ophyd_async.testing

Utilities for testing devices.

Package Contents

Classes

ManagedSubprocess	A started subprocess, owning its Popen and able to stop it cleanly.
ExampleEnum	Example of a strict Enum datatype.
ExampleSubsetEnum	Example of a subset Enum datatype: the backend may have extra choices.
ExampleSupersetEnum	Example of a superset Enum datatype: not all choices need exist in backend.
ExampleTable	An abstraction of a Table where each field is a column.
OneOfEverythingDevice	A device with one of every datatype allowed on signals.
ParentOfEverythingDevice	Device containing subdevices with one of every datatype allowed on signals.
MonitorQueue	Monitors a Signal and stores its updates.
ApproxTable	For approximating two tables are equivalent.
StatusWatcher	Watches an AsyncStatus, storing the calls within.
BeamstopPosition	All members should exist in the Backend, and there will be no extras.
Exploder	This one takes a value and sets all its signal to that value.
MovableBeamstop	As well as reads, this one allows you to move it.
ReadOnlyBeamstop	Reads from 2 motors to work out if the beamstop is in position.

Functions

approx_value	Allow any value to be compared to another in tests.
assert_value	Assert that a Signal has the given value.
assert_reading	Assert that a readable Device has the given reading.
assert_configuration	Assert that a configurable Device has the given configuration.
assert_describe_signal	Assert the describe of a signal matches the expected metadata.
assert_emitted	Assert emitted document generated by running a Bluesky plan.
partial_reading	Helper function for building expected reading or configuration mapping.
wait_for_pending_wakeups	Allow any ready asyncio tasks to be woken up.
find_free_port	Find a currently-unused TCP port on localhost.
start_subprocess	Start a backend server subprocess.
int_array_value
float_array_value
assert_has_calls	Check that the device connected in mock mode has seen the supplied calls.

API

ophyd_async.testing.approx_value(value: Any)

Allow any value to be compared to another in tests.

This is needed because numpy arrays give a numpy array back when compared, not a bool. This means that you can’t assert array1==array2. Numpy arrays can be wrapped with pytest.approx, but this doesn’t work for Table instances: in this case we use ApproxTable.

async ophyd_async.testing.assert_value(signal: SignalR[SignalDatatypeT], value: Any) → None

Assert that a Signal has the given value.

Parameters:

• signal – Signal with get_value.

• value – The expected value from the signal.

async ophyd_async.testing.assert_reading(readable: AsyncReadable, expected_reading: Mapping[str, Mapping[str, Any]], full_match: bool = True) → None

Assert that a readable Device has the given reading.

Parameters:

• readable – Device with an async read() method to get the reading from.

• expected_reading – The expected reading from the readable.

• full_match – if expected_reading keys set is same as actual keys set. true: exact match false: expected_reading keys is subset of actual reading keys

async ophyd_async.testing.assert_configuration(configurable: AsyncConfigurable, expected_configuration: Mapping[str, Mapping[str, Any]], full_match: bool = True) → None

Assert that a configurable Device has the given configuration.

Parameters:

• configurable – Device with an async read_configuration() method to get the configuration from.

• expected_configuration – The expected configuration from the configurable.

• full_match – if expected_reading keys set is same as actual keys set. true: exact match false: expected_reading keys is subset of actual reading keys

async ophyd_async.testing.assert_describe_signal(signal: SignalR, /, **metadata)

Assert the describe of a signal matches the expected metadata.

Parameters:

• signal – The signal to describe.

• metadata – The expected metadata.

ophyd_async.testing.assert_emitted(docs: Mapping[str, list[dict]], **numbers: int)

Assert emitted document generated by running a Bluesky plan.

Parameters:

• docs – A mapping of document type -> list of documents that have been emitted.

• numbers – The number of each document type expected.

Example:

docs = defaultdict(list)
RE.subscribe(lambda name, doc: docs[name].append(doc))
RE(my_plan())
assert_emitted(docs, start=1, descriptor=1, event=1, stop=1)

ophyd_async.testing.partial_reading(val: Any) → Mapping[str, Any]

Helper function for building expected reading or configuration mapping.

Parameters:

val – Value to be wrapped in mapping with “value” as the key.

Returns:

The mapping that has wrapped the val with key “value”.

async ophyd_async.testing.wait_for_pending_wakeups(max_yields=20, raise_if_exceeded=True)

Allow any ready asyncio tasks to be woken up.

Used in:

• Tests to allow tasks like set() to start so that signal puts can be tested

• observe_value to allow it to be wrapped in asyncio.wait_for with a timeout

class ophyd_async.testing.ManagedSubprocess(subprocess_args: Sequence[str], ready_marker: str, *, startup_timeout: float = 15.0, stop_input: str | None = None, shutdown_timeout: float = 10.0)

A started subprocess, owning its Popen and able to stop it cleanly.

start() → None

stop() → None

ophyd_async.testing.find_free_port() → int

Find a currently-unused TCP port on localhost.

For callers that need to tell a subprocess which port to listen on up front (e.g. ophyd_async.tango.testing.start_tango_device_servers, which - unlike an EPICS IOC - can’t rely on a fixed macro-based address scheme, since a Tango TRL bakes the port straight into the URL). The standard “bind to port 0, read back what the OS assigned, close it” trick - the same one pytest/tox use. There’s an unavoidable (tiny) race between this returning and whatever you start next actually binding that port - acceptable for test/demo subprocess launching, not a security boundary.

ophyd_async.testing.start_subprocess(subprocess_args: Sequence[str], ready_marker: str, *, startup_timeout: float = 15.0, stop_input: str | None = None, shutdown_timeout: float = 10.0) → ManagedSubprocess

Start a backend server subprocess.

Parameters:

• subprocess_args – Full argv to spawn, e.g. [sys.executable, "-m", "epicscorelibs.ioc", ...].

• ready_marker – Substring to watch for on the subprocess’s stdout (stderr is merged into stdout) indicating it’s ready to accept connections. Blocks until this appears.

• startup_timeout – Seconds to wait for ready_marker before giving up.

• stop_input – Written to the subprocess’s stdin (then stdin is closed) by .stop() to request a clean shutdown, e.g. "exit()" for an EPICS IOC shell. If None, stdin is just closed with nothing written first - the shutdown mechanism the Tango device server scripts in this repo use (they block reading stdin and exit on EOF).

• shutdown_timeout – Seconds to wait for a clean exit after requesting shutdown before killing.

Returns a handle whose .stop() requests that clean shutdown (killing it if it doesn’t exit in time); also usable as a context manager.

class ophyd_async.testing.ExampleEnum

Bases: ophyd_async.core.StrictEnum

Example of a strict Enum datatype.

A

‘Aaa’

B

‘Bbb’

C

‘Ccc’

class ophyd_async.testing.ExampleSubsetEnum

Bases: ophyd_async.core.SubsetEnum

Example of a subset Enum datatype: the backend may have extra choices.

A

‘Aaa’

B

‘Bbb’

class ophyd_async.testing.ExampleSupersetEnum

Bases: ophyd_async.core.SupersetEnum

Example of a superset Enum datatype: not all choices need exist in backend.

A

‘Aaa’

B

‘Bbb’

C

‘Ccc’

D

‘Ddd’

class ophyd_async.testing.ExampleTable(**kwargs)

Bases: ophyd_async.core.Table

An abstraction of a Table where each field is a column.

For example:

>>> from ophyd_async.core import Table, Array1D
>>> import numpy as np
>>> from collections.abc import Sequence
>>> class MyTable(Table):
...     a: Array1D[np.int8]
...     b: Sequence[str]
...
>>> t = MyTable(a=[1, 2], b=["x", "y"])
>>> len(t)  # the length is the number of rows
2
>>> t2 = t + t  # adding tables together concatenates them
>>> t2.a
array([1, 2, 1, 2], dtype=int8)
>>> t2.b
['x', 'y', 'x', 'y']
>>> t2[1]  # slice a row
array([(2, b'y')], dtype=[('a', 'i1'), ('b', 'S40')])

a_bool: Array1D[bool_]

None

a_int: Array1D[int32]

None

a_float: Array1D[float64]

None

a_str: Sequence[str]

None

a_enum: Sequence[ExampleEnum]

None

class ophyd_async.testing.OneOfEverythingDevice(name='')

Bases: ophyd_async.core.StandardReadable

A device with one of every datatype allowed on signals.

async get_signal_values()

class ophyd_async.testing.ParentOfEverythingDevice(name='')

Bases: ophyd_async.core.Device

Device containing subdevices with one of every datatype allowed on signals.

async get_signal_values()

class ophyd_async.testing.MonitorQueue(signal: SignalR)

Bases: contextlib.AbstractContextManager

Monitors a Signal and stores its updates.

async assert_updates(expected_value)

class ophyd_async.testing.ApproxTable(expected: Table, rel=None, abs=None, nan_ok: bool = False)

For approximating two tables are equivalent.

Parameters:

• expected – The expected table.

• rel – The relative tolerance.

• abs – The absolute tolerance.

• nan_ok – Whether NaNs are allowed.

class ophyd_async.testing.StatusWatcher(status: WatchableAsyncStatus)

Bases: ophyd_async.core.Watcher[ophyd_async.testing._utils.T]

Watches an AsyncStatus, storing the calls within.

mock

‘Mock(…)’

Mock that stores watcher updates from the status.

async wait_for_call(current: T | None = None, initial: T | None = None, target: T | None = None, name: str | None = None, unit: str | None = None, precision: int | None = None, fraction: float | None = None, time_elapsed: float | Any = None, time_remaining: float | Any = None)

ophyd_async.testing.int_array_value(dtype: type[DTypeScalar_co])

ophyd_async.testing.float_array_value(dtype: type[DTypeScalar_co])

class ophyd_async.testing.BeamstopPosition

Bases: ophyd_async.core.StrictEnum

All members should exist in the Backend, and there will be no extras.

IN_POSITION

‘In position’

OUT_OF_POSITION

‘Out of position’

class ophyd_async.testing.Exploder(num_signals: int, name='')

Bases: ophyd_async.core.StandardReadable

This one takes a value and sets all its signal to that value.

This allows convenience “set all” functions, while the individual signals are still free to be set to different values.

class ophyd_async.testing.MovableBeamstop(name='')

Bases: ophyd_async.core.Device

As well as reads, this one allows you to move it.

E.g. bps.mv(beamstop.position, BeamstopPosition.IN_POSITION)

class ophyd_async.testing.ReadOnlyBeamstop(name='')

Bases: ophyd_async.core.Device

Reads from 2 motors to work out if the beamstop is in position.

E.g. bps.rd(beamstop.position)

ophyd_async.testing.assert_has_calls(device: Device | Signal, calls: Sequence[Any], reset_after=True)

Check that the device connected in mock mode has seen the supplied calls.

Parameters:

• device – Device or Signal connected in mock mode to check.

• calls – Expected sequence of mock calls.

• reset_after – Whether to reset the mock after assertion so subsequent calls don’t see the same calls again.

Example:

    device, [
        call.device.num_frames.put(1),
        call.device.start_writing.put(None),
    ],
)