cipy/tests/action_test.py

import pytest
import unittest.mock

from unittest.mock import ANY

import cipy
import cipy.action


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


def mock_action(*, name: str = "Mock", id: str = "") -> unittest.mock.MagicMock:
    rval = unittest.mock.MagicMock(
        wraps=NoOp(name=name, id=id), spec=cipy.action.Action
    )
    rval.name = name
    rval.id = id
    rval.inputs = cipy.Inputs()
    rval.outputs = cipy.Outputs()
    return rval


def test_action() -> None:
    action = NoOp(name="Stub")
    assert action.run(cipy.Context()) is cipy.Status.NOT_RUN


def test_call_proxies_functions() -> None:
    action = mock_action()
    call = cipy.Call(action)

    call.is_enabled(cipy.Status.NOT_RUN, cipy.Context())
    action.is_enabled.assert_called_once()

    call.run(cipy.Context())
    action.run.assert_called_once()

    call.cleanup(cipy.Context())
    action.cleanup.assert_called_once()


def test_call_can_pass_args() -> None:
    class Inputs(cipy.Inputs):
        foo: int = cipy.required()

    action = mock_action()
    action.inputs = Inputs()
    call = cipy.Call(action, foo=3)

    call.run(cipy.Context())

    action.run.assert_called_once()
    assert action.inputs.foo == 3


def test_composite_runs_steps_in_order() -> None:
    run = unittest.mock.Mock()

    a1 = mock_action(name="A")
    a2 = mock_action(name="B")

    a1.run = run.a1
    a1.run.return_value = cipy.Status.SUCCESS
    a2.run = run.a2
    a2.run.return_value = cipy.Status.SUCCESS

    comp = cipy.Composite(a1, a2)
    comp.run(cipy.Context())

    run.assert_has_calls(
        [unittest.mock.call.a1(ANY), unittest.mock.call.a2(ANY)], any_order=False
    )


def test_composite_checks_is_enabled_for_each() -> None:
    a1 = mock_action(name="A")
    a1.run.return_value = cipy.Status.FAILURE
    a2 = mock_action(name="B")
    a2.run.return_value = cipy.Status.SUCCESS

    comp = cipy.Composite(a1, a2)
    comp.run(cipy.Context())

    a1.run.assert_called_once()
    a2.is_enabled.assert_called_once()
    a2.run.assert_not_called()


def test_composite_cleanup_goes_backwards() -> None:
    cleanup = unittest.mock.Mock()

    a1 = mock_action(name="A")
    a2 = mock_action(name="B")

    a1.run.return_value = cipy.Status.SUCCESS
    a2.run.return_value = cipy.Status.SUCCESS

    a1.cleanup = cleanup.a1
    a2.cleanup = cleanup.a2

    comp = cipy.Composite(a1, a2)
    comp.run(cipy.Context())
    comp.cleanup(cipy.Context())

    cleanup.assert_has_calls(
        [unittest.mock.call.a2(ANY), unittest.mock.call.a1(ANY)], any_order=False
    )


def test_composite_cleanup_ignores_skipped() -> None:
    a1 = mock_action(name="A")
    a1.run.return_value = cipy.Status.SKIPPED
    a2 = mock_action(name="B")
    a2.run.return_value = cipy.Status.SUCCESS

    comp = cipy.Composite(a1, a2)
    comp.run(cipy.Context())
    comp.cleanup(cipy.Context())

    a2.cleanup.assert_called_once()
    a1.cleanup.assert_not_called()