Skip to content
Snippets Groups Projects
Select Git revision
  • 009460e35a5ed0159e1e2ab5a2767be03caca799
  • master default protected
  • revert-cs032-ccd-ip
  • deploy-components-parallel
  • fix-chrony-exporter
  • L2SS-2407-swap-iers-caltable-monitoring-port
  • L2SS-2357-fix-ruff
  • sync-up-with-meta-pypcc
  • stabilise-landing-page
  • all-stations-lofar2
  • v0.39.7-backports
  • Move-sdptr-to-v1.5.0
  • fix-build-ubuntu
  • tokens-in-env-files
  • fix-build
  • L2SS-2214-deploy-cdb
  • fix-missing-init
  • add-power-hardware-apply
  • L2SS-2129-Add-Subrack-Routine
  • Also-listen-internal-to-rpc
  • fix-build-dind
  • v0.55.5-r2 protected
  • v0.52.8-rc1 protected
  • v0.55.5 protected
  • v0.55.4 protected
  • 0.55.2.dev0
  • 0.55.1.dev0
  • 0.55.0.dev0
  • v0.54.0 protected
  • 0.53.2.dev0
  • 0.53.1.dev0
  • v0.52.3-r2 protected
  • remove-snmp-client
  • v0.52.3 protected
  • v0.52.3dev0 protected
  • 0.53.1dev0
  • v0.52.2-rc3 protected
  • v0.52.2-rc2 protected
  • v0.52.2-rc1 protected
  • v0.52.1.1 protected
  • v0.52.1 protected
41 results

test_device_configuration.py

Blame
  • 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)
            )