Select Git revision
test_observation_client.py

L2SS-2154: Use LOTUS library for common metrics code
Jan David Mol authored
Code owners
Assign users and groups as approvers for specific file changes. Learn more.
test_observation_client.py 3.20 KiB
# Copyright (C) 2022 ASTRON (Netherlands Institute for Radio Astronomy)
# SPDX-License-Identifier: Apache-2.0
from json import loads
from os import environ
from integration_tests import base
from integration_tests.device_proxy import TestDeviceProxy
from tangostationcontrol.test.dummy_observation_settings import (
get_observation_settings_hba_core_immediate,
)
import grpc
from lofar_sid.interface.stationcontrol.observation_pb2 import (
StartObservationRequest,
StopObservationRequest,
)
from lofar_sid.interface.stationcontrol.observation_pb2_grpc import ObservationStub
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 = get_observation_settings_hba_core_immediate().to_json()
specification_dict = loads(specification)
# create an observation class using the dict and as host just get it using a
# util function
tango_host = environ["TANGO_HOST"]
obs = TestDeviceProxy(f"tango://{tango_host}/STAT/ObservationControl/1")
obs_id = specification_dict["antenna_fields"][0]["observation_id"]
with grpc.insecure_channel("rpc.service.consul:50051") as channel:
stub = ObservationStub(channel)
stub.StartObservation(StartObservationRequest(configuration=specification))
# Assert the observation is running after starting it
self.assertTrue(obs.is_observation_running(obs_id))
# Assert the observation has stopped after aborting
with grpc.insecure_channel("rpc.service.consul:50051") as channel:
stub = ObservationStub(channel)
stub.StopObservation(StopObservationRequest(observation_id=obs_id))
self.assertFalse(obs.is_observation_running(obs_id))