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

It is a very common pattern for craft tools to start the session
directly on a host machine, but launch another craft instance to
do the actual work inside LXD or Multipass.

In order for a seamless CLI experience (which looks identical
between a 'destructive build' on the host, LXD or Multipass),
we need support for progress messages (ephemeral=True) to seamlessly
cross the boundaries between the host and the container instance of
of the craft tool.

Please see examples.py canonical#26 for an example.

Add support for _Printer to respect ephemeral messages when
stopping. On printer stop(), these messages are no longer terminated
with a newline, but instead the printer will clear and reset the
cursor to the start of the line.

This behaviour change will allow subsequent terminal writes by a
different process the ability to reuse a terminal line previously
used as ephemeral output.

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:

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 is line inserted by the printer stop sequence to reset the
+        # last ephemeral print to terminal on exit.
+        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])
@@ -358,7 +365,10 @@ def test_progressbar_brief_terminal(capsys, monkeypatch):
         Line("Uploading stuff [████████████████████████       ] 1400/1788", permanent=False),
         Line("Uploading stuff [███████████████████████████████] 1788/1788", permanent=False),
         Line("Uploading stuff (<---)", permanent=False),
-        Line("And so on", permanent=True),
+        Line("And so on", permanent=False),
+        # This is line inserted by the printer stop sequence to reset the
+        # last ephemeral print to terminal on exit.
+        Line("", permanent=False),
     ]
     expected_log = [
         Line("Uploading stuff (--->)"),

tests/unit/test_messages_printer.py

@@ -905,7 +905,7 @@ 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_no_msg(capsys, log_filepath):
     """Stopping when stdout is not complete."""
     printer = _Printer(log_filepath)
     printer.unfinished_stream = sys.stdout
@@ -916,6 +916,31 @@ def test_stop_streams_unfinished_out(capsys, log_filepath):
     assert not err
 
 
+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)