Improve reliability of timing sensitive tests

Also add timeout on blocking subscriptions.

tests/conftest.py

@@ -38,6 +38,8 @@
     "base64": "AAAAAAAAAAA1XrpJDAL7PwAAAAAAAABA",
 }
 
+SUBSCRIPTION_TIMEOUT = 10
+
 
 def wait_for_ioc(ioc):
     while True:
@@ -344,6 +346,7 @@ def get_longout_subscription_query(pv_prefix):
     subscribeChannel(id: "ca://%sticking") {
         value {
             string(units: true)
+            float
         }
         display {
             precision
@@ -360,19 +363,19 @@ def get_ticking_subscription_result(startVal):
     return [
         {
             "subscribeChannel": {
-                "value": {"string": "{}0000 mm".format(startVal)},
+                "value": {"string": f"{startVal}0000 mm", "float": startVal},
                 "display": {"precision": 5, "units": "mm"},
             }
         },
         {
             "subscribeChannel": {
-                "value": {"string": "{}0000 mm".format(startVal + 1)},
+                "value": {"string": f"{startVal + 1}0000 mm", "float": startVal + 1},
                 "display": None,
             }
         },
         {
             "subscribeChannel": {
-                "value": {"string": "{}0000 mm".format(startVal + 2)},
+                "value": {"string": f"{startVal + 2}0000 mm", "float": startVal + 2},
                 "display": None,
             }
         },

tests/test_aiohttp.py

@@ -1,9 +1,9 @@
 import asyncio
+import time
 from subprocess import Popen
 from typing import Any, Dict, List, Optional
 
 import pytest
-from aioca import caget
 from aiohttp.test_utils import TestClient
 from strawberry.subscriptions import GRAPHQL_TRANSPORT_WS_PROTOCOL
 from strawberry.subscriptions.protocols.graphql_transport_ws.types import (
@@ -16,6 +16,7 @@
 
 from .conftest import (
     PV_PREFIX,
+    SUBSCRIPTION_TIMEOUT,
     base64_put_query,
     base64_put_query_result,
     check_put_timestamp,
@@ -130,16 +131,21 @@ async def test_subscribe_pv(ioc: Popen, client: TestClient, subscription_data):
         response = await ws.receive_json()
         assert response == msg_ack
         await ws.send_json(msg_send)
-        startVal = 0.0
         count = 0
+        start = time.time()
         while True:
+            if time.time() - start > SUBSCRIPTION_TIMEOUT:
+                pytest.fail("Timeout waiting for subscription data")
             if count > 2:
                 break
-            # Get the starting value in the subscription for checks later
-            if count == 0:
-                startVal = await caget(PV_PREFIX + "ticking")
-            result = await ws.receive_json()
 
+            # Set a timeout on wait for a websocket response as this blocks for the
+            # GRAPHQL_TRANSPORT_WS_PROTOCOL. A "keep alive" response is constantly
+            # sent for the GRAPHQL_WS_PROTOCOL so still need the above timeout to catch
+            # this case.
+            result = await asyncio.wait_for(
+                ws.receive_json(), timeout=SUBSCRIPTION_TIMEOUT
+            )
             if result["type"] == GQL_CONNECTION_KEEP_ALIVE:
                 continue
             results.append(result["payload"]["data"])
@@ -148,6 +154,10 @@ async def test_subscribe_pv(ioc: Popen, client: TestClient, subscription_data):
         await ws.close()
         assert ws.closed
     assert len(results) == 3
+    # Determine the starting value in the subscription
+    startSub = results[0]
+    assert startSub
+    startVal = startSub["subscribeChannel"]["value"]["float"]
     subscription_result = get_ticking_subscription_result(startVal)
     for i in range(3):
         assert results[i] == subscription_result[i]

tests/test_caplugin.py

@@ -1,14 +1,15 @@
+import asyncio
 from subprocess import Popen
 from typing import Any, AsyncIterator, Dict, List, Optional
 
 import pytest
-from aioca import caget
 from strawberry import Schema
 
 from coniql.app import create_schema
 
 from .conftest import (
     PV_PREFIX,
+    SUBSCRIPTION_TIMEOUT,
     base64_put_query,
     base64_put_query_result,
     check_put_timestamp,
@@ -108,17 +109,20 @@ async def test_subscribe_ticking(ioc: Popen, schema: Schema):
     results: List[Optional[Dict[str, Any]]] = []
     resp = await schema.subscribe(ticking_subscription_query)
     assert isinstance(resp, AsyncIterator)
-    startVal = 0.0
     count = 0
-    async for result in resp:
+    while True:
         if count > 2:
             break
-        # Get the starting value in the subscription for checks later
-        if count == 0:
-            startVal = await caget(PV_PREFIX + "ticking")
+        # Set a timeout on wait for a response as otherwise this call will
+        # block forever unless the schema.subscribe() receives data
+        result = await asyncio.wait_for(resp.__anext__(), timeout=SUBSCRIPTION_TIMEOUT)
         results.append(result.data)
         count += 1
     assert len(results) == 3
+    # Determine the starting value in the subscription
+    startSub = results[0]
+    assert startSub
+    startVal = startSub["subscribeChannel"]["value"]["float"]
     subscription_result = get_ticking_subscription_result(startVal)
     for i in range(3):
         assert results[i] == subscription_result[i]