Select Git revision
test_device_configuration.py

Hannes Feldt authored
Code owners
Assign users and groups as approvers for specific file changes. Learn more.
test_device_configuration.py 12.02 KiB
# Copyright (C) 2022 ASTRON (Netherlands Institute for Radio Astronomy)
# SPDX-License-Identifier: Apache-2.0
import json
from typing import Callable
import logging
try:
from importlib.resources import files
except ImportError:
from importlib_resources import files # type: ignore
from integration_test.default.devices.base import TestDeviceBase
from lofar_lotus.dict import CaseInsensitiveDict
from lofar_lotus.dict import CaseInsensitiveString
from tango import DevState
logger = logging.getLogger()
INITIAL_CONFIGURATION = None
class TestDeviceConfiguration(TestDeviceBase):
"""Integration test class for device Configuration"""
__test__ = True
TEST_CONFIGURATION = """{
"servers": {
"AFH": {
"STAT": {
"AFH": {
"STAT/AFH/HBA": {
"properties": {
"Control_Children": [ "STAT/MOCKRCU/1" ]
}
}
}
}
},
"ObservationControl": {
"STAT": {
"ObservationControl": {
"STAT/ObservationControl/11": {}
}
}
}
}
}"""
DB_DEFAULT_CONFIG_FILE = "LOFAR_ConfigDb.json"
DB_FILE_LIST = [
"test_environment_ConfigDb.json",
"simulators_ConfigDb.json",
"dummy_positions_ConfigDb.json",
]
CONTROL_CHILD_PROPERTY = CaseInsensitiveString("Control_Children")
def setUp(self):
super().setUp("STAT/Configuration/1")
# use a bigger timeout as even test configurations are getting big
self.proxy.set_timeout_millis(60000)
global INITIAL_CONFIGURATION
if not INITIAL_CONFIGURATION:
if self.proxy.state() == DevState.OFF:
self.proxy.boot()
INITIAL_CONFIGURATION = self.proxy.station_configuration_RW
self.proxy.off()
# Ensure that each test begins with the default station configuration
self.upload_test_lofar_configuration()
self.addCleanup(self.restore_initial_configuration)
@staticmethod
def _casefold_dbdata(dbdata):
return CaseInsensitiveDict(dbdata)
def restore_initial_configuration(self):
"""Restore initial devices configuration"""
global INITIAL_CONFIGURATION
if self.proxy.state() == DevState.OFF:
self.proxy.boot()
self.proxy.station_configuration_RW = INITIAL_CONFIGURATION
self.proxy.off()
def upload_test_lofar_configuration(self):
"""Upload station configuration with original default values"""
if self.proxy.state() == DevState.OFF:
self.proxy.boot()
config_file_path = files(__package__).joinpath(
f"configDB/{self.DB_DEFAULT_CONFIG_FILE}"
)
with config_file_path.open() as _file:
self.proxy.station_configuration_RW = json.dumps(json.load(_file))
for json_file in self.DB_FILE_LIST:
json_file_path = files(__package__).joinpath(f"configDB/{json_file}")
with json_file_path.open() as _file:
self.proxy.update_station_configuration(json.dumps(json.load(_file)))
self.proxy.off()
def configuration_base(self, _fn: Callable[[dict], None]):
"""Base method shared across read configuration tests"""
self.assertEqual(DevState.OFF, self.proxy.state())
self.proxy.boot()
self.assertEqual(DevState.ON, self.proxy.state())
station_configuration = self.proxy.station_configuration_RW
dbdata = json.loads(station_configuration)
_fn(dbdata)
def test_read_station_configuration(self):
"""Test whether the station control configuration is correctly retrieved as a JSON string"""
def read_station_configuration_test(dbdata: dict):
self.assertEqual(type(dbdata), dict)
self.assertGreater(len(dbdata["servers"]), 0)
# Verify if Configuration Device exists
self.assertTrue(
"configuration" in dbdata["servers"], msg=f"{dbdata}"
) # server-name
self.assertTrue(
"stat" in dbdata["servers"]["configuration"], msg=f"{dbdata}"
) # server-instance
self.assertTrue(
"configuration" in dbdata["servers"]["configuration"]["stat"],
msg=f"{dbdata}",
) # server-class
self.assertTrue(
"stat/configuration/1"
in dbdata["servers"]["configuration"]["stat"]["configuration"],
msg=f"{dbdata}",
) # device
self.configuration_base(read_station_configuration_test)
def test_load_station_configuration(self):
"""Test whether the station control configuration is correctly loaded into Tango Database"""
def load_station_configuration_test(dbdata: dict):
# So far configuration Database is set from files in the DB_FILE_LIST
self.assertTrue(
"stat/configuration/1"
in dbdata["servers"]["configuration"]["stat"]["configuration"],
msg=f"{dbdata}",
) # configuration device
self.assertFalse(
"stat/observationcontrol/11"
in dbdata["servers"]["observationcontrol"]["stat"][
"observationcontrol"
],
msg=f"{dbdata}",
) # observation device
self.assertTrue(
"stat/afh/hba" in dbdata["servers"]["afh"]["stat"]["afh"],
msg=f"{dbdata}",
) # antennafield device
antennafield_properties = CaseInsensitiveDict(
dbdata["servers"]["afh"]["stat"]["afh"]["stat/afh/hba"]["properties"]
)
self.assertIn(
self.CONTROL_CHILD_PROPERTY,
antennafield_properties,
msg=f"{antennafield_properties}",
)
self.assertEqual(
antennafield_properties[self.CONTROL_CHILD_PROPERTY][0], "STAT/RECVH/1"
)
# Load a full new configuration, and consequently erase the previous one
self.proxy.station_configuration_RW = self.TEST_CONFIGURATION
updated_dbdata = json.loads(self.proxy.station_configuration_RW)
# Test whether default 'protected' devices have not been deleted
self.assertTrue(
"stat/configuration/1"
in updated_dbdata["servers"]["configuration"]["stat"]["configuration"],
msg=f"{updated_dbdata}",
)
# Test whether new device has been added
self.assertTrue(
"stat/observationcontrol/11"
in updated_dbdata["servers"]["observationcontrol"]["stat"][
"observationcontrol"
],
msg=f"{updated_dbdata}",
) # observation device
# Test whether old device has been deleted
self.assertFalse(
"recvh" in updated_dbdata["servers"].keys(), msg=f"{updated_dbdata}"
) # recvh device
# Test whether old attribute has been updated
self.assertTrue(
"stat/afh/hba" in updated_dbdata["servers"]["afh"]["stat"]["afh"],
msg=f"{updated_dbdata}",
)
antennafield_properties = CaseInsensitiveDict(
updated_dbdata["servers"]["afh"]["stat"]["afh"]["stat/afh/hba"][
"properties"
]
)
self.assertIn(
self.CONTROL_CHILD_PROPERTY,
antennafield_properties,
msg=f"{antennafield_properties}",
)
self.assertEqual(
antennafield_properties[self.CONTROL_CHILD_PROPERTY][0],
"STAT/MOCKRCU/1",
)
self.configuration_base(load_station_configuration_test)
def test_update_station_configuration(self):
"""Test whether the station control configuration
is correctly updated into Tango Database"""
self.assertEqual(DevState.OFF, self.proxy.state())
self.proxy.boot()
self.assertEqual(DevState.ON, self.proxy.state())
# So far configuration Database is set from files in the DB_FILE_LIST
# Update the station configuration with a new JSON file
self.proxy.update_station_configuration(self.TEST_CONFIGURATION)
updated_dbdata = CaseInsensitiveDict(
json.loads(self.proxy.station_configuration_RW)
)
# Test whether default 'protected' devices have not been deleted
self.assertTrue(
"stat/configuration/1"
in updated_dbdata["servers"]["configuration"]["stat"]["configuration"],
msg=f"{updated_dbdata}",
)
# Test whether new device has been added
self.assertTrue(
"stat/observationcontrol/11"
in updated_dbdata["servers"]["observationcontrol"]["stat"][
"observationcontrol"
],
msg=f"{updated_dbdata}",
) # observation device
# Test whether old device has NOT been deleted
self.assertTrue(
"stat/recvh/h0" in updated_dbdata["servers"]["recvh"]["stat"]["recvh"],
msg=f"{updated_dbdata}",
) # recvh device
# Test whether old attribute has been updated
self.assertTrue(
"stat/afh/hba" in updated_dbdata["servers"]["afh"]["stat"]["afh"],
msg=f"{updated_dbdata}",
)
antennafield_properties = CaseInsensitiveDict(
updated_dbdata["servers"]["afh"]["stat"]["afh"]["stat/afh/hba"][
"properties"
]
)
self.assertIn(
self.CONTROL_CHILD_PROPERTY,
antennafield_properties,
msg=f"{antennafield_properties}",
)
self.assertEqual(
antennafield_properties[self.CONTROL_CHILD_PROPERTY][0], "STAT/MOCKRCU/1"
)
def test_backup_station_configuration(self):
"""Test whether the station control configuration is correctly saved
in the correspondent attribute as a backup"""
self.proxy.boot()
default_dbdata = json.loads(self.proxy.station_configuration_RW)
# Load a full new configuration, and consequently erase the previous one
self.proxy.station_configuration_RW = self.TEST_CONFIGURATION
new_dbdata = json.loads(self.proxy.station_configuration_RW)
# Check if device has saved the previous configuration as a backup copy
backup_dbdata = json.loads(self.proxy.backup_station_configuration_R)
self.assertNotEqual(new_dbdata, backup_dbdata)
self.assertEqual(default_dbdata, backup_dbdata)
def test_restore_station_configuration(self):
"""Test whether the backup station control configuration
can be correctly restored after a loading/update operation"""
self.proxy.boot()
default_dbdata = json.loads(self.proxy.station_configuration_RW)
backup_dbdata = json.loads(self.proxy.backup_station_configuration_R)
# Load a full new configuration, and consequently erase the previous one
self.proxy.station_configuration_RW = self.TEST_CONFIGURATION
new_dbdata = json.loads(self.proxy.station_configuration_RW)
self.assertNotEqual(new_dbdata, backup_dbdata)
# Restore the previous configuration
self.proxy.restore_station_configuration()
self.assertEqual(
default_dbdata, json.loads(self.proxy.station_configuration_RW)
)