ophyd_async.testing

Utilities for testing devices.

Package Contents

Classes

ExampleEnum	Example of a strict Enum datatype.
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.
get_mock	Return the mock (which may have child mocks attached) for a Device.
set_mock_value	Set the value of a signal that is in mock mode.
set_mock_values	Set a signal to a sequence of values, optionally repeating.
get_mock_put	Get the mock associated with the put call on the signal.
callback_on_mock_put	For setting a callback when a backend is put to.
mock_puts_blocked	Context manager to block puts at the start and unblock at the end.
set_mock_put_proceeds	Allow or block a put with wait=True from proceeding.
wait_for_pending_wakeups	Allow any ready asyncio tasks to be woken up.
int_array_value
float_array_value

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”.

ophyd_async.testing.get_mock(device: Device | Signal) → Mock

Return the mock (which may have child mocks attached) for a Device.

The device must have been connected in mock mode.

ophyd_async.testing.set_mock_value(signal: Signal[SignalDatatypeT], value: SignalDatatypeT)

Set the value of a signal that is in mock mode.

ophyd_async.testing.set_mock_values(signal: SignalR[SignalDatatypeT], values: Iterable[SignalDatatypeT], require_all_consumed: bool = False) → Iterator[SignalDatatypeT]

Set a signal to a sequence of values, optionally repeating.

Parameters:

• signal – A signal connected in mock mode.

• values – An iterable of the values to set the signal to, on each iteration the next value will be set.

• require_all_consumed – If True, an AssertionError will be raised if the iterator is deleted before all values have been consumed.

Example:

for value_set in set_mock_values(signal, range(3)):
    # do something

cm = set_mock_values(signal, [1, 3, 8], require_all_consumed=True):
next(cm) # do something

ophyd_async.testing.get_mock_put(signal: Signal) → AsyncMock

Get the mock associated with the put call on the signal.

ophyd_async.testing.callback_on_mock_put(signal: Signal[SignalDatatypeT], callback: Callable[[SignalDatatypeT, bool], SignalDatatypeT | None] | Callable[[SignalDatatypeT, bool], Awaitable[SignalDatatypeT | None]])

For setting a callback when a backend is put to.

Can either be used in a context, with the callback being unset on exit, or as an ordinary function.

The value that the callback returns (if not None) will be set to the signal readback. If None is returned then the readback will be set to the setpoint.

Parameters:

• signal – A signal with a MockSignalBackend backend.

• callback – The callback to call when the backend is put to during the context.

ophyd_async.testing.mock_puts_blocked(*signals: Signal)

Context manager to block puts at the start and unblock at the end.

ophyd_async.testing.set_mock_put_proceeds(signal: Signal, proceeds: bool)

Allow or block a put with wait=True from proceeding.

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.ExampleEnum

Bases: ophyd_async.core.StrictEnum

Example of a strict Enum datatype.

A

‘Aaa’

B

‘Bbb’

C

‘Ccc’

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)