From e5c03fe99ca69fa78964a690db32b7ce395089ae Mon Sep 17 00:00:00 2001 From: Mika Joenpera Date: Thu, 25 Apr 2024 11:34:32 +0300 Subject: [PATCH] UnitTests and Coverage reporting 4966 - Comms Service Discovery 4967 - Comms Settings 4968 - Comms Status Jira-Id: SECO-4966, SECO-4967, SECO-4968 Signed-off-by: Mika Joenpera --- .../src/nats/coverage_report.txt | 33 ++ .../src/nats/run_unittests_PC.sh | 35 +- .../src/nats/src/comms_command.py | 304 +++++++-------- .../src/nats/src/comms_settings.py | 21 +- .../src/nats/src/comms_status.py | 2 + .../src/nats/tests/test_service_discovery.py | 63 +++- .../src/nats/tests/test_settings.py | 352 +++++++++++++++--- .../src/nats/tests/test_status.py | 113 +++++- 8 files changed, 691 insertions(+), 232 deletions(-) create mode 100644 modules/sc-mesh-secure-deployment/src/nats/coverage_report.txt diff --git a/modules/sc-mesh-secure-deployment/src/nats/coverage_report.txt b/modules/sc-mesh-secure-deployment/src/nats/coverage_report.txt new file mode 100644 index 000000000..d3986d540 --- /dev/null +++ b/modules/sc-mesh-secure-deployment/src/nats/coverage_report.txt @@ -0,0 +1,33 @@ +Name Stmts Miss Cover Missing +----------------------------------------------------------------- +mdm_agent.py 440 306 30% 103-104, 143, 163-208, 234-239, 248-250, 258-261, 267, 274-342, 355, 364-368, 378-473, 481-526, 536-571, 582-605, 626, 637-643, 659-695, 713-717, 723-737, 750-811, 822-873, 877-889 +src/__init__.py 0 0 100% +src/cbma_adaptation.py 473 383 19% 94-106, 109-112, 122-177, 182-190, 193-217, 223-237, 240-247, 252-276, 282-319, 322-335, 338-348, 351-354, 357-379, 382-385, 388-425, 428-435, 438-448, 451-458, 462-509, 513-529, 532-571, 578-591, 594-618, 625-639, 643-670, 676-702, 712-723, 726-737, 748-775, 782-807, 819-879, 886-893, 900-914 +src/cbma_paths.py 7 0 100% +src/comms_command.py 192 131 32% 88, 90, 92, 94-102, 105, 107, 109, 111-114, 116-119, 125, 147-220, 230-317, 321-335, 338-344, 348-362, 365-371, 410-434 +src/comms_common.py 26 0 100% +src/comms_config_store.py 21 0 100% +src/comms_controller.py 38 5 87% 33, 48, 79-82 +src/comms_if_monitor.py 56 2 96% 95-96 +src/comms_service_discovery.py 88 14 84% 114-115, 147, 152-153, 162, 180-211 +src/comms_settings.py 253 27 89% 12-13, 148-149, 164-184, 249-256, 312-315, 355, 362 +src/comms_status.py 296 14 95% 70, 194, 225-227, 240-242, 258, 264, 294, 302, 321-323, 528 +src/constants.py 29 0 100% +src/interface.py 5 0 100% +src/validation.py 104 2 98% 221-222 +tests/__init__.py 0 0 100% +tests/service_discovery_helper.py 22 0 100% +tests/test_command.py 26 0 100% +tests/test_config_store.py 26 0 100% +tests/test_constants.py 27 0 100% +tests/test_controller.py 32 0 100% +tests/test_if_monitor.py 25 1 96% 32 +tests/test_mdm_agent.py 56 0 100% +tests/test_service_discovery.py 45 6 87% 32-33, 55-56, 78-79 +tests/test_settings.py 153 0 100% +tests/test_status.py 128 8 94% 28-35 +tests/test_validation.py 146 0 100% +----------------------------------------------------------------- +TOTAL 2714 899 67% +Not tested files as not MDM content or tested elsewhere: + batadvvis.py,batstat.py,fmo_agent.py,comms_nats_discovery.py,cbma/*,debug_tests/*,comms_mesh_telemetry.py,comms_interface_info.py diff --git a/modules/sc-mesh-secure-deployment/src/nats/run_unittests_PC.sh b/modules/sc-mesh-secure-deployment/src/nats/run_unittests_PC.sh index 2aee12fc7..d2532f1fd 100755 --- a/modules/sc-mesh-secure-deployment/src/nats/run_unittests_PC.sh +++ b/modules/sc-mesh-secure-deployment/src/nats/run_unittests_PC.sh @@ -1,5 +1,26 @@ #!/bin/bash +# Check if the script is being run by root +if [ "$(id -u)" -ne 0 ]; then + echo "This script must be run with root privileges." + exit 1 +fi + +# preconditions +if [ ! -f "$(pwd)/$(basename $0)" ]; then + echo "Script is not being executed in the same folder" + exit 1 +fi + +needed_apps="batctl swig" +for app in $needed_apps; do + if ! command -v $app &> /dev/null + then + echo "$app is not installed. exit. Tips: sudo apt-get install $app" + exit 1 + fi +done + # python virtualenv python3 -m venv unittest source unittest/bin/activate @@ -8,13 +29,17 @@ source unittest/bin/activate pip install coverage==7.4.4 # this is for testing purpose pip install -r requirements.txt -# discover and run unittests -coverage run -m unittest discover -v -report=$(coverage report -m) +# List of files not to used for coverage calculation. +# Files tested elsewhere or not needed to be tested or not mesh shield content +not_used="batadvvis.py,batstat.py,fmo_agent.py,comms_nats_discovery.py,cbma/*,debug_tests/*,comms_mesh_telemetry.py,comms_interface_info.py" -# print report lines starting with "TOTAL" -echo "$report" | grep -e "^src" -e "^mdm" -e "^tests" +# discover and run unittests +coverage run --omit="$not_used" -m unittest discover -v +REPORT=$(coverage report -m) +# print and save coverage report +echo "$REPORT" | tee coverage_report.txt +echo -e "Not tested files as not MDM content or tested elsewhere:\n $not_used" >> coverage_report.txt # deactivate virtualenv deactivate diff --git a/modules/sc-mesh-secure-deployment/src/nats/src/comms_command.py b/modules/sc-mesh-secure-deployment/src/nats/src/comms_command.py index 8be0fe083..0d16a5fea 100644 --- a/modules/sc-mesh-secure-deployment/src/nats/src/comms_command.py +++ b/modules/sc-mesh-secure-deployment/src/nats/src/comms_command.py @@ -117,20 +117,20 @@ def handle_command(self, msg: str, cc) -> Tuple[str, str, str]: ret, info = self.__radio_up_all(cc) else: ret, info = self.__radio_up_single() - elif self.command == COMMAND.reboot: - ret, info = "FAIL", "Command not implemented" - elif self.command == COMMAND.get_logs: - ret, info, data = self.__get_logs(self.param) + # elif self.command == COMMAND.reboot: + # ret, info = "FAIL", "Command not implemented" + # elif self.command == COMMAND.get_logs: + # ret, info, data = self.__get_logs(self.param) elif self.command == COMMAND.debug: ret, info, data = self.__debug(cc, self.param) - elif self.command == COMMAND.enable_visualisation: - ret, info = self.__enable_visualisation(cc) - elif self.command == COMMAND.disable_visualisation: - ret, info = self.__disable_visualisation(cc) - elif self.command == COMMAND.get_config: - ret, info, data = self.__get_configs(self.param) - elif self.command == COMMAND.get_identity: - ret, info, data = self.get_identity() # type: ignore[assignment] + # elif self.command == COMMAND.enable_visualisation: + # ret, info = self.__enable_visualisation(cc) + # elif self.command == COMMAND.disable_visualisation: + # ret, info = self.__disable_visualisation(cc) + # elif self.command == COMMAND.get_config: + # ret, info, data = self.__get_configs(self.param) + # elif self.command == COMMAND.get_identity: + # ret, info, data = self.get_identity() # type: ignore[assignment] else: ret, info = "FAIL", "Command not supported" return ret, info, data @@ -209,13 +209,13 @@ def __revoke(self, cc) -> Tuple[str, str]: self.logger.debug("Default mesh command applied") - if self.comms_status[int(self.radio_index)].is_visualisation_active: - status, _ = self.__disable_visualisation(cc) - if status == "FAIL": - return ( - "FAIL", - "Revoke failed partially." + " Visualisation is still active", - ) + # if self.comms_status[int(self.radio_index)].is_visualisation_active: + # status, _ = self.__disable_visualisation(cc) + # if status == "FAIL": + # return ( + # "FAIL", + # "Revoke failed partially." + " Visualisation is still active", + # ) return "OK", "Mesh settings revoked" @@ -370,41 +370,41 @@ def __radio_up_all(self, cc) -> Tuple[str, str]: self.logger.debug("All radios activated") return "OK", "All radios activated" - def __enable_visualisation(self, cc) -> Tuple[str, str]: - try: - cc.telemetry.run() - except Exception as e: - self.logger.error("Failed to enable visualisation, %s", e) - return "FAIL", "Failed to enable visualisation" - - self.logger.debug("Visualisation enabled") - self.comms_status[int(self.radio_index)].is_visualisation_active = True - return "OK", "Visualisation enabled" - - def __disable_visualisation(self, cc) -> Tuple[str, str]: - try: - cc.telemetry.stop() - cc.visualisation_enabled = False - except Exception as e: - self.logger.error("Failed to disable visualisation, %s", e) - return "FAIL", "Failed to disable visualisation" - - self.logger.debug("Visualisation disabled") - self.comms_status[int(self.radio_index)].is_visualisation_active = False - return "OK", "Visualisation disabled" - - @staticmethod - def __read_log_file(filename) -> bytes: - """ - read file and return the content as bytes and base64 encoded - - param: filename: str - return: (int, bytes) - """ - # read as bytes as b64encode expects bytes - with open(filename, "rb") as file: - file_log = file.read() - return base64.b64encode(file_log) + # def __enable_visualisation(self, cc) -> Tuple[str, str]: + # try: + # cc.telemetry.run() + # except Exception as e: + # self.logger.error("Failed to enable visualisation, %s", e) + # return "FAIL", "Failed to enable visualisation" + # + # self.logger.debug("Visualisation enabled") + # self.comms_status[int(self.radio_index)].is_visualisation_active = True + # return "OK", "Visualisation enabled" + # + # def __disable_visualisation(self, cc) -> Tuple[str, str]: + # try: + # cc.telemetry.stop() + # cc.visualisation_enabled = False + # except Exception as e: + # self.logger.error("Failed to disable visualisation, %s", e) + # return "FAIL", "Failed to disable visualisation" + # + # self.logger.debug("Visualisation disabled") + # self.comms_status[int(self.radio_index)].is_visualisation_active = False + # return "OK", "Visualisation disabled" + # + # @staticmethod + # def __read_log_file(filename) -> bytes: + # """ + # read file and return the content as bytes and base64 encoded + # + # param: filename: str + # return: (int, bytes) + # """ + # # read as bytes as b64encode expects bytes + # with open(filename, "rb") as file: + # file_log = file.read() + # return base64.b64encode(file_log) def __debug(self, cc, param) -> Tuple[str, str, str]: file = "" @@ -433,101 +433,101 @@ def __debug(self, cc, param) -> Tuple[str, str, str]: self.logger.debug("__debug done") return "OK", f"'{p}' DEBUG COMMAND done", file_b64.decode() - def __get_logs(self, param) -> Tuple[str, str, str]: - file = "" - try: - files = LogFiles() - if param == files.WPA: - file_b64 = self.__read_log_file( - files.WPA_LOG + "_id" + self.radio_index + ".log" - ) - elif param == files.HOSTAPD: - file_b64 = self.__read_log_file( - files.HOSTAPD_LOG + "_id" + self.radio_index + ".log" - ) - elif param == files.CONTROLLER: - file_b64 = self.__read_log_file(files.CONTROLLER_LOG) - elif param == files.DMESG: - ret = subprocess.run( - [files.DMESG_CMD], - shell=False, - check=True, - capture_output=True, - ) - if ret.returncode != 0: - return "FAIL", f"{file} file read failed", "" - file_b64 = base64.b64encode(ret.stdout) - else: - return "FAIL", "Log file not supported", "" - - except Exception as e: - self.logger.error("Log reading failed, %s", e) - return "FAIL", f"{param} log reading failed", "" - - self.logger.debug("__getlogs done") - return "OK", "wpa_supplicant log", file_b64.decode() - - def __get_configs(self, param) -> Tuple[str, str, str]: - file_b64 = b"None" - try: - files = ConfigFiles() - self.comms_status[int(self.radio_index)].refresh_status() - if param == files.WPA: - file_b64 = self.__read_log_file( - f"/var/run/wpa_supplicant-11s_id{self.radio_index}.conf" - ) - elif param == files.HOSTAPD: - file_b64 = self.__read_log_file( - f"/var/run/hostapd-{self.radio_index}.conf" - ) - else: - return "FAIL", "Parameter not supported", "" - - except Exception as e: - self.logger.error("Not able to get configs, %s", e) - return "FAIL", "Not able to get config file", "" - - self.logger.debug("__get_configs done") - return "OK", f"{param}", file_b64.decode() - - def get_identity(self) -> Tuple[str, str, Union[str, dict]]: - """ - Gathers identity, NATS server url and wireless interface - device information and returns that in JSON compatible - dictionary format. - - Returns: - tuple: (str, str, str | dict) -- A tuple that contains - always textual status and description elements. 3rd - element is JSON compatible in normal cases but in - case of failure it is an empty string. - """ - - identity_dict = {} - nats_ip = "NA" - try: - files = ConfigFiles() - for status in self.comms_status: - status.refresh_status() - - with open(files.IDENTITY, "rb") as file: - identity = file.read() - identity_dict["identity"] = identity.decode().strip() - - # pylint: disable=c-extension-no-member - ips = ni.ifaddresses("br-lan") - for item in ips[ni.AF_INET6]: - if item["addr"].startswith("fd"): - nats_ip = item["addr"] - break - - identity_dict["nats_url"] = f"nats://[{nats_ip}]:4222" - identity_dict[ - "interfaces" - ] = CommsInterfaces().get_wireless_device_info() # type: ignore - except Exception as e: - self.logger.error("Not able to get identity, %s", e) - return "FAIL", "Not able to get identity file", "" - - self.logger.debug("get_identity done") - return "OK", "Identity and NATS URL", identity_dict + # def __get_logs(self, param) -> Tuple[str, str, str]: + # file = "" + # try: + # files = LogFiles() + # if param == files.WPA: + # file_b64 = self.__read_log_file( + # files.WPA_LOG + "_id" + self.radio_index + ".log" + # ) + # elif param == files.HOSTAPD: + # file_b64 = self.__read_log_file( + # files.HOSTAPD_LOG + "_id" + self.radio_index + ".log" + # ) + # elif param == files.CONTROLLER: + # file_b64 = self.__read_log_file(files.CONTROLLER_LOG) + # elif param == files.DMESG: + # ret = subprocess.run( + # [files.DMESG_CMD], + # shell=False, + # check=True, + # capture_output=True, + # ) + # if ret.returncode != 0: + # return "FAIL", f"{file} file read failed", "" + # file_b64 = base64.b64encode(ret.stdout) + # else: + # return "FAIL", "Log file not supported", "" + # + # except Exception as e: + # self.logger.error("Log reading failed, %s", e) + # return "FAIL", f"{param} log reading failed", "" + # + # self.logger.debug("__getlogs done") + # return "OK", "wpa_supplicant log", file_b64.decode() + # + # def __get_configs(self, param) -> Tuple[str, str, str]: + # file_b64 = b"None" + # try: + # files = ConfigFiles() + # self.comms_status[int(self.radio_index)].refresh_status() + # if param == files.WPA: + # file_b64 = self.__read_log_file( + # f"/var/run/wpa_supplicant-11s_id{self.radio_index}.conf" + # ) + # elif param == files.HOSTAPD: + # file_b64 = self.__read_log_file( + # f"/var/run/hostapd-{self.radio_index}.conf" + # ) + # else: + # return "FAIL", "Parameter not supported", "" + # + # except Exception as e: + # self.logger.error("Not able to get configs, %s", e) + # return "FAIL", "Not able to get config file", "" + # + # self.logger.debug("__get_configs done") + # return "OK", f"{param}", file_b64.decode() + # + # def get_identity(self) -> Tuple[str, str, Union[str, dict]]: + # """ + # Gathers identity, NATS server url and wireless interface + # device information and returns that in JSON compatible + # dictionary format. + # + # Returns: + # tuple: (str, str, str | dict) -- A tuple that contains + # always textual status and description elements. 3rd + # element is JSON compatible in normal cases but in + # case of failure it is an empty string. + # """ + # + # identity_dict = {} + # nats_ip = "NA" + # try: + # files = ConfigFiles() + # for status in self.comms_status: + # status.refresh_status() + # + # with open(files.IDENTITY, "rb") as file: + # identity = file.read() + # identity_dict["identity"] = identity.decode().strip() + # + # # pylint: disable=c-extension-no-member + # ips = ni.ifaddresses("br-lan") + # for item in ips[ni.AF_INET6]: + # if item["addr"].startswith("fd"): + # nats_ip = item["addr"] + # break + # + # identity_dict["nats_url"] = f"nats://[{nats_ip}]:4222" + # identity_dict[ + # "interfaces" + # ] = CommsInterfaces().get_wireless_device_info() # type: ignore + # except Exception as e: + # self.logger.error("Not able to get identity, %s", e) + # return "FAIL", "Not able to get identity file", "" + # + # self.logger.debug("get_identity done") + # return "OK", "Identity and NATS URL", identity_dict 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 62a000a6b..9c48a51ef 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 @@ -65,21 +65,30 @@ def validate_mesh_settings(self, index: int) -> (str, str): return "FAIL", "Invalid mode" self.logger.debug("validate mesh settings mode ok") - if validation.validate_frequency(int(self.frequency[index])) is False: + try: + if validation.validate_frequency(int(self.frequency[index])) is False: + return "FAIL", "Invalid frequency" + self.logger.debug("validate mesh settings freq ok") + except ValueError: return "FAIL", "Invalid frequency" - self.logger.debug("validate mesh settings freq ok") - if validation.validate_frequency(int(self.frequency_mcc[index])) is False: + try: + if validation.validate_frequency(int(self.frequency_mcc[index])) is False: + return "FAIL", "Invalid mcc frequency" + self.logger.debug("validate mesh settings mcc freq ok") + except ValueError: return "FAIL", "Invalid mcc frequency" - self.logger.debug("validate mesh settings mcc freq ok") if validation.validate_country_code(self.country[index]) is False: return "FAIL", "Invalid country code" self.logger.debug("validate mesh settings country ok") - if validation.validate_tx_power(int(self.tx_power[index])) is False: + try: + if validation.validate_tx_power(int(self.tx_power[index])) is False: + return "FAIL", "Invalid tx power" + self.logger.debug("validate mesh settings tx power ok") + except ValueError: return "FAIL", "Invalid tx power" - self.logger.debug("validate mesh settings tx power ok") if validation.validate_priority(self.priority[index]) is False: return "FAIL", "Invalid priority" diff --git a/modules/sc-mesh-secure-deployment/src/nats/src/comms_status.py b/modules/sc-mesh-secure-deployment/src/nats/src/comms_status.py index e9015578e..ed206f3c2 100644 --- a/modules/sc-mesh-secure-deployment/src/nats/src/comms_status.py +++ b/modules/sc-mesh-secure-deployment/src/nats/src/comms_status.py @@ -495,6 +495,7 @@ def __get_mission_cfg_status(self): hash_file_path = f"/opt/{str(self.index)}_mesh.conf_hash" old_mesh_cfg_status = self.__mesh_cfg_status old_is_mission_cfg = self.__is_mission_cfg + try: with open(config_file_path, "rb") as f: config = f.read() @@ -539,3 +540,4 @@ def __get_mission_cfg_status(self): self.__mesh_cfg_status, self.__is_mission_cfg, ) + diff --git a/modules/sc-mesh-secure-deployment/src/nats/tests/test_service_discovery.py b/modules/sc-mesh-secure-deployment/src/nats/tests/test_service_discovery.py index 8e9307949..d69a9b363 100644 --- a/modules/sc-mesh-secure-deployment/src/nats/tests/test_service_discovery.py +++ b/modules/sc-mesh-secure-deployment/src/nats/tests/test_service_discovery.py @@ -1,5 +1,5 @@ import unittest -from unittest.mock import MagicMock +from unittest.mock import MagicMock, patch from src.comms_service_discovery import CommsServiceMonitor from tests.service_discovery_helper import (dns_service_register, @@ -7,18 +7,21 @@ service_discovery_cb, get_kwargs_dict, SERVICE_NAME, SERVICE_TYPE) -class CommsServiceDiscoveryProperties(unittest.TestCase): +class TestCommsServiceDiscovery(unittest.TestCase): - def test_comms_service_discovery_initialization(self): + @patch('src.comms_controller.logging.getLogger') + @patch('src.comms_controller.logging.debug') + def test_comms_service_discovery_initialization(self, mock_logger_debug, mock_get_logger): try: - logger = MagicMock() + mock_logger = MagicMock() + mock_logger_debug.return_value = None + mock_get_logger.return_value = mock_logger dns_service_register() comms_service_discovery = CommsServiceMonitor( service_name=SERVICE_NAME, service_type=SERVICE_TYPE, service_cb=service_discovery_cb, test=True, - logger=logger ) comms_service_discovery.run() @@ -29,5 +32,51 @@ def test_comms_service_discovery_initialization(self): except Exception as e: self.fail(f"An error occurred: {e}") - finally: - dns_service_unregister() + comms_service_discovery.close() + dns_service_unregister() + + def test_comms_service_discovery_initialization_with_logger(self): + try: + mock_logger = MagicMock() + dns_service_register() + comms_service_discovery = CommsServiceMonitor( + service_name=SERVICE_NAME, + service_type=SERVICE_TYPE, + service_cb=service_discovery_cb, + test=True, + logger=mock_logger + ) + + comms_service_discovery.run() + + kwargs_dict: dict = get_kwargs_dict() + self.assertEqual(kwargs_dict['service_name'], f'{SERVICE_NAME}.{SERVICE_TYPE}') + + except Exception as e: + self.fail(f"An error occurred: {e}") + + comms_service_discovery.close() + dns_service_unregister() + + def test_comms_service_discovery_initialization_with_interface(self): + try: + mock_logger = MagicMock() + dns_service_register() + comms_service_discovery = CommsServiceMonitor( + service_name=SERVICE_NAME, + service_type=SERVICE_TYPE, + service_cb=service_discovery_cb, + test=True, + logger=mock_logger, + interface='lo' + ) + + comms_service_discovery.run() + + kwargs_dict: dict = get_kwargs_dict() + self.assertEqual(kwargs_dict['service_name'], f'{SERVICE_NAME}.{SERVICE_TYPE}') + except Exception as e: + self.fail(f"An error occurred: {e}") + + comms_service_discovery.close() + dns_service_unregister() 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 6a0dde93b..7583c8681 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,17 +4,92 @@ # pylint: disable=import-error, wrong-import-position, unused-import, \ # disable=unresolved-reference, undefined-variable, too-long import unittest +from unittest.mock import mock_open +import os +from copy import deepcopy from unittest.mock import patch, MagicMock import json import warnings from src.comms_settings import CommsSettings from src.comms_status import CommsStatus +cmd_dict_org = { + "api_version": 1, + "role": "drone", # sleeve, drone, gcs + "radios": [ + { + "radio_index": "0", + "ssid": "test_mesh2", + "key": "1234567890", + "country": "US", # all radios must have the same country + "frequency": "2412", + "frequency_mcc": "2412", # multiradio not supporting + "priority": "long_range", + "tx_power": "15", + "mptcp": "disable", + "slaac": "usb0 wlp3s0", + "mode": "mesh", # ap+mesh_scc, mesh, halow + "mesh_vif": "wlp2s0", + }, + { + "radio_index": "1", + "ssid": "test_mesh", + "key": "1234567890", + "country": "US", # all radios must have the same country + "frequency": "5220", + "frequency_mcc": "2412", # multiradio not supporting + "priority": "long_range", + "slaac": "usb0 wlp3s0", + "tx_power": "15", + "mptcp": "disable", + "mode": "mesh", # ap+mesh_scc, mesh, halow + "mesh_vif": "wlp3s0", # this needs to be correct + }, + { + "radio_index": "2", + "ssid": "test_mesh3", + "key": "1234567890", + "country": "US", # all radios must have the same country + "frequency": "5190", + "frequency_mcc": "2412", # multiradio not supporting + "priority": "long_range", + "tx_power": "30", + "slaac": "usb0 wlp3s0", + "mptcp": "disable", + "mode": "halow", # ap+mesh_scc, mesh, halow + "mesh_vif": "halow1", + }, + ], +} + + class TestSettings(unittest.TestCase): """ Test cases for comms_settings.py """ + @classmethod + def tearDownClass(cls): + current_path = os.getcwd() + test_files = ["0_mesh.conf", "1_mesh.conf", "2_mesh.conf"] + + # Clean up the test files + for file_name in test_files: + file_path = os.path.join(current_path, "tests", file_name) + if os.path.exists(file_path): + os.remove(file_path) + + @classmethod + def setUpClass(cls): + logger = MagicMock() + + cls.cs: [CommsStatus, ...] = [ + CommsStatus(logger, "0"), + CommsStatus(logger, "1"), + CommsStatus(logger, "2"), + ] + cls.settings = CommsSettings(cls.cs, logger) + @patch("src.validation.is_valid_interface") def test_handle_mesh_settings(self, mock_is_valid_interface): """ @@ -26,82 +101,237 @@ def test_handle_mesh_settings(self, mock_is_valid_interface): mock_is_valid_interface.return_value = True warnings.simplefilter("ignore", ResourceWarning) - # settings json is valid - logger = MagicMock() - - cs: [CommsStatus, ...] = [ - CommsStatus(logger, "0"), - CommsStatus(logger, "1"), - CommsStatus(logger, "2"), - ] - settings = CommsSettings(cs, logger) - - cmd_dict = { - "api_version": 1, - "role": "drone", # sleeve, drone, gcs - "radios": [ - { - "radio_index": "0", - "ssid": "test_mesh2", - "key": "1234567890", - "country": "US", # all radios must have the same country - "frequency": "2412", - "frequency_mcc": "2412", # multiradio not supporting - "priority": "long_range", - "tx_power": "15", - "mptcp": "disable", - "slaac": "usb0 wlp3s0", - "mode": "mesh", # ap+mesh_scc, mesh, halow - "mesh_vif": "wlp2s0", - }, - { - "radio_index": "1", - "ssid": "test_mesh", - "key": "1234567890", - "country": "US", # all radios must have the same country - "frequency": "5220", - "frequency_mcc": "2412", # multiradio not supporting - "priority": "long_range", - "slaac": "usb0 wlp3s0", - "tx_power": "15", - "mptcp": "disable", - "mode": "mesh", # ap+mesh_scc, mesh, halow - "mesh_vif": "wlp3s0", # this needs to be correct - }, - { - "radio_index": "2", - "ssid": "test_mesh3", - "key": "1234567890", - "country": "US", # all radios must have the same country - "frequency": "5190", - "frequency_mcc": "2412", # multiradio not supporting - "priority": "long_range", - "tx_power": "30", - "slaac": "usb0 wlp3s0", - "mptcp": "disable", - "mode": "halow", # ap+mesh_scc, mesh, halow - "mesh_vif": "halow1", - }, - ], - } + + cmd_dict = deepcopy(cmd_dict_org) jsoned = json.dumps(cmd_dict) - ret, mesh_status = settings.handle_mesh_settings( + ret, mesh_status = self.settings.handle_mesh_settings( jsoned, "./tests", "mesh.conf" ) self.assertEqual(ret, "OK", msg=f"ret: {ret}, mesh_status: {mesh_status}") # settings json is invalid - ret, mesh_status = settings.handle_mesh_settings( + ret, mesh_status = self.settings.handle_mesh_settings( """{}""", "./tests", "mesh.conf" ) self.assertEqual(ret, "FAIL", msg=f"ret: {ret}, mesh_status: {mesh_status}") # # # # settings json is invalid - ret, mesh_status = settings.handle_mesh_settings( + ret, mesh_status = self.settings.handle_mesh_settings( """{"radio_index": "0", "api_version": 1,"ssid": "test;_mesh", "key":"1230","country": "fi","frequency": "5220","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}") + + del cmd_dict + + + def test_handle_mesh_settings_validate_parameters_ssid(self): + cmd_dict = deepcopy(cmd_dict_org) + + #ssid + cmd_dict["radios"][0]["ssid"] = "a" * 33 + jsoned = json.dumps(cmd_dict) + ret, mesh_status = self.settings.handle_mesh_settings( + jsoned, "./tests", "mesh.conf" + ) + self.assertEqual(ret, "FAIL", msg=f"ret: {ret}, mesh_status: {mesh_status}") + del cmd_dict + + def test_handle_mesh_settings_validate_parameters_key(self): + cmd_dict = deepcopy(cmd_dict_org) + + #key + cmd_dict["radios"][0]["key"] = "a" + jsoned = json.dumps(cmd_dict) + ret, mesh_status = self.settings.handle_mesh_settings( + jsoned, "./tests", "mesh.conf" + ) + self.assertEqual(ret, "FAIL", msg=f"ret: {ret}, mesh_status: {mesh_status}") + del cmd_dict + + def test_handle_mesh_settings_validate_parameters_mode(self): + cmd_dict = deepcopy(cmd_dict_org) + + #mode + cmd_dict["radios"][0]["key"] = "1234567890" + cmd_dict["radios"][0]["mode"] = "test" + jsoned = json.dumps(cmd_dict) + ret, mesh_status = self.settings.handle_mesh_settings( + jsoned, "./tests", "mesh.conf" + ) + self.assertEqual(ret, "FAIL", msg=f"ret: {ret}, mesh_status: {mesh_status}") + del cmd_dict + + def test_handle_mesh_settings_validate_parameters_frequency(self): + cmd_dict = deepcopy(cmd_dict_org) + + #frequency + cmd_dict["radios"][0]["frequency"] = "test" + jsoned = json.dumps(cmd_dict) + ret, mesh_status = self.settings.handle_mesh_settings( + jsoned, "./tests", "mesh.conf" + ) + self.assertEqual(ret, "FAIL", msg=f"ret: {ret}, mesh_status: {mesh_status}") + del cmd_dict + + cmd_dict = deepcopy(cmd_dict_org) + cmd_dict["radios"][0]["frequency"] = "2444" + jsoned = json.dumps(cmd_dict) + ret, mesh_status = self.settings.handle_mesh_settings( + jsoned, "./tests", "mesh.conf" + ) + self.assertEqual(ret, "FAIL", msg=f"ret: {ret}, mesh_status: {mesh_status}") + del cmd_dict + + def test_handle_mesh_settings_validate_parameters_frequency_mcc(self): + cmd_dict = deepcopy(cmd_dict_org) + + #frequency_mcc + cmd_dict["radios"][0]["frequency_mcc"] = "test" + jsoned = json.dumps(cmd_dict) + ret, mesh_status = self.settings.handle_mesh_settings( + jsoned, "./tests", "mesh.conf" + ) + self.assertEqual(ret, "FAIL", msg=f"ret: {ret}, mesh_status: {mesh_status}") + del cmd_dict + + cmd_dict = deepcopy(cmd_dict_org) + cmd_dict["radios"][0]["frequency_mcc"] = "2400" + jsoned = json.dumps(cmd_dict) + ret, mesh_status = self.settings.handle_mesh_settings( + jsoned, "./tests", "mesh.conf" + ) + self.assertEqual(ret, "FAIL", msg=f"ret: {ret}, mesh_status: {mesh_status}") + del cmd_dict + + def test_handle_mesh_settings_validate_parameters_cc(self): + cmd_dict = deepcopy(cmd_dict_org) + + #country + cmd_dict["radios"][0]["country"] = "test" + jsoned = json.dumps(cmd_dict) + ret, mesh_status = self.settings.handle_mesh_settings( + jsoned, "./tests", "mesh.conf" + ) + self.assertEqual(ret, "FAIL", msg=f"ret: {ret}, mesh_status: {mesh_status}") + del cmd_dict + + def test_handle_mesh_settings_validate_parameters_tx_power(self): + cmd_dict = deepcopy(cmd_dict_org) + + #tx_power + cmd_dict["radios"][0]["tx_power"] = "test" + jsoned = json.dumps(cmd_dict) + ret, mesh_status = self.settings.handle_mesh_settings( + jsoned, "./tests", "mesh.conf" + ) + self.assertEqual(ret, "FAIL", msg=f"ret: {ret}, mesh_status: {mesh_status}") + del cmd_dict + + cmd_dict = deepcopy(cmd_dict_org) + cmd_dict["radios"][0]["tx_power"] = "50" + jsoned = json.dumps(cmd_dict) + ret, mesh_status = self.settings.handle_mesh_settings( + jsoned, "./tests", "mesh.conf" + ) + self.assertEqual(ret, "FAIL", msg=f"ret: {ret}, mesh_status: {mesh_status}") + + def test_handle_mesh_settings_validate_parameters_priority(self): + cmd_dict = deepcopy(cmd_dict_org) + + #priority + cmd_dict["radios"][0]["priority"] = "test" + jsoned = json.dumps(cmd_dict) + ret, mesh_status = self.settings.handle_mesh_settings( + jsoned, "./tests", "mesh.conf" + ) + self.assertEqual(ret, "FAIL", msg=f"ret: {ret}, mesh_status: {mesh_status}") + del cmd_dict + + def test_handle_mesh_settings_validate_parameters_role(self): + cmd_dict = deepcopy(cmd_dict_org) + + #role + cmd_dict["role"] = "test" + jsoned = json.dumps(cmd_dict) + ret, mesh_status = self.settings.handle_mesh_settings( + jsoned, "./tests", "mesh.conf" + ) + self.assertEqual(ret, "FAIL", msg=f"ret: {ret}, mesh_status: {mesh_status}") + del cmd_dict + + def test_handle_mesh_settings_validate_parameters_mptcp(self): + cmd_dict = deepcopy(cmd_dict_org) + + #mptcp + cmd_dict["radios"][0]["mptcp"] = "test" + jsoned = json.dumps(cmd_dict) + ret, mesh_status = self.settings.handle_mesh_settings( + jsoned, "./tests", "mesh.conf" + ) + self.assertEqual(ret, "FAIL", msg=f"ret: {ret}, mesh_status: {mesh_status}") + del cmd_dict + + def test_handle_mesh_settings_validate_parameters_slaac(self): + cmd_dict = deepcopy(cmd_dict_org) + + #slaac + cmd_dict["radios"][0]["slaac"] = "test" + jsoned = json.dumps(cmd_dict) + ret, mesh_status = self.settings.handle_mesh_settings( + jsoned, "./tests", "mesh.conf" + ) + self.assertEqual(ret, "FAIL", msg=f"ret: {ret}, mesh_status: {mesh_status}") + del cmd_dict + + def test_handle_mesh_settings_validate_parameters_channel_change(self): + + channel_change_dict = {"frequency": "2452", "radio_index": "1"} + jsoned = json.dumps(channel_change_dict) + ret, mesh_status, trigger = self.settings.handle_mesh_settings_channel_change( + jsoned, "./tests", "mesh.conf" + ) + #other settings are clean and empty + self.assertEqual(ret, "FAIL", msg=f"ret: {ret}, mesh_status: {mesh_status}") + del channel_change_dict + + def test_handle_mesh_settings_load_settings(self): + # Prepare mock file content (simulate YAML content) + mock_file_content = """ + ROLE=DRONE + MSVERSION=nats + id0_MODE=ap+mesh_mcc + id0_KEY=1234567890 + id0_ESSID=gold + id0_FREQ=5805 + id0_FREQ_MCC=2412 + id0_TXPOWER=30 + id0_COUNTRY=US + id0_PRIORITY=high_throughput + id0_MESH_VIF=wlp1s0 + id0_MPTCP=disable + id0_SLAAC="wlan1 usb0" + """ + + with patch("os.path.exists", return_value=True): + # Patch the open function to return the mock file content + with patch("builtins.open", mock_open(read_data=mock_file_content)) as mock_file: + ret, info = self.settings._CommsSettings__load_settings() + + # Verify that open was called with the expected arguments for each file + mock_file.assert_any_call("/opt/0_mesh.conf", "r", encoding="utf-8") + mock_file.assert_any_call("/opt/1_mesh.conf", "r", encoding="utf-8") + mock_file.assert_any_call("/opt/2_mesh.conf", "r", encoding="utf-8") + + # Ensure that the method returns the expected values + self.assertEqual(ret, "OK", msg=f"ret: {ret}, info: {info}") + + with patch("os.path.exists", return_value=False): + # Patch the open function to return the mock file content + with patch("builtins.open", mock_open(read_data=mock_file_content)) as mock_file: + ret, info = self.settings._CommsSettings__load_settings() + mock_file.assert_called_once_with("/opt/mesh_default.conf", "r", encoding="utf-8") + self.assertEqual(ret, "OK", msg=f"ret: {ret}, info: {info}") diff --git a/modules/sc-mesh-secure-deployment/src/nats/tests/test_status.py b/modules/sc-mesh-secure-deployment/src/nats/tests/test_status.py index 35aee8938..bdc16d79a 100644 --- a/modules/sc-mesh-secure-deployment/src/nats/tests/test_status.py +++ b/modules/sc-mesh-secure-deployment/src/nats/tests/test_status.py @@ -1,5 +1,6 @@ import unittest from unittest.mock import patch +import tempfile import warnings import os from src.constants import Constants @@ -78,4 +79,114 @@ def test_comms_status_properties(self): assert comms_status.is_visualisation_active is not None assert comms_status.is_ap_radio_on is not None assert comms_status.ap_interface_name is not None - assert comms_status.mesh_interface_name is not None \ No newline at end of file + assert comms_status.mesh_interface_name is not None + + def test_comms_status_mission_config(self): + + logger = MagicMock() + comms_status = CommsStatus(logger=logger, index=0) + + # Define mock file content + config_content = b"test_config_content" + hash_content = b"1a709d4bfad22352b1136fff4061c3c29313e02d06113c18da16c94551a9d62c" + wrong_hash_content = b"1a1a1a1a" + + # Define file paths + config_file_path = "/opt/0_mesh.conf" + hash_file_path = "/opt/0_mesh.conf_hash" + + # Create temporary files with the specified paths + with open(config_file_path, "wb") as config_file: + config_file.write(config_content) + + with open(hash_file_path, "wb") as hash_file: + hash_file.write(hash_content) + + # Call the method under test with correct hash + comms_status._CommsStatus__get_mission_cfg_status() + self.assertTrue(comms_status.is_mission_cfg) # Mission config is expected + + with open(hash_file_path, "wb") as hash_file: + hash_file.write(wrong_hash_content) + + # Call the method under test with wrong hash + comms_status._CommsStatus__get_mission_cfg_status() + self.assertFalse(comms_status.is_mission_cfg) # Mission config is not expected + + os.unlink(hash_file_path) + # without hash file causing FileNotFoundError + comms_status._CommsStatus__get_mission_cfg_status() + self.assertFalse(comms_status.is_mission_cfg) # Mission config is not expected + + os.unlink(config_file_path) + # without files causing FileNotFoundError + comms_status._CommsStatus__get_mission_cfg_status() + self.assertFalse(comms_status.is_mission_cfg) # Mission config is not expected + + @patch('subprocess.Popen') + def test_get_wpa_cli_status_with_valid_output(self, mock_popen): + # Mocking the Popen call and setting the return value + mock_popen.return_value.communicate.return_value = ( + b'Selected interface \'wlan0\'\nbssid=00:00:00:00:00:00\nfreq=2412\nssid=test\nid=0\nmode=station\npairwise_cipher=CCMP\n' + b'group_cipher=CCMP\nkey_mgmt=WPA2-PSK\nwpa_state=COMPLETED\naddress=00:00:00:00:00:00\nuuid=00000000-0000-0000-0000-000000000000\n', + None + ) + + comms_status = CommsStatus(logger=MagicMock(), index=0) + comms_status._CommsStatus__get_wpa_cli_status() + + self.assertEqual(comms_status._CommsStatus__wpa_status.interface, 'wlan0') + self.assertEqual(comms_status._CommsStatus__wpa_status.bssid, '00:00:00:00:00:00') + self.assertEqual(comms_status._CommsStatus__wpa_status.freq, '2412') + self.assertEqual(comms_status._CommsStatus__wpa_status.ssid, 'test') + self.assertEqual(comms_status._CommsStatus__wpa_status.id, '0') + self.assertEqual(comms_status._CommsStatus__wpa_status.mode, 'station') + self.assertEqual(comms_status._CommsStatus__wpa_status.pairwise_cipher, 'CCMP') + self.assertEqual(comms_status._CommsStatus__wpa_status.group_cipher, 'CCMP') + self.assertEqual(comms_status._CommsStatus__wpa_status.key_mgmt, 'WPA2-PSK') + self.assertEqual(comms_status._CommsStatus__wpa_status.wpa_state, 'COMPLETED') + self.assertEqual(comms_status._CommsStatus__wpa_status.address, '00:00:00:00:00:00') + self.assertEqual(comms_status._CommsStatus__wpa_status.uuid, + '00000000-0000-0000-0000-000000000000') + + @patch('subprocess.Popen') + def test_get_wpa_cli_status_with_error(self, mock_popen): + # Mocking the Popen call and setting the return value + mock_popen.return_value.communicate.return_value = ( + None, + b'Error from wpa_cli' + ) + + comms_status = CommsStatus(logger=MagicMock(), index=0) + with self.assertRaises(RuntimeError): + comms_status._CommsStatus__get_wpa_cli_status() + + @patch('subprocess.Popen') + def test_hostapd_cli_status_with_valid_output(self, mock_popen): + mock_popen.return_value.communicate.return_value = ( + b'Selected interface \'wlan0\'\nstate=ENABLED\nphy=phy0\nfreq=2412\nchannel=6\nbeacon_int=100\nssid[0]=test\n', + None + ) + + comms_status = CommsStatus(logger=MagicMock(), index=0) + comms_status._CommsStatus__get_hostapd_cli_status() + + self.assertEqual(comms_status._CommsStatus__hostapd_status.interface, 'wlan0') + self.assertEqual(comms_status._CommsStatus__hostapd_status.state, 'ENABLED') + self.assertEqual(comms_status._CommsStatus__hostapd_status.phy, 'phy0') + self.assertEqual(comms_status._CommsStatus__hostapd_status.freq, '2412') + self.assertEqual(comms_status._CommsStatus__hostapd_status.channel, '6') + self.assertEqual(comms_status._CommsStatus__hostapd_status.beacon_int, '100') + self.assertEqual(comms_status._CommsStatus__hostapd_status.ssid, 'test') + + @patch('subprocess.Popen') + def test_hostapd_cli_status_with_error(self, mock_popen): + mock_popen.return_value.communicate.return_value = ( + None, + b'Error from hostapd_cli' + ) + + comms_status = CommsStatus(logger=MagicMock(), index=0) + with self.assertRaises(RuntimeError): + comms_status._CommsStatus__get_hostapd_cli_status() +