CSPFacade

class ska_csp_lmc_common.testing.k8s_test.csp_facade.CSPFacade(sut: SUTWrapper, num_subarrays, num_beams, logger, attr_timeout=20, teardown_timeout=30)

Bases: object

A facade to Central Signal Processor and its actions.

A facade to CSP sub-system, providing a simplified interface to the system and its actions. The facade is the entry point of CSP.LMC integration tests.

To initialize the CSP testing framework, instantiate CSPFacade with a configured CSPWrapper instance:

csp_facade = CSPFacade(csp_wrapper_instance)

Once testing is complete, the entire CSP test environment can be shut down and reset to its initial state by invoking the teardown method.

# Tear down the CSP test structure
csp_facade.teardown()
property lmc: SubsystemWrapper

The CSP.LMC wrapper

property cbf: SubsystemWrapper

The CBF wrapper

property pst: SubsystemWrapper

The PST wrapper

property pss: SubsystemWrapper

The PSS wrapper

turn_on_pst_beam(beam_id: int, check_initial: bool = True)

Turn on all or some of the PST beams.

Parameters:

  • beam_id – The PST beam ID to turn on.

  • check_initial – Whether to enforce an initial-status check before sending the command.

command_on_subarray(command_name: str, sub_id: str, postconditions: dict, preconditions: dict)

Executes a command on a given subarray and check for given postconditions.

Parameters:

  • command_name – The name of the command to execute.

  • sub_id – The id of the subarray to execute the command on.

  • postconditions – A dictionary of postconditions to check after the command has been executed.

Raises:

AssertionError – If any of the postconditions are not met after the command has been executed.

assign_resources_to_subarray(sub_id: str, beam_ids: List[int] | None = None)

Assigns multiple PST beams to a subarray.

Parameters:

  • sub_id – The id of the subarray to assign the PST beams to.

  • beam_ids – A list of PST beam IDs to assign. If None, no beams will be assigned.

The function first checks if the subarray is in its initial status. If not, it resets all devices and then turns on the requested PST beams.

Then it checks if any of the PST beams are already assigned to another subarray. If they are, it resets all devices and then turns on the requested PST beams.

Then it assigns the PST beams to the requested subarray, if they are not already assigned. It checks that the subarray is assigned with the PST beams by checking its state and assignedTimingBeamIDs attribute.

Raises:

AssertionError – If the subarray is not assigned with the PST beams after the assignment.

configure_subarray(sub_id: str, beam_ids: List[int] | None = None)

Configures a subarray with a PST beam.

Parameters:

  • sub_id – The id of the subarray to assign the PST beam to.

  • beam_ids – A list of PST beam IDs to assign. If None, no beams will be assigned.

The function first assigns the PST beam to the requested subarray. Then it configures the subarray. The function checks that the subarray is in the READY state after the configuration.

abort_during_configure(sub_id: str, configuring_timeout: float = 10.0, abort_timeout: int = 20, abort_delay_s: float = 0.0) → str

Start Configure on a subarray, wait until it enters CONFIGURING, then issue Abort.

This is intended for integration tests that need to verify abort behaviour during the Configure command without coupling the test to a specific internal implementation detail such as update_json.

Parameters:

  • sub_id – The id of the subarray to configure and abort.

  • configuring_timeout – Maximum time to wait for CONFIGURING.

  • abort_timeout – Timeout used for the Abort LRC postconditions.

  • abort_delay_s – Additional deterministic delay after CONFIGURING is observed and before Abort is sent.

Returns:

Configure command identifier.

start_scanning(sub_id: str, beam_ids: List[int] | None = None)

Initiates a scanning process on a subarray with a specified PST beam.

This function first configures the subarray with the given PST beam, ensuring that the subarray transitions to the READY state. It then issues a scan command to the subarray, transitioning its observation state to SCANNING.

Parameters:

  • sub_id – The ID of the subarray to scan.

  • beam_ids – A list of PST beam IDs to use for scanning.

end_scan(sub_id: str, beam_ids: List[int] | None = None)

Ends a scanning process on a subarray with a specified PST beam.

This function first starts a scanning process on the specified subarray with the given PST beam, and then ends the scanning process.

Parameters:

  • sub_id – The ID of the subarray to end the scan.

  • beam_ids – A list of PST beam IDs to use for scanning.

deconfigure_subarray(sub_id: str, beam_ids: List[int], when_deconfigured: str)

Deconfigures a subarray with a PST beam.

This function first configures or ends a scan on the specified subarray with the given PST beam, depending on the value of when_deconfigured. It then issues a GoToIdle command to the subarray, transitioning its observation state to IDLE.

Parameters:

  • sub_id – The ID of the subarray to deconfigure.

  • beam_ids – A list of PST beam IDs to use for deconfiguring.

  • when_deconfigured – The point at which the subarray is deconfigured. Must be either “before a scan” or “after a scan”.

abort_scan(sub_id: str, beam_ids: list[int])

Aborts a scan in the subarray with a PST beam.

This function first starts a scanning process on the specified subarray with the given PST beam, and then aborts the scanning process.

Parameters:

  • sub_id – The ID of the subarray to abort the scan.

  • beam_id – The ID of the PST beam to use for scanning.

generic_lrc_action(command_name: str, proxy: DeviceProxy, command_input: str | None = None, success: bool = True, failure: bool = False, rejection: bool = False, not_allowed: bool = False)

Create a TangoLRCAction that sends a command to a device and includes expected result codes and LRC error handling.

Parameters:

  • command_name – The command to send to the device

  • proxy – The device proxy that the command is sent to

  • command_input – The input to the command if required

Returns:

The TangoLRCAction

evaluate_command_input(command_name: str, proxy: DeviceProxy)

Evaluate the command input for a given command and proxy.

If the command_name is in self.command_inputs, this method evaluates the command input and returns the evaluated input.

If the command_name is Configure, Assign, Scan or ReleaseResources, and the proxy is a subarray, this method sets the subarray_id field of the command input to the subarray number of the proxy.

Parameters:

  • command_name – The name of the command to execute.

  • proxy – The DeviceProxy representing the target Tango device.

Returns:

The evaluated command input.

_format_command_input(command_name: str, proxy: DeviceProxy, command_input: str) → str

Hook for subclasses to modify the command input. Base implementation handles subarray_id injection.

set_adminmode(initial_adminmode: str, proxy: DeviceProxy)

Set the admin mode of a device to the specified initial mode.

This method checks the current admin mode of a device and sets it to the given initial admin mode if they are different. It logs the process and ensures the ‘isCommunicating’ attribute is consistent with the admin mode.

Parameters:

  • proxy – The device proxy to set the admin mode on.

  • initial_adminmode – The desired initial admin mode for the device.

perform_action(command_name: str, proxy: DeviceProxy, command_input: Any | None = None, success: bool = False, failure: bool = False, rejection: bool = False, not_allowed: bool = False, timeout: int = 20)

Execute a command successfully on a Tango device.

This method wraps the execution of a command as a LongRunningCommand (LRC) and ensures its completion by executing it with a specified postconditions timeout.

Parameters:

  • command_name – The name of the command to execute.

  • proxy – The DeviceProxy representing the target Tango device.

  • command_input – Optional input parameter for the command.

  • timeout – The time in seconds to wait for postconditions to be met.

hard_reset()

Reset all devices and then check their initial status

check_initial_status(controller_previous_value=DevState.DISABLE, subarrays_previous_value=DevState.DISABLE)

Check devices status after reset

translate_attributes_in_enum(attribute: str, value: str)

Translates a string attribute value to its corresponding enumeration type.

This function takes an attribute name and its string value, typically parsed from Gherkin, and translates it into the corresponding value. The translation is based on a predefined map of attribute names to their respective enumeration classes.

Parameters:

  • attribute – The name of the attribute to be translated.

  • value – The string value of the attribute to be translated.

Returns:

The translated enumeration value.

static _normalize_health_info(value: Any) → Dict[str, List[str]]

Normalize a healthInfo payload into dict[str, list[str]].

Accepted input types are JSON string or dict payloads.

static _health_info_contains(health_info: Dict[str, List[str]], expected_subset: Dict[str, List[str]]) → bool

Check whether health_info contains all entries in expected_subset.

For each expected fqdn, all expected messages must be present as substrings in that fqdn message list.

check_health_info_payload_on_device(proxy: DeviceProxy, expected_payload: Any, timeout: float | None = None) → None

Assert that device healthInfo equals the expected payload.

The check uses a direct-read fast path and, if needed, event-based waiting through check_attribute_on_device.

check_health_info_contains_on_device(proxy: DeviceProxy, expected_subset: Any, timeout: float | None = None) → None

Assert that device healthInfo contains the requested subset payload.

The check uses a direct-read fast path and, if needed, event-based waiting through check_attribute_on_device.

check_attribute_on_device(attribute: str, value: Any, proxy: DeviceProxy, previous_value: Any | None = None, new_tracer=False, timeout: float | None = None, custom_matcher: Callable[[Any], bool] | None = None)

Check the value of an attribute on a specific proxy

Param:

attribute: the name of Tango attribute to be checked

Param:

value: the value of the attribute to be checked

Param:

proxy: the Tango device on which the attribute has to be checked

check_attribute_on_device_group(subsystem, attribute, value)

Checks that all devices within a specified subsystem have the given attribute set to the specified value.

This function translates the attribute and value to appropriate enumeration representations, identifies the devices within the subsystem, and verifies their attribute values. It logs the process and raises an error if the subsystem is unknown.

Parameters:

  • subsystem – The name of the subsystem to check attributes for.

  • attribute – The attribute to verify on each device.

  • value – The expected value of the attribute.

subscribe_if_not_subscribed(proxy: DeviceProxy, attribute: str)

If the event for the given proxy and attribute is not yet subscribed, subscribe it.

Parameters:

  • proxy – the device proxy to subscribe to

  • attribute – the attribute to subscribe to

_define_proxy_map(pss=True, pst=True) → List[DeviceProxy]

Define the list of device proxies to subscribe to.

Parameters:

self – the instance of the class

Returns:

list of device proxies

subscribe_relevant_events()

Subscribe to the relevant events for the device proxies provided.

It will subscribe to the following events: - State - healthstate - obsstate for non control devices - obsmode for non control devices

Parameters:

self – the instance of the class

_ensure_controller_online_for_teardown()

Ensure CSP controller and all known devices are ONLINE in teardown.

_ensure_subarray_online_for_teardown(subarray, subarray_index: int)

Ensure LMC and CBF subarray proxies are ONLINE before recovery.

_recover_subarray_obsstate_for_teardown(subarray)

Drive a non-EMPTY subarray back to EMPTY during teardown.

_turn_off_pst_beams_in_teardown()

Turn off PST beams at teardown if any beam is still ON.

teardown()

Tears down the CSP in an orderly fashion.

This method will put the Low CSP in an empty state, and the Low PST beams in an off state. It will abort any active scans and restart the subarrays to ensure they are in the empty state. It will then turn off the Low PST beams.

archive_callback(event)

Callback function invoked when an archive event is received from the monitored devices.

If the event is error-free, the value of the attribute is logged and the counter of the archive events for the attribute is incremented.

If the event contains an error, the error is logged.

Parameters:

event (tango.events.Event) – the event received from the monitored device