graphql subscription test added

Author: dwsutherland

tests/integration/network/test_graphql.py

@@ -16,11 +16,21 @@
 
 """Test the top-level (root) GraphQL queries."""
 
+from contextlib import suppress
 import pytest
 from typing import TYPE_CHECKING
 
+from graphql import parse, MiddlewareManager
+
 from cylc.flow.id import Tokens
 from cylc.flow.network.client import WorkflowRuntimeClient
+from cylc.flow.network.schema import schema, SUB_RESOLVER_MAPPING
+from cylc.flow.network.graphql import (
+    CylcExecutionContext,
+    IgnoreFieldMiddleware,
+    instantiate_middleware,
+)
+from cylc.flow.network.graphql_subscribe import subscribe
 
 if TYPE_CHECKING:
     from cylc.flow.scheduler import Scheduler
@@ -362,3 +372,40 @@ async def test_jobs(harness):
     assert ret == {
         'job': {'id': f'{j_id}'}
     }
+
+
+@pytest.mark.asyncio(loop_scope="module")
+async def test_subscription(harness):
+    """Test the GraphQL subscription infrastructure.
+
+    (currently only used at UIS)
+    """
+    schd, client, w_tokens = harness
+
+    request_string = 'subscription { workflows (stripNull: true) { id } }'
+    kwargs = {
+        "variable_values": {},
+        "operation_name": None,
+        "context_value": {
+            'op_id': 1,
+            'resolvers': schd.server.resolvers,
+            'meta': {},
+        },
+        "subscribe_resolver_map": SUB_RESOLVER_MAPPING,
+        "middleware": MiddlewareManager(
+            instantiate_middleware(
+                [IgnoreFieldMiddleware]
+            )
+        ),
+        "execution_context_class": CylcExecutionContext,
+    }
+    document = parse(request_string)
+    result = await subscribe(
+        schema.graphql_schema,
+        document,
+        **kwargs
+    )
+    with suppress(GeneratorExit):
+        async for item in result:
+            assert item.data['workflows'][0]['id'] == w_tokens.id
+            await result.aclose()