Spaces:
Sleeping
Sleeping
File size: 4,091 Bytes
116524e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 | """Unit tests for StepProtocol and SampleResult."""
from __future__ import annotations
import pytest
from pipeline import SampleResult, StepContext, StepProtocol
# ---------------------------------------------------------------------------
# Helper objects
# ---------------------------------------------------------------------------
class ValidStep:
requires = frozenset({"a"})
provides = frozenset({"b"})
def __call__(self, ctx: StepContext) -> StepContext:
return ctx
class ValidStepWithPlainSets:
requires = {"a"} # plain set, not frozenset
provides = {"b"}
def __call__(self, ctx: StepContext) -> StepContext:
return ctx
class MissingRequires:
provides = frozenset({"b"})
def __call__(self, ctx: StepContext) -> StepContext:
return ctx
class MissingProvides:
requires = frozenset({"a"})
def __call__(self, ctx: StepContext) -> StepContext:
return ctx
class MissingCall:
requires = frozenset({"a"})
provides = frozenset({"b"})
class EmptyStep:
"""Valid step with empty requires/provides (e.g. a pure side-effect step)."""
requires = frozenset()
provides = frozenset()
def __call__(self, ctx: StepContext) -> StepContext:
return ctx
# ---------------------------------------------------------------------------
# StepProtocol
# ---------------------------------------------------------------------------
@pytest.mark.unit
class TestStepProtocol:
def test_valid_step_passes_isinstance(self):
assert isinstance(ValidStep(), StepProtocol)
def test_valid_step_with_plain_sets_passes_isinstance(self):
# AbstractSet[str] accepts both set and frozenset
assert isinstance(ValidStepWithPlainSets(), StepProtocol)
def test_empty_requires_provides_is_valid(self):
assert isinstance(EmptyStep(), StepProtocol)
def test_missing_requires_fails_isinstance(self):
assert not isinstance(MissingRequires(), StepProtocol)
def test_missing_provides_fails_isinstance(self):
assert not isinstance(MissingProvides(), StepProtocol)
def test_missing_call_fails_isinstance(self):
assert not isinstance(MissingCall(), StepProtocol)
def test_plain_object_fails_isinstance(self):
assert not isinstance(object(), StepProtocol)
def test_none_fails_isinstance(self):
assert not isinstance(None, StepProtocol)
# ---------------------------------------------------------------------------
# SampleResult
# ---------------------------------------------------------------------------
@pytest.mark.unit
class TestSampleResult:
def test_basic_construction(self):
ctx = StepContext(sample="s")
r = SampleResult(sample="s", output=ctx, error=None, failed_at=None)
assert r.sample == "s"
assert r.output is ctx
assert r.error is None
assert r.failed_at is None
def test_cause_defaults_to_none(self):
r = SampleResult(sample="s", output=None, error=None, failed_at=None)
assert r.cause is None
def test_is_mutable(self):
r = SampleResult(sample="s", output=None, error=None, failed_at=None)
exc = RuntimeError("oops")
r.error = exc
r.failed_at = "SomeStep"
assert r.error is exc
assert r.failed_at == "SomeStep"
def test_failure_result(self):
exc = RuntimeError("boom")
r = SampleResult(sample="x", output=None, error=exc, failed_at="BoomStep")
assert r.output is None
assert r.error is exc
assert r.failed_at == "BoomStep"
def test_branch_failure_has_cause(self):
inner = RuntimeError("inner")
from pipeline import BranchError
outer = BranchError([inner])
r = SampleResult(
sample="x", output=None, error=outer, failed_at="Branch", cause=inner
)
assert r.cause is inner
assert r.failed_at == "Branch"
|