scottk/wlan-testing
1
0
Fork 0
mirror of https://github.com/Telecominfraproject/wlan-testing.git synced 2025-10-30 02:22:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

nightly test run

Browse Source
This commit is contained in:
Gleb Boushev
2020-11-13 12:01:28 +03:00
parent 39679cec5e
commit 24eb168f27
4 changed files with 193 additions and 203 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
docker/Dockerfile
View File

@@ -1,13 +1,15 @@
FROM python:3.8
WORKDIR /ci
RUN apt update && apt install vim -y && rm -rf /var/lib/apt/lists/*
RUN pip3 install requests
COPY wlan-LANforge-scripts/py-json/LANforge /ci/LANforge
COPY wlan-LANforge-scripts/py-json/realm.py /ci
COPY wlan-LANforge-scripts/py-json/generic_cx.py /ci
COPY wlan-LANforge-scripts/py-scripts/sta_connect2.py /ci
COPY wlan-LANforge-scripts/py-scripts/__init__.py /ci
COPY wlan-lanforge-scripts/py-json/LANforge /ci/LANforge
COPY wlan-lanforge-scripts/py-json/realm.py /ci
COPY wlan-lanforge-scripts/py-json/generic_cx.py /ci
COPY wlan-lanforge-scripts/py-scripts/sta_connect2.py /ci
COPY wlan-lanforge-scripts/py-scripts/__init__.py /ci
COPY wlan-testing/docker/nightly.py /ci
COPY wlan-testing/docker/test_config.json /ci
ENTRYPOINT [ "/ci/nightly.py" ]

326
docker/nightly.py
View File

@@ -3,37 +3,43 @@ import re
import requests
import json
import logging
import configparser
from time import sleep, gmtime, strftime
import argparse
parser = argparse.ArgumentParser()
parser.add_argument('--sdk-base-url', type=str, help='cloudsdk base url',
default="https://wlan-portal-svc.cicd.lab.wlan.tip.build")
parser.add_argument('--sdk-user-id', type=str, help='cloudsdk user id',
default="support@example.com")
parser.add_argument('--sdk-user-password', type=str, help='cloudsdk user password',
default="support")
parser.add_argument('--jfrog-base-url', type=str, help='jfrog base url',
default="tip.jFrog.io/artifactory/tip-wlan-ap-firmware")
parser.add_argument('--jfrog-user-id', type=str, help='jfrog user id',
default="cicd_user")
parser.add_argument('--jfrog-user-password', type=str, help='jfrog user password',
default="fakepassword")
parser.add_argument('--testrail-base-url', type=str, help='testrail base url',
default="https://telecominfraproject.testrail.com")
parser.add_argument('--testrail-project', type=str, help='testrail project name',
default="opsfleet-wlan")
parser.add_argument('--testrail-user-id', type=str, help='testrail user id',
default="gleb@opsfleet.com")
parser.add_argument('--testrail-user-password', type=str, help='testrail user password',
default="fakepassword")
args = parser.parse_args()
# if you lack __init__.py in this directory you will not find sta_connect module
from time import sleep, gmtime, strftime
# if you lack __init__.py in this directory this import will fail
# This should be replaced with a module and added to dockerfile
from LANforge.LFUtils import *
from sta_connect2 import StaConnect2
parser = argparse.ArgumentParser()
parser.add_argument("--sdk-base-url", type=str, help="cloudsdk base url",
default="https://wlan-portal-svc.cicd.lab.wlan.tip.build")
parser.add_argument("--sdk-user-id", type=str, help="cloudsdk user id",
default="support@example.com")
parser.add_argument("--sdk-user-password", type=str, help="cloudsdk user password",
default="support")
parser.add_argument("--jfrog-base-url", type=str, help="jfrog base url",
default="tip.jFrog.io/artifactory/tip-wlan-ap-firmware")
parser.add_argument("--jfrog-user-id", type=str, help="jfrog user id",
default="cicd_user")
parser.add_argument("--jfrog-user-password", type=str, help="jfrog user password",
default="fepv6nj9guCPeEHC")
parser.add_argument("--testrail-base-url", type=str, help="testrail base url",
default="https://telecominfraproject.testrail.com")
parser.add_argument("--testrail-project", type=str, help="testrail project name",
default="opsfleet-wlan")
parser.add_argument("--testrail-user-id", type=str, help="testrail user id",
default="gleb@opsfleet.com")
parser.add_argument("--testrail-user-password", type=str, help="testrail user password",
default="@3x57XQmkKVGFt8Ujnyf")
parser.add_argument("--lanforge-ip-address", type=str, help="ip address of the lanforge gui",
default="10.28.3.6")
parser.add_argument("--lanforge-port-number", type=str, help="port of the lanforge gui",
default="8080")
parser.add_argument('--update-firmware', dest='update_firmware', action='store_true')
parser.add_argument('--skip-update-firmware', dest='update_firmware', action='store_false')
parser.set_defaults(update_firmware=True)
args = parser.parse_args()
logging.basicConfig(level=logging.DEBUG,
format="%(asctime)s %(levelname)-8s %(message)s file:%(pathname)s line:%(lineno)d",
datefmt="%m-%d %H:%M",
@@ -44,12 +50,24 @@ console = logging.StreamHandler()
console.setLevel(logging.INFO)
console.setFormatter(formatter)
logging.getLogger().addHandler(console)
logging.info("")
logging.info("------------------------")
logging.info("nightly sanity run start")
# Class to interact with Testrail
# I suspect it is probably better to replace this
# with pytest and nunit xml report
# ap details, test data and test config
with open("test_config.json") as json_file:
TEST_DATA = json.load(json_file)
TESTRAIL = {
True: {
"statusCode": 1, # status_id is 1 for Passed, 2 For Blocked, 4 for Retest, 5 for Failed
"message": "success"
},
False: {
"statusCode": 5,
"message": "failuse"
}
}
# Class to interact with Testrail; better to replace this with pytest\nunit
class TestRail_Client:
def __init__(self):
self.user = args.testrail_user_id
@@ -91,7 +109,7 @@ class TestRail_Client:
logging.debug(f"Method: {method}; Url: {url}; Data: {data}")
if method == "POST":
if uri[:14] == "add_attachment": # add_attachment API method
if uri[:14] == "add_attachment": # add_attachment API method
files = { "attachment": open(data, "rb") }
response = requests.post(url, headers=headers, files=files, auth=(self.user, self.password))
files["attachment"].close()
@@ -132,27 +150,13 @@ class TestRail_Client:
if project["name"] == project_name:
return project["id"]
# def get_run_id(self, test_run_name, project_id):
# try:
# test_runs = testrail.send_get(f"get_runs/{project_id}")
# except Exception as e:
# logging.error(f"Exception in get_run_id(): {e}")
# return None
# else:
# for test_run in test_runs:
# if test_run["name"] == test_run_name:
# logging.info(test_run)
# run_id = test_run["id"]
# logging.info(f"runId in Test Runs: {run_id}")
# return run_id
def update_testrail(self, case_id, run_id, status_id, msg):
logging.info(f"Update TestRail; result status: {status_id}; case id: {case_id}; run id: {run_id}")
if run_id is not None:
try:
result = testrail.send_post(
f"add_result_for_case/{run_id}/{case_id}",
{ "status_id": status_id, "comment": msg } # status_id is 1 for Passed, 2 For Blocked, 4 for Retest and 5 for Failed
{ "status_id": status_id, "comment": msg }
)
logging.info(f"Updating TestRail result: {result}")
except Exception as e:
@@ -162,8 +166,8 @@ class TestRail_Client:
def create_testrun(self, name, case_ids, project_id):
result = testrail.send_post(
f'add_run/{project_id}',
{'name': name, 'case_ids': case_ids, 'include_all': False}
f"add_run/{project_id}",
{"name": name, "case_ids": case_ids, "include_all": False}
)
logging.info(result)
return result["id"]
@@ -209,7 +213,7 @@ class CloudSDK_Client:
equip_fw_url = f"{self.baseUrl}/portal/status/forEquipment?customerId={customer_id}&equipmentId={equipment_id}&statusDataTypes="
status_response = requests.get(equip_fw_url, headers=self.headers)
logging.debug(status_response.json())
return "ea8300-2020-09-09-pending-a6348d3" # (status_response.json())[2]["details"]["reportedSwVersion"]
return (status_response.json())[2]["details"]["reportedSwVersion"]
def get_images(self, apModel):
getFW_url = f"{self.baseUrl}/portal/firmware/version/byEquipmentType?equipmentType=AP&modelId={apModel}"
@@ -271,197 +275,137 @@ class CloudSDK_Client:
self.headers.pop("Content-Type", None)
logging.debug(response.text)
# Class for tests (temp)
class RunTest:
def Single_Client_Connectivity(self, radio, ssid_name, ssid_psk, security, station, test_case, runId):
"""SINGLE CLIENT CONNECTIVITY using sta_connect2.py"""
staConnect = StaConnect2("10.10.10.201", 8080, debug_= False)
staConnect.sta_mode = 0
staConnect.upstream_resource = 1
staConnect.upstream_port = "eth2"
staConnect.radio = radio
staConnect.resource = 1
staConnect.dut_ssid = ssid_name
staConnect.dut_passwd = ssid_psk
staConnect.dut_security = security
staConnect.station_names = station
staConnect.runtime_secs = 30
staConnect.cleanup_on_exit = True
staConnect.setup()
staConnect.start()
logging.info(f"sleeping {staConnect.runtime_secs} seconds")
sleep(staConnect.runtime_secs)
staConnect.stop()
staConnect.cleanup()
run_results = staConnect.get_result_list()
for result in run_results:
logging.info("test result: " + result)
if staConnect.passes() == True:
logging.info(f"Single client connection to {ssid_name} successful. Test Passed")
testrail.update_testrail(case_id=test_case, run_id=runId, status_id=1, msg="Client connectivity passed")
else:
logging.info(f"Single client connection to {ssid_name} unsuccessful. Test Failed")
testrail.update_testrail(case_id=test_case, run_id=runId, status_id=5, msg="Client connectivity failed")
# instantiate clients and configuration
sdk: CloudSDK_Client = CloudSDK_Client()
testrail: TestRail_Client = TestRail_Client()
jFrog: jFrog_Client = jFrog_Client()
Test: RunTest = RunTest()
# models under test and their data
customer_id = "2"
ap_models = {
"ea8300": {
"id": "3",
"firmware": "unknown",
"info": {
"profile_id": "24",
"fiveG_WPA2_SSID": "EA8300_5G_WPA2",
"fiveG_WPA2_PSK": "Connectus123$",
"fiveG_WPA_SSID": "EA8300_5G_WPA",
"fiveG_WPA_PSK": "Connectus123$",
"fiveG_OPEN_SSID": "EA8300_5G_OPEN",
"twoFourG_OPEN_SSID": "EA8300_2dot4G_OPEN",
"twoFourG_WPA2_SSID": "EA8300_2dot4G_WPA2",
"twoFourG_WPA2_PSK": "Connectus123$",
"twoFourG_WPA_SSID":"EA8300_2dot4G_WPA",
"twoFourG_WPA_PSK": "Connectus123$"
}
},
# "ecw5410": {
# "id": "6",
# "firmware": "unknown",
# "info": {
# "profile_id": "7",
# "fiveG_WPA2_SSID": "ECW5410_5G_WPA2",
# "fiveG_WPA2_PSK": "Connectus123$",
# "fiveG_WPA_SSID": "ECW5410_5G_WPA",
# "fiveG_WPA_PSK": "Connectus123$",
# "fiveG_OPEN_SSID": "ECW5410_5G_OPEN",
# "twoFourG_OPEN_SSID": "ECW5410_2dot4G_OPEN",
# "twoFourG_WPA2_SSID": "ECW5410_2dot4G_WPA2",
# "twoFourG_WPA2_PSK": "Connectus123$",
# "twoFourG_WPA_SSID":"ECW5410_2dot4G_WPA",
# "twoFourG_WPA_PSK": "Connectus123$"
# }
# }
}
# 1. Find Latest firmware on jFrog for each AP Model
# 2. Find Available firmware on CloudSDK
# 3. If Latest firmware not present, upload
# 4. Update Firmware on each AP Model
# 5. Run tests
for model in ap_models.keys():
# Get latest firmware on jFrog
logging.info(f"Model: {model}")
for model in TEST_DATA["ap_models"].keys():
# Get latest firmware on jFrog and Cloud SDK
latest_image = jFrog.get_latest_image(model)
logging.info(f"Latest firmware on jFrog: {latest_image}")
# Get latest firmware on Cloud SDK
firmware_list_by_model = sdk.get_images(model)
logging.info(f"Latest firmware on Cloud SDK: {firmware_list_by_model}")
logging.info(f"Model: {model}; Latest firmware on jFrog: {latest_image}; Latest firmware on Cloud SDK: {firmware_list_by_model}")
if latest_image in firmware_list_by_model:
logging.info(f"Latest firmware {latest_image} present on CloudSDK!")
ap_models[model]["firmware"] = latest_image
TEST_DATA["ap_models"][model]["firmware"] = latest_image
else:
logging.info(f"Uploading {latest_image} firmware to CloudSDK")
fw_url = jFrog.get_latest_image_url(model, latest_image)
fw_upload_status = sdk.firwmare_upload(model, latest_image, fw_url)
logging.debug(fw_upload_status)
logging.info(f"Upload Complete. {latest_image}; firmware ID is {fw_upload_status['id']}")
ap_models[model]["firmware"] = latest_image
TEST_DATA["ap_models"][model]["firmware"] = latest_image
# Get Current AP Firmware and upgrade
equipment_id = ap_models[model]["id"]
ap_fw = sdk.ap_firmware(customer_id, equipment_id)
ap_fw = sdk.ap_firmware(TEST_DATA["customer_id"], TEST_DATA["ap_models"][model]["id"])
fw_model = ap_fw.partition("-")[0]
logging.info(f"Testing AP model: {fw_model}; firmware: {ap_fw}")
# Find Latest firmware for Current AP Model and Get FW ID
latest_ap_image = ap_models[fw_model]["firmware"]
latest_ap_image = TEST_DATA["ap_models"][fw_model]["firmware"]
model_firmware_id = sdk.get_firmware_id(latest_ap_image)
logging.info(f"Latest firmware ID is: {model_firmware_id}")
logging.info(f"Firmware: {ap_fw}; latest firmware ID is: {model_firmware_id}")
if ap_fw != latest_ap_image:
if ap_fw == latest_ap_image and args.update_firmware:
logging.info("Model does not require firmware upgrade, skipping sanity tests")
else:
logging.info("Model requires firmware update, will update and sleep for 5 minutes")
sdk.update_firmware(equipment_id, model_firmware_id)
# sleep(300) # probably waits for the AP to upgrade ??? this is really fragile, need to loop and check
if args.update_firmware:
logging.info("Model requires firmware update, will update and sleep")
sdk.update_firmware(TEST_DATA["ap_models"][model]["id"], model_firmware_id)
sleep(300)
test_cases_data = {
2832: { # 2.4 GHz Open
"radio": "wiphy0",
"radio": "wiphy4",
"station": ["sta2234"],
"ssid_name": ap_models[fw_model]["info"]["twoFourG_OPEN_SSID"],
"ssid_name": TEST_DATA["ap_models"][fw_model]["info"]["twoFourG_OPEN_SSID"],
"ssid_psk": "BLANK",
"security": "open"
},
2833: { # 5 GHz Open
"radio": "wiphy3",
"station": ["sta2235"],
"ssid_name": ap_models[fw_model]["info"]["fiveG_OPEN_SSID"],
"ssid_psk": "BLANK",
"security": "open"
},
2834: { # 5 GHz WPA2
"radio": "wiphy3",
"station": ["sta2236"],
"ssid_name": ap_models[fw_model]["info"]["fiveG_WPA2_SSID"],
"ssid_psk": ap_models[fw_model]["info"]["fiveG_WPA2_PSK"],
"security": "wpa2"
},
2835: { # 2.4 GHz WPA2
"radio": "wiphy0",
"radio": "wiphy4",
"station": ["sta2237"],
"ssid_name": ap_models[fw_model]["info"]["twoFourG_WPA2_SSID"],
"ssid_psk": ap_models[fw_model]["info"]["twoFourG_WPA2_PSK"],
"ssid_name": TEST_DATA["ap_models"][fw_model]["info"]["twoFourG_WPA2_SSID"],
"ssid_psk": TEST_DATA["ap_models"][fw_model]["info"]["twoFourG_WPA2_PSK"],
"security": "wpa2"
},
2836: { # 5 GHz WPA
"radio": "wiphy3",
"station": ["sta2419"],
"ssid_name": ap_models[fw_model]["info"]["fiveG_WPA_SSID"],
"ssid_psk": ap_models[fw_model]["info"]["fiveG_WPA_PSK"],
"security": "wpa"
},
2837: { # 2.4 GHz WPA
"radio": "wiphy0",
"station": ["sta2420"],
"ssid_name": ap_models[fw_model]["info"]["twoFourG_WPA_SSID"],
"ssid_psk": ap_models[fw_model]["info"]["twoFourG_WPA_PSK"],
"security": "wpa"
}
# 2833: { # 5 GHz Open
# "radio": "wiphy3",
# "station": ["sta2235"],
# "ssid_name": TEST_DATA["ap_models"][fw_model]["info"]["fiveG_OPEN_SSID"],
# "ssid_psk": "BLANK",
# "security": "open"
# },
# 2834: { # 5 GHz WPA2
# "radio": "wiphy3",
# "station": ["sta2236"],
# "ssid_name": TEST_DATA["ap_models"][fw_model]["info"]["fiveG_WPA2_SSID"],
# "ssid_psk": TEST_DATA["ap_models"][fw_model]["info"]["fiveG_WPA2_PSK"],
# "security": "wpa2"
# },
# 2836: { # 5 GHz WPA
# "radio": "wiphy3",
# "station": ["sta2419"],
# "ssid_name": TEST_DATA["ap_models"][fw_model]["info"]["fiveG_WPA_SSID"],
# "ssid_psk": TEST_DATA["ap_models"][fw_model]["info"]["fiveG_WPA_PSK"],
# "security": "wpa"
# },
# 2837: { # 2.4 GHz WPA
# "radio": "wiphy0",
# "station": ["sta2420"],
# "ssid_name": TEST_DATA["ap_models"][fw_model]["info"]["twoFourG_WPA_SSID"],
# "ssid_psk": TEST_DATA["ap_models"][fw_model]["info"]["twoFourG_WPA_PSK"],
# "security": "wpa"
# }
}
# Create Test Run
test_run_name = f'Daily_Sanity_{fw_model}_{strftime("%Y-%m-%d", gmtime())}_{ap_fw}_opsfleet'
testrail_project_id = testrail.get_project_id(project_name=args.testrail_project)
runId = testrail.create_testrun(name=test_run_name, case_ids=(list(test_cases_data) + [2831]), project_id=testrail_project_id)
logging.debug(f"Testrail project id: {testrail_project_id}")
logging.debug(f"TIP run ID is: {runId}")
logging.info(f"Testrail project id: {testrail_project_id}; run ID is: {runId}")
# Check if upgrade worked
ap_fw = sdk.ap_firmware(customer_id, equipment_id)
logging.info(f"Current AP Firmware: {ap_fw}")
if ap_fw == latest_ap_image:
testrail.update_testrail(case_id="2831", run_id=runId, status_id=1, msg="Upgrade successful")
logging.info("Upgrade SUCCESS. Proceeding with sanity testing for this AP variant")
else:
testrail.update_testrail(case_id="2831", run_id=runId, status_id=5, msg="Upgrade Failed")
logging.error("Upgrade FAILED. Updating TestRail and skipping sanity tests for this AP variant.")
continue
if args.update_firmware:
results = TESTRAIL[(sdk.ap_firmware(TEST_DATA["customer_id"], TEST_DATA["ap_models"][model]["id"]) == latest_ap_image)]
testrail.update_testrail(case_id="2831", run_id=runId, status_id=results["statusCode"], msg=f"Upgrade {results['message']}")
logging.info(f"Upgrade {results['statusCode']}")
if not test_result:
continue
# Set Proper AP Profile
test_profile_id = ap_models[fw_model]["info"]["profile_id"]
logging.info(test_profile_id)
sdk.set_ap_profile(equipment_id, test_profile_id)
test_profile_id = TEST_DATA["ap_models"][fw_model]["info"]["profile_id"]
sdk.set_ap_profile(TEST_DATA["ap_models"][model]["id"], test_profile_id)
logging.info(f"Test profile id: test_profile_id")
# Run Client Single Connectivity Test Cases
for testcase in test_cases_data.keys():
logging.info(f"radio: {test_cases_data[testcase]['radio']}; ssid_name: {test_cases_data[testcase]['ssid_name']}; ssid_psk: {test_cases_data[testcase]['ssid_psk']}; security: {test_cases_data[testcase]['security']}; station: {test_cases_data[testcase]['station']}; testcase: {testcase}; runId: {runId}")
Test.Single_Client_Connectivity(test_cases_data[testcase]["radio"], test_cases_data[testcase]["ssid_name"], test_cases_data[testcase]["ssid_psk"],
test_cases_data[testcase]["security"], test_cases_data[testcase]["station"], testcase, runId)
logging.info(f"Test parameters are\n radio: {test_cases_data[testcase]['radio']}\n ssid_name: {test_cases_data[testcase]['ssid_name']}\n ssid_psk: {test_cases_data[testcase]['ssid_psk']}\n security: {test_cases_data[testcase]['security']}\n station: {test_cases_data[testcase]['station']}\n testcase: {testcase}\n runId: {runId}")
staConnect = StaConnect2(args.lanforge_ip_address, args.lanforge_port_number, debug_ = True)
staConnect.sta_mode = 0
staConnect.upstream_resource = 1
staConnect.upstream_port = "eth2"
staConnect.radio = test_cases_data[testcase]["radio"]
staConnect.resource = 1
staConnect.dut_ssid = test_cases_data[testcase]["ssid_name"]
staConnect.dut_passwd = test_cases_data[testcase]["ssid_psk"]
staConnect.dut_security = test_cases_data[testcase]["security"]
staConnect.station_names = test_cases_data[testcase]["station"]
staConnect.runtime_secs = 30
staConnect.cleanup_on_exit = True
staConnect.setup()
staConnect.start()
logging.info(f"sleeping {staConnect.runtime_secs} seconds")
sleep(staConnect.runtime_secs)
staConnect.stop()
staConnect.cleanup()
for result in staConnect.get_result_list():
logging.info(f"test result: {result}")
results = TESTRAIL[staConnect.passes()]
logging.info(f"Single client connection to {test_cases_data[testcase]['ssid_name']} successful. Test {results['message']}")
testrail.update_testrail(case_id=testcase, run_id=runId, status_id=results["statusCode"], msg="Client connectivity {results['message']}")
logging.info("----------------------")
logging.info("End of Sanity Test run")

22
docker/test_config.json Normal file
View File

@@ -0,0 +1,22 @@
{
"customer_id": "2",
"ap_models": {
"ecw5410": {
"id": "5",
"firmware": "unknown",
"info": {
"profile_id": "6",
"fiveG_WPA2_SSID": "TipWlan-cloud-wifi",
"fiveG_WPA2_PSK": "w1r3l3ss-fr33d0m",
"fiveG_WPA_SSID": "TipWlan-cloud-wifi",
"fiveG_WPA_PSK": "w1r3l3ss-fr33d0m",
"fiveG_OPEN_SSID": "TipWlan-cloud-wifi",
"twoFourG_OPEN_SSID": "gt-open",
"twoFourG_WPA2_SSID": "gt-wpa2",
"twoFourG_WPA2_PSK": "gt-wpa2-key",
"twoFourG_WPA_SSID": "TipWlan-cloud-wifi",
"twoFourG_WPA_PSK": "w1r3l3ss-fr33d0m"
}
}
}
}