Assert Python workflow start against polyglot parity fixture

Assert Python workflow start parity fixture

Author: durable-workflow-ops

tests/fixtures/control-plane/workflow-start-parity.json

@@ -0,0 +1,115 @@
+{
+  "schema": "durable-workflow.polyglot.control-plane-request-fixture",
+  "version": 1,
+  "operation": "workflow.start",
+  "request": {
+    "method": "POST",
+    "path": "/workflows"
+  },
+  "semantic_body": {
+    "workflow_type": "orders.process",
+    "workflow_id": "wf-polyglot-231",
+    "task_queue": "orders",
+    "input": [
+      {
+        "order_id": 42,
+        "priority": "gold"
+      }
+    ],
+    "memo": {
+      "source": "polyglot-fixture"
+    },
+    "search_attributes": {
+      "CustomerId": "cust-42",
+      "Tier": "gold"
+    },
+    "execution_timeout_seconds": 300,
+    "run_timeout_seconds": 120,
+    "duplicate_policy": "use-existing"
+  },
+  "cli": {
+    "argv": {
+      "--type": "orders.process",
+      "--workflow-id": "wf-polyglot-231",
+      "--task-queue": "orders",
+      "--input": "[{\"order_id\":42,\"priority\":\"gold\"}]",
+      "--memo": "{\"source\":\"polyglot-fixture\"}",
+      "--search-attr": [
+        "CustomerId=cust-42",
+        "Tier=gold"
+      ],
+      "--execution-timeout": "300",
+      "--run-timeout": "120",
+      "--duplicate-policy": "use-existing"
+    },
+    "expected_body": {
+      "workflow_type": "orders.process",
+      "workflow_id": "wf-polyglot-231",
+      "task_queue": "orders",
+      "duplicate_policy": "use-existing",
+      "input": [
+        {
+          "order_id": 42,
+          "priority": "gold"
+        }
+      ],
+      "memo": {
+        "source": "polyglot-fixture"
+      },
+      "execution_timeout_seconds": 300,
+      "run_timeout_seconds": 120,
+      "search_attributes": {
+        "CustomerId": "cust-42",
+        "Tier": "gold"
+      }
+    }
+  },
+  "sdk_python": {
+    "kwargs": {
+      "workflow_type": "orders.process",
+      "workflow_id": "wf-polyglot-231",
+      "task_queue": "orders",
+      "input": [
+        {
+          "order_id": 42,
+          "priority": "gold"
+        }
+      ],
+      "memo": {
+        "source": "polyglot-fixture"
+      },
+      "search_attributes": {
+        "CustomerId": "cust-42",
+        "Tier": "gold"
+      },
+      "execution_timeout_seconds": 300,
+      "run_timeout_seconds": 120,
+      "duplicate_policy": "use-existing"
+    },
+    "expected_body": {
+      "workflow_type": "orders.process",
+      "workflow_id": "wf-polyglot-231",
+      "task_queue": "orders",
+      "duplicate_policy": "use-existing",
+      "memo": {
+        "source": "polyglot-fixture"
+      },
+      "search_attributes": {
+        "CustomerId": "cust-42",
+        "Tier": "gold"
+      },
+      "execution_timeout_seconds": 300,
+      "run_timeout_seconds": 120
+    },
+    "payload_envelope": {
+      "field": "input",
+      "codec": "avro",
+      "decoded": [
+        {
+          "order_id": 42,
+          "priority": "gold"
+        }
+      ]
+    }
+  }
+}

tests/test_client.py

@@ -2,6 +2,7 @@
 
 import json
 import logging
+from pathlib import Path
 from unittest.mock import AsyncMock, patch
 
 import httpx
@@ -102,6 +103,43 @@ async def test_success(self, client: Client) -> None:
             assert body["input"]["codec"] == "avro"
             assert serializer.decode(body["input"]["blob"], codec="avro") == ["hello"]
 
+    @pytest.mark.asyncio
+    async def test_start_request_matches_polyglot_fixture(self, client: Client) -> None:
+        fixture_path = Path(__file__).parent / "fixtures" / "control-plane" / "workflow-start-parity.json"
+        fixture = json.loads(fixture_path.read_text())
+        sdk = fixture["sdk_python"]
+        expected = sdk["expected_body"]
+        envelope_contract = sdk["payload_envelope"]
+
+        resp = _mock_response(201, {
+            "workflow_id": fixture["semantic_body"]["workflow_id"],
+            "run_id": "run-polyglot-231",
+            "workflow_type": fixture["semantic_body"]["workflow_type"],
+            "namespace": "ns1",
+            "status": "running",
+        })
+
+        with patch.object(client._http, "request", new_callable=AsyncMock, return_value=resp) as mock:
+            handle = await client.start_workflow(**sdk["kwargs"])
+
+        assert handle.workflow_id == fixture["semantic_body"]["workflow_id"]
+
+        call_args = mock.call_args
+        assert call_args.args[0] == fixture["request"]["method"]
+        assert call_args.args[1] == f"/api{fixture['request']['path']}"
+        body = call_args.kwargs.get("json") or call_args[1].get("json")
+
+        for field, value in expected.items():
+            assert body[field] == value
+
+        envelope = body[envelope_contract["field"]]
+        assert envelope["codec"] == envelope_contract["codec"]
+        assert serializer.decode(envelope["blob"], codec=envelope["codec"]) == envelope_contract["decoded"]
+
+        semantic = fixture["semantic_body"]
+        for field in ["workflow_type", "workflow_id", "task_queue", "memo", "search_attributes", "duplicate_policy"]:
+            assert body[field] == semantic[field]
+
     @pytest.mark.asyncio
     async def test_warns_when_start_input_approaches_payload_limit(
         self, caplog: pytest.LogCaptureFixture