diff --git a/.github/workflows/nightly.yml b/.github/workflows/nightly.yml index fd84574c6..432e18cc3 100644 --- a/.github/workflows/nightly.yml +++ b/.github/workflows/nightly.yml @@ -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 < 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") \ No newline at end of file diff --git a/docker/Dockerfile b/docker/Dockerfile new file mode 100644 index 000000000..2989edc6f --- /dev/null +++ b/docker/Dockerfile @@ -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" ] diff --git a/docker/nightly.py b/docker/nightly.py new file mode 100755 index 000000000..3944c29a8 --- /dev/null +++ b/docker/nightly.py @@ -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") diff --git a/docker/nightly_test_config.json b/docker/nightly_test_config.json new file mode 100644 index 000000000..e3075c9f2 --- /dev/null +++ b/docker/nightly_test_config.json @@ -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" + } + } + } +} \ No newline at end of file