=======
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, ...)

.. tabs::

   .. tab:: ds manager (no opcua server)

      .. code-block:: rst

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

   .. tab:: ds manager (with opcua server)

      .. code-block:: rst

        `deploy DS Simulator OPCUA server in ska-mid-dish-simulators/src/ska_mid_dish_simulators/devices/`
        $ python ds_opcua_server.py &

        `deploy DSManager`
        `keep the simulator running while continuously re-running DSManager to test new changes`
        $ 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:

   .. code-block:: rst

     $ 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.

.. code-block:: python
   :linenos:

   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])
 

.. _DeviceTestContext: https://gitlab.com/tango-controls/pytango/-/blob/v9.5.0/tango/test_context.py?ref_type=tags#L740
.. _simulated implementation: https://gitlab.com/ska-telescope/ska-mid-dish-simulators