cipy/tests/context_test.py

import functools
import pytest

from unittest.mock import Mock, patch

from pydantic import BaseModel, Field
from pydantic_core import ValidationError

import cipy
from cipy.context import _Stub, Results

from cipy.common import Inputs, Outputs


def test_required_is_secretly_none() -> None:
    class Model(BaseModel):
        inputs: Inputs = cipy.required()

    model = Model()
    assert model.inputs is None


def test_required_will_fail_if_reconstructed() -> None:
    class Model(BaseModel):
        inputs: Inputs = cipy.required()

    model = Model()
    with pytest.raises(ValidationError):
        model = Model(**vars(model))


def test_fabricate_empty_inputs() -> None:
    class Model(BaseModel):
        inputs: Inputs = cipy.required()

    model = Model()
    context = cipy.Context()
    assert context.fabricate(model, "inputs") is not None
    assert model.inputs is not None


def test_fabricate_empty_outputs() -> None:
    class Model(BaseModel):
        outputs: Outputs = cipy.required()

    model = Model()
    context = cipy.Context()
    assert context.fabricate(model, "outputs") is not None
    assert model.outputs is not None


def test_fabricate_data() -> None:
    class Model(BaseModel):
        class _Inputs(Inputs):
            foo: int = 0
            bar: str = ""

        inputs: _Inputs = cipy.required()

    model = Model()
    context = cipy.Context()

    assert model.inputs is None
    assert context.fabricate(model, "inputs") is not None
    assert model.inputs.foo == 0
    assert model.inputs.bar == ""


def test_fabricate_can_require_values() -> None:
    class Model(BaseModel):
        class _Inputs(Inputs):
            foo: int = cipy.required()
            bar: str = ""

        inputs: _Inputs = cipy.required()

    model = Model()
    context = cipy.Context()

    assert model.inputs is None
    with pytest.raises(ValidationError):
        context.fabricate(model, "inputs")


def test_fabricate_can_use_prior_values() -> None:
    class Model(BaseModel):
        class _Inputs(Inputs):
            foo: int
            bar: str = ""

        inputs: _Inputs = cipy.required()

    model = Model(inputs=Model._Inputs(foo=1))
    context = cipy.Context()

    assert context.fabricate(model, "inputs") is not None
    assert model.inputs.foo == 1


def test_fabricate_can_provide_extras() -> None:
    class Model(BaseModel):
        class _Inputs(Inputs):
            foo: int
            bar: str = ""

        inputs: _Inputs = cipy.required()

    model = Model()
    context = cipy.Context()

    assert context.fabricate(model, "inputs", {"foo": 1}) is not None
    assert model.inputs.foo == 1


def test_fabricate_can_use_reference() -> None:
    class Model(BaseModel):
        class _Inputs(Inputs):
            foo: int = cipy.context("example.foo")
            bar: str = ""

        inputs: _Inputs = cipy.required()

    model = Model()
    context = cipy.Context(example={"foo": 5})

    assert context.fabricate(model, "inputs") is not None
    assert model.inputs.foo == 5


def test_fabricate_reference_env(monkeypatch: pytest.MonkeyPatch) -> None:
    class Model(BaseModel):
        class _Inputs(Inputs):
            foo: int = cipy.context("env.FOO")
            bar: str = ""

        inputs: _Inputs = cipy.required()

    model = Model()
    context = cipy.Context()

    monkeypatch.setenv("FOO", "5")
    assert context.fabricate(model, "inputs") is not None
    assert model.inputs.foo == 5


def test_fabricate_throws_on_missing_item() -> None:
    class Model(BaseModel):
        class _Inputs(Inputs):
            foo: int = cipy.context("example.foo")
            bar: str = ""

        inputs: _Inputs = cipy.required()

    model = Model()
    context = cipy.Context(example={})

    with pytest.raises(AttributeError) as ex:
        context.fabricate(model, "inputs")
    assert "not found" in str(ex)


def test_fabricate_throws_on_null_parent() -> None:
    class Model(BaseModel):
        class _Inputs(Inputs):
            foo: int = cipy.context("example.foo")
            bar: str = ""

        inputs: _Inputs = cipy.required()

    model = Model()
    context = cipy.Context(example=None)

    with pytest.raises(AttributeError) as ex:
        context.fabricate(model, "inputs")
    assert "NULL" in str(ex)


def test_fabricate_stub() -> None:
    class Model(BaseModel, arbitrary_types_allowed=True):
        class _Inputs(Inputs):
            foo: int = Field(default=_Stub())  # type: ignore[assignment]
            bar: str = ""

        inputs: _Inputs = cipy.required()
        logger: Mock = Mock()

    model = Model()
    context = cipy.Context()

    assert context.fabricate(model, "inputs") is not None
    assert model.inputs.foo == 0


def test_fabricate_stub_union_with_first_type() -> None:
    class Model(BaseModel, arbitrary_types_allowed=True):
        class _Inputs(Inputs):
            foo: int | str = Field(default=_Stub())  # type: ignore[assignment]
            bar: str = ""

        inputs: _Inputs = cipy.required()
        logger: Mock = Mock()

    model = Model()
    context = cipy.Context()

    assert context.fabricate(model, "inputs") is not None
    assert model.inputs.foo == 0


def test_fabricate_stub_produces_logs() -> None:
    class Model(BaseModel, arbitrary_types_allowed=True):
        class _Inputs(Inputs):
            foo: int = Field(default=_Stub())  # type: ignore[assignment]
            bar: str = ""

        inputs: _Inputs = cipy.required()
        logger: Mock = Mock()

    model = Model()
    context = cipy.Context()

    assert context.fabricate(model, "inputs") is not None
    model.logger.warning.assert_called_once()
    model.logger.debug.assert_called_once()


def test_context_extend_does_not_pollute() -> None:
    context = cipy.Context()
    with context.extend(var=1) as ex:
        assert ex.var == 1

    assert not hasattr(context, "var")


def test_results_is_constructed_empty() -> None:
    results = Results()
    with pytest.raises(AttributeError):
        results["foo"]


def test_results_cannot_insert_empty_key() -> None:
    results = Results()
    results[""] = Results.Item()

    assert "" not in results
    with pytest.raises(AttributeError):
        results[""]


def test_results_can_insert_and_access_by_key_or_attr() -> None:
    results = Results()
    results["foo"] = Results.Item()

    assert "foo" in results
    assert hasattr(results, "foo")
    assert results.foo is results["foo"]


def test_fabricate_with_results_can_return_stub() -> None:
    class Model(BaseModel, arbitrary_types_allowed=True):
        class _Inputs(Inputs):
            foo: int = cipy.context("steps.foo.outputs.value")

        inputs: _Inputs = cipy.required()
        logger: Mock = Mock()

    model = Model()
    context = cipy.Context(steps=Results())
    context.steps["foo"] = Results.Item()

    call = functools.partial(cipy.Context.__call__, context)
    with patch.object(cipy.Context, "__call__", wraps=call) as mock:
        assert context.fabricate(model, "inputs") is not None
        mock.assert_called_once()