How to test a plugin¶
You can test the components and services of a plugin with the fixtures
redsun uses for its own tests, from
redsun.testing.
Prerequisites¶
You need a plugin package with pytest tests, as
Package components as a plugin describes. The blocks
below are parts of one test module, and the whole module is at the end.
Install the fixtures¶
Add redsun with the testing extra to the plugin's development
dependencies:
pytest-asyncio runs the async tests, such as the service test below.
Load them¶
Load the module as a pytest plugin in pyproject.toml:
From then on every test keeps its session settings and session log files under
its own tmp_path. So do the acquisition files and catalogs a session puts in
their default location, and every test drops the psygnal emissions it left
queued for another thread. A session whose configuration names its own
storage directory still writes there. Nothing loads the module unless a suite
asks for it, so these fixtures never reach a project that only installs
redsun.
A session built in a wider fixture isn't covered
The fixtures run for each test, so they don't cover a session built in a fixture scoped to a module or the whole run. Build the session in a per-test fixture.
The module defines no qapp fixture, and a session with a window needs a
QApplication. Take the qapp fixture of pytest-qt, or define one.
Build a session¶
Ask for build. It builds a session class with the
configuration you give it, and shuts the session down when the test ends:
def test_the_session_builds_every_component(
build: BuildSession, qapp: QApplication
) -> None:
"""Build every component of the session, its devices mocked."""
session = build(MyApp, {"mock": True, "strict": True})
assert set(session.devices) == {"motor"}
assert set(session.presenters) == {"ctrl"}
assert set(session.views) == {"panel"}
mock: True connects every device to a simulated backend, as in
Run without hardware. That doesn't suit a device
whose signals come from its service when it connects, as a fastcs device's
do, because a mocked one has none. Test such a device against its service
instead.
Start a service¶
Declare the service once, with the Launch a session class
uses:
Ask for start_service and give it the name
the service is declared under, the Launch, and the transport the session
file names under services.transport:
async def test_the_stage_moves(start_service: StartService) -> None:
"""Move the stage its service serves."""
service = start_service("stage_service", STAGE, transport="pv-access")
stage = MyStage(service.prefix, name="stage")
await stage.connect()
await stage.position.set(2.0)
assert await stage.position.get_value() == 2.0
When the test ends, the fixture stops the service, releases its transport as a session does, and restores the EPICS address list the service added itself to, so the next test starts from the same list. The test reads the prefix from the service it gets back.
Channel Access services started late are not found
A Channel Access client reads its address list once, the first time the
process uses Channel Access, so under channel-access a service started
after that isn't found. Prefer pv-access for services started in tests, or
start every Channel Access service the suite needs before its first client
connects.
The example in full¶
The whole module
"""The tests of the guide "How to test a plugin"."""
from __future__ import annotations
from typing import TYPE_CHECKING
from docs.examples.connect_on_demand import MyApp
from docs.examples.device_fastcs import MyStage
from redsun import Launch
if TYPE_CHECKING:
from qtpy.QtWidgets import QApplication
from redsun.testing import BuildSession, StartService
STAGE = Launch("stage_fastcs", ready="stage ready", prefix="STAGE:")
def test_the_session_builds_every_component(
build: BuildSession, qapp: QApplication
) -> None:
"""Build every component of the session, its devices mocked."""
session = build(MyApp, {"mock": True, "strict": True})
assert set(session.devices) == {"motor"}
assert set(session.presenters) == {"ctrl"}
assert set(session.views) == {"panel"}
async def test_the_stage_moves(start_service: StartService) -> None:
"""Move the stage its service serves."""
service = start_service("stage_service", STAGE, transport="pv-access")
stage = MyStage(service.prefix, name="stage")
await stage.connect()
await stage.position.set(2.0)
assert await stage.position.get_value() == 2.0