DID-9477: [QA-SDK-PHP] PHP server SDK release contract

codegen/didit_codegen/model.py

@@ -529,11 +529,28 @@ def _merge_body_schemas(doc: Document, content: dict[str, Any]) -> tuple[dict[st
     required_sets = []
     for raw in schemas:
         s = doc.deref(raw)
+        if isinstance(s, dict) and ("oneOf" in s or "anyOf" in s):
+            # A content type that accepts more than one shape (an object or a
+            # bare array, say) contributes the properties of every branch; a
+            # key is required only when every branch requires it.
+            branches = s.get("oneOf") if "oneOf" in s else s.get("anyOf")
+            flat: dict[str, Any] = {}
+            branch_required: list[set[str]] = []
+            for branch in branches if isinstance(branches, list) else []:
+                b = doc.deref(branch)
+                if not isinstance(b, dict):
+                    continue
+                for key, prop in b.get("properties", {}).items():
+                    flat.setdefault(key, prop)
+                branch_required.append(set(b.get("required", [])))
+            s = {"properties": flat,
+                 "required": sorted(set.intersection(*branch_required)) if branch_required else [],
+                 "description": s.get("description", "")}
         if isinstance(s, dict) and isinstance(s.get("allOf"), list):
             parts = [doc.deref(x) for x in s["allOf"]]
             s = {"properties": {k: v for p in parts for k, v in p.get("properties", {}).items()},
                  "required": [r for p in parts for r in p.get("required", [])], "description": s.get("description", "")}
-        for key, prop in s.get("properties", {}).items():
+        for key, prop in (s.get("properties", {}) if isinstance(s, dict) else {}).items():
             existing = props.get(key)
             if existing is None:
                 props[key] = prop
@@ -544,10 +561,11 @@ def _merge_body_schemas(doc: Document, content: dict[str, Any]) -> tuple[dict[st
                     props[key] = {"oneOf": [existing, prop], "description": ea.get("description", "")}
                 elif ea.get("format") == "binary" and eb.get("format") != "binary":
                     props[key] = {"oneOf": [existing, prop], "description": ea.get("description", "")}
-        required_sets.append(set(s.get("required", [])))
+        required_sets.append(set(s.get("required", [])) if isinstance(s, dict) else set())
     required = set.intersection(*required_sets) if required_sets else set()
     first = doc.deref(schemas[0])
-    return {"type": "object", "properties": props, "required": [k for k in props if k in required], "description": first.get("description", "")}, None
+    description = first.get("description", "") if isinstance(first, dict) else ""
+    return {"type": "object", "properties": props, "required": [k for k in props if k in required], "description": description}, None
 
 
 def _response(reg: Registry, o: Operation, prefix: str) -> Response:

codegen/tests/test_model.py

@@ -6,7 +6,7 @@
 import unittest
 from pathlib import Path
 
-from helpers import write_sessions
+from helpers import document, naming, ok, write_openapi, write_sessions
 from didit_codegen import model, php, spec
 from didit_codegen.model import AnyT, ArrayT, FreeObjT, MapT, Prop, Scalar, Shape, ShapeT
 
@@ -96,5 +96,57 @@ def test_hint_reaches_the_generated_operation(self) -> None:
         self.assertIn("jsonHints: ['contact_details' => ['o'], 'metadata' => ['o']]", source)
 
 
+class MultiContentOneOfBodyTest(unittest.TestCase):
+    """A body that is a union in one content type and has another form (DESIGN 5.1)."""
+
+    @classmethod
+    def setUpClass(cls) -> None:
+        row = {"$ref": "#/components/schemas/WalletAddressImportRow"}
+        paths = {
+            "/v3/travel-rule/wallet-addresses/import/": {
+                "post": {
+                    "summary": "Import wallet addresses",
+                    "requestBody": {
+                        "required": True,
+                        "content": {
+                            "application/json": {
+                                "schema": {"oneOf": [
+                                    {"type": "object", "required": ["entries"],
+                                     "properties": {"entries": {"type": "array", "minItems": 1, "items": row}}},
+                                    {"type": "array", "minItems": 1, "items": row},
+                                ]},
+                            },
+                            "multipart/form-data": {
+                                "schema": {"type": "object", "required": ["file"],
+                                           "properties": {"file": {"type": "string", "format": "binary"}}},
+                            },
+                        },
+                    },
+                    "responses": ok({"type": "object", "properties": {"total": {"type": "integer"}}}),
+                }
+            }
+        }
+        schemas = {"WalletAddressImportRow": {"type": "object", "properties": {"address": {"type": "string"}}}}
+        overlay = {"POST /v3/travel-rule/wallet-addresses/import/": naming("travel_rule.wallet_addresses", "bulk_import")}
+        with tempfile.TemporaryDirectory() as tmp:
+            write_openapi(Path(tmp), document(paths, schemas), overlay)
+            s = spec.load(Path(tmp))
+        cls.reg = model.Registry()
+        cls.op = next(o for o in model.build(cls.reg, s.server_operations))
+
+    def test_every_branch_of_the_union_reaches_the_params(self) -> None:
+        body = {p.wire: p for p in self.op.body.props}
+        self.assertEqual(sorted(body), ["entries", "file"])
+        self.assertIsInstance(body["entries"].type, ArrayT)
+        # Neither form is required of every caller: the JSON union keeps no
+        # key across its branches, and the multipart file is only one form.
+        self.assertFalse(body["entries"].required)
+        self.assertFalse(body["file"].required)
+
+    def test_the_generated_params_alias_lists_the_union_field(self) -> None:
+        source = php.emit_params(list(self.reg.shapes.values()), [])
+        self.assertIn("entries?: list<WalletAddressImportRow>,", source)
+
+
 if __name__ == "__main__":
     unittest.main()