Test API package

conftest module

Path: tests/

conftest module for Subarray Node.

tests.conftest.aggregation_process_low()

Create aggregation process low object

Yields

Aggregation process low object

Return type

Generator[AggregationProcessLow, None, None]

tests.conftest.aggregation_process_mid()

Create aggregation process mid object

Yields

Aggregation procees mid object

Return type

Generator[AggregationProcessMid, None, None]

tests.conftest.change_event_callbacks()

Return a dictionary of Tango device change event callbacks with asynchrony support.

Return type

MockTangoEventCallbackGroup

Returns

a collections.defaultdict that returns change event callbacks by name.

tests.conftest.cm_without_er_lp_low()

Fixture to create and yield a SubarrayNodeComponentManagerLow instance without event receiver or liveliness probe for testing purposes.

Yields

SubarrayNodeComponentManagerLow – Configured component manager instance.

Return type

Generator[SubarrayNodeComponentManagerLow, None, None]

tests.conftest.component_manager_low()

Create a component manager instance for LOW.

Yields

Instance of component manager low

Return type

Generator[SubarrayNodeComponentManagerLow, None, None]

tests.conftest.component_manager_mid()

Create a component manager instance for MID.

Yields

Instance of component manager

Return type

Generator[SubarrayNodeComponentManagerMid, None, None]

tests.conftest.component_state_changed_callback(cm, state_change)

Handle change in this device’s state.

This is a callback hook, called whenever the state changes. It is responsible for updating the tango side of things i.e. making sure the attribute is up to date, and events are pushed.

Parameters

  • cm – Component Manager

  • state_change – A dictionary containing the name of the state that changed and its new value.

Return type

None

tests.conftest.devices_to_load()

This method returns the devices whose context are required

Return type

dict

Returns

Dictionary of dictionaries containing the class and devices

tests.conftest.event_recorder()

Records events and allows assertions

Yields

EventRecorder instance

Return type

Generator[EventRecorder, None, None]

tests.conftest.get_input_str(path)

Returns input json string

Parameters

path (str) – File path

Returns

input json string

Return type

str

tests.conftest.get_scan_input_str(scan_input_file='Scan_mid.json')

Returns the scan input json string

Parameters

scan_input_file – name of the scan input file

Returns

scan input json string

Return type

str

tests.conftest.health_state_aggregation_process_low()

Create aggregation process low object

Yields

Healthstate aggregation process mid

Return type

Generator[HealthStateAggregationProcessor, None, None]

tests.conftest.health_state_aggregation_process_mid()

Create aggregation process mid object

Yields

Healthstate aggregation process mid

Return type

Generator[HealthStateAggregationProcessor, None, None]

tests.conftest.json_factory()

Json factory for getting json files

Returns

get json from files

Return type

str

tests.conftest.pytest_addoption(parser)

Pytest hook; implemented to add the –true-context option, used to indicate that a true Tango subsystem is available, so there is no need for a tango.test_context.MultiDeviceTestContext.

Parameters

parser (ArgumentParser) – the command line options parser

Return type

None

tests.conftest.pytest_sessionstart(session)

Pytest hook; prints info about tango version.

Parameters

session (Session) – a pytest Session object

Return type

None

tests.conftest.set_low_sdp_csp_leaf_node_availability_for_aggregation()

Setting low Csp subarray leaf node and Sdp subarray leaf node availabilty attribute isSubsystemAvailable as True for aggregation

Return type

None

tests.conftest.set_low_sdp_csp_mccs_admin_modes()

Setting low Csp subarray and Sdp subarray adminModes attribute as Online.

tests.conftest.set_mid_sdp_csp_admin_modes()

Setting mid Csp subarray and Sdp subarray adminModes attribute as Online.

tests.conftest.set_mid_sdp_csp_leaf_node_availability_for_aggregation()

Setting mid Csp subarray leaf node and Sdp subarray leaf node availabilty attribute isSubsystemAvailable as True for aggregation

Return type

None

tests.conftest.subarray_node(event_recorder, request)

Returns the instance of SubarrayNode class

Parameters

  • event_recorder – Event recorder instance

  • request – pytest request object

Yields

SubarrayNode instance

Return type

Generator[SubarrayNode, None, None]

tests.conftest.tango_context(devices_to_load, request)

This Fixture provide the tango context fixture

Parameters

  • devices_to_load – devices to load

  • request – pytest request object

Yields

Tango context

Return type

Generator[MultiDeviceTestContext, None, None]

tests.conftest.task_callback()

Creates a mock callable for asynchronous testing

Return type

MockCallable

Returns

MockCallable

common_utils module

Path: tests/test_helpers/common_utils.py

Common utils class for a simple test harness.

class tests.test_helpers.common_utils.SubarrayNode(deployment='MID')

Bases: object

Class for Subarray Node device for test harness

assert_command_completion(unique_id, event_recorder)

Checks if the command was completed successfully.

Raises AssertionError in case of failure.

Parameters

  • unique_id (str) – Unique id of the command

  • event_recorder (EventRecorder) – Event Recorder

Return type

None

assert_command_completion_on_device(device_name, expected_state)

Check for device obsState to transition to the correct state.

Parameters

  • device_name (str) – Name of the device

  • expected_state (ObsState) – Expected ObsState

Return type

None

assert_command_failure(unique_id, exception_message, event_recorder, result_code=ska_tango_base.commands.ResultCode.FAILED)

Checks if the command pushed the correct error message event on its longRunningCommandResult attribute.

Raises AssertionError in case of failure.

Parameters

  • unique_id (str) – Unique id of the command

  • exception_message (str) – Exception message in case of failure

  • event_recorder (EventRecorder) – Event Recorder

  • result_code (ResultCode) – ResultCode, default is ResultCode.FAILED

Return type

None

assert_health_state(healthState)

Asserts the SubarrayNode HealthState to be the HealthState specified.

Raises AssertionError in case of failure.

Parameters

healthState (HealthState) – Health State

Return type

None

assert_obs_state(obsstate)

Asserts the SubarrayNode ObsState to be the ObsState specified.

Raises AssertionError in case of failure.

Parameters

obsstate (ObsState) – ObsState

Return type

None

assert_obsstate_update(obs_state, event_recorder)

Checks if the command pushed the expected obsState event

Raises AssertionError in case of failure.

Parameters

  • obs_state (ObsState) – ObsState to be expected

  • event_recorder (EventRecorder) – Event Recorder

Return type

None

clear_command_call_info()

Clears the command call info attribute.

Return type

None

get_device_obs_states()

Returns a dictionary with current device ObsStates

Returns

Dictionary with device names as keys and their ObsStates

Return type

dict

invoke_abort_command(event_recorder)

Invoke Abort command on Subarray Node

Parameters

event_recorder (EventRecorder) – Event Recorder

Return type

None

invoke_and_assert_command_rejected(command_name, exception_message, **kwargs)

Invokes the command on Subarray Node and asserts its rejection. Raises AssertionError in case of failure.

Parameters

  • command_name (str) – Name of the command to be invoked.

  • exception_message (str) – Exception message to assert after the command is rejected

  • kwargs – Additional keyword arguments. Example- input_str which is the input parameter for command call.

Return type

None

invoke_command(command_name, input_str='')

Method to invoke given command on subarray node with given input json if any. Returns the unique id from command invocation.

Parameters

  • command_name (str) – Name of the command to invoke

  • input_str (str) – input json

Return type

str

Returns

Unique id

invoke_command_on_device(device_name, command, input_data=None)

Invokes the given command on the device with provided input.

Parameters

  • device_name (str) – Name of the device

  • command (str) – Command to be invoked

  • input_data – Input data for the command, default is None

Return type

None

perform_action(device_names, actions)

Performs the said action on the given device

Parameters

  • device_names (List[str]) – List of device names

  • actions (List[str]) – List of actions to be performed

Return type

None

reset_defects_for_devices(device_names)

Resets the defect parameters for given devices

Parameters

device_names (list) – List of device names

Return type

None

set_devices_availability(device_names, availability=True)

Set the device availability to the given value.

Parameters

  • device_names (list) – List of device names

  • availability (bool) – Availability status to be set Default is True (available).

Return type

None

set_devices_availablity(device_availability)

Sets the availability of devices with provided parameters.

Parameters

device_availability (dict) – Dictionary with device names as keys and their availability as values.

Return type

None

set_devices_defective(device_names, defective_params)

Sets the given devices defective with provided parameters.

Parameters

  • device_names (list) – List of device names

  • defective_params (str) – Defective parameters to be set

Return type

None

set_subarray_health_state_to_ok(event_recorder)

Set Subarray Health state to OK

Parameters

event_recorder (EventRecorder) – Event Recorder

Return type

None

updated_tear_down(event_recorder)

Teardown for integration and acceptance testing. Brings the subarray node back to EMPTY state.

Raises given exception if the flag is set True.

Parameters

event_recorder (EventRecorder) – Event Recorder

Return type

None

wait_for_attribute_to_change_to(device, attribute_name, attribute_value)

Wait for the attribute to change to given value.

Parameters

  • device (str) – Name of the device

  • attribute_name (str) – Attribute name as a string

  • attribute_value (Any) – Value of attribute to be asserted

Raises

AssertionError – If the attribute value is not equal to the given value within the timeout period.

Return type

None

wait_for_leaf_node_obsstate(device_name, obsstate, timeout=100)

Wait for Device ObsState to be the given obsstate till timeout occurs.

Raises Exception in case of a timeout.

Parameters

  • device_name (str) – Name of the device

  • obsstate (ObsState) – ObsState

  • timeout (int) – timeout in sec for waiting to change obsstate

Raises

Exception – If the timeout occurs while waiting for the Device to transition to the given ObsState.

Return type

None

wait_for_subarray_assign_resources(assign_resources_value, timeout=3)

Wait for assign resources value of Subarray

Parameters

  • assign_resources_value (dict) – expected assign resource value

  • timeout (int) – timeout in sec for waiting to change assign resource

Returns

True if assign resource value is same as expected value else False

Return type

bool

wait_for_subarray_healthstate(healthstate, timeout=60)

Wait for Subarray Node healthState to be the given Healthstate till timeout occurs.

Raises Exception in case of a timeout.

Parameters

  • healthstate (HealthState) – HealthState

  • timeout (int) – timeout in sec for waiting to change healthstate

Raises

Exception – If the timeout occurs while waiting for the Subarray Node to transition to the given HealthState.

Return type

None

wait_for_subarray_obsstate(obsstate, timeout=100)

Wait for Subarray Node ObsState to be the given obsstate till timeout occurs.

Raises Exception in case of a timeout.

Parameters

  • obsstate (ObsState) – ObsState

  • timeout (int) – timeout in sec for waiting to change obsstate

Raises

Exception – If the timeout occurs while waiting for the Subarray Node to transition to the given ObsState.

Return type

None

tests.test_helpers.common_utils.retry_invocation(function)

A decorator to retry command invocation if it fails due to unavailability.

Parameters

function (Callable) – The function to be decorated.

Return type

Callable

Returns

The decorated function.

constants module

Path: tests/test_helpers/constants.py

Define all constants required for testing

event_recorder module

Path: tests/test_helpers/event_recorder.py

Implement Event checker class which can be used to validate events

exception tests.test_helpers.event_recorder.AttributeNotSubscribed(message)

Bases: Exception

Raise this exception when attribute is not subscribed

class tests.test_helpers.event_recorder.EventRecorder

Bases: object

Implement method required for validating events

clear_events()

Clear Subscribed Events

Return type

None

has_change_event_occurred(device, attribute_name, attribute_value, lookahead=9)

Validate Change Event occurred for provided attribute This method check attribute value changed within number of lookahead events

Parameters

  • device (Any) – Tango Device Proxy Object.

  • attribute_name (str) – Name of the attribute.

  • attribute_value (Any) – Value of attribute.

  • lookahead (int) – Number of events to look ahead.

Returns

Change Event occurred True or False.

Return type

bool

Raises

AttributeNotSubscribed – If attribute is not subscribed.

subscribe_event(device, attribute_name, timeout=50.0)

Subscribe for change event for given attribute

Parameters

  • device (Any) – Tango Device Proxy Object

  • attribute_name (str) – Name of the attribute

  • timeout (float) – number of seconds to wait for the callable to be called

Module contents