feat(emitter): add a log dumping function (#331)

Author: bepri

craft_cli/messages.py

@@ -780,6 +780,18 @@ def error(self, error: errors.CraftError) -> None:
         self._report_error(error)
         self._stop()
 
+    @_active_guard()
+    def append_to_log(self, file: TextIO, prefix: str = ":: ") -> None:
+        """Dump the contents of an external log file into the emitter log.
+
+        :param file: A file I/O object to read from
+        :param prefix: A prefix for every line printed. Defaults to ":: ".
+        """
+        for line in file:
+            text = f"{prefix}{line}"
+            self._printer.log.write(text)
+        self._printer.log.flush()
+
     @_active_guard()
     def set_secrets(self, secrets: list[str]) -> None:
         """Set the list of strings that should be masked out in all output."""
@@ -830,6 +842,11 @@ def prompt(self, prompt_text: str, *, hide: bool = False) -> str:
             raise errors.CraftError("input cannot be empty")
         return val
 
+    @property
+    def log_filepath(self) -> pathlib.Path:
+        """The path to the log file."""
+        return self._log_filepath
+
 
 # module-level instantiated Emitter; this is the instance all code shall use and Emitter
 # shall not be instantiated again for the process' run

docs/changelog.rst

@@ -15,6 +15,12 @@ Breaking changes
     - Deprecates support for Python 3.8 and adds support for Python 3.11
       and 3.12.
 
+New features
+
+    - Add an ``append_to_log`` method to the emitter, which reads from a file
+      and dumps it directly into the log.
+    - Add a ``log_filepath`` read-only property to the emitter.
+
 2.15.0 (2025-Jan-23)
 --------------------
 

tests/integration/test_messages_integration.py

@@ -22,6 +22,7 @@
 
 import logging
 import os
+import pathlib
 import re
 import subprocess
 import sys
@@ -108,9 +109,9 @@ def compare_lines(expected_lines: Collection[Line], raw_stream: str, std_stream)
         else:
             # If the terminal doesn't support ANSI escape sequences, we fill the screen
             # width and don't terminate lines, so we split lines according to that length
-            assert (
-                len(raw_stream) % width == 0
-            ), f"Bad length {len(raw_stream)} ({width=}) {raw_stream=!r}"
+            assert len(raw_stream) % width == 0, (
+                f"Bad length {len(raw_stream)} ({width=}) {raw_stream=!r}"
+            )
             args = [iter(raw_stream)] * width
             lines = ["".join(x) for x in zip(*args)]  # pyright: ignore[reportGeneralTypeIssues]
     else:
@@ -909,7 +910,7 @@ def test_simple_errors_debugish(capsys, mode):
 @pytest.mark.parametrize("output_is_terminal", [True, False])
 def test_error_api_details_quiet(capsys):
     """Somewhat expected API error, final user modes.
-    
+
     Check that "quiet" is indeed quiet.
     """
     emit = Emitter()
@@ -929,7 +930,7 @@ def test_error_api_details_quiet(capsys):
     ]
     assert_outputs(capsys, emit, expected_err=expected_err, expected_log=expected_log)
 
-    
+
 @pytest.mark.parametrize(
     "mode",
     [
@@ -1684,3 +1685,50 @@ def test_open_stream_no_text(capsys, logger, monkeypatch, init_emitter):
         expected_err=expected_err,
         expected_log=expected_log,
     )
+
+
+@pytest.mark.parametrize(
+    "prefix",
+    [
+        pytest.param("", id="no_prefix"),
+        pytest.param(":: ", id="simple_prefix"),
+        pytest.param("2000-01-01 12:00:00.000", id="weird_prefix"),
+    ],
+)
+@pytest.mark.parametrize(
+    "file_contents",
+    [
+        pytest.param("", id="no_content"),
+        pytest.param("\n\n\n", id="only_newlines"),
+        pytest.param("one\ntwo\nthree", id="simple_content"),
+        pytest.param("one\ntwo\nthree\n", id="simple_with_newline"),
+    ],
+)
+@pytest.mark.parametrize(
+    "output_is_terminal", [pytest.param(True, id="is_term"), pytest.param(False, id="not_term")]
+)
+@pytest.mark.usefixtures("init_emitter")
+def test_append_to_log(tmp_path: pathlib.Path, prefix: str, file_contents: str) -> None:
+    file_to_dump = tmp_path / "file.txt"
+    file_to_dump.write_text(file_contents)
+
+    emit = Emitter()
+    emit.init(EmitterMode.QUIET, "testapp", GREETING)
+
+    emit.debug("We're having fun with it now!")
+    with file_to_dump.open() as file:
+        emit.append_to_log(file=file, prefix=prefix)
+    emit.debug("We are no longer having fun.")
+
+    prefixed_file_contents = [f"{prefix}{line}" for line in file_contents.splitlines()]
+    expected_output = [
+        rf"{TIMESTAMP_FORMAT}Specific greeting to be ignored",
+        rf"{TIMESTAMP_FORMAT}We're having fun with it now\!",
+        *prefixed_file_contents,
+        rf"{TIMESTAMP_FORMAT}We are no longer having fun\.",
+    ]
+
+    log_content = emit.log_filepath.read_text()
+
+    for expected, actual in zip(expected_output, log_content.splitlines()):
+        assert re.match(expected, actual)

tests/unit/test_messages_emitter.py

@@ -20,9 +20,8 @@
 import sys
 from unittest import mock
 from unittest.mock import call, patch
-from typing import cast, Callable
+from typing import Any, cast, Callable
 
-import craft_cli
 import pytest
 import pytest_mock
 
@@ -88,6 +87,34 @@ def func(mode, *, greeting="default greeting", **kwargs):
 
         yield func
 
+def emitter_methods(init: bool, stop: bool = True, exclude: list[str] = []) -> list[Callable[..., Any]]:
+    """Provide a list of all public methods on an Emitter object.
+
+    :param init: Whether or not to initialize the emitter first
+    :param stop: Whether or not to stop the emitter after initialization. Does nothing if init is False.
+        Defaults to true.
+    :param exclude: A list of method names to exclude from the final output. Defaults to empty.
+    """
+    emitter = Emitter()
+    if init:
+        emitter.init(EmitterMode.QUIET, "testappname", "default greeting")
+        if stop:
+            emitter.ended_ok()
+
+    # Collect all the public methods in Emitter
+    all_methods = [item for item in dir(Emitter) if item[0] != "_"]
+
+    # Filter out from the exclusion list
+    all_methods = [item for item in all_methods if item not in exclude]
+
+    # Get the actual attributes
+    all_methods = [getattr(emitter, item) for item in all_methods]
+
+    # Filter out anything that isn't actually a method
+    all_methods = [item for item in all_methods if isinstance(item, Callable)]
+
+    return all_methods
+
 
 # -- tests for init and setting/getting mode
 
@@ -180,12 +207,9 @@ def test_init_developer_modes(mode, tmp_path, monkeypatch):
     (handler,) = [x for x in logger.handlers if isinstance(x, _Handler)]
     assert handler.mode == mode
 
-
-@pytest.mark.parametrize("method_name", [x for x in dir(Emitter) if x[0] != "_" and x != "init"])
-def test_needs_init(method_name):
+@pytest.mark.parametrize("method", emitter_methods(init=False, exclude=["init"]))
+def test_needs_init(method):
     """Check that calling other methods needs emitter first to be initiated."""
-    emitter = Emitter()
-    method = getattr(emitter, method_name)
     with pytest.raises(RuntimeError, match="Emitter needs to be initiated first"):
         method()
 
@@ -775,17 +799,9 @@ def test_ended_double_after_error(get_initiated_emitter):
     emitter.ended_ok()
     assert emitter.printer_calls == []
 
-
-@pytest.mark.parametrize(
-    "method_name",
-    [x for x in dir(Emitter) if x[0] != "_" and x not in ("init", "ended_ok", "error")],
-)
-def test_needs_being_active(get_initiated_emitter, method_name):
+@pytest.mark.parametrize("method", emitter_methods(init=True, exclude=["init", "ended_ok", "error"]))
+def test_needs_being_active(method):
     """Check that calling public methods needs emitter to not be stopped."""
-    emitter = get_initiated_emitter(EmitterMode.QUIET)
-    emitter.ended_ok()
-
-    method = getattr(emitter, method_name)
     with pytest.raises(RuntimeError, match="Emitter is stopped already"):
         method()