Merge branch 'main' into constrained-global-args

.github/workflows/docs.yaml

@@ -21,7 +21,7 @@ jobs:
         with:
           fetch-depth: 0
       - name: Setup Python
-        uses: actions/setup-python@v4
+        uses: actions/setup-python@v5
         with:
           python-version: '3.11'
       - name: Install Tox
@@ -31,7 +31,7 @@ jobs:
       - name: Build documentation
         run: tox run -e build-docs
       - name: Upload documentation
-        uses: actions/upload-artifact@v3
+        uses: actions/upload-artifact@v4
         with:
           name: docs
           path: docs/_build/

.github/workflows/tests.yaml

@@ -17,7 +17,7 @@ jobs:
         with:
           fetch-depth: 0
       - name: Setup Python
-        uses: actions/setup-python@v4
+        uses: actions/setup-python@v5
         with:
           python-version: '3.10'
       - name: Configure environment
@@ -48,7 +48,7 @@ jobs:
         with:
           fetch-depth: 0
       - name: Set up Python versions on ${{ matrix.platform }}
-        uses: actions/setup-python@v4
+        uses: actions/setup-python@v5
         with:
           python-version: |
             3.8
@@ -73,7 +73,7 @@ jobs:
           files: coverage*.xml
       - name: Upload test results
         if: success() || failure()
-        uses: actions/upload-artifact@v3
+        uses: actions/upload-artifact@v4
         with:
           name: test-results-${{ matrix.platform }}
           path: results/

craft_cli/messages.py

@@ -239,8 +239,9 @@ def _write(self, data: bytes) -> None:
             useful_line = data[pointer:newline_position]
             pointer = newline_position + 1
 
-            # write the useful line to intended outputs
-            unicode_line = useful_line.decode("utf8")
+            # write the useful line to intended outputs. Decode with errors="replace"
+            # here because we don't know where this line is coming from.
+            unicode_line = useful_line.decode("utf8", errors="replace")
             # replace tabs with a set number of spaces so that the printer
             # can correctly count the characters.
             unicode_line = unicode_line.replace("\t", "  ")

craft_cli/printer.py

@@ -56,7 +56,7 @@ class _MessageInfo:
     bar_total: int | float | None = None
     use_timestamp: bool = False
     end_line: bool = False
-    created_at: datetime = field(default_factory=datetime.now)
+    created_at: datetime = field(default_factory=datetime.now, compare=False)
     terminal_prefix: str = ""
 
 
@@ -71,6 +71,27 @@ def _get_terminal_width() -> int:
     return shutil.get_terminal_size().columns
 
 
+def _format_term_line(prefix: str, text: str, spintext: str, *, ephemeral: bool) -> str:
+    """Format a line to print to the terminal."""
+    # fill with spaces until the very end, on one hand to clear a possible previous message,
+    # but also to always have the cursor at the very end
+    width = _get_terminal_width()
+    usable = width - len(spintext) - 1  # the 1 is the cursor itself
+    if len(text) > usable:
+        if ephemeral:
+            text = text[: usable - 1] + "…"
+        elif spintext:
+            # we need to rewrite the message with the spintext, use only the last line for
+            # multiline messages, and ensure (again) that the last real line fits
+            remaining_for_last_line = len(text) % width
+            text = text[-remaining_for_last_line:]
+            if len(text) > usable:
+                text = text[: usable - 1] + "…"
+    cleaner = " " * (usable - len(text) % width)
+
+    return prefix + text + spintext + cleaner
+
+
 class _Spinner(threading.Thread):
     """A supervisor thread that will repeat long-standing messages with a spinner besides it.
 
@@ -105,6 +126,9 @@ def __init__(self, printer: Printer) -> None:
         # a lock to wait the spinner to stop spinning
         self.lock = threading.Lock()
 
+        # Keep the message under supervision available for examination.
+        self._under_supervision: _MessageInfo | None = None
+
     def run(self) -> None:
         prv_msg = None
         t_init = time.time()
@@ -136,6 +160,11 @@ def run(self) -> None:
 
     def supervise(self, message: _MessageInfo | None) -> None:
         """Supervise a message to spin it if it remains too long."""
+        # Don't bother the spinner if we're repeating the same message
+        if message == self._under_supervision:
+            return
+
+        self._under_supervision = message
         self.queue.put(message)
         # (maybe) wait for the spinner to exit spinning state (which does some cleaning)
         self.lock.acquire()
@@ -200,7 +229,6 @@ def _get_prefixed_message_text(self, message: _MessageInfo) -> str:
     def _write_line_terminal(self, message: _MessageInfo, *, spintext: str = "") -> None:
         """Write a simple line message to the screen."""
         # prepare the text with (maybe) the timestamp
-
         text = self._get_prefixed_message_text(message)
 
         if message.use_timestamp:
@@ -226,24 +254,12 @@ def _write_line_terminal(self, message: _MessageInfo, *, spintext: str = "") ->
             maybe_cr = ""
             print(flush=True, file=self.prv_msg.stream)
 
-        # fill with spaces until the very end, on one hand to clear a possible previous message,
-        # but also to always have the cursor at the very end
-        width = _get_terminal_width()
-        usable = width - len(spintext) - 1  # the 1 is the cursor itself
-        if len(text) > usable:
-            if message.ephemeral:
-                text = text[: usable - 1] + "…"
-            elif spintext:
-                # we need to rewrite the message with the spintext, use only the last line for
-                # multiline messages, and ensure (again) that the last real line fits
-                remaining_for_last_line = len(text) % width
-                text = text[-remaining_for_last_line:]
-                if len(text) > usable:
-                    text = text[: usable - 1] + "…"
-        cleaner = " " * (usable - len(text) % width)
-
-        line = maybe_cr + text + spintext + cleaner
-        print(line, end="", flush=True, file=message.stream)
+        # We don't need to rewrite the same ephemeral message repeatedly.
+        should_overwrite = spintext or message.end_line or not message.ephemeral
+        if should_overwrite or message != self.prv_msg:
+            line = _format_term_line(maybe_cr, text, spintext, ephemeral=message.ephemeral)
+            print(line, end="", flush=True, file=message.stream)
+
         if message.end_line:
             # finish the just shown line, as we need a clean terminal for some external thing
             print(flush=True, file=message.stream)

docs/.sphinx/requirements.txt

@@ -14,5 +14,5 @@ sphinxcontrib-jquery
 sphinxext-opengraph
 # Extra, craft-cli-specific requirements for the docs
 sphinx-toolbox==3.5.0
-sphinx-lint==0.9.0
+sphinx-lint==0.9.1
 pytest>=7.0.0 # This is just because this is imported by the code

examples.py

@@ -486,6 +486,33 @@ def _call_lib(logger, index):
     time.sleep(2)
 
 
+def example_30():
+    """Message spamming, noting the different spinner behaviour"""
+    emit.progress(
+        "Message spamming example. The same message will be spammed for 10s, but "
+        "it will appear as one message with a spinner.",
+        permanent=True,
+    )
+    end_time = time.monotonic() + 10
+    while time.monotonic() < end_time:
+        emit.progress("SPAM SPAM SPAM SPAM")
+        time.sleep(0.001)
+    emit.progress(
+        "Now two separate messages will be spammed and no spinner appear.", permanent=True
+    )
+    end_time = time.monotonic() + 10
+    while time.monotonic() < end_time:
+        emit.progress("SPAM SPAM SPAM SPAM")
+        time.sleep(0.01)
+        emit.progress("SPAM SPAM SPAM baked beans")
+        time.sleep(0.01)
+    emit.progress("And back to the first message!", permanent=True)
+    end_time = time.monotonic() + 10
+    while time.monotonic() < end_time:
+        emit.progress("SPAM SPAM SPAM SPAM")
+        time.sleep(0.001)
+
+
 # -- end of test cases
 
 if len(sys.argv) < 2:

pyproject.toml

@@ -43,31 +43,31 @@ emitter = "craft_cli.pytest_plugin"
 
 [project.optional-dependencies]
 dev = [
-    "coverage[toml]==7.3.2",
+    "coverage[toml]==7.4.1",
     "hypothesis==6.92.9",
-    "pytest==7.4.3",
+    "pytest==8.0.0",
     "pytest-check==2.2.4",
     "pytest-cov==4.1.0",
     "pytest-mock==3.12.0",
     "pytest-subprocess"
 ]
 lint = [
-    "black==23.11.0",
+    "black==24.1.1",
     "codespell[toml]==2.2.6",
-    "ruff==0.1.7",
+    "ruff==0.1.15",
     "yamllint==1.33.0"
 ]
 types = [
-    "mypy[reports]==1.7.1",
-    "pyright==1.1.337",
+    "mypy[reports]==1.8.0",
+    "pyright==1.1.349",
     "types-Pygments",
     "types-colorama",
     "types-setuptools",
 ]
 
 [build-system]
 requires = [
-    "setuptools==67.7.2",
+    "setuptools==69.0.3",
     "setuptools_scm[toml]>=7.1"
 ]
 build-backend = "setuptools.build_meta"

tests/unit/test_messages_stream_cm.py

@@ -257,6 +257,24 @@ def test_pipereader_tabs(recording_printer, stream):
     assert msg.text == "::   123  456"  # tabs expanded into 2 spaces
 
 
+@pytest.mark.parametrize(
+    ("invalid_text", "expected"),
+    [(b"\xf0\x28\x8c\xbc", "�(��"), (b"\xf0\x90\x28\xbc", "�(�"), (b"\xf0\x90\x8c\x28", "�(")],
+)
+def test_pipereader_invalid_utf8(recording_printer, invalid_text, expected):
+    """Check that bytes that aren't valid utf-8 text don't crash."""
+    invalid_bytes = b"valid prefix " + invalid_text + b" valid suffix\n"
+
+    flags = {"use_timestamp": False, "ephemeral": False, "end_line": True}
+    prt = _PipeReaderThread(recording_printer, sys.stdout, flags)
+    prt.start()
+    os.write(prt.write_pipe, invalid_bytes)
+    prt.stop()
+
+    (msg,) = recording_printer.written_terminal_lines
+    assert msg.text == f":: valid prefix {expected} valid suffix"
+
+
 def test_pipereader_chunk_assembler(recording_printer, monkeypatch):
     """Converts ok arbitrary chunks to lines."""
     monkeypatch.setattr(messages, "_PIPE_READER_CHUNK_SIZE", 5)

tests/unit/test_printer.py

@@ -364,6 +364,82 @@ def test_writelineterminal_spintext_length_just_exceeded(capsys, monkeypatch, lo
     assert out == "\r0x1x2x3x4x… * 3.15s"
 
 
+@pytest.mark.parametrize("test_text", ["", "Some test text."])
+def test_writelineterminal_ephemeral_spam(capsys, monkeypatch, log_filepath, test_text):
+    """Spam _write_line_terminal with the same message over and over."""
+    monkeypatch.setattr(printermod, "_get_terminal_width", lambda: 40)
+    printer = Printer(log_filepath)
+
+    for _ in range(10):
+        # Recreate the message here so we're checking equality, not just identity.
+        msg = _MessageInfo(sys.stdout, test_text, end_line=False, ephemeral=True)
+        printer._write_line_terminal(msg)
+        printer.prv_msg = msg
+
+    assert printer.unfinished_stream == sys.stdout
+
+    out, err = capsys.readouterr()
+    assert not err
+
+    # output completes the terminal width (leaving space for the cursor), and
+    # without a finishing newline
+    # There will only be one copy of the text.
+    assert out == test_text[:40] + " " * (39 - len(test_text))
+
+
+@pytest.mark.parametrize(("ephemeral", "end_line"), [(False, False), (False, True), (True, True)])
+@pytest.mark.parametrize("text", ["", "Some test text"])
+def test_writelineterminal_rewrites_same_message(
+    capsys, monkeypatch, log_filepath, text, ephemeral, end_line
+):
+    """Spam _write_line_terminal with the same message and ensure it keeps writing."""
+    monkeypatch.setattr(printermod, "_get_terminal_width", lambda: 40)
+    printer = Printer(log_filepath)
+
+    for _ in range(10):
+        message = _MessageInfo(sys.stdout, text, ephemeral=ephemeral, end_line=end_line)
+        printer._write_line_terminal(message)
+        printer.prv_msg = message
+
+    out, err = capsys.readouterr()
+    assert not err
+
+    # output completes the terminal width (leaving space for the cursor), and
+    # without a finishing newline
+    assert out.strip() == "\n".join([text + " " * (39 - len(text))] * 10).strip()
+
+
+@pytest.mark.parametrize("ephemeral", [True, False])
+@pytest.mark.parametrize("text", ["", "Some test text"])
+@pytest.mark.parametrize(
+    "spintext",
+    [
+        "!!!!!!!!!!",
+        "\\|/-",
+        "1234567890",
+    ],
+)
+def test_writelineterminal_rewrites_same_message_with_spintext(
+    capsys, monkeypatch, log_filepath, text, spintext, ephemeral
+):
+    """Spam _write_line_terminal with the same message over and over."""
+    monkeypatch.setattr(printermod, "_get_terminal_width", lambda: 40)
+    printer = Printer(log_filepath)
+
+    for spin in spintext:
+        message = _MessageInfo(sys.stdout, text, ephemeral=ephemeral, end_line=False)
+        printer._write_line_terminal(message, spintext=spin)
+        printer.prv_msg = message
+
+    out, err = capsys.readouterr()
+    assert not err
+
+    # output completes the terminal width (leaving space for the cursor), and
+    # without a finishing newline
+    expected = "\r".join(text + s + " " * (39 - len(text) - len(s)) for s in spintext)
+    assert out.strip() == expected.strip()
+
+
 # -- tests for the writing line (captured version) function
 
 
@@ -1055,6 +1131,46 @@ def test_spinner_working_simple(spinner, monkeypatch):
     assert spinner.printer.spinned[-1] == (msg, " ")
 
 
+def test_spinner_spam(spinner, monkeypatch):
+    """Test that the spinner works properly when spamming the same message.
+
+    The expected behaviour is to ignore the existence of the fresh message and just
+    write when the spinner needs to update.
+    """
+    # set absurdly low times so we can have several spin texts in the test
+    monkeypatch.setattr(printermod, "_SPINNER_THRESHOLD", 0.001)
+    monkeypatch.setattr(printermod, "_SPINNER_DELAY", 0.001)
+
+    # send a message, wait enough until we have enough spinned to test, and turn it off
+    msg = _MessageInfo(sys.stdout, "test msg")
+    for _ in range(100):
+        spinner.supervise(_MessageInfo(sys.stdout, "test msg"))
+    for _ in range(100):
+        if len(spinner.printer.spinned) >= 6:
+            break
+        time.sleep(0.01)
+    else:
+        pytest.fail("Waited too long for the _Spinner to generate messages")
+    spinner.supervise(None)
+    to_check = spinner.printer.spinned[:5]
+
+    # check the initial messages complete the "spinner drawing" also showing elapsed time
+    spinned_messages, spinned_texts = list(zip(*to_check))
+    assert all(spinned_msg == msg for spinned_msg in spinned_messages)
+    expected_texts = (
+        r" - \(\d\.\ds\)",
+        r" \\ \(\d\.\ds\)",
+        r" | \(\d\.\ds\)",
+        r" / \(\d\.\ds\)",
+        r" - \(\d\.\ds\)",
+    )
+    for expected, real in list(zip(expected_texts, spinned_texts)):
+        assert re.match(expected, real)
+
+    # the last message should clean the spinner
+    assert spinner.printer.spinned[-1] == (msg, " ")
+
+
 def test_spinner_two_messages(spinner, monkeypatch):
     """Two consecutive messages with spinner."""
     # set absurdly low times so we can have several spin texts in the test