added system test for training rig

tests/system_tests/common.py

@@ -0,0 +1,81 @@
+import os
+
+import pytest
+
+from blueapi.client.client import BlueapiClient
+from blueapi.client.event_bus import AnyEvent
+from blueapi.core.bluesky_types import DataEvent
+from blueapi.worker.event import TaskStatus, WorkerEvent, WorkerState
+
+BEAMLINE = os.environ.get("BEAMLINE", "")
+
+DISABLE_SIDE_EFFECTS = bool(os.environ.get("DISABLE_SIDE_EFFECTS", 0))
+DISABLE_SIDE_EFFECTS_MESSAGE = """
+    This test would cause side effects on the beamline, it has been disabled
+    so as not to interfere with operation. To run tests that may interfere with
+    the beamline export DISABLE_SIDE_EFFECTS=0
+    """
+disable_side_effects = pytest.mark.skipif(
+    DISABLE_SIDE_EFFECTS, reason=DISABLE_SIDE_EFFECTS_MESSAGE
+)
+
+REQUIRES_AUTH = bool(os.environ.get("REQUIRES_AUTH", 0))
+REQUIRES_AUTH_MESSAGE = """
+Authentication credentials are required to run this test.
+The test has been skipped because authentication is currently disabled.
+For more details, see: https://github.com/DiamondLightSource/blueapi/issues/676.
+To enable and execute these tests, set `REQUIRES_AUTH=1` and provide valid credentials.
+"""
+requires_auth = pytest.mark.skipif(not REQUIRES_AUTH, reason=REQUIRES_AUTH_MESSAGE)
+
+# Mark for beamline-specific tests
+BEAMLINE_SPECIFIC_MESSAGE = """
+    This test is beamline-specific but no beamline has been set.
+    Set the BEAMLINE environment variable to enable this test.
+    """
+beamline_specific_test = pytest.mark.skipif(
+    not BEAMLINE, reason=BEAMLINE_SPECIFIC_MESSAGE
+)
+
+
+def clean_existing_tasks(client: BlueapiClient) -> None:
+    for task in client.get_all_tasks().tasks:
+        client.clear_task(task.task_id)
+
+
+def check_all_events(all_events: list[AnyEvent]):
+    assert isinstance(all_events[0], WorkerEvent) and all_events[0].task_status
+    task_id = all_events[0].task_status.task_id
+    # First event is WorkerEvent
+    assert all_events[0] == WorkerEvent(
+        state=WorkerState.RUNNING,
+        task_status=TaskStatus(
+            task_id=task_id,
+            task_complete=False,
+            task_failed=False,
+        ),
+    )
+
+    assert all(
+        isinstance(event, DataEvent) for event in all_events[1:-2]
+    ), "Middle elements must be DataEvents."
+
+    # Last 2 events are WorkerEvent
+    assert all_events[-2:] == [
+        WorkerEvent(
+            state=WorkerState.IDLE,
+            task_status=TaskStatus(
+                task_id=task_id,
+                task_complete=False,
+                task_failed=False,
+            ),
+        ),
+        WorkerEvent(
+            state=WorkerState.IDLE,
+            task_status=TaskStatus(
+                task_id=task_id,
+                task_complete=True,
+                task_failed=False,
+            ),
+        ),
+    ]

tests/system_tests/conftest.py

@@ -0,0 +1,154 @@
+import inspect
+import time
+from pathlib import Path
+
+import pytest
+from bluesky_stomp.models import BasicAuthentication
+from pydantic import TypeAdapter
+
+from blueapi.client.client import BlueapiClient
+from blueapi.config import ApplicationConfig, RestConfig, StompConfig
+from blueapi.service.model import DeviceResponse, PlanResponse
+from blueapi.worker.task import Task
+from tests.system_tests.utils import BEAMLINE
+
+# Step 1: Ensure a message bus that supports stomp is running and available:
+#   src/script/start_rabbitmq.sh
+#
+# Step 2: Start the BlueAPI server with valid configuration:
+#   blueapi -c tests/unit_tests/example_yaml/valid_stomp_config.yaml serve
+#
+# Step 3: Run the system tests using tox:
+#   tox -e system-test
+
+_DATA_PATH = Path(__file__).parent / "expected_data"
+
+
+@pytest.fixture(scope="module", autouse=True)
+def wait_for_server():
+    client = BlueapiClient.from_config(config=ApplicationConfig())
+    for _ in range(20):
+        try:
+            client.get_environment()
+            return
+        except ConnectionError:
+            ...
+        time.sleep(0.5)
+    raise TimeoutError("No connection to the blueapi server")
+
+
+@pytest.fixture
+def expected_plans() -> PlanResponse:
+    file_name = "plans.json" if not BEAMLINE else f"plans_{BEAMLINE}.json"
+    return TypeAdapter(PlanResponse).validate_json((_DATA_PATH / file_name).read_text())
+
+
+@pytest.fixture
+def expected_devices() -> DeviceResponse:
+    file_name = "devices.json" if not BEAMLINE else f"devices_{BEAMLINE}.json"
+    return TypeAdapter(DeviceResponse).validate_json(
+        (_DATA_PATH / file_name).read_text()
+    )
+
+
+@pytest.fixture
+def blueapi_client_get_methods() -> list[str]:
+    # Get a list of methods that take only one argument (self)
+    # This will currently return
+    # ['get_plans', 'get_devices', 'get_state', 'get_all_tasks',
+    # 'get_active_task','get_environment','resume', 'stop','get_oidc_config']
+    return [
+        method
+        for method in BlueapiClient.__dict__
+        if callable(getattr(BlueapiClient, method))
+        and not method.startswith("__")
+        and len(inspect.signature(getattr(BlueapiClient, method)).parameters) == 1
+        and "self" in inspect.signature(getattr(BlueapiClient, method)).parameters
+    ]
+
+
+@pytest.fixture
+def task_definition() -> dict[str, Task]:
+    return {
+        "simple_plan": Task(name="sleep", params={"time": 0.0}),
+        "long_plan": Task(name="sleep", params={"time": 1.0}),
+        "spec_scan": Task(
+            name="spec_scan",
+            params={
+                "detectors": ["det"],
+                "spec": {
+                    "axis": "sample_stage.x",
+                    "start": 1.0,
+                    "stop": 10.0,
+                    "num": 10,
+                    "type": "Line",
+                },
+            },
+        ),
+    }
+
+
+@pytest.fixture
+def config() -> ApplicationConfig:
+    if BEAMLINE == "p46":
+        return ApplicationConfig(
+            api=RestConfig(
+                host="p46-blueapi.diamond.ac.uk", port=443, protocol="https"
+            ),
+        )
+    else:
+        return ApplicationConfig()
+
+
+@pytest.fixture
+def config_without_auth(tmp_path: Path) -> ApplicationConfig:
+    if BEAMLINE == "p46":
+        return ApplicationConfig(
+            stomp=StompConfig(
+                host="172.23.168.198",
+                auth=BasicAuthentication(username="guest", password="guest"),  # type: ignore
+            ),
+            api=RestConfig(
+                host="p46-blueapi.diamond.ac.uk", port=443, protocol="https"
+            ),
+            auth_token_path=tmp_path,
+        )
+    else:
+        return ApplicationConfig(auth_token_path=tmp_path)
+
+
+@pytest.fixture
+def config_with_stomp() -> ApplicationConfig:
+    if BEAMLINE == "p46":
+        return ApplicationConfig(
+            stomp=StompConfig(
+                host="172.23.168.198",
+                auth=BasicAuthentication(username="p46", password="64p"),  # type: ignore
+            ),
+            api=RestConfig(
+                host="p46-blueapi.diamond.ac.uk", port=443, protocol="https"
+            ),
+        )
+    else:
+        return ApplicationConfig(
+            stomp=StompConfig(
+                host="localhost",
+                auth=BasicAuthentication(username="guest", password="guest"),  # type: ignore
+            )
+        )
+
+
+# This client will have auth enabled if it finds cached valid token
+@pytest.fixture
+def client(config) -> BlueapiClient:
+    return BlueapiClient.from_config(config=config)
+
+
+@pytest.fixture
+def client_without_auth(config_without_auth) -> BlueapiClient:
+    return BlueapiClient.from_config(config=config_without_auth)
+
+
+@pytest.fixture
+def client_with_stomp(config_with_stomp) -> BlueapiClient:
+    return BlueapiClient.from_config(config=config_with_stomp)