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
172class OrderInput(BaseModel):
    internal_order_id: StrictStr = Field(alias="order_id")
    customer_note: StrictStr | None = None
    model_config = ConfigDict(extra="forbid", populate_by_name=False)

    @model_validator(mode="before")
    @classmethod
    def preserve_legacy_mapping(cls, value):
        # v1 tries dict(value), so pairs, [] and "" can become objects.
        # Preserve successful conversions without changing failed-input errors.
        if not isinstance(value, dict):
            try:
                return dict(value)
            except (TypeError, ValueError):
                return value
        return value


def serialize_order(payload):
    return OrderInput(**payload).model_dump(by_alias=True)


def preserve_legacy_error_order(fields):
    # v1 reports declared-field errors first and unknown keys alphabetically.
    # v2 keeps unknown keys in input order; preserve the existing HTTP contract.
    known = [field for field in fields if field["code"] != "extra_forbidden"]
    extra = sorted(
        (field for field in fields if field["code"] == "extra_forbidden"),
        key=lambda field: field["path"],
    )
    return known + extra


def check_target_model():
    for payload, expected in (
        ({"order_id": "ord_123"}, {"order_id": "ord_123", "customer_note": None}),
        (
            {"order_id": "ord_123", "customer_note": "Leave at desk"},
            {"order_id": "ord_123", "customer_note": "Leave at desk"},
        ),
        (
            {"order_id": "ord_123", "customer_note": None},
            {"order_id": "ord_123", "customer_note": None},
        ),
    ):
        assert serialize_order(payload) == expected
    for payload, expected in (
        ({}, [("order_id", "missing")]),
        ({"order_id": 123}, [("order_id", "string_type")]),
        (
            {"order_id": "ord_123", "customer_note": 17},
            [("customer_note", "string_type")],
        ),
        (
            {"order_id": "ord_123", "unexpected": True},
            [("unexpected", "extra_forbidden")],
        ),
    ):
        try:
            OrderInput(**payload)
        except ValidationError as error:
            assert [
                (item["loc"][0], item["type"]) for item in error.errors()
            ] == expected
        else:
            raise AssertionError(f"Accepted invalid order: {payload}")
    print("PASS: target model validation and serialization (7 cases)")
    print("Model-only checks; use --compare-stages for HTTP migration parity.")


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--output", type=Path, default=ROOT / "evidence")
    parser.add_argument("--compare-stages", action="store_true")
    args = parser.parse_args()
    if not args.compare_stages:
        check_target_model()
        return
    args.output.mkdir(parents=True, exist_ok=True)
    reports = []
    commands = []
    for stage in ("legacy", "prepared", "target"):
        project = Path("support") / stage
        stage_commands = [
            [
                "uv",
                "sync",
                "--project",
                str(project),
                "--python",
                sys.executable,
                "--frozen",
            ],
            [
                "uv",
                "run",
                "--project",
                str(project),
                "--python",
                sys.executable,
                "--frozen",
                "python",
                "test/run_stage.py",
                stage,
                str(args.output / f"{stage}.json"),
            ],
        ]
        for command in stage_commands:
            completed = subprocess.run(
                command, cwd=ROOT, capture_output=True, text=True
            )
            label = f"{stage}-{len(commands)}"
            (args.output / f"{label}.log").write_text(
                completed.stdout + completed.stderr
            )
            commands.append(
                {
                    "command": command,
                    "exit_code": completed.returncode,
                    "log": f"{label}.log",
                }
            )
            (args.output / "commands.json").write_text(
                json.dumps(commands, indent=2, sort_keys=True) + "\n"
            )
            if completed.returncode:
                print(completed.stdout + completed.stderr, file=sys.stderr)
                raise SystemExit(completed.returncode)
            print(completed.stdout, end="")
        reports.append(json.loads((args.output / f"{stage}.json").read_text()))
    assert all(report["python"] == platform.python_version() for report in reports)
    for report in reports[1:]:
        for key in (
            "observations",
            "public_openapi",
            "direct_call",
            "fixture_sha256",
            "schema_fixture_sha256",
        ):
            assert report[key] == reports[0][key], f"Parity failed: {key}"
    assert reports[1]["lock_sha256"] == reports[2]["lock_sha256"]
    native_schema_differs = (
        reports[0]["native_model_schema"] != reports[2]["native_model_schema"]
    )
    assert native_schema_differs, "The example should expose native schema drift"
    manifest = {
        str(path.relative_to(ROOT)): digest(path)
        for path in sorted(ROOT.rglob("*"))
        if path.is_file()
        and not path.resolve().is_relative_to(args.output.resolve())
        and not any(
            part in {".venv", "__pycache__", "evidence", ".ruff_cache"}
            for part in path.relative_to(ROOT).parts
        )
    }
    summary = {
        "status": "pass",
        "scope": "synthetic in-process FastAPI TestClient execution",
        "stages": [report["stage"] for report in reports],
        "http_cases_per_stage": reports[0]["http_cases"],
        "native_model_schema_differs": native_schema_differs,
        "public_http_and_schema_parity": True,
        "commands": commands,
        "source_sha256": manifest,
    }
    (args.output / "summary.json").write_text(
        json.dumps(summary, indent=2, sort_keys=True) + "\n"
    )
    print("PASS: legacy, prepared, target HTTP/schema/serialization contracts match")
    print(
        "Native model schemas differ; the existing public schema adapter is preserved."
    )