Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

UnitTests and Coverage reporting #444

Merged
merged 5 commits into from
May 17, 2024
Merged

UnitTests and Coverage reporting #444

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
e5c03fe
UnitTests and Coverage reporting
joenpera Apr 25, 2024
11ea1d3
Merge pull request #452 from tiiuae/develop
joenpera May 3, 2024
de6e3ad
Add settings/command Unittest
joenpera May 10, 2024
8c633fd
Review fixes
joenpera May 14, 2024
89f65e5
Fix review findings
joenpera May 15, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
35 changes: 35 additions & 0 deletions modules/sc-mesh-secure-deployment/src/nats/coverage_report.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,35 @@
Name Stmts Miss Cover Missing
-----------------------------------------------------------------
mdm_agent.py 442 307 31% 104-105, 144, 164-209, 233, 237-240, 248-251, 259-262, 268, 275-343, 356-357, 366-370, 380-475, 483-528, 538-573, 584-607, 628, 639-645, 661-697, 715-719, 725-739, 752-813, 824-875, 879-891
src/__init__.py 0 0 100%
src/bat_ctrl_utils.py 130 19 85% 40-41, 78-79, 135-136, 161-162, 173, 193-194, 224-225, 254-263, 277-279
src/cbma_adaptation.py 454 368 19% 83-84, 92-104, 114-183, 188-196, 199-223, 229-243, 246-253, 261-298, 301-314, 317-327, 330-333, 336-346, 349-356, 360-407, 411-427, 430-469, 477-490, 493-518, 525-539, 543-570, 585-618, 626-656, 666-677, 680-691, 702-730, 737-764, 776-838, 845-852, 859-873
src/cbma_paths.py 7 0 100%
src/comms_command.py 192 39 80% 88, 90, 94-102, 105, 107, 112, 117, 121, 142-144, 147-150, 155-157, 164, 175, 189-190, 235-240, 251, 262, 291, 309-310, 322, 336-337, 349, 373-375
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, 156-157, 172-192, 257-264, 320-323, 363, 370
src/comms_status.py 296 12 96% 70, 194, 225-227, 240-242, 258, 294, 302, 321-323
src/constants.py 44 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_bat_ctrl_utils.py 129 10 92% 232-233, 241-242, 245-246, 249-250, 253-254
tests/test_command.py 210 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 173 0 100%
tests/test_status.py 128 8 94% 28-35
tests/test_validation.py 146 0 100%
-----------------------------------------------------------------
TOTAL 3175 820 74%
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
35 changes: 30 additions & 5 deletions modules/sc-mesh-secure-deployment/src/nats/run_unittests_PC.sh
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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
Expand All @@ -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

155 changes: 0 additions & 155 deletions modules/sc-mesh-secure-deployment/src/nats/src/comms_command.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -117,20 +117,8 @@ 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.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]
else:
ret, info = "FAIL", "Command not supported"
return ret, info, data
Expand Down Expand Up @@ -209,14 +197,6 @@ 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",
)

return "OK", "Mesh settings revoked"

def __apply_mission_config(self) -> Tuple[str, str]:
Expand Down Expand Up @@ -370,42 +350,6 @@ 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 __debug(self, cc, param) -> Tuple[str, str, str]:
file = ""
try:
Expand All @@ -432,102 +376,3 @@ 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
59 changes: 38 additions & 21 deletions modules/sc-mesh-secure-deployment/src/nats/src/comms_settings.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,47 +53,64 @@ def validate_mesh_settings(self, index: int) -> (str, str):
self.logger.debug("validate mesh settings")

# pylint: disable=too-many-return-statements
if validation.validate_ssid(self.ssid[index]) is False:
if index > len(self.ssid) or \
validation.validate_ssid(self.ssid[index]) is False:
return "FAIL", "Invalid SSID"
self.logger.debug("validate mesh settings ssid ok")

if validation.validate_wpa3_psk(self.key[index]) is False:
if index > len(self.key) or \
validation.validate_wpa3_psk(self.key[index]) is False:
return "FAIL", "Invalid WPA3 PSK"
self.logger.debug("validate mesh settings wpa3 ok")

if validation.validate_mode(self.mode[index]) is False:
if index > len(self.mode) or \
validation.validate_mode(self.mode[index]) is False:
return "FAIL", "Invalid mode"
self.logger.debug("validate mesh settings mode ok")

if validation.validate_frequency(int(self.frequency[index])) is False:
return "FAIL", "Invalid frequency"
self.logger.debug("validate mesh settings freq ok")

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")
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:
pentestiing marked this conversation as resolved.
Show resolved Hide resolved
return "FAIL", "Invalid frequency (can't convert to integer)"

if validation.validate_country_code(self.country[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:
joenpera marked this conversation as resolved.
Show resolved Hide resolved
return "FAIL", "Invalid mcc frequency (can't convert to integer)"

if index > len(self.country) or \
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:
return "FAIL", "Invalid tx power"
self.logger.debug("validate mesh settings tx power ok")

if validation.validate_priority(self.priority[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:
pentestiing marked this conversation as resolved.
Show resolved Hide resolved
return "FAIL", "Invalid tx power (can't convert to integer)"

if index > len(self.priority) or \
validation.validate_priority(self.priority[index]) is False:
return "FAIL", "Invalid priority"
self.logger.debug("validate mesh settings priority ok")

if validation.validate_role(self.role) is False:
if index > len(self.role) or \
validation.validate_role(self.role) is False:
return "FAIL", "Invalid role"
self.logger.debug("validate mesh settings role ok")

if validation.validate_mptcp(self.mptcp[index]) is False:
if index > len(self.mptcp) or \
validation.validate_mptcp(self.mptcp[index]) is False:
return "FAIL", "Invalid mptcp value"
self.logger.debug("validate mesh settings mptcp ok")

if validation.validate_slaac(self.slaac[index]) is False:
if index > len(self.slaac) or \
validation.validate_slaac(self.slaac[index]) is False:
return "FAIL", "Invalid slaac ifaces"
self.logger.debug("validate mesh settings slaac ifaces ok")

Expand Down Expand Up @@ -300,9 +317,9 @@ def __save_settings(self, path: str, file: str, index: int) -> (str, str):

mesh_conf.write(f'id{str(index)}_SLAAC="{self.slaac[index]}"\n')

except:
except Exception as error:
self.comms_status[index].mesh_cfg_status = comms.STATUS.mesh_cfg_not_stored
self.logger.error("not able to write new %s", f"{str(index)}_{file}")
self.logger.error("not able to write new %s", f"{str(index)}_{file}, {error}")
return "FAIL", "not able to write new mesh.conf"

self.comms_status[index].mesh_cfg_status = comms.STATUS.mesh_cfg_stored
Expand Down
2 changes: 2 additions & 0 deletions modules/sc-mesh-secure-deployment/src/nats/src/comms_status.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Expand Down Expand Up @@ -539,3 +540,4 @@ def __get_mission_cfg_status(self):
self.__mesh_cfg_status,
self.__is_mission_cfg,
)

Loading
Loading