Telecominfraproject/wlan-testing
1
0
Fork 0
mirror of https://github.com/Telecominfraproject/wlan-testing.git synced 2026-01-27 10:23:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #1 from Telecominfraproject/feature/WIFI-510

Browse Source 
Feature/wifi 510
This commit is contained in:
Ben Greear
2020-12-11 10:34:52 -08:00
committed by GitHub
parent fb4d573e9c b2db35d897
commit 3300bcc810
6 changed files with 733 additions and 545 deletions
Download Patch File Download Diff File Expand all files Collapse all files

126
.github/workflows/nightly.yml vendored
View File

@@ -1,27 +1,135 @@
name: nightly build
env:
DOCKER_SERVER: tip-tip-wlan-cloud-docker-repo.jfrog.io
DOCKER_USER_NAME: ${{ secrets.DOCKER_USER_NAME }}
DOCKER_USER_PASSWORD: ${{ secrets.DOCKER_USER_PASSWORD }}
TESTRAIL_USER_PASSWORD: ${{ secrets.TESTRAIL_USER_PASSWORD }}
JFROG_USER_PASSWORD: ${{ secrets.JFROG_USER_PASSWORD }}
# AWS credentials
AWS_EKS_NAME: tip-wlan-main
AWS_DEFAULT_OUTPUT: json
AWS_DEFAULT_REGION: us-east-2
AWS_ACCOUNT_ID: ${{ secrets.AWS_ACCOUNT_ID }}
AWS_ACCESS_KEY_ID: ${{ secrets.AWS_CLIENT_ID }}
AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_CLIENT_KEY }}
# Cloud SDK certs
CACERT: ${{ secrets.CACERT }}
CAKEY: ${{ secrets.CAKEY }}
on:
on:
workflow_dispatch:
schedule:
- cron: '15 0 * * *'
- cron: '15 0 * * *'
jobs:
build:
runs-on: ubuntu-latest
container: 4c74356b41/cloud-sdk-nightly
steps:
- name: Checkout
# checkout needed repositories
- name: Checkout Testing repo
uses: actions/checkout@v2
with:
path: wlan-testing
- name: Checkout LANforge scripts
uses: actions/checkout@v2
with:
path: wlan-lanforge-scripts
repository: Telecominfraproject/wlan-lanforge-scripts
token: ${{ secrets.GitHub_PAT }} # `GitHub_PAT` is a secret that contains your PAT
path: wlan-LANforge-scripts
# build and push docker image
- name: docker login
shell: bash
run: docker login ${{ env.DOCKER_SERVER }} -u ${{ env.DOCKER_USER_NAME }} -p ${{ env.DOCKER_USER_PASSWORD }}
- name: build docker image
shell: bash
run: ls -Rla; docker build -t tip/nightly -f wlan-testing/docker/Dockerfile
run: docker build -t ${{ env.DOCKER_SERVER }}/cloud-sdk-nightly:${{ github.run_id }} -f wlan-testing/docker/Dockerfile .
- name: push docker image
shell: bash
run: docker push ${{ env.DOCKER_SERVER }}/cloud-sdk-nightly:${{ github.run_id }}
cloudsdk:
runs-on: ubuntu-latest
steps:
- name: Checkout pki scripts repo
uses: actions/checkout@v2
with:
path: wlan-pki
repository: Telecominfraproject/wlan-pki-cert-scripts
ref: local/dev
- name: Checkout Cloud SDK repo
uses: actions/checkout@v2
with:
path: wlan-helm
repository: Telecominfraproject/wlan-cloud-helm
- name: Checkout helm values repo
uses: actions/checkout@v2
with:
path: toolsmith
repository: Telecominfraproject/Toolsmith
token: ${{ secrets.PAT_TOKEN }}
- name: Prepare environment
shell: bash
run: |
mkdir ./wlan-helm/tip-wlan/resources/certs
mkdir -p ./wlan-pki/testCA/private
mkdir -p ./wlan-pki/testCA/newcerts
mkdir -p ./wlan-pki/generated
touch ./wlan-pki/testCA/index.txt
echo "01" > ./wlan-pki/testCA/serial.txt
echo "${{ env.CACERT }}" | base64 -d > ./wlan-pki/testCA/cacert.pem
echo "${{ env.CAKEY }}" | base64 -d > ./wlan-pki/testCA/private/cakey.pem
cp ./toolsmith/helm-values/aws-cicd.yaml ./wlan-helm/tip-wlan/resources/environments/aws-cicd.yaml
- name: Generate certs
shell: bash
run: |
cd ./wlan-pki
./generate_all.sh true
./copy-certs-to-helm.sh "../wlan-helm"
- name: Deploy Cloud SDK
shell: bash
run: |
cd ./wlan-helm/tip-wlan
aws eks update-kubeconfig --name ${{ env.AWS_EKS_NAME }}
# the 2 lines below are needed for the new flow only, not needed for the old one
# helm dependency update
# helm upgrade --install tip-wlan . -f resources/environments/aws-cicd.yaml --create-namespace -n tip
helm upgrade --install tip-wlan . -f resources/environments/aws-cicd.yaml --namespace default
test:
runs-on: ubuntu-latest
needs: [ build, cloudsdk ]
steps:
- name: run tests
run: |
today=$(date +"%d-%m-%Y")
aws eks update-kubeconfig --name ${{ env.AWS_EKS_NAME }}
kubectl delete job nightly-ci-$today --wait=true --ignore-not-found=true
cat <<EOF | kubectl apply -f -
apiVersion: batch/v1
kind: Job
metadata:
name: nightly-ci-$today
spec:
template:
spec:
containers:
- name: tests
image: ${{ env.DOCKER_SERVER }}/cloud-sdk-nightly:${{ github.run_id }}
args:
- "--testrail-user-id"
- "gleb@opsfleet.com"
- "--testrail-user-password"
- "${{ env.TESTRAIL_USER_PASSWORD }}"
- "--skip-update-firmware"
imagePullSecrets:
- name: tip-docker-registry-key
restartPolicy: Never
backoffLimit: 0
EOF
# kubectl wait job/nightly-ci-$today --for=condition=complete --timeout=3600s
# kubectl logs -l job-name=nightly-ci-$today --since=1h
# kubectl delete job/nightly-ci-$today
shell: bash

142
.gitignore vendored Normal file
View File

@@ -0,0 +1,142 @@
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class
# C extensions
*.so
# Distribution / packaging
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
share/python-wheels/
*.egg-info/
.installed.cfg
*.egg
MANIFEST
# PyInstaller
# Usually these files are written by a python script from a template
# before PyInstaller builds the exe, so as to inject date/other infos into it.
*.manifest
*.spec
# Installer logs
pip-log.txt
pip-delete-this-directory.txt
# Unit test / coverage reports
htmlcov/
.tox/
.nox/
.coverage
.coverage.*
.cache
nosetests.xml
coverage.xml
*.cover
*.py,cover
.hypothesis/
.pytest_cache/
cover/
# Translations
*.mo
*.pot
# Django stuff:
*.log
local_settings.py
db.sqlite3
db.sqlite3-journal
# Flask stuff:
instance/
.webassets-cache
# Scrapy stuff:
.scrapy
# Sphinx documentation
docs/_build/
# PyBuilder
.pybuilder/
target/
# Jupyter Notebook
.ipynb_checkpoints
# IPython
profile_default/
ipython_config.py
# pyenv
# For a library or package, you might want to ignore these files since the code is
# intended to run in multiple environments; otherwise, check them in:
# .python-version
# pipenv
# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
# However, in case of collaboration, if having platform-specific dependencies or dependencies
# having no cross-platform support, pipenv may install dependencies that don't work, or not
# install all needed dependencies.
#Pipfile.lock
# PEP 582; used by e.g. github.com/David-OConnor/pyflow
__pypackages__/
# Celery stuff
celerybeat-schedule
celerybeat.pid
# SageMath parsed files
*.sage.py
# Environments
.env
.venv
env/
venv/
ENV/
env.bak/
venv.bak/
# Spyder project settings
.spyderproject
.spyproject
# Rope project settings
.ropeproject
# mkdocs documentation
/site
# mypy
.mypy_cache/
.dmypy.json
dmypy.json
# Pyre type checker
.pyre/
# pytype static type analyzer
.pytype/
# Cython debug symbols
cython_debug/
docker/*
!docker/nightly*
docker/nightly*log

536
CICD_AP_CLOUDSDK/CICD_Nightly_Sanity.py
View File

@@ -1,536 +0,0 @@
#!/usr/bin/python3
import base64
import urllib.request
from bs4 import BeautifulSoup
import ssl
import subprocess, os
from artifactory import ArtifactoryPath
import tarfile
import paramiko
from paramiko import SSHClient
from scp import SCPClient
import os
import pexpect
from pexpect import pxssh
import sys
import paramiko
from scp import SCPClient
import pprint
from pprint import pprint
from os import listdir
import re
import requests
import json
import testrail_api
import logging
import datetime
import time
local_dir="logs/"
#print("Local Directory where all files will be copied and logged", local_dir)
tr_user="syama.devi@connectus.ai"
tr_pw="Connectus123$"
#print ("Testrail password =", tr_pw)
aws_host='3.96.56.0'
aws_user='ubuntu'
logger = logging.getLogger('Nightly_Sanity')
hdlr = logging.FileHandler(local_dir+"/Nightly_Sanity.log")
formatter = logging.Formatter('%(asctime)s %(levelname)s %(message)s')
hdlr.setFormatter(formatter)
logger.addHandler(hdlr)
logger.setLevel(logging.INFO)
logger.info('Start Nightly Sanity run')
# For finding files
# https://stackoverflow.com/questions/3207219/how-do-i-list-all-files-of-a-directory
import glob
#external_results_dir=/var/tmp/lanforge
if sys.version_info[0] != 3:
print("This script requires Python 3")
exit(1)
if 'py-json' not in sys.path:
sys.path.append('../py-json')
from LANforge.LFUtils import *
# if you lack __init__.py in this directory you will not find sta_connect module#
import sta_connect
import sta_connect2
import testrail_api
from sta_connect import StaConnect
from sta_connect2 import StaConnect2
from testrail_api import APIClient
from datetime import date
client: APIClient = APIClient('https://telecominfraproject.testrail.com')
client.user = tr_user
client.password = tr_pw
###Class for jfrog Interaction
class GetBuild:
def __init__(self):
self.user = 'cicd_user'
self.password = 'fepv6nj9guCPeEHC'
ssl._create_default_https_context = ssl._create_unverified_context
def get_latest_image(self,url):
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(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("pending"))[-1]
latest_file=last_link['href']
latest_fw = latest_file.replace('.tar.gz','')
return latest_fw
###Class for CloudSDK Interaction via RestAPI
class CloudSDK:
def __init__(self):
self.user = 'support@example.com'
def get_bearer():
cloud_login_url = "https://wlan-portal-svc.demo.lab.wlan.tip.build/management/v1/oauth2/token"
payload = "{\n \"userId\": \"support@example.com\",\n \"password\": \"support\"\n}"
headers = {
'Content-Type': 'application/json'
}
try:
token_response = requests.request("POST", cloud_login_url, headers=headers, data=payload)
except requests.exceptions.RequestException as e:
raise SystemExit("Exiting Script! Cloud not get bearer token for reason:",e)
token_data = token_response.json()
bearer_token = token_data['access_token']
return(bearer_token)
def ap_firmware(customer_id,equipment_id):
equip_fw_url = "https://wlan-portal-svc.demo.lab.wlan.tip.build/portal/status/forEquipment?customerId="+customer_id+"&equipmentId="+equipment_id
payload = {}
headers = {
'Authorization': 'Bearer ' + bearer
}
status_response = requests.request("GET", equip_fw_url, headers=headers, data=payload)
status_data = status_response.json()
current_ap_fw = status_data[2]['details']['reportedSwVersion']
return current_ap_fw
def CloudSDK_images(apModel):
getFW_url = "https://wlan-portal-svc.demo.lab.wlan.tip.build/portal/firmware/version/byEquipmentType?equipmentType=AP&modelId=" + apModel
payload = {}
headers = {
'Authorization': 'Bearer ' + bearer
}
response = requests.request("GET", getFW_url, headers=headers, data=payload)
ap_fw_details = response.json()
###return ap_fw_details
fwlist = []
for version in ap_fw_details:
fwlist.append(version.get('versionName'))
return(fwlist)
#fw_versionNames = ap_fw_details[0]['versionName']
#return fw_versionNames
def firwmare_upload(commit, apModel,latest_image,fw_url):
fw_upload_url = "https://wlan-portal-svc.demo.lab.wlan.tip.build/portal/firmware/version"
payload = "{\n \"model_type\": \"FirmwareVersion\",\n \"id\": 0,\n \"equipmentType\": \"AP\",\n \"modelId\": \""+apModel+"\",\n \"versionName\": \""+latest_image+"\",\n \"description\": \"\",\n \"filename\": \""+fw_url+"\",\n \"commit\": \""+commit+"\",\n \"validationMethod\": \"MD5_CHECKSUM\",\n \"validationCode\": \"19494befa87eb6bb90a64fd515634263\",\n \"releaseDate\": 1596192028877,\n \"createdTimestamp\": 0,\n \"lastModifiedTimestamp\": 0\n}\n\n"
headers = {
'Content-Type': 'application/json',
'Authorization': 'Bearer ' + bearer
}
response = requests.request("POST", fw_upload_url, headers=headers, data=payload)
#print(response)
upload_result = response.json()
return(upload_result)
def get_firmware_id(latest_ap_image):
#print(latest_ap_image)
fw_id_url = "https://wlan-portal-svc.demo.lab.wlan.tip.build/portal/firmware/version/byName?firmwareVersionName="+latest_ap_image
payload = {}
headers = {
'Authorization': 'Bearer ' + bearer
}
response = requests.request("GET", fw_id_url, headers=headers, data=payload)
fw_data = response.json()
latest_fw_id = fw_data['id']
return latest_fw_id
def update_firmware(equioment_id, latest_firmware_id):
url = "https://wlan-portal-svc.demo.lab.wlan.tip.build/portal/equipmentGateway/requestFirmwareUpdate?equipmentId="+equipment_id+"&firmwareVersionId="+latest_firmware_id
payload = {}
headers = {
'Authorization': 'Bearer ' + bearer
}
response = requests.request("POST", url, headers=headers, data=payload)
print(response.text)
def set_ap_profile(equipment_id,test_profile_id):
###Get AP Info
url = "https://wlan-portal-svc.demo.lab.wlan.tip.build/portal/equipment?equipmentId="+equipment_id
payload = {}
headers = {
'Authorization': 'Bearer ' + bearer
}
response = requests.request("GET", url, headers=headers, data=payload)
print(response)
###Add Lab Profile ID to Equipment
equipment_info = response.json()
#print(equipment_info)
equipment_info["profileId"] = test_profile_id
#print(equipment_info)
###Update AP Info with Required Profile ID
url = "https://wlan-portal-svc.demo.lab.wlan.tip.build/portal/equipment"
headers = {
'Content-Type': 'application/json',
'Authorization': 'Bearer ' + bearer
}
response = requests.request("PUT", url, headers=headers, data=json.dumps(equipment_info))
#print(response)
###Class for Tests
class RunTest:
def Single_Client_Connectivity(self, radio, ssid_name, ssid_psk, security, station, test_case, rid):
'''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()
print("napping %f sec" % staConnect.runtime_secs)
time.sleep(staConnect.runtime_secs)
staConnect.stop()
staConnect.cleanup()
run_results = staConnect.get_result_list()
for result in run_results:
print("test result: " + result)
#result = 'pass'
print("Single Client Connectivity :",staConnect.passes)
if staConnect.passes() == True:
print("Single client connection to", ssid_name, "successful. Test Passed")
client.update_testrail(case_id=test_case, run_id=rid, status_id=1, msg='Client connectivity passed')
else:
client.update_testrail(case_id=test_case, run_id=rid, status_id=5, msg='Client connectivity failed')
print("Single client connection to", ssid_name, "unsuccessful. Test Failed")
def Client_Count(customer_id, equipment_id):
url = "https://wlan-portal-svc.demo.lab.wlan.tip.build/portal/status/forEquipment?customerId=" + customer_id + "&equipmentId=" + equipment_id
payload = {}
headers = {
'Authorization': 'Bearer ' + bearer
}
response = requests.request("GET", url, headers=headers, data=payload)
print(response.text.encode('utf8'))
def TestCase_941(self, rid):
#MULTI CLIENT CONNECTIVITY
staConnect = StaConnect2("10.10.10.201", 8080, debug_= True)
staConnect.sta_mode = 0
staConnect.upstream_resource = 1
staConnect.upstream_port = "eth2"
staConnect.radio = "wiphy0"
staConnect.resource = 1
staConnect.dut_ssid = "LAB_2dot4G_WPA2"
# staConnect.dut_passwd = "4C0nnectUS!"
staConnect.dut_passwd = "Connectus123$"
staConnect.dut_security = "wpa2"
staConnect.station_names = ["sta0020", 'sta0021', 'sta0022', 'sta0023']
staConnect.runtime_secs = 20
staConnect.cleanup_on_exit = True
staConnect.run()
run_results = staConnect.get_result_list()
for result in run_results:
print("test result: " + result)
if staConnect.passes() == True:
client.update_testrail(case_id=941, run_id=rid, status_id=1,
msg='client Connectivity to SSID Passed ')
else:
client.update_testrail(case_id=941, run_id=rid, status_id=5,
msg='client connectivity to SSID Failed')
######Testrail Project and Run ID Information ##############################
Test: RunTest = RunTest()
projId = client.get_project_id(project_name= 'WLAN')
print("TIP WLAN Project ID is:", projId)
logger.info('Start of Nightly Sanity')
###Dictionaries
ap_latest_dict = {
"ea8300": "Unknown",
"ecw5410": "unknown"
}
ap_updated_dict = {
"ea8300": "",
"ecw5410": ""
}
##Equipment IDs for Lab APs under test
equipment_id_dict = {
"ea8300": "3",
"ecw5410": "6"
}
###Testing AP Profile Information
profile_info_dict = {
"ecw5410": {
"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$"
},
"ea8300": {
"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$"
}
}
##Test Cases to be included in Test Runs
test_cases = [
2233,
2234,
2235,
2236,
2237,
2419,
2420
]
##AP models
ap_models = ["ea8300","ecw5410"]
#ap_models = ["ecw5410"]
###Get Cloud Bearer Token
bearer = CloudSDK.get_bearer()
#############################################################################
##################### CloudSDK Firmware Check ###############################
### 1) Get Token for CloudSDK ###############################################
### 2) Find Latest FW on jfrog for each AP Model ############################
### 3) Find Available FW on CloudSDK --> if Latest FW not present, upload ###
#############################################################################
###Check Latest FW Version on jfrog and CloudSDK for each model
for model in ap_models:
apModel = model
###Check Latest FW on jFrog
jfrog_url = 'https://tip.jfrog.io/artifactory/tip-wlan-ap-firmware/'
url = jfrog_url + apModel + "/dev/"
Build: GetBuild = GetBuild()
latest_image = Build.get_latest_image(url)
print(model,"Latest FW on jFrog:",latest_image)
###Check images already on CloudSDK
firmware_list_by_model = CloudSDK.CloudSDK_images(apModel)
print("Available",apModel,"Firmware on CloudSDK:",firmware_list_by_model)
if latest_image in firmware_list_by_model:
print("Latest Firmware",latest_image,"is on CloudSDK!")
ap_latest_dict[model] = latest_image
ap_updated_dict[model] = "Up to Date"
else:
print("Latest Firmware is not on CloudSDK! Uploading...")
fw_url = "https://cicd_user:fepv6nj9guCPeEHC@tip.jfrog.io/artifactory/tip-wlan-ap-firmware/" + apModel + "/dev/" + latest_image + ".tar.gz"
#print(fw_url)
commit = latest_image.split("-")[-1]
fw_upload_status = CloudSDK.firwmare_upload(commit, apModel,latest_image,fw_url)
#print(fw_upload_status)
fw_id = fw_upload_status['id']
print("Upload Complete.",latest_image,"FW ID is",fw_id)
ap_latest_dict[model] = latest_image
ap_updated_dict[model] = "Outdated"
sleep(1)
#print(ap_updated_dict)
if (ap_updated_dict["ea8300"] == "Up to Date") and (ap_updated_dict["ecw5410"] == "Up to Date"):
print("All AP FW loads up to date on CloudSDK")
logger.info("Newest builds on jfrog already loaded to CloudSDK")
else:
print("All new FW loads created on CloudSDK")
logger.info("New loads have been created on CloudSDK")
#print("Latest FW List:",ap_latest_dict)
####################################################################################
####################################################################################
############ Update FW and Run Test Cases on Each AP Variant #######################
####################################################################################
####################################################################################
for key in equipment_id_dict:
###Get Current AP Firmware and upgrade
customer_id = "2"
equipment_id = equipment_id_dict[key]
ap_fw = CloudSDK.ap_firmware(customer_id,equipment_id)
fw_model = ap_fw.partition("-")[0]
print("AP MODEL UNDER TEST IS", fw_model)
print ('Current AP Firmware:', ap_fw)
###Find Latest FW for Current AP Model and Get FW ID
latest_ap_image = ap_latest_dict[fw_model]
model_firmware_id = CloudSDK.get_firmware_id(latest_ap_image)
latest_firmware_id = str(model_firmware_id)
print("Latest FW ID is:",latest_firmware_id)
##Compare Latest and Current AP FW and Upgrade
if ap_fw == latest_ap_image:
print('FW does not require updating')
logger.info(fw_model + " does not require upgrade. Not performing sanity tests for this AP variant")
#raise SystemExit("No new FW loads available. Exiting Test...")
else:
print('FW needs updating')
#Upgrade AP firmware
upgrade_fw = CloudSDK.update_firmware(equipment_id, latest_firmware_id)
logger.info("Lab "+fw_model+" Requires FW update")
time.sleep(300)
###Create Test Run
today = str(date.today())
test_run_name = "Daily_Sanity_"+fw_model+"_"+today
client.create_testrun(name=test_run_name, case_ids=test_cases, project_id=projId)
rid = client.get_run_id(test_run_name="Daily_Sanity_"+fw_model+"_"+today)
print("TIP run ID is:", rid)
#Check if upgrade worked
ap_fw = CloudSDK.ap_firmware(customer_id, equipment_id)
print('Current AP Firmware:', ap_fw)
if ap_fw == latest_ap_image:
print("Upgrade Successful!")
client.update_testrail(case_id="2233", run_id=rid, status_id=1,msg='Upgrade successful')
logger.info("Lab "+fw_model+" upgrade SUCCESS. Proceeding with sanity testing for this AP variant")
else:
client.update_testrail(case_id="2233", run_id=rid, status_id=5, msg='Upgrade Failed')
logger.warning("Lab "+fw_model+" upgrade FAILED")
print("AP FW Upgrade Failed. Updating TestRail and skipping sanity testing for this AP variant...")
continue
###Set Proper AP Profile
test_profile_id = profile_info_dict[fw_model]["profile_id"]
print(test_profile_id)
ap_profile = CloudSDK.set_ap_profile(equipment_id,test_profile_id)
###Run Client Single Connectivity Test Cases
#TC 2237 - 2.4 GHz WPA2
test_case = "2237"
radio = "wiphy0"
station = ["sta2237"]
ssid_name = profile_info_dict[fw_model]["twoFourG_WPA2_SSID"]
ssid_psk = profile_info_dict[fw_model]["twoFourG_WPA2_PSK"]
security = "wpa2"
Test.Single_Client_Connectivity(radio, ssid_name, ssid_psk, security, station, test_case, rid)
time.sleep(10)
#TC 2420 - 2.4 GHz WPA
test_case = "2420"
radio = "wiphy0"
station = ["sta2420"]
ssid_name = profile_info_dict[fw_model]["twoFourG_WPA_SSID"]
ssid_psk = profile_info_dict[fw_model]["twoFourG_WPA_PSK"]
security = "wpa"
Test.Single_Client_Connectivity(radio, ssid_name, ssid_psk, security, station, test_case, rid)
time.sleep(10)
#TC 2234 - 2.4 GHz Open
test_case = "2234"
radio = "wiphy0"
station = ["sta2234"]
ssid_name = profile_info_dict[fw_model]["twoFourG_OPEN_SSID"]
ssid_psk = "BLANK"
security = "open"
Test.Single_Client_Connectivity(radio, ssid_name, ssid_psk, security, station, test_case, rid)
time.sleep(10)
#TC 2236 - 5 GHz WPA2
test_case = "2236"
radio = "wiphy3"
station = ["sta2236"]
ssid_name = profile_info_dict[fw_model]["fiveG_WPA2_SSID"]
ssid_psk = profile_info_dict[fw_model]["fiveG_WPA2_PSK"]
security = "wpa2"
Test.Single_Client_Connectivity(radio, ssid_name, ssid_psk, security, station, test_case, rid)
time.sleep(10)
#TC 2419 - 5 GHz WPA
test_case = "2419"
radio = "wiphy3"
station = ["sta2419"]
ssid_name = profile_info_dict[fw_model]["fiveG_WPA_SSID"]
ssid_psk = profile_info_dict[fw_model]["fiveG_WPA_PSK"]
security = "wpa"
Test.Single_Client_Connectivity(radio, ssid_name, ssid_psk, security, station, test_case, rid)
time.sleep(10)
#TC 2235 - 5 GHz Open
test_case = "2235"
radio = "wiphy3"
station = ["sta2235"]
ssid_name = profile_info_dict[fw_model]["fiveG_OPEN_SSID"]
ssid_psk = "BLANK"
security = "open"
Test.Single_Client_Connectivity(radio, ssid_name, ssid_psk, security, station, test_case, rid)
print(".....End of Sanity Test.....")
logger.info("End of Sanity Test run")

15
docker/Dockerfile Normal file
View File

@@ -0,0 +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-testing/docker/nightly.py /ci
COPY wlan-testing/docker/nightly_test_config.json /ci
ENTRYPOINT [ "/ci/nightly.py" ]

437
docker/nightly.py Executable file
View File

@@ -0,0 +1,437 @@
#!/usr/local/bin/python3
import re
import requests
import json
import logging
import argparse
from time import sleep, gmtime, strftime
from unittest.mock import Mock
# the below hack is only needed for local dev
# This should be replaced with a module and added to dockerfile
import sys
for folder in 'py-json','py-scripts':
if folder not in sys.path:
sys.path.append(f'../lanforge/lanforge/{folder}')
from LANforge.LFUtils import *
from sta_connect2 import StaConnect2
# Run from remote dev machine on ssh tunnel through to LANforge in TIP Lab, without testrails being used.
# SSH tunnel created with login to orchestrator with option: -L 10080:lf1:8080
# python3 ./nightly.py --skip-update-firmware --no-testrails --lanforge-ip-address localhost --lanforge-port-number 10080
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="tip-read")
parser.add_argument("--jfrog-user-password", type=str, help="jfrog user password",
default="tip-read")
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="password")
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('--skip-update-firmware', dest='update_firmware', action='store_false')
parser.add_argument('--no-testrails', dest='use_testrails', action='store_false')
parser.set_defaults(update_firmware=True)
parser.set_defaults(use_testrails=True)
command_line_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",
filename="nightly_cicd_sanity.log",
filemode="a")
formatter = logging.Formatter("%(asctime)s %(levelname)-8s %(message)s")
console = logging.StreamHandler()
console.setLevel(logging.INFO)
console.setFormatter(formatter)
logging.getLogger().addHandler(console)
logging.info("------------------------")
logging.info("nightly sanity run start")
# ap details, test data and test config
with open("nightly_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": "failure"
}
}
# Class to interact with Testrail; better to replace this with pytest\nunit
class TestRail_Client:
def __new__(cls, *args, **kwargs):
if command_line_args.use_testrails:
return super().__new__(cls, *args, **kwargs)
else:
mock = Mock()
mock.get_project_id.return_value = -1
mock.create_testrun.return_value = -1
return mock
def __init__(self):
self.user = command_line_args.testrail_user_id
self.password = command_line_args.testrail_user_password
self.__url = f"{command_line_args.testrail_base_url}/index.php?/api/v2/"
def send_get(self, uri, filepath=None):
"""Issue a GET request (read) against the API.
Args:
uri: The API method to call including parameters, e.g. get_case/1.
filepath: The path and file name for attachment download; used only
for "get_attachment/:attachment_id".
Returns:
A dict containing the result of the request.
"""
return self.__send_request("GET", uri, filepath)
def send_post(self, uri, data):
"""Issue a POST request (write) against the API.
Args:
uri: The API method to call, including parameters, e.g. add_case/1.
data: The data to submit as part of the request as a dict; strings
must be UTF-8 encoded. If adding an attachment, must be the
path to the file.
Returns:
A dict containing the result of the request.
"""
return self.__send_request("POST", uri, data)
def __send_request(self, method, uri, data):
url = self.__url + uri
headers = {
"Content-Type": "application/json"
}
logging.debug(f"Method: {method}; Url: {url}; Data: {data}")
if method == "POST":
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()
else:
payload = bytes(json.dumps(data), "utf-8")
response = requests.post(url, headers=headers, data=payload, auth=(self.user, self.password))
else:
response = requests.get(url, headers=headers, auth=(self.user, self.password))
logging.debug(f"headers: {headers}; response: {response.text}; response code: {response.status_code}")
if response.status_code > 201:
try:
error = response.json()
except: # response.content not formatted as JSON
error = str(response.content)
logging.info(f"TestRail API returned HTTP status code {response.status_code} with the following: ({error})")
return
else:
logging.debug(uri[:15])
if uri[:15] == "get_attachments": # Expecting file, not JSON
try:
logging.info (str(response.content))
open(data, "wb").write(response.content)
return (data)
except:
return ("Error saving attachment.")
else:
try:
return response.json()
except: # Nothing to return
return {}
def get_project_id(self, project_name):
"Get the project ID using project name"
projects = testrail.send_get("get_projects")
logging.debug(projects)
for project in projects:
if project["name"] == project_name:
return project["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 }
)
logging.info(f"Updating TestRail result: {result}")
except Exception as e:
logging.info(f"Exception in update_testrail(): {e}")
else:
logging.info(f"Updated test result for case: {case_id} in test run: {run_id} with msg: {msg}")
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}
)
logging.debug(result)
return result["id"]
# Class for jFrog Interaction
class jFrog_Client:
def __init__(self):
self.user = command_line_args.jfrog_user_id
self.password = command_line_args.jfrog_user_password
self.baseUrl = command_line_args.jfrog_base_url
def get_latest_image(self, model):
# todo handle auth errors
logging.debug(f"searching for the latest firmware on url: {model}")
response = requests.get(f"https://{self.baseUrl}/{model}/dev/", auth=(self.user, self.password))
return re.findall('href="(.+pending.+).tar.gz"', response.text)[-1]
def get_latest_image_url(self, model, latest_image):
return f"https://{self.user}:{self.password}@{self.baseUrl}/{model}/dev/{latest_image}.tar.gz"
# Class for CloudSDK Interaction via RestAPI
class CloudSDK_Client:
def __init__(self):
self.baseUrl = command_line_args.sdk_base_url
cloud_login_url = f"{self.baseUrl}/management/v1/oauth2/token"
payload = {
"userId": command_line_args.sdk_user_id,
"password": command_line_args.sdk_user_password
}
headers = {
"Content-Type": "application/json"
}
try:
token_response = requests.post(cloud_login_url, headers=headers, data=json.dumps(payload))
except requests.exceptions.RequestException as e:
raise SystemExit(f"Exiting Script! Cloud not get bearer token for reason: {e}")
token_data = token_response.json()
self.headers = {
"Authorization": f"Bearer {token_data['access_token']}"
}
def ap_firmware(self, customer_id, equipment_id):
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 (status_response.json())[2]["details"]["reportedSwVersion"]
def get_images(self, apModel):
getFW_url = f"{self.baseUrl}/portal/firmware/version/byEquipmentType?equipmentType=AP&modelId={apModel}"
status_response = requests.get(getFW_url, headers=self.headers)
logging.debug(status_response.json())
return([ version.get("versionName") for version in status_response.json()])
def firwmare_upload(self, apModel, latest_image, fw_url):
fw_upload_url = f"{self.baseUrl}/portal/firmware/version"
payload = {
"model_type": "FirmwareVersion",
"id": 0,
"equipmentType": "AP",
"modelId": apModel,
"versionName": latest_image,
"description": "",
"filename": fw_url,
"commit": latest_image.split("-")[-1],
"validationMethod": "MD5_CHECKSUM",
"validationCode": "19494befa87eb6bb90a64fd515634263",
"releaseDate": 1596192028877,
"createdTimestamp": 0,
"lastModifiedTimestamp": 0
}
self.headers["Content-Type"] = "application/json"
response = requests.post(fw_upload_url, headers=self.headers, data=json.dumps(payload))
self.headers.pop("Content-Type", None)
logging.debug(response)
return(response.json())
def get_firmware_id(self, latest_ap_image):
logging.debug(latest_ap_image)
fw_id_url = f"{self.baseUrl}/portal/firmware/version/byName?firmwareVersionName={latest_ap_image}"
response = requests.get(fw_id_url, headers=self.headers)
fw_data = response.json()
return fw_data["id"]
def update_firmware(self, equipment_id, latest_firmware_id):
url = f"{self.baseUrl}/portal/equipmentGateway/requestFirmwareUpdate?equipmentId={equipment_id}&firmwareVersionId={latest_firmware_id}"
response = requests.post(url, headers=self.headers)
logging.info(response.text)
def set_ap_profile(self, equipment_id, test_profile_id):
url = f"{self.baseUrl}/portal/equipment?equipmentId={equipment_id}"
response = requests.get(url, headers=self.headers)
logging.debug(response.json())
# Add Lab Profile ID to Equipment
equipment_info = response.json()
logging.debug(equipment_info)
equipment_info["profileId"] = test_profile_id
logging.debug(equipment_info)
# Update AP Info with Required Profile ID
url = f"{self.baseUrl}/portal/equipment"
self.headers["Content-Type"] = "application/json"
response = requests.put(url, headers=self.headers, data=json.dumps(equipment_info))
self.headers.pop("Content-Type", None)
logging.debug(response.text)
# instantiate clients and configuration
sdk: CloudSDK_Client = CloudSDK_Client()
testrail: TestRail_Client = TestRail_Client()
jFrog: jFrog_Client = jFrog_Client()
# 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 TEST_DATA["ap_models"].keys():
# Get latest firmware on jFrog and Cloud SDK
latest_image = jFrog.get_latest_image(model)
firmware_list_by_model = sdk.get_images(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!")
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.info(f"Upload Complete. {latest_image}; firmware ID is {fw_upload_status['id']}")
TEST_DATA["ap_models"][model]["firmware"] = latest_image
# Get Current AP Firmware and upgrade
ap_fw = sdk.ap_firmware(TEST_DATA["customer_id"], TEST_DATA["ap_models"][model]["id"])
fw_model = ap_fw.partition("-")[0]
latest_ap_image = TEST_DATA["ap_models"][fw_model]["firmware"]
model_firmware_id = sdk.get_firmware_id(latest_ap_image)
logging.info(f"Firmware: {ap_fw}; latest firmware ID is: {model_firmware_id}")
if ap_fw == latest_ap_image and command_line_args.update_firmware:
logging.info("Model does not require firmware upgrade, skipping sanity tests")
else:
if command_line_args.update_firmware:
firmware_update_case = [2831]
logging.info("Model requires firmware update, will update and sleep")
sdk.update_firmware(TEST_DATA["ap_models"][model]["id"], model_firmware_id)
sleep(300)
else:
firmware_update_case = []
test_cases_data = {
2832: { # 2.4 GHz Open
"radio": "wiphy4",
"station": ["sta2234"],
"ssid_name": TEST_DATA["ap_models"][fw_model]["info"]["twoFourG_OPEN_SSID"],
"ssid_psk": "BLANK",
"security": "open"
},
2835: { # 2.4 GHz WPA2
"radio": "wiphy4",
"station": ["sta2237"],
"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"
},
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'Nightly_model_{fw_model}_firmware_{ap_fw}_{strftime("%Y-%m-%d", gmtime())}'
testrail_project_id = testrail.get_project_id(project_name=command_line_args.testrail_project)
runId = testrail.create_testrun(name=test_run_name, case_ids=( [*test_cases_data] + firmware_update_case ), project_id=testrail_project_id)
logging.info(f"Testrail project id: {testrail_project_id}; run ID is: {runId}")
# Check if upgrade worked
if command_line_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 = 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():
if test_cases_data[testcase]["ssid_name"] != "skip":
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}")
staConnect = StaConnect2(command_line_args.lanforge_ip_address, command_line_args.lanforge_port_number, debug_ = False)
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.clean_all_sta = True
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=f"Test {results['message']}")
logging.info("----------------------")
logging.info("End of Sanity Test run")

22
docker/nightly_test_config.json Normal file
View File

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