diff --git a/libs/controller/controller_2x/controller.py b/libs/controller/controller_2x/controller.py index 29f820989..77d2d1969 100644 --- a/libs/controller/controller_2x/controller.py +++ b/libs/controller/controller_2x/controller.py @@ -5,18 +5,14 @@ """ import datetime import json -import ssl import sys import time -from urllib.parse import urlparse -import pytest -import allure -import requests from operator import itemgetter -from pathlib import Path +from urllib.parse import urlparse -from requests.adapters import HTTPAdapter -import logging +import allure +import pytest +import requests # logging.basicConfig(level=logging.DEBUG) @@ -67,8 +63,6 @@ class ConfigureController: print(new_uri) return new_uri - - def request(self, service, command, method, params, payload): if service == "sec": uri = self.build_uri_sec(command) @@ -468,229 +462,227 @@ class FMSUtils: return "error" -class ProvUtils(ConfigureController): - def __init__(self, controller_data=None): - super().__init__(controller_data) +class ProvUtils: - def build_url_prov(self, path): - new_uri = 'https://%s:%d/api/v1/%s' % (self.prov_host.hostname, self.prov_host.port, path) - print(new_uri) - return new_uri + def __init__(self, sdk_client=None, controller_data=None): + if sdk_client is None: + self.sdk_client = Controller(controller_data=controller_data) + self.sdk_client = sdk_client def get_inventory(self): - uri = self.build_url_prov("inventory") + uri = self.sdk_client.build_url_prov("inventory") print(uri) - resp = requests.get(uri, headers=self.make_headers(), verify=False, timeout=100) - self.check_response("GET", resp, self.make_headers(), "", uri) + resp = requests.get(uri, headers=self.sdk_client.make_headers(), verify=False, timeout=100) + self.sdk_client.check_response("GET", resp, self.sdk_client.make_headers(), "", uri) return resp def get_inventory_by_device(self, device_name): - uri = self.build_url_prov("inventory/" + device_name) + uri = self.sdk_client.build_url_prov("inventory/" + device_name) print(uri) - resp = requests.get(uri, headers=self.make_headers(), verify=False, timeout=100) - self.check_response("GET", resp, self.make_headers(), "", uri) + resp = requests.get(uri, headers=self.sdk_client.make_headers(), verify=False, timeout=100) + self.sdk_client.check_response("GET", resp, self.sdk_client.make_headers(), "", uri) return resp def get_system_prov(self): - uri = self.build_url_prov("system?command=info") + uri = self.sdk_client.build_url_prov("system?command=info") allure.attach(name="Url of Prov UI:", body=str(uri)) - resp = requests.get(uri, headers=self.make_headers(), verify=False, timeout=100) - self.check_response("GET", resp, self.make_headers(), "", uri) + resp = requests.get(uri, headers=self.sdk_client.make_headers(), verify=False, timeout=100) + self.sdk_client.check_response("GET", resp, self.sdk_client.make_headers(), "", uri) return resp def add_device_to_inventory(self, device_name, payload): - uri = self.build_url_prov("inventory/" + device_name) + uri = self.sdk_client.build_url_prov("inventory/" + device_name) print(uri) print(payload) payload = json.dumps(payload) - resp = requests.post(uri, data=payload, headers=self.make_headers(), verify=False, timeout=100) + resp = requests.post(uri, data=payload, headers=self.sdk_client.make_headers(), verify=False, timeout=100) print(resp) - self.check_response("POST", resp, self.make_headers(), payload, uri) + self.sdk_client.check_response("POST", resp, self.sdk_client.make_headers(), payload, uri) return resp def delete_device_from_inventory(self, device_name): - uri = self.build_url_prov("inventory/" + device_name) + uri = self.sdk_client.build_url_prov("inventory/" + device_name) print(uri) - resp = requests.delete(uri, headers=self.make_headers(), verify=False, timeout=100) - self.check_response("DELETE", resp, self.make_headers(), "", uri) + resp = requests.delete(uri, headers=self.sdk_client.make_headers(), verify=False, timeout=100) + self.sdk_client.check_response("DELETE", resp, self.sdk_client.make_headers(), "", uri) return resp def get_entity(self): - uri = self.build_url_prov("entity") + uri = self.sdk_client.build_url_prov("entity") print(uri) - resp = requests.get(uri, headers=self.make_headers(), verify=False, timeout=100) - self.check_response("GET", resp, self.make_headers(), "", uri) + resp = requests.get(uri, headers=self.sdk_client.make_headers(), verify=False, timeout=100) + self.sdk_client.check_response("GET", resp, self.sdk_client.make_headers(), "", uri) return resp - def get_entity_by_id(self,entity_id): - uri = self.build_url_prov("entity/" + entity_id) + def get_entity_by_id(self, entity_id): + uri = self.sdk_client.build_url_prov("entity/" + entity_id) print(uri) - resp = requests.get(uri, headers=self.make_headers(), verify=False, timeout=100) - self.check_response("GET", resp, self.make_headers(), "", uri) + resp = requests.get(uri, headers=self.sdk_client.make_headers(), verify=False, timeout=100) + self.sdk_client.check_response("GET", resp, self.sdk_client.make_headers(), "", uri) return resp def add_entity(self, payload): - uri = self.build_url_prov("entity/1") + uri = self.sdk_client.build_url_prov("entity/1") print(uri) print(payload) payload = json.dumps(payload) - resp = requests.post(uri, data=payload, headers=self.make_headers(), verify=False, timeout=100) + resp = requests.post(uri, data=payload, headers=self.sdk_client.make_headers(), verify=False, timeout=100) print(resp) - self.check_response("POST", resp, self.make_headers(), payload, uri) + self.sdk_client.check_response("POST", resp, self.sdk_client.make_headers(), payload, uri) return resp def delete_entity(self, entity_id): - uri = self.build_url_prov("entity/" + entity_id) + uri = self.sdk_client.build_url_prov("entity/" + entity_id) print(uri) - resp = requests.delete(uri, headers=self.make_headers(), verify=False, timeout=100) - self.check_response("DELETE", resp, self.make_headers(), "", uri) + resp = requests.delete(uri, headers=self.sdk_client.make_headers(), verify=False, timeout=100) + self.sdk_client.check_response("DELETE", resp, self.sdk_client.make_headers(), "", uri) return resp def edit_device_from_inventory(self, device_name, payload): - uri = self.build_url_prov("inventory/" + device_name) + uri = self.sdk_client.build_url_prov("inventory/" + device_name) print(uri) print(payload) payload = json.dumps(payload) - resp = requests.put(uri, data=payload, headers=self.make_headers(), verify=False, timeout=100) + resp = requests.put(uri, data=payload, headers=self.sdk_client.make_headers(), verify=False, timeout=100) print(resp) - self.check_response("PUT", resp, self.make_headers(), payload, uri) + self.sdk_client.check_response("PUT", resp, self.sdk_client.make_headers(), payload, uri) return resp def edit_entity(self, payload, entity_id): - uri = self.build_url_prov("entity/" + entity_id) + uri = self.sdk_client.build_url_prov("entity/" + entity_id) print(uri) print(payload) payload = json.dumps(payload) - resp = requests.put(uri, data=payload, headers=self.make_headers(), verify=False, timeout=100) + resp = requests.put(uri, data=payload, headers=self.sdk_client.make_headers(), verify=False, timeout=100) print(resp) - self.check_response("PUT", resp, self.make_headers(), payload, uri) + self.sdk_client.check_response("PUT", resp, self.sdk_client.make_headers(), payload, uri) return resp def get_contact(self): - uri = self.build_url_prov("contact") + uri = self.sdk_client.build_url_prov("contact") print(uri) - resp = requests.get(uri, headers=self.make_headers(), verify=False, timeout=100) - self.check_response("GET", resp, self.make_headers(), "", uri) + resp = requests.get(uri, headers=self.sdk_client.make_headers(), verify=False, timeout=100) + self.sdk_client.check_response("GET", resp, self.sdk_client.make_headers(), "", uri) return resp def get_contact_by_id(self, contact_id): - uri = self.build_url_prov("contact/" + contact_id) + uri = self.sdk_client.build_url_prov("contact/" + contact_id) print(uri) - resp = requests.get(uri, headers=self.make_headers(), verify=False, timeout=100) - self.check_response("GET", resp, self.make_headers(), "", uri) + resp = requests.get(uri, headers=self.sdk_client.make_headers(), verify=False, timeout=100) + self.sdk_client.check_response("GET", resp, self.sdk_client.make_headers(), "", uri) return resp def add_contact(self, payload): - uri = self.build_url_prov("contact/1") + uri = self.sdk_client.build_url_prov("contact/1") print(uri) print(payload) payload = json.dumps(payload) - resp = requests.post(uri, data=payload, headers=self.make_headers(), verify=False, timeout=100) + resp = requests.post(uri, data=payload, headers=self.sdk_client.make_headers(), verify=False, timeout=100) print(resp) - self.check_response("POST", resp, self.make_headers(), payload, uri) + self.sdk_client.check_response("POST", resp, self.sdk_client.make_headers(), payload, uri) return resp def delete_contact(self, contact_id): - uri = self.build_url_prov("contact/" + contact_id) + uri = self.sdk_client.build_url_prov("contact/" + contact_id) print(uri) - resp = requests.delete(uri, headers=self.make_headers(), verify=False, timeout=100) - self.check_response("DELETE", resp, self.make_headers(), "", uri) + resp = requests.delete(uri, headers=self.sdk_client.make_headers(), verify=False, timeout=100) + self.sdk_client.check_response("DELETE", resp, self.sdk_client.make_headers(), "", uri) return resp def edit_contact(self, payload, contact_id): - uri = self.build_url_prov("contact/" + contact_id) + uri = self.sdk_client.build_url_prov("contact/" + contact_id) print(uri) print(payload) payload = json.dumps(payload) - resp = requests.put(uri, data=payload, headers=self.make_headers(), verify=False, timeout=100) + resp = requests.put(uri, data=payload, headers=self.sdk_client.make_headers(), verify=False, timeout=100) print(resp) - self.check_response("PUT", resp, self.make_headers(), payload, uri) + self.sdk_client.check_response("PUT", resp, self.sdk_client.make_headers(), payload, uri) return resp def get_location(self): - uri = self.build_url_prov("location") + uri = self.sdk_client.build_url_prov("location") print(uri) - resp = requests.get(uri, headers=self.make_headers(), verify=False, timeout=100) - self.check_response("GET", resp, self.make_headers(), "", uri) + resp = requests.get(uri, headers=self.sdk_client.make_headers(), verify=False, timeout=100) + self.sdk_client.check_response("GET", resp, self.sdk_client.make_headers(), "", uri) return resp def get_location_by_id(self, location_id): - uri = self.build_url_prov("location/" + location_id) + uri = self.sdk_client.build_url_prov("location/" + location_id) print(uri) - resp = requests.get(uri, headers=self.make_headers(), verify=False, timeout=100) - self.check_response("GET", resp, self.make_headers(), "", uri) + resp = requests.get(uri, headers=self.sdk_client.make_headers(), verify=False, timeout=100) + self.sdk_client.check_response("GET", resp, self.sdk_client.make_headers(), "", uri) return resp def add_location(self, payload): - uri = self.build_url_prov("location/1") + uri = self.sdk_client.build_url_prov("location/1") print(uri) print(payload) payload = json.dumps(payload) - resp = requests.post(uri, data=payload, headers=self.make_headers(), verify=False, timeout=100) + resp = requests.post(uri, data=payload, headers=self.sdk_client.make_headers(), verify=False, timeout=100) print(resp) - self.check_response("POST", resp, self.make_headers(), payload, uri) + self.sdk_client.check_response("POST", resp, self.sdk_client.make_headers(), payload, uri) return resp def delete_location(self, location_id): - uri = self.build_url_prov("location/" + location_id) + uri = self.sdk_client.build_url_prov("location/" + location_id) print(uri) - resp = requests.delete(uri, headers=self.make_headers(), verify=False, timeout=100) - self.check_response("DELETE", resp, self.make_headers(), "", uri) + resp = requests.delete(uri, headers=self.sdk_client.make_headers(), verify=False, timeout=100) + self.sdk_client.check_response("DELETE", resp, self.sdk_client.make_headers(), "", uri) return resp def edit_location(self, payload, location_id): - uri = self.build_url_prov("location/" + location_id) + uri = self.sdk_client.build_url_prov("location/" + location_id) print(uri) print(payload) payload = json.dumps(payload) - resp = requests.put(uri, data=payload, headers=self.make_headers(), verify=False, timeout=100) + resp = requests.put(uri, data=payload, headers=self.sdk_client.make_headers(), verify=False, timeout=100) print(resp) - self.check_response("PUT", resp, self.make_headers(), payload, uri) + self.sdk_client.check_response("PUT", resp, self.sdk_client.make_headers(), payload, uri) return resp - def get_venue(self): - uri = self.build_url_prov("venue") + uri = self.sdk_client.build_url_prov("venue") print(uri) - resp = requests.get(uri, headers=self.make_headers(), verify=False, timeout=100) - self.check_response("GET", resp, self.make_headers(), "", uri) + resp = requests.get(uri, headers=self.sdk_client.make_headers(), verify=False, timeout=100) + self.sdk_client.check_response("GET", resp, self.sdk_client.make_headers(), "", uri) return resp def get_venue_by_id(self, venue_id): - uri = self.build_url_prov("venue/" + venue_id) + uri = self.sdk_client.build_url_prov("venue/" + venue_id) print(uri) - resp = requests.get(uri, headers=self.make_headers(), verify=False, timeout=100) - self.check_response("GET", resp, self.make_headers(), "", uri) + resp = requests.get(uri, headers=self.sdk_client.make_headers(), verify=False, timeout=100) + self.sdk_client.check_response("GET", resp, self.sdk_client.make_headers(), "", uri) return resp def add_venue(self, payload): - uri = self.build_url_prov("venue/0") + uri = self.sdk_client.build_url_prov("venue/0") print(uri) print(payload) payload = json.dumps(payload) - resp = requests.post(uri, data=payload, headers=self.make_headers(), verify=False, timeout=100) + resp = requests.post(uri, data=payload, headers=self.sdk_client.make_headers(), verify=False, timeout=100) print(resp) - self.check_response("POST", resp, self.make_headers(), payload, uri) + self.sdk_client.check_response("POST", resp, self.sdk_client.make_headers(), payload, uri) return resp def delete_venue(self, venue_id): - uri = self.build_url_prov("venue/" + venue_id) + uri = self.sdk_client.build_url_prov("venue/" + venue_id) print(uri) - resp = requests.delete(uri, headers=self.make_headers(), verify=False, timeout=100) - self.check_response("DELETE", resp, self.make_headers(), "", uri) + resp = requests.delete(uri, headers=self.sdk_client.make_headers(), verify=False, timeout=100) + self.sdk_client.check_response("DELETE", resp, self.sdk_client.make_headers(), "", uri) return resp def edit_venue(self, payload, venue_id): - uri = self.build_url_prov("venue/" + venue_id) + uri = self.sdk_client.build_url_prov("venue/" + venue_id) print(uri) print(payload) payload = json.dumps(payload) - resp = requests.put(uri, data=payload, headers=self.make_headers(), verify=False, timeout=100) + resp = requests.put(uri, data=payload, headers=self.sdk_client.make_headers(), verify=False, timeout=100) print(resp) - self.check_response("PUT", resp, self.make_headers(), payload, uri) + self.sdk_client.check_response("PUT", resp, self.sdk_client.make_headers(), payload, uri) return resp + class UProfileUtility: def __init__(self, sdk_client=None, controller_data=None): @@ -900,7 +892,6 @@ class UProfileUtility: # for keys in radio_config[band]: # base_radio_config_6g[keys] = radio_config[band][keys] - self.base_profile_config["radios"].append(base_radio_config_2g) self.base_profile_config["radios"].append(base_radio_config_5g) print(self.base_profile_config) @@ -1053,6 +1044,8 @@ if __name__ == '__main__': 'password': 'OpenWifi%123', } obj = Controller(controller_data=controller) + po = ProvUtils(sdk_client=obj) + print(po.get_inventory()) # up = UProfileUtility(sdk_client=obj, controller_data=controller) # up.set_mode(mode="BRIDGE") # up.set_radio_config() diff --git a/tests/conftest.py b/tests/conftest.py index 6f40d6557..67681060c 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -94,6 +94,12 @@ def pytest_addoption(parser): default=False, help="skip updating firmware on the AP (useful for local testing)" ) + parser.addoption( + "--skip-env", + action="store_true", + default=False, + help="skip adding to env data" + ) parser.addoption( "--skip-lanforge", @@ -239,12 +245,14 @@ def testbed(request): allure.attach(name="testbed name", body=var) yield var + @pytest.fixture(scope="session") def device(request): """yields the device option selection""" var = request.config.getoption("--device") yield var + @pytest.fixture(scope="session") def should_upload_firmware(request): """yields the --force-upload option for firmware upload selection""" @@ -257,12 +265,14 @@ def run_lf(request): var = request.config.getoption("--run-lf") yield var + @pytest.fixture(scope="session") def cc_1(request): """yields the --cc.1 option for skipping configuration on AP and using Cloud controller of available framework""" var = request.config.getoption("--cc.1") yield var + @pytest.fixture(scope="session") def roaming_delay(request): """yields the --roaming_delay option """ @@ -270,6 +280,7 @@ def roaming_delay(request): allure.attach(name="roaming delay provided in seconds", body=str(var)) yield var + @pytest.fixture(scope="session") def iteration(request): """yields the --iteration option for a test to provide how frequenty roam should happen """ @@ -277,6 +288,7 @@ def iteration(request): allure.attach(name="iteration", body=var) yield var + @pytest.fixture(scope="session") def duration(request): """yields the --duration option for a test to provide how long roam should happen """ @@ -284,6 +296,7 @@ def duration(request): allure.attach(name="duration in minutes", body=str(var)) yield var + @pytest.fixture(scope="session") def client(request): """yields the --client option for getting user specified client number""" @@ -291,6 +304,7 @@ def client(request): allure.attach(name="number of clients", body=var) yield var + @pytest.fixture(scope="session") def skip_pcap(request): """yields the --skip-pcap option for skipping the packet capture for sanity""" @@ -347,6 +361,7 @@ def get_configuration(testbed, request): LOGGER.info("Selected the lab Info data: " + str((CONFIGURATION[testbed]))) yield CONFIGURATION[testbed] + @pytest.fixture(scope="session") def get_device_configuration(device, request): """yields the selected device information from lab info file (configuration.py)""" @@ -391,6 +406,7 @@ def get_uci_show(fixtures_ver, get_apnos, get_configuration): uci_show = fixtures_ver.get_uci_show(get_apnos, get_configuration) yield uci_show + @pytest.fixture(scope="session") def get_ap_version(fixtures_ver, get_apnos, get_configuration, cc_1): if not cc_1: @@ -399,6 +415,7 @@ def get_ap_version(fixtures_ver, get_apnos, get_configuration, cc_1): else: yield True + @pytest.fixture(scope="session") def skip_lf(request): yield request.config.getoption("--skip-lanforge") @@ -418,6 +435,7 @@ def setup_controller(request, get_configuration, add_env_properties, fixtures_ve request.addfinalizer(fixtures_ver.disconnect) yield sdk_client + # Prov Controller Fixture @pytest.fixture(scope="session") def setup_prov_controller(request, get_configuration, add_env_properties, fixtures_ver): @@ -719,18 +737,19 @@ def lf_tools(get_configuration, testbed, skip_lf, run_lf, get_ap_version, cc_1): """ Create a DUT on LANforge""" if not skip_lf: obj = ChamberView(lanforge_data=get_configuration["traffic_generator"]["details"], - testbed=testbed, run_lf=run_lf, access_point_data=get_configuration["access_point"], cc_1=cc_1, ap_version=get_ap_version) + testbed=testbed, run_lf=run_lf, access_point_data=get_configuration["access_point"], + cc_1=cc_1, ap_version=get_ap_version) else: obj = False yield obj + @pytest.fixture(scope="session") def lf_reports(): obj = Report() yield obj - @pytest.fixture(scope="session") def lf_test(get_configuration, setup_influx, request, skip_lf, run_lf, skip_pcap): if not skip_lf: @@ -784,18 +803,22 @@ def add_allure_environment_property(request: SubRequest) -> Optional[Callable]: @fixture(scope='session') -def add_env_properties(get_configuration, get_sdk_version, get_apnos, fixtures_ver, cc_1, +def add_env_properties(request, get_configuration, get_sdk_version, get_apnos, fixtures_ver, cc_1, add_allure_environment_property: Callable) -> None: + if request.config.getoption("--skip-env"): + add_allure_environment_property('Cloud-Controller-SDK-URL', get_configuration["controller"]["url"]) + return if cc_1: for i in range(len(get_configuration["access_point"])): - add_allure_environment_property(str('Access-Point-Model'+ str(i+1)), get_configuration["access_point"][i]["model"]) + add_allure_environment_property(str('Access-Point-Model' + str(i + 1)), + get_configuration["access_point"][i]["model"]) else: add_allure_environment_property('Access-Point-Model', get_configuration["access_point"][0]["model"]) add_allure_environment_property('SDK-Version', get_sdk_version) try: if not cc_1: add_allure_environment_property('Access-Point-Firmware-Version', - fixtures_ver.get_ap_version(get_apnos, get_configuration)[0].split("\n")[1]) + fixtures_ver.get_ap_version(get_apnos, get_configuration)[0].split("\n")[1]) except Exception as e: print(e) pass @@ -930,6 +953,7 @@ def get_ap_config_slots(get_configuration): # print(slot) allure.attach(name="ap_slots", body=str(slot)) + @pytest.fixture(scope="session") def get_apnos_max_clients(get_apnos, get_configuration): all_logs = [] @@ -943,6 +967,7 @@ def get_apnos_max_clients(get_apnos, get_configuration): pass yield all_logs + @pytest.fixture(scope="function") def get_ap_channel(get_apnos, get_configuration): all_data = [] @@ -975,30 +1000,31 @@ def get_ap_channel(get_apnos, get_configuration): print(all_data) yield all_data + @pytest.fixture(scope="function") def disable_band5ghz(get_configuration): obj = CController(controller_data=get_configuration['controller'], ap_data=get_configuration['access_point']) - shut= obj.ap_5ghz_shutdown() + shut = obj.ap_5ghz_shutdown() print(shut) + @pytest.fixture(scope="function") def disable_band2ghz(get_configuration): obj = CController(controller_data=get_configuration['controller'], ap_data=get_configuration['access_point']) shut = obj.ap_2ghz_shutdown() print(shut) + @pytest.fixture(scope="function") def disable_band6ghz(get_configuration): obj = CController(controller_data=get_configuration['controller'], ap_data=get_configuration['access_point']) shut = obj.ap_6ghz_shutdown() print(shut) + @pytest.fixture(scope="function") def enable_all_bands(get_configuration): obj = CController(controller_data=get_configuration['controller'], ap_data=get_configuration['access_point']) obj.no_ap_5ghz_shutdown() obj.no_ap_2ghz_shutdown() obj.no_ap_6ghz_shutdown() - - - diff --git a/tests/controller_tests/ucentral_gateway/test_commands.py b/tests/controller_tests/ucentral_gateway/test_commands.py deleted file mode 100644 index e69de29bb..000000000 diff --git a/tests/controller_tests/ucentral_gateway/test_devices.py b/tests/controller_tests/ucentral_gateway/test_devices.py deleted file mode 100644 index 2836e54b0..000000000 --- a/tests/controller_tests/ucentral_gateway/test_devices.py +++ /dev/null @@ -1,29 +0,0 @@ -""" - - UCI Rest API Tests: Test Devices API - -""" -import pytest - - -class TestUCIAUTHDEAUTH(object): - """ - pytest -m "uci_login or uci_logout" --ucentral - """ - - @pytest.mark.uci_login - def test_get_all_devices(self, setup_controller): - """ - pytest -m "uci_login" --ucentral - """ - print(setup_controller.login_resp) - assert setup_controller.login_resp.status_code == 200 - - @pytest.mark.uci_logout - def test_get_device_by_serial(self, setup_controller): - """ - pytest -m "uci_logout" --ucentral - """ - resp = setup_controller.logout() - print(resp) - assert resp.status_code == 200 diff --git a/tests/controller_tests/ucentral_gateway/test_fms_service.py b/tests/controller_tests/ucentral_gateway/test_fms_service.py index f640398c5..696395816 100644 --- a/tests/controller_tests/ucentral_gateway/test_fms_service.py +++ b/tests/controller_tests/ucentral_gateway/test_fms_service.py @@ -4,16 +4,14 @@ """ -import string -import random - -import pytest -import json import allure +import pytest @pytest.mark.uc_sanity -@allure.feature("SDK REST API") +@pytest.mark.ow_sdk_load_tests +@allure.parent_suite("OpenWifi SDK Tests") +@allure.parent_suite("OpenWifi FMS Service Tests") class TestUcentralFMSService(object): @pytest.mark.system_info_fms diff --git a/tests/controller_tests/ucentral_gateway/test_gatewayservice.py b/tests/controller_tests/ucentral_gateway/test_gatewayservice.py index c953a6174..5e6aac065 100644 --- a/tests/controller_tests/ucentral_gateway/test_gatewayservice.py +++ b/tests/controller_tests/ucentral_gateway/test_gatewayservice.py @@ -4,118 +4,120 @@ """ -import string +import json import random -import pytest -import json import allure +import pytest @pytest.mark.uc_sanity +@pytest.mark.ow_sdk_tests @pytest.mark.ow_sanity_lf +@pytest.mark.ow_sdk_load_tests @pytest.mark.owgw_api_tests -@allure.feature("SDK REST API") +@allure.parent_suite("OpenWifi SDK Tests") +@allure.parent_suite("OpenWifi Gateway Service Tests") class TestUcentralGatewayService(object): """ """ configuration = { - "uuid": 1, - "radios": [ - { - "band": "5G", - "country": "CA", - "channel-mode": "HE", - "channel-width": 80 - } - ], + "uuid": 1, + "radios": [ + { + "band": "5G", + "country": "CA", + "channel-mode": "HE", + "channel-width": 80 + } + ], - "interfaces": [ - { - "name": "WAN", - "role": "upstream", - "services": [ "lldp" ], - "ethernet": [ - { - "select-ports": [ - "WAN*" - ] - } - ], - "ipv4": { - "addressing": "dynamic" - }, - "ssids": [ - { - "name": "OpenWifi", - "wifi-bands": [ - "5G" - ], - "bss-mode": "ap", - "encryption": { - "proto": "psk2", - "key": "OpenWifi", - "ieee80211w": "optional" - } - } - ] - }, - { - "name": "LAN", - "role": "downstream", - "services": [ "ssh", "lldp" ], - "ethernet": [ - { - "select-ports": [ - "LAN*" - ] - } - ], - "ipv4": { - "addressing": "static", - "subnet": "192.168.1.1/24", - "dhcp": { - "lease-first": 10, - "lease-count": 100, - "lease-time": "6h" - } - }, - "ssids": [ - { - "name": "OpenWifi", - "wifi-bands": [ - "5G" - ], - "bss-mode": "ap", - "encryption": { - "proto": "psk2", - "key": "OpenWifi", - "ieee80211w": "optional" - } - } - ] - - } - ], - "metrics": { - "statistics": { - "interval": 120, - "types": [ "ssids", "lldp", "clients" ] - }, - "health": { - "interval": 120 - } - }, - "services": { - "lldp": { - "describe": "2.x", - "location": "universe" - }, - "ssh": { - "port": 22 - } - } + "interfaces": [ + { + "name": "WAN", + "role": "upstream", + "services": ["lldp"], + "ethernet": [ + { + "select-ports": [ + "WAN*" + ] + } + ], + "ipv4": { + "addressing": "dynamic" + }, + "ssids": [ + { + "name": "OpenWifi", + "wifi-bands": [ + "5G" + ], + "bss-mode": "ap", + "encryption": { + "proto": "psk2", + "key": "OpenWifi", + "ieee80211w": "optional" } + } + ] + }, + { + "name": "LAN", + "role": "downstream", + "services": ["ssh", "lldp"], + "ethernet": [ + { + "select-ports": [ + "LAN*" + ] + } + ], + "ipv4": { + "addressing": "static", + "subnet": "192.168.1.1/24", + "dhcp": { + "lease-first": 10, + "lease-count": 100, + "lease-time": "6h" + } + }, + "ssids": [ + { + "name": "OpenWifi", + "wifi-bands": [ + "5G" + ], + "bss-mode": "ap", + "encryption": { + "proto": "psk2", + "key": "OpenWifi", + "ieee80211w": "optional" + } + } + ] + + } + ], + "metrics": { + "statistics": { + "interval": 120, + "types": ["ssids", "lldp", "clients"] + }, + "health": { + "interval": 120 + } + }, + "services": { + "lldp": { + "describe": "2.x", + "location": "universe" + }, + "ssh": { + "port": 22 + } + } + } @pytest.mark.sdk_restapi def test_gwservice_listdevices(self, setup_controller): @@ -136,8 +138,8 @@ class TestUcentralGatewayService(object): WIFI-3453 """ device_mac = "02:00:00:%02x:%02x:%02x" % (random.randint(0, 255), - random.randint(0, 255), - random.randint(0, 255)) + random.randint(0, 255), + random.randint(0, 255)) device_name = device_mac.replace(":", "") # device_name = "deadbeef0011" + testbed.replace("-","") payload = {'serialNumber': device_name, @@ -165,13 +167,13 @@ class TestUcentralGatewayService(object): assert False editing_payload = { - "id": device_name, - "notes": [ - { - "note": "Testing through Automation" - } - ] - } + "id": device_name, + "notes": [ + { + "note": "Testing through Automation" + } + ] + } print(json.dumps(editing_payload)) resp = setup_controller.edit_device_on_gw(device_name, editing_payload) allure.attach(name="response: ", body=str(resp.json())) @@ -194,19 +196,19 @@ class TestUcentralGatewayService(object): if resp.status_code != 200: assert False - @pytest.mark.system_info_gw def test_system_info_gw(self, setup_controller): system_info = setup_controller.get_system_gw() print(system_info.json()) - allure.attach(name="system info", body=str(system_info.json()),attachment_type=allure.attachment_type.JSON) + allure.attach(name="system info", body=str(system_info.json()), attachment_type=allure.attachment_type.JSON) assert system_info.status_code == 200 @pytest.mark.gw_commands def test_gw_commands(self, setup_controller): system_info = setup_controller.get_commands() print(system_info.json()) - allure.attach(name="Gateway list of commands", body=str(system_info.json()), attachment_type=allure.attachment_type.JSON) + allure.attach(name="Gateway list of commands", body=str(system_info.json()), + attachment_type=allure.attachment_type.JSON) assert system_info.status_code == 200 @pytest.mark.gw_device_logs @@ -217,7 +219,7 @@ class TestUcentralGatewayService(object): device_name = get_configuration['access_point'][0]['serial'] resp = setup_controller.get_device_logs(device_name) print(resp.json()) - allure.attach(name="Device Logs", body=str(resp.json()),attachment_type=allure.attachment_type.JSON) + allure.attach(name="Device Logs", body=str(resp.json()), attachment_type=allure.attachment_type.JSON) assert resp.status_code == 200 @pytest.mark.gw_device_health_checks @@ -270,8 +272,8 @@ class TestUcentralGatewayService(object): """ device_name = get_configuration['access_point'][0]['serial'] payload = { - "serialNumber": device_name - } + "serialNumber": device_name + } print(json.dumps(payload)) resp = setup_controller.ping_device(device_name, payload) print(resp.json()) @@ -284,15 +286,16 @@ class TestUcentralGatewayService(object): """ device_name = get_configuration['access_point'][0]['serial'] payload = { - "serialNumber": device_name, - "when": 0, - "duration": 1, - "pattern": "on" - } + "serialNumber": device_name, + "when": 0, + "duration": 1, + "pattern": "on" + } print(json.dumps(payload)) resp = setup_controller.led_blink_device(device_name, payload) print(resp.json()) - allure.attach(name="Device Blink led status", body=str(resp.json()), attachment_type=allure.attachment_type.JSON) + allure.attach(name="Device Blink led status", body=str(resp.json()), + attachment_type=allure.attachment_type.JSON) @pytest.mark.gw_trace_device def test_gw_service_trace_device(self, setup_controller, get_configuration): @@ -301,13 +304,13 @@ class TestUcentralGatewayService(object): """ device_name = get_configuration['access_point'][0]['serial'] payload = { - "serialNumber": device_name, - "when": 0, - "duration": 1, - "numberOfPackets": 0, - "network": "string", - "interface": "string" - } + "serialNumber": device_name, + "when": 0, + "duration": 1, + "numberOfPackets": 0, + "network": "string", + "interface": "string" + } print(json.dumps(payload)) resp = setup_controller.trace_device(device_name, payload) print(resp.json()) @@ -320,19 +323,20 @@ class TestUcentralGatewayService(object): """ device_name = get_configuration['access_point'][0]['serial'] payload = { - "serialNumber": device_name, - "verbose": True, - "activeScan": True, - "selector": { - "bands": [ - "2" - ] - } - } + "serialNumber": device_name, + "verbose": True, + "activeScan": True, + "selector": { + "bands": [ + "2" + ] + } + } print(json.dumps(payload)) resp = setup_controller.wifi_scan_device(device_name, payload) print(resp.json()) - allure.attach(name="Device Wifi scan status", body=str(resp.json()), attachment_type=allure.attachment_type.JSON) + allure.attach(name="Device Wifi scan status", body=str(resp.json()), + attachment_type=allure.attachment_type.JSON) @pytest.mark.gw_request_msg_device def test_gw_service_request_msg_device(self, setup_controller, get_configuration): @@ -341,14 +345,15 @@ class TestUcentralGatewayService(object): """ device_name = get_configuration['access_point'][0]['serial'] payload = { - "serialNumber": device_name, - "when": 0, - "message": "state" - } + "serialNumber": device_name, + "when": 0, + "message": "state" + } print(json.dumps(payload)) resp = setup_controller.request_specific_msg_from_device(device_name, payload) print(resp.json()) - allure.attach(name="Device Request specific msg status", body=str(resp.json()), attachment_type=allure.attachment_type.JSON) + allure.attach(name="Device Request specific msg status", body=str(resp.json()), + attachment_type=allure.attachment_type.JSON) @pytest.mark.gw_event_queue_device def test_gw_service_event_queue_device(self, setup_controller, get_configuration): @@ -357,15 +362,16 @@ class TestUcentralGatewayService(object): """ device_name = get_configuration['access_point'][0]['serial'] payload = { - "serialNumber": device_name, - "types": [ - "dhcp" - ] - } + "serialNumber": device_name, + "types": [ + "dhcp" + ] + } print(json.dumps(payload)) resp = setup_controller.event_queue(device_name, payload) print(resp.json()) - allure.attach(name="Device Request Event Queue status", body=str(resp.json()), attachment_type=allure.attachment_type.JSON) + allure.attach(name="Device Request Event Queue status", body=str(resp.json()), + attachment_type=allure.attachment_type.JSON) @pytest.mark.gw_telemetry_device def test_gw_service_telemetry_device(self, setup_controller, get_configuration): @@ -374,19 +380,20 @@ class TestUcentralGatewayService(object): """ device_name = get_configuration['access_point'][0]['serial'] payload = { - "serialNumber": device_name, - "interval": 0, - "lifetime": 0, - "kafka": False, - "types": [ - "dhcp-snooping" - ], - "uuid": "3fa85f64-5717-4562-b3fc-2c963f66afa6" - } + "serialNumber": device_name, + "interval": 0, + "lifetime": 0, + "kafka": False, + "types": [ + "dhcp-snooping" + ], + "uuid": "3fa85f64-5717-4562-b3fc-2c963f66afa6" + } print(json.dumps(payload)) resp = setup_controller.telemetry(device_name, payload) print(resp.json()) - allure.attach(name="Device telemetry status", body=str(resp.json()), attachment_type=allure.attachment_type.JSON) + allure.attach(name="Device telemetry status", body=str(resp.json()), + attachment_type=allure.attachment_type.JSON) @pytest.mark.gw_rtty def test_gw_service_get_rtty(self, setup_controller, get_configuration): diff --git a/tests/controller_tests/ucentral_gateway/test_provisionservice.py b/tests/controller_tests/ucentral_gateway/test_provisionservice.py index 0ec846339..f185bc680 100644 --- a/tests/controller_tests/ucentral_gateway/test_provisionservice.py +++ b/tests/controller_tests/ucentral_gateway/test_provisionservice.py @@ -4,21 +4,22 @@ """ -import string +import json import random -import pytest -import json import allure +import pytest @pytest.mark.ow_sanity_lf @pytest.mark.uc_sanity +@pytest.mark.ow_sdk_tests +@pytest.mark.ow_sdk_load_tests @pytest.mark.owprov_api_tests -@allure.feature("SDK PROV REST API") +@allure.parent_suite("OpenWifi SDK Tests") +@allure.parent_suite("OpenWifi Provisioning Service Tests") class TestUcentralProvisionService(object): - @pytest.mark.sdk_restapi @pytest.mark.prov_api def test_provservice_inventorylist(self, setup_prov_controller, get_configuration): @@ -28,7 +29,7 @@ class TestUcentralProvisionService(object): device_name = get_configuration['access_point'][0]['serial'] resp = setup_prov_controller.get_inventory_by_device(device_name) print(resp.json()) - allure.attach(name="Inventory", body=str(resp.json()),attachment_type=allure.attachment_type.JSON) + allure.attach(name="Inventory", body=str(resp.json()), attachment_type=allure.attachment_type.JSON) assert resp.status_code == 200 @pytest.mark.prov_api_test @@ -37,8 +38,8 @@ class TestUcentralProvisionService(object): Test the create device in provision Inventory """ device_mac = "02:00:00:%02x:%02x:%02x" % (random.randint(0, 255), - random.randint(0, 255), - random.randint(0, 255)) + random.randint(0, 255), + random.randint(0, 255)) device_name = device_mac.replace(":", "") # device_name = "deadbeef0011" + testbed.replace("-","") payload = {"serialNumber": device_name, @@ -77,15 +78,15 @@ class TestUcentralProvisionService(object): # This is for Edititng the information fo device in Inventory editing_payload = { - "description": "For testing API through automation after editing", - "devClass": "any", - "deviceType": "edgecore_eap101", - "entity": "", - "name": "Testing_to_add_device_through_automation", - "notes": [], - "rrm": "inherit", - "venue": "" - } + "description": "For testing API through automation after editing", + "devClass": "any", + "deviceType": "edgecore_eap101", + "entity": "", + "name": "Testing_to_add_device_through_automation", + "notes": [], + "rrm": "inherit", + "venue": "" + } print(json.dumps(editing_payload)) resp = setup_prov_controller.edit_device_from_inventory(device_name, editing_payload) allure.attach(name="response: ", body=str(resp.json())) @@ -121,10 +122,10 @@ class TestUcentralProvisionService(object): Test the create Entity in provision Inventory """ payload = {"name": "Testing_prov", - "rrm": "inherit", - "description": "For testing Purposes through Automation", - "notes": [{"note": "For testing Purposes through Automation"}], - "parent": "0000-0000-0000" + "rrm": "inherit", + "description": "For testing Purposes through Automation", + "notes": [{"note": "For testing Purposes through Automation"}], + "parent": "0000-0000-0000" } print(json.dumps(payload)) resp = setup_prov_controller.add_entity(payload) @@ -145,14 +146,14 @@ class TestUcentralProvisionService(object): # This to edit Entity editing_payload = { - "description": "For testing Purposes through Automation after edit", - "deviceConfiguration": [], - "name": "Testing_prov", - "notes": [], - "rrm": "inherit", - "sourceIP": [], - "uuid": entity_id - } + "description": "For testing Purposes through Automation after edit", + "deviceConfiguration": [], + "name": "Testing_prov", + "notes": [], + "rrm": "inherit", + "sourceIP": [], + "uuid": entity_id + } print(json.dumps(editing_payload)) resp = setup_prov_controller.edit_entity(editing_payload, entity_id) allure.attach(name="response: ", body=str(resp.json())) @@ -196,24 +197,24 @@ class TestUcentralProvisionService(object): Test the create Contact in provision Inventory """ payload = { - "name": "Prov-Testing-through-Automation", - "type": "USER", - "title": "Testing_contact", - "salutation": "", - "firstname": "ProvTesting", - "lastname": "Through Automation", - "initials": "", - "visual": "", - "phones": [], - "mobiles": [], - "primaryEmail": "tip@ucentral.com", - "secondaryEmail": "", - "accessPIN": "", - "description": "", - "initialNote": "", - "entity": "0000-0000-0000", - "notes": [{"note": ""}] - } + "name": "Prov-Testing-through-Automation", + "type": "USER", + "title": "Testing_contact", + "salutation": "", + "firstname": "ProvTesting", + "lastname": "Through Automation", + "initials": "", + "visual": "", + "phones": [], + "mobiles": [], + "primaryEmail": "tip@ucentral.com", + "secondaryEmail": "", + "accessPIN": "", + "description": "", + "initialNote": "", + "entity": "0000-0000-0000", + "notes": [{"note": ""}] + } print(json.dumps(payload)) resp = setup_prov_controller.add_contact(payload) allure.attach(name="response: ", body=str(resp.json())) @@ -233,22 +234,22 @@ class TestUcentralProvisionService(object): # This to edit Entity editing_payload = { - "accessPIN": "", - "description": "", - "entity": "0000-0000-0000", - "firstname": "ProvTesting", - "initials": "", - "lastname": "Through Automation", - "mobiles": [], - "name": "Prov-Testing-Automation API's", - "notes": [], - "phones": [], - "primaryEmail": "tip@ucentral.com", - "salutation": "", - "secondaryEmail": "", - "title": "Testing_contact", - "type": "USER" - } + "accessPIN": "", + "description": "", + "entity": "0000-0000-0000", + "firstname": "ProvTesting", + "initials": "", + "lastname": "Through Automation", + "mobiles": [], + "name": "Prov-Testing-Automation API's", + "notes": [], + "phones": [], + "primaryEmail": "tip@ucentral.com", + "salutation": "", + "secondaryEmail": "", + "title": "Testing_contact", + "type": "USER" + } print(json.dumps(editing_payload)) resp = setup_prov_controller.edit_contact(editing_payload, contact_id) allure.attach(name="response: ", body=str(resp.json())) @@ -285,22 +286,22 @@ class TestUcentralProvisionService(object): Test the create location in provision Inventory """ payload = { - "name": "TIP", - "type": "AUTO", - "buildingName": "", - "addressLines": ["Pedda Rushikonda"], - "city": "Visakhapatnam", - "state": "Andhra pradesh", - "postal": "530045", - "country": "IN", - "phones": [], - "mobiles": [], - "geoCode": "", - "description": "For Testing through Automation", - "initialNote": "Testing purposes through Automation", - "entity": "0000-0000-0000", - "notes": [{"note": "Testing purposes"}] - } + "name": "TIP", + "type": "AUTO", + "buildingName": "", + "addressLines": ["Pedda Rushikonda"], + "city": "Visakhapatnam", + "state": "Andhra pradesh", + "postal": "530045", + "country": "IN", + "phones": [], + "mobiles": [], + "geoCode": "", + "description": "For Testing through Automation", + "initialNote": "Testing purposes through Automation", + "entity": "0000-0000-0000", + "notes": [{"note": "Testing purposes"}] + } print(json.dumps(payload)) resp = setup_prov_controller.add_location(payload) allure.attach(name="response: ", body=str(resp.json())) @@ -320,24 +321,24 @@ class TestUcentralProvisionService(object): # This to edit Entity editing_payload = { - "addressLines": [ - "Madhurawada", - "" - ], - "buildingName": "", - "city": "Visakhapatnam", - "country": "IN", - "description": "Candela Testing", - "entity": "0000-0000-0000", - "geoCode": "", - "mobiles": [], - "name": "Candela IND", - "notes": [], - "phones": [], - "postal": "530048", - "state": "Andhra Pradesh", - "type": "SERVICE" - } + "addressLines": [ + "Madhurawada", + "" + ], + "buildingName": "", + "city": "Visakhapatnam", + "country": "IN", + "description": "Candela Testing", + "entity": "0000-0000-0000", + "geoCode": "", + "mobiles": [], + "name": "Candela IND", + "notes": [], + "phones": [], + "postal": "530048", + "state": "Andhra Pradesh", + "type": "SERVICE" + } print(json.dumps(editing_payload)) resp = setup_prov_controller.edit_location(editing_payload, location_id) allure.attach(name="response: ", body=str(resp.json())) @@ -374,18 +375,18 @@ class TestUcentralProvisionService(object): Test the create venue in provision Inventory """ payload = { - "description": "For testing Purposes", - "entity": "6a657863-9940-4303-ac68-4cc10d3078ec", - "location": "", - "name": "Testing Prov", - "notes": [ - { - "note": "For testing Purposes" - } - ], - "parent": "", - "rrm": "inherit" - } + "description": "For testing Purposes", + "entity": "6a657863-9940-4303-ac68-4cc10d3078ec", + "location": "", + "name": "Testing Prov", + "notes": [ + { + "note": "For testing Purposes" + } + ], + "parent": "", + "rrm": "inherit" + } print(json.dumps(payload)) resp = setup_prov_controller.add_venue(payload) allure.attach(name="response: ", body=str(resp.json())) @@ -405,13 +406,13 @@ class TestUcentralProvisionService(object): # This to edit venue editing_payload = { - "description": "For testing Purposes through Automation", - "location": "", - "name": "Testing Prov", - "notes": [], - "rrm": "inherit", - "sourceIP": [] - } + "description": "For testing Purposes through Automation", + "location": "", + "name": "Testing Prov", + "notes": [], + "rrm": "inherit", + "sourceIP": [] + } print(json.dumps(editing_payload)) resp = setup_prov_controller.edit_venue(editing_payload, venue_id) allure.attach(name="response: ", body=str(resp.json())) @@ -432,4 +433,4 @@ class TestUcentralProvisionService(object): body = resp.url + "," + str(resp.status_code) + ',' + resp.text allure.attach(name="Prov created venue-delete", body=body) if resp.status_code != 200: - assert False \ No newline at end of file + assert False diff --git a/tests/controller_tests/ucentral_gateway/test_securityservice.py b/tests/controller_tests/ucentral_gateway/test_securityservice.py index 14bf1e262..45a4d9e6b 100644 --- a/tests/controller_tests/ucentral_gateway/test_securityservice.py +++ b/tests/controller_tests/ucentral_gateway/test_securityservice.py @@ -10,8 +10,11 @@ import allure @pytest.mark.uc_sanity @pytest.mark.ow_sanity_lf +@pytest.mark.ow_sdk_tests +@pytest.mark.ow_sdk_load_tests @pytest.mark.owsec_api_tests -@allure.feature("SDK REST API") +@allure.parent_suite("OpenWifi SDK Tests") +@allure.parent_suite("OpenWifi Security Service Tests") class TestUcentralSecService(object): """ Test the oauth endpoint diff --git a/tests/e2e/basic/test_e2e_sdk_ap_tests.py b/tests/e2e/basic/test_e2e_sdk_ap_tests.py new file mode 100644 index 000000000..e6712f4a9 --- /dev/null +++ b/tests/e2e/basic/test_e2e_sdk_ap_tests.py @@ -0,0 +1,154 @@ +import sys +import time +from datetime import datetime + +import allure +import pytest + +setup_params = [ + { + "mode": "BRIDGE", + "ssids": [ + {"ssid_name": "ssid_psk_2g", "appliedRadios": ["2G"], "security_key": "something", "security": "psk"}, + {"ssid_name": "ssid_psk_5g", "appliedRadios": ["5G"], "security_key": "something", "security": "psk"}], + "radius": False + }, + + { + "mode": "BRIDGE", + "ssids": [ + {"ssid_name": "ssid_psk2_2g", "appliedRadios": ["2G"], "security_key": "something", "security": "psk2"}, + {"ssid_name": "ssid_psk2_5g", "appliedRadios": ["5G"], "security_key": "something", "security": "psk2"}], + "radius": False + }, + + { + "mode": "BRIDGE", + "ssids": [ + {"ssid_name": "ssid_sae_2g", "appliedRadios": ["2G"], "security_key": "something", "security": "sae"}, + {"ssid_name": "ssid_sae_5g", "appliedRadios": ["5G"], "security_key": "something", "security": "sae"}], + "radius": False + }, + + { + "mode": "BRIDGE", + "ssids": [ + {"ssid_name": "ssid_open_2g", "appliedRadios": ["2G"], "security_key": "something", "security": "none"}, + {"ssid_name": "ssid_open_5g", "appliedRadios": ["5G"], "security_key": "something", "security": "none"}], + "radius": False + }, + + { + "mode": "BRIDGE", + "ssids": [ + {"ssid_name": "ssid_wpa2_eap_2g", "appliedRadios": ["2G"], "security_key": "something", "security": "wpa2"}, + {"ssid_name": "ssid_wpa2_eap_5g", "appliedRadios": ["5G"], "security_key": "something", + "security": "wpa2"}], + "radius": True + }, + + { + "mode": "BRIDGE", + "ssids": [ + {"ssid_name": "ssid_wpa3_eap_2g", "appliedRadios": ["2G"], "security_key": "something", "security": "wpa3"}, + {"ssid_name": "ssid_wpa3_eap_5g", "appliedRadios": ["5G"], "security_key": "something", + "security": "wpa3"}], + "radius": True + }, + + { + "mode": "NAT", + "ssids": [ + {"ssid_name": "ssid_psk_2g", "appliedRadios": ["2G"], "security_key": "something", "security": "psk"}, + {"ssid_name": "ssid_psk_5g", "appliedRadios": ["5G"], "security_key": "something", "security": "psk"}], + "radius": False + }, + + { + "mode": "NAT", + "ssids": [ + {"ssid_name": "ssid_psk2_2g", "appliedRadios": ["2G"], "security_key": "something", "security": "psk2"}, + {"ssid_name": "ssid_psk2_5g", "appliedRadios": ["5G"], "security_key": "something", "security": "psk2"}], + "radius": False + }, + + { + "mode": "NAT", + "ssids": [ + {"ssid_name": "ssid_sae_2g", "appliedRadios": ["2G"], "security_key": "something", "security": "sae"}, + {"ssid_name": "ssid_sae_5g", "appliedRadios": ["5G"], "security_key": "something", "security": "sae"}], + "radius": False + }, + + { + "mode": "NAT", + "ssids": [ + {"ssid_name": "ssid_open_2g", "appliedRadios": ["2G"], "security_key": "something", "security": "none"}, + {"ssid_name": "ssid_open_5g", "appliedRadios": ["5G"], "security_key": "something", "security": "none"}], + "radius": False + }, + + { + "mode": "NAT", + "ssids": [ + {"ssid_name": "ssid_wpa2_eap_2g", "appliedRadios": ["2G"], "security_key": "something", "security": "wpa2"}, + {"ssid_name": "ssid_wpa2_eap_5g", "appliedRadios": ["5G"], "security_key": "something", + "security": "wpa2"}], + "radius": True + }, + + { + "mode": "NAT", + "ssids": [ + {"ssid_name": "ssid_wpa3_eap_2g", "appliedRadios": ["2G"], "security_key": "something", "security": "wpa3"}, + {"ssid_name": "ssid_wpa3_eap_5g", "appliedRadios": ["5G"], "security_key": "something", + "security": "wpa3"}], + "radius": True + }, + +] + + +@pytest.mark.ow_config_load_test +@pytest.mark.ow_sdk_load_tests +@allure.parent_suite("OpenWifi SDK E2E Tests") +@allure.parent_suite("OpenWifi Gateway E2E Configuration Test") +class TestBulkConfigTest(object): + + @allure.sub_suite("Back to Back config Apply test on single AP") + def test_config_apply_test(self, setup_controller, instantiate_profile, get_configuration, radius_info, + radius_accounting_info): + """ + Test the system endpoints to verify list of services present + WIFI-3449 + """ + PASS = [] + SERIAL = get_configuration["access_point"][0]["serial"] + for config in setup_params: + profile_obj = instantiate_profile(sdk_client=setup_controller) + profile_obj.set_mode(config["mode"]) + profile_obj.set_radio_config() + radius = config["radius"] + for ssid in config["ssids"]: + if radius: + profile_obj.add_ssid(ssid_data=ssid, radius=radius, radius_auth_data=radius_info, + radius_accounting_data=radius_accounting_info) + else: + profile_obj.add_ssid(ssid_data=ssid) + status = profile_obj.push_config(serial_number=SERIAL) + if status.status_code != 200: + allure.attach("Configure command Failed: ", SERIAL, " Time: " + str(datetime.utcnow())) + print(status.status_code + ":\t" + status.json()) + allure.attach(name=status.status_code, body=status.json()) + print("Configure command success: ", SERIAL, " Time: " + str(datetime.utcnow())) + PASS.append(False) + if status.status_code == 200: + print(status.status_code + ":\t" + status.json()) + allure.attach(name=status.status_code, body=status.json()) + allure.attach("Configure command success: ", SERIAL, " Time: " + str(datetime.utcnow())) + print("Configure command success: ", SERIAL, " Time: " + str(datetime.utcnow())) + PASS.append(True) + print("Sleeping 30 Sec before Next Config") + time.sleep(30) + + assert False not in PASS diff --git a/tests/fixtures_2x.py b/tests/fixtures_2x.py index 20fce5265..86df02a42 100644 --- a/tests/fixtures_2x.py +++ b/tests/fixtures_2x.py @@ -52,7 +52,7 @@ class Fixtures_2x: if not run_lf: try: self.controller_obj = Controller(controller_data=self.lab_info["controller"]) - self.prov_controller_obj = ProvUtils(controller_data=self.lab_info["controller"]) + self.prov_controller_obj = ProvUtils(sdk_client=self.controller_obj) self.fw_client = FMSUtils(sdk_client=self.controller_obj) except Exception as e: print(e)