do a sub class of unittest.TestCase for testing

Also add .nxdtest file that will be used during tests.

.nxdtest

+TestCase('test UE connect', ['python', '-m', 'openradio_ws_ue'])

main_test.py

+#!/usr/bin/env python3
+
+import argparse
+import openradio_ws_ue
+
+from websocket import create_connection
+
+parser = argparse.ArgumentParser(description='Connect to a lteue websocket to perform some testing.')
+parser.add_argument('url', help='url of the websocket in the form ws://[XXX]:XXX (eg "ws://[2a11:9ac0:d::1]:9002")')
+args = parser.parse_args()
+
+ue_test = openradio_ws_ue.UeTest()
+ue_test.ws_url = args.url
+ue_test.ws = create_connection(args.url)
+ue_test.test_ue_has_ip()
+

openradio_ws_ue.py

+import argparse
+import json
+import time
+import unittest
+
+from websocket import create_connection
+
+class UeError(Exception):
+    pass
+
+class UeTest(unittest.TestCase):
+    @classmethod
+    def setUpClass(self):
+        #self.ws_url = "ws://[2a11:9ac0:d::1]:9002"
+        self.ws_url = "ws://[2a11:9ac0:d::df82]:9002"
+        #TODO parse the instances.json file
+        # there should be a instances.json file containing a json dumped python dictionary like this:
+        # {
+        #   "instance1 name":
+        #   {
+        #     "url": SR_URL,
+        #     "software_type": instance_type,
+        #     "connection_parameters": instance1_connection_parameters_dict
+        #   },
+        #   ...
+        # }
+        self.ws = create_connection(self.ws_url)
+
+    @classmethod
+    def tearDownClass(self):
+        self.ws.close()
+
+    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()
+        self.assertEqual(result['message'], 'ue_get', "Unexpected response")
+        self.assertIn('ue_list', result, "No UE")
+        self.assertTrue(result['ue_list'], "No UE")
+        return result['ue_list'][0]
+
+    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()
+
+    def test_ue_has_ip(self):
+        result = self.recv()
+        self.assertEqual(result['message'], 'ready', "lteue not ready")
+        result = self.ue_get()
+        ue_id = result['ue_id']
+
+        try:
+            self.power_on(ue_id)
+            time.sleep(2)
+
+            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")
+            print(result['pdn_list'][0]['ipv4'])
+        finally:
+            self.power_off(ue_id)
+
+
+if __name__ == '__main__':
+    unittest.main()

ws.py

-import argparse
-import json
-import time
-
-from websocket import create_connection
-
-parser = argparse.ArgumentParser(description='Connect to a lteue websocket to perform some testing.')
-parser.add_argument('url', help='url of the websocket in the form ws://[XXX]:XXX (eg "ws://[2a11:9ac0:d::1]:9002")')
-args = parser.parse_args()
-
-class UeError(Exception):
-    pass
-
-def send(ws, msg):
-    ws.send(json.dumps(msg))
-def recv(ws):
-    return json.loads(ws.recv())
-
-def ue_get(ws):
-    send(ws, {
-        "message": "ue_get",
-    })
-    result = recv(ws)
-    if result['message'] != 'ue_get':
-        raise UeError("Unexpected response")
-    if not result['ue_list']:
-        raise UeError("No UE")
-    return result['ue_list'][0]
-
-def power_on(ws, ue_id):
-    if ue_get(ws)['power_on']:
-        raise UeError("UE already powered on")
-    send(ws, {
-        "message": "power_on",
-        "ue_id": ue_id,
-    })
-    recv(ws)
-
-def power_off(ws, ue_id):
-    if not ue_get(ws)['power_on']:
-        raise UeError("UE already powered off")
-    send(ws, {
-        "message": "power_off",
-        "ue_id": ue_id,
-    })
-    recv(ws)
-
-try:
-
-    try:
-        ws = create_connection(args.url)
-    except ConnectionRefusedError:
-        raise UeError("Couldn't connect to remote API")
-
-    try:
-        result = recv(ws)
-        if result['message'] != 'ready':
-            raise UeError("lteue not ready")
-
-        result = ue_get(ws)
-        ue_id = result['ue_id']
-
-        try:
-            power_on(ws, ue_id)
-            time.sleep(2)
-
-            result = ue_get(ws)
-            if 'pdn_list' not in result:
-                raise UeError("UE didn't connect")
-            print(result['pdn_list'][0]['ipv4'])
-        finally:
-            power_off(ws, ue_id)
-
-    finally:
-        ws.close()
-
-except UeError as e:
-    print(e)