diff --git a/postprocess_models.py b/postprocess_models.py index 84d35e7..d6aa55b 100644 --- a/postprocess_models.py +++ b/postprocess_models.py @@ -72,16 +72,27 @@ always treats it as optional. The script accepts only an unambiguous single required discriminator using ``const``/``enum`` and a ``then.required`` list, then injects a ``model_validator(mode="after")``. More complex conditions are - skipped rather than approximated. - -* Conditional numeric bounds are dropped for the same reason: ``total.json`` - requires a ``discount`` amount to be negative and a ``tax`` amount to be - non-negative via if/then branches, but the generated ``Total`` carries no - validator, so a positive discount validates. These rules are carried as - ``allOf`` branches, which have no sibling ``properties`` of their own, so the - scan validates them against the enclosing object's property set. A rule whose - fields were stripped by request-variant projection is inapplicable rather than - malformed and is skipped silently. + skipped rather than approximated. These rules are commonly carried as + ``allOf`` branches with no sibling ``properties`` of their own (every + conditional rule in ``profile.json``'s ``jwk_public_key`` is this shape), so + the scan validates them against the enclosing object's property set, the + same threading ``find_conditional_bounds`` (below) already used. A branch's + own documentation ``title`` (some carry one purely as rule prose, e.g. "EC + keys carry crv, x, y") is never adopted as the enclosing class name; see + ``_is_bare_conditional_branch``. + +* Conditional numeric bounds, and conditional exact-value (``const``) pins, + are dropped for the same reason: ``total.json`` requires a ``discount`` + amount to be negative and a ``tax`` amount to be non-negative via if/then + branches, and ``unit.json`` pins ``scale`` to exactly 0 when ``unit`` is + ``C62``, but the generated ``Total``/``Unit`` carry no validator, so a + positive discount or a nonzero C62 scale both validate. These rules are + carried as ``allOf`` branches, which have no sibling ``properties`` of + their own, so the scan validates them against the enclosing object's + property set. A rule whose fields were stripped by request-variant + projection is inapplicable rather than malformed and is skipped silently. + As with conditional required rules above, a branch's own documentation + ``title`` is never adopted as the enclosing class name. * ``additionalProperties: false`` on an object schema with named properties is normally overridden by the generator's ``--extra-fields=allow`` flag. The @@ -166,11 +177,15 @@ def {marker}(self): # Keyword -> (comparison rendered in the message, python operator name). The # operator is applied as "value limit" and a true result is a violation. +# "const" pins the field to an exact value (e.g. unit.json: scale must be +# exactly 0 when unit is C62) rather than bounding a range; it reuses the +# same "value limit -> violation" shape with not-equal as the operator. _BOUND_KEYWORDS = { "minimum": (">=", "lt"), "maximum": ("<=", "gt"), "exclusiveMinimum": (">", "le"), "exclusiveMaximum": ("<", "ge"), + "const": ("==", "ne"), } _CONDITIONAL_BOUNDS_TEMPLATE = ''' @@ -553,6 +568,27 @@ def _alias_name(title): return "".join(title.split()) +def _is_bare_conditional_branch(node): + """True when ``node`` is an if/then rule wrapper with no type of its own. + + UCP schemas sometimes give an allOf if/then branch its own human-readable + ``title`` purely as rule documentation (e.g. profile.json's jwk_public_key: + "EC keys carry crv, x, y", "P-256 pairs with ES256"). Such a branch has no + ``properties`` of its own -- it constrains the *enclosing* object -- so its + title never names a generated class. Adopting it as ``current_class_name`` + (the same code path real class-defining titles use) misattributes the + rule to a nonexistent class instead of the enclosing one. A node that + carries both an if/then rule AND its own ``properties`` is a genuine + titled type that happens to declare an inline conditional, and keeps + adopting its title as before. + """ + return ( + isinstance(node.get("if"), dict) + and isinstance(node.get("then"), dict) + and not isinstance(node.get("properties"), dict) + ) + + def _to_camel_case(string): """Convert a string (snake, kebab, space-separated) to CamelCase.""" parts = re.split(r"[^a-zA-Z0-9]", string) @@ -754,21 +790,32 @@ def describe(node, properties): "required": sorted(consequence_required), } - def walk(node, current_class_name, path_str): + def walk(node, current_class_name, path_str, enclosing_properties=None): if not isinstance(node, dict): return - if isinstance(node.get("title"), str): + if isinstance( + node.get("title"), str + ) and not _is_bare_conditional_branch(node): current_class_name = _alias_name(node["title"]) properties = node.get("properties") + # An if/then pair carried as an allOf branch has no sibling + # properties of its own (every JWK required-field rule is exactly + # this shape): the object it constrains is the enclosing schema, so + # its property set is what the rule must be validated against. This + # mirrors find_conditional_bounds's existing enclosing_properties + # threading. + scope = ( + properties if isinstance(properties, dict) else enclosing_properties + ) then = node.get("then") is_required_rule = isinstance(then, dict) and "required" in then - if isinstance(properties, dict) and is_required_rule: + if isinstance(scope, dict) and is_required_rule: if "else" in node: rule = None else: rule = describe( {key: node[key] for key in ("if", "then") if key in node}, - properties, + scope, ) if rule is None: sys.stderr.write( @@ -786,7 +833,7 @@ def walk(node, current_class_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, path_str) + walk(item, current_class_name, path_str, scope) for path in sorted(Path(schema_dir).rglob("*.json")): try: @@ -886,11 +933,20 @@ def describe(node, properties): or set(constraint) - set(_BOUND_KEYWORDS) ): return None - if any( - not isinstance(limit, (int, float)) or isinstance(limit, bool) - for limit in constraint.values() - ): - return None + # "const" pins an exact value and may legitimately be a string + # (unit.json's C62 pin is an int; JWK's algorithm pins are + # strings), so it is checked against the same scalar types + # already accepted for the discriminator's own const/enum + # values above. The numeric bound keywords keep their existing, + # narrower int/float (non-bool) requirement. + for keyword, limit in constraint.items(): + if keyword == "const": + if not isinstance(limit, (str, int, float, bool)): + return None + elif not isinstance(limit, (int, float)) or isinstance( + limit, bool + ): + return None bounds[name] = dict(constraint) # A request variant strips the fields a platform must not send, so a # rule naming one is inapplicable to that class rather than malformed. @@ -907,7 +963,9 @@ def describe(node, properties): def walk(node, current_class_name, path_str, enclosing_properties=None): if not isinstance(node, dict): return - if isinstance(node.get("title"), str): + if isinstance( + node.get("title"), str + ) and not _is_bare_conditional_branch(node): current_class_name = _alias_name(node["title"]) properties = node.get("properties") # An if/then pair carried as an allOf branch has no sibling properties: diff --git a/src/ucp_sdk/models/schemas/common/types/total.py b/src/ucp_sdk/models/schemas/common/types/total.py index 72b904e..d4c4c6e 100644 --- a/src/ucp_sdk/models/schemas/common/types/total.py +++ b/src/ucp_sdk/models/schemas/common/types/total.py @@ -63,6 +63,7 @@ def _enforce_conditional_bounds(self): "maximum": ("<=", "gt"), "exclusiveMinimum": (">", "le"), "exclusiveMaximum": ("<", "ge"), + "const": ("==", "ne"), } for rule in rules: actual = getattr(self, rule["discriminator"], None) diff --git a/src/ucp_sdk/models/schemas/common/types/totals.py b/src/ucp_sdk/models/schemas/common/types/totals.py index 642088a..bba32a9 100644 --- a/src/ucp_sdk/models/schemas/common/types/totals.py +++ b/src/ucp_sdk/models/schemas/common/types/totals.py @@ -79,6 +79,7 @@ def _enforce_conditional_bounds(self): "maximum": ("<=", "gt"), "exclusiveMinimum": (">", "le"), "exclusiveMaximum": ("<", "ge"), + "const": ("==", "ne"), } for rule in rules: actual = getattr(self, rule["discriminator"], None) diff --git a/src/ucp_sdk/models/schemas/common/types/unit.py b/src/ucp_sdk/models/schemas/common/types/unit.py index bddaa3b..478eccb 100644 --- a/src/ucp_sdk/models/schemas/common/types/unit.py +++ b/src/ucp_sdk/models/schemas/common/types/unit.py @@ -18,7 +18,9 @@ from __future__ import annotations -from pydantic import BaseModel, ConfigDict, Field +import operator + +from pydantic import BaseModel, ConfigDict, Field, model_validator class Unit(BaseModel): @@ -41,3 +43,37 @@ class Unit(BaseModel): """ Required printable unit label provided by the Business. The Platform MUST use it when it does not recognize `unit`; for a recognized UN/CEFACT Rec 20 Common Code, the Platform MAY substitute its own localized label. It does not participate in unit identity or mismatch comparison. """ + + @model_validator(mode="after") + def _enforce_conditional_bounds(self): + """JSON Schema if/then: enforce conditional numeric bounds.""" + rules = [ + { + "discriminator": "unit", + "values": ["C62"], + "bounds": {"scale": {"const": 0}}, + } + ] + checks = { + "minimum": (">=", "lt"), + "maximum": ("<=", "gt"), + "exclusiveMinimum": (">", "le"), + "exclusiveMaximum": ("<", "ge"), + "const": ("==", "ne"), + } + for rule in rules: + actual = getattr(self, rule["discriminator"], None) + if actual not in rule["values"]: + continue + for field, bounds in rule["bounds"].items(): + value = getattr(self, field, None) + if value is None: + continue + for keyword, limit in bounds.items(): + symbol, op_name = checks[keyword] + if getattr(operator, op_name)(value, limit): + raise ValueError( + f"Field {field!r} must be {symbol} {limit} " + f"when {rule['discriminator']} is {actual!r}" + ) + return self diff --git a/src/ucp_sdk/models/schemas/common/types/unit_create_request.py b/src/ucp_sdk/models/schemas/common/types/unit_create_request.py index 2fd5282..56b8475 100644 --- a/src/ucp_sdk/models/schemas/common/types/unit_create_request.py +++ b/src/ucp_sdk/models/schemas/common/types/unit_create_request.py @@ -18,7 +18,9 @@ from __future__ import annotations -from pydantic import BaseModel, ConfigDict, Field +import operator + +from pydantic import BaseModel, ConfigDict, Field, model_validator class UnitCreateRequest(BaseModel): @@ -41,3 +43,37 @@ class UnitCreateRequest(BaseModel): """ Required printable unit label provided by the Business. The Platform MUST use it when it does not recognize `unit`; for a recognized UN/CEFACT Rec 20 Common Code, the Platform MAY substitute its own localized label. It does not participate in unit identity or mismatch comparison. """ + + @model_validator(mode="after") + def _enforce_conditional_bounds(self): + """JSON Schema if/then: enforce conditional numeric bounds.""" + rules = [ + { + "discriminator": "unit", + "values": ["C62"], + "bounds": {"scale": {"const": 0}}, + } + ] + checks = { + "minimum": (">=", "lt"), + "maximum": ("<=", "gt"), + "exclusiveMinimum": (">", "le"), + "exclusiveMaximum": ("<", "ge"), + "const": ("==", "ne"), + } + for rule in rules: + actual = getattr(self, rule["discriminator"], None) + if actual not in rule["values"]: + continue + for field, bounds in rule["bounds"].items(): + value = getattr(self, field, None) + if value is None: + continue + for keyword, limit in bounds.items(): + symbol, op_name = checks[keyword] + if getattr(operator, op_name)(value, limit): + raise ValueError( + f"Field {field!r} must be {symbol} {limit} " + f"when {rule['discriminator']} is {actual!r}" + ) + return self diff --git a/src/ucp_sdk/models/schemas/common/types/unit_update_request.py b/src/ucp_sdk/models/schemas/common/types/unit_update_request.py index a38f7d5..1b52ca5 100644 --- a/src/ucp_sdk/models/schemas/common/types/unit_update_request.py +++ b/src/ucp_sdk/models/schemas/common/types/unit_update_request.py @@ -18,7 +18,9 @@ from __future__ import annotations -from pydantic import BaseModel, ConfigDict, Field +import operator + +from pydantic import BaseModel, ConfigDict, Field, model_validator class UnitUpdateRequest(BaseModel): @@ -41,3 +43,37 @@ class UnitUpdateRequest(BaseModel): """ Required printable unit label provided by the Business. The Platform MUST use it when it does not recognize `unit`; for a recognized UN/CEFACT Rec 20 Common Code, the Platform MAY substitute its own localized label. It does not participate in unit identity or mismatch comparison. """ + + @model_validator(mode="after") + def _enforce_conditional_bounds(self): + """JSON Schema if/then: enforce conditional numeric bounds.""" + rules = [ + { + "discriminator": "unit", + "values": ["C62"], + "bounds": {"scale": {"const": 0}}, + } + ] + checks = { + "minimum": (">=", "lt"), + "maximum": ("<=", "gt"), + "exclusiveMinimum": (">", "le"), + "exclusiveMaximum": ("<", "ge"), + "const": ("==", "ne"), + } + for rule in rules: + actual = getattr(self, rule["discriminator"], None) + if actual not in rule["values"]: + continue + for field, bounds in rule["bounds"].items(): + value = getattr(self, field, None) + if value is None: + continue + for keyword, limit in bounds.items(): + symbol, op_name = checks[keyword] + if getattr(operator, op_name)(value, limit): + raise ValueError( + f"Field {field!r} must be {symbol} {limit} " + f"when {rule['discriminator']} is {actual!r}" + ) + return self diff --git a/src/ucp_sdk/models/schemas/profile.py b/src/ucp_sdk/models/schemas/profile.py index 500c36d..91a2848 100644 --- a/src/ucp_sdk/models/schemas/profile.py +++ b/src/ucp_sdk/models/schemas/profile.py @@ -18,7 +18,9 @@ from __future__ import annotations -from pydantic import BaseModel, ConfigDict, Field +import operator + +from pydantic import BaseModel, ConfigDict, Field, model_validator from . import ucp as ucp_1 @@ -60,6 +62,75 @@ class JwkPublicKey(BaseModel): JWK public key use. UCP examples use sig for signatures. """ + @model_validator(mode="after") + def _enforce_conditional_required(self): + """JSON Schema if/then: enforce conditionally required fields.""" + rules = [ + { + "discriminator": "kty", + "values": ["EC"], + "required": ["crv", "x", "y"], + }, + { + "discriminator": "kty", + "values": ["OKP"], + "required": ["crv", "x"], + }, + ] + for rule in rules: + if getattr(self, rule["discriminator"], None) not in rule["values"]: + continue + for field in rule["required"]: + if field not in self.model_fields_set: + raise ValueError( + f"Field {field!r} is required by a schema condition" + ) + return self + + @model_validator(mode="after") + def _enforce_conditional_bounds(self): + """JSON Schema if/then: enforce conditional numeric bounds.""" + rules = [ + { + "discriminator": "crv", + "values": ["P-256"], + "bounds": {"alg": {"const": "ES256"}}, + }, + { + "discriminator": "crv", + "values": ["P-384"], + "bounds": {"alg": {"const": "ES384"}}, + }, + { + "discriminator": "crv", + "values": ["Ed25519"], + "bounds": {"alg": {"const": "EdDSA"}}, + }, + ] + checks = { + "minimum": (">=", "lt"), + "maximum": ("<=", "gt"), + "exclusiveMinimum": (">", "le"), + "exclusiveMaximum": ("<", "ge"), + "const": ("==", "ne"), + } + for rule in rules: + actual = getattr(self, rule["discriminator"], None) + if actual not in rule["values"]: + continue + for field, bounds in rule["bounds"].items(): + value = getattr(self, field, None) + if value is None: + continue + for keyword, limit in bounds.items(): + symbol, op_name = checks[keyword] + if getattr(operator, op_name)(value, limit): + raise ValueError( + f"Field {field!r} must be {symbol} {limit} " + f"when {rule['discriminator']} is {actual!r}" + ) + return self + class UcpProfileDocument(BaseModel): """ diff --git a/tests/test_codegen_pipeline.py b/tests/test_codegen_pipeline.py index eba1f0c..ac62ea4 100644 --- a/tests/test_codegen_pipeline.py +++ b/tests/test_codegen_pipeline.py @@ -31,12 +31,19 @@ try: from pydantic import TypeAdapter, ValidationError - from ucp_sdk.models.schemas.shopping.types.description import Description - from ucp_sdk.models.schemas.shopping.types.totals import Totals - from ucp_sdk.models.schemas.shopping.types.totals_create_request import ( + # NOTE(root-cause-0): these paths moved from shopping.types to + # common.types when #87 (2026-08-25 UCP release regen) restructured the + # schema tree. The old paths silently raise ModuleNotFoundError here, + # which the except clause below swallows as HAVE_SDK = False -- so every + # semantic test gated on HAVE_SDK skips instead of running, and CI is + # green on a suite that mostly never executed. See the sibling fixes to + # the other stale shopping.types.* imports later in this file. + from ucp_sdk.models.schemas.common.types.description import Description + from ucp_sdk.models.schemas.common.types.totals import Totals + from ucp_sdk.models.schemas.common.types.totals_create_request import ( TotalsCreateRequest, ) - from ucp_sdk.models.schemas.shopping.types.totals_update_request import ( + from ucp_sdk.models.schemas.common.types.totals_update_request import ( TotalsUpdateRequest, ) @@ -1014,7 +1021,7 @@ class SignalsPropertyNamesTest(unittest.TestCase): """ def _signals(self): - from ucp_sdk.models.schemas.shopping.types.signals import Signals + from ucp_sdk.models.schemas.common.types.signals import Signals return Signals @@ -1053,13 +1060,13 @@ def test_known_named_fields_still_work(self): def test_request_variants_enforce_property_names(self): # The gap and its fix travel to the generated request variants too. - from ucp_sdk.models.schemas.shopping.types.signals_complete_request import ( + from ucp_sdk.models.schemas.common.types.signals_complete_request import ( SignalsCompleteRequest, ) - from ucp_sdk.models.schemas.shopping.types.signals_create_request import ( + from ucp_sdk.models.schemas.common.types.signals_create_request import ( SignalsCreateRequest, ) - from ucp_sdk.models.schemas.shopping.types.signals_update_request import ( + from ucp_sdk.models.schemas.common.types.signals_update_request import ( SignalsUpdateRequest, ) @@ -1240,6 +1247,75 @@ def test_schema_scan_skips_else_branches(self): found = postprocess_models.find_conditional_required(Path(tmp)) self.assertEqual(found, {}) + def test_schema_scan_threads_enclosing_scope_into_titled_allof_branch( + self, + ): + """An allOf branch with neither its own `properties` nor a type + title of its own is invisible today: the scan only looks at + `node.get("properties")` on the branch itself, so a branch that + relies on the enclosing object's properties (JWK's five if/then + rules, none of which repeat `properties`) is silently dropped with + no warning. And when the branch DOES carry a human-readable + documentation `title` (JWK's branches are each titled, e.g. "EC + keys carry crv, x, y"), the current code adopts that title as the + class name via `_alias_name`, misattributing the rule to a + nonexistent class instead of the enclosing `JwkPublicKey`. This + mirrors profile.json's jwk_public_key def exactly. + """ + schema = { + "$defs": { + "jwk_public_key": { + "type": "object", + "required": ["kid", "kty"], + "properties": { + "kid": {"type": "string"}, + "kty": {"type": "string"}, + "crv": {"type": "string"}, + "x": {"type": "string"}, + "y": {"type": "string"}, + }, + "allOf": [ + { + "title": "EC keys carry crv, x, y", + "if": { + "properties": {"kty": {"const": "EC"}}, + "required": ["kty"], + }, + "then": {"required": ["crv", "x", "y"]}, + }, + { + "title": "OKP keys carry crv, x", + "if": { + "properties": {"kty": {"const": "OKP"}}, + "required": ["kty"], + }, + "then": {"required": ["crv", "x"]}, + }, + ], + } + } + } + with tempfile.TemporaryDirectory() as tmp: + Path(tmp, "profile.json").write_text(json.dumps(schema)) + found = postprocess_models.find_conditional_required(Path(tmp)) + self.assertEqual( + found, + { + "JwkPublicKey": [ + { + "discriminator": "kty", + "values": ["EC"], + "required": ["crv", "x", "y"], + }, + { + "discriminator": "kty", + "values": ["OKP"], + "required": ["crv", "x"], + }, + ] + }, + ) + def test_injection_is_idempotent(self): once = postprocess_models.inject_conditional_required( self.MODULE, "Response", self.RULES @@ -1349,6 +1425,97 @@ def test_schema_scan_warns_on_unsupported_shape(self): self.assertEqual(found, {"Total": [self.RULES[1]]}) self.assertIn("unsupported", stderr.getvalue()) + def test_schema_scan_recognizes_const_pinning(self): + """A `then.properties..const` pin is dropped today: describe() + only accepts the four numeric bound keywords in _BOUND_KEYWORDS, so + `set(constraint) - set(_BOUND_KEYWORDS)` is non-empty for a bare + `{"const": ...}` constraint and the whole rule returns None. This + mirrors unit.json exactly: when unit is C62, scale must be exactly + 0. The branch carries no title of its own (unlike the JWK case + below), isolating bug (b) from bug (c). + """ + schema = { + "title": "Unit", + "type": "object", + "properties": { + "unit": {"type": "string"}, + "scale": {"type": "integer"}, + }, + "allOf": [ + { + "if": { + "properties": {"unit": {"const": "C62"}}, + "required": ["unit"], + }, + "then": {"properties": {"scale": {"const": 0}}}, + } + ], + } + with tempfile.TemporaryDirectory() as tmp: + Path(tmp, "unit.json").write_text(json.dumps(schema)) + found = postprocess_models.find_conditional_bounds(Path(tmp)) + self.assertEqual( + found, + { + "Unit": [ + { + "discriminator": "unit", + "values": ["C62"], + "bounds": {"scale": {"const": 0}}, + } + ] + }, + ) + + def test_schema_scan_recognizes_const_pinning_in_titled_allof_branch( + self, + ): + """profile.json's JWK curve/algorithm pairing rules combine both + gaps at once: `then.properties.alg.const` (bug b, see above) inside + a branch that carries its own documentation `title` (bug c, see + ConditionalRequiredInjectorTest) which must not overwrite the + enclosing `JwkPublicKey` class name. + """ + schema = { + "$defs": { + "jwk_public_key": { + "type": "object", + "required": ["kid", "kty"], + "properties": { + "kid": {"type": "string"}, + "kty": {"type": "string"}, + "crv": {"type": "string"}, + "alg": {"type": "string"}, + }, + "allOf": [ + { + "title": "P-256 pairs with ES256", + "if": { + "properties": {"crv": {"const": "P-256"}}, + "required": ["crv"], + }, + "then": {"properties": {"alg": {"const": "ES256"}}}, + } + ], + } + } + } + with tempfile.TemporaryDirectory() as tmp: + Path(tmp, "profile.json").write_text(json.dumps(schema)) + found = postprocess_models.find_conditional_bounds(Path(tmp)) + self.assertEqual( + found, + { + "JwkPublicKey": [ + { + "discriminator": "crv", + "values": ["P-256"], + "bounds": {"alg": {"const": "ES256"}}, + } + ] + }, + ) + def test_injection_is_idempotent(self): once = postprocess_models.inject_conditional_bounds( self.MODULE, "Total", self.RULES @@ -1358,6 +1525,39 @@ def test_injection_is_idempotent(self): ) self.assertEqual(once, twice) + @unittest.skipUnless(HAVE_SDK, "executing the module needs pydantic") + def test_injected_validator_enforces_const_pinning(self): + module = ( + "from __future__ import annotations\n" + "\n" + "from pydantic import BaseModel, ConfigDict\n" + "\n" + "\n" + "class Unit(BaseModel):\n" + ' model_config = ConfigDict(extra="allow")\n' + " unit: str\n" + " scale: int | None = 0\n" + ) + rules = [ + { + "discriminator": "unit", + "values": ["C62"], + "bounds": {"scale": {"const": 0}}, + } + ] + out = postprocess_models.inject_conditional_bounds( + module, "Unit", rules + ) + namespace: dict = {} + exec(compile(out, "", "exec"), namespace) # noqa: S102 + unit = namespace["Unit"] + with self.assertRaises(ValidationError): + unit(unit="C62", scale=5) + unit(unit="C62", scale=0) + unit(unit="C62") + # A unit outside the pinned vocabulary is unconstrained. + unit(unit="KGM", scale=3) + @unittest.skipUnless(HAVE_SDK, "executing the module needs pydantic") def test_injected_validator_enforces_conditional_bounds(self): out = postprocess_models.inject_conditional_bounds( @@ -1870,18 +2070,41 @@ def test_injected_validator_rejects_duplicates(self) -> None: class UniqueItemsSemanticTest(unittest.TestCase): """Committed models enforce uniqueItems on declared array fields.""" + # NOTE(root-cause-0): card_payment_instrument.json no longer declares a + # Constraints.brands field (uniqueItems) as of the pinned 2026-08-25 UCP + # schema -- the module now generates only Display, ConstraintTarget and + # CardPaymentInstrument (verified against + # src/ucp_sdk/models/schemas/common/types/card_payment_instrument.py). + # These two tests exercised a schema shape that no longer exists; the + # HAVE_SDK import gate bug (see the top of this file) had been hiding + # that they could not pass, not just that they were unrelated to SDK + # availability. Documented skip rather than silent deletion: the + # uniqueItems mechanism itself stays covered by UniqueItemsInjectorTest + # (injector unit tests) and by other committed models with uniqueItems + # fields (e.g. common.types.constraint_expression, context, + # location_filter, request_constraints). + @unittest.skip( + "card_payment_instrument.Constraints.brands (uniqueItems) was " + "removed from the schema before the pinned 2026-08-25 UCP release; " + "no current committed model at this path carries a brands field" + ) def test_brands_rejects_duplicates(self) -> None: """card_payment_instrument brands rejects duplicate entries.""" - from ucp_sdk.models.schemas.shopping.types.card_payment_instrument import ( + from ucp_sdk.models.schemas.common.types.card_payment_instrument import ( Constraints, ) with self.assertRaisesRegex(ValidationError, "[Uu]nique"): Constraints(brands=["visa", "visa"]) + @unittest.skip( + "card_payment_instrument.Constraints.brands (uniqueItems) was " + "removed from the schema before the pinned 2026-08-25 UCP release; " + "no current committed model at this path carries a brands field" + ) def test_brands_accepts_unique_and_none(self) -> None: """Unique lists and missing values are accepted.""" - from ucp_sdk.models.schemas.shopping.types.card_payment_instrument import ( + from ucp_sdk.models.schemas.common.types.card_payment_instrument import ( Constraints, ) @@ -2024,7 +2247,7 @@ class AdditionalPropertiesForbidSemanticTest(unittest.TestCase): """Committed models reject unknown keys on additionalProperties:false.""" def test_error_response_rejects_unknown_keys(self) -> None: - from ucp_sdk.models.schemas.shopping.types.error_response import ( + from ucp_sdk.models.schemas.common.types.error_response import ( ErrorResponse, ) @@ -2045,7 +2268,7 @@ def test_error_response_rejects_unknown_keys(self) -> None: ) def test_error_response_accepts_declared_fields(self) -> None: - from ucp_sdk.models.schemas.shopping.types.error_response import ( + from ucp_sdk.models.schemas.common.types.error_response import ( ErrorResponse, ) @@ -2064,8 +2287,29 @@ def test_error_response_accepts_declared_fields(self) -> None: ) self.assertEqual(obj.messages[0].content, "boom") + # NOTE(root-cause-0): merchant_fulfillment_config.json was renamed and + # restructured to business_fulfillment_config.json before the pinned + # 2026-08-25 UCP release. The nested additionalProperties:false object + # these tests targeted (allows_multi_destination -> AllowsMultiDestination) + # is gone; the current schema's multi_destination field is a list of + # MultiDestinationItem (extra="allow", no nested forbid object) -- + # verified against + # src/ucp_sdk/models/schemas/shopping/types/business_fulfillment_config.py. + # The HAVE_SDK import gate bug (see the top of this file) had been + # hiding that these two tests could not pass at all, not just that they + # were unrelated to SDK availability. Documented skip rather than silent + # deletion: the additionalProperties:false -> extra="forbid" mechanism + # itself stays covered by test_error_response_rejects_unknown_keys above + # and by AdditionalPropertiesForbidInjectorTest/FinderTest. + @unittest.skip( + "merchant_fulfillment_config.AllowsMultiDestination was removed " + "when the schema was restructured to " + "business_fulfillment_config.MultiDestinationItem before the " + "pinned 2026-08-25 UCP release; no current committed model at " + "this path carries a nested additionalProperties:false object" + ) def test_allows_multi_destination_rejects_unknown_keys(self) -> None: - from ucp_sdk.models.schemas.shopping.types.merchant_fulfillment_config import ( + from ucp_sdk.models.schemas.shopping.types.business_fulfillment_config import ( AllowsMultiDestination, ) @@ -2074,12 +2318,19 @@ def test_allows_multi_destination_rejects_unknown_keys(self) -> None: {"shipping": True, "bogus": "x"} ) + @unittest.skip( + "merchant_fulfillment_config.MerchantFulfillmentConfig was renamed " + "and restructured to business_fulfillment_config." + "BusinessFulfillmentConfig before the pinned 2026-08-25 UCP " + "release; see test_allows_multi_destination_rejects_unknown_keys " + "above" + ) def test_sibling_config_keeps_extra_allow(self) -> None: - from ucp_sdk.models.schemas.shopping.types.merchant_fulfillment_config import ( - MerchantFulfillmentConfig, + from ucp_sdk.models.schemas.shopping.types.business_fulfillment_config import ( + BusinessFulfillmentConfig, ) - config = MerchantFulfillmentConfig.model_validate({"bogus": "x"}) + config = BusinessFulfillmentConfig.model_validate({"bogus": "x"}) self.assertEqual(config.model_extra, {"bogus": "x"}) @@ -2114,5 +2365,114 @@ def test_payment_handler_base_rejects_invalid_version(self) -> None: Base.model_validate({"version": {"not": "a version"}}) +@unittest.skipUnless( + HAVE_SDK, "requires the installed package (pip install -e .)" +) +class JwkConditionalRulesSemanticTest(unittest.TestCase): + """profile.json's jwk_public_key carries five if/then rules, all five + dropped by the generator today: two conditional-required rules (an EC + key needs crv/x/y, an OKP key needs crv/x) and three conditional + const-pin rules pairing a curve with its algorithm (P-256/ES256, + P-384/ES384, Ed25519/EdDSA). Security-adjacent: a profile publishing an + EC key with no curve, or an algorithm that does not match its curve, + currently passes SDK validation and would only fail (or silently + misverify) downstream at signature-verification time. + """ + + def _jwk(self): + from ucp_sdk.models.schemas.profile import JwkPublicKey + + return JwkPublicKey + + def test_ec_key_without_curve_and_coordinates_rejected(self): + with self.assertRaises(ValidationError): + self._jwk()(kid="k1", kty="EC") + + def test_ec_key_with_curve_and_coordinates_accepted(self): + key = self._jwk()( + kid="k1", kty="EC", crv="P-256", x="AA", y="BB", alg="ES256" + ) + self.assertEqual(key.crv, "P-256") + + def test_okp_key_without_curve_rejected(self): + with self.assertRaises(ValidationError): + self._jwk()(kid="k2", kty="OKP") + + def test_okp_key_with_curve_accepted(self): + key = self._jwk()(kid="k2", kty="OKP", crv="Ed25519", x="AA") + self.assertEqual(key.crv, "Ed25519") + + def test_p256_with_mismatched_algorithm_rejected(self): + with self.assertRaises(ValidationError): + self._jwk()( + kid="k3", kty="EC", crv="P-256", x="AA", y="BB", alg="EdDSA" + ) + + def test_p256_with_matching_algorithm_accepted(self): + self._jwk()( + kid="k3", kty="EC", crv="P-256", x="AA", y="BB", alg="ES256" + ) + + def test_p384_with_mismatched_algorithm_rejected(self): + with self.assertRaises(ValidationError): + self._jwk()( + kid="k4", kty="EC", crv="P-384", x="AA", y="BB", alg="ES256" + ) + + def test_p384_with_matching_algorithm_accepted(self): + self._jwk()( + kid="k4", kty="EC", crv="P-384", x="AA", y="BB", alg="ES384" + ) + + def test_ed25519_with_mismatched_algorithm_rejected(self): + with self.assertRaises(ValidationError): + self._jwk()(kid="k5", kty="OKP", crv="Ed25519", x="AA", alg="ES256") + + def test_ed25519_with_matching_algorithm_accepted(self): + self._jwk()(kid="k5", kty="OKP", crv="Ed25519", x="AA", alg="EdDSA") + + def test_algorithm_omitted_is_unconstrained(self): + # alg is optional; verifiers derive it from crv when absent. + self._jwk()(kid="k6", kty="EC", crv="P-256", x="AA", y="BB") + + def test_unrecognized_curve_is_unconstrained(self): + # The crv/kty/alg vocabularies are open (see the schema + # description); a curve outside the three well-known pairings + # carries no algorithm rule. + self._jwk()( + kid="k7", kty="EC", crv="secp256k1", x="AA", y="BB", alg="ES256K" + ) + + +@unittest.skipUnless( + HAVE_SDK, "requires the installed package (pip install -e .)" +) +class UnitScaleSemanticTest(unittest.TestCase): + """unit.json: when unit is C62, scale (if present) MUST be 0.""" + + def _unit(self): + from ucp_sdk.models.schemas.common.types.unit import Unit + + return Unit + + def test_c62_with_nonzero_scale_rejected(self): + with self.assertRaises(ValidationError): + self._unit()(unit="C62", scale=5, display_text="pieces") + + def test_c62_with_zero_scale_accepted(self): + unit = self._unit()(unit="C62", scale=0, display_text="pieces") + self.assertEqual(unit.scale, 0) + + def test_c62_with_scale_omitted_defaults_to_zero(self): + # scale defaults to 0, which already satisfies the C62 pin. + unit = self._unit()(unit="C62", display_text="pieces") + self.assertEqual(unit.scale, 0) + + def test_other_unit_with_nonzero_scale_accepted(self): + # The pin is C62-specific; any other unit is unconstrained. + unit = self._unit()(unit="GRM", scale=3, display_text="grams") + self.assertEqual(unit.scale, 3) + + if __name__ == "__main__": unittest.main()