Assert Python workflow list parity

Assert Python workflow list parity

Author: durable-workflow-ops

src/durable_workflow/client.py

@@ -25,7 +25,7 @@
 from importlib.metadata import PackageNotFoundError
 from importlib.metadata import version as _pkg_version
 from typing import Any
-from urllib.parse import quote
+from urllib.parse import quote, urlencode
 
 import httpx
 
@@ -992,7 +992,7 @@ async def list_workflows(
         if next_page_token is not None:
             params["next_page_token"] = next_page_token
 
-        qs = "&".join(f"{k}={v}" for k, v in params.items())
+        qs = urlencode(params)
         path = f"/workflows?{qs}" if qs else "/workflows"
         data = await self._request("GET", path)
         items = data.get("workflows", [])

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

@@ -0,0 +1,69 @@
+{
+  "schema": "durable-workflow.polyglot.control-plane-request-fixture",
+  "version": 1,
+  "operation": "workflow.list",
+  "request": {
+    "method": "GET",
+    "path": "/workflows"
+  },
+  "semantic_body": {
+    "workflow_type": "orders.process",
+    "status": "running",
+    "query": "CustomerId=\"42\"",
+    "page_size": 25,
+    "workflow_ids": [
+      "wf-polyglot-list-1",
+      "wf-polyglot-list-2"
+    ],
+    "next_page_token": "next-polyglot-page"
+  },
+  "response_body": {
+    "workflows": [
+      {
+        "workflow_id": "wf-polyglot-list-1",
+        "run_id": "run-polyglot-list-1",
+        "workflow_type": "orders.process",
+        "business_key": "order-42",
+        "status": "running",
+        "task_queue": "orders",
+        "started_at": "2026-04-22T05:00:00Z",
+        "closed_at": null
+      },
+      {
+        "workflow_id": "wf-polyglot-list-2",
+        "run_id": "run-polyglot-list-2",
+        "workflow_type": "orders.process",
+        "business_key": "order-43",
+        "status": "running",
+        "task_queue": "orders",
+        "started_at": "2026-04-22T05:01:00Z",
+        "closed_at": null
+      }
+    ],
+    "next_page_token": "next-polyglot-page"
+  },
+  "cli": {
+    "argv": {
+      "--type": "orders.process",
+      "--status": "running",
+      "--query": "CustomerId=\"42\"",
+      "--limit": "25",
+      "--json": true
+    },
+    "expected_query": {
+      "workflow_type": "orders.process",
+      "status": "running",
+      "query": "CustomerId=\"42\"",
+      "page_size": 25
+    }
+  },
+  "sdk_python": {
+    "method": "list_workflows",
+    "args": {
+      "workflow_type": "orders.process",
+      "status": "running",
+      "query": "CustomerId=\"42\"",
+      "page_size": 25
+    }
+  }
+}

tests/test_client.py

@@ -4,6 +4,7 @@
 import logging
 from pathlib import Path
 from unittest.mock import AsyncMock, patch
+from urllib.parse import parse_qs, urlsplit
 
 import httpx
 import pytest
@@ -686,6 +687,32 @@ async def test_list(self, client: Client) -> None:
             assert result.next_page_token == "abc"
             assert result.executions[0].workflow_id == "wf-1"
 
+    @pytest.mark.asyncio
+    async def test_list_request_matches_polyglot_fixture(self, client: Client) -> None:
+        fixture_path = Path(__file__).parent / "fixtures" / "control-plane" / "workflow-list-parity.json"
+        fixture = json.loads(fixture_path.read_text())
+        sdk_args = fixture["sdk_python"]["args"]
+
+        resp = _mock_response(200, fixture["response_body"])
+
+        with patch.object(client._http, "request", new_callable=AsyncMock, return_value=resp) as mock:
+            result = await client.list_workflows(**sdk_args)
+
+        call_args = mock.call_args
+        assert call_args.args[0] == fixture["request"]["method"]
+
+        url = urlsplit(call_args.args[1])
+        assert url.path == f"/api{fixture['request']['path']}"
+        assert parse_qs(url.query) == {
+            "workflow_type": [fixture["semantic_body"]["workflow_type"]],
+            "status": [fixture["semantic_body"]["status"]],
+            "query": [fixture["semantic_body"]["query"]],
+            "page_size": [str(fixture["semantic_body"]["page_size"])],
+        }
+
+        assert [execution.workflow_id for execution in result.executions] == fixture["semantic_body"]["workflow_ids"]
+        assert result.next_page_token == fixture["semantic_body"]["next_page_token"]
+
 
 class TestTaskQueues:
     @pytest.mark.asyncio