test: cover cipy.context

tests/context_test.py

@@ -0,0 +1,274 @@
+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()