ITH as a Platform: API Reference

(New) Core API Reference

The core of the Integration Test Harness as a Platform.

The core of the Integration Test Harness as a Platform is a set of classes and functions that provide the basic functionalities to create integration tests for complex Tango-based systems. The core has the following features:

The core is generic and by itself it doesn’t even talk about telescopes and SKA, but is more intended as a framework to integration test any kind of complex system based on Tango devices.
The core is extendable and adaptable to many different testing contexts, but also usable as it is.
The core relies on ska_tango_testing for the event tracing and assertion mechanisms.
The core is unit tested code, and it is expected to be reliable and robust.

The core is composed of the following main components:

ska_integration_test_harness.core.actions: a set of classes that represent the actions that can be performed on your system and the consequent synchronisation. The concept of action includes both very generic procedures and more specific ones (e.g., like a command execution). You can use the actions as building blocks to represent the interactions with your system in a structured way.
ska_integration_test_harness.core.assertions: a set of classes that represent verification procedures on your system. The assertions may be simple checks (e.g., a set of devices are in a certain state) or more complex event-based checks. You can use the assertions in actions as synchronisation points and preconditions/postconditions, but potentially also as standalone checks.

NOTE: all the core components are usually represented as class hierarchies, so you can:

use the given (non-abstract) classes as they are
extend the base or the intermediate classes to create your own custom pieces of code

In ska_integration_test_harness.extensions you can find some already implemented and tested extensions that may be useful for many SKA testing contexts.

Actions module

A framework for defining and executing actions in test cases.

This module provides base classes and implementations for defining and executing operations in test cases against a (Tango-based) system under test (SUT).

What is an action? In abstract terms, an action is any kind of self-contained operation that can be executed on the SUT. We chose to define an action class structure because we want to give a common interface to anything that holds test logic and/or domain knowledge about how you interact with the SUT. Concretely, an action can be something as simple as sending a command to a device or setting an attribute, or it can be something more complex like orchestrating a sequence of commands, operations, or other actions.

In our framework, we expect an action to have the following structure:

Expects some kind of preconditions to be met to be executed
Performs some operations on the SUT (e.g., invoking a command, writing one or more attributes on devices, orchestrating a sequence of other actions)
Verifies the effects of the operations on the SUT through a set of postconditions (and eventually synchronises with the SUT state)

How are actions implemented in the ITH? In the ITH framework, we choose to define actions as a class hierarchy. The base class for all actions is SUTAction.

This base class is essentially not much more than an empty shell that defines a common interface for the action lifecycle and so it needs to be extended to implement concretely the preconditions, the operations, and the postconditions. You can directly extend it and implement the steps from scratch or you can start from one of the provided subclasses. Three important subclasses are:

TracerAction: a subclass that essentially gives more structure to the way the preconditions and postconditions are defined. This class makes you define and set the preconditions and postconditions as a list of assertion objects (see assertions) and it uses a ska_tango_testing.integration.TangoEventTracer to perform the verifications. This is a good choice if you want to have a declarative way to define many simple and repetitive pre and post conditions.
TangoCommandAction: a subclass that provides a ready-to-use implementation for actions whose procedure is a simple Tango command invocation. It is a subclass of TracerAction so it inherits the structure for defining preconditions and postconditions. It is useful as a simple and quick way to send commands and verify their effects on the SUT.
SUTActionSequence: a subclass that provides a way to group multiple actions in a sequence and execute them in order within a shared timeout.

Further actions and assertions (e.g, for dealing with long-running commands and subarray commands) can be found in the ska_integration_test_harness.extensions module. We suggest you look at that module before implementing your own actions and assertions from scratch.

class ska_integration_test_harness.core.actions.SUTAction

A generic action on the System Under Test (SUT).

This class is the base class for the actions framework and it provides an empty shell for defining a generic interaction with the SUT.

An action is a self-contained operation that can be executed with the execute() method. An action is supposed to be made of:

A core procedure that acts on the SUT (method execute_procedure(), the only compulsory extension point)
The verification of some preconditions before the procedure is executed (method verify_preconditions(), optional extension point)
The verification of some postconditions after the procedure is executed and an eventual synchronisation with the SUT state within a timeout (method verify_postconditions(), optional extension point)
A setup phase that prepares the action to be executed and resets any internal resources (method setup(), optional extension point)
A name and a brief description of the action (methods name() and description(), optional extension points)

How to use an action as an end user: An end user can use actions simply by creating an instance of the action and calling the execute() method. The action will be executed, and the preconditions and postconditions will be verified. The execution can be repeated as many times as the user wants, given the preconditions are satisfied. Optionally, the user can also:

use a timeout for the postconditions verification by passing the timeout parameter to the execute() method (by default it is 0s, and to be working who developed the action should have implemented the postconditions verification within a timeout)
disable the verification of the preconditions or postconditions by passing the flags verify_preconditions and verify_postconditions to the execute() method
disable the logging messages generated by the action by calling set_logging() with the flag enable_logging set to False

How to extend an action: An action can be extended by subclassing it and overriding the extension points. The only compulsory extension point is the execute_procedure() method. The other extension points are optional and can be used to add custom preconditions, postconditions, setup, name, and description.

Extension and usage example:

from ska_integration_test_harness.core.actions import SUTAction

# define a new action
class MyAction(SUTAction):

    def setup(self):  # optional
        super().setup() # always call the superclass

        # e.g., reset a tracer, clear the events
        # subscribe to the attributes, etc.
        # ...

    def execute_procedure(self): # compulsory
        # (compulsory) act on the SUT

        # e.g., run some kind of interaction algorithm to reach
        # a certain state
        # ...

    # optional
    def verify_preconditions(self, timeout: SupportsFloat = 0):
        # always call the superclass
        super().verify_preconditions(timeout)

        # e.g., check that the device is in the desired state
        # within the timeout using the tracer
        # ...

# execute the action and verify the postconditions within 10s
action = MyAction()
action.execute(postconditions_timeout=10)

NOTE for those who extend this class: When you extend this class, always think about the value and semantic meaning. An action is supposed to be some meaningful procedure that acts on the SUT; it’s a self-contained piece of business logic, so choose a meaningful class name, write a good docstring, and provide a meaningful short description. Since an action implementation will likely not be unit tested code, prioritise clarity and readability. Think also about reusability: if you find yourself creating several actions that are very similar, consider refactoring the common logic into a superclass or implementing a single (possibly unit tested) parametrised action.

description()

A brief description of the action (optional extension point).

This method returns a string that is used to describe very briefly what this action is supposed to do in the logs and in the reports. The default implementation returns an empty string.

HOW TO EXTEND: Override this method in a subclass to provide a your custom description. Make it be a single small sentence that describes the action semantically (e.g., “Bring the telescope in X state”).

Return type:: str

execute(postconditions_timeout=0, verify_preconditions=True, verify_postconditions=True)

Execute the action and verify the postconditions.

This method is the entry point for the user to execute the action. An execution of an action consists of the following steps:

setup() is called to prepare the action to be executed (and reset eventual internal resources).
verify_preconditions() is called to check if the preconditions are satisfied. If the preconditions are not satisfied, an exception is raised.
execute_procedure() is called to act on the SUT.
verify_postconditions() is called to check if the postconditions are satisfied (within the given timeout). If the postconditions are not satisfied, an exception is raised.

An action is supposed to be executable multiple times, given the preconditions are satisfied.

The user can pass a timeout to the method to set a timeout for the postconditions verification. By default, the timeout is 0s.

Through the verify_preconditions and verify_postconditions flags, the user can decide if the preconditions and postconditions should be verified. By default, both preconditions and postconditions are verified.

Parameters:

postconditions_timeout (SupportsFloat) – The timeout for the postconditions verification. By default, the timeout is 0s. You can pass a float or an integer, or also a ska_tango_testing.integration.assertions.ChainedAssertionsTimeout object. If you pass a numerical value, internally it is converted into an object to make it be “shared by default” inside the action. The timeout will start before verifying the postconditions. If you want to start it before, you should create the timeout object by yourself and pass it already started.
verify_preconditions (bool) – True if the preconditions should be verified before executing the action, False otherwise. By default, the preconditions are verified.
verify_postconditions (bool) – True if the postconditions should be verified after executing the action, False otherwise. By default, the postconditions are verified.

Return type:

None

Raises:

AssertionError if the preconditions or postconditions are not satisfied. Additional exceptions can be raised by the execute_procedure() method if the action fails.

abstract execute_procedure()

Act on the SUT (compulsory extension point).

This method is the core of the action. It should execute the procedure of the action. The procedure is the sequence of commands, interactions, or any other operation that the action is supposed to perform on the SUT.

HOW TO EXTEND: Override this method in a subclass to add custom procedures. If it exists, you may consider to call the superclass method but it is not compulsory. You can assume setup() and verify_preconditions() have been called before this method and verify_postconditions() will be called after.

unless the action purpose is to embed conditional logic, assume the system is in the expected state defined by the preconditions through verify_preconditions()
if the operation you are done is asynchronous, make this method terminate quickly and put the synchronisation logic in verify_postconditions()
describe in the docstring what the action does (this method docstring, but also the class docstring)

Return type:: None
Raises:: AssertionError if the action fails.

is_logging_enabled()

Check if the logging is enabled for the action.

Return type:: bool
Returns:: True if the action logs the execution, False otherwise.

logger: The logger for the action.

name()

A name to identify the action (optional extension point).

This method returns a string that is used to identify the action in the logs and in the reports. The default implementation returns the class name.

HOW TO EXTEND: Override this method in a subclass to provide a your custom name. Make it be just 1 or few more words. We don’t really suggest to override this method unless you have a good reason to do so (override instead the description() method).

Return type:: str

set_logging(enable_logging)

Enable or disable logging for the action.

Enable or disable the logging messages the action generates during its execution. By default, logging is enabled.

TODO: should this be a runtime configuration or a class-level configuration?

Parameters:: enable_logging (bool) – True if the action should log the execution, False otherwise.
Return type:: None

setup()

Set up the action (optional extension point).

This method is called before the action is executed and any assertion is verified. It should set up action resources in a way such that all the preconditions can be verified, the action’s procedure executed, and the postconditions verified.

HOW TO EXTEND: Override this method in a subclass to add custom setup. Always call the superclass method when overriding this method. Some good practices if you override this method are:

make it be idempotent
after this method termination, the user may be able to assume the action instance is ready to verify preconditions and execute the procedure. Make it capable of resetting any kind of resources such as tracers, timeouts, etc.
always call the superclass method
in the docstring of the method, specify the resources that are set up (and briefly recap also what is done by superclasses, potentially referencing the superclasses method docstring)

Return type:: None

verify_postconditions(timeout=0)

Verify the postconditions of the action (extension point).

This method is called after the action is executed. It should verify that the system reaches the state that is expected after a successful execution of the action. If the postconditions are not satisfied, the method should raise an exception. The verification should be done within the given timeout.

HOW TO EXTEND: Override this method in a subclass to add custom postconditions. Always call the superclass method when overriding this method. You can assume setup(), verify_preconditions(), and execute_procedure() have been called before this method Some good practices if you override this method are:

make it be idempotent
always call the superclass method
the assertions you put here can be slow (e.g., with timeouts) and can be intended as a way to synchronise with the system (e.g., waiting for a device to be ready)
after this method termination, the user may be able to assume the system is in the state expected by the action
in the docstring of the method, specify which postcondition is verified (and briefly recap also what is done by superclasses, potentially referencing the superclasses method docstring)
use the given timeout to set a timeout for the verification (if you need to wait for something to happen)

Parameters:: timeout (SupportsFloat) – the time in seconds to wait for the postconditions to be verified. If not specified, it defaults to 0.
Return type:: None
Raises:: AssertionError if the postconditions are not satisfied.

verify_preconditions()

Verify the preconditions (optional extension point).

This method is called before the action is executed. It should verify that all the system in the state required by the action. If the preconditions are not satisfied, the method should raise an exception.

HOW TO EXTEND: Override this method in a subclass to add custom preconditions. Always call the superclass method when overriding this method. You can assume setup() has been called before this method. Some good practices if you override this method are:

make it be idempotent
always call the superclass method
try to put here only assertions that are fast to compute (i.e., without timeouts)
after this method termination, the user may be able to assume the system is in the state expected by the action to be executed correctly
in the docstring of the method, specify which precondition is verified (and briefly recap also what is done by superclasses, potentially referencing the superclasses method docstring)

Return type:: None
Raises:: AssertionError if the preconditions are not satisfied.

class ska_integration_test_harness.core.actions.SUTActionSequence

Compose a sequence of actions, under a same common timeout.

This class represents a sequence of ska_integration_test_harness.core.actions.SUTAction instances grouped together in a sequence to be executed in order. The actions have the following properties:

Each action is a SUTAction instance, and can be executed within a shared timeout by calling execute(). The verification includes a setup, a verification of preconditions, the execution of the procedure, and a verification of postconditions (with an eventual synchronisation within the timeout).
Execution parameters such as the timeout and the flags to enable/disable precondition and postcondition verification are shared among all the actions in the sequence. The shared timeout is started before the first postcondition evaluation.
You can build the sequence by adding actions with add_actions().
The sequence can be executed in the given order by calling execute(). Every time you call the method, the sequence is executed from the beginning, within a shared timeout (an action remaining time for verifying postconditions is the remaining time left by the previous)
Optionally, you can override this class to add further global setup, precondition, or postcondition verification.

Example:

from ska_integration_test_harness.core.actions import SUTActionSequence

# create a sequence of actions
sequence = SUTActionSequence()

# add actions to the sequence
sequence.add_actions(action1, action2, action3)

# execute the sequence (within the same shared timeout)
sequence.execute(postconditions_timeout=10)

property actions: list[SUTAction]

The ordered list of actions in the sequence.

Returns:: The list of actions in the sequence.

add_actions(*actions, put_them_at_beginning=False)

Add one or more actions to the sequence.

Parameters:

actions (SUTAction) – The actions to add to the sequence.
put_them_at_beginning – If True, the actions are added at the beginning of the sequence.

description()

A brief description of the action (optional extension point).

This method returns a string that is used to describe very briefly what this action is supposed to do in the logs and in the reports. The default implementation returns an empty string.

HOW TO EXTEND: Override this method in a subclass to provide a your custom description. Make it be a single small sentence that describes the action semantically (e.g., “Bring the telescope in X state”).

execute_procedure(): Run the sequence of actions within the same shared timeout.

set_logging(enable_logging)

Set the logging for this instance and for the whole sequence.

This method sets the logging for this instance and for all the actions in the sequence. See set_logging() for more details.

Parameters:: enable_logging (bool) – If True, the logging is enabled for this instance and for all the actions in the sequence.

class ska_integration_test_harness.core.actions.TangoCommandAction(target_device, command_name, command_param=None, command_kwargs=None)

Send a command to a Tango device and synchronise using a tracer.

This class represents an action that sends a command to a Tango device and then performs a series of checks using the ska_tango_testing.integration.TangoEventTracer.

The command:

is sent to a target tango.DeviceProxy instance;
can be any Tango command;
can be sent with or without parameters;
may produce a result which is stored in this action instance’s last_command_result attribute.

The preconditions and postconditions verification are performed using the same mechanics as the superclass ska_integration_test_harness.core.actions.TracerAction. Contrary to the superclass, this action does not need to be extended and is ready to be used as it is.

Usage example:

from ska_integration_test_harness.core.actions import TangoCommandAction
from ska_integration_test_harness.core.assertions import AssertDevicesAreInState
from ska_integration_test_harness.core.assertions import AssertDevicesStateChanges

# Then you can build action instances and add preconditions and
# postconditions to them according to your needs.

action = TangoCommandAction(
    target_device=dev1,
    command_name="IncreaseAttribute",
    command_param=2,
).add_preconditions(
    AssertDevicesAreInState(
        devices=[dev1, dev2],
        attribute_name="attr1",
        attribute_value=42
    ),
).add_postconditions(
    AssertDevicesStateChanges(
        devices=[dev1, dev2],
        attribute_name="attr1",
        attribute_value=43
    ),
    AssertDevicesAreInState(
        devices=[dev1, dev2],
        attribute_name="attr1",
        attribute_value=44
    ),
).add_early_stop(
    lambda e: e.attribute_value < 42
)

# execute the action within a timeout of 5 seconds
action.execute(postconditions_timeout=5)

command_kwargs

Additional keyword arguments.

(they will be passed to tango.DeviceProxy.command_inout())

command_name: The name of the command to execute.

command_param: The parameter to pass to the command.

description()

Describe the sent command and its arguments.

Returns:: a string describing the command and its arguments.

execute_procedure()

Call the command on the target device and store the result.

Call the command on the target device with the provided arguments and keyword arguments. The result of the command is stored in the last_command_result attribute.

last_command_result

The result of the last command execution.

It is set to None until the command is executed.

name()

Return the name of the action.

Returns:: the name of the action.

target_device: The target device on which to execute the command.

class ska_integration_test_harness.core.actions.TracerAction

An event-based action which uses TangoEventTracer to synchronise.

This class represents an action where the preconditions and postconditions have a strong dependency on the events emitted by the SUT, and so this class provides a structure to manage them.

Concretely, this action:

Makes you define the preconditions and postconditions using respectively ska_integration_test_harness.core.assertions.SUTAssertion and ska_integration_test_harness.core.assertions.TracerAssertion objects (through the add_preconditions() and add_postconditions() methods).
Manages a ska_tango_testing.integration.TangoEventTracer to trace the events emitted by the SUT and automatically makes all the preconditions and postconditions use the same tracer.
Manages a shared timeout for the postconditions, so all of them are verified within the same time interval.
Allows you to define early stop conditions for the postconditions, through the add_early_stop() method (to read more about early stop conditions, see ska_tango_testing.integration.assertions.with_early_stop()).

To do this, the class implements many of the lifecycle methods of the SUTAction class, except for the execute_procedure() method, which is left to be implemented by the subclasses.

This class is very useful if combined with built-in assertions like ska_integration_test_harness.core.assertions.AssertDevicesAreInState and ska_integration_test_harness.core.assertions.AssertDevicesStateChanges.

Usage example:

from ska_integration_test_harness.core.actions import TracerAction
from ska_integration_test_harness.core.assertions import AssertDevicesAreInState
from ska_integration_test_harness.core.assertions import AssertDevicesStateChanges

# To use this class, you need to create a subclass and implement the
# execute_procedure method.
class IncrementAttributeBy2(TracerAction):
    def execute_procedure(self):
        # your action logic here to increment the attribute by 2

# Then you can build action instances and add preconditions and
# postconditions to them according to your needs.

action = MyAction().add_preconditions(
    AssertDevicesAreInState(
        devices=[dev1, dev2],
        attribute_name="attr1",
        attribute_value=42
    ),
).add_postconditions(
    AssertDevicesStateChanges(
        devices=[dev1, dev2],
        attribute_name="attr1",
        attribute_value=43
    ),
    AssertDevicesAreInState(
        devices=[dev1, dev2],
        attribute_name="attr1",
        attribute_value=44
    ),
)

# execute the action within a timeout of 5 seconds
action.execute(postconditions_timeout=5)

# ---------------------------------------------------------------
# alternatively, if you don't want also the pre and postconditions
# logic in the action class, you can just use the constructor

class IncrAttrByN(TracerAction):

    def __init__(self, device, attribute_name, n):
        super().__init__()
        self.device = device
        self.attribute_name = attribute_name
        self.n = n

        for incr in range(1, n+1):
            self.add_postconditions(
                AssertDevicesStateChanges(
                    devices=[self.device],
                    attribute_name=self.attribute_name,
                    custom_matcher=lambda event:
                        self.is_attr_incremented_by_N(event, incr)
                )
            )

    def setup(self):
        super().setup()

        # store the initial value of the attribute
        # (will be useful to verify the increments)
        self.initial_value = self.device.read_attribute(self.attribute_name).value

    def execute_procedure(self):
        # the action logic that increments the attribute by n

    # define a custom matcher to verify the attribute increments
    def is_attr_incremented_by_N(self, event, incr):
        return event.attribute_value == self.initial_value + incr


action = IncrAttrByN(dev1, "attr1", 3)

# (here we can still add more preconditions if needed)

# execute (this time without a timeout)
action.execute()

NOTE: At the moment, the tracer is managed exclusively by the action itself. The timeout instead is potentially injectable from the outside as an object of type ChainedAssertionsTimeout.

NOTE: The action setters are chainable, so you can chain the calls to add preconditions and postconditions, set the timeout, etc. Example:

# this is valid code
MyTracerAction().add_preconditions(
    # ...
).add_postconditions(
    # ...
).execute(postconditions_timeout=10)

NOTE: This kind of action is particularly useful when you have to create factories of base actions whose pre and post conditions are “enriched” or “customised” according to the context in which they are used.

add_early_stop(early_stop)

Add an early stop condition for the postconditions.

Add a new early stop condition for the postconditions, which will be combined with the existing one (if any) using a logical OR.

To read more about what is an early stop condition, see ska_tango_testing.integration.assertions.with_early_stop().

Parameters:: early_stop (Callable[[ReceivedEvent], bool]) – the early stop condition to add.
Return type:: TracerAction
Returns:: the action itself, to allow chaining the calls.

add_postconditions(*postconditions, put_them_at_beginning=False)

Add one or more postconditions to the action.

Add more postconditions to the action, to be verified after the action is executed. The postconditions are verified in the order they are added, unless the put_them_at_beginning parameter is set in which case they are verified before the ones already existing.

Your postcondition tracer and timeout will be overridden by the action’s tracer and timeout. The post-condition eventual early stop will be combined with the action’s early stop.

Parameters:

postconditions (TracerAssertion) – the postconditions to add.
put_them_at_beginning (bool) – whether to put the postconditions at the beginning of the list (default is False, i.e., append them at the end). If True, the postconditions will be verified before the ones already existing.

Return type:

TracerAction

Returns:

the action itself, to allow chaining the calls.

add_preconditions(*preconditions, put_them_at_beginning=False)

Add one or more preconditions to the action.

Add more preconditions to the action, to be verified before the action is executed. The preconditions are verified in the order they are added, unless the put_them_at_beginning parameter is set in which case they are verified before the ones already existing.

Try to add preconditions that terminate immediately. If a precondition requires a tracer, it will use the same tracer as the action.

Parameters:

preconditions (SUTAssertion) – the preconditions to add.
put_them_at_beginning (bool) – whether to put the preconditions at the beginning of the list (default is False, i.e., append them at the end). If True, the preconditions will be verified before the ones already existing.

Return type:

TracerAction

Returns:

the action itself, to allow chaining the calls.

property early_stop: Callable[[ska_tango_testing.integration.event.ReceivedEvent], bool] | None

The early stop condition for the postconditions.

This early stop condition is used to stop the verification of the postconditions before the timeout expires if some kind of error condition is detected. Each postcondition may have its own early stop condition, but this one is applied to all of them.

Use add_early_stop() to add a new early stop condition, which will be combined with the existing one (if any) using a logical OR.

Returns:: the early stop condition for the postconditions.

log_postconditions

Whether to log the postconditions when verifying them.

Defaults to True.

log_preconditions

Whether to log the preconditions when verifying them.

Defaults to False.

property postconditions: list[TracerAssertion]

The postconditions of the action.

The postconditions are the assertions that need to be verified after the action is executed, that guarantee that the system is in the expected state after the action. Use add_postconditions() to add new postconditions.

Postconditions are all verified using the timeout value from the execute() method, and they all use the same tracer.

Returns:: the postconditions of the action.

property preconditions: list[SUTAssertion]

The preconditions of the action.

The preconditions are the assertions that need to be verified before the action is executed, that guarantee that the system is in a state that supports the action. Use add_preconditions() to add new preconditions. Eventual preconditions will use the same tracer.

Returns:: the preconditions of the action.

setup()

Reset the tracer and the timeout, and setup the pre/post conditions.

This setup method:

resets the tracer, unsubscribing all the events and clearing the event list;
sets up the preconditions and the postconditions (also configuring the timeout for the postconditions and the tracer for the event-based assertions).

tracer: The (managed) tracer instance to use for the pre/post conditions.

verify_postconditions(timeout=0)

Verify all the configured postconditions, within the given timeout.

This method verifies all the postconditions configured for the action, one by one and in the given order. If one of the postconditions fails, the method raises an exception and stops the verification.

NOTE: all the postconditions are verified within the given timeout, using the same tracer and considering the early stop condition.

If specified in the initialisation, the postconditions will be logged (By default, they are logged).

Parameters:

timeout (SupportsFloat) –

the time in seconds to wait for the postconditions to be verified. If not specified, it defaults to 0. Potentially, it can be:

a number, specifying the timeout in seconds,
a ChainedAssertionsTimeout object, potentially shared among multiple actions.

Raises:

AssertionError – if one of the postconditions fails.

verify_preconditions()

Verify all the configured preconditions.

This method verifies all the preconditions configured for the action, one by one and in the given order. If one of the preconditions fails, the method raises an exception and stops the verification.

If specified in the initialisation, the preconditions will be logged (By default, they are not logged).

Raises:: AssertionError – if one of the preconditions fails.

Assertions module

A collection of assertion objects and functions for use in test cases.

This module provides a collection of assertion (abstract and concrete) classes, for defining and executing verifications and synchronisations against a (Tango-based) system under test (SUT).

What is an assertion? In abstract terms, an assertion is supposed to be a generic verification of a system state or property. In an event-based system like Tango, an assertion is often a verification that some events are emitted, in the past or within a certain time frame.

Representing assertions as classes has not been an easy choice, since the risk of producing or encouraging the production of over-engineered boilerplate code is high. Before going further, we want to remind you that in testing there are many tools to assert things:

using the built-in Python assert statement
using various assertion methods provided by libraries such as assertpy,
using ska_tango_testing and ska_tango_testing.integration utilities for specifically asserting over Tango devices.

However, we decided to define a set of classes for defining assertion classes in the context of the ITH for the following reasons:

The SUT is a complex system made of many components, so sometimes you may want to group multiple lower-level checks into a few re-usable high-level assertions. Having a class structure helps in organising the assertion code into standard modular pieces.
The ITH has the purpose of engineering some aspects of the testing process, and for implementing some mechanisms you may want to separate the definition of an assertion from its execution. For example, you may want to compose a more complex object (e.g., an action - see ska_integration_test_harness.core.actions.TracerAction) injecting assertions as parameters. Representing assertions as classes makes it easier to define objects, pass them around and operate on them using a standard interface.
The SUT is a distributed system, so often the assertions are event-based. An assertion in an event-based context often needs first to subscribe to some event, and then wait for the event to happen. Having a class structure helps in defining a standard interface for the various steps required to perform the assertion (e.g., setup and verification).

Just to be clear: those classes are meant primarily to engineer internal mechanisms inside the core and inside your own extensions of the ITH. If you are writing a test case, especially if you are in a THEN step, you should prefer to directly use the assertion methods provided by assertpy and the ska-tango-testing utilities (to facilitate test readability and debugging).

So, given those premises, how are assertions implemented in the ITH? In this module we provide essentially two base classes for assertions:

SUTAssertion: a base class that defines the common interface for all the assertions in the ITH. It serves as an empty shell that defines the common shape of all assertion objects.
TracerAssertion: a subclass that represents event-based assertions that use the ska_tango_testing.integration.TangoEventTracer to perform the verifications. This class explicitly introduces the concept of a timeout and of an early stop condition, and provides you with utilities to operate with the tracer (potentially injecting it and the timeout from the outside to share them among different assertions).

Some concrete and ready-to-use assertions are:

AssertDevicesAreInState: a simple assertion that checks if a set of devices’ attributes have a specific value. Useful for defining simple non-event-based checks.
AssertDevicesStateChanges: an assertion that checks if a set of devices’ attributes change their value in a specific way. Useful for defining simple event-based checks.

Further actions and assertions (e.g, for dealing with long-running commands and subarray commands) can be found in the ska_integration_test_harness.extensions module. We suggest you look at that module before implementing your own actions and assertions from scratch.

class ska_integration_test_harness.core.assertions.AssertDevicesAreInState(devices, attribute_name, attribute_value)

Verify a set of devices are in a certain state.

This assertion verifies that the given devices right now are in a certain state (i.e., their given attribute has a certain value). This assertion is defined by:

the devices where the attribute is expected to have a value
the attribute name (attribute_name)
the expected value (attribute_value)

This assertion is expected to be verified considering the current value of the attribute, not the value recorded by some kind of event.

attribute_name: The name of the attribute to assert.

attribute_value: The value of the attribute to assert.

describe_assumption()

Describe the assertion’s assumption.

This assertion verifies that the devices have a certain attribute value.

If you extend this class, please check SUTAssertion.describe_assumption() to see how to extend it properly.

Returns:: the description of the assumption

devices: The list of devices to verify.

verify()

Verify the devices are in the expected state.

This assertion verifies that all the given devices have the attribute value equal to the expected value. If any of the devices does not have the expected value (or if the attribute is not readable), an AssertionError is raised.

Raises:: AssertionError – if the current state of the devices does not match the expected state or if the attribute is not readable.
Return type:: None

class ska_integration_test_harness.core.assertions.AssertDevicesStateChanges(devices, attribute_name, attribute_value=None, previous_value=None, custom_matcher=None, **kwargs)

Verify there are recorded state changes for the given devices.

This assertion verifies that the given devices have recorded state changes for a certain attribute, within a given timeout and without early stop events. A state change is expected to happen in one or more devices, regarding an attribute_name and could be defined by the following parameters (all optional and combinable):

by a certain value (attribute_value)
by a certain previous value (previous_value)
by a custom matcher (custom_matcher)

This assertion extends TracerAssertion and uses ska_tango_testing.integration.assertions.has_change_event_occurred() assertions to detect the events.

attribute_name: The name of the attribute that changes value.

attribute_value: The expected new value (Optional).

custom_matcher: A custom matcher for the expected event (Optional).

describe_assumption()

Describe the assumption of the assertion.

This assertion verifies that the devices {devices} have recorded state changes for the attribute {attribute_name}.

If you extend this class, please check SUTAssertion.describe_assumption() to see how to extend it properly.

Return type:: str
Returns:: the description of the assumption

devices: The devices where the state change is expected.

previous_value: The expected previous value than the expected event (Optional).

setup()

Subscribe to the state change events of the devices.

Return type:: None

verify()

Verify the devices have recorded state changes.

Verify that the devices have recorded state changes in the TangoEventTracer within the given timeout and without early stop events. The events should be related to the attribute specified in the constructor and, according to the specified parameters: :rtype: None

have a certain value (if specified)
have a certain previous value (if specified)
match a custom matcher (if specified)

Raises:: AssertionError – if the timeout is reached before the expected events are recorded or if the events are not as expected or if some early stop events are recorded.

class ska_integration_test_harness.core.assertions.SUTAssertion

A base class for an assertion on the SUT.

This class is the base class for the ska_integration_test_harness.core.assertions framework and it provides an empty shell for defining a generic assertion on the SUT.

An assertion is a verification of some kind of wide condition, probably made by many lower-level assertions grouped together, eventually event-based. An assertion:

may need a setup phase to prepare the assertion for the verification (method setup(), optional extension point)
must have a verification phase to verify the assertion (method verify(), required implementation)
should have a description of the assumption that the assertion verifies (method describe_assumption(), required implementation)

How to use an assertion as an end user: An end user can use assertions by calling in sequence the setup() method to prepare the assertion for the verification (and ensure that the assertion is in a clean state) and then the verify() method to verify the assertion.

How to extend this class: This class is basically just an empty skeleton, so subclass it and implement the compulsory verify() and describe_assumption() methods. Optionally, you can also override the setup() method.

import logging
from ska_integration_test_harness.core.assertions import SUTAssertion

class MyAssertion(SUTAssertion):

    def setup(self):
        super().setup()

        # your setup code here (e.g., subscribe to events)

    def verify(self):
        # your verification code here

    def describe_assumption(self):
        return "My assertion description"

assertion = MyAssertion()
assertion.setup()

logging.info(f"Verifying: {assertion.describe_assumption()}")
assertion.verify()

# an assertion can be reused multiple times, given that you call
# the setup method before each verification
assertion.setup()
assertion.verify()

abstract describe_assumption()

Describe the assertion’s assumption (required implementation).

This method should return a string that describes briefly what the assertion verifies. This is useful to understand the context of the assertion and to give a semantic meaning to it.

HOW TO EXTEND: override this method in your subclass to implement the description of the assumption. Return a string that describes briefly what you are verifying. The string may be single line, or multiline if needed. If you are extending from a subclass, consider calling the superclass method and appending your description to it (if you think it is useful).

Return type:: str
Returns:: the description of the assumption

setup()

Set up the assertion (optional extension point).

This method should be called before the verification procedure. It should be used to prepare the instance for a new, clear verification procedure (e.g., subscribe to events, reset some state, etc.). By default, no setup is done.

HOW TO EXTEND: override this method in your subclass to implement the setup phase. Always call the superclass method when overriding this method. Some good practices if you want to override are:

make it idempotent
inside this method, clear and set up all the subscriptions you need for the verification procedure
in the docstring of the method, specify the resources that are set up (and briefly recap also what is done by superclasses, potentially referencing their method docstring)

abstract verify()

Verify the assertion.

This method should be called to verify the assertion. It should be implemented to verify the assertion and raise an AssertionError if the assertion fails.

HOW TO EXTEND: override this method in your subclass to implement the verification procedure. Always call the superclass method when overriding this method. Some good practices when you implement are: :rtype: None

make it idempotent
inside this method, make all your assertions and fail if something is wrong
the verification can be a blocking operation that waits for some event to happen. If it is, consider SUTAssertionWTimeout
if you fail, produce a meaningful error message
in the docstring of the method, specify the resources that are verified (and briefly recap also what is done by superclasses, potentially referencing their method docstring)

Raises:: AssertionError – if the assertion fails

class ska_integration_test_harness.core.assertions.TracerAssertion(tracer=None, timeout=0, early_stop=None)

An event-based assertion based on TangoEventTracer.

This class is a SUTAssertion that is based on the events emitted by the TangoEventTracer. The assertion is event-based, so it needs to be configured with:

a valid tracer instance
a timeout for the assertion to be verified
an optional early stop condition

The tracer instance and the timeout can be managed by the assertion itself, or injected from the outside. A managed tracer will be automatically cleared of events and subscriptions during the setup phase, while an injected tracer will be left untouched. A managed timeout will be re-generated as a new fresh timeout object during the setup phase, while an injected timeout will be left untouched.

To make the tracer unmanaged, you can pass a custom tracer instance to the constructor. To make the timeout unmanaged, you can pass a custom timeout object to the constructor (if you pass a numeric value, the timeout object will be created for you and will be managed). Both the tracer and the timeout can be shared among multiple assertions (to share a timeout, check ska_tango_testing.integration.assertions.ChainedAssertionsTimeout).

NOTE: you still have to extend this class and implement the verify() and the describe_assumption() methods to make it work. In verify() implementation you can use the get_assertpy_context() to get the assertpy context you need to verify the assertion.

Usage Example:

from ska_integration_test_harness.core.assertions import (
    TracerAssertion
)

class MyAssertion(TracerAssertion):

    def verify(self):
        # your verification code here
        # - use get_assertpy_context to get the assertpy context
        #   (already configured with the tracer, the timeout and
        #   the early stop condition)
        # - use the assertpy context to verify the assertion
        #   (you can store it in a variable if you need to chain
        #   dynamically multiple assertions - e.g., in a for loop)
        context = get_assertpy_context()

        context.has_change_event_occurred(
            "my/device/1", "my_attr", 42
        ).has_change_event_occurred(
            "my/device/2", "my_attr", 43, previous_value=42
        ).has_change_event_occurred(
            "my/device/3", "my_attr", 44, previous_value=43
        )

assertion = MyAssertion()

# ---------------------------------------------------------------
# simple verification within a timeout
assertion.timeout = 5
assertion.setup()
assertion.verify()

# ---------------------------------------------------------------
# simple verification on the current events on a given
# tracer instance (setup will NOT clear up the given instance!)
assertion.tracer = my_tracer
assertion.setup()
assertion.verify()

# ---------------------------------------------------------------
# verification with an early stop condition
assertion.tracer = None  # This way the tracer will be managed
assertion.early_stop = lambda event:
    "ERROR" in str(event.attribute_value)

# ---------------------------------------------------------------
# verification of this assertion and many others within a same
# common timeout (setup will NOT clear up the timeout!)

from ska_tango_testing.integration.assertions import (
    ChainedAssertionsTimeout
)
timeout = ChainedAssertionsTimeout(5)

assertions = [assertion, MyOtherAssertion(), ...]

for assertion in assertions:
    assertion.timeout = timeout
    # (potentially they could share also the same tracer instance)
    assertion.setup()
    assertion.verify()

early_stop

The early stop condition for the assertion.

(see ska_tango_testing.integration.assertions.with_early_stop() for more details)

get_assertpy_context()

Get the assertpy context to use for the assertion.

Use this method to access a valid assertpy context that already includes: the tracer instance, the timeout and the early stop condition. Multiple calls of this method will share the same timeout object.

You can use the returned context to verify the assertion. Example:

# what you will probably call in your ``verify`` method
# implementation
self.get_assertpy_context().has_change_event_occurred(
    "my/device/1", "my_attr", 42
)

# or in a more sophisticated way...
context = self.get_assertpy_context()
for device, attr, value in [
    ("my/device/1", "my_attr", 42),
    ("my/device/2", "my_attr", 43),
    ("my/device/3", "my_attr", 44),
]:
    context.has_change_event_occurred(device, attr, value)

Return type:: Any
Returns:: the assertpy context to use for the assertion

is_timeout_managed()

Check if the used timeout is managed by the assertion.

Return type:: bool
Returns:: True if the timeout is managed by the assertion, False otherwise.

is_tracer_managed()

Check if the used tracer is managed by the assertion.

Return type:: bool
Returns:: True if the tracer is managed by the assertion, False otherwise.

setup()

Reset managed resources before the assertion is verified.

The resources that can be managed are: :rtype: None

the tracer, which if managed will be cleared of events and subscriptions;
the timeout, which if managed will be re-generated as a new fresh object (which still needs to be started).

Please, override this and subscribe to the events you need to verify the assertion.

property timeout: ska_tango_testing.integration.assertions.ChainedAssertionsTimeout

Get the timeout for the assertion to be verified.

Returns:: the timeout for the assertion to be verified

property tracer: ska_tango_testing.integration.TangoEventTracer

Get the tracer instance to use for the assertion.

Returns:: the tracer instance to use for the assertion

State Machine module

State machine implementation for managing transitions in a complex system.

This module provides a graph-based state machine implementation for navigating complex systems through sequences of state transitions. Rather than hardcoding linear action sequences, you define the possible state transitions and the machine automatically computes and executes the lowest-cost path to your target state using Dijkstra’s algorithm.

See StateMachine for the main class and usage examples.

class ska_integration_test_harness.core.state_machine.StateMachine(state_getter, transitions=None)

Automated state navigation and transition management.

This class solves the problem of navigating a complex system between states without hardcoding action sequences. Rather than manually orchestrating a series of commands to move from state A to state B, you define the possible transitions (state pairs and their associated actions), and the state machine automatically computes and executes the shortest path to your target state.

How it works: The state machine represents your system’s state space as a directed graph where states are nodes and transitions are edges. When you request to reach a target state, it uses Dijkstra’s algorithm to find the lowest-cost path from the current state to the target, then executes each transition in sequence. After each transition, the actual system state is validated to ensure it matches expectations.

Key concepts:

States: Represented as Enum members (e.g., STANDBY, ACTIVE, ERROR). You provide a callable that reads the current state from the system under test.
Transitions: A pair of (source_state, target_state) with an associated SUTAction that performs the state change. Each transition has an optional cost (default 1) that can be used to encourage or discourage certain paths. You can specify the transitions during initialization or add them later using define_transition().
Reaching a state: Call reach() with your target state. The machine finds the path and executes it, or raises an exception if no path exists.

Critical assumption: The system’s actual state must match what the state machine expects after each transition. If validation fails (actual state differs from expected), navigation stops and raises an error. This defensive design prevents silent failures due to unexpected system behaviour.

Important Associated Classes and dependencies:

StateMachineTransition: Represents individual transitions with their target state, action, accepted sources, and cost.
StateMachineNoPathError: Raised when no path exists to reach the target state.
StateMachineNavigationError: Raised when an error occurs during navigation (e.g., unexpected state after a transition).
networkx: Used for graph representation and pathfinding (Dijkstra’s algorithm) - read more: networkx.org.

Usage Example:

Consider a Tango subarray device with multiple states. You want to navigate between them using long-running commands, but the state space is non-linear and you want to avoid hardcoding sequences.

NOTE: This example is illustrative and may not correspond to the actual subarray state machine. Ask Team Badger for a more realistic example, if needed.

from enum import Enum
from ska_integration_test_harness.core.state_machine import (
    StateMachine
)
from ska_integration_test_harness.extensions.lrc import TangoLRCAction
from <...> import ObsState  # Example enum for subarray states

# Define a state getter function that reads the subarray's current
# observation state from the Tango device
from ska_control_model import ObsState
def get_current_state() -> ObsState:
    return ObsState(subarray_device.obsState)

# Create the state machine with the getter
sm = StateMachine(state_getter=get_current_state)

# (assume we have some predefined JSON inputs
# for resources, config, and scan)
resources_json = "{...}"
config_json = "{...}"
scan_json = "{...}"

# Define the main transitions following the SKA subarray pattern:
# EMPTY → IDLE (allocate resources)
sm.define_transition(
    target=ObsState.IDLE,
    action=TangoLRCAction(
        target_device=subarray_device,
        command_name="AssignResources",
        command_param=resources_json,
    ).add_lrc_completion_to_postconditions(),
    accepted_sources=[ObsState.EMPTY],
    cost=1.0,
)
# IDLE → READY (configure for observation)
sm.define_transition(
    target=ObsState.READY,
    action=TangoLRCAction(
        target_device=subarray_device,
        command_name="Configure",
        command_param=config_json,
    ).add_lrc_completion_to_postconditions(),
    accepted_sources=[ObsState.IDLE],
    cost=1.0,
)
# READY → SCANNING (start observation)
sm.define_transition(
    target=ObsState.SCANNING,
    action=TangoLRCAction(
        target_device=subarray_device,
        command_name="Scan",
        command_param=scan_json,
    ), # (no sync here, just call the command)
    accepted_sources=[ObsState.READY],
    cost=1.0,
)
# SCANNING → READY (end scan)
sm.define_transition(
    target=ObsState.READY,
    action=TangoLRCAction(
        target_device=subarray_device,
        command_name="EndScan",
    ).add_lrc_completion_to_postconditions(),
    accepted_sources=[ObsState.SCANNING],
    cost=1.0,
)
# READY → IDLE (release configuration)
sm.define_transition(
    target=ObsState.IDLE,
    action=TangoLRCAction(
        target_device=subarray_device,
        command_name="End",
    ).add_lrc_completion_to_postconditions(),
    accepted_sources=[ObsState.READY],
    cost=1.0,
)
# IDLE → EMPTY (release all resources)
sm.define_transition(
    target=ObsState.EMPTY,
    action=TangoLRCAction(
        target_device=subarray_device,
        command_name="ReleaseAllResources",
    ).add_lrc_completion_to_postconditions(),
    accepted_sources=[ObsState.IDLE],
    cost=1.0,
)
# Abort from most states to ABORTED (high cost, use only if needed)
sm.define_transition(
    target=ObsState.ABORTED,
    action=TangoLRCAction(
        target_device=subarray_device,
        command_name="Abort",
    ).add_lrc_completion_to_postconditions(),
    accepted_sources=[
        ObsState.IDLE,
        ObsState.READY,
        ObsState.SCANNING,
    ],
    cost=20.0,
)
# Restart from ABORTED or FAULT to EMPTY
sm.define_transition(
    target=ObsState.EMPTY,
    action=TangoLRCAction(
        target_device=subarray_device,
        command_name="Restart",
    ).add_lrc_completion_to_postconditions(),
    accepted_sources=[ObsState.ABORTED, ObsState.FAULT],
    cost=10.0,
)

# Check if SCANNING is reachable from the current state
if sm.is_reachable(ObsState.SCANNING):
    print("Can reach SCANNING from current state")

# Compute the path to SCANNING without executing it
path = sm.path_to(ObsState.SCANNING)
if path:
    print(f"Path to SCANNING requires {len(path)} transitions")

# Navigate to SCANNING from any current state
sm.reach(ObsState.SCANNING, timeout=60)
print(f"Subarray is now in state: {sm.current_state()}")

current_state()

Get the current state of the state machine.

Return type:: TypeVar(S, bound= Enum)
Returns:: The current state.

define_transition(target, action, accepted_sources=None, cost=1.0)

Add a transition to the state machine.

Parameters:

target (TypeVar(S, bound= Enum)) – The target state of the transition.
action (SUTAction) – The action to be performed during the transition.
accepted_sources (Optional[Collection[TypeVar(S, bound= Enum)]]) – A collection of source states from which this transition can be initiated. If None, all states are considered valid sources.
cost (float) – The cost associated with the transition (default is 1.0). It can represent time, resources, or any other metric relevant to the state machine’s operation. It should be a non-negative float value.

Return type:

StateMachine[TypeVar(S, bound= Enum)]

Returns:

The StateMachine instance (for method chaining).

is_reachable(target, source=None)

Check if the target state is reachable from the source state.

Parameters:

target (TypeVar(S, bound= Enum)) – The target state to reach.
source (Optional[TypeVar(S, bound= Enum)]) – The source state to start from. If None, uses the current state of the state machine.

Return type:

bool

Returns:

True if the target state is reachable from the source state, False otherwise.

path_to(target, source=None)

Find a path of transitions from the source state to the target state.

Parameters:

target (TypeVar(S, bound= Enum)) – The target state to reach.
source (Optional[TypeVar(S, bound= Enum)]) – The source state to start from. If None, uses the current state of the state machine.

Return type:

list[StateMachineTransition[TypeVar(S, bound= Enum)]] | None

Returns:

A list of StateMachineTransition instances representing the path from source to target. None if no valid path exists. An empty list if source and target are the same.

reach(target, timeout=0)

Reach the target state from the current state by executing the necessary transitions.

Parameters:

target (TypeVar(S, bound= Enum)) – The target state to reach.
timeout (SupportsFloat) – Optional shared timeout for the entire navigation process. The timeout is used by individual transitions during execution for any time-based assertions. If the timeout is or becomes zero before completion, navigation proceeds but individual transitions may fail if time-based assertions are not immediately satisfied.

Raises:

StateMachineNoPathError – If no valid path exists from the current state to the target state.
StateMachineNavigationError – If an error occurs during navigation through the state machine.

Return type:

None

exception ska_integration_test_harness.core.state_machine.StateMachineError: General exception for state machine errors.

exception ska_integration_test_harness.core.state_machine.StateMachineNavigationError: Exception raised when an error occurs during navigation through the state machine.

exception ska_integration_test_harness.core.state_machine.StateMachineNoPathError: Exception raised when no valid path exists from the current state to the target state in the state machine.

class ska_integration_test_harness.core.state_machine.StateMachineTransition(target, action, accepted_sources=None, cost=1.0)

Represents a transition in a state machine.

A transition is one (or possibly more) directed edge(s) in the state machine graph, connecting one or more source states to a single target state.

See StateMachine for the overall state machine implementation and usage example. See StateMachine.define_transition() for a detailed explanation of the parameters and their purpose.

execute(timeout=0)

Execute the action associated with this transition.

Parameters:: timeout (SupportsFloat) – Optional timeout for the action execution.
Return type:: None

is_destination(state)

Check if the given state is the target state of this transition.

Parameters:: state (TypeVar(S, bound= Enum)) – The state to check.
Return type:: bool
Returns:: True if the state is the target state, False otherwise.

is_valid_source(source)

Check if the given source state is valid for this transition.

Parameters:: source (TypeVar(S, bound= Enum)) – The source state to check.
Return type:: bool
Returns:: True if the source state is valid, False otherwise.

Configuration module

Configuration management for the Integration Test Harness (ITH).

This module provides classes for loading, representing, and managing configuration data for integration test environments that interact with (Tango-based) systems under test (SUT).

What is configuration in an ITH context? A configuration is a structured representation of some details you may want to isolate and encapsulate in your test code to avoid hardcoding them in the tests. Good candidates are:

Tango Resource Locators (TRLs - e.g., sys/controller/1) that you need to connect to in order to interact with the SUT. These are typically grouped in subsystems (e.g., a “CSP.LMC” subsystem may include a controller device and multiple subarray devices).
Other arbitrary configuration details that may be relevant for the test environment, such as flags to tell the test code to activate or not certain “modes” (e.g., “emulated” vs “real hardware”, “mid” vs “low”, etc.).

All these details can be fixed values or read and derived from environment variables.

How is the configuration represented? The configuration is represented as a YAML file, which can be loaded and parsed by the test code. The YAML file can contain Jinja2 templates to allow dynamic content based on environment variables or on other extras you decide to pass to the Jinja2 rendering.

Starting from the YAML file, your configuration will be loaded into an instance of ITHConfig, which acts as the main entry point to access the configuration data. Inside that class you will find a list of subsystems, each represented by an instance of SubsystemConfig, which encapsulates the details of a specific subsystem (e.g., “CSP.LMC”) and whose main structures are the devices.

Check ITHConfig and SubsystemConfig for more details and examples on how to use them.

Important

Before going further ask yourself: do you really need a configuration class? Do you really need to load the configuration from some YAML file or is it enough to just define some variables in your test code? It’s your choice, that depends if you fear more the risk of having a messy sparse hardcoded configuration in your test code, or the risk of over-engineering something that should stay simple.

Take it this way: choose to adopt it based on whether you think it will simplify your test code.

class ska_integration_test_harness.core.config.ITHConfig(*args, **kwargs)

The configuration for the Integration Test Harness (ITH).

This class loads and represents configuration data that test code uses to interact with an integration test deployment or environment. Configuration is typically loaded from a YAML file, which may contain Jinja2 templates to allow dynamic content based on environment variables.

The main content of the configuration is a list of subsystems (see SubsystemConfig), each representing a collection of Tango devices and their associated configuration details. This class also accepts arbitrary extra fields to accommodate any other configuration data that may be relevant for the test environment.

This class is strongly tied to its YAML representation, as the configuration is typically loaded from a YAML file like this:

arbitrary_config: 'value'
other_arbitrary_config: {{ VALUE_FROM_ENV }}

# nice use case to have generic tests that can be run against
# different targets (e.g., mid or low) and where behaviour will
# slightly change in the two cases based on the target:
target: {{ MID_OR_LOW | default('mid') }}

# predefined section for subsystems:
subsystems:

    # a subsystem entry
    - name: CSP.LMC
      # definition of its devices, associating a device role
      # (arbitrary string - e.g., 'controller')
      # to a Tango Resource Locator (TRL - e.g., `sys/controller/1`)
      devices:
        controller: {{ MID_OR_LOW | default('mid') }}-csp/control/0
        subarray_1: {{ MID_OR_LOW | default('mid') }}-csp/subarray/01
        subarray_2: {{ MID_OR_LOW | default('mid') }}-csp/subarray/02

    # another subsystem entry, that will exist only
    # if MID_OR_LOW is not set to 'low'
    # (e.g., it's 'mid' or not set at all)
    {% if MID_OR_LOW != 'low' %}
    - name: Dishes
      devices:
        dish_1: mid-dish/dish-manager/001
        dish_36: mid-dish/dish-manager/036
        dish_63: mid-dish/dish-manager/063
        dish_100: mid-dish/dish-manager/100
    {% endif %}

You can read such configuration from a YAML file using the from_yaml_file() class method:

from ska_integration_test_harness.core.config import ITHConfig

# Let's say environment variable VALUE_FROM_ENV is set to 42
# and MID_OR_LOW is not set, so it will default to 'mid'

ith_config = ITHConfig.from_yaml_file('path/to/config.yaml')

ith_config.arbitrary_config  # 'value'
ith_config.other_arbitrary_config  # 42

csp_lmc_config = ith_config.subsystem('CSP.LMC')
csp_lmc_config.device_trl('controller')  # 'mid-csp/control/0'
csp_lmc_config.devices_by_prefix('subarray')
# {
#   'subarray_1': 'mid-csp/subarray/01',
#   'subarray_2': 'mid-csp/subarray/02',
# }

dishes_config = ith_config.subsystem('Dishes')
dishes_config.device_trl('dish_36')
# 'mid-dish/dish-manager/036'

classmethod from_dict(config_data)

Create an ITHConfig instance from a dictionary.

This method takes care of converting the list of subsystems from dictionaries to SubsystemConfig instances

Parameters:: config_data (dict[str, Any]) – A dictionary representing the configuration data.
Return type:: ITHConfig
Returns:: An instance of ITHConfig populated with the data from the dictionary.

classmethod from_yaml_file(filename, extra_variables=None)

Load the ITH configuration from a YAML file, with Jinja2 templating.

Usage example:

let’s say we have a YAML file config.yaml with the following content:

arbitrary_config: 'value'
subsystems:
    - name: ExampleSubsystem
      other_field: {{ VALUE_FROM_ENV }}
      devices:
        controller: example/controller/1

And we have an environment variable VALUE_FROM_ENV set to 42. We can load the configuration as follows:

from ska_integration_test_harness.core.config import ITHConfig

ith_config = ITHConfig.from_yaml_file(‘config.yaml’) print(ith_config.arbitrary_config) # Output: ‘value’ example_subsystem = ith_config.subsystem(‘ExampleSubsystem’) print(example_subsystem.other_field) # Output: 42

Parameters:

filename (str) – The path to the YAML configuration file.
extra_variables (Optional[dict[str, Any]]) – Optional dictionary of extra variables to be used in Jinja2 rendering, in addition to environment variables.

Return type:

ITHConfig

Returns:

An instance of ITHConfig populated with the data from the file, including Jinja2 template rendering.

has_subsystem(name)

Check if a subsystem with the given name exists in the config.

Parameters:: name (str) – The name of the subsystem. Case insensitive.
Return type:: bool
Returns:: True if the subsystem exists, False otherwise.

subsystem(name)

Get a subsystem configuration by name.

Parameters:: name (str) – The name of the subsystem. Case insensitive.
Return type:: SubsystemConfig
Returns:: The SubsystemConfig instance.
Raises:: KeyError – If the subsystem with the given name does not exist.

to_yaml()

Serialise the ITH configuration to a YAML string representation.

Example of the resulting YAML:

arbitrary_config: 'value'
other_arbitrary_config: '42'
subsystems:
  - name: example_subsystem
    arbitrary_field: some_value
    devices:
      controller: sys/controller/1
      subarray_1: sys/subarray/1
      subarray_2: sys/subarray/2

Return type:: str
Returns:: A YAML string representation of the ITH configuration.

class ska_integration_test_harness.core.config.SubsystemConfig(*args, **kwargs)

A configuration for a specific subsystem within an integration test env.

A subsystem is a collection of Tango devices that work together to provide a specific functionality within the test environment. This class encapsulates the configuration details for connecting and interacting with such a subsystem.

The main content of the subsystem configuration is a mapping of device roles (e.g., ‘controller’, ‘subarray_1’) to their corresponding Tango Resource Locators (TRLs - e.g., sys/controller/1, sys/subarray/1), that can be useful to retrieve the devices needed for testing. Then, this class accepts arbitrary extra fields to accommodate any other configuration data that may be relevant for the subsystem.

This class is strongly tied to its YAML representation, as the configuration is typically loaded from a YAML file (see ITHConfig for details).

device_trl(role)

Get the Tango device TRL for a given role.

Parameters:: role (str) – The role of the device (e.g., ‘controller’).
Return type:: str
Returns:: The Tango device TRL as a string.
Raises:: KeyError – If the role or key does not exist.

devices_by_prefix(prefix)

Get all devices whose roles start with the given prefix.

E.g., if the devices mapping is:

{
    'controller': 'sys/controller/1',
    'subarray_1': 'sys/subarray/1',
    'subarray_2': 'sys/subarray/2',
}

and the prefix is 'subarray', the result will be:

{
    'subarray_1': 'sys/subarray/1',
    'subarray_2': 'sys/subarray/2',
}

Parameters:: prefix (str) – The prefix to filter device roles.
Return type:: dict[str, str]
Returns:: A dictionary of role to Tango device TRL mappings.

name: str = Ellipsis

to_yaml()

Serialise the subsystem configuration to a YAML string representation.

Example of the resulting YAML:

name: example_subsystem
arbitrary_field: some_value
devices:
  controller: sys/controller/1
  subarray_1: sys/subarray/1
  subarray_2: sys/subarray/2

Return type:: str
Returns:: A YAML string representation of the subsystem configuration.

Subsystem module

Subsystem device management and proxy caching.

This module provides the Subsystem class, which acts as a runtime wrapper around subsystem configuration, materialising static device definitions into live Tango device proxies.

What is a Subsystem? A Subsystem manages a collection of related Tango devices, providing lazy-loaded and cached access to them. Rather than creating device proxies eagerly (which can fail if devices are temporarily unavailable), proxies are created on first access and reused thereafter.

When to use it:

Use Subsystem when you have multiple related devices that belong together conceptually, especially if you may need to manage a variable number of them.
When you want to centralise and organise device access in a YAML configuration file instead of hardcoding TRLs scattered throughout your test code (see also config).
When you want to defer device proxy creation until a specific point in test execution, then cache and reuse those proxies across multiple steps.

Note that you can use this class standalone without the configuration system, although it is designed to work well with it.

You can either use the Subsystem base class directly or extend it with convenience methods for your specific subsystem (e.g., to provide typed accessors for specific devices or groups of devices).

Important

Do you really need this class? To avoid unnecessary complexity and over-engineering, use it only if it would help you manage a collection of related devices more effectively. If you are testing a small subsystem with few fixed devices, you will likely not need it. For larger subsystems with many devices or dynamic device sets (e.g., multiple subarrays), ask yourself: will it help you make your test code simpler and more maintainable?

See Subsystem and Subsystem.from_config() for usage examples and API details.

class ska_integration_test_harness.core.subsystem.Subsystem(device_locators, subsystem_name=None, dev_factory=None)

Runtime wrapper for subsystem device management.

This class manages the creation and caching of Tango device proxies for all devices in a subsystem. It provides convenient methods to access individual devices or groups of devices by key or prefix.

Device proxies are created lazily on first access and cached for reuse. A custom device proxy factory can be created for testing or special scenarios (e.g., mocking).

Instances can be created from a ska_integration_test_harness.core.config.SubsystemConfig using the from_config() class method, allowing you to define subsystem details in a YAML configuration file and instantiate objects directly from it. See ska_integration_test_harness.core.config for further details.

Usage example:

Let’s say you have a subsystem configuration like this:

subsystems:
    - name: CSP.LMC
      devices:
        controller: mid-csp/control/0
        subarray_1: mid-csp/subarray/01
        subarray_2: mid-csp/subarray/02

You can extend the base Subsystem class with convenience methods for your subsystem:

from ska_integration_test_harness.core.config import ITHConfig
from ska_integration_test_harness.core.subsystem import Subsystem
from <...> import AdminMode

# Optional: extend the base Subsystem class with convenience methods
# specific to your subsystem. You can also use the base class directly.
class CSPLMC(Subsystem):

    def controller(self) -> tango.DeviceProxy:
        '''Get the controller device proxy.'''
        return self.get_device("controller")

    def subarray(self, subarray_id: int) -> tango.DeviceProxy:
        '''Get a subarray device proxy by index.'''
        return self.get_device(f"subarray_{subarray_id}")

    def subarray_indices(self) -> list[int]:
        '''Get the available subarray indices.'''
        subarray_keys = self.get_device_keys(prefix="subarray_")
        return [
            int(key.split("_")[1]) for key in subarray_keys
            if key.startswith("subarray_")
        ]

# load the configuration
config = ITHConfig.from_yaml_file('config.yaml')
csp_lmc_config = config.subsystem('CSP.LMC')

# create an instance of the subsystem from the configuration
csp_lmc = CSPLMC.from_config(csp_lmc_config)

# access the device using the convenience methods
controller = csp_lmc.controller()
controller.ping()

# Note: lazy creation and caching of device proxies avoids
# connection failures if devices are temporarily unavailable
controller.adminMode = AdminMode.ONLINE

for idx in csp_lmc.subarray_indices():
    subarray = csp_lmc.subarray(idx)
    print(f"Subarray {idx}: {subarray.name()}")

If you are not interested in the configuration but you are in the subsystem infrastructure you can also use this class alone:

from ska_integration_test_harness.core.subsystem import Subsystem

my_subsystem = Subsystem(
    device_locators={
        "controller": "sys/controller/1",
        "subarray_1": "sys/subarray/1",
        "subarray_2": "sys/subarray/2",
    },
    subsystem_name="my_subsystem",
)

my_subsystem.get_device("controller").ping()
print(
    "Number of subarrays:",
    len(my_subsystem.get_device_keys(prefix="subarray_"))
)

classmethod from_config(config, dev_factory=None)

Create a Subsystem instance from a SubsystemConfig.

NOTE: Since this method is a class method, you can use it also to create subclasses instances.

Parameters:

config (SubsystemConfig) – The SubsystemConfig containing the configuration for the subsystem.
dev_factory (Optional[Callable[[str], DeviceProxy]]) – Optional factory function for creating device proxies. If not provided, tango.DeviceProxy will be used by default.

Return type:

Subsystem

Returns:

An instance of Subsystem initialized with the provided config.

get_device(device_key)

Get a device proxy for the given device key.

Parameters:: device_key (str) – The key identifying the device in the subsystem.
Return type:: DeviceProxy
Returns:: A Tango DeviceProxy for the specified device key.
Raises:: ValueError – If the device key is not found in the subsystem.

get_device_keys(prefix=None)

Get a list of device keys, optionally filtered by a prefix.

Parameters:: prefix (Optional[str]) – Optional filter to select only device keys that start with the given prefix (e.g., "subarray_" to match "subarray_1", "subarray_2", etc.).
Return type:: list[str]
Returns:: A list of device keys matching the optional prefix filter. If no prefix is provided, all device keys in the subsystem are returned.

get_devices(prefix=None)

Get a dictionary of device proxies, optionally filtered by a prefix.

Parameters:: prefix (Optional[str]) – Optional filter to select only device keys that start with the given prefix (e.g., "subarray_" to match "subarray_1", "subarray_2", etc.).
Return type:: dict[str, DeviceProxy]

(New) Common Extensions API Reference

A collection of Common Extensions for the ITH Core.

The Common Extensions are a collection of classes that extend and combine the functionalities of the ITH core module for typical SKA use cases.

While the core by itself doesn’t know anything about telescopes, this module instead embeds some knowledge about the SKA telescope and its mechanisms. That said, what is in this module is still pretty generic and is supposed to be reusable across different SKA testing repositories and SKAO teams.

Long Running Command (LRC) module

Common extension to send Long Running Commands and synchronizing on them.

This module contains actions and assertions to send Long Running Commands (LRC) to a Tango device, synchronizing on it’s completion and optionally failing early if some error code is detected in the LRC events. The library implements LRC verification assuming your devices are compliant with the SKA LMC (Long Running Command) standard.

The main access point to this module is the TangoLRCAction class, which can be used like a TangoCommandAction but with additional features to synchronize on the LRC completion and detect LRC errors to eventually fail early.

The class relies on AssertLRCCompletion to implement LRC completion verification and error detection. Probably you will not need to use this class directly.

If for some reason your devices emit LRC in different formats, with a bit of handwork you can extend the classes and override the methods to adapt them to your needs.

class ska_integration_test_harness.extensions.lrc.AssertLRCCompletion(device, expected_result_codes=ska_control_model.ResultCode.OK, **kwargs)

Assert that the LongRunningCommand has completed successfully.

This assertion is an extension of TracerAssertion that is specifically designed to verify the completion of a LongRunningCommand through monitoring longRunningCommandResult events from a Tango device.

The usage is very similar to the superclass, with the only difference that there is a further method called monitor_lrc() which will allow you to set the LRC to monitor. This is necessary because probably when you create the assertion instance, the LRC ID is not yet available.

In the creation phase, you should specify the target device and the expected result code(s). In the setup(), the assertion will subscribe the tracer to the longRunningCommandResult event of the target device. Then, after you have called the command and the LRC ID is available, you can call monitor_lrc() to set the LRC to monitor. If you don’t do it, the assertion will monitor any LRC completion events from the target device’s longRunningCommandResult attribute (which is probably not what you want).

NOTE: if for some reason your own device emits LRC in different formats, you can extend this class and override the match_lrc_completion() method to implement your own custom matcher.

NOTE: this assertion is designed to be used in the TangoLRCAction to verify the completion of a LongRunningCommand. If you use it standalone (e.g., passing it directly to a TracerAction), you should take care of the LRC ID management (or accept that any LRC ID will match).

describe_assumption()

Describe the assertion’s assumption (required implementation).

This method should return a string that describes briefly what the assertion verifies. This is useful to understand the context of the assertion and to give a semantic meaning to it.

HOW TO EXTEND: override this method in your subclass to implement the description of the assumption. Return a string that describes briefly what you are verifying. The string may be single line, or multiline if needed. If you are extending from a subclass, consider calling the superclass method and appending your description to it (if you think it is useful).

Returns:: the description of the assumption

device: The target device to verify.

expected_attribute_name

The expected attribute name for the LRC completion event.

By default, it’s longRunningCommandResult.

expected_result_codes

The result codes that are accepted.

By default, it accepts only the OK result code.

lrc_id: str | None

The LRC ID to monitor.

We don’t know it yet, so it’s None by default. Until it’s not set, the assertion will match any LRC completion event from the target device.

It is returned when calling the Tango command that starts the LRC.

match_lrc_completion(event)

Check if an event is a LRC completion event (with expected result).

This method is a custom matcher that will be used by the assertion to check if an event is a LRC completion event, the format is the expected one and:

if given, check the LRC ID is the expected one (self.lrc_id)
if given, check the result code is one of the expected ones (self.expected_result_codes)

The following is the expected format of the event value:

(lrc_id, '[RESULT_CODE, "result message"]')

NOTE: this method can be overridden to implement custom matchers for different LRC completion event formats. This method can also be used as a standalone predicate to be passed as an early stop predicate in ska_integration_test_harness.core.actions.TracerAction or ska_integration_test_harness.core.assertions.TracerAssertion instances. When the method is used as a standalone matcher, you can defer the LRC ID setting and the expected result codes setting.

Parameters:: event (ReceivedEvent) – the event to check.
Return type:: bool
Returns:: True if the event is a LRC completion event with the expected result code, False otherwise.

monitor_lrc(lrc_id)

Set the LRC to monitor.

NOTE: this method supports the chaining.

Parameters:: lrc_id (str) – the LRC ID to monitor.
Return type:: AssertLRCCompletion
Returns:: this instance for chaining.

setup(): Subscribe to the LRC completion event.

verify()

Verify the LRC completes and has the expected result code.

Verify that the device reports an event on the longRunningCommandResult attribute with the expected ska_control_model.ResultCode value for the LRC ID configured through monitor_lrc().

NOTE: before calling this method, you should call monitor_lrc() to set the LRC ID to monitor. If you don’t call it, the assertion will match any LRC completion event from the target device.

class ska_integration_test_harness.extensions.lrc.TangoLRCAction(target_device, command_name, command_param=None, command_kwargs=None)

Send a LongRunningCommand to a Tango device and synchronise.

This class represents an action that sends a LongRunningCommand to a Tango device and then synchronises on its successful completion (and possibly on its errors too). This class is an extension of ska_integration_test_harness.core.actions.TangoCommandAction, which inherits the capability of sending Tango commands and synchronising through a set of ska_integration_test_harness.core.assertions.TracerAssertion, but adds two additional features:

the possibility to add to postconditions a check on the completion of the LRC, through the method add_lrc_completion_to_postconditions();
the possibility to add to early stop a check on the errors of the LRC, through the method add_lrc_errors_to_early_stop().

Usage example:

from ska_integration_test_harness.core.actions import TangoCommandAction
from ska_integration_test_harness.core.assertions import AssertDevicesAreInState
from ska_integration_test_harness.core.assertions import AssertDevicesStateChanges

# Then you can build action instances and add preconditions and
# postconditions to them according to your needs.

action = TangoCommandAction(
    target_device=dev1,
    command_name="IncreaseAttributeLRC",
    command_param=2,
).add_preconditions(
    AssertDevicesAreInState(
        devices=[dev1, dev2],
        attribute_name="attr1",
        attribute_value=42
    ),
).add_postconditions(
    AssertDevicesStateChanges(
        devices=[dev1, dev2],
        attribute_name="attr1",
        attribute_value=43
    ),
    AssertDevicesAreInState(
        devices=[dev1, dev2],
        attribute_name="attr1",
        attribute_value=44
    ),
).add_early_stop(
    lambda e: e.attribute_value < 42

# synchronise on LRC completion (after the state changes)
).add_lrc_completion_to_postconditions(

# monitor LRC errors (if any)
).add_lrc_errors_to_early_stop()

# execute the action within a timeout of 5 seconds
# (which will stop early if the LRC fails or if it detects
# an event with an attribute value less than 42)
action.execute(postconditions_timeout=5)

Two additional notes:

add_lrc_completion_to_postconditions() and add_lrc_errors_to_early_stop() can be customised passing sets of result codes that you want to consider as successful or as errors. See the methods documentation for more information.
add_lrc_completion_to_postconditions() can be called multiple times with different result codes to track different stages of the LRC completion.

What if your LRC is not exactly as expected?

If your code does not support LRC the way we expect, but it still does that in a slightly different way, you can still use this class by extending it and overriding it where necessary. For example, let’s say:

The command result code is returned in a slightly different format
The event that signals the LRC completion is not emitted on longRunningCommandResult, but on a different attribute
The LRC events have a different format than the one supported by AssertLRCCompletion

To handle this:

You override get_last_lrc_id() to extract the LRC ID from your own command result format
You subclass AssertLRCCompletion and override the method match_lrc_completion() to match your own LRC event format.
You also override the constructor of the assertion class to set your own expected attribute name.
You override _create_assert_lrc_completion_instance() to use your own subclass instead of the default one.

Example:

from ska_integration_test_harness.extensions.lrc import (
    AssertLRCCompletion, TangoLRCAction
)

# 2,3. Subclass AssertLRCCompletion to match your own LRC event format
# and set your own expected attribute name
class MyAssertLRCCompletion(AssertLRCCompletion):

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.expected_attribute_name = "myLRCEvent"

    def match_lrc_completion(self, event: ReceivedEvent) -> bool:
        # your custom implementation here

# 1,4. Subclass TangoLRCAction
class MyTangoLRCAction(TangoLRCAction):

    def get_last_lrc_id(self) -> str:
        # your custom implementation here to extract the LRC ID
        return # ...

    def _create_assert_lrc_completion_instance(
        self,
        expected_result_codes: "ResultCode | list[ResultCode] | None",
    ) -> MyAssertLRCCompletion:
        return MyAssertLRCCompletion(
            # don't forget to pass the target device
            device=self.target_device,
            expected_result_codes=expected_result_codes,
        )

    # Now you can use MyTangoLRCAction instead of TangoLRCAction
    my_lrc_action = MyTangoLRCAction(
        target_device=dev1,
        command_name="IncreaseAttributeLRC",
        command_param=2,
    ).add_lrc_completion_to_postconditions(
        expected_result_codes=ResultCode.OK
    ).add_lrc_errors_to_early_stop()

add_early_stop(early_stop)

Add an early stop condition for the postconditions.

Add a new early stop condition for the postconditions, which will be combined with the existing one (if any) using a logical OR.

To read more about what is an early stop condition, see ska_tango_testing.integration.assertions.with_early_stop().

Parameters:: early_stop – the early stop condition to add.
Return type:: TangoLRCAction
Returns:: the action itself, to allow chaining the calls.

add_lrc_completion_to_postconditions(expected_result_codes=ska_control_model.ResultCode.OK, put_at_beginning=False)

Add a postcondition to verify the completion of the LRC.

Call this method to append to postconditions an assertion to check the LRC completes with some ska_control_model.ResultCode. This is done through an AssertLRCCompletion instance, which assert over events on the longRunningCommandResult attribute of the target device. The LRC ID will be extracted from the last command execution result and this class will automatically be configured in all the assertions of this type you add to the action.

NOTE: to accept multiple result codes as successful, you can pass a list of them when calling this method. If you instead want to track a whole sequence of LRC events with different result codes, you can call this method multiple times.

Parameters:

expected_result_code –
the expected result code of the LRC. You can:
- leave it to the default value (ResultCode.OK) to accept only the OK result code;
- set it to None to accept any result code;
- set it to one or more result codes to accept multiple result codes (by passing a single ResultCode or a list of them).
put_at_beginning – if True, you will verify the LRC completion before the other postconditions you configured. By default, it is added at the end.

Returns:

the action itself, to allow chaining the calls.

add_lrc_errors_to_early_stop(error_result_codes=None)

Add an early stop condition to stop early if the LRC fails.

Call this method to add an early stop condition to stop the action early if some kind of LRC ska_control_model.ResultCode is detected among events. This method will build an early stop predicate that will applied in all the existing an future postconditions. The LRC ID will be extracted from the last command execution result and this class will automatically be configured in all the early stop assertions of this type you add to the action.

Parameters:

error_result_codes (Optional[list[ResultCode]]) –

the result codes to consider as errors. By default, the following result codes are considered as errors:

[ResultCode.FAILED, ResultCode.REJECTED, ResultCode.NOT_ALLOWED]

You can override this default by passing a result code or a list of them that you want to consider as errors.

ResultCode.FAILED
ResultCode.REJECTED
ResultCode.NOT_ALLOWED

Return type:

TangoLRCAction

add_postconditions(*postconditions, put_them_at_beginning=False)

Add one or more postconditions to the action.

Add more postconditions to the action, to be verified after the action is executed. The postconditions are verified in the order they are added, unless the put_them_at_beginning parameter is set in which case they are verified before the ones already existing.

Your postcondition tracer and timeout will be overridden by the action’s tracer and timeout. The post-condition eventual early stop will be combined with the action’s early stop.

Parameters:

postconditions – the postconditions to add.
put_them_at_beginning – whether to put the postconditions at the beginning of the list (default is False, i.e., append them at the end). If True, the postconditions will be verified before the ones already existing.

Return type:

TangoLRCAction

Returns:

the action itself, to allow chaining the calls.

add_preconditions(*preconditions, put_them_at_beginning=False)

Add one or more preconditions to the action.

Add more preconditions to the action, to be verified before the action is executed. The preconditions are verified in the order they are added, unless the put_them_at_beginning parameter is set in which case they are verified before the ones already existing.

Try to add preconditions that terminate immediately. If a precondition requires a tracer, it will use the same tracer as the action.

Parameters:

preconditions – the preconditions to add.
put_them_at_beginning – whether to put the preconditions at the beginning of the list (default is False, i.e., append them at the end). If True, the preconditions will be verified before the ones already existing.

Return type:

TangoLRCAction

Returns:

the action itself, to allow chaining the calls.

get_last_lrc_id()

Extract the LRC ID from the last command result.

Or fail if the last command result is not as expected.

Return type:: str
Returns:: the LRC ID from the last command result.
Raises:: AssertionError – if the last command result is not as expected.

setup()

Subscribe to all the necessary events, including the LRC event.

See ska_integration_test_harness.core.actions.TracerAction.setup() for more information.

verify_postconditions(timeout=0)

Verify the postconditions of the action.

But before, ensure all the LRC assertions are configured with the LRC ID from the last command result.

See ska_integration_test_harness.core.actions.TracerAction.verify_postconditions() for more information.

Parameters:

timeout (SupportsFloat) –

the time in seconds to wait for the postconditions to be verified. If not specified, it defaults to 0. Potentially, it can be:

a number, specifying the timeout in seconds,
a ChainedAssertionsTimeout object, potentially shared among multiple actions.

Raises:

AssertionError – if one of the postconditions fails.

Return type:

None