Testing

Unit Tests

Since the component manager handle the interactions with the DS Simulator, which implements an OPCUA server, we are able to check the robustness of our component manager and the business rules captured in our model without spinning up any tango infrastructure.

These unit tests are captured in the python-test job. Additionally, the device server interface is tested (using a DeviceTestContext) without having to set up client connections to the components (DS Simulator). The necessary triggers on the components needed to effect a transition on DishManager are manipulated from weak references to the component managers.

Acceptance Tests

Acceptance tests are run using a simulated implementation of the OPCUA server and the DS Manager. This simulator maintains an interface as close enough to the server to be deployed by CETC (mainly tracked in an xml interface description file).

Testing Locally without Kubernetes

DishManager is packaged as a helm chart to be deployed in a kubernetes cluster. Beyond verifying changes based on pipeline outputs from python-test and k8s-test jobs, it’s beneficial (in some cases) to be able to deploy the devices locally without needing to spin up a kubernetes cluster to quickly verify changes. This is not meant to rival our deployment process in the project but rather, provide alternatives for the developer to verify their changes locally before pushing them upstream.

Deploy DSManager with no DB

This can be run in an isolated environment (virtual environment, docker container, …)

ds manager (no opcua server)ds manager (with opcua server)

$ python DSManager.py SKA001 -v4 -nodb -host 127.0.0.1 -port 12345 -dlist mid-dish/ds-manager/SKA001

Tip

Device server can be deployed directly from docker image as:

$ docker run -p 45450:45450 -it <image-name:tag>  /usr/bin/python3 /app/src/ska_mid_dish_ds_manager/devices/DSManagerDS.py SKA001 -v4 -nodb -host 127.0.0.1 -port 12345 -dlist mid-dish/ds-manager/SKA001

Testing Tracking with Append

The script below illustrates appending to the track table while actively tracking. Only two points in each table are used for simplification. Run itango in the root folder of the repository to reach dependencies and paste the script below for testing this behaviour.

import time
from src.ska_mid_dish_ds_manager.models.dish_enums import TrackTableLoadMode
from math import isclose

# get tai offset of current time with SKA epoch
def get_current_tai_timestamp():
   return time.time() - 946684763.0

ds_manager_proxy = DeviceProxy("mid-dish/ds-manager/SKA001")
ds_manager_proxy.SetStandbyFPMode()

# subscribe to azimuth and elevation attributes to view variables
ds_manager_proxy.subscribe_event(
   "achievedPointing",
   tango.EventType.CHANGE_EVENT,
   tango.utils.EventCallback(),
)

# get current position
_, az_init, el_init = ds_manager_proxy.achievedPointing

# create an initial table to load with New
start_delay = 10
tai_start_time = get_current_tai_timestamp() + start_delay
start_time_first_table = tai_start_time
duration_first_table = 10
end_time_first_table = start_time_first_table + duration_first_table
track_table = [
   TrackTableLoadMode.NEW,
   2,
   start_time_first_table, az_init + 1, el_init + 1,
   end_time_first_table, az_init + 2, el_init + 2]
ds_manager_proxy.TrackLoadTable(track_table)
# start tracking with new table
ds_manager_proxy.Track()
# wait for 2 seconds into tracking
time.sleep(start_time_first_table - get_current_tai_timestamp() + 2)

# create table that can be appended
start_time_second_table = tai_start_time + duration_first_table + 1
duration_second_table = 5
end_time_second_table = start_time_second_table + duration_second_table
track_table_additional = [
   TrackTableLoadMode.APPEND,
   2,
   start_time_second_table, az_init + 2, el_init + 2,
   end_time_second_table, az_init + 1, el_init + 1,
]
ds_manager_proxy.TrackLoadTable(track_table_additional)
time.sleep(end_time_second_table - get_current_tai_timestamp() + 2)

# check that the achieved points match the last point of the appended table
assert isclose((ds_manager_proxy.achievedPointing)[1], track_table_additional[-2])
assert isclose((ds_manager_proxy.achievedPointing)[2], track_table_additional[-1])