From c88dc6a525d1ed6e306d8e591a2318bf9d41eba4 Mon Sep 17 00:00:00 2001 From: FanouZeng-TT <18280587072@163.com> Date: Thu, 13 Aug 2026 14:00:27 +0800 Subject: [PATCH] fix: enforce conditional total amount bounds --- postprocess_models.py | 230 +++++++++++++++++- .../models/schemas/shopping/types/total.py | 40 ++- .../models/schemas/shopping/types/totals.py | 46 +++- tests/test_codegen_pipeline.py | 141 +++++++++++ 4 files changed, 453 insertions(+), 4 deletions(-) diff --git a/postprocess_models.py b/postprocess_models.py index 6b08219..b4d8309 100644 --- a/postprocess_models.py +++ b/postprocess_models.py @@ -14,7 +14,7 @@ """Post-generation fixes for constraints datamodel-code-generator ignores. -Six constraint families are handled: +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 @@ -74,6 +74,12 @@ then injects a ``model_validator(mode="after")``. More complex conditions are skipped rather than approximated. +* Simple conditional numeric bounds are also dropped: well-known ``Total`` + categories constrain ``amount`` to negative or non-negative values. The script + accepts only one discriminator, one target field, and one numeric bound in each + ``if``/``then`` rule, and injects a ``model_validator(mode="after")``. Rules + with ``else`` or multiple consequence fields are skipped. + * ``additionalProperties: false`` on an object schema with named properties is normally overridden by the generator's ``--extra-fields=allow`` flag. The script detects schemas with ``additionalProperties: false`` and flips their @@ -148,6 +154,33 @@ def {marker}(self): return self ''' +_CONDITIONAL_NUMERIC_MARKER = "_enforce_conditional_numeric_bounds" + +_CONDITIONAL_NUMERIC_TEMPLATE = ''' + @model_validator(mode="after") + def {marker}(self): + """JSON Schema if/then: enforce conditional numeric bounds.""" + rules = {rules!r} + operators = {{ + "minimum": lambda value, bound: value >= bound, + "exclusiveMinimum": lambda value, bound: value > bound, + "maximum": lambda value, bound: value <= bound, + "exclusiveMaximum": lambda value, bound: value < bound, + }} + for rule in rules: + if getattr(self, rule["discriminator"], None) not in rule["values"]: + continue + value = getattr(self, rule["field"], None) + if value is not None and not operators[rule["bound"]]( + value, rule["value"] + ): + raise ValueError( + f"Field {{rule['field']!r}} violates conditional " + f"{{rule['bound']}}={{rule['value']}}" + ) + return self +''' + _UNIQUE_VALIDATOR_TEMPLATE = ''' @field_validator("{field}", mode="after") def {marker}_{field}(cls, value): # noqa: N805 @@ -690,6 +723,162 @@ def inject_conditional_required(source, class_name, rules): return _ensure_pydantic_import(out, "model_validator") +def find_conditional_numeric_bounds(schema_dir): + """Map generated classes to simple if/then numeric-bound rules.""" + rules_by_class = {} + bound_names = { + "minimum", + "exclusiveMinimum", + "maximum", + "exclusiveMaximum", + } + + def describe(node, properties): + if not isinstance(node, dict) or set(node) != {"if", "then"}: + return None + condition = node["if"] + consequence = node["then"] + if ( + not isinstance(condition, dict) + or set(condition) != {"properties", "required"} + or not isinstance(consequence, dict) + or set(consequence) != {"properties"} + ): + return None + condition_props = condition["properties"] + condition_required = condition["required"] + consequence_props = consequence["properties"] + if ( + not isinstance(condition_props, dict) + or len(condition_props) != 1 + or not isinstance(condition_required, list) + or len(condition_required) != 1 + or not isinstance(consequence_props, dict) + or len(consequence_props) != 1 + ): + return None + discriminator, predicate = next(iter(condition_props.items())) + field, constraint = next(iter(consequence_props.items())) + if condition_required != [discriminator] or not isinstance( + predicate, dict + ): + return None + if set(predicate) == {"const"}: + values = [predicate["const"]] + elif ( + set(predicate) == {"enum"} + and isinstance(predicate["enum"], list) + and predicate["enum"] + ): + values = predicate["enum"] + else: + return None + if ( + not isinstance(constraint, dict) + or len(constraint) != 1 + or not set(constraint) <= bound_names + ): + return None + bound, value = next(iter(constraint.items())) + if ( + discriminator not in properties + or field not in properties + or isinstance(value, bool) + or not isinstance(value, (int, float)) + or any( + not isinstance(item, (str, int, float, bool)) for item in values + ) + ): + return None + return { + "discriminator": discriminator, + "values": values, + "field": field, + "bound": bound, + "value": value, + } + + def walk(node, current_class_name, class_properties, path_str): + if not isinstance(node, dict): + return + if isinstance(node.get("title"), str): + current_class_name = _alias_name(node["title"]) + properties = node.get("properties") + if isinstance(properties, dict): + class_properties = properties + if "if" in node or "then" in node: + consequence = node.get("then") + consequence_props = ( + consequence.get("properties") + if isinstance(consequence, dict) + else None + ) + is_numeric_rule = isinstance(consequence_props, dict) and any( + isinstance(constraint, dict) + and bool(set(constraint) & bound_names) + for constraint in consequence_props.values() + ) + if is_numeric_rule: + rule = ( + None if "else" in node else describe(node, class_properties) + ) + if rule is None: + sys.stderr.write( + f" ! {path_str}: unsupported conditional numeric " + "rule; skipped\n" + ) + elif current_class_name is not None: + rules_by_class.setdefault(current_class_name, []).append( + rule + ) + if isinstance(properties, dict): + for name, prop in properties.items(): + walk(prop, _to_camel_case(name), properties, path_str) + defs = node.get("$defs") + if isinstance(defs, dict): + for def_name, def_node in defs.items(): + walk(def_node, _to_camel_case(def_name), {}, path_str) + for key in ("allOf", "anyOf", "oneOf"): + if isinstance(node.get(key), list): + for item in node[key]: + walk(item, current_class_name, class_properties, path_str) + + 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 + root_title = schema.get("title") + initial_class = ( + _alias_name(root_title) if root_title else _to_camel_case(path.stem) + ) + walk(schema, initial_class, {}, str(path)) + return rules_by_class + + +def inject_conditional_numeric_bounds(source, class_name, rules): + """Inject simple conditional numeric checks into one generated class.""" + class_re = re.compile(rf"^class {re.escape(class_name)}\(", re.M) + match = class_re.search(source) + if not match: + return source + tail = re.compile(r"^\S", re.M) + end_match = tail.search(source, match.end()) + end = end_match.start() if end_match else len(source) + if f"def {_CONDITIONAL_NUMERIC_MARKER}(" in source[match.start() : end]: + return source + method = _CONDITIONAL_NUMERIC_TEMPLATE.format( + marker=_CONDITIONAL_NUMERIC_MARKER, + rules=rules, + ) + 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 find_unique_items_fields(schema_dir): """Map generated class names to fields carrying ``uniqueItems``. @@ -982,6 +1171,41 @@ def _patch_conditional_required(): return patched, 0 +def _patch_conditional_numeric_bounds(): + """Inject conditional numeric validators; return counts and status.""" + rules_by_class = find_conditional_numeric_bounds(RAW_SCHEMA_DIR) + if not rules_by_class: + sys.stdout.write( + "postprocess: no simple conditional numeric rules found\n" + ) + return 0, 0 + patched = 0 + for class_name, rules in sorted(rules_by_class.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(class_name)}\(", source, re.M + ): + continue + updated = inject_conditional_numeric_bounds( + source, class_name, rules + ) + if updated != source: + path.write_text(updated, encoding="utf-8") + patched += 1 + hits.append(path) + label = ( + ", ".join(str(path) for path in hits) or "NO GENERATED CLASS FOUND" + ) + sys.stdout.write( + f" conditional numeric bounds on '{class_name}' -> {label}\n" + ) + if not hits: + return patched, 1 + return patched, 0 + + def _patch_unique_items(): """Inject uniqueItems validators; return (patched_count, exit_code).""" unique_fields_by_class = find_unique_items_fields(SCHEMA_DIR) @@ -1121,6 +1345,7 @@ def main(): patched_pn, rc_pn = _patch_property_names() patched_ac, rc_ac = _patch_array_contains() patched_cr, rc_cr = _patch_conditional_required() + patched_cn, rc_cn = _patch_conditional_numeric_bounds() patched_ui, rc_ui = _patch_unique_items() patched_ef, rc_ef = _patch_extra_forbid() total = ( @@ -1128,11 +1353,12 @@ def main(): + patched_pn + patched_ac + patched_cr + + patched_cn + patched_ui + 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_ui or rc_ef + return rc_mp or rc_pn or rc_ac or rc_cr or rc_cn or rc_ui or rc_ef if __name__ == "__main__": diff --git a/src/ucp_sdk/models/schemas/shopping/types/total.py b/src/ucp_sdk/models/schemas/shopping/types/total.py index 006d68b..b8f33da 100644 --- a/src/ucp_sdk/models/schemas/shopping/types/total.py +++ b/src/ucp_sdk/models/schemas/shopping/types/total.py @@ -18,7 +18,7 @@ from __future__ import annotations -from pydantic import BaseModel, ConfigDict +from pydantic import BaseModel, ConfigDict, model_validator from . import signed_amount @@ -40,3 +40,41 @@ class Total(BaseModel): Text to display against the amount. Should reflect appropriate method (e.g., 'Shipping', 'Delivery'). """ amount: signed_amount.SignedAmount + + @model_validator(mode="after") + def _enforce_conditional_numeric_bounds(self): + """JSON Schema if/then: enforce conditional numeric bounds.""" + rules = [ + { + "discriminator": "type", + "values": ["discount", "items_discount"], + "field": "amount", + "bound": "exclusiveMaximum", + "value": 0, + }, + { + "discriminator": "type", + "values": ["subtotal", "fulfillment", "tax", "fee"], + "field": "amount", + "bound": "minimum", + "value": 0, + }, + ] + operators = { + "minimum": lambda value, bound: value >= bound, + "exclusiveMinimum": lambda value, bound: value > bound, + "maximum": lambda value, bound: value <= bound, + "exclusiveMaximum": lambda value, bound: value < bound, + } + for rule in rules: + if getattr(self, rule["discriminator"], None) not in rule["values"]: + continue + value = getattr(self, rule["field"], None) + if value is not None and not operators[rule["bound"]]( + value, rule["value"] + ): + raise ValueError( + f"Field {rule['field']!r} violates conditional " + f"{rule['bound']}={rule['value']}" + ) + return self diff --git a/src/ucp_sdk/models/schemas/shopping/types/totals.py b/src/ucp_sdk/models/schemas/shopping/types/totals.py index 0011b1d..5a6c7fa 100644 --- a/src/ucp_sdk/models/schemas/shopping/types/totals.py +++ b/src/ucp_sdk/models/schemas/shopping/types/totals.py @@ -20,7 +20,13 @@ from typing import Annotated -from pydantic import BaseModel, ConfigDict, Field, AfterValidator +from pydantic import ( + BaseModel, + ConfigDict, + Field, + AfterValidator, + model_validator, +) from typing_extensions import TypeAliasType from . import signed_amount @@ -51,6 +57,44 @@ class Total(Total_1): Optional itemized breakdown. The parent entry is always rendered; lines are supplementary. Sum of line amounts MUST equal the parent entry amount. """ + @model_validator(mode="after") + def _enforce_conditional_numeric_bounds(self): + """JSON Schema if/then: enforce conditional numeric bounds.""" + rules = [ + { + "discriminator": "type", + "values": ["discount", "items_discount"], + "field": "amount", + "bound": "exclusiveMaximum", + "value": 0, + }, + { + "discriminator": "type", + "values": ["subtotal", "fulfillment", "tax", "fee"], + "field": "amount", + "bound": "minimum", + "value": 0, + }, + ] + operators = { + "minimum": lambda value, bound: value >= bound, + "exclusiveMinimum": lambda value, bound: value > bound, + "maximum": lambda value, bound: value <= bound, + "exclusiveMaximum": lambda value, bound: value < bound, + } + for rule in rules: + if getattr(self, rule["discriminator"], None) not in rule["values"]: + continue + value = getattr(self, rule["field"], None) + if value is not None and not operators[rule["bound"]]( + value, rule["value"] + ): + raise ValueError( + f"Field {rule['field']!r} violates conditional " + f"{rule['bound']}={rule['value']}" + ) + return self + def _enforce_contains_totals(value): """JSON Schema contains/minContains/maxContains (see #49).""" diff --git a/tests/test_codegen_pipeline.py b/tests/test_codegen_pipeline.py index ccdef8e..13221d8 100644 --- a/tests/test_codegen_pipeline.py +++ b/tests/test_codegen_pipeline.py @@ -31,6 +31,7 @@ from pydantic import TypeAdapter, ValidationError from ucp_sdk.models.schemas.shopping.types.description import Description + from ucp_sdk.models.schemas.shopping.types.total import Total from ucp_sdk.models.schemas.shopping.types.totals import Totals from ucp_sdk.models.schemas.shopping.types.totals_create_request import ( TotalsCreateRequest, @@ -1073,6 +1074,146 @@ def test_injected_validator_enforces_conditional_required(self): response(has_next_page=False) +class ConditionalNumericBoundsInjectorTest(unittest.TestCase): + """Simple JSON Schema if/then numeric bounds are restored.""" + + MODULE = ( + "from __future__ import annotations\n" + "\n" + "from pydantic import BaseModel, ConfigDict\n" + "\n" + "\n" + "class Total(BaseModel):\n" + ' model_config = ConfigDict(extra="allow")\n' + " type: str\n" + " amount: int\n" + ) + RULES = [ + { + "discriminator": "type", + "values": ["discount", "items_discount"], + "field": "amount", + "bound": "exclusiveMaximum", + "value": 0, + }, + { + "discriminator": "type", + "values": ["subtotal", "tax"], + "field": "amount", + "bound": "minimum", + "value": 0, + }, + ] + + def test_schema_scan_finds_allof_numeric_bounds(self): + schema = { + "title": "Total", + "type": "object", + "properties": { + "type": {"type": "string"}, + "amount": {"type": "integer"}, + }, + "allOf": [ + { + "if": { + "properties": { + "type": {"enum": ["discount", "items_discount"]} + }, + "required": ["type"], + }, + "then": {"properties": {"amount": {"exclusiveMaximum": 0}}}, + }, + { + "if": { + "properties": {"type": {"enum": ["subtotal", "tax"]}}, + "required": ["type"], + }, + "then": {"properties": {"amount": {"minimum": 0}}}, + }, + ], + } + with tempfile.TemporaryDirectory() as tmp: + Path(tmp, "total.json").write_text(json.dumps(schema)) + found = postprocess_models.find_conditional_numeric_bounds( + Path(tmp) + ) + self.assertEqual(found, {"Total": self.RULES}) + + def test_schema_scan_skips_else_and_multiple_target_fields(self): + schema = { + "title": "Total", + "type": "object", + "properties": { + "type": {"type": "string"}, + "amount": {"type": "integer"}, + "other": {"type": "integer"}, + }, + "allOf": [ + { + "if": { + "properties": {"type": {"const": "discount"}}, + "required": ["type"], + }, + "then": {"properties": {"amount": {"maximum": 0}}}, + "else": {"properties": {"amount": {"minimum": 0}}}, + }, + { + "if": { + "properties": {"type": {"const": "fee"}}, + "required": ["type"], + }, + "then": { + "properties": { + "amount": {"minimum": 0}, + "other": {"minimum": 0}, + } + }, + }, + ], + } + with tempfile.TemporaryDirectory() as tmp: + Path(tmp, "total.json").write_text(json.dumps(schema)) + found = postprocess_models.find_conditional_numeric_bounds( + Path(tmp) + ) + self.assertEqual(found, {}) + + def test_injection_is_idempotent(self): + once = postprocess_models.inject_conditional_numeric_bounds( + self.MODULE, "Total", self.RULES + ) + twice = postprocess_models.inject_conditional_numeric_bounds( + once, "Total", self.RULES + ) + self.assertEqual(once, twice) + + @unittest.skipUnless(HAVE_SDK, "executing the module needs pydantic") + def test_injected_validator_enforces_numeric_bounds(self): + out = postprocess_models.inject_conditional_numeric_bounds( + self.MODULE, "Total", self.RULES + ) + namespace: dict = {} + exec(compile(out, "", "exec"), namespace) # noqa: S102 + total_cls = namespace["Total"] + with self.assertRaises(ValidationError): + total_cls(type="discount", amount=1) + with self.assertRaises(ValidationError): + total_cls(type="subtotal", amount=-1) + total_cls(type="discount", amount=-1) + total_cls(type="subtotal", amount=0) + total_cls(type="custom", amount=-1) + + @unittest.skipUnless(HAVE_SDK, "executing the model needs pydantic") + def test_generated_total_enforces_schema_bounds(self): + with self.assertRaises(ValidationError): + Total(type="discount", amount=1) + with self.assertRaises(ValidationError): + Total(type="subtotal", amount=-1) + Total(type="discount", amount=-1) + Total(type="subtotal", amount=0) + Total(type="custom", amount=-1) + + class InjectorTest(unittest.TestCase): """The post-generation injector's own behavior."""