-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtest_projection.py
More file actions
170 lines (122 loc) · 5.85 KB
/
Copy pathtest_projection.py
File metadata and controls
170 lines (122 loc) · 5.85 KB
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
"""Projection strategy unit coverage: FieldNameMatching + ExplicitMapping."""
import pytest
from pydantic import Field
from openarmature.graph import (
ExplicitMapping,
FieldNameMatching,
MappingReferencesUndeclaredField,
State,
)
class Parent(State):
shared: str = "parent-shared"
parent_only: int = 0
class ChildOverlap(State):
shared: str = "child-shared"
child_only: list[str] = Field(default_factory=list)
class ChildNoOverlap(State):
completely_different: str = "x"
# ===== FieldNameMatching =====
def test_project_in_returns_subgraph_defaults() -> None:
"""Per spec v0.2.0 §2: default projection-in is no projection — subgraph
starts from its own field defaults regardless of parent state."""
proj = FieldNameMatching[Parent, ChildOverlap]()
sub = proj.project_in(Parent(shared="ignored"), ChildOverlap)
assert isinstance(sub, ChildOverlap)
assert sub.shared == "child-shared"
def test_project_out_returns_only_shared_field_names() -> None:
proj = FieldNameMatching[Parent, ChildOverlap]()
sub_final = ChildOverlap(shared="child-final", child_only=["a"])
out = proj.project_out(sub_final, Parent(), ChildOverlap)
# Only `shared` is in both schemas; `child_only` is dropped.
assert dict(out) == {"shared": "child-final"}
def test_project_out_is_empty_when_no_overlap() -> None:
proj = FieldNameMatching[Parent, ChildNoOverlap]()
sub_final = ChildNoOverlap(completely_different="y")
out = proj.project_out(sub_final, Parent(), ChildNoOverlap)
assert dict(out) == {}
def test_field_name_matching_has_no_validate_method() -> None:
"""`validate` is an optional duck-typed compile hook — strategies with
nothing declarative to check (FieldNameMatching, custom imperative
projections) simply omit it. `compile()` skips them via `getattr`."""
assert not hasattr(FieldNameMatching[Parent, ChildOverlap](), "validate")
# ===== ExplicitMapping =====
class ParentEM(State):
a: int = 5
b: int = 7
captured: int = -1
note: str = "outer"
class ChildEM(State):
input: int = 3
result: int = 0
note: str = "child-default"
def test_explicit_mapping_inputs_copies_named_parent_fields() -> None:
"""`inputs: {input: a}` overlays parent.a onto subgraph.input; other
subgraph fields receive their schema-declared defaults (no field-name
fallback)."""
proj = ExplicitMapping[ParentEM, ChildEM](inputs={"input": "a"})
sub = proj.project_in(ParentEM(a=42, note="ignored-no-fallback"), ChildEM)
assert sub.input == 42
# `note` is declared on both schemas but not in `inputs` — must NOT be
# filled by name matching. It uses the subgraph's schema default.
assert sub.note == "child-default"
assert sub.result == 0
def test_explicit_mapping_outputs_projects_only_named_pairs() -> None:
"""`outputs: {captured: input, ...}` projects only the listed
parent←subgraph pairs; other subgraph fields are discarded (replacement
of field-name matching)."""
proj = ExplicitMapping[ParentEM, ChildEM](outputs={"captured": "input", "a": "result"})
sub_final = ChildEM(input=42, result=99, note="written-but-discarded")
out = proj.project_out(sub_final, ParentEM(), ChildEM)
assert dict(out) == {"captured": 42, "a": 99}
def test_explicit_mapping_outputs_absent_falls_back_to_field_name_matching() -> None:
"""`outputs=None` (absent) falls back to spec default field-name matching;
`outputs={}` (present, empty) projects nothing."""
sub_final = ChildEM(input=1, result=2, note="from-child")
fallback = ExplicitMapping[ParentEM, ChildEM](inputs={"input": "a"}).project_out(
sub_final, ParentEM(), ChildEM
)
# Only `note` is shared by name with the parent; `input` and `result` are not.
assert dict(fallback) == {"note": "from-child"}
explicit_empty = ExplicitMapping[ParentEM, ChildEM](inputs={"input": "a"}, outputs={}).project_out(
sub_final, ParentEM(), ChildEM
)
assert dict(explicit_empty) == {}
def test_explicit_mapping_no_args_behaves_like_default_strategies_per_direction() -> None:
"""Both `inputs` and `outputs` absent: project_in returns schema defaults;
project_out falls back to field-name matching."""
proj = ExplicitMapping[ParentEM, ChildEM]()
sub_in = proj.project_in(ParentEM(a=1, b=2), ChildEM)
assert sub_in == ChildEM()
sub_final = ChildEM(input=10, result=20, note="from-child")
out = proj.project_out(sub_final, ParentEM(), ChildEM)
# Only `note` is shared by name.
assert dict(out) == {"note": "from-child"}
@pytest.mark.parametrize(
("inputs", "outputs", "expected_direction", "expected_side", "expected_field"),
[
({"missing_sub": "a"}, None, "inputs", "subgraph", "missing_sub"),
({"input": "missing_parent"}, None, "inputs", "parent", "missing_parent"),
(None, {"missing_parent": "input"}, "outputs", "parent", "missing_parent"),
(None, {"a": "missing_sub"}, "outputs", "subgraph", "missing_sub"),
],
ids=["inputs-sub", "inputs-parent", "outputs-parent", "outputs-sub"],
)
def test_explicit_mapping_validate_raises_on_undeclared_field(
inputs: dict[str, str] | None,
outputs: dict[str, str] | None,
expected_direction: str,
expected_side: str,
expected_field: str,
) -> None:
proj = ExplicitMapping[ParentEM, ChildEM](inputs=inputs, outputs=outputs)
with pytest.raises(MappingReferencesUndeclaredField) as excinfo:
proj.validate(ParentEM, ChildEM)
assert excinfo.value.direction == expected_direction
assert excinfo.value.side == expected_side
assert excinfo.value.field_name == expected_field
def test_explicit_mapping_validate_passes_when_all_fields_declared() -> None:
proj = ExplicitMapping[ParentEM, ChildEM](
inputs={"input": "a"},
outputs={"captured": "result", "note": "note"},
)
proj.validate(ParentEM, ChildEM)