cipy/tests/test_workflow.py

import logging
import pytest

from unittest.mock import ANY, Mock, call, patch

import cipy
import cipy.workflow

from cipy.workflow import Job

from matchers import HasAttributes


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


class MockRunAction(cipy.action.Action, arbitrary_types_allowed=True):
    mock: Mock

    def run(self, context: cipy.Context) -> cipy.Status:
        return self.mock.run(name=self.name, context=context)


def test_workflow_constructs_depends_map() -> None:
    workflow = cipy.workflow.Workflow(
        name="Test",
        jobs=[
            Job(id="A", action=NoOp(name="A")),
            Job(id="B", needs=["A"], action=NoOp(name="B")),
            Job(id="C", needs=["A"], action=NoOp(name="C")),
            Job(id="D", needs=["B", "C"], action=NoOp(name="D")),
        ],
    )

    assert workflow._depends == {
        "A": [],
        "B": ["A"],
        "C": ["A"],
        "D": ["B", "C"],
    }


def test_workflow_calls_in_order() -> None:
    mock = Mock()
    mock.run.return_value = cipy.Status.SUCCESS

    workflow = cipy.workflow.Workflow(
        name="Test",
        jobs=[
            Job(id="A", action=MockRunAction(name="A", mock=mock)),
            Job(id="B", needs=["A"], action=MockRunAction(name="B", mock=mock)),
            Job(id="C", needs=["A"], action=MockRunAction(name="C", mock=mock)),
            Job(id="D", needs=["B", "C"], action=MockRunAction(name="D", mock=mock)),
        ],
    )

    workflow.run(cipy.Context())
    mock.run.assert_has_calls(
        [
            call(name="A", context=ANY),
            call(name="B", context=ANY),
            call(name="C", context=ANY),
            call(name="D", context=ANY),
        ],
        any_order=False,
    )


def test_workflow_stops_after_failure() -> None:
    mock = Mock()
    mock.run.side_effect = [cipy.Status.SUCCESS, cipy.Status.FAILURE]

    workflow = cipy.workflow.Workflow(
        name="Test",
        jobs=[
            Job(id="A", action=MockRunAction(name="A", mock=mock)),
            Job(id="B", needs=["A"], action=MockRunAction(name="B", mock=mock)),
            Job(id="C", needs=["A"], action=MockRunAction(name="C", mock=mock)),
            Job(id="D", needs=["B", "C"], action=MockRunAction(name="D", mock=mock)),
        ],
    )

    workflow.run(cipy.Context())
    mock.run.assert_has_calls(
        [
            call(name="A", context=ANY),
            call(name="B", context=ANY),
        ],
        any_order=False,
    )
    assert len(mock.run.call_args_list) == 2


def test_matrix_expands_combinatorally(ci_logger) -> None:
    mock = Mock()
    mock.run.return_value = cipy.Status.SUCCESS

    matrix = cipy.workflow.Matrix(
        on={
            "compiler": ["clang12", "clang18"],
            "os": ["ubuntu22", "ubuntu24"],
            "release": [True, False],
        },
        uses=MockRunAction(name="Test", mock=mock),
    )

    with patch("copy.deepcopy", side_effect=lambda x: x):
        matrix.run(cipy.Context())
        assert len(mock.run.call_args_list) == 8

    def call_with_args(**kwargs):
        return call(HasAttributes(levelno=logging.INFO, args=({**kwargs})))

    ci_logger.format.assert_has_calls(
        [
            call_with_args(compiler="clang12", os="ubuntu22", release=True),
            call_with_args(compiler="clang12", os="ubuntu22", release=False),
            call_with_args(compiler="clang12", os="ubuntu24", release=True),
            call_with_args(compiler="clang12", os="ubuntu24", release=False),
            call_with_args(compiler="clang18", os="ubuntu22", release=True),
            call_with_args(compiler="clang18", os="ubuntu22", release=False),
            call_with_args(compiler="clang18", os="ubuntu24", release=True),
            call_with_args(compiler="clang18", os="ubuntu24", release=False),
        ],
        any_order=True,
    )


def test_matrix_deep_copies_used_action(ci_logger) -> None:
    mock = Mock()
    mock.run.return_value = cipy.Status.SUCCESS

    matrix = cipy.workflow.Matrix(
        on={
            "compiler": ["clang12", "clang18"],
            "os": ["ubuntu22", "ubuntu24"],
            "release": [True, False],
        },
        uses=MockRunAction(name="Test", mock=mock),
    )

    matrix.run(cipy.Context())
    assert len(mock.run.call_args_list) == 0


def test_matrix_fails_fast(ci_logger) -> None:
    mock = Mock()
    mock.run.side_effect = [cipy.Status.SUCCESS, cipy.Status.FAILURE]
    mock.run.return_value = cipy.Status.SUCCESS

    matrix = cipy.workflow.Matrix(
        on={
            "compiler": ["clang12", "clang18"],
            "os": ["ubuntu22", "ubuntu24"],
            "release": [True, False],
        },
        uses=MockRunAction(name="Test", mock=mock),
    )

    with patch("copy.deepcopy", side_effect=lambda x: x):
        matrix.run(cipy.Context())
        assert len(mock.run.call_args_list) == 2


def test_matrix_can_disable_failfast(ci_logger) -> None:
    mock = Mock()
    mock.run.side_effect = [
        cipy.Status.SUCCESS,
        cipy.Status.FAILURE,
        cipy.Status.SUCCESS,
        cipy.Status.SUCCESS,
    ]

    matrix = cipy.workflow.Matrix(
        on={
            "compiler": ["clang12", "clang18"],
            "os": ["ubuntu22", "ubuntu24"],
        },
        uses=MockRunAction(name="Test", mock=mock),
        fail_fast=False,
    )

    with patch("copy.deepcopy", side_effect=lambda x: x):
        matrix.run(cipy.Context())
        assert len(mock.run.call_args_list) == 4