""" Telecom Infra Project OpenWifi 2.X (Ucentral libraries for Test Automation) """ import importlib import json import random import string import time import allure import pytest import requests logging = importlib.import_module("logging") ap_lib = importlib.import_module("ap_lib") controller = importlib.import_module("controller") """ Custom Class Imports needed for OpenWifi 2.X """ ConfigureController = controller.ConfigureController Controller = controller.Controller FMSUtils = controller.FMSUtils ProvUtils = controller.ProvUtils UProfileUtility = controller.UProfileUtility APLIBS = ap_lib.APLIBS class dut_lib_template: """ Standard OpenWifi wlan-testing specific variables """ controller_data = {} device_under_tests_info = [] """ OpenWifi 2.x Specific Variables that will be only scoped in dut_lib_template Library """ ow_sec_url = "" ow_sec_login_username = "" ow_sec_login_password = "" target = "dut_lib_template" controller_library_object = object() prov_library_object = object() firmware_library_object = object() dut_library_object = object() supported_bands = ["2G", "5G", "6G", "5G-lower", "5G-upper"] supported_modes = ["BRIDGE", "NAT", "VLAN"] supported_encryption = ["open", "wpa", "wpa2_personal", "wpa3_personal", "wpa_wpa2_personal_mixed", "wpa3_personal_mixed", "wpa_enterprise", "wpa2_enterprise", "wpa3_enterprise", "wpa_wpa2_enterprise_mixed", "wpa3_enterprise_mixed", "wpa3_enterprise_192" ] dut_lib_specific_encryption_translation = {"open": "none", "wpa": "psk", "wpa2_personal": "psk2", "wpa3_personal": "sae", "wpa3_personal_mixed": "sae-mixed", "wpa_wpa2_personal_mixed": "psk-mixed", "wpa_enterprise": "wpa", "wpa2_enterprise": "wpa2", "wpa3_enterprise": "wpa3", "wpa_wpa2_enterprise_mixed": "wpa-mixed", "wpa3_enterprise_mixed": "wpa3-mixed", "wpa3_enterprise_192": "wpa3-192" } def __init__(self, controller_data=None, target=None, device_under_tests_info=[], logging_level=logging.INFO): logging.basicConfig(format='%(asctime)s - %(message)s', level=logging_level) if target != self.target: logging.error("Target version is : " + target + " Expected target is dut_lib_template") pytest.exit("Target should be 'dut_lib_template', Current Target is : " + target) if controller_data is None: controller_data = {} self.controller_data = controller_data self.device_under_tests_info = device_under_tests_info self.setup_metadata() self.setup_objects() self.setup_environment_properties() """ Controller and Access Point specific metadata that is related to OpenWifi 2.x """ def setup_metadata(self): logging.info( "Setting up the Controller metadata for dut_lib_template Library: " + str( json.dumps(self.controller_data, indent=2))) logging.info("Setting up the DUT metadata for dut_lib_template Library: " + str( json.dumps(self.device_under_tests_info, indent=2))) logging.info("Number of DUT's in lab_info.json: " + str(len(self.device_under_tests_info))) self.ow_sec_url = self.controller_data["url"] self.ow_sec_login_username = self.controller_data["username"] self.ow_sec_login_password = self.controller_data["password"] def setup_objects(self): try: self.controller_library_object = Controller(controller_data=self.controller_data) self.prov_library_object = ProvUtils(sdk_client=self.controller_library_object) self.firmware_library_object = FMSUtils(sdk_client=self.controller_library_object) except Exception as e: pytest.fail("Unable to setup Controller Objects") logging.error("Exception in setting up Controller objects:" + str(e)) try: self.dut_library_object = APLIBS(dut_data=self.device_under_tests_info) except Exception as e: logging.error("Exception in setting up Access Point Library object:" + str(e)) pytest.fail("Unable to setup AP Objects") def teardown_objects(self): self.controller_library_object.logout() """ Standard getter methods. Should be available for all type of libraries. Commonly used by wlan-testing""" def get_dut_library_object(self): return self.dut_library_object def get_controller_library_object(self): return self.controller_library_object def get_controller_data(self): return self.controller_data def get_device_under_tests_info(self): return self.device_under_tests_info def get_number_of_dut(self): return len(self.device_under_tests_info) def get_dut_logs(self, dut_idx=0): return self.dut_library_object.get_logs(idx=0) def get_controller_logs(self): pass def get_dut_max_clients(self): pass """ standard method for translating the configurations from pytest: no need to change this """ def setup_configuration_data(self, configuration=None, requested_combination=None): """Predefined function for getting configuration data for applied data""" c_data = configuration.copy() if c_data is None: pytest.exit("No Configuration Received") if requested_combination is None: pytest.exit("No requested_combination Received") rf_data = None if c_data.keys().__contains__("rf"): rf_data = c_data["rf"] # base_band_keys = ["2G", "5G", "6G", "5G-lower", "5G-upper"] base_dict = dict.fromkeys(self.supported_bands) for i in base_dict: base_dict[i] = [] for i in requested_combination: if i[0] in self.supported_bands: base_dict[i[0]].append(self.dut_lib_specific_encryption_translation[i[1]]) if i[1] in self.supported_bands: base_dict[i[1]].append((self.dut_lib_specific_encryption_translation[i[0]])) temp = [] for i in list(base_dict.values()): for j in i: temp.append(j) temp_conf = c_data["ssid_modes"].copy() for i in temp_conf: if self.dut_lib_specific_encryption_translation[i] not in temp: c_data["ssid_modes"].pop(i) temp_conf = c_data["ssid_modes"].copy() print(self.dut_lib_specific_encryption_translation) for i in temp_conf: for j in range(len(temp_conf[i])): for k in temp_conf[i][j]["appliedRadios"]: if self.dut_lib_specific_encryption_translation[i] not in base_dict[k]: c_data["ssid_modes"][i][j]["appliedRadios"].remove(k) if c_data["ssid_modes"][i][j]["appliedRadios"] == []: c_data["ssid_modes"][i][j] = {} # .popi.popitem()) # .popitem() for i in c_data["ssid_modes"]: c_data["ssid_modes"][i] = [x for x in c_data["ssid_modes"][i] if x != {}] for ssids in c_data["ssid_modes"]: for i in c_data["ssid_modes"][ssids]: if i is not {}: i["security"] = self.dut_lib_specific_encryption_translation[ssids] temp_conf = c_data.copy() for i in range(0, len(self.device_under_tests_info)): if c_data["mode"] not in self.device_under_tests_info[i]["supported_modes"]: pytest.skip(c_data["mode"] + " is not Supported by DUT") for enc in c_data["ssid_modes"]: for idx in c_data["ssid_modes"][enc]: check = all( item in self.device_under_tests_info[i]["supported_bands"] for item in idx["appliedRadios"]) if not check: temp_conf["ssid_modes"][enc].remove(idx) for key in c_data["rf"]: if key not in self.device_under_tests_info[i]["supported_bands"]: print(key) temp_conf["rf"][key] = None return temp_conf """ setup_basic_configuration - Method to configure AP in basic operating modes with multiple SSID's and multiple AP's This covers, basic and advanced test cases """ def setup_basic_configuration(self, configuration=None, requested_combination=None): f_conf = self.setup_configuration_data(configuration=configuration, requested_combination=requested_combination) logging.info("Selected Configuration: " + str(json.dumps(f_conf, indent=2))) final_configuration = f_conf.copy() """ Write your logic here to setup the configuration on dut """ # Setup Mode # Setup Radio Scenario # Setup Config Apply on all AP's ret_val = dict() for i in range(0, len(self.device_under_tests_info)): self.pre_apply_check(idx=i) # Do check AP before pushing the configuration # Write your logic here to push the configuration on DUT. self.post_apply_check(idx=i) # Do check AP after pushing the configuration ret_val[self.device_under_tests_info[i]["identifier"]] = {} # Write the Logic to create the data in below format. This should be returned from this function """ dut = { '903cb36c4301': {'ssid_data': {0: {'ssid': 'ssid_wpa_2g_br', 'encryption': 'wpa', 'password': 'something', 'band': '2G', 'bssid': '90:3C:B3:6C:43:04'} }, 'radio_data':{ '2G': {'channel': 6, 'bandwidth': 20, 'frequency': 2437}, '5G': {'channel': None, 'bandwidth': None, 'frequency': None}, '6G': {'channel': None, 'bandwidth': None, 'frequency': None} } } } """ return ret_val """ setup_special_configuration - Method to configure APs in mesh operating modes with multiple SSID's and multiple AP's This covers, mesh and other roaming scenarios which includes any special type of modes multiple AP's with WDS and Wifi Steering scenarios are also covered here """ def setup_special_configuration(self, configuration=None, requested_combination=None): final_configuration = self.setup_configuration_data(configuration=configuration, requested_combination=requested_combination) logging.info("Selected Configuration: " + str(json.dumps(final_configuration, indent=2))) profile_object = UProfileUtility(sdk_client=self.controller_library_object) if final_configuration["mode"] in self.supported_modes: profile_object.set_mode(mode=final_configuration["mode"]) else: pytest.skip(final_configuration["mode"] + " Mode is not supported") # Setup Radio Scenario if final_configuration["rf"] != {}: profile_object.set_radio_config(radio_config=final_configuration["rf"]) else: profile_object.set_radio_config() for ssid in final_configuration["ssid_modes"]: for ssid_data in final_configuration["ssid_modes"][ssid]: profile_object.add_ssid(ssid_data=ssid_data) logging.info( "Configuration That is getting pushed: " + json.dumps(profile_object.base_profile_config, indent=2)) r_val = False # Do check AP before pushing the configuration # TODO self.dut_library_object.check_serial_connection() """ serial connection check ubus call ucentral status save the current uuid uci show ucentral ifconfig wifi status start logger to collect ap logs before config apply Timestamp before doing config apply """ for dut in self.device_under_tests_info: resp = profile_object.push_config(serial_number=dut["identifier"]) logging.info("Response for Config apply: " + str(resp.status_code)) if resp.status_code != 200: logging.info("Failed to apply Configuration to AP. Response Code" + str(resp.status_code) + "Retrying in 5 Seconds... ") time.sleep(5) resp = profile_object.push_config(serial_number=dut["identifier"]) if resp.status_code != 200: logging.info("Failed to apply Config, Response code:" + str(resp.status_code)) pytest.fail("Failed to apply Config, Response code :" + str(resp.status_code)) if resp.status_code == 200: r_val = True # TODO """ serial connection check ubus call ucentral status save the current uuid and compare with the one before config apply save the active config and compare with the latest apply uci show ifconfig iwinfo wifi status start logger to collect ap logs before config apply Timestamp after doing config apply """ return r_val def get_dut_version(self): """ get_dut_version write your code to get ap version in self.dut_library_object.get_ap_version() """ version_info = [] for ap in range(len(self.device_under_tests_info)): version_info.append(self.dut_library_object.get_ap_version(idx=ap, print_log=True)) return version_info def get_controller_version(self): version_info = dict() # version_info["ow_fms"] = self.controller_library_object.get_sdk_version_fms() # version_info["ow_gw"] = self.controller_library_object.get_sdk_version_gw() # version_info["ow_sec"] = self.controller_library_object.get_sdk_version_sec() # version_info["ow_prov"] = self.controller_library_object.get_sdk_version_prov() # version_info["ow_rrm"] = self.controller_library_object.get_sdk_version_owrrm() # version_info["ow_analytics"] = self.controller_library_object.get_sdk_version_ow_analytics() # version_info["ow_sub"] = self.controller_library_object.get_sdk_version_owsub() return version_info def pre_apply_check(self, idx=0): """ Write your code that you need to use """ check = "PASSED" if check is "FAILED": pytest.fail("Write the message here for why precheck is failed") def post_apply_check(self, idx=0): check = "PASSED" if check is "FAILED": pytest.fail("Write the message here for why post apply is failed") def setup_environment_properties(self, add_allure_environment_property=None): # Write the logic to add data to environment data if add_allure_environment_property is None: return add_allure_environment_property('Cloud-Controller-SDK-URL', self.controller_data.get("url")) sdk_version_data = self.get_controller_version() for microservice in sdk_version_data: add_allure_environment_property(microservice + '-version', str(sdk_version_data.get(microservice))) dut_versions = self.get_dut_version() for i in range(len(self.device_under_tests_info)): add_allure_environment_property("Firmware-Version_" + self.device_under_tests_info[i]["identifier"], str(dut_versions[i])) for dut in self.device_under_tests_info: models = [] identifiers = [] models.append(dut["model"]) identifiers.append(dut["identifier"]) add_allure_environment_property('DUT-Model/s', ", ".join(models)) add_allure_environment_property('Serial-Number/s', ", ".join(identifiers)) def setup_firmware(self): # Write the logic to upgrade the firmware on DUT upgrade_status = [] return upgrade_status def simulate_radar(self, idx=0): """Simulate radar command for DFS""" ret = self.dut_library_object.dfs(idx=idx) return ret def get_dfs_logs(self, idx=0): """Get the ap logs after Simulate radar command""" logs = self.dut_library_object.dfs_logread(idx=idx) return logs def reboot(self, idx=0): """Reboot the AP""" ret = self.dut_library_object.reboot(idx=idx) return ret if __name__ == '__main__': basic_05 = { "target": "dut_lib_template", "controller": { "url": "https://sec-qa01.cicd.lab.wlan.tip.build:16001", "username": "tip@ucentral.com", "password": "OpenWifi%123" }, "device_under_tests": [{ "model": "cig_wf188n", "supported_bands": ["2G", "5G", "6G"], "supported_modes": ["BRIDGE", "NAT", "VLAN"], "wan_port": "1.1.eth2", "lan_port": "1.1.eth1", "ssid": { "mode": "BRIDGE", "2g-ssid": "OpenWifi", "5g-ssid": "OpenWifi", "6g-ssid": "OpenWifi", "2g-password": "OpenWifi", "5g-password": "OpenWifi", "6g-password": "OpenWifi", "2g-encryption": "WPA2", "5g-encryption": "WPA2", "6g-encryption": "WPA3", "2g-bssid": "68:7d:b4:5f:5c:31", "5g-bssid": "68:7d:b4:5f:5c:3c", "6g-bssid": "68:7d:b4:5f:5c:38" }, "mode": "wifi6", "identifier": "0000c1018812", # identifier is important and should be unique for each dut within the device_under_tests in single testbed "method": "serial", # serial | ssh | telnet "host_ip": "10.28.3.103", # if method == serial, then host_ip - ip for remote system where dut is connected via serial "host_username": "lanforge", "host_password": "pumpkin77", "host_ssh_port": 22, "serial_tty": "/dev/ttyAP1", # if method == serial, then tty port for this dut "firmware_version": "next-latest" }], "traffic_generator": { "name": "lanforge", "testbed": "basic", "scenario": "dhcp-bridge", # dhcp-bridge | dhcp-external "details": { "manager_ip": "10.28.3.28", "http_port": 8080, "ssh_port": 22, "setup": {"method": "build", "DB": "Test_Scenario_Automation"}, # method - build | load "wan_ports": { # addressing - dhcp-server | static | dynamic "1.1.eth2": {"addressing": "dhcp-server", "subnet": "172.16.0.1/16", "dhcp": { "lease-first": 10, "lease-count": 10000, "lease-time": "6h" } } }, "lan_ports": { "1.1.eth1": {"addressing": "dynamic"} }, "uplink_nat_ports": { "1.1.eth1": {"addressing": "static", "subnet": "10.28.2.16/24", "gateway_ip": "10.28.2.1", "ip_mask": "255.255.255.0" } } } } } var = dut_lib_template(controller_data=basic_05["controller"], device_under_tests_info=basic_05["device_under_tests"], target=basic_05["target"]) # var.setup_objects() setup_params_general_two = { "mode": "BRIDGE", "ssid_modes": { "wpa3_personal": [ {"ssid_name": "ssid_wpa3_p_2g_br", "appliedRadios": ["2G"], "security_key": "something"}, {"ssid_name": "ssid_wpa3_p_5g_br", "appliedRadios": ["5G"], "security_key": "something"}, {"ssid_name": "ssid_wpa3_p_6g_br", "appliedRadios": ["6G"], "security_key": "something"}], "wpa3_personal_mixed": [ {"ssid_name": "ssid_wpa3_p_m_2g_br", "appliedRadios": ["2G"], "security_key": "something"}, {"ssid_name": "ssid_wpa3_p_m_5g_br", "appliedRadios": ["5G"], "security_key": "something"}], "wpa_wpa2_personal_mixed": [ {"ssid_name": "ssid_wpa_wpa2_p_m_2g_br", "appliedRadios": ["2G"], "security_key": "something"}, {"ssid_name": "ssid_wpa_wpa2_p_m_5g_br", "appliedRadios": ["5G"], "security_key": "something"}] }, "rf": {}, "radius": False } x = setup_params_general_two.copy() target = [['6G', 'wpa3_personal']] d = var.setup_configuration_data(configuration=setup_params_general_two, requested_combination=target) # d = var.setup_basic_configuration(configuration=setup_params_general_two, requested_combination=target) print(x) # var.setup_firmware() # var.teardown_objects()