From 3c9dbe30534053ba894a49c3f52655c3c3fa7042 Mon Sep 17 00:00:00 2001 From: Mika Joenpera Date: Tue, 24 Oct 2023 08:35:31 +0300 Subject: [PATCH] Fix UnitTests and bug in the comms_settings - UnitTests now supports multiradio - bug found and fixed from comms_settings Jira-Id: SCDI-43 Signed-off-by: Mika Joenpera --- .../src/nats/src/comms_settings.py | 8 +- .../src/nats/src/validation.py | 27 ++-- .../src/nats/tests/test_settings.py | 137 +++++++++++++++++- .../src/nats/tests/test_validation.py | 81 +++++++++++ 4 files changed, 235 insertions(+), 18 deletions(-) diff --git a/modules/sc-mesh-secure-deployment/src/nats/src/comms_settings.py b/modules/sc-mesh-secure-deployment/src/nats/src/comms_settings.py index d89317418..c29ffcd23 100644 --- a/modules/sc-mesh-secure-deployment/src/nats/src/comms_settings.py +++ b/modules/sc-mesh-secure-deployment/src/nats/src/comms_settings.py @@ -146,6 +146,9 @@ def handle_mesh_settings(self, msg: str, path="/opt", """ Handle mesh settings """ + + _index = 0 + try: parameters_set = json.loads(msg) @@ -179,6 +182,7 @@ def handle_mesh_settings(self, msg: str, path="/opt", self.bridge = quote(str(parameters_set["bridge"])) for index in self.radio_index: + _index = index self.logger.debug("Mesh settings validation index: %s", str(index)) ret, info = self.validate_mesh_settings(index) @@ -193,9 +197,9 @@ def handle_mesh_settings(self, msg: str, path="/opt", except (json.decoder.JSONDecodeError, KeyError, TypeError, AttributeError) as error: - self.comms_status.mesh_cfg_status = \ + self.comms_status[_index].mesh_cfg_status = \ comms.STATUS.mesh_cfg_not_stored - ret, _ = "FAIL", self.comms_status.mesh_cfg_status + ret, _ = "FAIL", self.comms_status[_index].mesh_cfg_status info = "JSON format not correct" + str(error) self.logger.error("Mesh settings validation: %s, %s", ret, info) diff --git a/modules/sc-mesh-secure-deployment/src/nats/src/validation.py b/modules/sc-mesh-secure-deployment/src/nats/src/validation.py index 8a5084548..5a1bded71 100644 --- a/modules/sc-mesh-secure-deployment/src/nats/src/validation.py +++ b/modules/sc-mesh-secure-deployment/src/nats/src/validation.py @@ -220,9 +220,12 @@ def validate_role(role: str) -> bool: Validates a given role. Returns True if the role is valid, False otherwise. """ - if role in ("drone", "sleeve", "gcs"): - return True - return False + try: + if role in ("drone", "sleeve", "gcs"): + return True + return False + except (ValueError, TypeError, AttributeError): + return False def validate_delay(delay: str) -> bool: """ @@ -265,15 +268,21 @@ def validate_mesh_vif(mesh_vif: str) -> bool: Returns True if the mesh vif is valid, False otherwise. """ # todo add more checks - if mesh_vif.startswith("wl") or mesh_vif.startswith("halow"): - return True - return False + try: + if mesh_vif.startswith("wl") or mesh_vif.startswith("halow"): + return True + return False + except (ValueError, TypeError, AttributeError): + return False def validate_batman_iface(batman_iface: str) -> bool: """ Validates a given batman iface. Returns True if the batman iface is valid, False otherwise. """ - if "bat" in batman_iface and int(batman_iface.replace("bat", "")) >= 0: - return True - return False \ No newline at end of file + try: + if "bat" in batman_iface and int(batman_iface.replace("bat", "")) >= 0: + return True + return False + except (ValueError, TypeError, AttributeError): + return False \ No newline at end of file diff --git a/modules/sc-mesh-secure-deployment/src/nats/tests/test_settings.py b/modules/sc-mesh-secure-deployment/src/nats/tests/test_settings.py index e86abecd7..2e2e604f5 100644 --- a/modules/sc-mesh-secure-deployment/src/nats/tests/test_settings.py +++ b/modules/sc-mesh-secure-deployment/src/nats/tests/test_settings.py @@ -4,6 +4,7 @@ # pylint: disable=import-error, wrong-import-position, unused-import, \ # disable=unresolved-reference, undefined-variable, too-long import unittest +import json import logging from .context import comms_settings as comms from .context import comms_status as comms_status @@ -18,21 +19,143 @@ def test_handle_mesh_settings(self): Test cases for handle_mesh_settings() """ - - # settings json is valid logger = logging.getLogger("test") - cs = comms_status.CommsStatus(logger) + cs: [comms_status, ...] = [comms_status.CommsStatus(logger, "0"), + comms_status.CommsStatus(logger, "1"), + comms_status.CommsStatus(logger, "2")] settings = comms.CommsSettings(cs, logger) - ret, mesh_status = settings.handle_mesh_settings("""{"api_version": 1,"ssid": "test_mesh", "key": "1234567890","ap_mac": "00:11:22:33:44:55","country": "fi","frequency": "5220","frequency_mcc": "2412","ip": "192.168.1.2","subnet": "255.255.255.0","tx_power": "5","mode": "mesh","routing": "batman-adv", "priority":"long_range","role":"gcs"}""", "./tests", "test.conf") + + cmd_dict = { + "api_version": 1, + "role": "drone", # sleeve, drone, gcs + "radios": [ + { + "radio_index": "0", + "ssid": "test_mesh2", + "key": "1234567890", + "ap_mac": "00:11:22:33:44:55", + "country": "US", # all radios must have the same country + "frequency": "2412", + "frequency_mcc": "2412", # multiradio not supporting + "routing": "batman-adv", + "priority": "long_range", + "ip": "10.20.15.3", + "subnet": "255.255.255.0", + "tx_power": "15", + "mode": "mesh", # ap+mesh_scc, mesh, halow + "mesh_vif": "wlp2s0", + "batman_iface": "bat0", + }, + { + "radio_index": "1", + "ssid": "test_mesh", + "key": "1234567890", + "ap_mac": "00:11:22:33:44:55", + "country": "US", # all radios must have the same country + "frequency": "5220", + "frequency_mcc": "2412", # multiradio not supporting + "routing": "batman-adv", + "priority": "long_range", + "ip": "10.20.15.3", + "subnet": "255.255.255.0", + "tx_power": "15", + "mode": "mesh", # ap+mesh_scc, mesh, halow + "mesh_vif": "wlp3s0", # this needs to be correct + "batman_iface": "bat0", + }, + { + "radio_index": "2", + "ssid": "test_mesh3", + "key": "1234567890", + "ap_mac": "00:11:22:33:44:55", + "country": "US", # all radios must have the same country + "frequency": "5190", + "frequency_mcc": "2412", # multiradio not supporting + "routing": "batman-adv", + "priority": "long_range", + "ip": "10.20.15.3", + "subnet": "255.255.255.0", + "tx_power": "30", + "mode": "halow", # ap+mesh_scc, mesh, halow + "mesh_vif": "halow1", + "batman_iface": "bat0", + }, + ], + "bridge": "br-lan bat0 eth1 lan1 eth0 usb0", + } + + jsoned = json.dumps(cmd_dict) + ret, mesh_status = settings.handle_mesh_settings(jsoned, "./tests", "mesh.conf") + print(f"ret: {ret}, mesh_status: {mesh_status}") self.assertEqual(ret, "OK", msg=f"ret: {ret}, mesh_status: {mesh_status}") # settings json is invalid - ret, mesh_status = settings.handle_mesh_settings("""{}""", "./tests", "test.conf") + ret, mesh_status = settings.handle_mesh_settings("""{}""", "./tests", "mesh.conf") self.assertEqual(ret, "FAIL", msg=f"ret: {ret}, mesh_status: {mesh_status}") + # + # # # settings json is invalid + + cmd_dict = { + "api_version": 1, + "role": "drone", # sleeve, drone, gcs + "radios": [ + { + "radio_index": "0", + "ssid": "test_mesh2", + "key": "1234567890", + "ap_mac": "00:11:22:33:44:55", + "country": "US", # all radios must have the same country + "frequency": "2400", + "frequency_mcc": "2412", # multiradio not supporting + "routing": "batman-adv", + "priority": "long_range", + "ip": "10.20.15.3", + "subnet": "255.255.255.0", + "tx_power": "15", + "mode": "mesh", # ap+mesh_scc, mesh, halow + "mesh_vif": "wlp2s0.2", + "batman_iface": "bat", + }, + { + "radio_index": "1", + "ssid": "test_mesh", + "key": "1234567890", + "ap_mac": "00:11:22:33:44:55", + "country": "US", # all radios must have the same country + "frequency": "5220", + "frequency_mcc": "2412", # multiradio not supporting + "routing": "batman", + "priority": "long_range", + "ip": "10.20.15.3", + "subnet": "255.255.255.0", + "tx_power": "15", + "mode": "mesh", # ap+mesh_scc, mesh, halow + "mesh_vif": "wlp3s0", # this needs to be correct + "batman_iface": "bat0", + }, + { + "radio_index": "2", + "ssid": "test_mesh3", + "key": "123456 7890", + "ap_mac": "00:11:22:33:44:55", + "country": "US", # all radios must have the same country + "frequency": "5190", + "frequency_mcc": "2412", # multiradio not supporting + "routing": "None", + "priority": "looong_range", + "ip": "10.20.15.3", + "subnet": "255.255.255.0", + "tx_power": "30", + "mode": "halow", # ap+mesh_scc, mesh, halow + "mesh_vif": "halow1", + "batman_iface": "bat0", + }, + ], + "bridge to far": "br-lan bat0 eth1 lan1 eth0 usb0", + } - # # settings json is invalid - ret, mesh_status = settings.handle_mesh_settings("""{"api_version": 1,"ssid": "test;_mesh", "key":"1230","ap_mac": "00:11:22:33:44:55","country": "fi","frequency": "5220","ip": "2522.168.1.2","subnet": "a.255.255.0","tx_power": "5","mode": "mesh","priority":"long_range","role":"gcs"}}""", "./tests", "test.conf") + ret, mesh_status = settings.handle_mesh_settings("""{"radio_index": "0", "api_version": 1,"ssid": "test;_mesh", "key":"1230","ap_mac": "00:11:22:33:44:55","country": "fi","frequency": "5220","ip": "2522.168.1.2","subnet": "a.255.255.0","tx_power": "5","mode": "mesh","priority":"long_range","role":"gcs"}}""", "./tests", "test.conf") self.assertEqual(ret, "FAIL", msg=f"ret: {ret}, mesh_status: {mesh_status}") if __name__ == '__main__': diff --git a/modules/sc-mesh-secure-deployment/src/nats/tests/test_validation.py b/modules/sc-mesh-secure-deployment/src/nats/tests/test_validation.py index 193a56d0c..fddbb47fc 100644 --- a/modules/sc-mesh-secure-deployment/src/nats/tests/test_validation.py +++ b/modules/sc-mesh-secure-deployment/src/nats/tests/test_validation.py @@ -160,5 +160,86 @@ def test_validate_routing(self): # routing is invalid self.assertFalse(validation.validate_routing(1)) + def test_validate_priority(self): + """ + Test cases for validate_priority(priority) + """ + # priority is invalid + self.assertFalse(validation.validate_priority('none')) + # priority is valid + self.assertTrue(validation.validate_priority('long_range')) + # priority is valid + self.assertFalse(validation.validate_priority('high')) + # priority is invalid + self.assertFalse(validation.validate_priority('nonee')) + # priority is invalid + self.assertFalse(validation.validate_priority('long_rang')) + # priority is invalid + self.assertFalse(validation.validate_priority('hig')) + # priority is invalid + self.assertFalse(validation.validate_priority(1)) + # priotity is valid + self.assertTrue(validation.validate_priority("high_throughput")) + + def test_validate_radio_index(self): + """ + Test cases for validate_radio_index(radio_index) + """ + # radio index is valid + self.assertTrue(validation.validate_radio_index("0")) + # radio index is valid + self.assertTrue(validation.validate_radio_index("1")) + # radio index is valid + self.assertTrue(validation.validate_radio_index("2")) + # radio index is invalid + self.assertTrue(validation.validate_radio_index("3")) + # radio index is invalid + self.assertFalse(validation.validate_radio_index("-1")) + self.assertFalse(validation.validate_radio_index(-1)) + + def test_validate_mesh_vif(self): + """ + Test cases for validate_mesh_vif(mesh_vif) + """ + # mesh vif is valid + self.assertTrue(validation.validate_mesh_vif("wlp2s0")) + # mesh vif is valid + self.assertTrue(validation.validate_mesh_vif("wlp3s0")) + # mesh vif is valid + self.assertTrue(validation.validate_mesh_vif("halow1")) + # mesh vif is valid + self.assertTrue(validation.validate_mesh_vif("wlp2s1")) + # mesh vif is valid + self.assertTrue(validation.validate_mesh_vif("wlp3s1")) + # mesh vif is valid + self.assertTrue(validation.validate_mesh_vif("halow2")) + # mesh vif is invalid + self.assertFalse(validation.validate_mesh_vif("lan1")) + # mesh vif is invalid + self.assertFalse(validation.validate_mesh_vif("eth0")) + # mesh vif is invalid + self.assertFalse(validation.validate_mesh_vif("usb0")) + # mesh vif is invalid + self.assertFalse(validation.validate_mesh_vif(0)) + + def test_validate_batman_iface(self): + """ + Test cases for validate_batman_iface(batman_iface) + """ + # batman iface is valid + self.assertTrue(validation.validate_batman_iface("bat0")) + # batman iface is valid + self.assertTrue(validation.validate_batman_iface("bat1")) + # batman iface is valid + self.assertTrue(validation.validate_batman_iface("bat2")) + # batman iface is invalid + self.assertFalse(validation.validate_batman_iface("3")) + # batman iface is invalid + self.assertFalse(validation.validate_batman_iface(0)) + + + + + if __name__ == '__main__': unittest.main()