|
| 1 | +"""Test harnesses to help with writitng tests for things..""" |
| 2 | + |
| 3 | +from __future__ import annotations |
| 4 | +from concurrent.futures import Future |
| 5 | +from typing import ( |
| 6 | + TYPE_CHECKING, |
| 7 | + Any, |
| 8 | + Awaitable, |
| 9 | + Callable, |
| 10 | + Mapping, |
| 11 | + ParamSpec, |
| 12 | + TypeVar, |
| 13 | + Iterable, |
| 14 | +) |
| 15 | +from tempfile import TemporaryDirectory |
| 16 | +from unittest.mock import Mock |
| 17 | + |
| 18 | +from .utilities import class_attributes |
| 19 | +from .thing_slots import ThingSlot |
| 20 | +from .thing_server_interface import ThingServerInterface |
| 21 | + |
| 22 | +if TYPE_CHECKING: |
| 23 | + from .thing import Thing |
| 24 | + |
| 25 | +Params = ParamSpec("Params") |
| 26 | +ReturnType = TypeVar("ReturnType") |
| 27 | + |
| 28 | + |
| 29 | +class MockThingServerInterface(ThingServerInterface): |
| 30 | + r"""A mock class that simulates a ThingServerInterface without the server. |
| 31 | +
|
| 32 | + This allows a `.Thing` to be instantiated but not connected to a server. |
| 33 | + The methods normally provided by the server are mocked, specifically: |
| 34 | +
|
| 35 | + * The `name` is set by an argument to `__init__`\ . |
| 36 | + * `start_async_task_soon` silently does nothing, i.e. the async function |
| 37 | + will not be run. |
| 38 | + * The settings folder will either be specified when the class is initialised, |
| 39 | + or a temporary folder will be created. |
| 40 | + * `get_thing_states` will return an empty dictionary. |
| 41 | + """ |
| 42 | + |
| 43 | + def __init__(self, name: str, settings_folder: str | None = None) -> None: |
| 44 | + """Initialise a ThingServerInterface. |
| 45 | +
|
| 46 | + :param name: The name of the Thing we're providing an interface to. |
| 47 | + :param settings_folder: The location where we should save settings. |
| 48 | + By default, this is a temporary directory. |
| 49 | + """ |
| 50 | + # We deliberately don't call super().__init__(), as it won't work without |
| 51 | + # a server. |
| 52 | + self._name: str = name |
| 53 | + self._settings_tempdir: TemporaryDirectory | None = None |
| 54 | + self._settings_folder = settings_folder |
| 55 | + self._mocks: list[Mock] = [] |
| 56 | + |
| 57 | + def start_async_task_soon( |
| 58 | + self, async_function: Callable[Params, Awaitable[ReturnType]], *args: Any |
| 59 | + ) -> Future[ReturnType]: |
| 60 | + r"""Do nothing, as there's no event loop to use. |
| 61 | +
|
| 62 | + This returns a `concurrent.futures.Future` object that is already cancelled, |
| 63 | + in order to avoid accidental hangs in test code that attempts to wait for |
| 64 | + the future object to resolve. Cancelling it may cause errors if you need |
| 65 | + the return value. |
| 66 | +
|
| 67 | + If you need the async code to run, it's best to add the `.Thing` to a |
| 68 | + `lt.ThingServer` instead. Using a test client will start an event loop |
| 69 | + in a background thread, and allow you to use a real `.ThingServerInterface` |
| 70 | + without the overhead of actually starting an HTTP server. |
| 71 | +
|
| 72 | + :param async_function: the asynchronous function to call. |
| 73 | + :param \*args: positional arguments to be provided to the function. |
| 74 | +
|
| 75 | + :returns: a `concurrent.futures.Future` object that has been cancelled. |
| 76 | + """ |
| 77 | + f: Future[ReturnType] = Future() |
| 78 | + f.cancel() |
| 79 | + return f |
| 80 | + |
| 81 | + @property |
| 82 | + def settings_folder(self) -> str: |
| 83 | + """The path to a folder where persistent files may be saved. |
| 84 | +
|
| 85 | + This will create a temporary folder the first time it is called, |
| 86 | + and return the same folder on subsequent calls. |
| 87 | +
|
| 88 | + :returns: the path to a temporary folder. |
| 89 | + """ |
| 90 | + if self._settings_folder: |
| 91 | + return self._settings_folder |
| 92 | + if not self._settings_tempdir: |
| 93 | + self._settings_tempdir = TemporaryDirectory() |
| 94 | + return self._settings_tempdir.name |
| 95 | + |
| 96 | + @property |
| 97 | + def path(self) -> str: |
| 98 | + """The path, relative to the server's base URL, of the Thing. |
| 99 | +
|
| 100 | + A ThingServerInterface is specific to one Thing, so this path points |
| 101 | + to the base URL of the Thing, i.e. the Thing Description's endpoint. |
| 102 | + """ |
| 103 | + return f"/{self.name}/" |
| 104 | + |
| 105 | + def get_thing_states(self) -> Mapping[str, Any]: |
| 106 | + """Return an empty dictionary to mock the metadata dictionary. |
| 107 | +
|
| 108 | + :returns: an empty dictionary. |
| 109 | + """ |
| 110 | + return {} |
| 111 | + |
| 112 | + |
| 113 | +ThingSubclass = TypeVar("ThingSubclass", bound="Thing") |
| 114 | + |
| 115 | + |
| 116 | +def create_thing_without_server( |
| 117 | + cls: type[ThingSubclass], |
| 118 | + *args: Any, |
| 119 | + settings_folder: str | None = None, |
| 120 | + mock_all_slots: bool = False, |
| 121 | + **kwargs: Any, |
| 122 | +) -> ThingSubclass: |
| 123 | + r"""Create a `.Thing` and supply a mock ThingServerInterface. |
| 124 | +
|
| 125 | + This function is intended for use in testing, where it will enable a `.Thing` |
| 126 | + to be created without a server, by supplying a `.MockThingServerInterface` |
| 127 | + instead of a real `.ThingServerInterface`\ . |
| 128 | +
|
| 129 | + The name of the Thing will be taken from the class name, lowercased. |
| 130 | +
|
| 131 | + :param cls: The `.Thing` subclass to instantiate. |
| 132 | + :param \*args: positional arguments to ``__init__``. |
| 133 | + :param settings_folder: The path to the settings folder. A temporary folder |
| 134 | + is used by default. |
| 135 | + :param mock_all_slots: Set to True to create a `unittest.mock.Mock` object |
| 136 | + connected to each thing slot. It follows the default of the specified |
| 137 | + to the slot. So if an optional slot has a default of `None`, no mock |
| 138 | + will be provided. |
| 139 | + :param \**kwargs: keyword arguments to ``__init__``. |
| 140 | +
|
| 141 | + :returns: an instance of ``cls`` with a `.MockThingServerInterface` |
| 142 | + so that it will function without a server. |
| 143 | +
|
| 144 | + :raises ValueError: if a keyword argument called 'thing_server_interface' |
| 145 | + is supplied, as this would conflict with the mock interface. |
| 146 | + """ |
| 147 | + name = cls.__name__.lower() |
| 148 | + if "thing_server_interface" in kwargs: |
| 149 | + msg = "You may not supply a keyword argument called 'thing_server_interface'." |
| 150 | + raise ValueError(msg) |
| 151 | + |
| 152 | + msi = MockThingServerInterface(name=name, settings_folder=settings_folder) |
| 153 | + # Note: we must ignore misc typing errors above because mypy flags an error |
| 154 | + # that `thing_server_interface` is multiply specified. |
| 155 | + # This is a conflict with *args, if we had only **kwargs it would not flag |
| 156 | + # any error. |
| 157 | + # Given that args and kwargs are dynamically typed anyway, this does not |
| 158 | + # lose us much. |
| 159 | + thing = cls(*args, **kwargs, thing_server_interface=msi) # type: ignore[misc] |
| 160 | + if mock_all_slots: |
| 161 | + _mock_slots(thing) |
| 162 | + return thing |
| 163 | + |
| 164 | + |
| 165 | +def _mock_slots(thing: Thing) -> None: |
| 166 | + """Mock the slots of a thing created by create_thing_without_server. |
| 167 | +
|
| 168 | + :param thing: The thing to mock the slots of. |
| 169 | + :raises TypeError: If this was called on a Thing with a real ThingServerInterface |
| 170 | + """ |
| 171 | + # Populate a mapping of mocks pretending to be the things on the server |
| 172 | + mocks = {} |
| 173 | + for attr_name, attr in class_attributes(thing): |
| 174 | + if isinstance(attr, ThingSlot): |
| 175 | + # Simply use the class of the first type that can be used. |
| 176 | + mock_class = attr.thing_type[0] |
| 177 | + |
| 178 | + # The names of the mocks we need to create to make a mapping of mock |
| 179 | + # things for the slot to connect to. |
| 180 | + mock_names = [] |
| 181 | + if attr.default is ...: |
| 182 | + # if default use the name of the slot with mock |
| 183 | + mock_names.append(f"mock-{attr_name}") |
| 184 | + elif isinstance(attr.default, str): |
| 185 | + mock_names.append(attr.default) |
| 186 | + elif isinstance(attr.default, Iterable): |
| 187 | + mock_names = list(attr.default) |
| 188 | + # Note: If attr.default is None it will connect to None so no need for |
| 189 | + # adding anything mapping of mocks. |
| 190 | + |
| 191 | + # Add mock to dictionary |
| 192 | + for name in mock_names: |
| 193 | + mock = Mock(spec=mock_class) |
| 194 | + mock.name = name |
| 195 | + mocks[name] = mock |
| 196 | + # Store a copy of this mock in the mock server interface so it isn't |
| 197 | + # garbage collected. |
| 198 | + interface = thing._thing_server_interface |
| 199 | + if isinstance(interface, MockThingServerInterface): |
| 200 | + interface._mocks.append(mock) |
| 201 | + else: |
| 202 | + raise TypeError( |
| 203 | + "Slots may not be mocked when a Thing is attached to a real " |
| 204 | + "server." |
| 205 | + ) |
| 206 | + |
| 207 | + # Finally connect the mocked slots. |
| 208 | + for _attr_name, attr in class_attributes(thing): |
| 209 | + if isinstance(attr, ThingSlot): |
| 210 | + attr.connect(thing, mocks, ...) |
0 commit comments