printer: enhance ephemeral message handover after stop() or pause()

craft_cli/messages.py

@@ -234,7 +234,7 @@ def __init__(self, log_filepath: pathlib.Path) -> None:
         # open the log file (will be closed explicitly later)
         self.log = open(log_filepath, "at", encoding="utf8")  # pylint: disable=consider-using-with
 
-        # keep account of output streams with unfinished lines
+        # keep account of output terminal streams with unfinished lines
         self.unfinished_stream: Optional[TextIO] = None
 
         # run the spinner supervisor
@@ -439,7 +439,18 @@ def stop(self) -> None:
         if not TESTMODE:
             self.spinner.stop()
         if self.unfinished_stream is not None:
-            print(flush=True, file=self.unfinished_stream)
+            # With unfinished_stream set, the prv_msg object is valid.
+            if self.prv_msg.ephemeral:  # type: ignore
+                # If the last printed message is of 'ephemeral' type, the stop
+                # request must clean and reset the line.
+                cleaner = " " * (_get_terminal_width() - 1)
+                line = "\r" + cleaner + "\r"
+                print(line, end="", flush=True, file=self.prv_msg.stream)  # type: ignore
+            else:
+                # The last printed message is permanent. Leave the cursor on
+                # the next clean line.
+                print(flush=True, file=self.unfinished_stream)
+
         self.log.close()
         self.stopped = True
 

examples.py

@@ -335,6 +335,45 @@ def example_25():
     emit.message("The meaning of life is 42.")
 
 
+def example_26():
+    """Show emitter progress message handover.
+
+    This example demonstrates seamless emitter progress message handover
+    between two craft tools. Handover uses emit.pause() on the local
+    craft tool before an LXD launched craft tool takes over, and hands back.
+    """
+    emit.set_mode(EmitterMode.BRIEF)
+
+    lxd_craft_tool = textwrap.dedent(
+        """
+        import time
+
+        from craft_cli import emit, EmitterMode
+
+        emit.init(EmitterMode.BRIEF, "subapp", "An example sub application.")
+        emit.progress("seamless progress #2")
+        time.sleep(2)
+        emit.progress("seamless progress #3")
+        time.sleep(2)
+        emit.ended_ok()
+    """
+    )
+    temp_fh, temp_name = tempfile.mkstemp()
+    with open(temp_fh, "wt", encoding="utf8") as fh:
+        fh.write(lxd_craft_tool)
+
+    emit.message("Application Start.")
+    emit.progress("seamless progress #1")
+    time.sleep(2)
+    with emit.pause():
+        cmd = [sys.executable, temp_name]
+        subprocess.run(cmd, env={"PYTHONPATH": os.getcwd()}, capture_output=False, text=True)
+        os.unlink(temp_name)
+    emit.progress("seamless progress #4")
+    time.sleep(2)
+    emit.message("Application End.")
+
+
 # -- end of test cases
 
 if len(sys.argv) != 2:

setup.cfg

@@ -88,6 +88,7 @@ universal = 1
 [codespell]
 quiet-level = 3
 skip = ./docs/_build,.direnv,.git,.mypy_cache,.pytest_cache,.venv,__pycache__,venv
+ignore-words-list = dedented
 
 [flake8]
 exclude = .direnv .git .mypy_cache .pytest_cache .venv __pycache__ venv

tests/unit/test_messages_integration.py

@@ -230,11 +230,18 @@ def test_progress_brief_terminal(capsys):
     emit.progress("Another message.")
     emit.ended_ok()
 
-    expected = [
+    expected_term = [
         Line("The meaning of life is 42.", permanent=False),
-        Line("Another message.", permanent=True),  # stays as it's the last message
+        Line("Another message.", permanent=False),
+        # This cleaner line is inserted by the printer stop
+        # sequence to reset the last ephemeral print to terminal.
+        Line("", permanent=False),
     ]
-    assert_outputs(capsys, emit, expected_err=expected, expected_log=expected)
+    expected_log = [
+        Line("The meaning of life is 42.", permanent=False),
+        Line("Another message.", permanent=False),
+    ]
+    assert_outputs(capsys, emit, expected_err=expected_term, expected_log=expected_log)
 
 
 @pytest.mark.parametrize("output_is_terminal", [False])
@@ -352,6 +359,49 @@ def test_progressbar_brief_terminal(capsys, monkeypatch):
     emit.progress("And so on")  # just a line so last progress line is not artificially permanent
     emit.ended_ok()
 
+    expected_screen = [
+        Line("Uploading stuff (--->)", permanent=False),
+        Line("Uploading stuff [████████████                    ] 700/1788", permanent=False),
+        Line("Uploading stuff [████████████████████████       ] 1400/1788", permanent=False),
+        Line("Uploading stuff [███████████████████████████████] 1788/1788", permanent=False),
+        Line("Uploading stuff (<---)", permanent=False),
+        Line("And so on", permanent=False),
+        # This cleaner line is inserted by the printer stop
+        # sequence to reset the last ephemeral print to terminal.
+        Line("", permanent=False),
+    ]
+    expected_log = [
+        Line("Uploading stuff (--->)"),
+        Line("Uploading stuff (<---)"),
+        Line("And so on"),
+    ]
+    assert_outputs(capsys, emit, expected_err=expected_screen, expected_log=expected_log)
+
+
+@pytest.mark.parametrize("output_is_terminal", [True])
+def test_progressbar_brief_permanent_terminal(capsys, monkeypatch):
+    """Show a progress bar in brief mode."""
+    # fake size so lines to compare are static
+    monkeypatch.setattr(messages, "_get_terminal_width", lambda: 60)
+
+    emit = Emitter()
+
+    # patch `set_mode` so it's not really run and set the mode manually, as we do NOT want
+    # the "Logging execution..." message to be sent to screen because it's too long and will
+    # break the tests. Note we want the fake terminal width to be small so we can "draw" here
+    # in the test the progress bar we want to see.
+    emit.set_mode = lambda mode: None
+    emit.init(EmitterMode.BRIEF, "testapp", GREETING)
+    emit._mode = EmitterMode.BRIEF
+
+    with emit.progress_bar("Uploading stuff", 1788) as progress:
+        for uploaded in [700, 700, 388]:
+            progress.advance(uploaded)
+    emit.progress(
+        "And so on", permanent=True
+    )  # just a line so last progress line is not artificially permanent
+    emit.ended_ok()
+
     expected_screen = [
         Line("Uploading stuff (--->)", permanent=False),
         Line("Uploading stuff [████████████                    ] 700/1788", permanent=False),

tests/unit/test_messages_printer.py

@@ -905,21 +905,36 @@ def test_stop_streams_ok(capsys, log_filepath):
     assert not err
 
 
-def test_stop_streams_unfinished_out(capsys, log_filepath):
+def test_stop_streams_unfinished_out_non_ephemeral(capsys, log_filepath, monkeypatch):
     """Stopping when stdout is not complete."""
     printer = _Printer(log_filepath)
     printer.unfinished_stream = sys.stdout
+    printer.prv_msg = _MessageInfo(sys.stdout, "test")
     printer.stop()
 
     out, err = capsys.readouterr()
     assert out == "\n"
     assert not err
 
 
+def test_stop_streams_unfinished_out_ephemeral(capsys, log_filepath, monkeypatch):
+    """Stopping when stdout is not complete."""
+    monkeypatch.setattr(messages, "_get_terminal_width", lambda: 10)
+    printer = _Printer(log_filepath)
+    printer.unfinished_stream = sys.stdout
+    printer.prv_msg = _MessageInfo(sys.stdout, "test", ephemeral=True)
+    printer.stop()
+
+    out, err = capsys.readouterr()
+    assert out == "\r         \r"  # 9 spaces
+    assert not err
+
+
 def test_stop_streams_unfinished_err(capsys, log_filepath):
     """Stopping when stderr is not complete."""
     printer = _Printer(log_filepath)
     printer.unfinished_stream = sys.stderr
+    printer.prv_msg = _MessageInfo(sys.stderr, "test")
     printer.stop()
 
     out, err = capsys.readouterr()