Simple websocket test, no instance creation, works

software/end-to-end-testing/tests/test_ors2.py

+import json
+import hashlib
+import hmac
+import time
+import slapos.testing.e2e as e2e
+from websocket import create_connection
+
+class WebsocketTestClass(e2e.EndToEndTestCase):
+    @classmethod
+    def setUpClass(cls):
+        try:
+            super().setUpClass()
+
+            create_instances = False
+            if create_instances:
+                cls.enb_instance_name = time.strftime('e2e-ors70-enb-%Y-%B-%d-%H:%M:%S')
+                cls.product = 'https://lab.nexedi.com/nexedi/slapos/-/raw/1.0.371/software/simpleran/software-ors.cfg'
+                # Component GUIDs and configurations
+                cls.comp_enb = "COMP-4296"
+                # Retry configurations
+                cls.max_retries = 10
+                cls.retry_delay = 180  # seconds
+                # Setup instances
+                cls.setup_instances()
+                cls.waitUntilGreen(cls.enb_instance_name)
+                cls.waitUntilGreen(cls.cn_instance_name)
+            else:
+                cls.enb_instance_name = 'e2e-ors70-enb-1733404208'
+                cls.ue_instance_name = 'e2e-sb005-ue-1733403458'
+
+                cls.waitUntilGreen(cls.enb_instance_name)
+                cls.waitUntilGreen(cls.ue_instance_name)
+                cls.setup_websocket_connection()
+
+        except Exception as e:
+            cls.logger.error("Error during setup: " + str(e))
+            # Ensure cleanup
+            cls.tearDownClass()
+            raise
+
+    @classmethod
+    def setup_websocket_connection(cls):
+        connection_params = cls.getInstanceInfos(cls.ue_instance_name).connection_dict
+        cls.waitUntilGreen(cls.ue_instance_name)
+        cls.ws_host = connection_params.get('websocket-hostname')
+        cls.ws_port = connection_params.get('websocket-port')
+        cls.ws_pass = connection_params.get('websocket-password')
+        cls.ws_url = f"wss://{cls.ws_host}/websocket:{cls.ws_port}"
+
+        cls.logger.info(f"Websocket URL: {cls.ws_url}")
+
+        cls.ws = create_connection(cls.ws_url)
+        cls.logger.info("Websocket connection established.")
+        data = json.loads(cls.ws.recv())
+        res = hmac.new(
+          "{}:{}:{}".format(data['type'], cls.ws_pass, data['name']).encode(),
+          msg=data['challenge'].encode(),
+          digestmod=hashlib.sha256
+        ).hexdigest()
+        msg = {"message": "authenticate", "res": res}
+        cls.ws.send(json.dumps(msg))
+        cls.ws.recv()
+        cls.logger.info("Websocket authentication established.")
+
+    @classmethod
+    def retry_request(cls, func, *args, **kwargs):
+        for attempt in range(cls.max_retries):
+            try:
+                result = func(*args, **kwargs)
+                if result:
+                    return result
+            except Exception as e:
+                cls.logger.error(f"Error on attempt {attempt + 1}: {e}")
+            if attempt < cls.max_retries - 1:
+                time.sleep(cls.retry_delay)
+        return None
+
+    @classmethod
+    def setup_instances(cls):
+        cls.request_enb()
+
+    @classmethod
+    def request_enb(cls, custom_params=None):
+        cls.logger.info("Request "+ cls.enb_instance_name)
+        enb_parameters = {}
+        if custom_params:
+            enb_parameters.update(custom_params)
+
+        json_enb_parameters = json.dumps(enb_parameters)
+
+        cls.retry_request(cls.request, cls.product, cls.enb_instance_name,
+                          filter_kw={"computer_guid": cls.comp_enb},
+                          partition_parameter_kw={'_': json_enb_parameters},
+                          software_type='enb')
+
+    @classmethod
+    def tearDownClass(cls):
+        if hasattr(cls, 'ws') and cls.ws is not None:
+            cls.logger.info("Closing websocket")
+            cls.ws.close()
+        super().tearDownClass()
+
+    def send(self, msg):
+        self.ws.send(json.dumps(msg))
+    def recv(self):
+        return json.loads(self.ws.recv())
+
+    def ue_get(self):
+        self.send({"message": "ue_get"})
+        result = self.recv()
+
+        if 'message' not in result:
+            raise ValueError(f"Unexpected response format: {result}")
+
+        if 'ue_list' in result:
+            if not result['ue_list']:
+                raise ValueError(f"No UE found in response: {result}")
+            return result['ue_list'][0]
+        else:
+            return result
+
+    def power_on(self, ue_id):
+        self.assertFalse(self.ue_get()['power_on'], "UE already powered on")
+        self.send({"message": "power_on", "ue_id": ue_id})
+        self.recv()
+
+    def power_off(self, ue_id):
+        self.assertTrue(self.ue_get()['power_on'], "UE already powered off")
+        self.send({"message": "power_off", "ue_id": ue_id})
+        self.recv()
+
+class ORSTest(WebsocketTestClass):
+    def test_ue_has_ip(self):
+        result = self.ue_get()
+        ue_id = result['ue_id']
+
+        try:
+            self.power_on(ue_id)
+            time.sleep(5)
+            result = self.ue_get()
+            self.assertIn('pdn_list', result, "UE didn't connect")
+            self.assertIn('ipv4', result['pdn_list'][0], "UE didn't get IPv4")
+            self.logger.info("UE connected with ip: " + result['pdn_list'][0]['ipv4'])
+        finally:
+            self.power_off(ue_id)