From 19449a32e53f9e2d6f1dd0afeb7b3142d194dac3 Mon Sep 17 00:00:00 2001 From: Yevhenii Vaskivskyi Date: Thu, 21 Dec 2023 07:08:15 +0100 Subject: [PATCH] Test and refactor `onboarding` endpoint module (#425) --- asusrouter/modules/endpoint/onboarding.py | 77 +++--- tests/modules/endpoint/test_onboarding.py | 276 ++++++++++++++++++++++ 2 files changed, 320 insertions(+), 33 deletions(-) create mode 100644 tests/modules/endpoint/test_onboarding.py diff --git a/asusrouter/modules/endpoint/onboarding.py b/asusrouter/modules/endpoint/onboarding.py index b405985..9e1318a 100644 --- a/asusrouter/modules/endpoint/onboarding.py +++ b/asusrouter/modules/endpoint/onboarding.py @@ -1,6 +1,5 @@ """Onboarding endpoint module.""" - from __future__ import annotations from typing import Any @@ -20,9 +19,36 @@ } +CONST_AP = { + "2g": "2ghz", + "5g": "5ghz", + "5g1": "5ghz2", + "6g": "6ghz", + "dwb": "dwb", +} + +CONST_FREQ = [ + "2g", + "5g", + "6g", +] + + def read(content: str) -> dict[str, Any]: """Read onboarding data""" + # Preprocess the content + content = read_preprocess(content) + + # Read the json content + onboarding: dict[str, Any] = read_json_content(content) + + return onboarding + + +def read_preprocess(content: str) -> str: + """Preprocess the content before reading it.""" + # Prepare the content content = ( clean_content(content) @@ -35,10 +61,7 @@ def read(content: str) -> dict[str, Any]: # In case we have a trailing comma inside a dict content = content.replace(",}", "}") - # Read the json content - onboarding: dict[str, Any] = read_json_content(content) - - return onboarding + return content def process(data: dict[str, Any]) -> dict[AsusData, Any]: @@ -55,14 +78,14 @@ def process(data: dict[str, Any]) -> dict[AsusData, Any]: # Client list clients = {} - client_list = data["get_allclientlist"][0] + client_list = data.get("get_allclientlist", [{}])[0] for node in client_list: for connection in client_list[node]: convert = process_connection(connection) for mac in client_list[node][connection]: description = { - "connection_type": convert["connection_type"], - "guest": convert["guest"], + "connection_type": convert.get("connection_type"), + "guest": convert.get("guest"), "ip": safe_return( client_list[node][connection][mac].get("ip", None) ), @@ -81,33 +104,24 @@ def process(data: dict[str, Any]) -> dict[AsusData, Any]: def process_aimesh_node(data: dict[str, Any]) -> AiMeshDevice: """Process AiMesh node data.""" - const_ap = { - "2g": "2ghz", - "5g": "5ghz", - "5g1": "5ghz2", - "6g": "6ghz", - "dwb": "dwb", - } - - const_freq = [ - "2g", - "5g", - "6g", - ] - + # Get the list of WLAN APs ap = {} - for el, value in const_ap.items(): + for el, value in CONST_AP.items(): if f"ap{el}" in data and data[f"ap{el}"] is not str(): ap[value] = data[f"ap{el}"] + # Get the parent data parent = {} - for el in const_freq: + for el in CONST_FREQ: if f"pap{el}" in data and data[f"pap{el}"] is not str(): - parent["connection"] = const_ap[el] + parent["connection"] = CONST_AP[el] parent["mac"] = data[f"pap{el}"] parent["rssi"] = safe_return(data.get(f"rssi{el}")) parent["ssid"] = safe_return(data.get(f"pap{el}_ssid")) + # Stop the loop since we found the parent + break + level = safe_int(data.get("level", "0")) node_type = "router" if level == 0 else "node" @@ -138,11 +152,8 @@ def process_connection(data: str) -> dict[str, int]: if data == "wired_mac": return {"connection_type": 0, "guest": 0} - try: - temp = data.split("_") - return { - "connection_type": CONNECTION_TYPE.get(temp[0]) or 0, - "guest": int(temp[1]) if len(temp) > 1 else 0, - } - except Exception as ex: - raise ex + temp = data.split("_") + return { + "connection_type": CONNECTION_TYPE.get(temp[0]) or 0, + "guest": int(temp[1]) if len(temp) > 1 else 0, + } diff --git a/tests/modules/endpoint/test_onboarding.py b/tests/modules/endpoint/test_onboarding.py new file mode 100644 index 0000000..28b8264 --- /dev/null +++ b/tests/modules/endpoint/test_onboarding.py @@ -0,0 +1,276 @@ +"""Tests for the onboarding endpoint module.""" + +from unittest.mock import patch + +import pytest + +from asusrouter import AsusData +from asusrouter.modules.aimesh import AiMeshDevice +from asusrouter.modules.endpoint.onboarding import ( + CONNECTION_TYPE, + process, + process_aimesh_node, + process_connection, + read, + read_preprocess, +) + +CONTENT_RAW = ( + "get_onboardinglist = [{}][0];\n" + "get_cfg_clientlist = [[{}]][0];\n" + "get_onboardingstatus = [{}][0];\n" + "get_wclientlist = [{}][0];\n" + "get_wiredclientlist = [{}][0];\n" + "get_allclientlist = [{}][0];\n" + 'cfg_note = "1";\n' + 'cfg_obre = "";\n' + "\n" + "\n" +) + +CONTENT_PREPROCESSED = ( + '{"get_onboardinglist":[{}],' + '"get_cfg_clientlist":[[{}]],' + '"get_onboardingstatus":[{}],' + '"get_wclientlist":[{}],' + '"get_wiredclientlist":[{}],' + '"get_allclientlist":[{}],' + '"cfg_note":"1",' + '"cfg_obre":""}' +) + +CONTENT_READ = { + "get_onboardinglist": [{}], + "get_cfg_clientlist": [[{}]], + "get_onboardingstatus": [{}], + "get_wclientlist": [{}], + "get_wiredclientlist": [{}], + "get_allclientlist": [{}], + "cfg_note": "1", + "cfg_obre": "", +} + + +def test_read(): + """Test read function.""" + + with patch( + "asusrouter.modules.endpoint.onboarding.read_preprocess", + return_value=CONTENT_PREPROCESSED, + ) as mock_preprocess, patch( + "asusrouter.modules.endpoint.onboarding.read_json_content", + return_value=CONTENT_READ, + ) as mock_read_json_content: + # Get the result + result = read(CONTENT_RAW) + + # Check the result + assert result == CONTENT_READ + + # Check the calls + mock_preprocess.assert_called_once_with(CONTENT_RAW) + mock_read_json_content.assert_called_once_with(CONTENT_PREPROCESSED) + + +def test_read_preprocess(): + """Test read_preprocess function.""" + + # Get the result + result = read_preprocess(CONTENT_RAW) + + # Check the result + assert result == CONTENT_PREPROCESSED + + +DATA_CLIENTLIST = [[{"mac": "00:aa:11:bb:22:cc", "ip": "192.168.1.1"}]] +DATA_ALLCLIENTLIST = [ + { + "00:aa:11:bb:22:cc": { + "2G": {"00:aa:11:bb:22:dd": {"ip": "192.168.1.12", "rssi": -34}} + } + } +] + +RESULT_CLIENTLIST = { + "00:aa:11:bb:22:cc": AiMeshDevice( + status=False, + alias=None, + model=None, + product_id=None, + ip="192.168.1.1", + fw=None, + fw_new=None, + mac="00:aa:11:bb:22:cc", + ap={}, + parent={}, + type="router", + level=0, + config=None, + ), +} + +RESULT_ALLCLIENTLIST = { + "00:aa:11:bb:22:dd": { + "connection_type": None, + "guest": None, + "ip": "192.168.1.12", + "mac": "00:aa:11:bb:22:dd", + "node": "00:aa:11:bb:22:cc", + "online": True, + "rssi": -34, + } +} + + +@pytest.mark.parametrize( + "data_clientlist, data_allclientlist, result_clientlist, result_allclientlist", + [ + # Correct data + (DATA_CLIENTLIST, DATA_ALLCLIENTLIST, RESULT_CLIENTLIST, RESULT_ALLCLIENTLIST), + # AiMesh node data missing + (None, DATA_ALLCLIENTLIST, None, RESULT_ALLCLIENTLIST), + # Clients data missing + (DATA_CLIENTLIST, None, RESULT_CLIENTLIST, None), + # No data + (None, None, None, None), + ], +) +def test_process( + data_clientlist, data_allclientlist, result_clientlist, result_allclientlist +): + """Test process function.""" + + # Test data and expected result + data = {} + expected_result = {} + + # AiMesh nodes data + if data_clientlist is not None: + data["get_cfg_clientlist"] = data_clientlist + expected_result[AsusData.AIMESH] = result_clientlist or {} + + # Clients data + if data_allclientlist is not None: + data["get_allclientlist"] = data_allclientlist + expected_result[AsusData.CLIENTS] = result_allclientlist or {} + + # Define the side effects for the mocked functions + def mock_aimesh_node_side_effect(device): + """Mock the process_aimesh_node function.""" + + if result_clientlist and device.get("mac") in result_clientlist: + return result_clientlist[device.get("mac")] + return None + + # Patch the functions and set their return values + with patch( + "asusrouter.modules.endpoint.onboarding.process_aimesh_node", + side_effect=mock_aimesh_node_side_effect, + ) as mock_aimesh_node, patch( + "asusrouter.modules.endpoint.onboarding.process_connection", + return_value=result_allclientlist, + ) as mock_connection: + # Get the result + result = process(data) + + # Assert that the result is as expected + assert result == expected_result + + # Check that mock_aimesh_node was called (if data_clientlist was provided) + if data_clientlist is not None: + assert len(mock_aimesh_node.call_args_list) == len(data_clientlist[0]) + for i, call in enumerate(mock_aimesh_node.call_args_list): + assert call.args[0] == data_clientlist[0][i] + else: + mock_aimesh_node.assert_not_called() + + # Check that mock_connection was called appropriate number of times + if data_allclientlist is not None: + assert mock_connection.call_count == len(data_allclientlist) + else: + mock_connection.assert_not_called() + + +def test_process_aimesh_node(): + """Test process_aimesh_node function.""" + + # Test data and expected result + data = { + "ap2g": "ap2g_value", + "ap5g": "ap5g_value", + "ap5g1": "ap5g1_value", + "ap6g": "ap6g_value", + "apdwb": "apdwb_value", + "pap2g": "", + "pap5g": "pap5g_value", + "pap6g": "pap6g_value", + "rssi2g": "", + "rssi5g": "rssi5g_value", + "rssi6g": "rssi6g_value", + "pap2g_ssid": "", + "pap5g_ssid": "pap5g_ssid_value", + "pap6g_ssid": "pap6g_ssid_value", + "level": "1", + "online": "1", + "alias": "alias_value", + "ui_model_name": "ui_model_name_value", + "model_name": "model_name_value", + "product_id": "product_id_value", + "ip": "192.168.0.2", + "fwver": "fwver_value", + "newfwver": "newfwver_value", + "mac": "00:aa:11:bb:22:cc", + "config": {"config_key": "config_value"}, + } + expected_result = AiMeshDevice( + status=True, + alias="alias_value", + model="ui_model_name_value", + product_id="product_id_value", + ip="192.168.0.2", + fw="fwver_value", + fw_new="newfwver_value", + mac="00:aa:11:bb:22:cc", + ap={ + "2ghz": "ap2g_value", + "5ghz": "ap5g_value", + "5ghz2": "ap5g1_value", + "6ghz": "ap6g_value", + "dwb": "apdwb_value", + }, + parent={ + "connection": "5ghz", + "mac": "pap5g_value", + "rssi": "rssi5g_value", + "ssid": "pap5g_ssid_value", + }, + type="node", + level=1, + config={"config_key": "config_value"}, + ) + + # Get the result + result = process_aimesh_node(data) + + # Check the result + assert result == expected_result + + +@pytest.mark.parametrize( + "data, expected", + [ + # An empty string + ("", {}), + # Data for a wired connection + ("wired_mac", {"connection_type": 0, "guest": 0}), + # Data for a wireless guest connection + ("type_1", {"connection_type": CONNECTION_TYPE.get("type") or 0, "guest": 1}), + # Wrong data + (None, {}), + (123, {}), + ], +) +def test_process_connection(data, expected): + """Test process_connection function.""" + result = process_connection(data) + assert result == expected