Extensions and specific logic API (TMC Monolith)

This section is a sort of enumeration of the main concrete non-generic classes and components that are part of the ska-integration-test-harness for testing TMC-CSP in Mid and Low (at the time of writing).

IMPORTANT NOTE: A very crucial difference from the generic mechanisms section is that the classes and components described here are not unit tested code, since they are considered more related to the specific SUT logic than to the test harness itself. In a certain sense, the integration tests that use the test harness are the validation of the correctness of the implementation of the classes and components described in this section. The coverage metrics of the test harness are calculated only on the first section (at the time of writing, the coverage on the first section is approximately 90%, while the overall coverage is around 60%).

An example of tests that use what is described in the second section are the ones introduced in ska-tmc-mid-integration by this merge request.

Facades module

A set of facades to interact with the telescope devices.

A set of facades to interact with the telescope devices, abstracting the complexities of the interaction with and between the devices.

class ska_integration_test_harness.facades.CSPFacade(telescope)

A facade to expose the CSP devices to the tests.

property csp_master: tango.DeviceProxy: A Tango proxy to the CSP master device.

property csp_subarray: tango.DeviceProxy: A Tango proxy to the CSP subarray device.

property pst: tango.DeviceProxy | None

A Tango proxy to the PST device.

It is available only when the target is ‘low’.

class ska_integration_test_harness.facades.DishesFacade(telescope)

A facade to expose the Dish devices to the tests.

property dish_master_dict: dict[str, tango.DeviceProxy]

The dish master Tango device proxies as a dictionary.

The key is the dish name, as a string in the format “dish_XXX”.

property dish_master_list: list[tango.DeviceProxy]: The dish master Tango device proxies as a list (sorted by key).

class ska_integration_test_harness.facades.MCCSFacade(telescope)

A facade to expose MCCS devices to the tests.

property mccs_controller: tango.DeviceProxy: The MCCS controller Tango device proxy.

property mccs_subarray: tango.DeviceProxy: The MCCS subarray Tango device proxy.

class ska_integration_test_harness.facades.SDPFacade(telescope)

A facade to expose the SDP devices to the tests.

property sdp_master: A Tango proxy to the SDP master device.

property sdp_subarray: A Tango proxy to the SDP subarray device.

class ska_integration_test_harness.facades.TMCCentralNodeFacade(telescope)

A facade to TMC Central Node device and its actions.

A facade to TMC subsystem, providing a simplified interface to the central node device and its actions. It contains:

references to central node device,
references to master leaf devices to interact with CSP and SDP,
properties to get and set telescope state,
actions to move the telescope to ON, OFF and STANDBY states,
actions to load dish VCC configuration, assign and release resources,
a generic action to perform any action on central node.

IMPORTANT NOTE: This facade is deprecated, since we want to use a facade for each subsystem. Please use TMCFacade instead.

assign_resources(assign_json, wait_termination=True)

Invoke Assign Resource command on central Node.

Parameters:

assign_json (str) – Assign resource input json.
wait_termination (bool) – set to False if you don’t want to wait for the termination condition. By default the termination condition is waited.

Return type:

Tuple[ResultCode, str]

Returns:

result, message

property central_node: tango.DeviceProxy: The central node Tango device proxy.

property csp_master_leaf_node: tango.DeviceProxy: The CSP master leaf node Tango device proxy.

load_dish_vcc_configuration(dish_vcc_config, wait_termination=True)

Invoke LoadDishCfg command on central Node.

Parameters:

dish_vcc_config (str) – Dishes VCC configuration json.
wait_termination (bool) – set to False if you don’t want to wait for the termination condition. By default the termination condition is waited.

Return type:

Tuple[ResultCode, str]

Returns:

result, message

move_to_off(wait_termination=True)

Move the telescope to OFF state.

Parameters:: wait_termination (bool) – set to False if you don’t want to wait for the termination condition. By default the termination condition is waited.
Return type:: None

move_to_on(wait_termination=True)

Move the telescope to ON state.

Parameters:: wait_termination (bool) – set to False if you don’t want to wait for the termination condition. By default the termination condition is waited.
Return type:: None

perform_action(command_name, input_json, wait_termination=True)

Execute provided command on central node.

Parameters:

command_name (str) – Name of command to execute.
input_json (JSONInput) – Json send as input to execute command.
wait_termination (bool) – set to False if you don’t want to wait for the termination condition. By default the termination condition is waited.

Return type:

Tuple[ResultCode, str]

release_resources(input_string, wait_termination=True)

Invoke Release Resource command on central Node.

Parameters:

input_string (str) – Release resource input json.
wait_termination (bool) – set to False if you don’t want to wait for the termination condition. By default the termination condition is waited.

Return type:

Tuple[ResultCode, str]

Returns:

result, message

property sdp_master_leaf_node: tango.DeviceProxy: The SDP master leaf node Tango device proxy.

set_standby(wait_termination=True)

Set the telescope to STANDBY state.

Parameters:: wait_termination (bool) – set to False if you don’t want to wait for the termination condition. By default the termination condition is waited.
Return type:: None

class ska_integration_test_harness.facades.TMCFacade(telescope)

A facade to TMC devices and the actions on them.

A facade to TMC subsystem, providing a simplified interface to:

references to central node device,
references to subarray devices and their leaf devices,
references to leaf devices to the controllers of CSP and SDP,
actions to move the telescope to ON, OFF and STANDBY states,
actions to load dish VCC configuration, assign and release resources,
actions to configure, scan, end scan, end observation, abort, restart the subarray,
an action to force the subarray to a certain obs state,
a generic action to perform any action on central node or subarray node.

All the actions can be executed synchronously or asynchronously. When executed synchronously, the action waits for the termination condition to occur. When executed asynchronously, the action returns immediately and the termination condition is not waited. The timeout for the termination condition can be customized (otherwise the default action timeout is used).

Some of the given actions can be long running commands (LRC). In most of the cases when you call them and you decide to wait for the termination also the LRC successful completion is waited. In two tricky cases:

MoveToOn
CentralNodeLoadDishConfig

the LRC completion is not waited by default, since at the moment failures can be observed (TODO: solve the issue and set to True). In any case, you can always set the argument is_long_running_command to True or to False to include or exclude the LRC completion in the synchronisation. At the moment the customisation of this possibility is not instead available for actions called in force_change_of_obs_state method and in the system teardown.

abort(wait_termination=True, custom_timeout=None)

Invoke Abort command on subarray Node.

Parameters:

wait_termination (bool) – set to False if you don’t want to wait for the termination condition. By default the termination condition is waited.
custom_timeout (Optional[int]) – A custom timeout (in seconds) to wait for the termination condition to occur. If None, the default action timeout is used. This parameter is useful only when wait_termination=True.

Return type:

tuple[Any, list[str]]

Returns:

result, message

assign_resources(assign_input, wait_termination=True, custom_timeout=None, is_long_running_command=True)

Invoke Assign Resource command on central Node.

Parameters:

assign_input (JSONInput) – Assign resource input json.
wait_termination (bool) – set to False if you don’t want to wait for the termination condition. By default the termination condition is waited.
custom_timeout (Optional[int]) – A custom timeout (in seconds) to wait for the termination condition to occur. If None, the default action timeout is used. This parameter is useful only when wait_termination=True.
is_long_running_command (bool) – set to True if you want to include in the synchronisation the long running command. This parameter is useful only when wait_termination=True. By default is True.

Return type:

tuple[Any, list[str]]

Returns:

result, message

property central_node: tango.DeviceProxy: The central node Tango device proxy.

configure(configure_input, wait_termination=True, custom_timeout=None, is_long_running_command=True)

Invoke configure command on subarray Node.

Parameters:

configure_input (JSONInput) – json input for configure command.
wait_termination (bool) – set to False if you don’t want to wait for the termination condition. By default the termination condition is waited.
custom_timeout (Optional[int]) – A custom timeout (in seconds) to wait for the termination condition to occur. If None, the default action timeout is used. This parameter is useful only when wait_termination=True.
is_long_running_command (bool) – set to True if you want to include in the synchronisation the long running command. This parameter is useful only when wait_termination=True. By default is True.

Return type:

tuple[Any, list[str]]

Returns:

result, message

property csp_master_leaf_node: tango.DeviceProxy: The CSP master leaf node Tango device proxy.

property csp_subarray_leaf_node: Return CSP Subarray Leaf Node Proxy

property dish_leaf_node_list: Return Dish Leaf Node List

end_observation(wait_termination=True, custom_timeout=None, is_long_running_command=True)

Invoke End command on subarray Node.

Parameters:

wait_termination (bool) – set to False if you don’t want to wait for the termination condition. By default the termination condition is waited.
custom_timeout (Optional[int]) – A custom timeout (in seconds) to wait for the termination condition to occur. If None, the default action timeout is used. This parameter is useful only when wait_termination=True.
is_long_running_command (bool) – set to True if you want to include in the synchronisation the long running command. This parameter is useful only when wait_termination=True. By default is True.

Return type:

tuple[Any, list[str]]

Returns:

result, message

end_scan(wait_termination=True, custom_timeout=None, is_long_running_command=True)

Invoke EndScan command on subarray Node.

Parameters:

wait_termination (bool) – set to False if you don’t want to wait for the termination condition. By default the termination condition is waited.
custom_timeout (Optional[int]) – A custom timeout (in seconds) to wait for the termination condition to occur. If None, the default action timeout is used. This parameter is useful only when wait_termination=True.
is_long_running_command (bool) – set to True if you want to include in the synchronisation the long running command. This parameter is useful only when wait_termination=True. By default is True.

Return type:

tuple[Any, list[str]]

Returns:

result, message

force_change_of_obs_state(dest_state_name, commands_inputs, wait_termination=True, custom_timeout=None)

Force SubarrayNode obsState to provided obsState.

Parameters:

dest_state_name (ObsState) – Name of the destination obsState.
commands_inputs (TestHarnessInputs) – The JSON inputs for the commands to bring the subarray in a certain obs state. You can pass just the JSON inputs you need, but if one of them is missing, you may occur in an error.
wait_termination (bool) – set to False if you don’t want to wait for the termination condition. By default the termination condition is waited.
custom_timeout (Optional[int]) – A custom timeout (in seconds) to wait for the termination condition to occur. If None, the default action timeout is used. This parameter is useful only when wait_termination=True.

Raises:

NotImplementedError – If the procedure to reach the target state is not implemented.

Return type:

None

load_dish_vcc_configuration(dish_vcc_config, wait_termination=True, custom_timeout=None, is_long_running_command=True)

Invoke LoadDishCfg command on central Node.

Parameters:

dish_vcc_config (str) – dish vcc configuration json string.
wait_termination (bool) – set to False if you don’t want to wait for the termination condition. By default the termination condition is waited.
custom_timeout (Optional[int]) – A custom timeout (in seconds) to wait for the termination condition to occur. If None, the default action timeout is used. This parameter is useful only when wait_termination=True.
is_long_running_command (bool) – set to True if you want to include in the synchronisation the long running command. This parameter is useful only when wait_termination=True. By default is True.

Return type:

tuple[Any, list[str]]

Returns:

result, message

property mccs_master_leaf_node: Return MCCS Master Leaf Node Proxy

property mccs_subarray_leaf_node: Return MCCS Subarray Leaf Node Proxy

move_to_off(wait_termination=True, custom_timeout=None, is_long_running_command=True)

Move the telescope to OFF state.

NOTE: if the telescope is already in OFF state, the action returns immediately without sending the command to the central node.

Parameters:

wait_termination (bool) – set to False if you don’t want to wait for the termination condition. By default the termination condition is waited.
custom_timeout (Optional[int]) – A custom timeout (in seconds) to wait for the termination condition to occur. If None, the default action timeout is used. This parameter is useful only when wait_termination=True.
is_long_running_command (bool) – set to True if you want to include in the synchronisation the long running command. This parameter is useful only when wait_termination=True. By default is True.

Return type:

None | tuple[Any, list[str]]

move_to_on(wait_termination=True, custom_timeout=None, is_long_running_command=False)

Move the telescope to ON state.

NOTE: if the telescope is already in ON state, the action returns immediately without sending the command to the central node.

Parameters:

wait_termination (bool) – set to False if you don’t want to wait for the termination condition. By default the termination condition is waited.
custom_timeout (Optional[int]) – A custom timeout (in seconds) to wait for the termination condition to occur. If None, the default action timeout is used. This parameter is useful only when wait_termination=True.
is_long_running_command (bool) – set to True if you want to include in the synchronisation the long running command. This parameter is useful only when wait_termination=True. By default is False for MoveToOn action, since at the moment a failure can be observed. TODO: solve the issue and set to True.

Return type:

None | tuple[Any, list[str]]

release_resources(release_input, wait_termination=True, custom_timeout=None, is_long_running_command=True)

Invoke Release Resource command on central Node.

Parameters:

release_input (JSONInput) – Release resource input json.
wait_termination (bool) – set to False if you don’t want to wait for the termination condition. By default the termination condition is waited.
custom_timeout (Optional[int]) – A custom timeout (in seconds) to wait for the termination condition to occur. If None, the default action timeout is used. This parameter is useful only when wait_termination=True.
is_long_running_command (bool) – set to True if you want to include in the synchronisation the long running command. This parameter is useful only when wait_termination=True. By default is True.

Return type:

tuple[Any, list[str]]

Returns:

result, message

restart(wait_termination=True, custom_timeout=None, is_long_running_command=True)

Invoke Restart command on subarray Node.

Parameters:

wait_termination (bool) – set to False if you don’t want to wait for the termination condition. By default the termination condition is waited.
custom_timeout (Optional[int]) – A custom timeout (in seconds) to wait for the termination condition to occur. If None, the default action timeout is used. This parameter is useful only when wait_termination=True.
is_long_running_command (bool) – set to True if you want to include in the synchronisation the long running command. This parameter is useful only when wait_termination=True. By default is True.

Return type:

tuple[Any, list[str]]

Returns:

result, message

run_command_on_central_node(command_name, command_input=None, is_long_running_command=False)

Run a generic command on central node.

NOTE: the termination condition is empty, so the action will return immediately.

Parameters:

command_name (str) – Name of command to execute.
command_input (Optional[JSONInput]) – Json send as input to execute command.
is_long_running_command (bool) – set to True if you want to include in the synchronisation the long running command. This parameter is useful only when wait_termination=True. By default is False (we assume the command is not a LRC).

Return type:

tuple[Any, list[str]]

run_command_on_subarray_node(command_name, command_input=None, expected_obs_state=None, custom_timeout=None, is_long_running_command=False)

Run a generic command on subarray node.

NOTE: the termination condition by default is empty. You can although wait for a certain obsState to be reached setting expected_obs_state or the LRC completion.

Parameters:

command_name – Name of command to execute
command_input – JSON input for the command. By default None.
expected_obs_state – Expected obsState after command execution. By default no expected obsState (=> no waiting for termination condition).
custom_timeout – A custom timeout (in seconds) to wait for the termination condition to occur. If None, the default action timeout is used. This parameter is useful only when expected_obs_state is specified or when is_long_running_command=True.
is_long_running_command – set to True if you want to include in the synchronisation the long running command. This parameter is useful only when wait_termination=True. By default is False (we assume the command is not a LRC).

Returns:

result, message

scan(scan_input, wait_termination=True, custom_timeout=None, is_long_running_command=True)

Invoke Scan command on subarray Node.

Parameters:

scan_input (JSONInput) – json input for scan command.
wait_termination (bool) – set to False if you don’t want to wait for the termination condition. By default the termination condition is waited.
custom_timeout (Optional[int]) – A custom timeout (in seconds) to wait for the termination condition to occur. If None, the default action timeout is used.
is_long_running_command (bool) – set to True if you want to include in the synchronisation the long running command. This parameter is useful only when wait_termination=True. By default is True.

Return type:

tuple[Any, list[str]]

Returns:

result, message

property sdp_master_leaf_node: tango.DeviceProxy: The SDP master leaf node Tango device proxy.

property sdp_subarray_leaf_node: Return SDP Subarray Leaf Node Proxy

set_standby(wait_termination=True, custom_timeout=None, is_long_running_command=True)

Set the telescope to STANDBY state.

Parameters:

wait_termination (bool) – set to False if you don’t want to wait for the termination condition. By default the termination condition is waited.
custom_timeout (Optional[int]) – A custom timeout (in seconds) to wait for the termination condition to occur. If None, the default action timeout is used. This parameter is useful only when wait_termination=True.
is_long_running_command (bool) – set to True if you want to include in the synchronisation the long running command. This parameter is useful only when wait_termination=True. By default is True.

Return type:

tuple[Any, list[str]]

set_subarray_id(requested_subarray_id)

Make the ITH connect to the subarray devices with the given ID.

By default, the ITH is connected to a certain subarray device. This method allows to change the subarray devices that are used by the various test harness actions.

Parameters:: requested_subarray_id (str) – Subarray ID to set.
Return type:: None

property subarray_node: Return Subarray Node Proxy

class ska_integration_test_harness.facades.TMCSubarrayNodeFacade(telescope)

A facade to TMC Subarray Node device and its actions.

A facade to TMC subsystem, providing a simplified interface to the subarray node devices and their actions. It contains:

references to subarray node device,
references to leaf devices to interact with CSP and SDP subarrays,
an action to initialise the subarray setting the subarray ID,
actions to move the subarray to ON and OFF states,
actions to interact with the obs state of the subarray, making individual state changes (through command calls) or forcing the change of the obs state to a target state whatever the current state is,
various other actions (e.g., five point calibration scan).

IMPORTANT NOTE: This facade is deprecated, since we want to use a facade for each subsystem. Please use TMCFacade instead.

abort(wait_termination=True)

Invoke Abort command on subarray Node.

Parameters:: wait_termination (bool) – set to False if you don’t want to wait for the termination condition. By default the termination condition is waited.
Returns:: result, message

assign_resources(assign_input, wait_termination=True)

Invoke Assign Resource command on subarray Node

Parameters:

assign_input (JSONInput) – Assign resource input json.
wait_termination (bool) – set to False if you don’t want to wait for the termination condition. By default the termination condition is waited.

Returns:

result, message

configure(configure_input, wait_termination=True)

Invoke configure command on subarray Node.

Parameters:

configure_input (JSONInput) – json input for configure command.
wait_termination (bool) – set to False if you don’t want to wait for the termination condition. By default the termination condition is waited.

Returns:

result, message

property csp_master_leaf_node: Return CSP Master Leaf Node Proxy

property csp_subarray_leaf_node: Return CSP Subarray Leaf Node Proxy

property dish_leaf_node_list: Return the dish leaf nodes as a list of device proxies.

end_observation(wait_termination=True)

Invoke End command on subarray Node.

Parameters:: wait_termination (bool) – set to False if you don’t want to wait for the termination condition. By default the termination condition is waited.
Returns:: result, message

end_scan(wait_termination=True)

Invoke EndScan command on subarray Node.

Parameters:: wait_termination (bool) – set to False if you don’t want to wait for the termination condition. By default the termination condition is waited.
Returns:: result, message

execute_transition(command_name, command_input=None, expected_obs_state=None)

Execute provided command on subarray

Parameters:

command_name – Name of command to execute
command_input – JSON input for the command. By default None.
expected_obs_state – Expected obsState after command execution. By default no expected obsState (=> no waiting for termination condition).

Returns:

result, message

force_change_of_obs_state(dest_state_name, commands_inputs, wait_termination=True)

Force SubarrayNode obsState to provided obsState.

Parameters:

commands_inputs (TestHarnessInputs) – The JSON inputs for the commands to bring the subarray in a certain obs state. You can pass just the JSON inputs you need, but if one of them is missing, you may occur in an error.
dest_state_name (ObsState) – Name of the destination obsState.
commands_inputs – The JSON inputs for the commands to bring the subarray in a certain obs state. You can pass just the JSON inputs you need, but if one of them is missing, you may occur in an error.
wait_termination (bool) – set to False if you don’t want to wait for the termination condition. By default the termination condition is waited.

Return type:

None

move_to_off(wait_termination=True)

Move Subarray to OFF state.

Parameters:: wait_termination (bool) – set to False if you don’t want to wait for the termination condition. By default the termination condition is waited.
Returns:: result, message

move_to_on(wait_termination=True)

Move subarray to ON state.

Parameters:: wait_termination (bool) – set to False if you don’t want to wait for the termination condition. By default the termination condition is waited.
Returns:: result, message

release_all_resources(wait_termination=True)

Invoke Release Resource command on subarray Node.

Parameters:: wait_termination (bool) – set to False if you don’t want to wait for the termination condition. By default the termination condition is waited.
Returns:: result, message

restart(wait_termination=True)

Invoke Restart command on subarray Node.

Parameters:: wait_termination (bool) – set to False if you don’t want to wait for the termination condition. By default the termination condition is waited.
Returns:: result, message

scan(scan_input, wait_termination=True)

Invoke Scan command on subarray Node.

Parameters:

scan_input (JSONInput) – json input for scan command.
wait_termination (bool) – set to False if you don’t want to wait for the termination condition. By default the termination condition is waited.

Returns:

result, message

property sdp_master_leaf_node: Return SDP Master Leaf Node Proxy

property sdp_subarray_leaf_node: Return SDP Subarray Leaf Node Proxy

set_subarray_id(requested_subarray_id)

Create subarray devices for the requested subarray.

Parameters:: requested_subarray_id (str) – Subarray ID to set.
Return type:: None

property subarray_node: Return Subarray Node Proxy

Central Node actions

A collection of actions performed on TMC Central Node.

Some of the given actions are basic command calls (often followed by a synchronisation on one or more expected state changes), while others are more complex orchestrations (which occasionally include calls on the wrappers to deal with complexities derived by the emulated-not emulated status of the involved subsystems).

class ska_integration_test_harness.actions.central_node.CentralNodeAssignResources(assign_input)

Invoke Assign Resource command on CentralNode.

termination_condition_for_quiescent_state(): All subarrays must reach the IDLE state and LRC must terminate.

termination_condition_for_transient_state(): All subarrays must reach the RESOURCING state.

class ska_integration_test_harness.actions.central_node.CentralNodeLoadDishConfig(dish_vcc_config)

Invoke LoadDishCfg command on CentralNode.

termination_condition(): The dishes configuration has been changed and LRC has terminated.

class ska_integration_test_harness.actions.central_node.CentralNodeReleaseResources(release_input)

Invoke ReleaseResources on the CentralNode.

termination_condition()

All subarrays are in EMPTY state and resources are released.

(and LRC must terminate).

class ska_integration_test_harness.actions.central_node.CentralNodeRunCommand(command_name, command_input=None, is_long_running_command=False)

Invoke a generic command on CentralNode.

This action is used to execute any command on CentralNode. The command will be called on the Tango device which is pointed to be the TMC CentralNode.

The command can optionally have a JSON input. If it does, the input will be sent to the command as a string.

The command can be a long running command or not. If it is a long running command, the action will wait for the command to terminate.

command_input: The JSON input for the command (if any).

command_name: The name of the command to be executed (e.g., “AssignResources”).

class ska_integration_test_harness.actions.central_node.MoveToOff

Ensure the central node is in OFF state.

This action is a wrapper around the MoveToOffCommand action, which sends the TelescopeOff command to the central node only if the central node is not already in OFF state. This action is useful when you want to ensure that the central node is in OFF state, but you don’t want to send the command if the central node is already in OFF state.

The result of this action is None if the central node is already in OFF state, otherwise it is the result of the MoveToOffCommand action.

The termination conditions always include the expected side effects of the MoveToOffCommand action. The LRC instead is verified only if the MoveToOffCommand action is executed.

Policies about the termination condition are propagated to the MoveToOffCommand action.

termination_condition()

Master and subarray devices are in OFF state.

Master and subarray devices should be in OFF state, while all dishes should be in STANDBY_LP mode.

class ska_integration_test_harness.actions.central_node.MoveToOffCommand

Send TelescopeOff to the central node and wait for termination.

expected_side_effects()

Expected command side-effects, excluding LRC termination.

Devices are in OFF state, dishes are in STANDBY_LP mode.

termination_condition()

Master and subarray devices are in OFF state (+ LRC terminates).

Master and subarray devices should be in OFF state, while all dishes should be in STANDBY_LP mode and LRC must terminate.

class ska_integration_test_harness.actions.central_node.MoveToOn

Ensure the central node is in ON state.

This action is a wrapper around the MoveToOnCommand action, which sends the TelescopeOn command to the central node only if the central node is not already in ON state. This action is useful when you want to ensure that the central node is in ON state, but you don’t want to send the command if the central node is already in ON state.

The result of this action is None if the central node is already in ON state, otherwise it is the result of the MoveToOnCommand action.

The termination conditions always include the expected side effects of the MoveToOnCommand action. The LRC instead is verified only if the MoveToOnCommand action is executed.

Policies about the termination condition are propagated to the MoveToOnCommand action.

termination_condition()

Master and subarray devices are in ON state.

Master and subarray devices should be in ON state, while all dishes should be in STANDBY_FP mode.

class ska_integration_test_harness.actions.central_node.MoveToOnCommand

Send TelescopeOn to the central node and wait for termination.

expected_side_effects()

Expected command side-effects, excluding LRC termination.

Devices are in ON state, dishes are in STANDBY_FP mode.

termination_condition()

Master and subarray devices are in ON state (+ LRC terminates).

Master and subarray devices should be in ON state, while all dishes should be in STANDBY_FP mode and LRC must terminate.

class ska_integration_test_harness.actions.central_node.SetStandby

An action to set the central node to STANDBY State.

termination_condition(): Central node should be in STANDBY state and so also SDP all dishes should be in STANDBY_LP mode and LRC must terminate.

Subarray actions

A collection of actions performed on TMC subarray Node.

The following actions are basic command calls (often followed by a synchronisation on one or more expected state changes):

SubarrayAbort
SubarrayRestart
SubarrayAssignResources
SubarrayReleaseAllResources
SubarrayConfigure
SubarrayScan
SubarrayEndObservation
SubarrayEndScan
SubarrayMoveToOn
SubarrayMoveToOff
SubarrayExecuteTransition (which is a generic action for executing any transition on the subarray)

The following instead are more complex orchestrations (or tools for facilitating those orchestrations):

ForceChangeOfObsState
SubarrayObsStateResetterFactory
SubarrayForceAbort

class ska_integration_test_harness.actions.subarray.ForceChangeOfObsState(dest_state_name, commands_input)

Force the change of the ObsState in Subarray to a target state.

This action is used to force the change of the ObsState in TMC Subarray, regardless of the current state of the Subarray. The action will move the state machine to the target state, by executing the necessary steps to reach it (e.g., assign, configure, scan) in a correct and consistent way.

This action has no termination condition, but you can customise:

the wait timeout of each of the steps, by calling the method
set_termination_condition_timeout(timeout).
the termination condition policy of the last step, by calling the method
set_termination_condition_policy(True) (to ensure all steps wait for the termination condition) or set_termination_condition_policy(False) (to make the last step not wait for the termination condition).

By default each step will keep the default wait termination condition and each step will wait for its termination condition.

If the procedure to reach the target state is not implemented, the action will raise a NotImplementedError.

termination_condition(): No expected outcome for this action.

class ska_integration_test_harness.actions.subarray.SubarrayAbort

Invoke Abort command on subarray Node.

This action is expected to move the subarray to the ABORTING state (transient) and then to the ABORTED state (quiescent and stable).

termination_condition_for_quiescent_state()

All subarrays must be in ABORTED state.

Return type:: list[ExpectedEvent]

termination_condition_for_transient_state()

All subarrays must be in ABORTING state.

Return type:: list[ExpectedEvent]

class ska_integration_test_harness.actions.subarray.SubarrayAssignResources(assign_input)

Invoke Assign Resource command on subarray Node.

This action is expected to move the subarray to the RESOURCING state (transient) and then to the IDLE state (quiescent and stable).

This action is not supposed to be used to release resources, unless you subclass it and override the termination conditions.

termination_condition_for_quiescent_state(): All subarrays must reach the IDLE state (and LRC must terminate).

termination_condition_for_transient_state(): All subarrays must reach the RESOURCING state.

class ska_integration_test_harness.actions.subarray.SubarrayConfigure(configure_input)

Invoke configure command on subarray Node.

This action is expected to move the subarray to the CONFIGURING state (transient) and then to the READY state (quiescent and stable).

termination_condition_for_quiescent_state()

All subarrays must reach the READY state (and LRC must terminate).

Also, all dishes must be in OPERATE dishMode and TRACK pointingState.

Return type:: list[ExpectedEvent]

termination_condition_for_transient_state()

All subarrays must reach the CONFIGURING state.

Return type:: list[ExpectedEvent]

class ska_integration_test_harness.actions.subarray.SubarrayEndObservation

Invoke End command on subarray Node.

termination_condition(): All subarrays must be in IDLE state (and LRC must terminate).

class ska_integration_test_harness.actions.subarray.SubarrayEndScan

Invoke EndScan command on subarray Node.

termination_condition(): All subarrays must be in READY state (and LRC must terminate).

class ska_integration_test_harness.actions.subarray.SubarrayForceAbort

Foreach of the individual subarray devices, force an abort if an abort process is not ongoing.

termination_condition(): All subarrays must be in ABORTED state.

class ska_integration_test_harness.actions.subarray.SubarrayMoveToOff: Invoke MoveToOff command on subarray Node.

class ska_integration_test_harness.actions.subarray.SubarrayMoveToOn: Invoke MoveToOn command on subarray Node.

class ska_integration_test_harness.actions.subarray.SubarrayObsStateResetterFactory(commands_inputs)

Factory to create Telescope Actions to bring subarray in a obs state.

This factory is used to create TelescopeAction s to bring the subarray in a certain obs state. The factory provides methods to create composite actions that can move subarray to:

Empty state
Resourcing state
Idle state
Aborting state
Aborted state
Configuring state
Ready state
Scanning state

The starting state of the subarray is not considered while creating, because all the actions are designed to reset the subarray to EMPTY state first and then move to the target state.

create_action_to_reset_subarray_to_aborted()

Create a TelescopeAction to reset the subarray to ABORTED.

Return type:: TelescopeAction
Returns:: A TelescopeAction to reset the subarray to the ObsState.ABORTED state.

create_action_to_reset_subarray_to_aborting()

Create a TelescopeAction to reset the subarray to ABORTING.

Return type:: TelescopeAction
Returns:: A TelescopeAction to reset the subarray to the ObsState.ABORTING state.

create_action_to_reset_subarray_to_configuring()

Create a TelescopeAction to reset the subarray to CONFIGURING.

Return type:: TelescopeAction
Returns:: A TelescopeAction to reset the subarray to the ObsState.CONFIGURING state.

create_action_to_reset_subarray_to_empty()

Create a TelescopeAction to reset the subarray to EMPTY.

Return type:: TelescopeAction
Returns:: A TelescopeAction to reset the subarray to the ObsState.EMPTY state.

create_action_to_reset_subarray_to_idle()

Create a TelescopeAction to reset the subarray to the IDLE.

Return type:: TelescopeAction
Returns:: A TelescopeAction to reset the subarray to the ObsState.IDLE state.

create_action_to_reset_subarray_to_ready()

Create a TelescopeAction to reset the subarray to READY.

Return type:: TelescopeAction
Returns:: A TelescopeAction to reset the subarray to the ObsState.READY state.

create_action_to_reset_subarray_to_resourcing()

Create a TelescopeAction to reset the subarray to RESOURCING.

Return type:: TelescopeAction
Returns:: A TelescopeAction to reset the subarray to the ObsState.RESOURCING state.

create_action_to_reset_subarray_to_restarting()

Create a TelescopeAction to reset the subarray to RESTARTING.

Return type:: TelescopeAction
Returns:: A TelescopeAction to reset the subarray to the ObsState.RESTARTING state.

create_action_to_reset_subarray_to_scanning()

Create a TelescopeAction to reset the subarray to SCANNING.

Return type:: TelescopeAction
Returns:: A TelescopeAction to reset the subarray to the ObsState.SCANNING state.

create_action_to_reset_subarray_to_state(target_state)

Create a TelescopeAction to reset the subarray to the given state.

Parameters:: target_state (ObsState) – The target state to which the subarray should be reset.
Return type:: TelescopeAction
Returns:: A TelescopeAction to reset the subarray to the given state.
Raises:: NotImplementedError – If the procedure to reach the target state is not implemented.

class ska_integration_test_harness.actions.subarray.SubarrayReleaseAllResources

Invoke Release Resource command on subarray Node.

termination_condition(): Subarrays are in EMPTY, resources are released, LRC terminates.

class ska_integration_test_harness.actions.subarray.SubarrayRestart

Invoke Restart command on subarray Node.

This action is expected to be called when the subarray is in ABORTED state. This action will move the subarray to the RESTARTING state (transient) and then to the EMPTY state (quiescent and stable).

termination_condition_for_quiescent_state()

All subarrays must be in EMPTY state (and LRC must terminate).

Return type:: list[ExpectedEvent]

termination_condition_for_transient_state()

All subarrays must be in RESTARTING state.

Return type:: list[ExpectedEvent]

class ska_integration_test_harness.actions.subarray.SubarrayRunCommand(command_name, expected_obs_state=None, command_input=None, is_long_running_command=False)

Invoke a generic command on the TMC SubarrayNode.

This action is used to execute any command on the TMC SubarrayNode. The command will be called on the Tango device which is pointed to be the TMC SubarrayNode.

The command can optionally have a JSON input. If it does, the input will be sent to the command as a string.

The command can be a long running command or not. If it is a long running command, the action will wait for the command to terminate.

Optionally, you can specify the expected obsState after the command is called. If you do, the action will wait for all subarrays to have the expected obsState before it terminates.

command_input: The JSON input for the command (if any).

command_name: The name of the command to be executed (e.g., “Configure”).

expected_obs_state

The expected obsState after the command is executed.

(e.g., ObsState.READY). If not provided, the action will have an empty termination condition. If provided, the termination condition will be that all subarrays must have the expected obsState.

termination_condition()

Eventual LRC termination check + eventual expected obsState check.

This action may have those two termination conditions:

The LRC must terminate (if the command is a LRC).
All subarrays must have the expected obsState (if provided).

class ska_integration_test_harness.actions.subarray.SubarrayScan(scan_input)

Invoke Scan command on subarray Node.

termination_condition(): All subarrays must be in SCANNING state (and LRC must terminate).

Wrappers to emulated subsystems

A collection of wrapper of emulated subsystem wrappers (+ utils).

class ska_integration_test_harness.emulated.EmulatedCSPWrapper(csp_configuration)

A wrapper for an emulated CSP.

Differently from the production CSP wrapper, when a move to on or move to off command is called, the CSP master is set to the desired state directly. Moreover, the tear down implements the usual procedure for emulated devices.

clear_command_call()

Clear the command call on the CSP.

Return type:: None

is_emulated()

Check if the subsystem is emulated.

Return type:: bool
Returns:: True if the subsystem is emulated, False otherwise.

tear_down()

Tear down the CSP.

The procedure is the following: - Reset the health state for the CSP master and the CSP subarray. - Clear the command call on the CSP subarray. - Reset the transitions data for the CSP subarray. - Reset the delay for the CSP subarray.

Return type:: None

class ska_integration_test_harness.emulated.EmulatedDishesWrapper(dishes_configuration)

A wrapper for emulated dishes.

Differently from the production dishes wrapper, there is no need to initialise the dishes in a Tango database (just an initial state setup is needed). The tear down procedure is the usual one for emulated devices.

clear_command_call()

Clear the command call on the Dishes.

Return type:: None

is_emulated()

Check if the subsystem is emulated.

Return type:: bool
Returns:: True if the subsystem is emulated, False otherwise.

tear_down()

Tear down the dishes.

The procedure is the following: :rtype: None

Reset the health state for the dishes.
Clear the command call on the dishes.
Reset the transitions data for the dishes.
Reset the delay for the dishes.
Reset the attributes dish mode and state on the dishes.

class ska_integration_test_harness.emulated.EmulatedSDPWrapper(*args, **kwargs)

A wrapper for an emulated SDP.

Differently from the production SDP wrapper, the tear down implements the usual procedure for emulated devices.

clear_command_call()

Clear the command call on the SDP.

Return type:: None

configure_receive_address(): Configure the receive address for the SDP subarray.

is_emulated()

Check if the subsystem is emulated.

Return type:: bool
Returns:: True if the subsystem is emulated, False otherwise.

tear_down()

Tear down the SDP.

The procedure is the following: - Reset the health state for the SDP master and the SDP subarray. - Clear the command call on the SDP subarray. - Reset the transitions data for the SDP subarray. - Reset the delay for the SDP subarray.

Return type:: None

class ska_integration_test_harness.emulated.EmulatedTeardownHelper

Tear down helper for emulated telescope subsystems.

static clear_command_call(devices)

Clear the command call of the given devices.

Return type:: None

static reset_delay(devices)

Reset the delay of the given devices.

Return type:: None

static reset_health_state(devices)

Reset the health state of the given devices.

Return type:: None

static reset_transitions_data(devices)

Reset the transition data of the given devices.

Return type:: None

static unset_defective_status(devices)

Unset the defective status of the given devices.

Return type:: None

Wrappers to production subsystems

A collection of production subsystem wrappers.

class ska_integration_test_harness.production.ProductionCSPWrapper(csp_configuration, all_production=False)

A wrapper for a production CSP.

Differently from the emulated CSP wrapper, when a move to on command is called, the CSP master is supposed to be in OFF state, so it moves to OFF state.

ensure_admin_mode_online()

Ensure the CSP master is in ONLINE admin mode.

Return type:: None

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.production.ProductionDishesWrapper(dishes_configuration)

A wrapper for production dishes.

Different from the emulated dishes wrapper, this class initialises in the Tango database the real Dish devices.

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.production.ProductionSDPWrapper(sdp_configuration)

A wrapper for a production SDP.

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.production.ProductionTMCWrapper(tmc_configuration, default_commands_input)

Production wrapper for TMC devices.

is_emulated()

Check if the subsystem is emulated.

Return type:: bool
Returns:: True if the subsystem is emulated, False otherwise.

tear_down()

Tear down the TMC devices.

This method will: :rtype: None

Release resources from the central node (if the subarray is in a IDLE state).
Force the subarray state to EMPTY.
Move the central node to OFF.
Move the subarray to OFF.
Load the default dish VCC configuration (if needed and if in Mid).