Generic mechanisms API (TMC Monolith)

Here below are documented the generic mechanisms of the ska-integration-test-harness. Implementation of specific actions and structures are currently left outside, see directly the code for more details.

IMPORTANT NOTE: A very crucial aspect of the generic mechanisms is that they are unit tested code, while the specific actions and structures are not unit tested (and in a certain sense, the integration tests that use the test harness are the validation of the correctness of the implementation of the specific actions and structures). The coverage metrics of the test harness are calculated only on the generic mechanisms (at the time of writing, the coverage on the generic mechanisms is approximately 90%, while the overall coverage is around 60%).

Structure module

A collection of wrappers for various system components.

TelescopeWrapper is the main entry point for the telescope system.

Right now, it exposes the following subsystems (through a wrapper for each):

  • TMC

  • SDP

  • CSP

  • Dishes

All the subsystem wrappers are represented by abstract classes, so you can create your own implementation for each one (e.g., an implementation that points to the devices of a real subsystem, or an implementation that points to the devices of an emulated subsystem).

The module also contains a generic SubsystemWrapper class that can be used as a base class for any subsystem wrapper.

class ska_integration_test_harness.structure.CSPWrapper(csp_configuration)

A test wrapper for the CSP.

clear_command_call()

Clear the command call on the CSP (if needed).

Return type:

None

get_all_devices()

Get all the subsystem devices as a dictionary.

Return type:

dict[str, DeviceProxy]

get_subsystem_name()

Get the name of the subsystem.

Return type:

str

set_subarray_id(subarray_id)

Set the subarray ID on the CSP subarray.

Return type:

None

tear_down()

Tear down the CSP (if needed).

Return type:

None

class ska_integration_test_harness.structure.DishesWrapper(dishes_configuration)

A test wrapper for the dishes.

clear_command_call()

Clear the command call on the Dishes (if needed).

Return type:

None

property dish_master_list: list[tango.DeviceProxy]

Dish Master device proxies as a list (sorted by key).

get_all_devices()

Get all the subsystem devices as a dictionary.

Return type:

dict[str, DeviceProxy]

get_subsystem_name()

Get the name of the subsystem.

Return type:

str

tear_down()

Tear down the dishes (if needed).

Return type:

None

class ska_integration_test_harness.structure.SDPWrapper(sdp_configuration)

A test wrapper for the SDP.

clear_command_call()

Clear the command call on the SDP (if needed).

Return type:

None

get_all_devices()

Get all the subsystem devices as a dictionary.

Return type:

dict[str, DeviceProxy]

get_subsystem_name()

Get the name of the subsystem.

Return type:

str

set_subarray_id(subarray_id)

Set the subarray ID on the SDP subarray.

Return type:

None

tear_down()

Tear down the SDP (if needed).

Return type:

None

class ska_integration_test_harness.structure.SubsystemWrapper

The abstract definition of a subsystem wrapper.

A subsystem is a part of the telescope involved in the tests (e.g. CSP, TMC, Dishes, etc.). A subsystem has the following properties:

  • has a name

  • has Tango devices

  • can be emulated or not

On a subsystem, we can perform the following operations:

  • access the properties described above

  • get a recap of the subsystem information (system name, emulated or not, devices information)

This class represents the abstract definition of a subsystem wrapper. Extend this class and implement the abstract methods to create a subsystem wrapper for a specific subsystem. Your child class may be also abstract, if you want for example to distinguish between emulated and production subsystems.

abstract get_all_devices()

Get all the subsystem devices as a dictionary.

Return type:

dict[str, DeviceProxy]

Returns:

A dictionary of device proxies, where the key is an unique identifier of the device and the value is the device proxy.

NOTE: the unique identifier is not necessarily the device name, but more something that explains “what role” the device has in the subsystem (e.g., “central_node”, “dish_1”, etc.).

get_recap(devices_info_provider=None)

Get a (string) recap of the subsystem information.

The recap contains the subsystem name, the emulated status and the devices information. The devices information is a list of the devices in the subsystem, eventually enriched with further information (e.g., version, etc.).

Parameters:

devices_info_provider (Optional[DevicesInfoProvider]) – The provider to get info about Tango devices from ska-k8s-config-exporter. If None, only the device names are included in the recap.

Return type:

str

abstract get_subsystem_name()

Get the name of the subsystem.

Return type:

str

Returns:

The name of the subsystem.

abstract is_emulated()

Check if the subsystem is emulated.

Return type:

bool

Returns:

True if the subsystem is emulated, False otherwise.

class ska_integration_test_harness.structure.TMCWrapper(tmc_configuration)

A wrapper for the TMC component.

get_all_devices()

Get all the subsystem devices as a dictionary.

Return type:

dict[str, DeviceProxy]

get_recap(devices_info_provider=None)

Get a (string) recap of the subsystem information.

The recap contains the subsystem name, the emulated status and the devices information. The devices information is a list of the devices in the subsystem, eventually enriched with further information (e.g., version, etc.).

Parameters:

devices_info_provider (Optional[DevicesInfoProvider]) – The provider to get info about Tango devices from ska-k8s-config-exporter. If None, only the device names are included in the recap.

Return type:

str

get_subsystem_name()

Get the name of the subsystem.

Return type:

str

is_subarray_initialised()

Check if the subarray is initialised

Return type:

bool

set_subarray_id(subarray_id)

Set subarray ID

supports_low()

Check if the configuration supports the low target environment.

Return type:

bool

supports_mid()

Check if the configuration supports the mid target environment.

Return type:

bool

abstract tear_down()

Reset the TMC devices to their initial state.

Return type:

None

class ska_integration_test_harness.structure.TelescopeWrapper

A wrapper class that contains all the telescope subsystems.

This infrastructural class is used as an unique entry point to access all the telescope subsystems and its devices. Given an instance of this class, using its properties, it is possible to access the TMC, SDP, CSP, and Dishes subsystems.

This class is a Singleton, so this mean that there is only one instance of it in the entire code. This is done to avoid the creation of multiple “telescope” instances, potentially inconsistently initialised with different subsystem instances (which may be configured differently).

To initialise the telescope test structure, create an instance of this class and call the set_up method with the instances of the TMC, SDP, CSP, and Dishes subsystems. After the initialisation, in any point of the code you can create an instance of this class and have it already initialised with the subsystems.

# Initialise the telescope test structure
telescope = TelescopeWrapper()
telescope.set_up(tmc, sdp, csp, dishes)

# ...

# In any point of the code, you have access to the subsystems
# (and their devices) using the properties of the telescope instance.
telescope = TelescopeWrapper()
do_something(telescope.tmc.central_node)

When you are done with testing, you can tear down the entire telescope test structure and reset it to the initial state calling the tear_down method.

# Tear down the telescope test structure
telescope = TelescopeWrapper()
telescope.tear_down()

The Singleton is a pretty much standard design pattern. To learn more about it, you can refer to the following resources.

actions_default_timeout: int = 60

The default timeout (in seconds) used in the telescope actions.

clear_command_call()

Clear the command call on the telescope (if needed).

Raises:

ValueError – If one or more subsystems are missing.

Return type:

None

property csp: CSPWrapper

A wrapper for the CSP subsystem and its devices.

Raises:

ValueError – If one or more subsystems are missing.

devices_info_provider: DevicesInfoProvider | None = None

The devices info provider used to access the devices information.

(Used for recap purposes).

property dishes: DishesWrapper

A wrapper for the dishes subsystem and its devices.

Returns:

The DishesDevices instance.

Raises:

ValueError – If one or more subsystems are missing.

fail_if_not_set_up()

Fail if a valid structure is not set up.

Raises:

ValueError – If one or more subsystems are missing.

Return type:

None

get_active_subsystems()

Get all the active subsystems.

Return type:

list[SubsystemWrapper]

Returns:

The list of all the subsystems.

get_required_subsystems()

Get all the required subsystems (based on the TMC configuration).

Return type:

dict[SubsystemWrapper]

Returns:

The list of all the required subsystems.

get_subsystems_recap(update_devices_info=True)

Get a recap of the active subsystems and their devices.

Get a recap of the active subsystems, their production-emulated status, and the devices they contain.

Parameters:

update_devices_info (bool) – If True, the devices info provider is updated before getting the subsystems recaps.

Return type:

str

Returns:

The recap string.

property mccs: MCCSWrapper

A wrapper for the MCCS subsystem and its devices.

Returns:

The MCCSDevices instance.

Raises:

ValueError – If one or more subsystems are missing.

property sdp: SDPWrapper

A wrapper for the SDP subsystem and its devices.

Raises:

ValueError – If one or more subsystems are missing.

set_subarray_id(subarray_id)

Create subarray devices for the requested subarray.

Parameters:

subarray_id (int) – The Subarray ID to set.

Raises:

ValueError – If one or more subsystems are missing.

Return type:

None

set_up(tmc, sdp, csp, dishes=None, mccs=None)

Initialise the telescope test structure with the given devices.

Parameters:

  • tmc (TMCWrapper) – The TMC subsystem wrapper.

  • sdp (SDPWrapper) – The SDP subsystem wrapper.

  • csp (CSPWrapper) – The CSP subsystem wrapper.

  • dishes (Optional[DishesWrapper]) – The Dishes subsystem wrapper (required for mid).

  • mccs (Optional[MCCSWrapper]) – The MCCS subsystem wrapper (required for low).

Return type:

None

tear_down()

Tear down the entire telescope test structure.

Raises:

ValueError – If one or more subsystems are missing.

Return type:

None

property tmc: TMCWrapper

A wrapper for the TMC subsystem and its devices.

Returns:

The TMCDevices instance.

Raises:

ValueError – If one or more subsystems are missing.

Actions module

A structure for performing actions on the telescope system.

This module contains the structure for implementing actions on the telescope. An action is essentially an abstraction of a procedure that can be executed on the telescope system and - essentially - it is made by a business logic and a termination condition to synchronise the execution. Actions then can be represented through a hierarchy of classes, which have as a root the TelescopeAction.

The main mechanism is the one provided by the ska_tango_testing.integration.tracer.TangoEventTracer that allows the implementation of a StateChangeWaiter to wait for a specific state change in the system and so to synchronise after an action is executed.

Further details in the classes documentation.

class ska_integration_test_harness.actions.StateChangeWaiter

A tool to wait for a set of state changes from multiple Tango devices.

This class is used to wait for a set of state changes to occur on multiple Tango devices. You use this class by creating an instance of it, progressively adding state changes to wait for, and then calling the wait_all method to wait for all the state changes to occur (or for a timeout to occur). If all the state changes occur before the timeout, the execution continues, else it will be raised an exception.

An instance of this class can be shared among multiple classes, so you can separate the responsibility of knowing what conditions to wait foreach group of devices (e.g., a TMC class may know what to wait for the TMC devices, and a CSP class may know what to wait for the CSP devices, etc.) and also move the wait_all call in a common orchestrator class (that doesn’t need to know the details of what to wait for).

Calling the method reset you will clear the list of expected state changes, so you can reuse the same instance for multiple actions.

add_expected_state_changes(state_changes)

Add a list of expected state changes to wait for.

Parameters:

state_changes (list[ExpectedEvent]) – A list of expected state changes to wait for.

Return type:

None

reset()

Clear the list of expected state changes.

wait_all(timeout)

Wait for all the expected state changes to occur.

Parameters:

timeout (int | float) – The maximum time (in seconds) to wait for the state changes.

Raises:

TimeoutError – If not all the expected state changes occurred within the timeout.

Return type:

None

class ska_integration_test_harness.actions.TelescopeAction

A generic action executed on the telescope and its subsystems.

An action is made by:

  • the action itself, which is the procedure that interacts with telescope subsystems (TMC, CSP, SDP, Dishes);

  • a termination condition, which is a set of expected events that should occur after the action is executed and which define a successful completion of the action;

  • a return type T, which is the type of the expected result of the action.

This class is a template for such actions.

SUBCLASS AN ACTION

To create a new action, you create a subclass of TelescopeAction and you implement the abstract methods _action() and termination_condition(). You define your business logic in the _action() method and you define the termination condition in the termination_condition() method, as a list of tests.test_harness3.telescope_actions.expected_events.ExpectedEvent instances (or subclasses).

Here a few guidelines about the usage.

  • If your action needs some parameters, you can override the __init__() method to accept them and store them as instance attributes.

  • When you implement an action you have to specify a return type. If your action doesn’t return anything, you can specify None as return type.

  • Even if the action has no termination condition, you should still specify that by returning an empty list of expected events in the termination_condition() method.

  • If in your action you need to access the telescope instance, you can do it by using the attribute self.telescope. If you need to log something (only if the logging policy is active), you can use the method _log(). If for some reason you need, you can also access the internal configuration and the components.

Usage example:

from tango import DevState
from ska_integration_test_harness.actions.telescope_action import (
    TelescopeAction
)
from ska_integration_test_harness.actions.expected_events import (
    ExpectedStateChange
)


# create an action that returns nothing
# and has no termination condition. It takes a parameter through.
class MyAction(TelescopeAction[None]):
    def __init__(self, my_parameter: int):
        super().__init__()
        self.my_parameter = my_parameter

    def _action(self):
        # your business logic here
        pass

    def termination_condition(self):
        return []

# create an action that - if necessary - runs
# a certain Tango command on TMC central node and terminates
# when that device reaches a certain state. The action then
# returns true or false depending if the command was necessary or not.
# Some further logging is done if the command wasn't necessary.
class CentralNodeMoveToOn(TelescopeAction[bool]):
    def _action(self):
        if self.telescope.tmc.central_node.State != DevState.ON:
            self.telescope.tmc.central_node.On()
            return True

        self._log(
            "Central node is already ON. No need to run the command."
        )
        return False

    def termination_condition(self):
        return [
            ExpectedStateChange(
                device=self.telescope.tmc.central_node,
                attribute="State",
                expected_value=DevState.ON,
            )
        ]

EXECUTE AN ACTION

To run an action, you call the method execute(). By default, when called the method:

  • the action is executed;

  • the termination condition is waited for until it occurs;

  • some logging is performed to report the execution beginning and end;

  • if the termination condition does not occur within a timeout, a TimeoutError is raised.

Before calling the execute() method, you can customise some configurations of the action:

NOTE: An already configured action can be executed multiple times.

Usage example:

# create an instance of the action
action = MyAction(my_parameter=42)

# [optional] customise the timeout for the termination condition
action.set_termination_condition_timeout(10)

# [optional] execute the action without waiting for the
# termination condition to occur
action.set_termination_condition_policy(False)

# [optional] deactivate the logging policy
action.set_logging_policy(False)

# execute the action
action.execute()

# execute the action again
action.execute()

RATIONALE AND INSPIRATIONS

This class is strongly inspired by the Command design pattern (https://refactoring.guru/design-patterns/command), since it abstracts a potentially complex action into a class, incapsulating and hiding the execution details of the action itself, such as having available a telescope instance and waiting for the termination condition to occur. Since it leaves you some methods to implement (that will instead be called by this class itself), it is also inspired by the Template Method design pattern (https://refactoring.guru/design-patterns/template-method).

do_logging

If True, the action will log its execution beginning and end.

execute()

Execute the action.

By default, when called the method: :rtype: Optional[Any]

  • the action is executed;

  • the termination condition is waited for until it occurs;

  • some logging is performed to report the execution beginning and end;

  • if the termination condition does not occur within a timeout, a TimeoutError is raised.

Before calling the execute() method, you can customise some configurations of the action:

NOTE: An already configured action can be executed multiple times.

Usage example:

# create an instance of the action
action = MyAction(my_parameter=42)

# [optional] customise the timeout for the termination condition
action.set_termination_condition_timeout(10)

# [optional] execute the action without waiting for the
# termination condition to occur
action.set_termination_condition_policy(False)

# [optional] deactivate the logging policy
action.set_logging_policy(False)

# execute the action
action.execute()

# execute the action again
action.execute()
Raises:

TimeoutError – If the expected outcome does not occur within a timeout.

get_last_execution_result()

Get the result of the last execution of the action (if any).

Return type:

Optional[TypeVar(T, bound= object)]

Returns:

The result of the last execution of the action (if any).

set_logging_policy(do_logging)

Change the policy for logging the action.

If you call this method with do_logging=False, the action will not log its execution beginning and end. If you call this method with do_logging=True, the action will log its execution beginning and end.

Parameters:

do_logging (bool) – If True, the action will log its execution beginning and end. If False, it will not.

Return type:

None

set_termination_condition_policy(wait)

Change the policy for waiting for the termination condition.

If you call this method with wait=False, the termination condition will not be waited for when calling the execute() method. If you call this method with wait=True, the termination condition will be waited for when calling the execute() method.

Parameters:

wait (bool) – If True, the termination condition will be waited for when calling the execute() method. If False, the termination condition will not be waited for.

Return type:

None

set_termination_condition_timeout(timeout)

Change the timeout for the termination condition.

The timeout is the maximum time to wait for the termination condition to occur. You can change it at any time before calling the execute() method.

Parameters:

timeout (int | float) – The new timeout for the termination condition (in seconds).

Return type:

None

telescope

The telescope instance, which you can use to access all the subsystem devices (TMC, CSP, SDP, Dishes).

abstract termination_condition()

The termination condition of the action.

The termination condition is a list of expected events. Override this method to define the expected events that should occur. Remember that by default this method is called BEFORE the action is executed, so you can access the devices states and attributes and store some useful information to define the termination condition.

Also remember that the termination condition is being waited AFTER the action procedure is executed. So, you can define a termination condition that is based on the action result (through the use opportune lambda functions based on the call of the action result getter get_last_execution_result).

Return type:

list[ExpectedEvent]

Returns:

A list of expected events that define the termination condition of the action. They should be instances of tests.test_harness3.telescope_actions.expected_events.ExpectedEvent or subclasses, like the very useful tests.test_harness3.telescope_actions.expected_events.ExpectedStateChange.

termination_condition_timeout: float

The timeout for the termination condition (in seconds).

It defaults to the one specified in the telescope wrapper class (at the moment of the action creation).

wait_termination: bool

If True, the termination condition will be waited for.

class ska_integration_test_harness.actions.TelescopeActionSequence(steps)

A sequence of TelescopeAction, executed in order.

This action is used to group a sequence of actions together, so that the can be executed as a single action. The sub-actions are executed in the order they are provided and the synchronisation is done after each sub action (step). By default, this action does not have further termination conditions.

By default, each step keeps the default wait termination condition timeout. Calling the set_termination_condition_timeout(timeout) method you will apply the change to each of the steps.

By default, the termination condition policy is set to wait for the termination condition of each of the steps. Calling set_termination_condition_policy(False) you will make the last step to not wait for the termination condition (but all the others will still keep their previous policy). Calling set_termination_condition_policy(True) you will make all the steps to wait for the termination condition. If you need, you can set the termination condition policy for each step by calling the method on each of them (you can access them through steps).

Usage example:

# Create a sequence of actions
sequence = TelescopeActionSequence([
    action1,
    action2,
    action3,
])

# Execute the sequence
result = sequence.execute()

# (you can always access the steps, set policies, etc.)
sequence.steps[0].set_termination_condition_timeout(10)
sequence.set_logging_policy(True)
# ...
set_logging_policy(do_logging)

Propagate a new logging policy to each of the steps.

Parameters:

do_logging (bool) – The new logging policy value.

Return type:

None

set_termination_condition_policy(wait)

Set the termination condition policy for the steps as follows.

  • set to True: all the steps will wait for the termination condition.

  • set to False: the last step will not wait for the

    termination condition.

Parameters:

wait (bool) – The new policy value.

Return type:

None

set_termination_condition_timeout(timeout)

Propagate a new timeout value to each of the steps.

Parameters:

timeout (int | float) – The new timeout value.

Return type:

None

termination_condition()

The sequence by itself does not have a termination condition.

The termination condition is handled by the steps. If configured, the termination condition of the last step will be used.

Returns:

An empty list.

class ska_integration_test_harness.actions.TelescopeCommandAction(target_device=None, is_long_running_command=False)

An action that send a command to some telescope subsystem.

Such an action:

  • may have a target encoded in the action, which can be a Tango device

  • may be a LRC (long running command) or not

  • returns as a tuple the command result

If the command is a LRC (and a target device is provided), by default the action will wait for the LRC to terminate. If you want to extend the termination condition to wait for more things or not to wait for the LRC to terminate, you can override the termination_condition method (and include or not the result of the superclass method in your own returned list). Actually, just changing the is_long_running_command attribute to False you can avoid waiting for the LRC to terminate.

The result of the command is a tuple with two elements:

  • the first element is the result of the command (e.g., ResultCode.OK)

  • the second element is one or more messages about the command execution.

If the command is a LRC, the second element can be used as a reference to check the command status through the longRunningCommandResult attribute. In fact, the waiting condition for the LRC termination is based on this attribute.

IMPORTANT NOTE: at the moment, the command itself is not part of this action. This is because I don’t yet want to interfere with the way you call it (e.g., with the command name, the command input, etc.). This may be subject to change in the future, but for now, you have to implement the command execution in the _action method as you would do in a normal action.

is_long_running_command

Whether the command is a long running command or not.

target_device

The Tango device to which the command will be sent.

termination_condition()

Wait for the LRC to terminate.

If the command is a LRC (and a target device is provided), by default the action will wait for the LRC to terminate. If you want to extend the termination condition to wait for more things or not to wait for the LRC to terminate, you can override the termination_condition method (and include or not the result of the superclass method in your own returned list).

Return type:

list[ExpectedEvent]

Returns:

A list of ExpectedEvent objects to wait for (which is empty by default, or includes the termination condition for the LRC if the command is marked as a LRC).

termination_condition_for_lrc()

Wait for the LRC to terminate.

Implement this method to specify the termination condition for when the command is a long running command.

Return type:

list[ExpectedEvent]

Returns:

A list of ExpectedEvent objects to wait for, when the command is a long running command.

class ska_integration_test_harness.actions.TransientQuiescentCommandAction(target_device=None, is_long_running_command=False, synchronise_on_transient_state=False)

A command which can synchronise on a quiescent or on a transient state.

This class represents a command that when called can be synchronised on a quiescent state (the final expected state) or on a transient state (an intermediate state in the command execution). Concretely, this means that the action has two instead of one termination conditions: one for the quiescent state and one for the transient state. By default, the action will synchronise on the quiescent state. If you want to synchronise on the transient state, you can set the synchronise_on_transient_state attribute to True.

This is a subclass of TelescopeCommandAction, so it still inherits the fact of having a target device and being a LRC or not. If this action is set as a LRC, when you synchronise on the quiescent state, the action termination condition by default will include the LRC termination condition.

set_synchronise_on_transient_state(value)

Set the synchronise_on_transient_state attribute.

Parameters:

value (bool) – If True, the action will synchronise on the next transient state instead of the next quiescent state. By default, it is False, so the action will synchronise on the next quiescent state.

synchronise_on_transient_state

If True, the action will synchronise on the next transient state instead of the next quiescent state. By default, it is False, so the action will synchronise on the next quiescent state.

termination_condition()

Wait for the quiescent or transient state of the subarray.

By default, the action will synchronise on the next quiescent state, but if the synchronise_on_transient_state attribute is True, it will synchronise on the next transient state. Eventual inherited termination conditions are also include when synchronising on the quiescent state.

Return type:

list[ExpectedEvent]

Returns:

A list of ExpectedEvent objects to wait for.

abstract termination_condition_for_quiescent_state()

Wait for the quiescent state of the subarray.

Implement this method to specify the termination condition for when you want to synchronise on the quiescent state.

Return type:

list[ExpectedEvent]

Returns:

A list of ExpectedEvent objects to wait for, when you want to synchronise on the quiescent state.

abstract termination_condition_for_transient_state()

Wait for the transient state of the subarray.

Implement this method to specify the termination condition for when you want to synchronise on the transient state.

Return type:

list[ExpectedEvent]

Returns:

A list of ExpectedEvent objects to wait for, when you want to synchronise on the transient state.

Inputs module

A collection of (mostly abstract) classes to represent the required inputs.

This module contains classes that represent the required inputs for the telescope actions. Mainly, those inputs come in the form of JSON strings, for which we provide an abstraction to represent them.

Currently, it contains also some classes (left outside from the documentation on purpose) to represent various data types that - in theory - they should belong to external repositories but that for some reason they are here.

class ska_integration_test_harness.inputs.DictJSONInput(json_dict)

A JSON input that is represented as a dictionary.

This class is a concrete implementation of the JSONInput abstract class which uses a dictionary to represent the JSON input. This class is useful when the JSON input you want to provide is already a dictionary.

Since a dictionary can be always converted to a valid JSON string, this class is guaranteed to always return a valid JSON string representation. Since the representation is valid, as_dict and with_attribute will never raise an exception.

as_dict()

Return the JSON dictionary representation of the input.

Return type:

dict

as_str()

Return the JSON string representation of the input.

Return type:

str

with_attribute(attr_name, attr_value)

Decorate the input with an additional attribute.

Return type:

JSONInput

class ska_integration_test_harness.inputs.FileJSONInput(json_file, encoding='utf-8')

A JSON input that is represented as a file path.

This class is a concrete implementation of the JSONInput abstract class which uses a file path to represent the JSON input. This class is useful when the JSON input is already a file path and you don’t want to necessarily parse it to a dictionary.

IMPORTANT NOTE: The string representation of the JSON input is not guaranteed to be a valid JSON string. This means that the as_dict and with_attribute methods may raise an exception if the JSON string is invalid. However, the as_str method should always return a string.

OTHER IMPORTANT NOTE: Since the file path is used to represent the JSON input, this class constructor may raise the exceptions you would expect when working with files (e.g., FileNotFoundError, PermissionError).

class ska_integration_test_harness.inputs.JSONInput

Template for a generic JSON input.

This class is an abstract class that defines a template for a JSON input for a command on the telescope. It is meant to be used as a base class to implement your own way to provide that JSON value.

IMPORTANT NOTE: This class instances are meant to be immutable. If you need to change the value of the JSON input, you should create a new instance with the new value (see the with_attribute method).

abstract as_dict()

Return the JSON dictionary representation of the input.

This representation is guaranteed to be a valid JSON dictionary, but it may raise an exception if this instance represents an invalid JSON string.

Return type:

dict

Returns:

The JSON dictionary representation of the input.

Raises:

json.JSONDecodeError – if the JSON string is invalid.

abstract as_str()

Return the JSON string representation of the input.

This representation is not guaranteed to be valid JSON, but it should always be guaranteed to be returned as a string without raising any exceptions.

Return type:

str

Returns:

The JSON string representation of the input.

is_equal_to_json(json_data)

Check if this input is equal to the provided JSON data.

Return type:

bool

abstract with_attribute(attr_name, attr_value)

Decorate the input with an additional attribute.

This method should return a new instance of the JSON input with the specified attribute set to the specified value.

This representation is guaranteed to be a valid JSON dictionary, but it may raise an exception if this instance represents an invalid JSON string.

Parameters:

  • attr_name (str) – the name of the attribute to set

  • attr_value (Any) – the value to set for the attribute

Raises:

json.JSONDecodeError – if the JSON string is invalid.

Return type:

JSONInput

class ska_integration_test_harness.inputs.StrJSONInput(json_str)

A JSON input that is represented as a string.

This class is a concrete implementation of the JSONInput abstract class which uses a string to represent the JSON input. This class is useful when the JSON input is already a string and you don’t want to necessarily parse it to a dictionary.

IMPORTANT NOTE: The string representation of the JSON input is not guaranteed to be a valid JSON string. This means that the as_dict and with_attribute methods may raise an exception if the JSON string is invalid. However, the as_str method should always return a string.

as_dict()

Return the JSON dictionary representation of the input.

Return type:

dict

as_str()

Return the JSON string representation of the input.

Return type:

str

with_attribute(attr_name, attr_value)

Decorate the input with an additional attribute.

Return type:

JSONInput

class ska_integration_test_harness.inputs.TestHarnessInputs(default_vcc_config_input=None, assign_input=None, configure_input=None, scan_input=None, release_input=None)

A collection of inputs for the test harness.

This data class is a collection of inputs that may be needed for some processes in the test harness. For example, it can be used to specify the inputs for the subarray observation state resetter, or also the default inputs injected in the test harness to perform various operations (such as the teardown).

The contained inputs are:

  • a default VCC configuration as a JSON, needed for the teardown procedure,

  • a set of JSON inputs for the obsState state machine: - assign - configure - scan - release

A support enum is provided to specify the input names, and allow access to the inputs in a parametric way.

class InputName(value)

An enumeration of the possible input names.

An enumeration of the possible input names that could be included as test harness inputs.

ASSIGN = 'assign'
CONFIGURE = 'configure'
DEFAULT_VCC_CONFIG = 'default_vcc_config'
RELEASE = 'release'
SCAN = 'scan'
assign_input: JSONInput | None = None

The input for the AssignResources command.

configure_input: JSONInput | None = None

The input for the Configure command.

default_vcc_config_input: JSONInput | None = None

The default VCC configuration input.

get_input(input_name, fail_if_missing=False)

Get the input with the given name.

Parameters:

  • input_name (InputName) – The name of the input to get.

  • fail_if_missing (bool) – If True, raise an exception if the input is missing.

Return type:

JSONInput

Returns:

The input with the given name.

Raises:

ValueError – If the input is missing and fail_if_missing is set to True.

get_non_none_json_inputs()

Get the non-None JSON inputs as a dictionary.

Return type:

dict[InputName, JSONInput]

Returns:

A dictionary with the non-None JSON inputs.

release_input: JSONInput | None = None

The input for the ReleaseResources command.

scan_input: JSONInput | None = None

The input for the Scan command.

Inputs validation

Tools to validate the inputs to the test harness.

class ska_integration_test_harness.inputs.validation.BasicInputValidator(logger=None)

A basic validator for the input data.

For how it is implemented now:

  • it checks the presence of the inputs for: default VCC config, assign, configure, scan, release;

  • it checks each of those input is a valid JSON.

Instead, it doesn’t check the semantic correctness of the JSONs.

additional_inputs_for_mid

The list of the additional inputs required for the MID.

required_inputs

The list of the required inputs.

validate_inputs_correctness(inputs)

Ensure all the passed inputs are valid JSONs.

Return type:

None

validate_inputs_presence(inputs, is_mid=True)

Validate the presence of the required inputs.

Return type:

None

class ska_integration_test_harness.inputs.validation.InputValidator(logger=None)

A validator for the input data.

It provides interfaces for two main validation steps:

  • a step to validate the presence of a needed set of inputs

  • a step to validate each input individually

It provides also a minimal log.info utility to log the validation steps.

logger

An optional logger used to log the errors and warnings, while they are found during the validation.

logger_prefix

The prefix used to log the validation messages.

abstract validate_inputs_correctness(inputs)

Check the correctness of the given inputs.

Validate all inputs using the configured validation rules. If any critical error is found, a ValueError is raised.

Raises:

ValueError – If any critical error is found.

Return type:

None

abstract validate_inputs_presence(inputs, is_mid=True)

Validate the presence of the required inputs.

A given set of inputs must contain all the needed inputs to run the tests.

If the logger is set, all the errors and warnings are logged.

Raises:

ValueError – If any critical error is found.

Return type:

None

Config(uration) module

Test harness configurations.

This module contains classes that represent, read and validate the configuration files for the test harness.

class ska_integration_test_harness.config.CSPConfiguration(is_emulated=True, target='mid', csp_master_name=None, csp_subarrays_names=<factory>, pst_name=None)

Configuration for a CSP device.

It contains the names of the various devices that make up the CSP. It is initialised with default values.

csp_master_name: str = None
property csp_subarray1_name: str

Get the name of the first subarray.

csp_subarrays_names: dict[int, str]
get_device_names()

Return all the device names.

(associated with they “keyword” name in the configuration)

Return type:

dict[str, str]

Returns:

List of attribute names.

mandatory_attributes()

Return the names of the mandatory attributes.

Return type:

list[str]

Returns:

List of attribute names.

pst_name: str | None = None
class ska_integration_test_harness.config.DishesConfiguration(is_emulated=True, target='mid', dish_master1_name=None, dish_master2_name=None, dish_master3_name=None, dish_master4_name=None)

Configuration for the dishes.

It contains the names of the various dishes. It is initialised with default values (which are the ones you will have when the dishes are emulated).

dish_master1_name: str = None
dish_master2_name: str = None
dish_master3_name: str = None
dish_master4_name: str = None
get_device_names()

Return all the device names.

(associated with they “keyword” name in the configuration)

Return type:

dict[str, str]

Returns:

List of attribute names.

mandatory_attributes()

Return the names of the mandatory attributes.

Return type:

list[str]

Returns:

List of attribute names.

class ska_integration_test_harness.config.SDPConfiguration(is_emulated=True, target='mid', sdp_master_name=None, sdp_subarrays_names=<factory>)

Configuration for a SDP device.

It contains the names of the various devices that make up the SDP. It is initialised with default values.

get_device_names()

Return all the device names.

(associated with they “keyword” name in the configuration)

Return type:

dict[str, str]

Returns:

List of attribute names.

mandatory_attributes()

Return the names of the mandatory attributes.

Return type:

list[str]

Returns:

List of attribute names.

sdp_master_name: str = None
property sdp_subarray1_name: str

Get the name of the first subarray.

sdp_subarrays_names: dict[int, str]
class ska_integration_test_harness.config.SubsystemConfiguration(is_emulated=True, target='mid')

A generic configuration for a telescope subsystem.

A subsystem is a collection of devices that are logically grouped together, such as CSP, SDP, TMC and the Dishes. This class provides a way to encapsulate the names of the devices that make up the subsystem.

To use this class:

  • extend it,

  • add as attributes your own configuration parameters,

  • implement a few methods to expose attributes containing device names and the mandatory attributes.

A subsystem in the context of SKA can be emulated or a production one. This class contains a boolean flag that will specify that. By default, all configurations point to emulated devices. You don’t need to list the is_emulated attribute in the all_attributes

abstract get_device_names()

Return all the device names.

(associated with they “keyword” name in the configuration)

Return type:

dict[str, str]

Returns:

List of attribute names.

is_emulated: bool = True
abstract mandatory_attributes()

Return the names of the mandatory attributes.

Return type:

list[str]

Returns:

List of attribute names.

supports_low()

Check if the configuration supports the low target environment.

Return type:

bool

supports_mid()

Check if the configuration supports the mid target environment.

Return type:

bool

target: str = 'mid'

The target environment for the subsystem.

It can be “mid” (default) or “low”.

class ska_integration_test_harness.config.TMCConfiguration(is_emulated=True, target='mid', centralnode_name=None, tmc_csp_master_leaf_node_name=None, tmc_sdp_master_leaf_node_name=None, tmc_mccs_master_leaf_node_name=None, subarrays_names=<factory>, tmc_csp_subarrays_leaf_nodes_names=<factory>, tmc_sdp_subarrays_leaf_nodes_names=<factory>, tmc_mccs_subarrays_leaf_nodes_names=<factory>, tmc_dish_leaf_node1_name=None, tmc_dish_leaf_node2_name=None, tmc_dish_leaf_node3_name=None, tmc_dish_leaf_node4_name=None)

Configuration for a TMC device.

It contains the names of the various devices that make up the TMC. It is initialised with default values.

centralnode_name: str = None
get_device_names()

Return all the device names.

(associated with they “keyword” name in the configuration)

Return type:

dict[str, str]

Returns:

List of attribute names.

mandatory_attributes()

Return the names of the mandatory attributes.

Return type:

list[str]

Returns:

List of attribute names.

subarrays_names: dict[int, str]
supports_low()

Check if the configuration supports the low target environment.

Return type:

bool

supports_mid()

Check if the configuration supports the mid target environment.

Return type:

bool

target: str = 'mid'

The target environment for the TMC. It can be “mid” (default) or “low”.

tmc_csp_master_leaf_node_name: str = None
property tmc_csp_subarray_leaf_node_name: str

Get the name of the first subarray node.

tmc_csp_subarrays_leaf_nodes_names: dict[int, str]
tmc_dish_leaf_node1_name: str = None
tmc_dish_leaf_node2_name: str = None
tmc_dish_leaf_node3_name: str = None
tmc_dish_leaf_node4_name: str = None
tmc_mccs_master_leaf_node_name: str | None = None
property tmc_mccs_subarray_leaf_node_name: str

Get the name of the first subarray node.

tmc_mccs_subarrays_leaf_nodes_names: dict[int, str]
tmc_sdp_master_leaf_node_name: str = None
property tmc_sdp_subarray_leaf_node_name: str

Get the name of the first subarray node.

tmc_sdp_subarrays_leaf_nodes_names: dict[int, str]
property tmc_subarraynode1_name: str

Get the name of the first subarray node.

class ska_integration_test_harness.config.TestHarnessConfiguration(tmc_config=None, csp_config=None, sdp_config=None, dishes_config=None, mccs_config=None)

A wrapper to all the configurations needed to setup the test harness.

This class is a wrapper to all the configurations needed to setup the test harness. It is used to pass all the configurations to the test harness factory, so it can create the needed test harness wrappers.

Each field of this class is a configuration for a different subsystem of the test harness, such as the TMC, CSP, SDP, and the dishes. Even if they are marked as optional, for now they are all required, as the test harness is strictly built for TMC-Mid integration tests. In the future, we may support an elastic choice of which subsystems to include, so this configuration may be useful to select only the needed subsystems.

A support enum is provided to specify the subsystem names, and allow access to the configurations in a parametric way.

class SubsystemName(value)

An enumeration of the possible subsystem names.

An enumeration of the possible subsystems that could be included in the the test harness configuration.

CSP = 'csp'
DISHES = 'dishes'
MCCS = 'mccs'
SDP = 'sdp'
TMC = 'tmc'
all_emulated()

Check if, among the included subsystems, all are emulated.

Return type:

bool

Returns:

True if all the subsystems are emulated, False otherwise.

all_production()

Check if, among the included subsystems, all are in production.

Return type:

bool

Returns:

True if all the subsystems are in production, False otherwise.

csp_config: CSPConfiguration | None = None
dishes_config: DishesConfiguration | None = None
get_included_subsystems()

Get the list of subsystems that are included in the configuration.

Return type:

list[SubsystemConfiguration]

Returns:

A list with the names of the included subsystems.

get_subsystem_config(subsystem_name)

Get the configuration for a specific subsystem.

Parameters:

subsystem_name (SubsystemName) – The name of the subsystem you want to get the configuration for.

Return type:

SubsystemConfiguration

Returns:

The configuration for the specified subsystem.

Raises:

ValueError – If the specified subsystem is not included in the configuration.

mccs_config: MCCSConfiguration | None = None
sdp_config: SDPConfiguration | None = None
tmc_config: TMCConfiguration | None = None

Config reader

Concrete and abstract tools to read the test harness configuration.

class ska_integration_test_harness.config.reader.ConfigurationReader

A factory that reads from somewhere the test harness configuration.

This abstract class defines the interface of a configuration reader that is able to collect subsystem configurations and create a TestHarnessConfiguration object.

Currently, the supported subsystems are:

  • TMC

  • CSP

  • SDP

  • the dishes

  • the emulation configuration

None of those configurations are mandatory and the correctness of the configuration is not checked by this class, since it is delegated to a separate validator that can make its own checks according to the testing context.

The concrete implementations of this class may read the configuration from different sources, such as environment variables and configuration files.

abstract get_csp_configuration()

Get the configuration for the CSP.

return: The CSP configuration (if any).

Return type:

CSPConfiguration | None

abstract get_dish_configuration()

Get the configuration for the dishes.

return: The dishes configuration (if any).

Return type:

DishesConfiguration | None

abstract get_mccs_configuration()

Get the configuration for the MCCS.

return: The MCCS configuration (if any).

Return type:

MCCSConfiguration | None

abstract get_sdp_configuration()

Get the configuration for the SDP.

return: The SDP configuration (if any).

Return type:

SDPConfiguration | None

abstract get_target()

Get the target environment for the configuration.

return: The target environment for the configuration.

Return type:

str

get_test_harness_configuration()

Get all the configurations needed for the test harness.

Return type:

TestHarnessConfiguration

return: A collection of all the test harness configurations that

have been found by the reader.

abstract get_tmc_configuration()

Get the configuration for the TMC.

return: The TMC configuration (if any).

Return type:

TMCConfiguration | None

class ska_integration_test_harness.config.reader.YAMLConfigurationReader

A configuration reader that reads from a YAML file.

A configuration reader that creates the needed test harness configurations reading everything from a YAML file.

The YAML file must have the following structure:

tmc:
    is_emulated: false # Not supported otherwise, default is false

    # Expected device names (Required)
    centralnode_name: "ska_mid/tm_central/central_node"
    tmc_subarraynode1_name: "ska_mid/tm_subarray_node/1"
    tmc_csp_master_leaf_node_name: "ska_mid/tm_leaf_node/csp_master"
    tmc_csp_subarray_leaf_node_name: "ska_mid/tm_leaf_node/csp_subarray01"
    tmc_sdp_master_leaf_node_name: "ska_mid/tm_leaf_node/sdp_master"
    tmc_sdp_subarray_leaf_node_name: "ska_mid/tm_leaf_node/sdp_subarray01"
    tmc_dish_leaf_node1_name: "ska_mid/tm_leaf_node/d0001"
    tmc_dish_leaf_node2_name: "ska_mid/tm_leaf_node/d0036"
    tmc_dish_leaf_node3_name: "ska_mid/tm_leaf_node/d0063"
    tmc_dish_leaf_node4_name: "ska_mid/tm_leaf_node/d0100"

csp:
    is_emulated: false # Supported true too, default is true

    # Expected device names
    csp_master_name: "mid-csp/control/0"
    csp_subarray1_name: "mid-csp/subarray/01"

sdp:
    is_emulated: true # Supported false too, default is true

    # Expected device names (Required)
    sdp_master_name: "mid-sdp/control/0"
    sdp_subarray1_name: "mid-sdp/subarray/01"

dishes:
    is_emulated: true # Supported false too, default is true

    # Expected device names (Required)
    dish_master1_name: "ska001/elt/master"
    dish_master2_name: "ska036/elt/master"
    dish_master3_name: "ska063/elt/master"
    dish_master4_name: "ska100/elt/master"

IMPORTANT NOTE: currently, the required fields and sections are hardcoded, but in the future, we may want to make them optional and elastic to permit the user to define the configuration only for the subsystems they need. This could be the starting point to make all the subsystem to be potentially optional and elastic.

config_as_dict: dict

The content of the YAML file as a dictionary.

filename: str | None

The path to the YAML file you read.

get_csp_configuration()

Get the configuration for the CSP.

return: The CSP configuration (if any).

Return type:

CSPConfiguration | None

get_dish_configuration()

Get the configuration for the dishes.

return: The dishes configuration (if any).

Return type:

DishesConfiguration | None

get_mccs_configuration()

Get the configuration for the MCCS.

return: The MCCS configuration (if any).

Return type:

MCCSConfiguration | None

get_sdp_configuration()

Get the configuration for the SDP.

return: The SDP configuration (if any).

Return type:

SDPConfiguration | None

get_target()

Get the target environment for the subsystem (“mid” or “low”).

return: The “mid” or “low” target environment. Default is “mid”.

Return type:

str

get_tmc_configuration()

Get the configuration for the TMC.

return: The TMC configuration (if any).

Return type:

TMCConfiguration | None

read_configuration_file(filename)

Read the YAML file and store the content as a dictionary.

Read the YAML file and store the content as a dictionary in the config_as_dict attribute. Run this method before calling any other method that generates configurations.

Parameters:

filename (str) – The path to the YAML file you want to read

Raises:

  • FileNotFoundError – If the file doesn’t exist.

  • yaml.YAMLError – If the file is not a valid YAML file.

Return type:

None

Config validation

Concrete and abstract tools to validate the test harness configuration.

class ska_integration_test_harness.config.validation.BasicConfigurationValidator(logger=None)

A basic configuration validator for the test harness.

This validator:

  • checks the presence of TMC, CSP, SDP and Dishes or MCCS configurations (depending on the target being Mid or Low)

  • ensure the required fields are set for each subsystem

  • ensure the device names are valid and that they point to reachable devices

  • ensure the consistency of the emulation settings

subsystem_validators: SubsystemConfigurationValidator

The validators used to validate the subsystem configurations.

validate_subsystems_configurations(config)

Validate each individual subsystem configuration.

Return type:

None

validate_subsystems_presence(config)

Validate the presence of TMC, CSP, SDP, and Dishes configs.

Return type:

None

class ska_integration_test_harness.config.validation.ConfigurationError(message)

A critical issue found in the configuration.

is_critical()

Return True if the anomaly is critical, False otherwise.

Return type:

bool

log(logger)

Log the anomaly using the provided logger.

Return type:

None

class ska_integration_test_harness.config.validation.ConfigurationIssue(message)

A data structure to represent an issue found in a configuration.

A configuration issue is some anomaly in the configuration that may prevent the test harness from working correctly. It is generated by a validation procedure, it is defined by a message and it can be classified as critical (error) or non-critical (warning).

A critical issue will stop the test harness from starting, while a non-critical issue will only generate a warning.

All the issues will be logged (so this class instances - given a logger - they can log themselves).

abstract is_critical()

Return True if the anomaly is critical, False otherwise.

Return type:

bool

abstract log(logger)

Log the anomaly using the provided logger.

Return type:

None

message

The message describing the anomaly.

class ska_integration_test_harness.config.validation.ConfigurationValidator(logger=None)

A generic validator for the whole test harness configuration.

It provides interfaces for two main validation steps:

  • a step to validate the presence of a coherent set of subsystems (e.g., for TMC-Mid tests, we need TMC, CSP, SDP, and Dishes)

  • a step to validate the individual subsystems configurations (e.g., check the required fields, check the device are reachable, etc.)

It provides also a minimal log.info utility to log the validation steps.

logger

An optional logger used to log the errors and warnings, while they are found during the validation.

logger_prefix

The prefix used to log the validation messages.

abstract validate_subsystems_configurations(config)

Validate the individual subsystem configurations.

Validate all subsystem configurations using the configured validators. If any critical error is found, a ValueError is raised.

If the logger is set, all the errors and warnings are logged.

Raises:

ValueError – If any critical error is found.

Return type:

None

abstract validate_subsystems_presence(config)

Validate the presence of the required subsystems.

A given configuration must contain a valid subset of subsystems (i,e., there must be all the needed dependencies to run the tests).

If the logger is set, all the errors and warnings are logged.

Raises:

ValueError – If any critical error is found.

Return type:

None

class ska_integration_test_harness.config.validation.ConfigurationWarning(message)

A non-critical issue found in the configuration.

is_critical()

Return True if the anomaly is critical, False otherwise.

Return type:

bool

log(logger)

Log the anomaly using the provided logger.

Return type:

None

class ska_integration_test_harness.config.validation.SubsystemConfigurationValidator(logger=None)

A generic validator for a subsystem configuration.

This is an abstraction of a class that can be used to validate one or more subsystem configurations. The idea is that you can implement your own validation logic by subclassing this class and implementing the validate method. The validation logic should scan a given configuration and populate the errors_and_warnings list with the errors and warnings found during the validation (using the add_error and add_warning helper methods).

With a validator instance, you can run multiple validations on different configurations, and collect all the errors and warnings in the errors_and_warnings list, or you can reset the validator and start a new validation. After one or more runs, you can check if the validation was successful by calling the is_valid method.

When you initialise an instance of this class, you can pass an optional logger that will be used to log the errors and warnings found during the validation.

NOTE: maybe in future this could be refactored with the https://refactoring.guru/design-patterns/visitor design pattern (and so implement custom checks for each different kind of subsystem configuration).

add_error(message)

Add an error to the errors and warnings list.

Parameters:

message (str) – The error message to add.

Return type:

None

add_warning(message)

Add a warning to the errors and warnings list.

Parameters:

message (str) – The warning message to add.

Return type:

None

errors_and_warnings: list[ConfigurationIssue]

The errors and warnings found during one or more validations.

get_critical_errors()

Return the critical errors found during the validation.

Return type:

list[ConfigurationIssue]

Returns:

A list of critical errors.

is_valid()

Return True if there are no critical issues (i.e., errors).

Return type:

bool

logger

An optional logger used to log the errors and warnings, while they are found during the validation.

reset()

Reset the validator, clearing the errors and warnings list.

Return type:

None

abstract validate(config)

Validate a generic subsystem configuration.

Validate a generic subsystem configuration, implementing your own checks to ensure that the configuration is correct. As a side effect, this method should populate the errors_and_warnings list with the errors and warnings found during the validation.

NOTE: this method is not expected to return anything, nor to raise exceptions.

Parameters:

config (SubsystemConfiguration) – The configuration to validate.

Init(ialisation) module

Tools to initialise the test harness.

(see Installation and Usage for more information on how to use them)

class ska_integration_test_harness.init.TestHarnessBuilder

A builder class for configuring and building a test harness.

This class allows you to configure various subsystems such as TMC, CSP, SDP, and Dishes, validate the configurations, and build the test harness. You can/should also specify some default JSON inputs for various commands/purposes.

The class uses also a set of tools to read and validate the configurations and the inputs. Potentially, you can inject your own tools to customise the behaviour of the builder. Potentially, you can also subclass this builder to customise its behaviour.

Usage example:

builder = TestHarnessBuilder()
builder.read_config_file("configurations.yaml")
builder.validate_configurations()

default_inputs = TestHarnessInputs(
    assign_input=FileJSONInput("path/to/assign.json"),
    # ...
)
builder.set_default_inputs(default_inputs)
builder.validate_default_inputs()

telescope = builder.build()
are_default_inputs_validated()

Check if the default inputs have been validated.

Return type:

bool

build()

Build the test harness.

Return type:

TelescopeWrapper

Returns:

a wrapper for the telescope and its subsystems.

config: TestHarnessConfiguration

The configuration used to build the test harness.

config_reader: YAMLConfigurationReader

The tool used to read the configurations.

config_validator: ConfigurationValidator

The tool used to validate the configurations.

default_inputs: TestHarnessInputs

The default inputs used to build the test harness.

input_validator: InputValidator

The tool used to validate the inputs.

is_config_validated()

Check if the configurations have been validated.

Return type:

bool

kube_namespace: str | None

The Kubernetes namespace where the SUT is deployed.

It is optional, but if you set it, it will be used to connect to get more information about the various subsystem devices.

read_config_file(filepath)
Read the configuration from a YAML file and set the configurations

accordingly.

Parameters:

filepath (str) – The path to the YAML file containing the configurations.

Return type:

TestHarnessBuilder

Returns:

The current instance of TestHarnessBuilder with configurations set.

Raises:

  • FileNotFoundError – If the YAML file does not exist.

  • yaml.YAMLError – If the YAML file contains errors.

set_default_inputs(inputs)

Set the default inputs for the test harness.

NOTE: all the defaults input specified by the TestHarnessInputs class are required, since they are used in test harness tear down procedures.

Parameters:

inputs (TestHarnessInputs) – The default inputs to be used.

Return type:

TestHarnessBuilder

Returns:

The current instance of TestHarnessBuilder with the default inputs set.

set_kubernetes_namespace(kube_namespace)

Set the Kubernetes namespace where the SUT is deployed.

It will help to connect to the ska-k8s-config-exporter service.

Parameters:

namespace – The Kubernetes namespace.

Return type:

TestHarnessBuilder

Returns:

The current instance of TestHarnessBuilder with the Kubernetes namespace set.

test_harness_factory: TestHarnessFactory

The factory used to create the subsystems and the overall telescope wrapper.

validate_configurations()

Validate all subsystem configurations.

Logs errors and warnings as necessary.

If any configuration is invalid, a ValueError is raised, containing a detailed list of all validation errors and warnings.

Return type:

TestHarnessBuilder

Returns:

The current instance of TestHarnessBuilder if all configurations are valid.

Raises:

ValueError – If any configuration is invalid, with the details of the errors.

validate_default_inputs()

Check if all default inputs are valid.

Right now this method checks: :rtype: TestHarnessBuilder

  • all the required default inputs are passed

  • all the required default inputs have the right type

Raises:

ValueError – if some input is missing.

class ska_integration_test_harness.init.TestHarnessFactory(config, default_inputs)

Factory to build the test harness wrappers.

Given the necessary inputs and configurations, build the various test harness subsystems (dealing with emulation directives, variants coming from configuration settings, etc.).

Main methods:

  • create_tmc_wrapper: Create a wrapper for the TMC subsystem.

  • create_csp_wrapper: Create a wrapper for the CSP subsystem.

  • create_sdp_wrapper: Create a wrapper for the SDP subsystem.

  • create_dishes_wrapper: Create a wrapper for the Dishes subsystem.

create_csp_wrapper()

Create a CSP wrapper with the given configuration.

Return type:

CSPWrapper

Returns:

The CSP wrapper.

create_dishes_wrapper()

Create a dishes wrapper with the given configuration.

Return type:

DishesWrapper

Returns:

The dishes wrapper.

create_mccs_wrapper()

Create a MCCS wrapper with the given configuration.

Return type:

MCCSWrapper

Returns:

The MCCS wrapper.

create_sdp_wrapper()

Create a SDP wrapper with the given configuration.

Return type:

SDPWrapper

Returns:

The SDP wrapper.

create_telescope_wrapper()

Create a telescope wrapper with the given configuration.

Return type:

TelescopeWrapper

Returns:

The telescope wrapper.

create_tmc_wrapper()

Create a TMC wrapper with the given configuration.

Return type:

TMCWrapper

Returns:

The TMC wrapper.

set_config(config)

Set the configuration for the factory.

Parameters:

config (TestHarnessConfiguration) – The new configuration.

set_default_inputs(default_inputs)

Set the default inputs for the factory.

Parameters:

default_inputs (TestHarnessInputs) – The new default inputs.

Common utils module

A collection of common utilities for the test harness.

At the moment, this module contains the following utilities:

  • DevicesInfoProvider: A class that can be used to get information about Tango devices from the k8s-config-info service (and various support classes).

class ska_integration_test_harness.common_utils.DevicesInfoProvider(kube_namespace, service_name='ska-k8s-config-exporter-service', port=8080, path='tango_devices')

Provider to get info about Tango devices from ska-k8s-config-exporter.

This class communicates with the ska-k8s-config-exporter service to fetch and store the information about Tango devices. It also provides methods to retrieve device information and recaps.

DEFAULT_PATH = 'tango_devices'

Default path to interrogate the ska-k8s-config-exporter service.

DEFAULT_PORT = 8080

Default port where the ska-k8s-config-exporter service is listening.

DEFAULT_SERVICE_NAME = 'ska-k8s-config-exporter-service'

Default name of the ska-k8s-config-exporter service.

get_device_recap(device_name)

Get a recap of the given device information.

The recap contains the device name and its information (for now, only the version). If the device is not found, it is reported in the recap.

Parameters:

device_name (str) – Name of the device to get the recap.

Return type:

str

Returns:

Recap of the device information.

get_update_service_url()

Get the URL to update the devices information.

Return type:

str

Returns:

URL to update the devices information.

kube_namespace

Kubernetes namespace where the service is running.

last_devices_info: dict[str, TangoDeviceInfo]

Last devices information fetched from the service.

last_update_time: datetime | None

Time when the last update was done.

path

Path to interrogate on the service to get all devices.

port

Port where the service is listening.

service_name

Name of the ska-k8s-config-exporter service.

update(timeout=10)

Update the local devices information by calling the service.

Fetches the latest devices information from the ska-k8s-config-exporter service and updates the internal state.

Parameters:

timeout (int) – Timeout in seconds for the request (default: 10 seconds).

Raises:

DevicesInfoServiceException – If the service is not available, or it does not respond as expected.

Return type:

None

exception ska_integration_test_harness.common_utils.DevicesInfoServiceException(message='')

The devices information is not available or something went wrong.

i.e., when the ska-k8s-config-exporter service is not available.

class ska_integration_test_harness.common_utils.TangoDeviceInfo(name, version=None)

Information about a Tango device from ska-k8s-config-exporter.

Information related to a Tango device collected by the ska-k8s-config-exporter tool. The information includes:

  • the name of the device (always present)

  • the version of the device (may be None)

get_recap(include_version=True)

Get a recap of the device information.

Get a recap of the device information. The recap contains the device name and its information (for now, only the version if available).

Parameters:

include_version (bool) – If True, the version is included in the recap.

Return type:

str

Returns:

Recap of the device information.

name: str

Name of the device in the Tango database.

version: str | None = None

Version of the device, if available.