L2SS-780: fix test env path

tangostationcontrol/tangostationcontrol/integration_test/default/prometheus/test_tango_prometheus_client.py

@@ -13,18 +13,31 @@ from tangostationcontrol.integration_test.device_proxy import TestDeviceProxy
 from tango import Database
 import importlib
 import sys
-sys.path.append('/opt/lofar/tango/docker-compose/tango-prometheus-exporter/code')
-tpc = importlib.import_module('tango-prometheus-client')
+#sys.path.append('/opt/lofar/tango/docker-compose/tango-prometheus-exporter/code')
+#tpc = importlib.import_module('tango-prometheus-client')
+
+module_name = 'ArchiverPolicy'
+file_path = '/opt/lofar/tango/docker-compose/tango-prometheus-exporter/code/tango-prometheus-client.py'
+spec = importlib.util.spec_from_file_location(module_name, file_path)
+tpc_policy = importlib.util.module_from_spec(spec)
+sys.modules[module_name] = tpc_policy
+spec.loader.exec_module(tpc_policy)
+
+module_name = 'CustomCollector'
+spec = importlib.util.spec_from_file_location(module_name, file_path)
+tpc_cc = importlib.util.module_from_spec(spec)
+sys.modules[module_name] = tpc_cc
+spec.loader.exec_module(tpc_cc)
 
 class TestPrometheusClient(BaseIntegrationTestCase):
 
-    CONFIG = tpc.ArchiverPolicy.load_config('/opt/lofar/tango/docker-compose/tango-prometheus-exporter/lofar2-policy.json')
+    CONFIG = tpc_policy.ArchiverPolicy.load_config('/opt/lofar/tango/docker-compose/tango-prometheus-exporter/lofar2-policy.json')
     
     def setUp(self):
         super().setUp()
         db = Database()
         station = db.get_property("station","name")["name"][0]
-        custom_collector = tpc.CustomCollector(self.CONFIG, station)
+        custom_collector = tpc_cc.CustomCollector(self.CONFIG, station)
         self.assertIsNotNone(custom_collector)
         return custom_collector
     
@@ -36,12 +49,29 @@ class TestPrometheusClient(BaseIntegrationTestCase):
         recv_proxy.set_defaults()
         return recv_proxy
     
+    def test_tango_db_devices(self):
+        """ Test if device names are correctly retrieved from Tango DB """
+        policy = tpc_policy.ArchiverPolicy(self.CONFIG)
+        db_devices = policy.device_list()
+        self.assertNotEqual(len(db_devices), 0)
+    
+    def test_policy_devices(self):
+        """ Test if device names are correctly filtered with policy file """
+        policy = tpc_policy.ArchiverPolicy(self.CONFIG)
+        db_devices = policy.device_list()
+        policy_devices = policy.devices()
+        self.assertLessEqual(len(policy_devices), len(db_devices))
+        config_retrieved_devices = [*policy.config['devices'].keys()]   # list of device names from policy file
+        for d in config_retrieved_devices:
+            if '*' not in d:                                            # filter out wildcards
+                self.assertIn(d, policy_devices)
+    
     def test_archiver_policy_attribute_list(self):
         """ Test if the full set of archiving policy for the given device is retrieved """
         device_name = 'stat/recv/1'
         recv_proxy = self.setup_recv_proxy(device_name)
-        policy = tpc.ArchiverPolicy(self.CONFIG)
-        attribute_list = policy.policy.attribute_list(device_name, recv_proxy.get_attribute_list())
+        policy = tpc_policy.ArchiverPolicy(self.CONFIG)
+        attribute_list = policy.attribute_list(device_name, recv_proxy.get_attribute_list())
         include = policy.config['devices']['stat/recv/1']['include'] # attribute that must be included
         for i in include:
             if '*' not in i:    # exclude wildcard

tangostationcontrol/tangostationcontrol/test/prometheus/test_archiver_policy.py

@@ -8,10 +8,17 @@
 # See LICENSE.txt for more info.
 
 from tangostationcontrol.test import base
-import importlib
+#import importlib
 import sys
-sys.path.append('/opt/lofar/tango/docker-compose/tango-prometheus-exporter/code')
-tpc = importlib.import_module('tango-prometheus-client')
+#sys.path.append('/opt/lofar/tango/docker-compose/tango-prometheus-exporter/code')
+#tpc = importlib.import_module('tango-prometheus-client')
+import importlib.util
+module_name = 'ArchiverPolicy'
+file_path = '/opt/lofar/tango/docker-compose/tango-prometheus-exporter/code/tango-prometheus-client.py'
+spec = importlib.util.spec_from_file_location(module_name, file_path)
+tpc = importlib.util.module_from_spec(spec)
+sys.modules[module_name] = tpc
+spec.loader.exec_module(tpc)
 
 class TestArchiverPolicy(base.TestCase):
 
@@ -25,20 +32,3 @@ class TestArchiverPolicy(base.TestCase):
         policy = tpc.ArchiverPolicy(self.CONFIG)
         self.assertEqual([*policy.config], ['default', 'devices'])  # dict keys
 
-    def test_tango_db_devices(self):
-        """ Test if device names are correctly retrieved from Tango DB """
-        policy = tpc.ArchiverPolicy(self.CONFIG)
-        db_devices = policy.device_list()
-        self.assertNotEqual(len(db_devices), 0)
-    
-    def test_policy_devices(self):
-        """ Test if device names are correctly filtered with policy file """
-        policy = tpc.ArchiverPolicy(self.CONFIG)
-        db_devices = policy.device_list()
-        policy_devices = policy.devices()
-        self.assertLessEqual(len(policy_devices), len(db_devices))
-        config_retrieved_devices = [*policy.config['devices'].keys()]   # list of device names from policy file
-        for d in config_retrieved_devices:
-            if '*' not in d:                                            # filter out wildcards
-                self.assertIn(d, policy_devices)
-