ska_pst.testutils.tango

Submodule for Tango related code.

class ska_pst.testutils.tango.AttributesMonitor(device_proxy: PstDeviceProxy, attribute_names: List[str], logger: Optional[Logger] = None)

Class used to monitor the attributes of a Tango device.

This class can be used to track multiple attributes of a Tango class and then be used to assert values or wait for when an attribute is updated.

Creating the instance of this class does nothing. The setup method must be called afterwards to ensure that attributes are monitored.

assert_attribute(attribute: str, value_assertion: Callable[[...], bool]) → None

Assert and attribute has a given value.

This is a helper method to get the attribute’s history and then passes it to the value_assertion callable.

Parameters

• attribute – the name of the attribute to assert against.

• value_assertion – a callable to assert against the latest value of the attribute.

assert_attribute_values_changed() → None

Assert that attribute values have changed since last check.

assert_attribute_values_not_changed() → None

Assert that attribute values have not changed since last check.

capture_current_values() → None

Capture the current values to allow for asserting of updates later.

property current_attribute_values: dict

Get current attribute values for device.

get_attribute_history_events(attribute_name: str) → List[AttributeHistoryEvent]

Get list of history events for an attribute.

setup() → None

Set up monitoring for attributes.

teardown() → None

Teardown the monitor.

This will unsubscribe from Tango events of the attributes.

wait_for_attribute_update(attribute_name: str, timeout: float) → None

Wait for attribute to be updated.

Waits until there has been an update for the specific attribute or a timeout has occurred.

Parameters

• attribute_name – the attribute to wait for an update of.

• timeout – how long to wait for an update before raising an exception.

class ska_pst.testutils.tango.CommandTracker(device_proxy: PstDeviceProxy, change_event_callbacks: ska_tango_testing.mock.tango.MockTangoEventCallbackGroup, logger: Optional[Logger] = None, default_timeout: float = 60.0)

Class to track the progress and results of commands on a PstDeviceProxy.

This class also uses the TangoDeviceCommandChecker which is more low level to check for updates of the long running process values. This provides a high level view of a command. It also will record if the commands fail and what was the state of the device proxy before the command was executed.

assert_previous_command_error_message_matches(expected_message_regexp: str) → None

Assert previous command error message is equal to the argument.

assert_previous_command_failed() → None

Assert previous command failed.

assert_previous_command_rejected() → None

Assert previous command was rejected due to invalid state.

perform_command(command: str, raise_exception: bool = False, timeout: Optional[float] = None, **kwargs: Any) → None

Execute the command.

This method can also optionally reraise the exception during the call rather than storing the exception. This should be used during the setup and teardown of the system.

teardown() → None

Teardown the command tracker.

This releases all of the subscriptions and change events used by the tracker.

class ska_pst.testutils.tango.PstTestDeviceProxy(fqdn: str, logger: Optional[Logger] = None, command_timeout: float = 30.0)

A class for use when testing a PST Tango Device.

This class is designed as a lightweight proxy that wraps a tango.DeviceProxy. This allows for automated testing or using in a notebook to block on the command calls rather than assuming the calls have finished.

All the commands exposed on this class do not check the current state of the remote device. The remote device is in control of checking state and rejecting methods that are or are not allowed.

Abort() → None

Call Abort on remote device.

ConfigureScan(scan_configuration: str) → None

Call ConfigureScan on remote device.

This method takes a scan configuration that is passed to the device proxy.

Parameters

scan_configuration – a JSON string of the scan configuration to be sent to the remote device.

EndScan() → None

Call EndScan on remote device.

GoToFault(fault_msg: str) → None

Call GoToFault on remote device.

GoToIdle() → None

Call GoToIdle on remote device.

ObsReset() → None

Call ObsReset on remote device.

Off() → None

Call Off command on remote device.

On() → None

Call On command on remote device.

Reset() → None

Call Reset on remote device.

Scan(scan_id: str) → None

Call Scan on remote device.

This will put the remote device in to a SCANNING state.

display_monitoring() → None

Display current values of some monitored attributes on remote device.

get_property(propname: str) → Any

Get the value of a device property.

This just proxies through to the tango.DeviceProxy.get_property

monitor() → None

Start background monitoring of values of remote device.

This method will start a background process to log out the current monitored values. This is done at a rate given by the monitoring polling rate on the remote device.

state() → tango.AdminMode

Get the current admin mode state of the remote device.

class ska_pst.testutils.tango.TangoChangeEventHelper(device_under_test: tango.DeviceProxy, change_event_callbacks: ska_tango_testing.mock.tango.MockTangoEventCallbackGroup, logger: Optional[Logger] = None)

Internal testing class used for handling change events.

assert_change_event(attribute_name: str, attribute_value: Any, **kwargs: Any) → dict

Assert that the callback received a change event with the given value.

This is a helper method that delegates to the MockTangoEventCallbackGroup.assert_change_event

Parameters

• attribute_name (str) – name of the attribute to assert a change event against.

• attribute_value (Any) – new value of the attribute for which the change event has been sent

Returns

details of the change event

Raises

AssertionError – if the asserted call has not occurred within the timeout period

release() → None

Release any subscriptions that are held.

subscribe(attribute_name: str) → None

Subscribe to change events of an attribute.

This returns a MockChangeEventCallback that can then be used to verify changes.

class ska_pst.testutils.tango.TangoDeviceCommandChecker(tango_change_event_helper: TangoChangeEventHelper, change_event_callbacks: ska_tango_testing.mock.tango.MockTangoEventCallbackGroup, logger: Optional[Logger] = None)

A convenience class used to help check a Tango Device command.

This class can be used to check that a command executed on a DeviceProxy fires the correct change events for task status, the completion state, and any changes through the ObsState.

assert_command(command: str, command_args: Optional[tuple[Any]] = None, expected_command_result: str | None = 'Completed', expected_command_status_events: List[ska_tango_base.executor.TaskStatus] = [ska_tango_base.executor.TaskStatus.STAGING, ska_tango_base.executor.TaskStatus.QUEUED, ska_tango_base.executor.TaskStatus.IN_PROGRESS, ska_tango_base.executor.TaskStatus.COMPLETED], expected_obs_state_events: List[ska_control_model.ObsState] = [], timeout: float = 10.0, **kwargs: Any) → None

Assert that the command has the correct result and events.

This method has sensible defaults of the expected result code, the overall result, and the status events that the command goes through, and by default asserts that the ObsState model doesn’t change.

Parameters

• command (str) – the name of the command to call on the remote TANGO device

• command_args (tuple[Any] | None, optional) – the optional arguments to pass through to TANGO device, defaults to None

• expected_command_result (str | None, optional) – the expected command result when the command completes, defaults to “Completed”.

• expected_command_status_events (List[TaskStatus], optional) – a list of expected status events of the command, these should be in the order the events happen. Default expected events are: [ TaskStatus.STAGING, TaskStatus.QUEUED, TaskStatus.IN_PROGRESS, TaskStatus.COMPLETED, ]

• expected_obs_state_events (List[ObsState], optional) – the expected events of the ObsState model. The default is an empty list, meaning no events expected.

• timeout (float, optional) – expected length of time for the results of the command to take, defaults to 10.0.