From 19e5f1eeab4f9a93974672ce9c246829c4475397 Mon Sep 17 00:00:00 2001
From: "tyrion.lh" <tyrion.lh@bytedance.com>
Date: Tue, 14 Jul 2026 13:55:08 -0700
Subject: [PATCH] test: add schema preprocessing coverage and CI

---
 .github/workflows/tests.yml      |  42 +++
 README.md                        |   8 +
 tests/test_preprocess_schemas.py | 539 +++++++++++++++++++++++++++++++
 3 files changed, 589 insertions(+)
 create mode 100644 .github/workflows/tests.yml
 create mode 100644 tests/test_preprocess_schemas.py

diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml
new file mode 100644
index 0000000..2e6a3a2
--- /dev/null
+++ b/.github/workflows/tests.yml
@@ -0,0 +1,42 @@
+# Copyright 2026 UCP Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+name: Tests
+
+on:
+  push:
+    branches: [main]
+  pull_request:
+    branches: [main]
+
+permissions:
+  contents: read
+
+jobs:
+  schema-preprocessing:
+    name: Schema preprocessing (Python ${{ matrix.python-version }})
+    runs-on: ubuntu-latest
+    strategy:
+      fail-fast: false
+      matrix:
+        python-version: ["3.10", "3.11", "3.12"]
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v5
+      - name: Set up Python
+        uses: actions/setup-python@v6
+        with:
+          python-version: ${{ matrix.python-version }}
+      - name: Run preprocessing tests
+        run: python -m unittest discover -s tests -p "test_*.py"
diff --git a/README.md b/README.md
index 41290a1..8438438 100644
--- a/README.md
+++ b/README.md
@@ -140,6 +140,14 @@ If no version is specified, the `main` branch of the
 
 The generated code is automatically formatted using `ruff`.
 
+### Running Tests
+
+Run the schema preprocessing test suite with:
+
+```bash
+python -m unittest discover -s tests -p "test_*.py"
+```
+
 ## Contributing
 
 We welcome community contributions. See our
diff --git a/tests/test_preprocess_schemas.py b/tests/test_preprocess_schemas.py
new file mode 100644
index 0000000..92fda86
--- /dev/null
+++ b/tests/test_preprocess_schemas.py
@@ -0,0 +1,539 @@
+# Copyright 2026 UCP Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Tests for the schema preprocessing pipeline."""
+
+import contextlib
+import copy
+import io
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+from unittest import mock
+
+import preprocess_schemas
+
+
+class SchemaNormalizationTest(unittest.TestCase):
+    """Tests schema flattening and reference normalization."""
+
+    def test_resolve_local_ref_supports_objects_and_arrays(self) -> None:
+        """Local JSON pointers resolve object keys and array indexes."""
+        schema = {"$defs": {"choices": [{"const": "first"}]}}
+
+        resolved = preprocess_schemas.resolve_local_ref(
+            "#/$defs/choices/0", schema
+        )
+
+        self.assertEqual(resolved, {"const": "first"})
+        self.assertIsNone(
+            preprocess_schemas.resolve_local_ref("#/$defs/choices/1", schema)
+        )
+        self.assertIsNone(
+            preprocess_schemas.resolve_local_ref("other.json", schema)
+        )
+
+    def test_preprocess_flattens_and_distributes_properties(self) -> None:
+        """Flattened base fields are distributed to polymorphic branches."""
+        schema = {
+            "$defs": {
+                "base": {
+                    "type": "object",
+                    "properties": {"id": {"type": "string"}},
+                    "required": ["id"],
+                }
+            },
+            "allOf": [{"$ref": "#/$defs/base"}],
+            "oneOf": [
+                {
+                    "properties": {"kind": {"const": "physical"}},
+                    "required": ["kind"],
+                }
+            ],
+        }
+
+        preprocess_schemas.preprocess_full_schema(schema)
+
+        self.assertNotIn("allOf", schema)
+        self.assertEqual(schema["required"], ["id"])
+        branch = schema["oneOf"][0]
+        self.assertEqual(set(branch["properties"]), {"id", "kind"})
+        self.assertEqual(set(branch["required"]), {"id", "kind"})
+        self.assertEqual(branch["type"], "object")
+
+    def test_preprocess_inlines_entity_fields(self) -> None:
+        """The shared entity definition is inlined without its metadata."""
+        entity = {
+            "title": "Entity",
+            "description": "Shared entity fields.",
+            "type": "object",
+            "properties": {"id": {"type": "string"}},
+            "required": ["id"],
+        }
+        schema = {
+            "allOf": [
+                {"$ref": "ucp.json#/$defs/entity"},
+                {
+                    "type": "object",
+                    "properties": {"value": {"type": "integer"}},
+                    "required": ["value"],
+                },
+            ]
+        }
+
+        preprocess_schemas.preprocess_full_schema(schema, entity)
+
+        self.assertEqual(set(schema["properties"]), {"id", "value"})
+        self.assertEqual(set(schema["required"]), {"id", "value"})
+        self.assertNotIn("title", schema)
+        self.assertNotIn("description", schema)
+
+    def test_flatten_dotted_defs_rewrites_local_refs(self) -> None:
+        """Dotted definition names and local references stay aligned."""
+        schema = {
+            "$defs": {
+                "checkout": {"type": "string"},
+                "dev.ucp.shopping.checkout": {"type": "object"},
+            },
+            "properties": {
+                "checkout": {"$ref": "#/$defs/dev.ucp.shopping.checkout"}
+            },
+        }
+
+        rename_map = preprocess_schemas.flatten_dotted_defs(schema)
+
+        self.assertEqual(
+            rename_map,
+            {"dev.ucp.shopping.checkout": "dev_ucp_shopping_checkout"},
+        )
+        self.assertIn("dev_ucp_shopping_checkout", schema["$defs"])
+        self.assertEqual(
+            schema["properties"]["checkout"]["$ref"],
+            "#/$defs/dev_ucp_shopping_checkout",
+        )
+
+    def test_rewrite_external_defs_refs_uses_target_rename_map(self) -> None:
+        """External references follow renames made in the target schema."""
+        with tempfile.TemporaryDirectory() as temp_dir:
+            root = Path(temp_dir)
+            source_path = root / "source.json"
+            target_path = root / "target.json"
+            schema = {
+                "properties": {
+                    "checkout": {
+                        "$ref": ("target.json#/$defs/dev.ucp.shopping.checkout")
+                    }
+                }
+            }
+
+            preprocess_schemas._rewrite_external_defs_refs(
+                source_path,
+                schema,
+                {
+                    str(target_path.resolve()): {
+                        "dev.ucp.shopping.checkout": "checkout"
+                    }
+                },
+            )
+
+        self.assertEqual(
+            schema["properties"]["checkout"]["$ref"],
+            "target.json#/$defs/checkout",
+        )
+
+
+class RequestMetadataTest(unittest.TestCase):
+    """Tests operation-specific request metadata rules."""
+
+    def test_get_required_ops_collects_all_declared_operations(self) -> None:
+        """String and mapping markers contribute their operations."""
+        schema = {
+            "properties": {
+                "id": {"ucp_request": "omit"},
+                "payment": {
+                    "ucp_request": {
+                        "complete": "required",
+                        "update": "optional",
+                    }
+                },
+                "plain": {"type": "string"},
+            }
+        }
+
+        self.assertEqual(
+            preprocess_schemas.get_required_ops(schema),
+            {"create", "update", "complete"},
+        )
+
+    def test_eval_prop_inclusion_applies_operation_overrides(self) -> None:
+        """Operation markers override base required and inclusion rules."""
+        cases = [
+            ("default-required", {}, "create", ["field"], (True, True)),
+            (
+                "simple-optional",
+                {"ucp_request": "optional"},
+                "create",
+                ["field"],
+                (True, False),
+            ),
+            (
+                "simple-omit",
+                {"ucp_request": "omit"},
+                "create",
+                [],
+                (False, False),
+            ),
+            (
+                "operation-required",
+                {"ucp_request": {"create": "required"}},
+                "create",
+                [],
+                (True, True),
+            ),
+            (
+                "operation-omit",
+                {"ucp_request": {"create": "omit"}},
+                "create",
+                ["field"],
+                (False, True),
+            ),
+            (
+                "undeclared-operation",
+                {"ucp_request": {"update": "required"}},
+                "create",
+                [],
+                (False, False),
+            ),
+        ]
+
+        for name, data, operation, required, expected in cases:
+            with self.subTest(name=name):
+                actual = preprocess_schemas.eval_prop_inclusion(
+                    "field", data, operation, required
+                )
+                self.assertEqual(actual, expected)
+
+
+class VariantGenerationTest(unittest.TestCase):
+    """Tests request variant construction and output."""
+
+    def test_object_variant_filters_fields_and_rewrites_refs(self) -> None:
+        """Object variants filter fields and target child variants."""
+        schema = {
+            "$id": "https://ucp.dev/schemas/checkout.json",
+            "title": "Checkout",
+            "type": "object",
+            "properties": {
+                "id": {
+                    "type": "string",
+                    "ucp_request": {
+                        "create": "omit",
+                        "update": "required",
+                    },
+                },
+                "currency": {
+                    "type": "string",
+                    "ucp_request": "required",
+                },
+                "note": {
+                    "type": "string",
+                    "ucp_request": "optional",
+                },
+                "server_only": {
+                    "type": "string",
+                    "ucp_request": "omit",
+                },
+                "child": {
+                    "$ref": "child.json",
+                    "ucp_request": "required",
+                },
+            },
+            "required": ["id", "note"],
+        }
+        original = copy.deepcopy(schema)
+        file_path = Path("/schemas/checkout.json")
+        child_path = str((file_path.parent / "child.json").resolve())
+
+        variant = preprocess_schemas._create_single_variant(
+            schema,
+            "create",
+            "checkout",
+            file_path,
+            {child_path: {"create"}},
+        )
+
+        self.assertEqual(schema, original)
+        self.assertEqual(variant["title"], "Checkout Create Request")
+        self.assertEqual(
+            variant["$id"],
+            "https://ucp.dev/schemas/checkout_create_request.json",
+        )
+        self.assertEqual(
+            set(variant["properties"]), {"currency", "note", "child"}
+        )
+        self.assertEqual(set(variant["required"]), {"currency", "child"})
+        self.assertEqual(
+            variant["properties"]["child"]["$ref"],
+            "child_create_request.json",
+        )
+        for data in variant["properties"].values():
+            self.assertNotIn("ucp_request", data)
+
+    def test_array_variant_preserves_root_and_filters_nested_objects(
+        self,
+    ) -> None:
+        """Array roots stay arrays while nested request fields are filtered."""
+        schema = {
+            "$id": "https://ucp.dev/schemas/totals.json",
+            "title": "Totals",
+            "type": "array",
+            "items": {
+                "allOf": [
+                    {
+                        "type": "object",
+                        "properties": {
+                            "amount": {"type": "integer"},
+                            "label": {
+                                "type": "string",
+                                "ucp_request": {"create": "required"},
+                            },
+                            "lines": {
+                                "type": "array",
+                                "ucp_request": {"create": "omit"},
+                            },
+                        },
+                        "required": ["amount"],
+                    }
+                ]
+            },
+        }
+
+        variant = preprocess_schemas._create_single_variant(
+            schema,
+            "create",
+            "totals",
+            Path("/schemas/totals.json"),
+            {},
+        )
+
+        self.assertEqual(variant["type"], "array")
+        self.assertNotIn("properties", variant)
+        self.assertNotIn("required", variant)
+        item_schema = variant["items"]["allOf"][0]
+        self.assertEqual(set(item_schema["properties"]), {"amount", "label"})
+        self.assertEqual(set(item_schema["required"]), {"amount", "label"})
+        self.assertEqual(variant["title"], "Totals Create Request")
+
+    def test_generate_variants_writes_operation_specific_files(self) -> None:
+        """Variant generation writes one filtered file per operation."""
+        schema = {
+            "title": "Product",
+            "type": "object",
+            "properties": {
+                "id": {
+                    "type": "string",
+                    "ucp_request": {
+                        "create": "omit",
+                        "update": "required",
+                    },
+                }
+            },
+            "required": ["id"],
+        }
+
+        with tempfile.TemporaryDirectory() as temp_dir:
+            source_path = Path(temp_dir) / "product.json"
+            with contextlib.redirect_stdout(io.StringIO()):
+                preprocess_schemas.generate_variants(
+                    source_path,
+                    schema,
+                    {"create", "update"},
+                    {},
+                )
+
+            create_variant = preprocess_schemas.load_json(
+                Path(temp_dir) / "product_create_request.json"
+            )
+            update_variant = preprocess_schemas.load_json(
+                Path(temp_dir) / "product_update_request.json"
+            )
+
+        self.assertEqual(create_variant["properties"], {})
+        self.assertEqual(create_variant["required"], [])
+        self.assertEqual(set(update_variant["properties"]), {"id"})
+        self.assertEqual(update_variant["required"], ["id"])
+
+
+class PipelineDependencyTest(unittest.TestCase):
+    """Tests metadata normalization and transitive variant dependencies."""
+
+    def test_normalize_metadata_schemas_sets_root_union_and_ucp_refs(
+        self,
+    ) -> None:
+        """Metadata schemas expose the root union and normalized references."""
+        target_dir = Path("/schemas")
+        ucp_path = str((target_dir / "ucp.json").resolve())
+        checkout_path = str((target_dir / "checkout.json").resolve())
+        request_path = str(
+            (target_dir / "checkout_create_request.json").resolve()
+        )
+        schemas = {
+            ucp_path: {"$defs": {}},
+            checkout_path: {
+                "properties": {
+                    "ucp": {"$ref": "ucp.json#/$defs/response_schema"}
+                }
+            },
+            request_path: {
+                "properties": {
+                    "ucp": {"$ref": "ucp.json#/$defs/request_schema"}
+                }
+            },
+        }
+
+        preprocess_schemas.normalize_metadata_schemas(schemas, target_dir)
+
+        self.assertEqual(
+            schemas[ucp_path]["oneOf"],
+            [
+                {"$ref": "#/$defs/platform_schema"},
+                {"$ref": "#/$defs/business_schema"},
+                {"$ref": "#/$defs/response_checkout_schema"},
+                {"$ref": "#/$defs/response_order_schema"},
+                {"$ref": "#/$defs/response_cart_schema"},
+            ],
+        )
+        self.assertEqual(
+            schemas[checkout_path]["properties"]["ucp"]["$ref"],
+            "ucp.json",
+        )
+        self.assertEqual(
+            schemas[request_path]["properties"]["ucp"]["$ref"],
+            "ucp.json#/$defs/request_schema",
+        )
+
+    def test_variant_needs_propagate_transitively_and_respect_omit(
+        self,
+    ) -> None:
+        """Variant dependencies propagate only through included properties."""
+        parent_path = "/schemas/parent.json"
+        child_path = "/schemas/child.json"
+        grandchild_path = "/schemas/grandchild.json"
+        schemas = {
+            parent_path: {
+                "properties": {
+                    "child": {
+                        "$ref": "child.json",
+                        "ucp_request": {
+                            "create": "required",
+                            "update": "omit",
+                        },
+                    }
+                }
+            },
+            child_path: {
+                "properties": {"grandchild": {"$ref": "grandchild.json"}}
+            },
+            grandchild_path: {"properties": {}},
+        }
+        schema_refs = {
+            parent_path: [("child", child_path)],
+            child_path: [("grandchild", grandchild_path)],
+            grandchild_path: [],
+        }
+        variant_needs = {parent_path: {"create", "update"}}
+
+        preprocess_schemas.propagate_needs_transitive(
+            variant_needs, schema_refs, schemas
+        )
+
+        self.assertEqual(variant_needs[child_path], {"create"})
+        self.assertEqual(variant_needs[grandchild_path], {"create"})
+
+    def test_main_preprocesses_schema_tree_end_to_end(self) -> None:
+        """The full pipeline normalizes schemas and writes linked variants."""
+        with tempfile.TemporaryDirectory() as temp_dir:
+            root = Path(temp_dir)
+            preprocess_schemas.save_json(
+                {
+                    "$defs": {
+                        "entity": {
+                            "type": "object",
+                            "properties": {"id": {"type": "string"}},
+                            "required": ["id"],
+                        }
+                    }
+                },
+                root / "ucp.json",
+            )
+            preprocess_schemas.save_json(
+                {
+                    "$id": "https://ucp.dev/schemas/child.json",
+                    "title": "Child",
+                    "type": "object",
+                    "properties": {
+                        "value": {
+                            "type": "string",
+                            "ucp_request": {"create": "required"},
+                        }
+                    },
+                },
+                root / "child.json",
+            )
+            preprocess_schemas.save_json(
+                {
+                    "$id": "https://ucp.dev/schemas/parent.json",
+                    "title": "Parent",
+                    "allOf": [{"$ref": "ucp.json#/$defs/entity"}],
+                    "properties": {
+                        "child": {
+                            "$ref": "child.json",
+                            "ucp_request": {"create": "required"},
+                        }
+                    },
+                },
+                root / "parent.json",
+            )
+
+            with (
+                mock.patch.object(
+                    sys,
+                    "argv",
+                    ["preprocess_schemas.py", str(root)],
+                ),
+                contextlib.redirect_stdout(io.StringIO()),
+            ):
+                preprocess_schemas.main()
+
+            parent = preprocess_schemas.load_json(root / "parent.json")
+            parent_variant = preprocess_schemas.load_json(
+                root / "parent_create_request.json"
+            )
+            child_variant = preprocess_schemas.load_json(
+                root / "child_create_request.json"
+            )
+
+        self.assertNotIn("allOf", parent)
+        self.assertEqual(set(parent["properties"]), {"id", "child"})
+        self.assertEqual(
+            parent_variant["properties"]["child"]["$ref"],
+            "child_create_request.json",
+        )
+        self.assertEqual(set(parent_variant["required"]), {"id", "child"})
+        self.assertEqual(child_variant["required"], ["value"])
+
+
+if __name__ == "__main__":
+    unittest.main()