Test API package

conftest module

Path: tests/

conftest module for Subarray Node.

tests.conftest.aggregation_process_low()

Create aggregation process low object

tests.conftest.aggregation_process_mid()

Create aggregation process mid object

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.component_manager_low()

Create a component manager instance for LOW.

tests.conftest.component_manager_mid()

Create a component manager instance for MID.

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

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

  • fqdn – The fqdn of the device.

Return type

None

tests.conftest.devices_to_load()

This method returns the devices whose context are required

tests.conftest.event_recorder()

Records events and allows assertions

tests.conftest.get_input_str(path)

Returns input json string :rtype: String

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

Json factory for getting json files

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. :type parser: :param parser: the command line options parser :type parser: argparse.ArgumentParser

tests.conftest.pytest_sessionstart(session)

Pytest hook; prints info about tango version. :type session: :param session: a pytest Session object :type session: pytest.Session

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

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

tests.conftest.subarray_node(event_recorder, request)

Returns the instance of SubarrayNode class

tests.conftest.tango_context(devices_to_load, request)

This Fixture provide the tango context fixture

tests.conftest.task_callback()

Creates a mock callable for asynchronous testing

Return type

MockCallable

common_utils module

Path: tests/test_helpers/

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.

assert_command_completion_on_device(device_name, expected_state)

Check for device obsState to transition to the correct state.

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.

assert_health_state(healthState)

Asserts the SubarrayNode HealthState to be the HealthState specified.

Raises AssertionError in case of failure.

assert_obs_state(obsstate)

Asserts the SubarrayNode ObsState to be the ObsState specified.

Raises AssertionError in case of failure.

clear_command_call_info()

Clears the command call info attribute.

get_device_obs_states()

Returns a dictionary with current device ObsStates

Return type

dict

invoke_abort_command(event_recorder)

Invoke Abort command on Subarray Node

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 – Addtional 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.

Return type

str

invoke_command_on_device(device_name, command, input_data=None)

Invokes the given command on the device with provided input.

perform_action(device_names, actions)

Performs the said action on the given device

reset_defects_for_devices(device_names)

Resets the defect parameters for given devices

set_devices_availability(device_names, availability=True)

Set the device availability to the given value.

set_devices_availablity(device_availability)

Sets the availability of devices with provided parameters. :param device_availability: Dictionary with device names as keys and their availability as values. :dtype: dict

set_devices_defective(device_names, defective_params)

Sets the given devices defective with provided parameters.

set_subarray_health_state_to_ok(event_recorder)

Set Subarray Health state to OK

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.

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

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.

wait_for_subarray_assign_resources(assign_resources_value, timeout=3)

Wait for assign resources value of Subarray :type assign_resources_value: dict :param assign_resources_value: expected assign resource value :type timeout: int :param timeout: timeout in sec for waiting to change assign resource

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.

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.

tests.test_helpers.common_utils.retry_invocation(function)

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

Return type

Callable

Module contents