mirror of
https://github.com/Telecominfraproject/wlan-testing.git
synced 2025-10-31 02:48:11 +00:00
WIFI-9958 : Added a marker for ow_sdk_load_tests for sdk tests (#649)
Signed-off-by: shivam <shivam.thakur@candelatech.com>
This commit is contained in:
Shivam Thakur
@@ -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()
|
||||
|
||||
@@ -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()
|
||||
|
||||
|
||||
|
||||
|
||||
@@ -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
|
||||
@@ -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
|
||||
|
||||
@@ -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):
|
||||
|
||||
@@ -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
|
||||
assert False
|
||||
|
||||
@@ -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
|
||||
|
||||
154
tests/e2e/basic/test_e2e_sdk_ap_tests.py
Normal file
154
tests/e2e/basic/test_e2e_sdk_ap_tests.py
Normal file
@@ -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
|
||||
@@ -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)
|
||||
|
||||
Reference in New Issue
Block a user