diff --git a/postprocess_models.py b/postprocess_models.py index d6aa55b..6541ad8 100644 --- a/postprocess_models.py +++ b/postprocess_models.py @@ -14,16 +14,23 @@ """Post-generation fixes for constraints datamodel-code-generator ignores. -Seven constraint families are handled: - -* ``minProperties`` on an object schema WITH declared properties is dropped by - the generator (issue #49): every field is optional, so an empty instance - passes validation in violation of the schema. (``minProperties`` on a - free-form object property is already handled natively — the generator maps it - to ``Field(min_length=...)`` on the dict field.) The script scans the - preprocessed schemas for root-level ``minProperties`` constraints and injects - a ``model_validator(mode="after")`` into the matching generated classes. - JSON Schema counts the keys present on the object, so the validator counts +Eight constraint families are handled: + +* ``minProperties`` / ``maxProperties`` on an object schema WITH declared + properties are dropped by the generator: every field is optional, so an + empty instance (or, for ``maxProperties``, an over-full one) passes + validation in violation of the schema. ``minProperties`` support (issue + #49, PR #55) never grew a ``maxProperties`` counterpart, so + ``location_serves.json``'s ``maxProperties: 1`` ("the Platform MUST + supply exactly one target form") went unenforced even though its sibling + ``minProperties: 1`` on the same schema was already caught. (Either bound + on a free-form object property is already handled natively — the + generator maps it to ``Field(min_length=..., max_length=...)`` on the + dict field.) The script scans the preprocessed schemas for root-level + ``minProperties``/``maxProperties`` constraints and injects a + ``model_validator(mode="after")`` into the matching generated classes, + one validator per bound so both can coexist on the same class. JSON + Schema counts the keys present on the object, so the validator counts provided fields (``model_fields_set``) unioned with extra keys (``model_extra``) — an explicit null is a present key, and unknown keys on ``extra="allow"`` models count too. @@ -131,6 +138,22 @@ def {marker}(self): return self ''' +_MAX_MARKER = "_enforce_max_properties" + +_MAX_VALIDATOR_TEMPLATE = ''' + @model_validator(mode="after") + def {marker}(self): + """JSON Schema maxProperties: allow at most {maximum} + provided {properties_noun}.""" + provided = self.model_fields_set | set(self.model_extra or {{}}) + if len(provided) > {maximum}: + raise ValueError( + "At most {maximum} {properties_noun} may be provided " + "(schema maxProperties={maximum})" + ) + return self +''' + _PROPNAMES_MARKER = "_enforce_property_names" _PROPNAMES_VALIDATOR_TEMPLATE = ''' @@ -253,6 +276,41 @@ def find_root_min_properties(schema_dir): return found +def find_root_max_properties(schema_dir): + """Map schema title -> maxProperties for root-level object constraints. + + Symmetric twin of find_root_min_properties (see #49/#55, which added + minProperties support but never a maxProperties counterpart): + maxProperties on an object schema WITH declared properties is dropped by + the generator the same way minProperties is, so + location_serves.json's maxProperties: 1 ("the Platform MUST supply + exactly one target form") was silently unenforced. As with the min + side, maxProperties on a free-form object property (no named + properties) is already handled natively by the generator + (Field(max_length=...) on the dict field), so it is out of scope here. + """ + found = {} + for path in sorted(Path(schema_dir).rglob("*.json")): + try: + schema = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + continue + if not isinstance(schema, dict): + continue + maximum = schema.get("maxProperties") + if not isinstance(maximum, int) or not schema.get("properties"): + continue + title = schema.get("title") + if not title: + sys.stderr.write( + f" ! {path}: root maxProperties but no title; " + "cannot map to a class\n" + ) + continue + found[_alias_name(title)] = maximum + return found + + def _ensure_pydantic_import(source, symbol): """Add ``symbol`` to the ``from pydantic import`` line if absent.""" if re.search( @@ -438,6 +496,35 @@ def inject_min_properties(source, class_name, minimum): return _ensure_pydantic_import(out, "model_validator") +def inject_max_properties(source, class_name, maximum): + """Inject the maxProperties validator at the end of ``class_name``. + + Symmetric twin of inject_min_properties; both validators can be + injected into the same class (location_serves.json declares both + minProperties: 1 and maxProperties: 1), each guarded by its own marker + so neither injection clobbers the other or re-runs on a second pass. + """ + if f"def {_MAX_MARKER}(" in source: + return source + class_re = re.compile(rf"^class {re.escape(class_name)}\(", re.M) + match = class_re.search(source) + if not match: + return source + # The class body ends at the next top-level statement or EOF. + tail = re.compile(r"^\S", re.M) + end_match = tail.search(source, match.end()) + end = end_match.start() if end_match else len(source) + method = _MAX_VALIDATOR_TEMPLATE.format( + marker=_MAX_MARKER, + maximum=maximum, + properties_noun="property" if maximum == 1 else "properties", + ) + body = source[:end].rstrip("\n") + rest = source[end:] + out = body + "\n" + method + ("\n" + rest if rest else "") + return _ensure_pydantic_import(out, "model_validator") + + def _extract_contains_groups(schema, path=None): """Collect every array ``contains`` group from a schema's root + allOf. @@ -1195,6 +1282,37 @@ def _patch_min_properties(): return patched, 0 +def _patch_max_properties(): + """Inject maxProperties validators; return (patched_count, exit_code).""" + constraints = find_root_max_properties(SCHEMA_DIR) + if not constraints: + sys.stdout.write( + "postprocess: no root-level maxProperties constraints found\n" + ) + return 0, 0 + patched = 0 + for title, maximum in sorted(constraints.items()): + hits = [] + for path in sorted(OUTPUT_DIR.rglob("*.py")): + source = path.read_text(encoding="utf-8") + if not re.search(rf"^class {re.escape(title)}\(", source, re.M): + continue + updated = inject_max_properties(source, title, maximum) + if updated != source: + path.write_text(updated, encoding="utf-8") + patched += 1 + hits.append(path) + label = ", ".join(str(h) for h in hits) or "NO GENERATED CLASS FOUND" + sys.stdout.write(f" maxProperties={maximum} on '{title}' -> {label}\n") + if not hits: + sys.stderr.write( + f" ! '{title}' has no generated class; " + "constraint not enforced\n" + ) + return patched, 1 + return patched, 0 + + def _array_contains_targets(): """Resolve ``title -> groups`` for every model needing a contains bound. @@ -1514,6 +1632,7 @@ def _patch_extra_forbid(): def main(): """Main entry point to scan schemas and patch generated models.""" patched_mp, rc_mp = _patch_min_properties() + patched_xp, rc_xp = _patch_max_properties() patched_pn, rc_pn = _patch_property_names() patched_ac, rc_ac = _patch_array_contains() patched_cr, rc_cr = _patch_conditional_required() @@ -1522,6 +1641,7 @@ def main(): patched_ef, rc_ef = _patch_extra_forbid() total = ( patched_mp + + patched_xp + patched_pn + patched_ac + patched_cr @@ -1530,7 +1650,7 @@ def main(): + patched_ef ) sys.stdout.write(f"postprocess: {total} module(s) patched\n") - return rc_mp or rc_pn or rc_ac or rc_cr or rc_cb or rc_ui or rc_ef + return rc_mp or rc_xp or rc_pn or rc_ac or rc_cr or rc_cb or rc_ui or rc_ef if __name__ == "__main__": diff --git a/src/ucp_sdk/models/schemas/common/types/location_serves.py b/src/ucp_sdk/models/schemas/common/types/location_serves.py index c5c9f4b..4b70a5a 100644 --- a/src/ucp_sdk/models/schemas/common/types/location_serves.py +++ b/src/ucp_sdk/models/schemas/common/types/location_serves.py @@ -84,3 +84,14 @@ def _enforce_min_properties(self): "At least 1 property must be provided (schema minProperties=1)" ) return self + + @model_validator(mode="after") + def _enforce_max_properties(self): + """JSON Schema maxProperties: allow at most 1 + provided property.""" + provided = self.model_fields_set | set(self.model_extra or {}) + if len(provided) > 1: + raise ValueError( + "At most 1 property may be provided (schema maxProperties=1)" + ) + return self diff --git a/src/ucp_sdk/models/schemas/common/types/location_serves_create_request.py b/src/ucp_sdk/models/schemas/common/types/location_serves_create_request.py index 2c1e4fb..89d4b13 100644 --- a/src/ucp_sdk/models/schemas/common/types/location_serves_create_request.py +++ b/src/ucp_sdk/models/schemas/common/types/location_serves_create_request.py @@ -84,3 +84,14 @@ def _enforce_min_properties(self): "At least 1 property must be provided (schema minProperties=1)" ) return self + + @model_validator(mode="after") + def _enforce_max_properties(self): + """JSON Schema maxProperties: allow at most 1 + provided property.""" + provided = self.model_fields_set | set(self.model_extra or {}) + if len(provided) > 1: + raise ValueError( + "At most 1 property may be provided (schema maxProperties=1)" + ) + return self diff --git a/src/ucp_sdk/models/schemas/common/types/location_serves_update_request.py b/src/ucp_sdk/models/schemas/common/types/location_serves_update_request.py index 99f3dd1..94f4ab5 100644 --- a/src/ucp_sdk/models/schemas/common/types/location_serves_update_request.py +++ b/src/ucp_sdk/models/schemas/common/types/location_serves_update_request.py @@ -84,3 +84,14 @@ def _enforce_min_properties(self): "At least 1 property must be provided (schema minProperties=1)" ) return self + + @model_validator(mode="after") + def _enforce_max_properties(self): + """JSON Schema maxProperties: allow at most 1 + provided property.""" + provided = self.model_fields_set | set(self.model_extra or {}) + if len(provided) > 1: + raise ValueError( + "At most 1 property may be provided (schema maxProperties=1)" + ) + return self diff --git a/tests/test_codegen_pipeline.py b/tests/test_codegen_pipeline.py index ac62ea4..f144070 100644 --- a/tests/test_codegen_pipeline.py +++ b/tests/test_codegen_pipeline.py @@ -1638,6 +1638,176 @@ def test_schema_scan_finds_root_constraints(self): self.assertEqual(found, {"Sample": 2}) +class MaxPropertiesInjectorTest(unittest.TestCase): + """maxProperties is the symmetric twin of minProperties (see #49/#55), + but only minProperties was ever scanned: find_root_min_properties reads + schema.get("minProperties") and there is no find_root_max_properties at + all, so location_serves.json's maxProperties: 1 -- "the Platform MUST + supply exactly one target form" -- is silently dropped. This mirrors + InjectorTest above one for one, for the max side. + """ + + SCHEMA = { + "title": "Sample", + "type": "object", + "maxProperties": 1, + "properties": {"a": {"type": "string"}, "b": {"type": "string"}}, + } + + MODULE = ( + "from __future__ import annotations\n" + "\n" + "from pydantic import BaseModel, ConfigDict\n" + "\n" + "\n" + "class Sample(BaseModel):\n" + ' """A sample."""\n' + "\n" + " model_config = ConfigDict(\n" + ' extra="allow",\n' + " )\n" + " a: str | None = None\n" + " b: str | None = None\n" + ) + + def test_injects_validator_with_declared_maximum(self): + out = postprocess_models.inject_max_properties(self.MODULE, "Sample", 1) + self.assertIn("model_validator", out) + self.assertIn("at most 1", out.lower()) + + @unittest.skipUnless(HAVE_SDK, "executing the module needs pydantic") + def test_injected_validator_enforces_count(self): + out = postprocess_models.inject_max_properties(self.MODULE, "Sample", 1) + namespace: dict = {} + exec(compile(out, "", "exec"), namespace) # noqa: S102 + sample_cls = namespace["Sample"] + with self.assertRaises(ValidationError): + sample_cls(a="one", b="two") + sample_cls(a="only-one") + sample_cls() + + def test_injection_is_idempotent(self): + once = postprocess_models.inject_max_properties( + self.MODULE, "Sample", 1 + ) + twice = postprocess_models.inject_max_properties(once, "Sample", 1) + self.assertEqual(once, twice) + + def test_schema_scan_finds_root_constraints(self): + with tempfile.TemporaryDirectory() as tmp: + sub = Path(tmp) / "sub" + sub.mkdir() + (sub / "sample.json").write_text(json.dumps(self.SCHEMA)) + (sub / "plain.json").write_text( + json.dumps( + {"title": "Plain", "type": "object", "properties": {}} + ) + ) + found = postprocess_models.find_root_max_properties(Path(tmp)) + self.assertEqual(found, {"Sample": 1}) + + def test_schema_scan_ignores_object_without_declared_properties(self): + # Mirrors find_root_min_properties: maxProperties on a free-form + # object property (no named properties) is already handled natively + # by the generator (Field(max_length=...) on the dict field), so a + # bare maxProperties with no properties is out of scope here. + schema = { + "title": "OpenMap", + "type": "object", + "maxProperties": 3, + } + with tempfile.TemporaryDirectory() as tmp: + Path(tmp, "open_map.json").write_text(json.dumps(schema)) + found = postprocess_models.find_root_max_properties(Path(tmp)) + self.assertEqual(found, {}) + + def test_both_bounds_coexist_on_the_same_class(self): + """location_serves.json declares both minProperties: 1 AND + maxProperties: 1 on the same object; both validators must be + injectable into the same class without clobbering each other.""" + module = postprocess_models.inject_min_properties( + self.MODULE, "Sample", 1 + ) + module = postprocess_models.inject_max_properties(module, "Sample", 1) + self.assertIn("_enforce_min_properties", module) + self.assertIn("_enforce_max_properties", module) + if HAVE_SDK: + namespace: dict = {} + exec(compile(module, "", "exec"), namespace) # noqa: S102 + sample_cls = namespace["Sample"] + with self.assertRaises(ValidationError): + sample_cls() + with self.assertRaises(ValidationError): + sample_cls(a="one", b="two") + sample_cls(a="only-one") + + +@unittest.skipUnless( + HAVE_SDK, "requires the installed package (pip install -e .)" +) +class LocationServesMaxPropertiesSemanticTest(unittest.TestCase): + """location_serves.json: "The Platform MUST supply exactly one target + form" -- minProperties: 1 AND maxProperties: 1 together. Only the + minimum was ever enforced (see MaxPropertiesInjectorTest above), so a + map naming both point and address currently validates in violation of + the schema. + """ + + def _location_serves(self): + from ucp_sdk.models.schemas.common.types.location_serves import ( + LocationServes, + ) + + return LocationServes + + def _geo(self): + from ucp_sdk.models.schemas.common.types.geo import Geo + + return Geo + + def _address(self): + from ucp_sdk.models.schemas.common.types.location_serves import ( + Address, + ) + + return Address + + def test_both_point_and_address_rejected(self): + with self.assertRaises(ValidationError): + self._location_serves()( + point=self._geo()(latitude=1.0, longitude=2.0), + address=self._address()(address_country="US"), + ) + + def test_point_only_accepted(self): + location = self._location_serves()( + point=self._geo()(latitude=1.0, longitude=2.0) + ) + self.assertIsNotNone(location.point) + + def test_address_only_accepted(self): + location = self._location_serves()( + address=self._address()(address_country="US") + ) + self.assertIsNotNone(location.address) + + def test_empty_still_rejected_by_the_existing_minimum(self): + # Unaffected by this fix; confirms minProperties: 1 still holds. + with self.assertRaises(ValidationError): + self._location_serves()() + + def test_extension_key_alongside_point_rejected(self): + # extra="allow": an extension form key still counts toward the + # maxProperties=1 total per JSON Schema's key-counting semantics. + with self.assertRaises(ValidationError): + self._location_serves().model_validate( + { + "point": {"latitude": 1.0, "longitude": 2.0}, + "dev.example.custom_target": {"foo": "bar"}, + } + ) + + @unittest.skipUnless( HAVE_SDK, "requires the installed package (pip install -e .)" )