Testing
For this guide we will build upon the example AwesomeInstrumentProtocol that we declared in the Walkthrough.
Add a simulation mode
Every connector should run without the physical device attached. Even a simulation that only returns plausible mock data lets you run CI, onboard developers without hardware, and refactor safely.
The requirements:
- Register the
SimulationControllercore feature, so clients can switch between simulation and real mode. - Simulate in the protocol, not in the features. Swap the protocol's transport for a mock. The SiLA interface then runs exactly the same code in both modes, and simulation tests exercise the real interface.
- Keep the simulation stubs in
io/, next to the protocol they mock, not in a separate top-level package. - Default to simulation mode with a config value. The
connector devcommand relies on this.
1. Write the stubs
A stub receives the request bytes and returns the device's response. stub_maker builds one from a fixed request and response:
from unitelabs.bus.testing import stub_maker
DEVICE_DATA: dict[bytes, bytes] = {
b"status": b"ok",
b"getSerial\r": b"0000012345\r\n"
}
stubs = [stub_maker(k, v) for k, v in DEVICE_DATA.items()]
2. Add the config value
import dataclasses
from unitelabs.cdk import ConnectorBaseConfig
@dataclasses.dataclass
class AwesomeInstrumentConfig(ConnectorBaseConfig):
"""Configuration for the Awesome Instrument."""
simulation_mode: bool = True
"""Whether to start the connector in simulation mode, without hardware."""
3. Implement the simulation controller
Subclass SimulationControllerBase and swap the protocol's transport when the mode changes. Protocol.connect returns the previous transport factory and its arguments, so you can restore the real connection later.
import typing
from unitelabs.bus.testing import create_mock_connection
from unitelabs.cdk.features.core.simulation_controller import SimulationControllerBase, StartRealModeFailed
from ...io.protocol import AwesomeInstrumentProtocol
from ...io.simulation import stubs
class SimulationController(SimulationControllerBase):
"""Switch the connector between simulated responses and the real device."""
def __init__(
self,
protocol: AwesomeInstrumentProtocol,
simulation_mode: bool = True,
):
super().__init__()
self.protocol = protocol
self._og_identity = self.protocol.identity
self._og_transport_factory = None
self._og_transport_kwargs: dict = {}
if simulation_mode:
self._loop = asyncio.get_event_loop_policy().get_event_loop()
asyncio.run_coroutine_threadsafe(self.start_simulation_mode(), self._loop)
@typing.override
async def simulation_mode(self) -> bool:
return self._simulation_mode
@typing.override
async def start_real_mode(self) -> None:
if self._simulation_mode:
self.protocol.logger.info("Starting real mode.")
if hasattr(self.protocol, "transport") and self.protocol.transport:
self.protocol.transport.close()
self.protocol.identity = self._og_identity
if self._og_transport_factory is None:
msg = "Cannot start real mode."
raise StartRealModeFailed(msg)
await self.protocol.connect(self._og_transport_factory, **self._og_transport_kwargs)
await self.protocol.open()
self._simulation_mode = False
@typing.override
async def start_simulation_mode(self) -> None:
if self._simulation_mode:
return
self.protocol.logger.info("Starting simulation mode.")
# Backup real identity (only used for serial communication, autoconnection)
self._og_identity = self.protocol.identity
self.protocol.identity = AsyncMock(return_value=True)
if hasattr(self.protocol, "transport") and self.protocol.transport:
self.protocol.transport.close()
self._og_transport_factory, self._og_transport_kwargs = await self.protocol.connect(
create_mock_connection, stubs=stubs
)
await self.protocol.open()
self._simulation_mode = True
4. Register it
Start in the configured mode and register the controller next to your other features:
async def create_app(config: AwesomeInstrumentConfig) -> collections.abc.AsyncGenerator[Connector]:
app = Connector(config)
protocol = AwesomeInstrumentProtocol()
simulation_controller = SimulationController(protocol, simulation_mode=config.simulation_mode)
try:
if not config.simulation_mode:
await protocol.open()
app.register(simulation_controller)
app.register(DeviceController(protocol))
yield app
finally:
await protocol.close()
Test layout
Lay out tests/ to mirror the package, with each test file named test_ plus the name of the file it tests:
| Source file | Test file |
|---|---|
io/protocol.py | tests/io/test_protocol.py |
features/device_controller/device_controller.py | tests/features/test_device_controller.py |
Protocol Unit Testing
For a device with a byte-level protocol, write unit tests for every method in io/ against a stubbed protocol. Map request bytes to response bytes so the tests cover the full range of device responses, including error responses and edge cases.
The Omnibus package ships pytest fixtures for this. Enable them in your conftest file:
pytest_plugins = ["unitelabs.bus.testing.fixtures"]
The mk_stubbed_protocol fixture takes a Protocol subclass, a dictionary mapping request bytes to response bytes, and the protocol's constructor kwargs. It returns an instance of your protocol that answers every execute call from that mapping instead of talking to a device.
1. Protocol Test Fixture
Wrap mk_stubbed_protocol in a fixture for your own protocol:
import typing
import pytest
from unitelabs.awesome_instrument.io.protocol import AwesomeInstrumentProtocol
pytest_plugins = ["unitelabs.bus.testing.fixtures"]
@pytest.fixture
def mk_my_protocol(mk_stubbed_protocol):
def _mk_my_protocol(data: dict[bytes, typing.Union[bytes, list[bytes]]], **kwargs) -> AwesomeInstrumentProtocol:
return mk_stubbed_protocol(AwesomeInstrumentProtocol, data, **kwargs)
return _mk_my_protocol
Because the fixture returns a function, each test can pass its own response mapping. You could also use mk_stubbed_protocol directly; the wrapper only reduces repetition.
2. Mock Device Communication
import pytest
from unitelabs.awesome_instrument.io.errors import NotOkException
from unitelabs.awesome_instrument.io.protocol import AwesomeInstrumentProtocol
async def test_should_get_status(mk_my_protocol):
protocol: AwesomeInstrumentProtocol = mk_my_protocol({b"status": b"ok"})
await protocol.open()
response = await protocol.get_status()
assert response == b"ok"
async def test_should_raise_if_not_ok(mk_my_protocol):
protocol: AwesomeInstrumentProtocol = mk_my_protocol({b"status": b"not ok"})
await protocol.open()
with pytest.raises(NotOkException, match=r"Everything is not ok\."):
await protocol.get_status()
Each test feeds the protocol one possible device response and checks that the method handles it as expected.
Last updated