test_observation_client.py

# Copyright (C) 2022 ASTRON (Netherlands Institute for Radio Astronomy)
# SPDX-License-Identifier: Apache-2.0

from json import loads
from os import environ

from lofar_station_client.observation.station_observation import (
    StationObservation,
)
from integration_test import base
from integration_test.device_proxy import TestDeviceProxy
from tangostationcontrol.test.dummy_observation_settings import (
    get_observation_settings_hba_immediate,
)

from tango import DevState


class TestObservation(base.IntegrationTestCase):
    def setUp(self):
        self.setup_stationmanager_proxy()
        self.observation_control_proxy = TestDeviceProxy("STAT/ObservationControl/1")
        self.observation_control_proxy.off()
        self.observation_control_proxy.boot()
        self.addCleanup(
            self.observation_control_proxy.test_device_turn_off,
            self.observation_control_proxy,
        )

        # make sure any devices we depend on are also started
        for device in [
            "STAT/RECVH/H0",
            "STAT/SDPFirmware/HBA0",
            "STAT/SDP/HBA0",
            "STAT/Beamlet/HBA0",
            "STAT/DigitalBeam/HBA0",
            "STAT/TileBeam/HBA0",
            "STAT/AFH/HBA0",
            "STAT/SST/HBA0",
            "STAT/XST/HBA0",
        ]:
            proxy = TestDeviceProxy(device)
            proxy.off()
            proxy.boot()
            self.addCleanup(proxy.test_device_turn_off, proxy)

    def setup_stationmanager_proxy(self):
        """Setup StationManager"""
        stationmanager_proxy = TestDeviceProxy("STAT/StationManager/1")
        stationmanager_proxy.off()
        stationmanager_proxy.boot()
        self.addCleanup(stationmanager_proxy.test_device_turn_off, stationmanager_proxy)
        self.assertEqual(stationmanager_proxy.state(), DevState.ON)
        return stationmanager_proxy

    def test_observation(self):
        """Test of the observation_wrapper class basic functionality"""

        # convert the JSON specification to a dict for this class
        specification_dict = loads(get_observation_settings_hba_immediate().to_json())

        # create an observation class using the dict and as host just get it using a
        # util function
        observation = StationObservation(
            specification=specification_dict, host=environ["TANGO_HOST"]
        )

        # Assert the observation is running after starting it
        observation.start()
        self.assertTrue(observation.is_running)

        # Assert the proxy is on
        station = observation.observation
        proxy = station.observation_field_proxies[0]
        self.assertTrue(proxy.state() == DevState.ON)

        # Assert the observation has stopped after aborting
        observation.stop()
        self.assertFalse(observation.is_running)