Added Bridge mode fixtures

Signed-off-by: shivamcandela <shivam.thakur@candelatech.com>
This commit is contained in:
shivamcandela
2021-03-16 19:18:54 +05:30
parent 42671838fd
commit 462c450f3c
5 changed files with 304 additions and 118 deletions

View File

@@ -18,7 +18,7 @@ class APNOS:
print("Connecting to jumphost: %s@%s:%s with password: %s" % (
self.jumphost_username, self.jumphost_ip, self.jumphost_port, self.jumphost_password))
client.connect(self.jumphost_ip, username=self.jumphost_username, password=self.jumphost_password,
port=self.jumphost_port, timeout=10)
port=self.jumphost_port, timeout=10, allow_agent=False, banner_timeout=200)
return client
@@ -73,5 +73,13 @@ class APNOS:
return ssid_list
APNOS_CREDENTIAL_DATA = {
'jumphost_ip': "192.168.200.80",
'jumphost_username': "lanforge",
'jumphost_password': "lanforge",
'jumphost_port': 22
}
obj = APNOS(jumphost_cred=APNOS_CREDENTIAL_DATA)
print(obj.get_vif_config_ssids())
# print(get_vif_config_ssids())
# print(get_vif_state_ssids())

View File

@@ -4,13 +4,18 @@
1. testbed/ sdk_base_url
2. login credentials
"""
import base64
import datetime
import json
import re
import ssl
import time
import urllib
import requests
import swagger_client
from bs4 import BeautifulSoup
from testbed_info import SDK_BASE_URLS
from testbed_info import LOGIN_CREDENTIALS
@@ -54,14 +59,14 @@ class ConfigureCloudSDK:
"""
Library for cloudsdk generic usages, it instantiate the bearer and credentials.
Library for cloudsdk_tests generic usages, it instantiate the bearer and credentials.
It provides the connectivity to the cloud.
"""
class CloudSDK(ConfigureCloudSDK):
"""
constructor for cloudsdk library : can be used from pytest framework
constructor for cloudsdk_tests library : can be used from pytest framework
"""
def __init__(self, testbed=None, customer_id=None):
@@ -123,12 +128,25 @@ class CloudSDK(ConfigureCloudSDK):
pagination_context=pagination_context)
return equipment_data._items
def validate_equipment_availability(self, equipment_id=None):
data = self.get_equipment_by_customer_id()
for i in data:
if i._id == equipment_id:
return i._id
return -1
def request_ap_reboot(self):
pass
def request_firmware_update(self):
pass
def get_model_name(self, equipment_id=None):
if equipment_id is None:
return None
data = self.equipment_client.get_equipment_by_id(equipment_id=equipment_id)
return str(data._details._equipment_model)
"""
Profile Utilities
"""
@@ -180,6 +198,17 @@ class ProfileUtility:
if i._name == profile_name:
return i
return None
def get_profile_by_id(self, profile_id=None):
# pagination_context = """{
# "model_type": "PaginationContext",
# "maxItemsPerPage": 10
# }"""
profiles = self.profile_client.get_profile_by_id(profile_id=profile_id)
print(profiles)
# for i in profiles._items:
# if i._name == profile_name:
# return i
# return None
def get_default_profiles(self):
pagination_context = """{
@@ -196,7 +225,7 @@ class ProfileUtility:
if i._name == "TipWlan-3-Radios":
self.default_profiles['equipment_ap_3_radios'] = i
if i._name == "TipWlan-2-Radios":
self.default_profiles['equipment_ap_3_radios'] = i
self.default_profiles['equipment_ap_2_radios'] = i
if i._name == "Captive-Portal":
self.default_profiles['captive_portal'] = i
if i._name == "Radius-Profile":
@@ -451,34 +480,145 @@ class JFrogUtility:
self.password = credentials["password"]
self.jfrog_url = "https://tip.jfrog.io/artifactory/tip-wlan-ap-firmware/"
self.build = "pending"
ssl._create_default_https_context = ssl._create_unverified_context
def list_revisions(self):
def get_latest_build(self, model=None):
jfrog_url = self.jfrog_url + model + "/dev/"
auth = str(
base64.b64encode(
bytes('%s:%s' % (self.user, self.password), 'utf-8')
),
'ascii'
).strip()
headers = {'Authorization': 'Basic ' + auth}
''' FIND THE LATEST FILE NAME'''
# print(url)
req = urllib.request.Request(jfrog_url, headers=headers)
response = urllib.request.urlopen(req)
html = response.read()
soup = BeautifulSoup(html, features="html.parser")
##find the last pending link on dev
last_link = soup.find_all('a', href=re.compile(self.build))[-1]
latest_file = last_link['href']
latest_fw = latest_file.replace('.tar.gz', '')
return latest_fw
def get_revisions(self, model=None):
pass
def get_latest_build(self):
pass
def create_bridge_profile(get_testbed_name="nola-ext-04"):
# SSID and AP name shall be used as testbed_name and mode
sdk_client = CloudSDK(testbed=get_testbed_name, customer_id=2)
profile_object = ProfileUtility(sdk_client=sdk_client)
profile_object.get_default_profiles()
profile_object.set_rf_profile()
ssid_list = []
profile_data = {
"profile_name": "%s-%s-%s" % (get_testbed_name, "ecw5410", '5G_WPA2_BR'),
"ssid_name": "%s-%s-%s" % (get_testbed_name, "ecw5410", '5G_WPA2_BR'),
"mode": "BRIDGE",
"security_key": "%s-%s" % ("ecw5410", "5G_WPA2_BR")
}
profile_object.create_wpa2_personal_ssid_profile(profile_data=profile_data, fiveg=False)
ssid_list.append(profile_data["profile_name"])
profile_data = {
"profile_name": "%s-%s-%s" % (get_testbed_name, "ecw5410", '2G_WPA2_BR'),
"ssid_name": "%s-%s-%s" % (get_testbed_name, "ecw5410", '2G_WPA2_BR'),
"mode": "BRIDGE",
"security_key": "%s-%s" % ("ecw5410", "2G_WPA2_BR")
}
profile_object.create_wpa2_personal_ssid_profile(profile_data=profile_data, two4g=False)
ssid_list.append(profile_data["profile_name"])
# Create a wpa2 profile
pass
profile_data = {
"profile_name": "%s-%s-%s" % (get_testbed_name, "ecw5410", 'BRIDGE'),
}
profile_object.set_ap_profile(profile_data=profile_data)
profile_object.push_profile_old_method(equipment_id='13')
# create an equipment ap profile
return ssid_list
def vif(profile_data=[]):
import sys
if 'apnos' not in sys.path:
sys.path.append(f'../apnos')
from apnos import APNOS
APNOS_CREDENTIAL_DATA = {
'jumphost_ip': "192.168.200.80",
'jumphost_username': "lanforge",
'jumphost_password': "lanforge",
'jumphost_port': 22
}
obj = APNOS(APNOS_CREDENTIAL_DATA)
# data = obj.get_vif_config_ssids()
cur_time = datetime.datetime.now()
print(profile_data)
for i in range(18):
vif_config = list(obj.get_vif_config_ssids())
vif_config.sort()
vif_state = list(obj.get_vif_state_ssids())
vif_state.sort()
profile_data = list(profile_data)
profile_data.sort()
print("create_bridge_profile: ", profile_data)
print("vif config data: ", vif_config)
print("vif state data: ", vif_state)
if profile_data == vif_config:
print("matched")
if vif_config == vif_state:
status = True
print("matched 1")
break
else:
print("matched 2")
status = False
else:
status = False
time.sleep(10)
def main():
# credentials = {
# "user": "tip-read",
# "password": "tip-read"
# }
# obj = JFrogUtility(credentials=credentials)
# print(obj.get_latest_build(model="ecw5410"))
# sdk_client = CloudSDK(testbed="nola-ext-04", customer_id=2)
# ap_utils = ProfileUtility(sdk_client=sdk_client)
# ap_utils.get_profile_by_id(profile_id=5)
# # sdk_client.get_model_name()
# sdk_client.disconnect_cloudsdk()
cur_time = datetime.datetime.now()
data = create_bridge_profile()
vif(profile_data=data)
print(cur_time)
if __name__ == "__main__":
testbeds = ["nola-01", "nola-02", "nola-04", "nola-ext-01", "nola-ext-02", "nola-ext-03", "nola-ext-04",
"nola-ext-05"]
for i in testbeds:
sdk_client = CloudSDK(testbed=i, customer_id=2)
print(sdk_client.get_equipment_by_customer_id())
print(sdk_client.portal_ping() is None)
break
# ap_utils = ProfileUtility(sdk_client=sdk_client)
# ap_utils.get_default_profiles()
# for j in ap_utils.default_profiles:
# print(ap_utils.default_profiles[j]._id)
main()
# testbeds = ["nola-01", "nola-02", "nola-04", "nola-ext-01", "nola-ext-02", "nola-ext-03", "nola-ext-04",
# "nola-ext-05"]
# for i in testbeds:
# sdk_client = CloudSDK(testbed=i, customer_id=2)
# print(sdk_client.get_equipment_by_customer_id())
# print(sdk_client.portal_ping() is None)
# break
# # ap_utils = ProfileUtility(sdk_client=sdk_client)
# ap_utils.get_default_profiles()
# for j in ap_utils.default_profiles:
# print(ap_utils.default_profiles[j]._id)
# data = sdk_client.get_equipment_by_customer_id()
# equipment_ids = []
# for i in data:
# equipment_ids.append(i)
# print(equipment_ids[0]._details._equipment_model)
sdk_client.disconnect_cloudsdk()
time.sleep(2)
# data = sdk_client.get_equipment_by_customer_id()
# equipment_ids = []
# for i in data:
# equipment_ids.append(i)
# print(equipment_ids[0]._details._equipment_model)
# sdk_client.disconnect_cloudsdk()
# time.sleep(2)
# sdk_client.get_equipment_by_customer_id(customer_id=2)
# ap_utils = APUtils(sdk_client=sdk_client)
@@ -546,43 +686,3 @@ if __name__ == "__main__":
# time.sleep(1)
# ap_utils.delete_profile(profile_id=ap_utils.profile_ids)
def test_open_ssid():
sdk_client = CloudSDK(testbed="nola-ext-04")
ap_utils = APUtils(sdk_client=sdk_client)
print(sdk_client.configuration.api_key_prefix)
ap_utils.select_rf_profile(profile_data=None)
profile_data = {
"profile_name": "test-ssid-open",
"ssid_name": "test_open",
"mode": "BRIDGE"
}
ap_utils.create_open_ssid_profile(profile_data=profile_data)
profile_data = {
"profile_name": "test-ap-profile",
}
ap_utils.set_ap_profile(profile_data=profile_data)
ap_utils.push_profile_old_method(equipment_id='12')
sdk_client.disconnect_cloudsdk()
pass
def test_wpa_ssid():
pass
def test_wpa2_personal_ssid():
pass
def test_wpa3_personal_ssid():
pass
def test_wpa2_enterprise_ssid():
pass
def test_wpa3_enterprise_ssid():
pass

View File

@@ -1,5 +1,6 @@
import pytest
@pytest.mark.usefixtures('setup_cloudsdk')
@pytest.mark.usefixtures('update_firmware')
@pytest.mark.bridge_mode_client_connectivity
@@ -8,14 +9,14 @@ class TestBridgeModeClientConnectivity(object):
@pytest.mark.bridge_mode_single_client_connectivity
@pytest.mark.nightly
@pytest.mark.nightly_bridge
def test_single_client(self, setup_cloudsdk, update_firmware, setup_bridge_profile, disconnect_cloudsdk):
assert setup_cloudsdk != -1
def test_single_client(self, setup_cloudsdk, update_firmware, setup_bridge_mode, disconnect_cloudsdk):
print("I am Iron Man")
assert setup_bridge_mode is True
@pytest.mark.bridge_mode_multi_client_connectivity
def test_multi_client(self):
pass
# """
# Bridge mode:
# testbed name, customer_id, equipment_id, jfrog-credentials, cloudsdk_tests-credentials, skip-open, skip-wpa, skip-wpa2, skip-radius
@@ -35,4 +36,4 @@ class TestBridgeModeClientConnectivity(object):
# Bridge Mode:
# WPA, WPA2-PERSONAL, WPA2-ENTERPRISE
# 2.4/5, 2.4/5, 2.4/5
# """
# """

View File

@@ -1,13 +1,15 @@
# import files in the current directory
import datetime
import sys
import os
import time
sys.path.append(
os.path.dirname(
os.path.realpath( __file__ )
os.path.realpath(__file__)
)
)
if 'cloudsdk' not in sys.path:
sys.path.append(f'../libs/cloudsdk')
if 'apnos' not in sys.path:
@@ -17,8 +19,10 @@ from apnos import APNOS
from cloudsdk import CloudSDK
from cloudsdk import ProfileUtility
import pytest
import logging
from configuration_data import PROFILE_DATA
def pytest_addoption(parser):
parser.addini("jfrog-base-url", "jfrog base url")
parser.addini("jfrog-user-id", "jfrog username")
@@ -51,7 +55,7 @@ def pytest_addoption(parser):
parser.addini("skip-wpa", "skip wpa ssid mode")
parser.addini("skip-wpa2", "skip wpa2 ssid mode")
parser.addini("skip-eap", "skip eap ssid mode")
# change behaviour
# change behaviour
parser.addoption(
"--skip-update-firmware",
action="store_true",
@@ -63,75 +67,108 @@ def pytest_addoption(parser):
parser.addoption(
"--access-points",
# nargs="+",
default=[ "ECW5410" ],
default=["ECW5410"],
help="list of access points to test"
)
"""
Fixtures for Instantiate the Objects
"""
@pytest.fixture(scope="session")
def instantiate_cloudsdk(request):
sdk_client = CloudSDK(testbed=request.config.getini("testbed-name"),
customer_id=request.config.getini("sdk-customer-id"))
yield sdk_client
@pytest.fixture(scope="session")
def instantiate_jFrog(request):
yield "instantiate_jFrog"
"""
Fixtures for Getting Essentials from ini
"""
@pytest.fixture(scope="session")
def get_testbed_name(request):
yield request.config.getini("testbed-name")
@pytest.fixture(scope="session")
def get_customer_id(request):
yield request.config.getini("sdk-customer-id")
"""
Fixtures for CloudSDK Utilities
"""
@pytest.fixture(scope="session")
def get_equipment_model(request, instantiate_cloudsdk, get_equipment_id):
yield request.config.getini("testbed-name")
@pytest.fixture(scope="session")
def get_current_firmware(request, instantiate_cloudsdk, get_equipment_model):
yield request.config.getini("testbed-name")
def pytest_generate_tests(metafunc):
if 'access_points' in metafunc.fixturenames:
metafunc.parametrize("access_points", metafunc.config.getoption('--access-points'), scope="session")
# run something after all tests are done regardless of the outcome
def pytest_unconfigure(config):
# cleanup or reporting
print("Tests cleanup done")
@pytest.fixture(scope="function")
def setup_cloudsdk(request, instantiate_cloudsdk):
equipment_id = instantiate_cloudsdk.validate_equipment_availability(equipment_id=int(request.config.getini("sdk-equipment-id")))
equipment_id = instantiate_cloudsdk.validate_equipment_availability(
equipment_id=int(request.config.getini("sdk-equipment-id")))
if equipment_id == -1:
yield -1
else:
yield equipment_id
@pytest.fixture(scope="session")
def update_firmware(request, instantiate_jFrog, instantiate_cloudsdk, retrieve_latest_image, access_points):
if request.config.getoption("--skip-update-firmware"):
return
yield "update_firmware"
@pytest.fixture(scope="session")
def retrieve_latest_image(request, access_points):
if request.config.getoption("--skip-update-firmware"):
return
yield "retrieve_latest_image"
@pytest.fixture(scope="session")
def instantiate_cloudsdk(request):
sdk_client = CloudSDK(testbed=request.config.getini("testbed-name"), customer_id=request.config.getini("sdk-customer-id"))
yield sdk_client
@pytest.fixture(scope="session")
def instantiate_jFrog(request):
yield "instantiate_jFrog"
@pytest.fixture(scope="session")
def get_customer_id(request):
yield request.config.getini("sdk-customer-id")
@pytest.fixture(scope="session")
def get_testbed_name(request):
yield request.config.getini("testbed-name")
@pytest.fixture(scope="session")
def get_equipment_model(request, instantiate_cloudsdk):
yield request.config.getini("testbed-name")
@pytest.fixture(scope="session")
def get_current_firmware(request, instantiate_cloudsdk, get_equipment_model):
yield request.config.getini("testbed-name")
@pytest.fixture(scope="session")
def get_latest_firmware(request, instantiate_cloudsdk, get_equipment_model):
yield request.config.getini("testbed-name")
@pytest.fixture(scope="function")
def disconnect_cloudsdk(instantiate_cloudsdk):
instantiate_cloudsdk.disconnect_cloudsdk()
@pytest.fixture(scope="function")
def setup_bridge_mode(request, instantiate_cloudsdk, setup_profile_data, create_bridge_profile):
# vif config and vif state logic here
logging.basicConfig(level=logging.DEBUG)
log = logging.getLogger('test_1')
APNOS_CREDENTIAL_DATA = {
'jumphost_ip': request.config.getini("jumphost_ip"),
'jumphost_username': request.config.getini("jumphost_username"),
@@ -139,18 +176,46 @@ def setup_bridge_mode(request, instantiate_cloudsdk, setup_profile_data, create_
'jumphost_port': request.config.getini("jumphost_port")
}
obj = APNOS(APNOS_CREDENTIAL_DATA)
obj.get_vif_config_ssids()
condition = ""
if condition: # Condition that matches the vif config data with pushed profile data
yield True
else:
yield False
profile_data = create_bridge_profile
vif_config = list(obj.get_vif_config_ssids())
vif_config.sort()
vif_state = list(obj.get_vif_state_ssids())
vif_state.sort()
profile_data = list(profile_data)
profile_data.sort()
for i in range(18):
print("profiles pushed: ", profile_data)
print("vif config data: ", vif_config)
print("vif state data: ", vif_state)
if profile_data == vif_config:
print("matched")
if vif_config == vif_state:
status = True
print("matched 1")
break
else:
print("matched 2")
status = False
else:
status = False
time.sleep(10)
vif_config = list(obj.get_vif_config_ssids())
vif_config.sort()
vif_state = list(obj.get_vif_state_ssids())
vif_state.sort()
profile_data = list(profile_data)
profile_data.sort()
def delete_profile(profile_data, sdk_client):
print(f"Cloud SDK cleanup for {request.node.originalname}, {profile_data}")
delete_profiles(profile_names=profile_data, sdk_client=sdk_client)
request.addfinalizer(delete_profile(profile_data, instantiate_cloudsdk))
yield [profile_data, vif_config, vif_state]
@pytest.fixture(scope="function")
def setup_profile_data(request, setup_profile_data):
def setup_profile_data(request):
# logic to setup bridge mode ssid and parameters
pass
yield True
@pytest.fixture(scope="function")
@@ -159,6 +224,7 @@ def create_bridge_profile(request, instantiate_cloudsdk, get_testbed_name):
profile_object = ProfileUtility(sdk_client=instantiate_cloudsdk)
profile_object.get_default_profiles()
profile_object.set_rf_profile()
ssid_list = []
if request.config.getini("skip-open") == 'False':
profile_data = {
"profile_name": "%s-%s-%s" % (get_testbed_name, "ecw5410", '2G_O_BR'),
@@ -166,12 +232,14 @@ def create_bridge_profile(request, instantiate_cloudsdk, get_testbed_name):
"mode": "BRIDGE"
}
profile_object.create_open_ssid_profile(profile_data=profile_data, fiveg=False)
ssid_list.append(profile_data["profile_name"])
profile_data = {
"profile_name": "%s-%s-%s" % (get_testbed_name, "ecw5410", '5G_O_BR'),
"ssid_name": "%s-%s-%s" % (get_testbed_name, "ecw5410", '5G_O_BR'),
"mode": "BRIDGE"
}
profile_object.create_open_ssid_profile(profile_data=profile_data, two4g=False)
ssid_list.append(profile_data["profile_name"])
# Create an open ssid profile
if request.config.getini("skip-wpa") == 'False':
profile_data = {
@@ -181,6 +249,7 @@ def create_bridge_profile(request, instantiate_cloudsdk, get_testbed_name):
"security_key": "%s-%s" % ("ecw5410", "2G_WPA_BR")
}
profile_object.create_wpa_ssid_profile(profile_data=profile_data, fiveg=False)
ssid_list.append(profile_data["profile_name"])
profile_data = {
"profile_name": "%s-%s-%s" % (get_testbed_name, "ecw5410", '5G_WPA_BR'),
"ssid_name": "%s-%s-%s" % (get_testbed_name, "ecw5410", '5G_WPA_BR'),
@@ -188,6 +257,7 @@ def create_bridge_profile(request, instantiate_cloudsdk, get_testbed_name):
"security_key": "%s-%s" % ("ecw5410", "5G_WPA_BR")
}
profile_object.create_wpa_ssid_profile(profile_data=profile_data, two4g=False)
ssid_list.append(profile_data["profile_name"])
# Create a wpa profile
pass
if request.config.getini("skip-wpa2") == 'False':
@@ -198,6 +268,7 @@ def create_bridge_profile(request, instantiate_cloudsdk, get_testbed_name):
"security_key": "%s-%s" % ("ecw5410", "5G_WPA2_BR")
}
profile_object.create_wpa2_personal_ssid_profile(profile_data=profile_data, fiveg=False)
ssid_list.append(profile_data["profile_name"])
profile_data = {
"profile_name": "%s-%s-%s" % (get_testbed_name, "ecw5410", '2G_WPA2_BR'),
"ssid_name": "%s-%s-%s" % (get_testbed_name, "ecw5410", '2G_WPA2_BR'),
@@ -205,6 +276,7 @@ def create_bridge_profile(request, instantiate_cloudsdk, get_testbed_name):
"security_key": "%s-%s" % ("ecw5410", "2G_WPA2_BR")
}
profile_object.create_wpa2_personal_ssid_profile(profile_data=profile_data, two4g=False)
ssid_list.append(profile_data["profile_name"])
# Create a wpa2 profile
pass
if request.config.getini("skip-eap") == 'False':
@@ -221,8 +293,10 @@ def create_bridge_profile(request, instantiate_cloudsdk, get_testbed_name):
}
profile_object.set_ap_profile(profile_data=profile_data)
profile_object.push_profile_old_method(equipment_id='13')
# create an equipment ap profile
yield profile_object
yield ssid_list
@pytest.fixture(scope="function")
def setup_nat_profile(request, instantiate_cloudsdk, get_testbed_name):
@@ -295,6 +369,7 @@ def setup_nat_profile(request, instantiate_cloudsdk, get_testbed_name):
# create an equipment ap profile
yield profile_object
@pytest.fixture(scope="function")
def setup_vlan_profile(request, instantiate_cloudsdk):
# SSID and AP name shall be used as testbed_name and mode
@@ -316,6 +391,7 @@ def setup_vlan_profile(request, instantiate_cloudsdk):
# create an equipment ap profile
yield profile_object
@pytest.fixture(scope="function")
def apply_default_profile(instantiate_cloudsdk):
profile_object = ProfileUtility(sdk_client=instantiate_cloudsdk)
@@ -323,8 +399,9 @@ def apply_default_profile(instantiate_cloudsdk):
profile_object.profile_creation_ids['ap'] = profile_object.default_profiles['equipment_ap_3_radios']
profile_object.push_profile_old_method(equipment_id='13')
@pytest.fixture(scope="function")
def delete_profiles(instantiate_cloudsdk):
profile_object = ProfileUtility(sdk_client=instantiate_cloudsdk)
def delete_profiles(profile_names, sdk_client=None):
profile_object = ProfileUtility(sdk_client=sdk_client)
profile_object.get_default_profiles()
profile_object.profile_creation_ids['ap'] = profile_object.default_profiles['equipment_ap_3_radios']
profile_object.push_profile_old_method()

View File

@@ -18,10 +18,10 @@ testrail-user-id=gleb@opsfleet.com
testrail-user-password=use_command_line_to_pass_this
# Jumphost
jumphost-address=192.168.200.80
jumphost-port=22
jumphost-username=root
jumphost-password=lanforge
jumphost_ip=192.168.200.80
jumphost_port=22
jumphost_username=lanforge
jumphost_password=lanforge
# LANforge
lanforge-ip-address=localhost
@@ -34,7 +34,7 @@ sdk-customer-id=2
sdk-equipment-id=12
# Profile
skip-open=False
skip-open=True
skip-wpa=False
skip-wpa2=False
skip-eap=False