"""Test utilities for interacting with the labthings Server.""" import tempfile import time from types import TracebackType from typing import Any, Mapping, Optional, Self, TypeVar import requests from fastapi.testclient import TestClient import labthings_fastapi as lt ThingSubclass = TypeVar("ThingSubclass", bound=lt.Thing) ACTION_RUNNING_KEYWORDS = ["idle", "pending", "running"] class LabThingsTestEnv: """A Labthings Server running in a FastAPI Test Client. This can be used to create ThingClients for testing, to to enable direct access to the server, or things on the server. It also provides high level functions to start actions without blocking the test thread. Use this as a context manager: .. code-block:: python with LabThingsTestEnv(things={"counter", CountingThing}) as env: ... """ def __init__( self, things: Mapping[str, lt.Thing | str], settings_folder: Optional[str] = None, ) -> None: """Initialise the test environment. The server and FastAPI TestClient are server are not created until this is used as a context manager: :param things: The thing configuration dictionary used to initialise the server. :param settings_folder: The settings folder to use. """ self._server: Optional[lt.ThingServer] self._test_client: Optional[TestClient] self._things_config = things self._settings_folder = settings_folder self._tmp_dir_obj: Optional[tempfile.TemporaryDirectory] = None def __enter__(self) -> Self: """Create server and run it in the TestClient.""" if self._settings_folder is None: self._tmp_dir_obj = tempfile.TemporaryDirectory() self._settings_folder = self._tmp_dir_obj.name self._server = lt.ThingServer( things=self._things_config, settings_folder=self._settings_folder ) self._test_client = TestClient(self._server.app) self._test_client.__enter__() return self def __exit__( self, exc_type: type[BaseException], exc_value: Optional[BaseException], traceback: Optional[TracebackType], ) -> None: """Close the TestClient and cleanup.""" self._test_client.__exit__(exc_type, exc_value, traceback) if self._tmp_dir_obj is not None: self._tmp_dir_obj.cleanup() @property def server(self) -> lt.ThingServer: """The LabThings Server.""" if self._server is None: raise RuntimeError("No server was started.") return self._server @property def client(self) -> TestClient: """The FastAPI TestClient running the server.""" if self._test_client is None: raise RuntimeError("No server was started.") return self._test_client def check_thing_exists(self, thing_name: str): """Raise a ValueError if no thing of with a matching name exists.""" if thing_name not in self.server.things: raise ValueError(f"No Thing named {thing_name}") def get_thing_by_name(self, thing_name: str) -> lt.Thing: """Get a Thing from the server by name. :param thing_name: The name of the thing to on the server. :return: The Thing with the specified name. """ self.check_thing_exists(thing_name) return self.server.things[thing_name] def get_thing_by_type(self, thing_class: type[ThingSubclass]) -> ThingSubclass: """Get a thing by type. :param thing_class: The subclass of thing to match. :return: The Thing that matches the subclass. :raises RuntimeError: If there are multiple things of the same type, or no matching thing. If there are multiple things of this type use ``get_thing_by_name`` or ``get_all_things_by_type``. """ matching = self.get_all_things_by_type(thing_class) n_things = len(matching) if n_things == 0: raise RuntimeError(f"No Thing of type {thing_class} on this server.") if n_things > 1: raise RuntimeError( f"Cannot get Thing by type as there are {n_things} of type " f"{thing_class} on this server." ) return next(iter(matching.values())) def get_all_things_by_type( self, thing_class: type[ThingSubclass] ) -> Mapping[str, ThingSubclass]: """Get a dictionary of all things by matching a type. :param thing_class: The subclass of thing to match. :return: A dictionary of Things that match the subclass. If none match this will be and empty dictionary. """ matching = {} for thing_name, thing in self.server.things.items(): if isinstance(thing, thing_class): matching[thing_name] = thing return matching def get_thing_client(self, thing_name: str) -> lt.ThingClient: """Get a ThingClient for a Thing by name. :param thing_name: The name of the thing to on the server. :return: A LabThings ThingClient for the Thing with the specified name. """ self.check_thing_exists(thing_name) thing = self.server.things[thing_name] return lt.ThingClient.from_url(thing.path, self.client) def start_action( self, thing_name: str, action_name: str, action_kwargs: Optional[Mapping[str, Any]] = None, ) -> requests.Response: """Start an action and return the server response. For most purposes the best way to run an action is to use ``get_thing_client`` to create a ThingClient. At this point any actions can be run with a similar Python API to calling directly. However, using ThingClient blocks the test thread. This function provides an alternative way to start actions without blocking the test thread. It will return the HTTP response, this response can be used to poll or cancel the action. Use this method if you: * Want to test cancelling an action. * Want to inspect the actions logs exactly as they would come to a web client * Direcltly interact with the HTTP API :param thing_name: The name of the Thing on the server. :param action_name: The name of the action to start. :action_kwargs: The keyword inputs to the action. :return: A Response object with the HTTP response. """ self.check_thing_exists(thing_name) url = f"/{thing_name}/{action_name}" json_payload = {} if action_kwargs is None else action_kwargs response = self.client.post(url, json=json_payload) response.raise_for_status() return response def poll_action( self, response: requests.Response, interval: float = 0.01 ) -> Mapping[str, Any]: """Poll an action until it completes and return the final response data. :param response: The response from starting this action with ``start_action``. """ invocation_data = response.json() if "status" not in invocation_data: raise ValueError(f"Response json has no status: {invocation_data}") first_run = True while invocation_data["status"] in ACTION_RUNNING_KEYWORDS: if first_run: first_run = False else: time.sleep(interval) response = self.client.get(_invocation_href(invocation_data)) response.raise_for_status() invocation_data = response.json() return invocation_data def cancel_action(self, response: requests.Response) -> None: """Cancel an ongoing action. :param response: The response from starting this action with ``start_action``. """ invocation_data = response.json() response = self.client.delete(_invocation_href(invocation_data)) response.raise_for_status() def _get_link(obj: Mapping[str, Any], rel: str) -> Mapping[str, Any]: """Retrieve a link from an object's `links` list, by its `rel` attribute.""" return next(link for link in obj["links"] if link["rel"] == rel) def _invocation_href(invocation_data: Mapping[str, Any]) -> str: """Get the invocation href from the invocation response data.""" return _get_link(invocation_data, "self")["href"]