pytest: Start integrating with existing libraries.

This is going to take more work, but getting somewhat closer.

Stop using pytest/helpers, they are replaced by libraries.

Existing libraries are often taking a command-line-args option
as easy way to pass something to various methods.  This will
have to be resolved to work with the way pytest wants to
run.

Much hard-coding needs to be removed, and specifically,
the 'set profile' logic is completley  not working.

Need to use library logic that can query the eq-id based
on name (and query eq-name based on serial port/ssh, perhaps).

Need to resolve how cmd-line can take multiple APs, but
each will have different eq-id.

Signed-off-by: Ben Greear <greearb@candelatech.com>
This commit is contained in:
Ben Greear
2021-02-11 13:18:38 -08:00
parent 2ea71662c3
commit 55d9e99978
12 changed files with 236 additions and 109 deletions

View File

@@ -9,6 +9,7 @@ sudo pip3 install paramiko
sudo pip3 install scp
sudo pip3 install pexpect
sudo pip3 install pexpect-serial
sudo yum install pytest
# Clone these repositories to get started:
git@github.com:Telecominfraproject/wlan-testing.git # This repo

View File

@@ -8,13 +8,22 @@ from lab_ap_info import *
class GetBuild:
def __init__(self, jfrog_user, jfrog_passwd, build):
def __init__(self, jfrog_user, jfrog_passwd, build, url=None):
self.user = jfrog_user
self.password = jfrog_passwd
ssl._create_default_https_context = ssl._create_unverified_context
self.jfrog_url = 'https://tip.jfrog.io/artifactory/tip-wlan-ap-firmware/'
if url:
self.jfrog_url = url
else:
self.jfrog_url = 'https://tip.jfrog.io/artifactory/tip-wlan-ap-firmware/'
self.build = build
def get_user(self):
return self.user
def get_passwd(self):
return self.password
def get_latest_image(self, model, for_build=None):
build_name = self.build
@@ -22,6 +31,7 @@ class GetBuild:
build_name = for_build
url = self.jfrog_url + model + "/dev/"
print("JfrogHelper::get_latest_image, url: ", url)
auth = str(
base64.b64encode(

View File

@@ -44,6 +44,21 @@ class CloudSDK:
self.password = command_line_args.sdk_user_password
self.assert_bad_response = False
self.verbose = command_line_args.verbose
self.base_url = command_line_args.sdk_base_url
self.cloud_type = "v1"
self.refresh_bearer()
def __init__(self, id, passwd, url, verbose):
self.user = id
self.password = passwd
self.base_url = url
self.assert_bad_response = False
self.verbose = verbose
self.cloud_type = "v1"
self.refresh_bearer()
def refresh_bearer(self):
self.bearer = self.get_bearer(self.base_url, self.cloud_type)
def get_bearer(self, cloudSDK_url, cloud_type):
cloud_login_url = cloudSDK_url+"/management/"+cloud_type+"/oauth2/token"
@@ -63,6 +78,7 @@ class CloudSDK:
raise SystemExit("Exiting Script! Cloud not get bearer token for reason:",e)
token_data = token_response.json()
bearer_token = token_data['access_token']
bearer = bearer_token
return(bearer_token)
def check_response(self, cmd, response, headers, data_str, url):
@@ -84,14 +100,15 @@ class CloudSDK:
return False
return True
def should_upgrade_ap_fw(self, bearer, command_line_args, report_data, latest_ap_image, fw_model, ap_cli_fw,
def should_upgrade_ap_fw(self, force_upgrade, skip_upgrade, report_data, latest_ap_image, fw_model, ap_cli_fw,
logger):
do_upgrade = False
if ap_cli_fw == latest_ap_image and command_line_args.force_upgrade != True:
if ap_cli_fw == latest_ap_image and force_upgrade != True:
print('FW does not require updating')
if report_data:
report_data['fw_available'][key] = "No"
logger.info(fw_model + " does not require upgrade.")
if logger:
logger.info(fw_model + " does not require upgrade.")
cloudsdk_cluster_info = {
"date": "N/A",
"commitId": "N/A",
@@ -100,11 +117,12 @@ class CloudSDK:
if report_data:
report_data['cloud_sdk'][key] = cloudsdk_cluster_info
if ap_cli_fw != latest_ap_image and command_line_args.skip_upgrade == True:
if ap_cli_fw != latest_ap_image and skip_upgrade == True:
print('FW needs updating, but skip_upgrade is True, so skipping upgrade')
if report_data:
report_data['fw_available'][key] = "No"
logger.info(fw_model + " firmware upgrade skipped, running with " + ap_cli_fw)
if logger:
logger.info(fw_model + " firmware upgrade skipped, running with " + ap_cli_fw)
cloudsdk_cluster_info = {
"date": "N/A",
"commitId": "N/A",
@@ -113,7 +131,7 @@ class CloudSDK:
if report_data:
report_data['cloud_sdk'][key] = cloudsdk_cluster_info
if (ap_cli_fw != latest_ap_image or command_line_args.force_upgrade == True) and not command_line_args.skip_upgrade:
if (ap_cli_fw != latest_ap_image or force_upgrade == True) and not skip_upgrade:
print('Updating firmware, old: %s new: %s'%(ap_cli_fw, latest_ap_image))
do_upgrade = True
if report_data:
@@ -124,13 +142,15 @@ class CloudSDK:
# client is testrail client
def do_upgrade_ap_fw(self, bearer, command_line_args, report_data, test_cases, client, ap_image, cloudModel, model,
def do_upgrade_ap_fw(self, command_line_args, report_data, test_cases, testrail_client, ap_image, cloudModel, model,
jfrog_user, jfrog_pwd, testrails_rid, customer_id, equipment_id, logger):
###Test Create Firmware Version
key = model
rid = testrails_rid
cloudSDK_url = command_line_args.sdk_base_url
test_id_fw = test_cases["create_fw"]
cloudSDK_url = self.base_url
bearer = self.bearer
if test_cases:
test_id_fw = test_cases["create_fw"]
print(cloudModel)
firmware_list_by_model = self.CloudSDK_images(cloudModel, cloudSDK_url, bearer)
print("Available", cloudModel, "Firmware on CloudSDK:", firmware_list_by_model)
@@ -146,15 +166,17 @@ class CloudSDK:
bearer)
fw_id = fw_upload_status['id']
print("Upload Complete.", ap_image, "FW ID is", fw_id)
client.update_testrail(case_id=test_id_fw, run_id=rid, status_id=1,
msg='Create new FW version by API successful')
if testrail_client:
testrail_client.update_testrail(case_id=test_id_fw, run_id=rid, status_id=1,
msg='Create new FW version by API successful')
if report_data:
report_data['tests'][key][test_id_fw] = "passed"
except:
fw_upload_status = 'error'
print("Unable to upload new FW version. Skipping Sanity on AP Model")
client.update_testrail(case_id=test_id_fw, run_id=rid, status_id=5,
msg='Error creating new FW version by API')
if testrail_client:
testrail_client.update_testrail(case_id=test_id_fw, run_id=rid, status_id=5,
msg='Error creating new FW version by API')
if report_data:
report_data['tests'][key][test_id_fw] = "failed"
return False
@@ -167,15 +189,17 @@ class CloudSDK:
bearer)
fw_id = fw_upload_status['id']
print("Upload Complete.", ap_image, "FW ID is", fw_id)
client.update_testrail(case_id=test_id_fw, run_id=rid, status_id=1,
msg='Create new FW version by API successful')
if testrail_client:
testrail_client.update_testrail(case_id=test_id_fw, run_id=rid, status_id=1,
msg='Create new FW version by API successful')
if report_data:
report_data['tests'][key][test_id_fw] = "passed"
except:
fw_upload_status = 'error'
print("Unable to upload new FW version. Skipping Sanity on AP Model")
client.update_testrail(case_id=test_id_fw, run_id=rid, status_id=5,
msg='Error creating new FW version by API')
if testrail_client:
testrail_client.update_testrail(case_id=test_id_fw, run_id=rid, status_id=5,
msg='Error creating new FW version by API')
if report_data:
report_data['tests'][key][test_id_fw] = "failed"
return False
@@ -183,7 +207,8 @@ class CloudSDK:
# Upgrade AP firmware
print("Upgrading...firmware ID is: ", fw_id)
upgrade_fw = self.update_firmware(equipment_id, str(fw_id), cloudSDK_url, bearer)
logger.info("Lab " + model + " Requires FW update")
if logger:
logger.info("Lab " + model + " Requires FW update")
print(upgrade_fw)
if "success" in upgrade_fw:
@@ -191,25 +216,31 @@ class CloudSDK:
print("CloudSDK Upgrade Request Success")
if report_data and test_cases:
report_data['tests'][key][test_cases["upgrade_api"]] = "passed"
client.update_testrail(case_id=test_cases["upgrade_api"], run_id=rid, status_id=1, msg='Upgrade request using API successful')
logger.info('Firmware upgrade API successfully sent')
if testrail_client:
testrail_client.update_testrail(case_id=test_cases["upgrade_api"], run_id=rid, status_id=1, msg='Upgrade request using API successful')
if logger:
logger.info('Firmware upgrade API successfully sent')
else:
print("Cloud SDK Upgrade Request Error!")
# mark upgrade test case as failed with CloudSDK error
if test_cases:
client.update_testrail(case_id=test_cases["upgrade_api"], run_id=rid, status_id=5, msg='Error requesting upgrade via API')
if testrail_client:
testrail_client.update_testrail(case_id=test_cases["upgrade_api"], run_id=rid, status_id=5, msg='Error requesting upgrade via API')
if report_data:
report_data['tests'][key][test_cases["upgrade_api"]] = "failed"
logger.warning('Firmware upgrade API failed to send')
if logger:
logger.warning('Firmware upgrade API failed to send')
return False
else:
print("Cloud SDK Upgrade Request Error!")
# mark upgrade test case as failed with CloudSDK error
if test_cases:
client.update_testrail(case_id=test_cases["upgrade_api"], run_id=rid, status_id=5,msg='Error requesting upgrade via API')
if testrail_client:
testrail_client.update_testrail(case_id=test_cases["upgrade_api"], run_id=rid, status_id=5,msg='Error requesting upgrade via API')
if report_data:
report_data['tests'][key][test_cases["upgrade_api"]] = "failed"
logger.warning('Firmware upgrade API failed to send')
if logger:
logger.warning('Firmware upgrade API failed to send')
return False
sdk_ok = False
@@ -220,7 +251,8 @@ class CloudSDK:
test_id_cloud = test_cases["cloud_fw"]
cloud_ap_fw = self.ap_firmware(customer_id, equipment_id, cloudSDK_url, bearer)
print('Current AP Firmware from CloudSDK: %s requested-image: %s'%(cloud_ap_fw, ap_image))
logger.info('AP Firmware from CloudSDK: ' + cloud_ap_fw)
if logger:
logger.info('AP Firmware from CloudSDK: ' + cloud_ap_fw)
if cloud_ap_fw == "ERROR":
print("AP FW Could not be read from CloudSDK")
@@ -242,7 +274,8 @@ class CloudSDK:
ap_cli_info = ssh_cli_active_fw(command_line_args)
ap_cli_fw = ap_cli_info['active_fw']
print("CLI reporting AP Active FW as:", ap_cli_fw)
logger.info('Firmware from CLI: ' + ap_cli_fw)
if logger:
logger.info('Firmware from CLI: ' + ap_cli_fw)
if ap_cli_fw == ap_image:
cli_ok = True
break
@@ -254,10 +287,12 @@ class CloudSDK:
print(ex)
logging.error(logging.traceback.format_exc())
print("Cannot Reach AP CLI to confirm upgrade!")
logger.warning('Cannot Reach AP CLI to confirm upgrade!')
if logger:
logger.warning('Cannot Reach AP CLI to confirm upgrade!')
if test_cases:
client.update_testrail(case_id=test_id_cli, run_id=rid, status_id=4,
msg='Cannot reach AP after upgrade to check CLI - re-test required')
if testrail_client:
testrail_client.update_testrail(case_id=test_id_cli, run_id=rid, status_id=4,
msg='Cannot reach AP after upgrade to check CLI - re-test required')
continue
time.sleep(30)
@@ -271,10 +306,11 @@ class CloudSDK:
if cloud_ap_fw == ap_image and ap_cli_fw == ap_image:
print("CloudSDK and AP CLI both show upgrade success, passing upgrade test case")
if test_cases:
client.update_testrail(case_id=test_id_cli, run_id=rid, status_id=1,
msg='Upgrade to ' + ap_image + ' successful')
client.update_testrail(case_id=test_id_cloud, run_id=rid, status_id=1,
msg='CLOUDSDK reporting correct firmware version.')
if testrail_client:
testrail_client.update_testrail(case_id=test_id_cli, run_id=rid, status_id=1,
msg='Upgrade to ' + ap_image + ' successful')
testrail_client.update_testrail(case_id=test_id_cloud, run_id=rid, status_id=1,
msg='CLOUDSDK reporting correct firmware version.')
if report_data:
report_data['tests'][key][test_id_cli] = "passed"
report_data['tests'][key][test_id_cloud] = "passed"
@@ -285,10 +321,11 @@ class CloudSDK:
print("AP CLI shows upgrade success - CloudSDK reporting error!")
##Raise CloudSDK error but continue testing
if test_cases:
client.update_testrail(case_id=test_id_cli, run_id=rid, status_id=1,
msg='Upgrade to ' + ap_image + ' successful.')
client.update_testrail(case_id=test_id_cloud, run_id=rid, status_id=5,
msg='CLOUDSDK reporting incorrect firmware version.')
if testrail_client:
testrail_client.update_testrail(case_id=test_id_cli, run_id=rid, status_id=1,
msg='Upgrade to ' + ap_image + ' successful.')
testrail_client.update_testrail(case_id=test_id_cloud, run_id=rid, status_id=5,
msg='CLOUDSDK reporting incorrect firmware version.')
if report_data:
report_data['tests'][key][test_id_cli] = "passed"
report_data['tests'][key][test_id_cloud] = "failed"
@@ -299,10 +336,11 @@ class CloudSDK:
print("AP CLI shows upgrade failed - CloudSDK reporting error!")
# Testrail TC fail
if test_cases:
client.update_testrail(case_id=test_id_cli, run_id=rid, status_id=5,
msg='AP failed to download or apply new FW. Upgrade to ' + ap_image + ' Failed')
client.update_testrail(case_id=test_id_cloud, run_id=rid, status_id=5,
msg='CLOUDSDK reporting incorrect firmware version.')
if testrail_client:
testrail_client.update_testrail(case_id=test_id_cli, run_id=rid, status_id=5,
msg='AP failed to download or apply new FW. Upgrade to ' + ap_image + ' Failed')
testrail_client.update_testrail(case_id=test_id_cloud, run_id=rid, status_id=5,
msg='CLOUDSDK reporting incorrect firmware version.')
if report_data:
report_data['tests'][key][test_id_cli] = "failed"
report_data['tests'][key][test_id_cloud] = "failed"
@@ -313,8 +351,9 @@ class CloudSDK:
print("Upgrade Failed! Confirmed on CloudSDK and AP CLI. Upgrade test case failed.")
##fail TR testcase and exit
if test_cases:
client.update_testrail(case_id=test_id_cli, run_id=rid, status_id=5,
msg='AP failed to download or apply new FW. Upgrade to ' + ap_image + ' Failed')
if testrail_client:
testrail_client.update_testrail(case_id=test_id_cli, run_id=rid, status_id=5,
msg='AP failed to download or apply new FW. Upgrade to ' + ap_image + ' Failed')
if report_data:
report_data['tests'][key][test_id_cli] = "failed"
print(report_data['tests'][key])
@@ -324,8 +363,9 @@ class CloudSDK:
print("Unable to determine upgrade status. Skipping AP variant")
# update TR testcase as error
if test_cases:
client.update_testrail(case_id=test_id_cli, run_id=rid, status_id=4,
msg='Cannot determine upgrade status - re-test required')
if testrail_client:
testrail_client.update_testrail(case_id=test_id_cli, run_id=rid, status_id=4,
msg='Cannot determine upgrade status - re-test required')
if report_data:
report_data['tests'][key][test_id_cli] = "error"
print(report_data['tests'][key])
@@ -613,12 +653,12 @@ class CloudSDK:
#print(response.text)
return response.json()
def set_ap_profile(self, equipment_id, test_profile_id, cloudSDK_url, bearer):
def set_ap_profile(self, equipment_id, test_profile_id):
###Get AP Info
url = cloudSDK_url+"/portal/equipment?equipmentId="+equipment_id
url = self.base_url + "/portal/equipment?equipmentId=" + equipment_id
payload = {}
headers = {
'Authorization': 'Bearer ' + bearer
'Authorization': 'Bearer ' + self.bearer
}
response = requests.request("GET", url, headers=headers, data=payload)
@@ -635,7 +675,7 @@ class CloudSDK:
url = cloudSDK_url+"/portal/equipment"
headers = {
'Content-Type': 'application/json',
'Authorization': 'Bearer ' + bearer
'Authorization': 'Bearer ' + self.bearer
}
response = requests.request("PUT", url, headers=headers, data=json.dumps(equipment_info))

View File

@@ -28,6 +28,15 @@ class TestRail_Client:
if command_line_args.testrail_user_id == "NONE":
self.use_testrails = False
def __init__(self, url, id, passwd):
self.user = id
self.password = passwd
base_url = url
if not base_url.endswith('/'):
base_url += '/'
self.__url = base_url + 'index.php?/api/v2/'
self.use_testrails = True
def send_get(self, uri, filepath=None):
"""Issue a GET request (read) against the API.

View File

@@ -78,7 +78,7 @@ class NightlySanity:
# Check if AP needs Upgrade
if (self.firmware["current"] is not None) and self.firmware["latest"] != self.firmware["current"]:
do_upgrade = self.cloud.should_upgrade_ap_fw(self.bearer, self.args, self.report_data,
do_upgrade = self.cloud.should_upgrade_ap_fw(self.args.force_upgrade, self.args.skip_upgrade, self.report_data,
self.firmware["latest"],
self.args.model,
self.firmware["current"], self.logger)
@@ -95,7 +95,7 @@ class NightlySanity:
if do_upgrade:
cloud_model = self.cloud_sdk_models[self.args.model]
pf = self.cloud.do_upgrade_ap_fw(self.bearer, self.args, self.report_data, test_cases, self.client,
pf = self.cloud.do_upgrade_ap_fw(self.args, self.report_data, test_cases, self.client,
self.firmware["latest"], cloud_model, self.args.model,
self.args.jfrog_user_id, self.args.jfrog_user_password, self.rid,
self.customer_id, self.equipment_id, self.logger)

View File

@@ -6,8 +6,6 @@ if sys.version_info[0] != 3:
print("This script requires Python 3")
exit(1)
import sys
for folder in 'py-json', 'py-scripts':
if folder not in sys.path:
sys.path.append(f'../lanforge/lanforge-scripts/{folder}')

View File

@@ -3,8 +3,27 @@ from time import sleep, gmtime, strftime
import sys
import os
sys.path.append(os.path.join(os.path.dirname(__file__), 'helpers'))
from utils import CloudSDK_Client, TestRail_Client, jFrog_Client
#sys.path.append(os.path.join(os.path.dirname(__file__), 'helpers'))
sys.path.append(f'..')
for folder in 'py-json', 'py-scripts':
if folder not in sys.path:
sys.path.append(f'../../lanforge/lanforge-scripts/{folder}')
sys.path.append(f'../../libs/lanforge/')
sys.path.append(f'../../libs/cloudsdk/')
sys.path.append(f'../../libs/apnos/')
sys.path.append(f'../../libs/testrails/')
sys.path.append(f'../../libs/')
sys.path.append(f'../test_utility/')
from UnitTestBase import *
from JfrogHelper import *
from cloudsdk import *
from testrail_api import TestRail_Client
def pytest_addoption(parser):
parser.addini("jfrog-base-url", "jfrog base url")
@@ -14,6 +33,7 @@ def pytest_addoption(parser):
parser.addini("sdk-user-id", "cloud sdk username")
parser.addini("sdk-user-password", "cloud sdk user password")
parser.addini("sdk-customer-id", "cloud sdk customer id for the access points")
parser.addini("sdk-equipment-id", "cloud sdk equipment id for the access point")
parser.addini("testrail-base-url", "testrail base url")
parser.addini("testrail-project", "testrail project name to use to generate test reports")
parser.addini("testrail-user-id", "testrail username")
@@ -31,6 +51,14 @@ def pytest_addoption(parser):
type=str
)
parser.addoption(
"--sdk-equipment-id",
action="store",
default="-1",
help="SDK equipment ID for AP",
type=str
)
# # Cloud SDK
# parser.addoption(
# "--sdk-base-url",
@@ -178,62 +206,72 @@ def setup_testrails(request, instantiate_testrail, access_points):
)
yield runId
# TODO: Should not be session wide I think, you will want to run different
# configurations (bridge, nat, vlan, wpa/wpa2/eap, etc
@pytest.fixture(scope="session")
def setup_cloudsdk(request, instantiate_cloudsdk):
# snippet to do cleanup after all the tests are done
def fin():
print("Cloud SDK cleanup done")
request.addfinalizer(fin)
instantiate_cloudsdk.set_ap_profile(3, 6)
yield {
"LANforge": {
"host": request.config.getini("lanforge-ip-address"),
"port": request.config.getini("lanforge-port-number"),
"radio": request.config.getini("lanforge-radio"),
"eth_port": request.config.getini("lanforge-ethernet-port"),
"runtime_duration": 15
},
"24ghz": {
"ssid": "TipWlan-cloud-wifi",
"password": "w1r3l3ss-fr33d0m",
"station_names": [ "sta2237" ]
}
}
# This is broken, see sdk_set_profile for correct way to do this.
#instantiate_cloudsdk.set_ap_profile(3, 6)
#yield {
# "LANforge": {
# "host": request.config.getini("lanforge-ip-address"),
# "port": request.config.getini("lanforge-port-number"),
# "radio": request.config.getini("lanforge-radio"),
# "eth_port": request.config.getini("lanforge-ethernet-port"),
# "runtime_duration": 15
# },
# "24ghz": {
# "ssid": "TipWlan-cloud-wifi",
# "password": "w1r3l3ss-fr33d0m",
# "station_names": [ "sta2237" ]
# }
#}
@pytest.fixture(scope="session")
def update_firmware(request, setup_testrails, instantiate_jFrog, instantiate_cloudsdk, access_points):
if request.config.getoption("--skip-update-firmware"):
return
latest_image = instantiate_jFrog.get_latest_image(access_points)
if latest_image in instantiate_cloudsdk.get_images(access_points):
model_firmware_id = instantiate_cloudsdk.get_firmware_id(latest_image)
else:
fw_url = instantiate_jFrog.get_latest_image_url(access_points, latest_image)
fw_upload_status = instantiate_cloudsdk.firwmare_upload(access_points, latest_image, fw_url)
model_firmware_id = fw_upload_status['id']
return True
# Get Current AP Firmware and upgrade\run tests if needed
# currently the AP id is hardcoded, but it should be looked up during the tests and not hardcoded in the config files or parameters
ap_fw = instantiate_cloudsdk.ap_firmware(request.config.getini("sdk-customer-id"), 3)
if ap_fw == latest_image:
pytest.skip("Do not need to upgrade firmware")
else:
instantiate_cloudsdk.update_firmware(3, model_firmware_id)
sleep_counter = 0
while True:
sleep_counter += 1
if instantiate_cloudsdk.ap_firmware(2, 3) == latest_image:
return
if sleep_counter > 0:
return
sleep(60)
#access_points is really a single AP.
ap = access_points
if True:
latest_image = instantiate_jFrog.get_latest_image(ap)
if latest_image is None:
print("AP Model: %s doesn't match the available Models"%(ap))
sys.exit(1) # TODO: How to return error properly here?
cloudModel = cloud_sdk_models[ap]
logger = None
report_data = None
test_cases = None
testrail_client = None
jfrog_user = instantiate_jFrog.get_user()
jfrog_pwd = instantiate_jFrog.get_passwd()
testrail_rid = 0
customer_id = request.config.getini("sdk-customer-id")
equipment_id = request.config.getoption("--sdk-equipment-id")
if equipment_id == "-1":
print("EQ ID invalid: ", equipment_id)
sys.exit(1)
pf = instantiate_cloudsdk.do_upgrade_ap_fw(request.config, report_data, test_cases, testrail_client,
latest_image, cloudModel, ap, jfrog_user, jfrog_pwd, testrail_rid,
customer_id, equipment_id, logger)
return pf
@pytest.fixture(scope="session")
def instantiate_cloudsdk(request):
yield CloudSDK_Client(
request.config.getini("sdk-base-url"),
yield CloudSDK(
request.config.getini("sdk-user-id"),
request.config.getini("sdk-user-password")
request.config.getini("sdk-user-password"),
request.config.getini("sdk-base-url"),
False # verbose TODO: Make this configurable
)
@pytest.fixture(scope="session")
@@ -246,8 +284,9 @@ def instantiate_testrail(request):
@pytest.fixture(scope="session")
def instantiate_jFrog(request):
yield jFrog_Client(
request.config.getini("jfrog-base-url"),
yield GetBuild(
request.config.getini("jfrog-user-id"),
request.config.getini("jfrog-user-password")
request.config.getini("jfrog-user-password"),
"pending", # TODO make this optional
url=request.config.getini("jfrog-base-url")
)

View File

@@ -109,6 +109,16 @@ class jFrog_Client:
return f"https://{self.user}:{self.password}@{self.baseUrl}/{model}/dev/{latest_image}.tar.gz"
# Class for CloudSDK Interaction via RestAPI
# Map firmware directory name to cloud's model name.
cloud_sdk_models = {
"ec420": "EC420-G1",
"ea8300": "EA8300-CA",
"ecw5211": "ECW5211",
"ecw5410": "ECW5410",
"wf188n": "WF188N"
}
class CloudSDK_Client:
def __init__(self, url, user, password):
self.baseUrl = f"https://{url}"

View File

@@ -1,15 +1,15 @@
[pytest]
addopts= --junitxml=test_everything.xml
# jFrog parameters
jfrog-base-url=tip.jFrog.io/artifactory/tip-wlan-ap-firmware
jfrog-base-url=https://tip.jfrog.io/artifactory/tip-wlan-ap-firmware/
jfrog-user-id=tip-read
jfrog-user-password=tip-read
# Cloud SDK parameters
sdk-base-url=wlan-portal-svc.cicd.lab.wlan.tip.build
sdk-base-url=https://wlan-portal-svc.cicd.lab.wlan.tip.build
sdk-user-id=support@example.com
sdk-user-password=support
# Testrails parameters
testrail-base-url=telecominfraproject.testrail.com
testrail-base-url=https://telecominfraproject.testrail.com
testrail-project=opsfleet-wlan
testrail-user-id=gleb@opsfleet.com
testrail-user-password=use_command_line_to_pass_this
@@ -21,10 +21,12 @@ lanforge-ethernet-port=eth2
# Cloud SDK settings
sdk-customer-id=2
# equipment ID is unique for each AP, have to be told what to use or query it based on other info.
sdk-equipment-id=-1
markers =
featureA: marks tests as slow (deselect with '-m "not slow"')
featureB
featureC
featureD
featureE
featureE

View File

@@ -2,6 +2,8 @@
# https://docs.pytest.org/en/latest/usage.html
# http://pythontesting.net/framework/pytest/pytest-introduction/
import sys
import pytest
from time import sleep, gmtime, strftime
from sta_connect2 import StaConnect2

View File

@@ -8,6 +8,7 @@ ssh -C -L 8800:lf1:4002 -L 8801:lf1:5901 -L 8802:lf1:8080 -L 8803:lab-ctlr:22 \
-L 8890:lf9:4002 -L 8891:lf9:5901 -L 8892:lf9:8080 -L 8893:lab-ctlr3:22 \
-L 8900:lf10:4002 -L 8901:lf10:5901 -L 8902:lf10:8080 -L 8903:lab-ctlr3:22 \
-L 8910:lf11:4002 -L 8911:lf11:5901 -L 8912:lf11:8080 -L 8913:lab-ctlr3:22 \
-L 8950:lf15:4002 -L 8951:lf15:5901 -L 8952:lf115:8080 -L 8953:lab-ctlr4:22 \
-L 8820:lf12:4002 -L 8821:lf12:5901 -L 8822:lf12:8080 -L 8823:lab-ctlr4:22 \
ubuntu@orch
@@ -83,3 +84,18 @@ Testbed 12 (Basic, wf188n)
./query_sdk.py --testrail-user-id NONE --model wf188n --sdk-base-url https://wlan-portal-svc-ben-testbed.cicd.lab.wlan.tip.build \
--sdk-user-id support@example.com --sdk-user-password support --equipment_id 3 --type profile --cmd get --object_id 11
Testbed-15
# Set AP profile on NOLA-15
./sdk_set_profile.py --testrail-user-id NONE --model ecw5410 --ap-jumphost-address localhost --ap-jumphost-port 8953 \
--ap-jumphost-password pumpkin77 --ap-jumphost-tty /dev/ttyAP4 --lanforge-ip-address localhost --lanforge-port-number 8952 \
--default-ap-profile TipWlan-2-Radios --sdk-base-url https://wlan-portal-svc.cicd.lab.wlan.tip.build \
--skip-radius --skip-wpa --verbose --testbed "NOLA-15" --ssid-5g-wpa2 Default-SSID-5gl --psk-5g-wpa2 12345678 \
--ssid-2g-wpa2 Default-SSID-2g --psk-2g-wpa2 12345678 --mode bridge
# Update firmware
./sdk_upgrade_fw.py --testrail-user-id NONE --model ecw5410 --ap-jumphost-address localhost --ap-jumphost-port 8953 \
--ap-jumphost-password pumpkin77 --ap-jumphost-tty /dev/ttyAP4 --testbed \"NOLA-15\" \
--sdk-base-url https://wlan-portal-svc-ben-testbed.cicd.lab.wlan.tip.build --force-upgrade true

View File

@@ -96,7 +96,7 @@ if equipment_id == "-1":
print("Found equipment ID: %s inventoryId: %s"%(e['id'], e['inventoryId']))
equipment_id = str(e['id'])
if equipment_id == -1:
if equipment_id == "-1":
print("ERROR: Could not find equipment-id.")
sys.exit(1)
@@ -166,11 +166,11 @@ print('Current Active AP FW from CLI:', ap_cli_fw)
##Compare Latest and Current AP FW and Upgrade
report_data = None
do_upgrade = cloud.should_upgrade_ap_fw(bearer, command_line_args, report_data, ap_image, fw_model, ap_cli_fw,
logger)
do_upgrade = cloud.should_upgrade_ap_fw(command_line_args.force_upgrade, command_line_args.skip_upgrade,
report_data, ap_image, fw_model, ap_cli_fw, logger)
cloudModel = cloud_sdk_models[model_id]
pf = cloud.do_upgrade_ap_fw(bearer, command_line_args, report_data, test_cases, client,
pf = cloud.do_upgrade_ap_fw(command_line_args, report_data, test_cases, client,
ap_image, cloudModel, model_id, jfrog_user, jfrog_pwd, rid,
customer_id, equipment_id, logger)