test(integration): poll for query-exchange visibility after POST (orders, order_groups)

tests/integration/helpers.py

@@ -5,15 +5,56 @@
 import asyncio
 import functools
 import logging
-from collections.abc import Callable
+import time
+from collections.abc import Awaitable, Callable
 from decimal import ROUND_HALF_UP, Decimal
 from typing import Any, TypeVar
 
 import pytest
 from websockets.exceptions import ConnectionClosed
 
 from kalshi.client import KalshiClient
-from kalshi.errors import KalshiConnectionError, KalshiError
+from kalshi.errors import KalshiConnectionError, KalshiError, KalshiNotFoundError
+
+
+def wait_for_resource[T](
+    fetch: Callable[[], T],
+    *,
+    timeout: float = 15.0,
+    interval: float = 0.5,
+) -> T:
+    """Poll ``fetch()`` until it stops raising KalshiNotFoundError.
+
+    Demo's ``query-exchange`` replica lags writes by ~10 seconds: orders
+    and order_groups created via POST are not immediately readable via
+    GET-by-id. Use this anywhere a test writes a resource and then
+    immediately reads it back.
+    """
+    deadline = time.monotonic() + timeout
+    while True:
+        try:
+            return fetch()
+        except KalshiNotFoundError:
+            if time.monotonic() >= deadline:
+                raise
+            time.sleep(interval)
+
+
+async def await_resource[T](
+    fetch: Callable[[], Awaitable[T]],
+    *,
+    timeout: float = 15.0,
+    interval: float = 0.5,
+) -> T:
+    """Async counterpart of :func:`wait_for_resource` — same ~10s demo lag rationale."""
+    deadline = time.monotonic() + timeout
+    while True:
+        try:
+            return await fetch()
+        except KalshiNotFoundError:
+            if time.monotonic() >= deadline:
+                raise
+            await asyncio.sleep(interval)
 
 logger = logging.getLogger(__name__)
 

tests/integration/test_order_groups.py

@@ -2,7 +2,6 @@
 
 from __future__ import annotations
 
-import asyncio
 import logging
 from collections.abc import Iterator
 
@@ -19,6 +18,7 @@
 from tests.integration.assertions import assert_model_fields
 from tests.integration.conftest import skip_if_low_balance
 from tests.integration.coverage_harness import register
+from tests.integration.helpers import await_resource, wait_for_resource
 
 logger = logging.getLogger(__name__)
 
@@ -69,15 +69,20 @@ def test_list(self, sync_client: KalshiClient) -> None:
     def test_create_and_get(
         self, sync_client: KalshiClient, ephemeral_group: str,
     ) -> None:
-        resp = sync_client.order_groups.get(ephemeral_group)
+        # Demo query-exchange lags writes by ~10s after create.
+        resp = wait_for_resource(
+            lambda: sync_client.order_groups.get(ephemeral_group),
+        )
         assert isinstance(resp, GetOrderGroupResponse)
         assert_model_fields(resp)
 
     def test_update_limit(
         self, sync_client: KalshiClient, ephemeral_group: str,
     ) -> None:
         sync_client.order_groups.update_limit(ephemeral_group, contracts_limit=5)
-        resp = sync_client.order_groups.get(ephemeral_group)
+        resp = wait_for_resource(
+            lambda: sync_client.order_groups.get(ephemeral_group),
+        )
         # Server may normalize the limit — just assert round-trip works
         assert resp.contracts_limit is not None
 
@@ -117,9 +122,10 @@ async def test_create_get_delete(
         resp = await async_client.order_groups.create(contracts_limit=1)
         assert isinstance(resp, CreateOrderGroupResponse)
         try:
-            # Demo server may need a moment to propagate the new group.
-            await asyncio.sleep(0.5)
-            got = await async_client.order_groups.get(resp.order_group_id)
+            # Demo query-exchange lags writes by ~10s after create.
+            got = await await_resource(
+                lambda: async_client.order_groups.get(resp.order_group_id),
+            )
             assert isinstance(got, GetOrderGroupResponse)
         finally:
             await async_client.order_groups.delete(resp.order_group_id)

tests/integration/test_orders.py

@@ -15,7 +15,7 @@
 from tests.integration.assertions import assert_model_fields
 from tests.integration.conftest import skip_if_low_balance
 from tests.integration.coverage_harness import register
-from tests.integration.helpers import fill_guarantee
+from tests.integration.helpers import await_resource, fill_guarantee, wait_for_resource
 
 logger = logging.getLogger(__name__)
 
@@ -92,9 +92,10 @@ def test_order_fill_lifecycle(
         )
 
         # Check order statuses — on demo, self-trade prevention may
-        # cancel one side immediately
-        buy_order = sync_client.orders.get(buy_id)
-        sell_order = sync_client.orders.get(sell_id)
+        # cancel one side immediately. Demo's query-exchange replica
+        # lags writes by ~10s, so poll for visibility.
+        buy_order = wait_for_resource(lambda: sync_client.orders.get(buy_id))
+        sell_order = wait_for_resource(lambda: sync_client.orders.get(sell_id))
         assert_model_fields(buy_order)
         assert_model_fields(sell_order)
 
@@ -151,7 +152,9 @@ def test_create_get_cancel(
         assert order.order_id
 
         try:
-            retrieved = sync_client.orders.get(order.order_id)
+            retrieved = wait_for_resource(
+                lambda: sync_client.orders.get(order.order_id),
+            )
             assert isinstance(retrieved, Order)
             assert_model_fields(retrieved)
             assert retrieved.order_id == order.order_id
@@ -262,7 +265,9 @@ async def test_create_get_cancel(
         assert order.order_id
 
         try:
-            retrieved = await async_client.orders.get(order.order_id)
+            retrieved = await await_resource(
+                lambda: async_client.orders.get(order.order_id),
+            )
             assert isinstance(retrieved, Order)
             assert_model_fields(retrieved)
             assert retrieved.order_id == order.order_id