Add tests covering the performance enhancements

These cover multiple subscribers to the same PV, unsubscribing, and
automatic camel casing.

Also remove unnecessary run_ioc function.

tests/conftest.py

@@ -6,6 +6,7 @@
 import sys
 from datetime import datetime, timedelta
 from pathlib import Path
+from typing import cast
 from unittest.mock import ANY
 
 import pytest
@@ -28,6 +29,8 @@
 )
 
 from coniql.app import create_app
+from coniql.caplugin import CAPlugin, CASubscriptionManager
+from coniql.strawberry_schema import store_global
 
 SOFT_RECORDS = str(Path(__file__).parent / "soft_records.db")
 
@@ -58,7 +61,7 @@ def event_loop():
     loop.close()
 
 
-def ioc_creator(pv_prefix=PV_PREFIX):
+def ioc_creator(pv_prefix=PV_PREFIX) -> subprocess.Popen:
     process = subprocess.Popen(
         [
             sys.executable,
@@ -78,10 +81,6 @@ def ioc_creator(pv_prefix=PV_PREFIX):
     return process
 
 
-def run_ioc(process):
-    yield process
-
-
 def ioc_cleanup(process):
     purge_channel_caches()
     try:
@@ -94,7 +93,7 @@ def ioc_cleanup(process):
 @pytest.fixture(scope="module")
 def ioc():
     process = ioc_creator()
-    yield run_ioc(process)
+    yield process
     ioc_cleanup(process)
 
 
@@ -415,3 +414,13 @@ def get_ticking_subscription_result(startVal):
 def subscription_data(request):
     """Fixture for the possible subscription types"""
     return request.param
+
+
+@pytest.fixture(autouse=True)
+def clear_subscription_manager() -> None:
+    """Reset the CASubscriptionManager inside the CAPlugin
+
+    This ensures there's no record of PVs between tests"""
+
+    ca_plugin: CAPlugin = cast(CAPlugin, store_global.plugins["ca"])
+    ca_plugin.subscription_manager = CASubscriptionManager()

tests/test_aiohttp.py

@@ -37,7 +37,6 @@
     longout_subscription_result,
     nan_get_query,
     nan_get_query_result,
-    run_ioc,
 )
 
 
@@ -82,7 +81,6 @@ async def test_aiohttp_client_put_pv(
 async def test_subscribe_disconnect(client: TestClient):
     pv_prefix = PV_PREFIX + "EXTRA:"
     ioc_process = ioc_creator(pv_prefix)
-    run_ioc(ioc_process)
     query = get_longout_subscription_query(pv_prefix)
     results: List[Dict[str, Any]] = []
     async with client.ws_connect(

tests/test_caplugin.py

@@ -1,11 +1,14 @@
 import asyncio
 from subprocess import Popen
-from typing import Any, AsyncIterator, Dict, List, Optional
+from typing import Any, AsyncIterator, Dict, List, Optional, cast
 
 import pytest
+from aioca import Subscription
 from strawberry import Schema
 
 from coniql.app import create_schema
+from coniql.caplugin import CAPlugin
+from coniql.strawberry_schema import store_global
 
 from .conftest import (
     PV_PREFIX,
@@ -29,7 +32,6 @@
     longout_subscription_result,
     nan_get_query,
     nan_get_query_result,
-    run_ioc,
     ticking_subscription_query,
 )
 
@@ -82,7 +84,6 @@ async def test_schema_put_pv(
 async def test_subscribe_disconnect(schema: Schema):
     pv_prefix = PV_PREFIX + "EXTRA:"
     ioc_process = ioc_creator(pv_prefix)
-    run_ioc(ioc_process)
     query = get_longout_subscription_query(pv_prefix)
     results: List[Dict[str, Any]] = []
     resp = await schema.subscribe(query)
@@ -126,3 +127,85 @@ async def test_subscribe_ticking(ioc: Popen, schema: Schema):
     subscription_result = get_ticking_subscription_result(startVal)
     for i in range(3):
         assert results[i] == subscription_result[i]
+
+
+@pytest.mark.asyncio
+async def test_subscribe_multiple(schema: Schema):
+    """Test that multiple subscriptions to the same PV all receive the right data"""
+
+    pv_prefix = PV_PREFIX + "MULTIPLE:"
+    ioc_process = ioc_creator(pv_prefix)
+
+    query = get_longout_subscription_query(pv_prefix)
+
+    responses = await asyncio.gather(
+        schema.subscribe(query), schema.subscribe(query), schema.subscribe(query)
+    )
+
+    # Check initial response is correct for all
+    for resp in responses:
+        results: List[Dict[str, Any]] = []
+        assert isinstance(resp, AsyncIterator)
+        async for result in resp:
+            assert result.errors is None
+            assert result.data
+            results.append(result.data)
+            break
+        assert len(results) == 1
+        assert results[0] == longout_subscription_result[0]
+
+    ioc_process.communicate("exit()")
+
+    # And check they all get the disconnect message
+    for resp in responses:
+        results = []
+        assert isinstance(resp, AsyncIterator)
+        async for result in resp:
+            assert result.errors is None
+            assert result.data
+            results.append(result.data)
+            break
+        assert len(results) == 1
+        assert results[0] == longout_subscription_result[1]
+
+
+async def subscribe_task_wrapper(schema: Schema, query: str):
+    resp = await schema.subscribe(query)
+    assert isinstance(resp, AsyncIterator)
+    async for _ in resp:
+        pass
+
+
+@pytest.mark.parametrize("num_subscribers", [1, 5])
+async def test_subscribe_unsubscribe(ioc: Popen, schema: Schema, num_subscribers: int):
+    """Test that cancelling a subscription correctly closes channel monitors"""
+
+    # Create subscription(s)
+    query = get_longout_subscription_query(PV_PREFIX)
+    tasks: List[asyncio.Task] = []
+    for _ in range(num_subscribers):
+        tasks.append(asyncio.create_task(subscribe_task_wrapper(schema, query)))
+
+    await asyncio.sleep(1)
+
+    # Check the subscription is logged in the manager
+    ca_plugin: CAPlugin = cast(CAPlugin, store_global.plugins["ca"])
+    pvs = ca_plugin.subscription_manager.pvs
+    assert len(pvs.keys()) == 1
+    for pv in pvs.values():
+        assert pv.meta_monitor.state == Subscription.OPEN
+        assert pv.time_monitor.state == Subscription.OPEN
+        assert pv.subscribers == num_subscribers
+
+    # Close subscriptions
+    for task in tasks:
+        task.cancel()
+
+    await asyncio.sleep(1)
+
+    # Check subscription has been closed
+    pvs = ca_plugin.subscription_manager.pvs
+    for pv in pvs.values():
+        assert pv.meta_monitor.state == Subscription.CLOSED
+        assert pv.time_monitor.state == Subscription.CLOSED
+        assert pv.subscribers == 0

tests/test_coniql.py

@@ -189,3 +189,26 @@ async def test_get_sim_sinewave(schema: Schema):
 def test_cli_version():
     cmd = [sys.executable, "-m", "coniql", "--version"]
     assert subprocess.check_output(cmd).decode().strip() == __version__
+
+
+def test_schema_contains_no_snake_case(schema: Schema):
+    """Test that the Schema contains no snake_case members which may need to be
+    converted to camelCase (as per GraphQL convention).
+
+    Currently we set auto_camel_case=False when constructing the Schema, as currently
+    there are no variables that need to be converted.
+
+    If this test fails due to additions to the schema, this can probably be deleted
+    (as long as we also re-enable auto_camel_case)"""
+
+    introspected_schema = schema.introspect()
+
+    for type in introspected_schema["__schema"]["types"]:
+        name = type["name"]
+        # Some inbuilt types with leading doule underscores exist, e.g. __Schema, __Type
+        if name[0:2] != "__":
+            assert "_" not in name
+
+        if type["fields"] is not None:
+            for field in type["fields"]:
+                assert "_" not in field["name"]