Assert Python activity task parity fixtures

Author: durable-workflow-ops

tests/fixtures/control-plane/activity-complete-parity.json

@@ -0,0 +1,67 @@
+{
+  "schema": "durable-workflow.polyglot.control-plane-request-fixture",
+  "version": 1,
+  "operation": "activity.complete",
+  "request": {
+    "method": "POST",
+    "path": "/worker/activity-tasks/activity-task-231/complete"
+  },
+  "semantic_body": {
+    "task_id": "activity-task-231",
+    "activity_attempt_id": "activity-attempt-231",
+    "lease_owner": "polyglot-worker-231",
+    "result": {
+      "items_processed": 3,
+      "checkpoint": "inventory-page-7"
+    },
+    "outcome": "completed"
+  },
+  "response_body": {
+    "task_id": "activity-task-231",
+    "activity_attempt_id": "activity-attempt-231",
+    "outcome": "completed"
+  },
+  "cli": {
+    "argv": {
+      "task-id": "activity-task-231",
+      "attempt-id": "activity-attempt-231",
+      "--lease-owner": "polyglot-worker-231",
+      "--input": "{\"items_processed\":3,\"checkpoint\":\"inventory-page-7\"}",
+      "--json": true
+    },
+    "expected_body": {
+      "activity_attempt_id": "activity-attempt-231",
+      "lease_owner": "polyglot-worker-231",
+      "result": {
+        "items_processed": 3,
+        "checkpoint": "inventory-page-7"
+      }
+    }
+  },
+  "sdk_python": {
+    "method": "complete_activity_task",
+    "kwargs": {
+      "task_id": "activity-task-231",
+      "activity_attempt_id": "activity-attempt-231",
+      "lease_owner": "polyglot-worker-231",
+      "result": {
+        "items_processed": 3,
+        "checkpoint": "inventory-page-7"
+      },
+      "codec": "json",
+      "activity_name": "inventory.sync-page"
+    },
+    "expected_body": {
+      "activity_attempt_id": "activity-attempt-231",
+      "lease_owner": "polyglot-worker-231"
+    },
+    "payload_envelope": {
+      "field": "result",
+      "codec": "json",
+      "decoded": {
+        "items_processed": 3,
+        "checkpoint": "inventory-page-7"
+      }
+    }
+  }
+}

tests/fixtures/control-plane/activity-fail-parity.json

@@ -0,0 +1,54 @@
+{
+  "schema": "durable-workflow.polyglot.control-plane-request-fixture",
+  "version": 1,
+  "operation": "activity.fail",
+  "request": {
+    "method": "POST",
+    "path": "/worker/activity-tasks/activity-task-231/fail",
+    "body": {
+      "activity_attempt_id": "activity-attempt-231",
+      "lease_owner": "polyglot-worker-231",
+      "failure": {
+        "message": "Inventory service rejected the batch.",
+        "type": "UpstreamRejected",
+        "non_retryable": true
+      }
+    }
+  },
+  "semantic_body": {
+    "task_id": "activity-task-231",
+    "activity_attempt_id": "activity-attempt-231",
+    "lease_owner": "polyglot-worker-231",
+    "message": "Inventory service rejected the batch.",
+    "type": "UpstreamRejected",
+    "non_retryable": true,
+    "outcome": "failed"
+  },
+  "response_body": {
+    "task_id": "activity-task-231",
+    "activity_attempt_id": "activity-attempt-231",
+    "outcome": "failed"
+  },
+  "cli": {
+    "argv": {
+      "task-id": "activity-task-231",
+      "attempt-id": "activity-attempt-231",
+      "--lease-owner": "polyglot-worker-231",
+      "--message": "Inventory service rejected the batch.",
+      "--type": "UpstreamRejected",
+      "--non-retryable": true,
+      "--json": true
+    }
+  },
+  "sdk_python": {
+    "method": "fail_activity_task",
+    "kwargs": {
+      "task_id": "activity-task-231",
+      "activity_attempt_id": "activity-attempt-231",
+      "lease_owner": "polyglot-worker-231",
+      "message": "Inventory service rejected the batch.",
+      "failure_type": "UpstreamRejected",
+      "non_retryable": true
+    }
+  }
+}

tests/test_client.py

@@ -1336,6 +1336,41 @@ async def test_body_shape(self, client: Client) -> None:
 
 
 class TestFailActivityTask:
+    @pytest.mark.asyncio
+    async def test_complete_activity_matches_polyglot_fixture(self, client: Client) -> None:
+        fixture_path = Path(__file__).parent / "fixtures" / "control-plane" / "activity-complete-parity.json"
+        fixture = json.loads(fixture_path.read_text())
+        sdk = fixture["sdk_python"]
+        expected = sdk["expected_body"]
+
+        resp = _mock_response(200, fixture["response_body"])
+        with patch.object(client._http, "request", new_callable=AsyncMock, return_value=resp) as mock:
+            result = await client.complete_activity_task(**sdk["kwargs"])
+
+        assert result == fixture["response_body"]
+        assert mock.call_args.args[:2] == (fixture["request"]["method"], f"/api{fixture['request']['path']}")
+
+        body = mock.call_args.kwargs["json"]
+        assert body["activity_attempt_id"] == expected["activity_attempt_id"]
+        assert body["lease_owner"] == expected["lease_owner"]
+
+        envelope = body[sdk["payload_envelope"]["field"]]
+        assert envelope["codec"] == sdk["payload_envelope"]["codec"]
+        assert serializer.decode_envelope(envelope) == sdk["payload_envelope"]["decoded"]
+
+    @pytest.mark.asyncio
+    async def test_fail_activity_matches_polyglot_fixture(self, client: Client) -> None:
+        fixture_path = Path(__file__).parent / "fixtures" / "control-plane" / "activity-fail-parity.json"
+        fixture = json.loads(fixture_path.read_text())
+
+        resp = _mock_response(200, fixture["response_body"])
+        with patch.object(client._http, "request", new_callable=AsyncMock, return_value=resp) as mock:
+            result = await client.fail_activity_task(**fixture["sdk_python"]["kwargs"])
+
+        assert result == fixture["response_body"]
+        assert mock.call_args.args[:2] == (fixture["request"]["method"], f"/api{fixture['request']['path']}")
+        assert mock.call_args.kwargs["json"] == fixture["request"]["body"]
+
     @pytest.mark.asyncio
     async def test_body_shape(self, client: Client) -> None:
         resp = _mock_response(200, {"task_id": "t1", "outcome": "failed"})