Add files via upload

2025-10-31 02:48:11 +00:00 · 2020-08-31 12:44:38 -04:00
parent cd4b75001c
commit ae39a55dec
1 changed files with 536 additions and 0 deletions
--- a/CICD_AP_CLOUDSDK/CICD_Nightly_Sanity.py
+++ b/CICD_AP_CLOUDSDK/CICD_Nightly_Sanity.py
@@ -0,0 +1,536 @@
+#!/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")