cipy/tests/runner_test.py

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"))),
        ]
    )


class NoOp(cipy.action.Action):
    def run(self, context: cipy.Context) -> cipy.Status:
        return cipy.Status.NOT_RUN


def test_run_logs_when_pipe_is_false(ci_logger, mock_subprocess) -> None:
    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")),
        ]
    )


def test_run_returns_when_pipe_is_true(mock_subprocess) -> None:
    mock_subprocess.stdout = "Lorem Ipsum"
    mock_subprocess.stderr = "Ooga Booga!"

    action = NoOp(name="Test")
    _, stdout, stderr = cipy.runner.run(action, [], pipe=True)

    assert stdout == "Lorem Ipsum"
    assert stderr == "Ooga Booga!"


def test_preamble_logs_input_and_output(ci_logger) -> None:
    class PreambleAction(cipy.Action):
        inputs: IPCInputAction.Inputs = IPCInputAction.Inputs(foo=1, bar="Hello")
        outputs: IPCOutputAction.Outputs = IPCOutputAction.Outputs(
            foo=5, bar="Lorem Ipsum"
        )

        @cipy.runner.preamble
        def run(self, context: cipy.Context) -> cipy.Status:
            return cipy.Status.SUCCESS

    action = PreambleAction(name="Test")
    action.run(cipy.Context())

    ci_logger.format.assert_has_calls(
        [
            call(HasAttributes(levelno=logging.INFO, message="with:")),
            call(HasAttributes(levelno=logging.INFO, args=("foo", 1))),
            call(HasAttributes(levelno=logging.INFO, args=("bar", "Hello"))),
            call(HasAttributes(levelno=logging.DEBUG, message="outputs:")),
            call(HasAttributes(levelno=logging.DEBUG, args=("foo", 5))),
            call(HasAttributes(levelno=logging.DEBUG, args=("bar", "Lorem Ipsum"))),
        ]
    )