Spaces:
Sleeping
Sleeping
File size: 12,460 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 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 | """Unit tests for pipeline.StepContext.
Tests cover the generic base class (sample + metadata) and the subclassing
pattern that consuming applications use to add domain fields.
"""
from __future__ import annotations
import dataclasses
from dataclasses import dataclass
from types import MappingProxyType
from typing import Any
import pytest
from pipeline import StepContext
# ---------------------------------------------------------------------------
# Test-local subclass — validates the subclassing pattern
# ---------------------------------------------------------------------------
@dataclass(frozen=True)
class DomainContext(StepContext):
"""Minimal subclass used only in these tests."""
output: Any = None
score: float = 0.0
# ---------------------------------------------------------------------------
# Base class defaults
# ---------------------------------------------------------------------------
@pytest.mark.unit
class TestStepContextDefaults:
def test_sample_defaults_to_none(self):
ctx = StepContext()
assert ctx.sample is None
def test_metadata_defaults_to_empty_mappingproxy(self):
ctx = StepContext(sample="s")
assert ctx.metadata == MappingProxyType({})
assert isinstance(ctx.metadata, MappingProxyType)
def test_only_two_fields_on_base_class(self):
field_names = {f.name for f in dataclasses.fields(StepContext)}
assert field_names == {"sample", "metadata"}
# ---------------------------------------------------------------------------
# Immutability
# ---------------------------------------------------------------------------
@pytest.mark.unit
class TestStepContextImmutability:
def test_setting_sample_raises(self):
ctx = StepContext(sample="s")
with pytest.raises(
(dataclasses.FrozenInstanceError, AttributeError, TypeError)
):
ctx.sample = "other" # type: ignore[misc]
def test_setting_metadata_raises(self):
ctx = StepContext(sample="s")
with pytest.raises(
(dataclasses.FrozenInstanceError, AttributeError, TypeError)
):
ctx.metadata = MappingProxyType({"x": 1}) # type: ignore[misc]
def test_metadata_mappingproxy_is_not_mutable(self):
ctx = StepContext(sample="s", metadata={"k": "v"})
with pytest.raises(TypeError):
ctx.metadata["k"] = "overwrite" # type: ignore[index]
# ---------------------------------------------------------------------------
# Coercion
# ---------------------------------------------------------------------------
@pytest.mark.unit
class TestStepContextCoercion:
def test_plain_dict_metadata_coerced_to_mappingproxy(self):
ctx = StepContext(sample="s", metadata={"x": 1})
assert isinstance(ctx.metadata, MappingProxyType)
assert ctx.metadata["x"] == 1
def test_existing_mappingproxy_not_double_wrapped(self):
mp = MappingProxyType({"x": 1})
ctx = StepContext(sample="s", metadata=mp)
assert ctx.metadata is mp
# ---------------------------------------------------------------------------
# replace()
# ---------------------------------------------------------------------------
@pytest.mark.unit
class TestStepContextReplace:
def test_replace_returns_new_object(self):
ctx = StepContext(sample="s")
ctx2 = ctx.replace(sample="t")
assert ctx2 is not ctx
def test_replace_does_not_mutate_original(self):
ctx = StepContext(sample="s")
ctx.replace(sample="t")
assert ctx.sample == "s"
def test_replace_updates_target_field(self):
ctx = StepContext(sample="s")
ctx2 = ctx.replace(sample="t")
assert ctx2.sample == "t"
def test_replace_preserves_other_fields(self):
ctx = StepContext(sample="s", metadata={"k": 1})
ctx2 = ctx.replace(sample="t")
assert ctx2.metadata["k"] == 1
def test_replace_metadata_immutable_pattern(self):
ctx = StepContext(sample="s", metadata={"x": 1})
ctx2 = ctx.replace(metadata=MappingProxyType({**ctx.metadata, "y": 2}))
assert ctx2.metadata["x"] == 1
assert ctx2.metadata["y"] == 2
assert "y" not in ctx.metadata # original unchanged
# ---------------------------------------------------------------------------
# Equality
# ---------------------------------------------------------------------------
@pytest.mark.unit
class TestStepContextEquality:
def test_equal_contexts(self):
ctx1 = StepContext(sample="s")
ctx2 = StepContext(sample="s")
assert ctx1 == ctx2
def test_different_sample_not_equal(self):
assert StepContext(sample="a") != StepContext(sample="b")
def test_context_not_hashable(self):
"""StepContext is frozen but NOT hashable: MappingProxyType wraps a dict,
which is unhashable, so Python cannot derive a hash for the dataclass."""
ctx = StepContext(sample="s")
with pytest.raises(TypeError):
hash(ctx)
# ---------------------------------------------------------------------------
# Subclassing pattern
# ---------------------------------------------------------------------------
@pytest.mark.unit
class TestStepContextSubclassing:
def test_subclass_has_base_fields(self):
ctx = DomainContext(sample="s")
assert ctx.sample == "s"
assert ctx.metadata == MappingProxyType({})
def test_subclass_has_domain_fields(self):
ctx = DomainContext(sample="s", output="answer", score=0.95)
assert ctx.output == "answer"
assert ctx.score == 0.95
def test_subclass_defaults(self):
ctx = DomainContext(sample="s")
assert ctx.output is None
assert ctx.score == 0.0
def test_subclass_is_frozen(self):
ctx = DomainContext(sample="s", output="x")
with pytest.raises(
(dataclasses.FrozenInstanceError, AttributeError, TypeError)
):
ctx.output = "y" # type: ignore[misc]
def test_subclass_replace_returns_same_type(self):
ctx = DomainContext(sample="s")
ctx2 = ctx.replace(output="answer")
assert isinstance(ctx2, DomainContext)
assert ctx2.output == "answer"
def test_subclass_replace_preserves_base_fields(self):
ctx = DomainContext(sample="s", metadata={"k": 1})
ctx2 = ctx.replace(output="x")
assert ctx2.sample == "s"
assert ctx2.metadata["k"] == 1
def test_subclass_replace_preserves_domain_fields(self):
ctx = DomainContext(sample="s", output="a", score=0.9)
ctx2 = ctx.replace(sample="t")
assert ctx2.output == "a"
assert ctx2.score == 0.9
def test_subclass_metadata_coercion(self):
ctx = DomainContext(sample="s", metadata={"x": 1})
assert isinstance(ctx.metadata, MappingProxyType)
def test_subclass_isinstance_of_step_context(self):
ctx = DomainContext(sample="s")
assert isinstance(ctx, StepContext)
def test_subclass_equality(self):
a = DomainContext(sample="s", output="x")
b = DomainContext(sample="s", output="x")
assert a == b
def test_subclass_inequality_on_domain_field(self):
a = DomainContext(sample="s", output="x")
b = DomainContext(sample="s", output="y")
assert a != b
# ---------------------------------------------------------------------------
# Multi-level subclassing (SubSubContext)
# ---------------------------------------------------------------------------
@dataclass(frozen=True)
class ExtendedContext(DomainContext):
"""Two-level subclass: StepContext → DomainContext → ExtendedContext."""
label: str = ""
@pytest.mark.unit
class TestMultiLevelSubclassing:
def test_has_all_ancestor_fields(self):
ctx = ExtendedContext(sample="s", output="x", score=0.5, label="L")
assert ctx.sample == "s"
assert ctx.metadata == MappingProxyType({})
assert ctx.output == "x"
assert ctx.score == 0.5
assert ctx.label == "L"
def test_defaults_from_all_levels(self):
ctx = ExtendedContext(sample="s")
assert ctx.output is None # from DomainContext
assert ctx.score == 0.0 # from DomainContext
assert ctx.label == "" # from ExtendedContext
def test_replace_returns_correct_type(self):
ctx = ExtendedContext(sample="s")
ctx2 = ctx.replace(label="new")
assert isinstance(ctx2, ExtendedContext)
assert ctx2.label == "new"
def test_replace_preserves_all_levels(self):
ctx = ExtendedContext(sample="s", output="x", score=0.9, label="L")
ctx2 = ctx.replace(sample="t")
assert ctx2.output == "x"
assert ctx2.score == 0.9
assert ctx2.label == "L"
def test_isinstance_chain(self):
ctx = ExtendedContext(sample="s")
assert isinstance(ctx, StepContext)
assert isinstance(ctx, DomainContext)
assert isinstance(ctx, ExtendedContext)
def test_is_frozen(self):
ctx = ExtendedContext(sample="s", label="L")
with pytest.raises(
(dataclasses.FrozenInstanceError, AttributeError, TypeError)
):
ctx.label = "new" # type: ignore[misc]
def test_metadata_coercion(self):
ctx = ExtendedContext(sample="s", metadata={"k": 1})
assert isinstance(ctx.metadata, MappingProxyType)
def test_field_count(self):
field_names = {f.name for f in dataclasses.fields(ExtendedContext)}
assert field_names == {"sample", "metadata", "output", "score", "label"}
# ---------------------------------------------------------------------------
# Multi-field replace
# ---------------------------------------------------------------------------
@pytest.mark.unit
class TestMultiFieldReplace:
def test_replace_multiple_fields_at_once(self):
ctx = DomainContext(sample="s", output="a", score=0.1)
ctx2 = ctx.replace(sample="t", output="b", score=0.9)
assert ctx2.sample == "t"
assert ctx2.output == "b"
assert ctx2.score == 0.9
def test_replace_base_and_domain_fields_together(self):
ctx = DomainContext(sample="s", metadata={"k": 1}, output="a")
ctx2 = ctx.replace(
sample="t",
metadata=MappingProxyType({"k": 2}),
output="b",
)
assert ctx2.sample == "t"
assert ctx2.metadata["k"] == 2
assert ctx2.output == "b"
def test_replace_all_fields_on_multi_level_subclass(self):
ctx = ExtendedContext(sample="s", output="a", score=0.1, label="L")
ctx2 = ctx.replace(sample="t", output="b", score=0.9, label="M")
assert isinstance(ctx2, ExtendedContext)
assert ctx2.sample == "t"
assert ctx2.output == "b"
assert ctx2.score == 0.9
assert ctx2.label == "M"
def test_original_unchanged_after_multi_field_replace(self):
ctx = DomainContext(sample="s", output="a", score=0.1)
ctx.replace(sample="t", output="b", score=0.9)
assert ctx.sample == "s"
assert ctx.output == "a"
assert ctx.score == 0.1
# ---------------------------------------------------------------------------
# Cross-type equality
# ---------------------------------------------------------------------------
@dataclass(frozen=True)
class OtherContext(StepContext):
"""A different subclass with the same field name as DomainContext."""
output: Any = None
@pytest.mark.unit
class TestCrossTypeEquality:
def test_different_subclass_types_not_equal(self):
a = DomainContext(sample="s", output="x")
b = OtherContext(sample="s", output="x")
assert a != b
def test_base_not_equal_to_subclass(self):
base = StepContext(sample="s")
sub = DomainContext(sample="s")
assert base != sub
def test_subclass_not_equal_to_sub_subclass(self):
parent = DomainContext(sample="s")
child = ExtendedContext(sample="s")
assert parent != child
|