test: capture output as log lines

tests/conftest.py

@@ -1,12 +1,34 @@
 import logging
 import pytest
+import subprocess
+import typing
 import unittest.mock
 
 from cipy import _handler
+from mocks import mock_run
 
 
 @pytest.fixture(scope="function")
-def ci_logger(monkeypatch: pytest.MonkeyPatch) -> unittest.mock.Mock:
+def mock_subprocess() -> typing.Iterator[unittest.mock.Mock]:
+    mock = unittest.mock.Mock()
+
+    def run(*args, **kwargs):
+        return mock_run(
+            *args,
+            **kwargs,
+            _poison_stdout=mock.stdout.encode("utf-8"),
+            _poison_stderr=mock.stderr.encode("utf-8"),
+        )
+
+    mock.side_effect = run
+    mock.stderr = ""
+    mock.stdout = ""
+    with unittest.mock.patch("subprocess.run", mock):
+        yield mock
+
+
+@pytest.fixture(scope="function")
+def ci_logger(monkeypatch: pytest.MonkeyPatch) -> typing.Iterator[unittest.mock.Mock]:
     mock = unittest.mock.Mock(spec=logging.Formatter)
     monkeypatch.setattr(_handler, "formatter", mock)
     level = logging.getLogger().level

tests/mocks.py

@@ -0,0 +1,23 @@
+import typing
+import subprocess
+
+
+def mock_run(
+    args: typing.Sequence[str],
+    *,
+    stdout: int | None = None,
+    stderr: int | None = None,
+    check: bool,  # Force an error if any of our mocked runs doesnt specify check
+    _poison_stdout: bytes = b"",
+    _poison_stderr: bytes = b"",
+    **kwargs: typing.Any,
+):
+    if stderr is not subprocess.PIPE:
+        assert _poison_stderr == b"", "Cannot poison STDERR when piping to STDOUT"
+
+    return subprocess.CompletedProcess(
+        args=args,
+        stdout=None if stdout is None else _poison_stdout,
+        stderr=None if stderr is None else _poison_stderr,
+        returncode=0,
+    )

tests/runner_test.py

@@ -104,3 +104,24 @@ def test_ipc_logs_outputs(ci_logger) -> None:
             call(HasAttributes(levelno=logging.DEBUG, args=("bar", "Lorem Ipsum"))),
         ]
     )
+
+
+def test_run_logs_when_pipe_is_false(ci_logger, mock_subprocess) -> None:
+    class NoOp(cipy.action.Action):
+        def run(self, context: cipy.Context) -> cipy.Status:
+            return cipy.Status.NOT_RUN
+
+    mock_subprocess.stdout = """
+This is a sentence
+Lorem Ipsum
+"""
+
+    action = NoOp(name="Test")
+    cipy.runner.run(action, [])
+
+    ci_logger.format.assert_has_calls(
+        [
+            call(HasAttributes(levelno=logging.INFO, message="This is a sentence")),
+            call(HasAttributes(levelno=logging.INFO, message="Lorem Ipsum")),
+        ]
+    )

tests/script_test.py

@@ -1,9 +1,7 @@
 # mypy: disable-error-code="attr-defined"
 import io
 import pytest
-import subprocess
 import typing
-import unittest.mock
 
 from pathlib import Path
 
@@ -11,56 +9,33 @@ import cipy
 import cipy.script
 
 from matchers import HasAttributes
+from mocks import mock_run
 
 
-def mock_run(
-    args: typing.Sequence[str],
-    *,
-    stdout: int | None = None,
-    stderr: int | None = None,
-    check: bool,  # Force an error if any of our mocked runs doesnt specify check
-    **kwargs: typing.Any,
-):
-    return subprocess.CompletedProcess(
-        args=args,
-        stdout=None if stdout is None else "",
-        stderr=None if stderr is None else "",
-        returncode=0,
-    )
-
-
-@pytest.fixture(scope="function", autouse=True)
-def disable_subprocess() -> typing.Iterator[None]:
-    mock = unittest.mock.Mock()
-    mock.side_effect = mock_run
-    with unittest.mock.patch("subprocess.run", mock):
-        yield
-
-
-def test_node_run_runs_main() -> None:
+def test_node_run_runs_main(mock_subprocess) -> None:
     action = cipy.script.NodeScript(name="example", main=Path("/dev/null"))
     action.run(cipy.Context())
 
-    assert subprocess.run.call_args_list[0].args == (["node", "/dev/null"],)
+    assert mock_subprocess.call_args_list[0].args == (["node", "/dev/null"],)
 
 
-def test_node_cleanup_noop_without_post() -> None:
+def test_node_cleanup_noop_without_post(mock_subprocess) -> None:
     action = cipy.script.NodeScript(name="example", main=Path("/dev/null"))
     action.cleanup(cipy.Context())
 
-    subprocess.run.assert_not_called()
+    mock_subprocess.assert_not_called()
 
 
-def test_node_cleanup_runs_post() -> None:
+def test_node_cleanup_runs_post(mock_subprocess) -> None:
     action = cipy.script.NodeScript(
         name="example", main=Path("/dev/null"), post=Path("/dev/zero")
     )
     action.cleanup(cipy.Context())
 
-    assert subprocess.run.call_args_list[0].args == (["node", "/dev/zero"],)
+    assert mock_subprocess.call_args_list[0].args == (["node", "/dev/zero"],)
 
 
-def test_script_writes_to_a_file() -> None:
+def test_script_writes_to_a_file(mock_subprocess) -> None:
     action = cipy.script.Script(script="""
     echo Lorem Ipsum
     """)
@@ -71,12 +46,12 @@ def test_script_writes_to_a_file() -> None:
             assert file.read() == action.script
         return mock_run(args, **kwargs)
 
-    subprocess.run.side_effect = check
+    mock_subprocess.side_effect = check
 
     action.run(cipy.Context())
 
 
-def test_script_performs_mustache_substitution() -> None:
+def test_script_performs_mustache_substitution(mock_subprocess) -> None:
     action = cipy.script.Script(script="""
     echo {{ foo }}
     """)
@@ -87,17 +62,17 @@ def test_script_performs_mustache_substitution() -> None:
             assert file.read() == "echo 1"
         return mock_run(args, **kwargs)
 
-    subprocess.run.side_effect = check
+    mock_subprocess.side_effect = check
 
     action.run(cipy.Context(foo=1))
 
 
-def test_script_propogates_mustache_error(ci_logger) -> None:
+def test_script_propogates_mustache_error(ci_logger, mock_subprocess) -> None:
     action = cipy.script.Script(script="""
     echo {{ foo }}
     """)
 
-    subprocess.run.side_effect = RuntimeError("Should not be reached")
+    mock_subprocess.side_effect = RuntimeError("Should not be reached")
 
     assert action.run(cipy.Context()) is cipy.Status.FAILURE
     ci_logger.format.assert_called_with(