test: cover cipy.runner.ipc and cipy.runner.environ

tests/conftest.py

@@ -9,4 +9,7 @@ from cipy import _handler
 def ci_logger(monkeypatch: pytest.MonkeyPatch) -> unittest.mock.Mock:
     mock = unittest.mock.Mock(spec=logging.Formatter)
     monkeypatch.setattr(_handler, "formatter", mock)
+    level = logging.getLogger().level
+    logging.getLogger().setLevel(logging.NOTSET)
     yield mock
+    logging.getLogger().setLevel(level)

tests/runner_test.py

@@ -0,0 +1,106 @@
+import logging
+import os
+import pytest
+
+from pathlib import Path
+from unittest.mock import call
+
+import cipy
+import cipy.runner
+
+from matchers import HasAttributes
+
+
+def test_environ_will_not_clobber_by_default(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setenv("EXAMPLE_FOR_TEST", "5")
+
+    assert os.environ["EXAMPLE_FOR_TEST"] == "5"
+    with pytest.raises(ValueError):
+        with cipy.runner.environ(EXAMPLE_FOR_TEST=2):
+            pass
+    assert os.environ["EXAMPLE_FOR_TEST"] == "5"
+
+
+def test_environ_applies_overrides_in_scope(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setenv("EXAMPLE_FOR_TEST", "5")
+
+    assert os.environ["EXAMPLE_FOR_TEST"] == "5"
+    with cipy.runner.environ(error_on_override=False, EXAMPLE_FOR_TEST=2):
+        assert os.environ["EXAMPLE_FOR_TEST"] == "2"
+    assert os.environ["EXAMPLE_FOR_TEST"] == "5"
+
+
+def test_environ_deletes_new_env_when_done(monkeypatch: pytest.MonkeyPatch) -> None:
+    assert "EXAMPLE_FOR_TEST" not in os.environ
+    with cipy.runner.environ(EXAMPLE_FOR_TEST=2):
+        assert os.environ["EXAMPLE_FOR_TEST"] == "2"
+    assert "EXAMPLE_FOR_TEST" not in os.environ
+
+
+class IPCAssertAction(cipy.action.Action):
+    @cipy.runner.ipc
+    def run(self, context: cipy.Context) -> cipy.Status:
+        assert Path(os.environ["CI_OUTPUT"]).is_file()
+        assert Path(os.environ["CI_ENVIRON"]).is_file()
+        return cipy.Status.SUCCESS
+
+
+class IPCInputAction(cipy.action.Action):
+    class Inputs(cipy.Inputs):
+        foo: int = 5
+        bar: str = "Lorem Ipsum"
+
+    inputs: Inputs = Inputs()
+
+    @cipy.runner.ipc
+    def run(self, context: cipy.Context) -> cipy.Status:
+        assert os.environ["INPUT_foo"] == "5"
+        assert os.environ["INPUT_bar"] == "Lorem Ipsum"
+        return cipy.Status.SUCCESS
+
+
+class IPCOutputAction(cipy.action.Action):
+    class Outputs(cipy.Outputs):
+        foo: int
+        bar: str
+
+    outputs: Outputs = cipy.required()
+
+    @cipy.runner.ipc
+    def run(self, context: cipy.Context) -> cipy.Status:
+        with open(os.environ["CI_OUTPUT"], "w", encoding="utf8") as file:
+            print("foo=5", file=file)
+            print('bar="Lorem Ipsum"', file=file)
+        return cipy.Status.SUCCESS
+
+
+def test_ipc_output_files_environs_are_paths() -> None:
+    action = IPCAssertAction(name="Test")
+    assert action.run(cipy.Context()) is cipy.Status.SUCCESS
+
+
+def test_ipc_injects_inputs_as_environ() -> None:
+    action = IPCInputAction(name="Test")
+    assert action.run(cipy.Context()) is cipy.Status.SUCCESS
+
+
+def test_ipc_extracts_outputs_from_tmpfile() -> None:
+    action = IPCOutputAction(name="Test")
+    assert action.outputs is None
+    assert action.run(cipy.Context()) is cipy.Status.SUCCESS
+    assert action.outputs.foo == 5
+    assert action.outputs.bar == "Lorem Ipsum"
+
+
+def test_ipc_logs_outputs(ci_logger) -> None:
+    action = IPCOutputAction(name="Test")
+    assert action.outputs is None
+    assert action.run(cipy.Context()) is cipy.Status.SUCCESS
+
+    ci_logger.format.assert_has_calls(
+        [
+            call(HasAttributes(levelno=logging.DEBUG, message="outputs:")),
+            call(HasAttributes(levelno=logging.DEBUG, args=("foo", 5))),
+            call(HasAttributes(levelno=logging.DEBUG, args=("bar", "Lorem Ipsum"))),
+        ]
+    )

tests/script_test.py

@@ -100,7 +100,6 @@ def test_script_propogates_mustache_error(ci_logger) -> None:
     subprocess.run.side_effect = RuntimeError("Should not be reached")
 
     assert action.run(cipy.Context()) is cipy.Status.FAILURE
-    record = ci_logger.format.call_args_list[0][0][0]
     ci_logger.format.assert_called_with(
         HasAttributes(name="Script", message="Could not find key 'foo'\n")
     )