diff --git a/.github/workflows/cloud-controller-build.yaml b/.github/workflows/cloud-controller-build.yaml
index de7187296..3dd9de42e 100644
--- a/.github/workflows/cloud-controller-build.yaml
+++ b/.github/workflows/cloud-controller-build.yaml
@@ -19,43 +19,23 @@ env:
testbeds: '[
{
"number": "ext-01",
- "version": "1.0.0-SNAPSHOT-2021-04-06"
+ "version": "1.1.0-SNAPSHOT-2021-04-27"
},
{
"number": "ext-02",
- "version": "1.0.0-SNAPSHOT-2021-04-06"
+ "version": "1.1.0-SNAPSHOT-2021-04-27"
},
{
"number": "ext-03",
- "version": "1.0.0-SNAPSHOT-2021-04-06"
+ "version": "1.1.0-RC1"
},
{
"number": "ext-04",
- "version": "1.0.0-SNAPSHOT-2021-04-06"
- },
- {
- "number": "ext-05",
- "version": "1.0.0-SNAPSHOT-2021-04-06T"
+ "version": "1.1.0-RC1"
},
{
"number": "01",
- "version": "1.0.0-SNAPSHOT-2021-04-06"
- },
- {
- "number": "02",
- "version": "1.0.0-SNAPSHOT-2021-04-06"
- },
- {
- "number": "04",
- "version": "1.0.0-SNAPSHOT-2021-04-06"
- },
- {
- "number": "05",
- "version": "1.0.0-SNAPSHOT-2021-04-06"
- },
- {
- "number": "15",
- "version": "1.0.0-SNAPSHOT-2021-04-06"
+ "version": "1.1.0-RC1"
}
]'
diff --git a/.github/workflows/quali.yml b/.github/workflows/quali.yml
new file mode 100644
index 000000000..c958f1c98
--- /dev/null
+++ b/.github/workflows/quali.yml
@@ -0,0 +1,33 @@
+name: Quali example pipeline
+
+on:
+ workflow_dispatch:
+
+jobs:
+ quali:
+ runs-on: ubuntu-latest
+ env:
+ CLOUDSHELL_URL: quali-cloudshell.lab.wlan.tip.build
+ CLOUDSHELL_USER: admin
+ CLOUDSHELL_PASSWORD: ${{ secrets.CLOUDSHELL_PASSWORD }}
+ steps:
+ - uses: actions/checkout@v2
+ - name: Set up Python
+ uses: actions/setup-python@v2
+ with:
+ python-version: '3.8'
+
+ - name: Install dependencies
+ run: pip install -r .quali/requirements.txt
+
+ - name: Start reservation
+ id: start
+ run: echo ::set-output name=res_id::$(python .quali/start_reservation.py)
+
+ - name: Run tests
+ run: |
+ echo "simulating test execution"
+ sleep 30
+
+ - name: Stop reservation
+ run: python .quali/stop_reservation.py ${{ steps.start.outputs.res_id }}
\ No newline at end of file
diff --git a/.quali/common.py b/.quali/common.py
new file mode 100644
index 000000000..a3f4b1d53
--- /dev/null
+++ b/.quali/common.py
@@ -0,0 +1,34 @@
+import os
+import time
+
+from cloudshell.api.cloudshell_api import CloudShellAPISession
+
+TIMEOUT=1200
+
+def get_session() -> CloudShellAPISession:
+ url = os.environ['CLOUDSHELL_URL']
+ user = os.environ['CLOUDSHELL_USER']
+ password = os.environ['CLOUDSHELL_PASSWORD']
+
+ return CloudShellAPISession(url, user, password, "Global")
+
+def __wait_for_status(session, res_id, field, target_status):
+ timer = 0
+ sleep_time = 5
+ while True:
+ status = session.GetReservationStatus(res_id).ReservationSlimStatus.__dict__[field]
+
+ if status == target_status:
+ break
+
+ if timer >= TIMEOUT:
+ raise RuntimeError(f'waiting for reservation to reach status {target_status} timed out')
+
+ time.sleep(sleep_time)
+ timer += sleep_time
+
+def wait_for_provisioning_status(session, res_id, target_status):
+ __wait_for_status(session, res_id, 'ProvisioningStatus', target_status)
+
+def wait_for_reservation_status(session, res_id, target_status):
+ __wait_for_status(session, res_id, 'Status', target_status)
diff --git a/.quali/requirements.txt b/.quali/requirements.txt
new file mode 100644
index 000000000..02044f576
--- /dev/null
+++ b/.quali/requirements.txt
@@ -0,0 +1 @@
+cloudshell-automation-api==2021.1.0.181140
\ No newline at end of file
diff --git a/.quali/start_reservation.py b/.quali/start_reservation.py
new file mode 100644
index 000000000..479eed961
--- /dev/null
+++ b/.quali/start_reservation.py
@@ -0,0 +1,31 @@
+import os
+
+from cloudshell.api.cloudshell_api import UpdateTopologyGlobalInputsRequest, UpdateTopologyRequirementsInputsRequest
+
+from common import wait_for_provisioning_status, get_session
+
+run_id = os.environ.get('GITHUB_RUN_NUMBER', 1)
+
+def main():
+ session = get_session()
+
+ reservation = session.CreateImmediateTopologyReservation(
+ reservationName=f'sanity-{run_id}',
+ owner=session.username,
+ durationInMinutes=60,
+ topologyFullPath='Basic Lab',
+ globalInputs=[
+ UpdateTopologyGlobalInputsRequest('Cloud Controller Version', '1.1.0-RC1'),
+ ],
+ requirementsInputs=[
+ UpdateTopologyRequirementsInputsRequest('Access Point', 'Ap.Wifi type', 'Wifi5', 'Attributes'),
+ UpdateTopologyRequirementsInputsRequest('Access Point', 'Ap.AP Model', 'ECW5410', 'Attributes')
+ ]
+ ).Reservation
+
+ print(reservation.Id)
+
+ wait_for_provisioning_status(session, reservation.Id, 'Ready')
+
+if __name__ == '__main__':
+ main()
diff --git a/.quali/stop_reservation.py b/.quali/stop_reservation.py
new file mode 100644
index 000000000..c9733d788
--- /dev/null
+++ b/.quali/stop_reservation.py
@@ -0,0 +1,12 @@
+import sys
+
+from common import wait_for_reservation_status, get_session
+
+def main():
+ session = get_session()
+ res_id = sys.argv[1]
+ session.EndReservation(res_id)
+ wait_for_reservation_status(session, res_id, 'Completed')
+
+if __name__ == '__main__':
+ main()
diff --git a/README.md b/README.md
index d9100cc00..47076c543 100644
--- a/README.md
+++ b/README.md
@@ -137,7 +137,7 @@ pip3 install xlsxwriter
```
#### Step 3
-Please ensure you follow the steps outlined in [here](./libs/README.md)
+Please ensure you follow the steps outlined in [here](./libs/controller/README.md)
```shell
Lets Install Controller Libraries, follow below steps:
diff --git a/docker/Dockerfile b/docker/Dockerfile
index 2242e4d45..6a8d112f1 100644
--- a/docker/Dockerfile
+++ b/docker/Dockerfile
@@ -1,15 +1,13 @@
FROM python:3.8
-WORKDIR /ci
-
-RUN apt update && apt install vim -y && rm -rf /var/lib/apt/lists/*
-RUN pip3 install requests xlsxwriter pandas
-
-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" ]
+WORKDIR /wlan-testing
+RUN apt update && rm -rf /var/lib/apt/lists/*
+RUN pip3 install pytest==6.2.2 bs4 paramiko xlsxwriter requests pandas influxdb scp allure-pytest
+RUN mkdir ~/.pip
+RUN echo "[global]" > ~/.pip/pip.conf
+RUN echo "index-url = https://pypi.org/simple" >> ~/.pip/pip.conf
+RUN echo "extra-index-url = https://tip-read:tip-read@tip.jfrog.io/artifactory/api/pypi/tip-wlan-python-pypi-local/simple" >> ~/.pip/pip.conf
+RUN echo "tip-wlan-cloud" > ~/.pip/requirements.txt
+RUN pip3 install -r ~/.pip/requirements.txt
+COPY lanforge /wlan-testing/lanforge
+COPY tests /wlan-testing/tests
+COPY libs /wlan-testing/libs
diff --git a/docker/README.md b/docker/README.md
new file mode 100644
index 000000000..1b754a33e
--- /dev/null
+++ b/docker/README.md
@@ -0,0 +1,19 @@
+## Docker based environment
+
+### Building a docker image
+
+From the root directory of this repository (wlan-testing) run the following command:
+```bash
+docker build -f ./docker/Dockerfile -t wlantest .
+```
+This will produce a docker image, which you can verify by running docker images command.
+
+### Running a docker image
+
+From the root directory of this repository (wlan-testing) run the following command. This command executes
+connectivity tests on a specific lab. **NOTE:** Use appropriate marker for your pytest execution,
+configuration.py and replace ${YOUR_ALLURE_RESULTS_DIR} with your allure result dir.
+
+```bash
+docker run -i -t -v $(YOUR_ALLURE_RESULT_DIR):/allure-results -v $(pwd)/configuration.py:/wlan-testing/configuration.py wlantest /bin/bash -c "cd tests; pytest -s -vvv --testbed=basic-02 -m client_connectivity_test --skip-testrail --alluredir=/allure-result"
+```
\ No newline at end of file
diff --git a/docker/nightly.py b/docker/nightly.py
deleted file mode 100755
index 56a8abcb2..000000000
--- a/docker/nightly.py
+++ /dev/null
@@ -1,438 +0,0 @@
-#!/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")
-
-# Initialize constants
-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, image):
- logging.debug(image)
- fw_id_url = f"{self.baseUrl}/portal/firmware/version/byName?firmwareVersionName={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
-
- # 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)
- TEST_DATA["ap_models"][model]["firmware"] = latest_image
- logging.info(f"Model: {model}; Latest firmware on jFrog: {latest_image}; Firmware on Cloud SDK: {firmware_list_by_model}")
-
- if latest_image in firmware_list_by_model:
- model_firmware_id = sdk.get_firmware_id(latest_image)
- logging.info(f"Latest firmware {latest_image} present on CloudSDK!")
- 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)
- model_firmware_id = fw_upload_status['id']
- logging.info(f"Upload Complete. {latest_image}; firmware ID is {model_firmware_id}")
-
- # Get Current AP Firmware and upgrade\run tests if needed
- ap_fw = sdk.ap_firmware(TEST_DATA["customer_id"], TEST_DATA["ap_models"][model]["id"])
- logging.info(f"Firmware: {ap_fw}; latest firmware is: {latest_image} with ID: {model_firmware_id}")
-
- if ap_fw == latest_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) # need to have a proper wait\retry here
- else:
- firmware_update_case = []
-
- test_cases_data = {
- 2832: { # 2.4 GHz Open
- "radio": "wiphy4",
- "station": ["sta2234"],
- "ssid_name": TEST_DATA["ap_models"][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"][model]["info"]["twoFourG_WPA2_SSID"],
- "ssid_psk": TEST_DATA["ap_models"][model]["info"]["twoFourG_WPA2_PSK"],
- "security": "wpa2"
- },
- 2833: { # 5 GHz Open
- "radio": "wiphy3",
- "station": ["sta2235"],
- "ssid_name": TEST_DATA["ap_models"][model]["info"]["fiveG_OPEN_SSID"],
- "ssid_psk": "BLANK",
- "security": "open"
- },
- 2834: { # 5 GHz WPA2
- "radio": "wiphy3",
- "station": ["sta2236"],
- "ssid_name": TEST_DATA["ap_models"][model]["info"]["fiveG_WPA2_SSID"],
- "ssid_psk": TEST_DATA["ap_models"][model]["info"]["fiveG_WPA2_PSK"],
- "security": "wpa2"
- },
- 2836: { # 5 GHz WPA
- "radio": "wiphy3",
- "station": ["sta2419"],
- "ssid_name": TEST_DATA["ap_models"][model]["info"]["fiveG_WPA_SSID"],
- "ssid_psk": TEST_DATA["ap_models"][model]["info"]["fiveG_WPA_PSK"],
- "security": "wpa"
- },
- 2837: { # 2.4 GHz WPA
- "radio": "wiphy0",
- "station": ["sta2420"],
- "ssid_name": TEST_DATA["ap_models"][model]["info"]["twoFourG_WPA_SSID"],
- "ssid_psk": TEST_DATA["ap_models"][model]["info"]["twoFourG_WPA_PSK"],
- "security": "wpa"
- }
- }
-
- # Create Test Run
- testrail_project_id = testrail.get_project_id(project_name=command_line_args.testrail_project)
- runId = testrail.create_testrun(
- name=f'Nightly_model_{model}_firmware_{ap_fw}_{strftime("%Y-%m-%d", gmtime())}',
- 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_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 results["message"] == "failure": # might want to fail all the other tests in testrails
- continue
-
- # Set Proper AP Profile
- test_profile_id = TEST_DATA["ap_models"][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": # to be refactored with pytest, good enough for now
- 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")
-logging.info("----------------------")
diff --git a/docker/nightly_test_config.json b/docker/nightly_test_config.json
deleted file mode 100644
index 6ca89951b..000000000
--- a/docker/nightly_test_config.json
+++ /dev/null
@@ -1,22 +0,0 @@
-{
- "customer_id": "2",
- "ap_models": {
- "ecw5410": {
- "id": "3",
- "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"
- }
- }
- }
-}
diff --git a/libs/apnos/apnos.py b/libs/apnos/apnos.py
index dadf73cf0..67da615ea 100644
--- a/libs/apnos/apnos.py
+++ b/libs/apnos/apnos.py
@@ -1,17 +1,25 @@
"""
APNOS Library : Used to execute SSH Commands in AP Using Direct-AP-SSH/ Jumphost-Serial Console
+
Currently Having Methods:
1. Get iwinfo
2. AP Manager Satus
3. Vif Config ssid's
4. Vif State ssid's
5. Get current Firmware
+
"""
+
import paramiko
from scp import SCPClient
import os
+import allure
+
+
class APNOS:
+
def __init__(self, credentials=None, pwd=os.getcwd()):
+ allure.attach(name="APNOS LIbrary: ", body=str(credentials))
self.owrt_args = "--prompt root@OpenAp -s serial --log stdout --user root --passwd openwifi"
if credentials is None:
print("No credentials Given")
@@ -34,14 +42,18 @@ class APNOS:
if str(stdout.read()).__contains__("False"):
print("Copying openwrt_ctl serial control Script...")
with SCPClient(client.get_transport()) as scp:
- scp.put(pwd+'openwrt_ctl.py', '~/cicd-git/openwrt_ctl.py') # Copy my_file.txt to the server
+ scp.put(pwd + 'openwrt_ctl.py', '~/cicd-git/openwrt_ctl.py') # Copy my_file.txt to the server
cmd = '[ -f ~/cicd-git/openwrt_ctl.py ] && echo "True" || echo "False"'
stdin, stdout, stderr = client.exec_command(cmd)
var = str(stdout.read())
+ print(var)
if var.__contains__("True"):
+ allure.attach(name="openwrt_ctl Setup", body=str(var))
print("APNOS Serial Setup OK")
else:
+ allure.attach(name="openwrt_ctl Setup", body=str(var))
print("APNOS Serial Setup Fail")
+
# Method to connect AP-CLI/ JUMPHOST-CLI
def ssh_cli_connect(self):
client = paramiko.SSHClient()
@@ -50,9 +62,12 @@ class APNOS:
self.username, self.ip, self.port, self.password))
client.connect(self.ip, username=self.username, password=self.password,
port=self.port, timeout=10, allow_agent=False, banner_timeout=200)
+
return client
+
def reboot(self):
client = self.ssh_cli_connect()
+
cmd = "reboot"
if self.mode:
cmd = f"cd ~/cicd-git/ && ./openwrt_ctl.py {self.owrt_args} -t {self.tty} --action " \
@@ -60,7 +75,9 @@ class APNOS:
stdin, stdout, stderr = client.exec_command(cmd)
output = stdout.read()
client.close()
+ allure.attach(name="AP Reboot", body=str(output))
return output
+
# Method to get the iwinfo status of AP using AP-CLI/ JUMPHOST-CLI
def iwinfo_status(self):
client = self.ssh_cli_connect()
@@ -71,7 +88,10 @@ class APNOS:
stdin, stdout, stderr = client.exec_command(cmd)
output = stdout.read()
client.close()
+ allure.attach(name="iwinfo Output Msg: ", body=str(output))
+ allure.attach(name="iwinfo config Err Msg: ", body=str(stderr))
return output
+
# Method to get the vif_config of AP using AP-CLI/ JUMPHOST-CLI
def get_vif_config(self):
client = self.ssh_cli_connect()
@@ -82,7 +102,11 @@ class APNOS:
stdin, stdout, stderr = client.exec_command(cmd)
output = stdout.read()
client.close()
+ allure.attach(name="vif config Output Msg: ", body=str(output))
+ allure.attach(name="vif config Err Msg: ", body=str(stderr))
+
return output
+
# Method to get the vif_state of AP using AP-CLI/ JUMPHOST-CLI
def get_vif_state(self):
client = self.ssh_cli_connect()
@@ -93,7 +117,10 @@ class APNOS:
stdin, stdout, stderr = client.exec_command(cmd)
output = stdout.read()
client.close()
+ allure.attach(name="vif state Output Msg: ", body=str(output))
+ allure.attach(name="vif state Err Msg: ", body=str(stderr))
return output
+
# Method to get the vif_config ssid's of AP using AP-CLI/ JUMPHOST-CLI
def get_vif_config_ssids(self):
stdout = self.get_vif_config()
@@ -102,7 +129,9 @@ class APNOS:
ssid = str(i).replace(" ", "").split(".")
if ssid[0].split(":")[0] == "b'ssid":
ssid_list.append(ssid[0].split(":")[1].replace("'", ""))
+ allure.attach(name="get_vif_config_ssids ", body=str(ssid_list))
return ssid_list
+
# Method to get the vif_state ssid's of AP using AP-CLI/ JUMPHOST-CLI
def get_vif_state_ssids(self):
stdout = self.get_vif_state()
@@ -111,7 +140,9 @@ class APNOS:
ssid = str(i).replace(" ", "").split(".")
if ssid[0].split(":")[0] == "b'ssid":
ssid_list.append(ssid[0].split(":")[1].replace("'", ""))
+ allure.attach(name="get_vif_state_ssids ", body=str(ssid_list))
return ssid_list
+
# Method to get the active firmware of AP using AP-CLI/ JUMPHOST-CLI
def get_active_firmware(self):
try:
@@ -129,8 +160,11 @@ class APNOS:
client.close()
except Exception as e:
print(e)
+ allure.attach(name="get_active_firmware - Exception ", body=str(e))
cli_active_fw = "Error"
+ allure.attach(name="get_active_firmware ", body=str(cli_active_fw))
return cli_active_fw
+
# Method to get the manager state of AP using AP-CLI/ JUMPHOST-CLI
def get_manager_state(self):
try:
@@ -146,5 +180,86 @@ class APNOS:
client.close()
except Exception as e:
print(e)
+ allure.attach(name="get_active_firmware - Exception ", body=str(e))
status = "Error"
+ allure.attach(name="get_active_firmware ", body=str(status))
return status
+
+ def get_serial_number(self):
+ try:
+ client = self.ssh_cli_connect()
+ cmd = "node | grep serial_number"
+ if self.mode:
+ cmd = f"cd ~/cicd-git/ && ./openwrt_ctl.py {self.owrt_args} -t {self.tty} --action " \
+ f"cmd --value \"{cmd}\" "
+ stdin, stdout, stderr = client.exec_command(cmd)
+ output = stdout.read()
+ output = output.decode('utf-8').splitlines()
+ allure.attach(name="get_serial_number output ", body=str(stderr))
+ serial = output[1].replace(" ", "").split("|")[1]
+ client.close()
+ except Exception as e:
+ print(e)
+ allure.attach(name="get_serial_number - Exception ", body=str(e))
+ serial = "Error"
+ allure.attach(name="get_serial_number ", body=str(serial))
+ return serial
+
+ def get_redirector(self):
+ try:
+ client = self.ssh_cli_connect()
+ cmd = "node | grep redirector_addr"
+ if self.mode:
+ cmd = f"cd ~/cicd-git/ && ./openwrt_ctl.py {self.owrt_args} -t {self.tty} --action " \
+ f"cmd --value \"{cmd}\" "
+ stdin, stdout, stderr = client.exec_command(cmd)
+ output = stdout.read()
+ status = output.decode('utf-8').splitlines()
+ allure.attach(name="get_redirector output ", body=str(stderr))
+ redirector = status[1].replace(" ", "").split("|")[1]
+ client.close()
+ except Exception as e:
+ print(e)
+ allure.attach(name="get_redirector - Exception ", body=str(e))
+ redirector = "Error"
+ allure.attach(name="get_redirector ", body=redirector)
+ return redirector
+
+ def run_generic_command(self, cmd=""):
+ allure.attach(name="run_generic_command ", body=cmd)
+ try:
+ client = self.ssh_cli_connect()
+ cmd = cmd
+ if self.mode:
+ cmd = f"cd ~/cicd-git/ && ./openwrt_ctl.py {self.owrt_args} -t {self.tty} --action " \
+ f"cmd --value \"{cmd}\" "
+ stdin, stdout, stderr = client.exec_command(cmd)
+ input = stdin.read().decode('utf-8').splitlines()
+ output = stdout.read().decode('utf-8').splitlines()
+ error = stderr.read().decode('utf-8').splitlines()
+ client.close()
+ except Exception as e:
+ print(e)
+ allure.attach(name="run_generic_command - Exception ", body=str(e))
+ input = "Error"
+ output = "Error"
+ error = "Error"
+ allure.attach(name="run_generic_command ", body=input)
+ allure.attach(name="run_generic_command ", body=str(output))
+ allure.attach(name="run_generic_command ", body=error)
+ return [input, output, error]
+
+
+if __name__ == '__main__':
+ obj = {
+ 'jumphost': True,
+ 'ip': "192.168.200.230",
+ 'username': "lanforge",
+ 'password': "lanforge",
+ 'port': 22,
+ 'jumphost_tty': '/dev/ttyAP1',
+
+ }
+ var = APNOS(credentials=obj)
+ r = var.get_redirector()
+ print(r)
\ No newline at end of file
diff --git a/libs/controller/controller.py b/libs/controller/controller.py
index 438285352..e5f77c4c0 100644
--- a/libs/controller/controller.py
+++ b/libs/controller/controller.py
@@ -4,8 +4,6 @@
1. controller_data/sdk_base_url
2. login credentials
"""
-import base64
-import datetime
import json
import re
import ssl
@@ -14,9 +12,10 @@ import urllib
import requests
import swagger_client
-from swagger_client import FirmwareManagementApi
-from swagger_client import EquipmentGatewayApi
from bs4 import BeautifulSoup
+from swagger_client import EquipmentGatewayApi
+from swagger_client import FirmwareManagementApi
+import allure
class ConfigureController:
@@ -25,7 +24,8 @@ class ConfigureController:
self.configuration = swagger_client.Configuration()
def set_credentials(self, controller_data=None):
- if dict(controller_data).keys().__contains__("username") and dict(controller_data).keys().__contains__("password"):
+ if dict(controller_data).keys().__contains__("username") and dict(controller_data).keys().__contains__(
+ "password"):
self.configuration.username = "support@example.com"
self.configuration.password = "support"
print("Login Credentials set to default: \n user_id: %s\n password: %s\n" % ("support@example.com",
@@ -108,6 +108,28 @@ class Controller(ConfigureController):
}
return self.login_client.get_access_token(request_body)
+ def refresh_instance(self):
+ # Connecting to Controller
+ self.api_client = swagger_client.ApiClient(self.configuration)
+ self.login_client = swagger_client.LoginApi(api_client=self.api_client)
+ self.bearer = self.get_bearer_token()
+
+ self.api_client.default_headers['Authorization'] = "Bearer " + self.bearer._access_token
+ self.status_client = swagger_client.StatusApi(api_client=self.api_client)
+ self.equipment_client = swagger_client.EquipmentApi(self.api_client)
+ self.profile_client = swagger_client.ProfileApi(self.api_client)
+ self.api_client.configuration.api_key_prefix = {
+ "Authorization": "Bearer " + self.bearer._access_token
+ }
+ self.api_client.configuration.refresh_api_key_hook = self.get_bearer_token()
+ self.ping_response = self.portal_ping()
+ self.default_profiles = {}
+ # print(self.bearer)
+ if self.ping_response._application_name != 'PortalServer':
+ print("Server not Reachable")
+ exit()
+ print("Connected to Controller Server")
+
def portal_ping(self):
return self.login_client.portal_ping()
@@ -140,8 +162,9 @@ class Controller(ConfigureController):
# Get the equipment id, of a equipment with a serial number
def get_equipment_id(self, serial_number=None):
equipment_data = self.get_equipment_by_customer_id(max_items=100)
- print(len(equipment_data))
+ # print(equipment_data)
for equipment in equipment_data:
+ print(equipment._id)
if equipment._serial == serial_number:
return equipment._id
@@ -150,6 +173,7 @@ class Controller(ConfigureController):
if equipment_id is None:
return None
data = self.equipment_client.get_equipment_by_id(equipment_id=equipment_id)
+ print(str(data._details._equipment_model))
return str(data._details._equipment_model)
# Needs Bug fix from swagger code generation side
@@ -173,7 +197,7 @@ class Controller(ConfigureController):
current_ap_fw = status_data[2]['details']['reportedSwVersion']
print(current_ap_fw)
return current_ap_fw
- except:
+ except Exception as e:
current_ap_fw = "error"
return False
@@ -322,13 +346,13 @@ class ProfileUtility:
self.push_profile_old_method(equipment_id=equipment_id)
self.delete_profile(profile_id=delete_ids)
- # This will delete all the profiles on an controller_tests instance, except the default profiles
+ # This will delete all the profiles on an controller instance, except the default profiles
def cleanup_profiles(self):
try:
self.get_default_profiles()
pagination_context = """{
"model_type": "PaginationContext",
- "maxItemsPerPage": 5000
+ "maxItemsPerPage": 10000
}"""
skip_delete_id = []
for i in self.default_profiles:
@@ -344,14 +368,13 @@ class ProfileUtility:
for i in self.default_profiles:
skip_delete_id.append(self.default_profiles[i]._id)
delete_ids = list(set(delete_ids) - set(delete_ids).intersection(set(skip_delete_id)))
+ print(delete_ids)
for i in delete_ids:
self.set_equipment_to_profile(profile_id=i)
- try:
- self.delete_profile(profile_id=delete_ids)
- except Exception as e:
- pass
+ self.delete_profile(profile_id=delete_ids)
status = True
- except:
+ except Exception as e:
+ print(e)
status = False
return status
@@ -393,11 +416,36 @@ class ProfileUtility:
Library method to create a new rf profile: Now using default profile
"""
- def set_rf_profile(self, profile_data=None):
- default_profile = self.default_profiles['rf']
- if profile_data is None:
- self.profile_creation_ids['rf'].append(default_profile._id)
- return default_profile
+ def set_rf_profile(self, profile_data=None, mode=None):
+ self.get_default_profiles()
+ if mode == "wifi5":
+ default_profile = self.default_profiles['rf']
+ default_profile._name = profile_data["name"]
+
+ default_profile._details["rfConfigMap"]["is2dot4GHz"]["rf"] = profile_data["name"]
+ default_profile._details["rfConfigMap"]["is5GHz"]["rf"] = profile_data["name"]
+ default_profile._details["rfConfigMap"]["is5GHzL"]["rf"] = profile_data["name"]
+ default_profile._details["rfConfigMap"]["is5GHzU"]["rf"] = profile_data["name"]
+ profile = self.profile_client.create_profile(body=default_profile)
+ self.profile_creation_ids['rf'].append(profile._id)
+ return profile
+
+ if mode == "wifi6":
+ default_profile = self.default_profiles['rf']
+ default_profile._name = profile_data["name"]
+ default_profile._details["rfConfigMap"]["is2dot4GHz"]["activeScanSettings"]["enabled"] = False
+ default_profile._details["rfConfigMap"]["is2dot4GHz"]["radioMode"] = 'modeAX'
+ default_profile._details["rfConfigMap"]["is5GHz"]["radioMode"] = 'modeAX'
+ default_profile._details["rfConfigMap"]["is5GHzL"]["radioMode"] = 'modeAX'
+ default_profile._details["rfConfigMap"]["is5GHzU"]["radioMode"] = 'modeAX'
+ default_profile._details["rfConfigMap"]["is2dot4GHz"]["rf"] = profile_data["name"]
+ default_profile._details["rfConfigMap"]["is5GHz"]["rf"] = profile_data["name"]
+ default_profile._details["rfConfigMap"]["is5GHzL"]["rf"] = profile_data["name"]
+ default_profile._details["rfConfigMap"]["is5GHzU"]["rf"] = profile_data["name"]
+ default_profile._name = profile_data["name"]
+ profile = self.profile_client.create_profile(body=default_profile)
+ self.profile_creation_ids['rf'].append(profile._id)
+ return profile
"""
method call: used to create a ssid profile with the given parameters
@@ -592,6 +640,7 @@ class ProfileUtility:
return default_profile
"""
+
method to push the profile to the given equipment
"""
@@ -671,31 +720,32 @@ class JFrogUtility:
def __init__(self, credentials=None):
if credentials is None:
exit()
- self.user = credentials["username"]
- self.password = credentials["password"]
self.jfrog_url = credentials["jfrog-base-url"]
self.build = credentials["build"]
self.branch = credentials["branch"]
ssl._create_default_https_context = ssl._create_unverified_context
- def get_latest_build(self, model=None):
+ def get_build(self, model=None, version=None):
jfrog_url = self.jfrog_url + "/" + model + "/" + self.branch + "/"
- auth = str(
- base64.b64encode(
- bytes('%s:%s' % (self.user, self.password), 'utf-8')
- ),
- 'ascii'
- ).strip()
- headers = {'Authorization': 'Basic ' + auth}
''' FIND THE LATEST FILE NAME'''
- req = urllib.request.Request(jfrog_url, headers=headers)
+ print(jfrog_url)
+ req = urllib.request.Request(jfrog_url)
response = urllib.request.urlopen(req)
+ # print(response)
html = response.read()
soup = BeautifulSoup(html, features="html.parser")
- last_link = soup.find_all('a', href=re.compile(self.build))[-1]
- latest_file = last_link['href']
- latest_fw = latest_file.replace('.tar.gz', '')
+ if self.branch == "trunk":
+ self.build = model
+ last_link = soup.find_all('a', href=re.compile(self.build))
+ latest_fw = None
+ for i in last_link:
+
+ if str(i['href']).__contains__(version):
+ latest_fw = i['href']
+
+ latest_fw = latest_fw.replace('.tar.gz', '')
+ print("Using Firmware Image: ", latest_fw)
return latest_fw
@@ -710,7 +760,13 @@ class JFrogUtility:
class FirmwareUtility(JFrogUtility):
- def __init__(self, sdk_client=None, jfrog_credentials=None, controller_data=None, customer_id=None):
+ def __init__(self,
+ sdk_client=None,
+ jfrog_credentials=None,
+ controller_data=None,
+ customer_id=None,
+ model=None,
+ version=None):
super().__init__(credentials=jfrog_credentials)
if sdk_client is None:
sdk_client = Controller(controller_data=controller_data, customer_id=customer_id)
@@ -718,13 +774,17 @@ class FirmwareUtility(JFrogUtility):
self.firmware_client = FirmwareManagementApi(api_client=sdk_client.api_client)
self.jfrog_client = JFrogUtility(credentials=jfrog_credentials)
self.equipment_gateway_client = EquipmentGatewayApi(api_client=sdk_client.api_client)
+ self.model = model
+ self.fw_version = version
- def get_latest_fw_version(self, model="ecw5410"):
+ def get_fw_version(self):
# Get The equipment model
- self.latest_fw = self.get_latest_build(model=model)
+ self.latest_fw = self.get_build(model=self.model, version=self.fw_version)
return self.latest_fw
def upload_fw_on_cloud(self, fw_version=None, force_upload=False):
+ print("Upload fw version :", fw_version)
+ # force_upload = True
# if fw_latest available and force upload is False -- Don't upload
# if fw_latest available and force upload is True -- Upload
# if fw_latest is not available -- Upload
@@ -745,11 +805,10 @@ class FirmwareUtility(JFrogUtility):
"equipmentType": "AP",
"modelId": fw_version.split("-")[0],
"versionName": fw_version + ".tar.gz",
- "description": "ECW5410 FW VERSION TEST",
- "filename": "https://tip.jfrog.io/artifactory/tip-wlan-ap-firmware/" + fw_version.split("-")[
- 0] + "/dev/" + fw_version + ".tar.gz",
- "commit": fw_version.split("-")[5]
+ "description": fw_version + " FW VERSION",
+ "filename": self.jfrog_url + "/" + self.model + "/" + self.branch + "/" + fw_version + ".tar.gz",
}
+ print(firmware_data["filename"])
firmware_id = self.firmware_client.create_firmware_version(body=firmware_data)
print("Force Upload :", force_upload, " Uploaded the Image")
return firmware_id._id
@@ -760,23 +819,25 @@ class FirmwareUtility(JFrogUtility):
exit()
if (force_upgrade is True) or (self.should_upgrade_ap_fw(equipment_id=equipment_id)):
model = self.sdk_client.get_model_name(equipment_id=equipment_id).lower()
- latest_fw = self.get_latest_fw_version(model=model)
+ latest_fw = self.get_fw_version()
firmware_id = self.upload_fw_on_cloud(fw_version=latest_fw, force_upload=force_upload)
time.sleep(5)
try:
obj = self.equipment_gateway_client.request_firmware_update(equipment_id=equipment_id,
firmware_version_id=firmware_id)
+
print("Request firmware upgrade Success! waiting for 300 sec")
time.sleep(300)
except Exception as e:
+ print(e)
obj = False
return obj
# Write the upgrade fw logic here
def should_upgrade_ap_fw(self, equipment_id=None):
- current_fw = self.get_current_fw_version(equipment_id=equipment_id)
- model = self.sdk_client.get_model_name(equipment_id=equipment_id).lower()
- latest_fw = self.get_latest_fw_version(model=model)
+ current_fw = self.sdk_client.get_ap_firmware_old_method(equipment_id=equipment_id)
+ latest_fw = self.get_fw_version()
+ print(self.model, current_fw, latest_fw)
if current_fw == latest_fw:
return False
else:
@@ -790,7 +851,24 @@ class FirmwareUtility(JFrogUtility):
firmware_version_name=fw_version + ".tar.gz")
firmware_version = firmware_version._id
print("Firmware ID: ", firmware_version)
- except:
+ except Exception as e:
+ print(e)
firmware_version = False
print("firmware not available: ", firmware_version)
return firmware_version
+
+
+if __name__ == '__main__':
+ """
+ Examples to Try Out
+ """
+ controller = {
+ 'url': "https://wlan-portal-svc-nola-ext-03.cicd.lab.wlan.tip.build", # API base url for the controller
+ 'username': 'support@example.com',
+ 'password': 'support',
+ 'version': "1.1.0-SNAPSHOT",
+ 'commit_date': "2021-04-27"
+ }
+ sdk_client = Controller(controller_data=controller)
+ # Use Library/ Method Here
+ sdk_client.disconnect_Controller()
diff --git a/tests/README.md b/tests/README.md
index e2a1407a9..05b02619b 100644
--- a/tests/README.md
+++ b/tests/README.md
@@ -1,10 +1,10 @@
-# wlan-testing framework Information
+## wlan-testing framework Information
-## pytest uses setup > test > tear_down
-#### Fixtures : Code that needs to be part of more than 1 test cases, Setup and teardown is Implemented in Fixtures
+**_pytest uses setup > test > tear_down_**
+**_Fixtures : Code that needs to be part of more than 1 test cases, Setup and teardown is Implemented in Fixtures_**
-###Test cases are structured across different directories
+### Test cases are structured across different directories
```
├── wlan-testing
├── tests /* Root directory for tests */
@@ -88,8 +88,8 @@ Read the README.md in each e2e directory to get sample test case.
```
-#### For any Clarifications, regarding Framework,
-#### Email : shivam.thakur@candelatech.com
+**_For any Clarifications, regarding Framework,_**
+**_Email : shivam.thakur@candelatech.com_**
diff --git a/tests/configuration.py b/tests/configuration.py
index 7283f61af..9120ea34d 100644
--- a/tests/configuration.py
+++ b/tests/configuration.py
@@ -1,6 +1,44 @@
CONFIGURATION = {
- "interop": {
+
+ "ext-05": {
+ "controller": {
+ 'url': "https://wlan-portal-svc-nola-ext-04.cicd.lab.wlan.tip.build", # API base url for the controller
+ 'username': 'support@example.com',
+ 'password': 'support',
+ 'version': '1.0.0-SNAPSHOT',
+ 'commit_date': '2021-03-01'
+ },
+ 'access_point': [
+ {
+ 'model': 'ecw5410',
+ 'mode': 'wifi5',
+ 'serial': '903cb3944817',
+ 'jumphost': True,
+ 'ip': "192.168.200.82",
+ 'username': "lanforge",
+ 'password': "lanforge",
+ 'port': 22,
+ 'jumphost_tty': '/dev/ttyAP1',
+ 'version': "ecw5410-2021-04-26-pending-3fc41fa"
+ }
+ ],
+ "traffic_generator": {
+ "name": "lanforge",
+ "details": {
+ "ip": "192.168.200.82",
+ "port": 8080,
+ "2.4G-Radio": ["wiphy0"],
+ "5G-Radio": ["wiphy1"],
+ "AX-Radio": ["wiphy2"],
+ "upstream": "eth1",
+ "2.4G-Station-Name": "wlan0",
+ "5G-Station-Name": "wlan1",
+
+ }
+ }
+ },
+ "interop": {
"controller": {
'url': "https://wlan-portal-svc-nola-01.cicd.lab.wlan.tip.build", # API base url for the controller
'username': 'support@example.com',
@@ -34,6 +72,7 @@ CONFIGURATION = {
}
+
FIRMWARE = {
# jFrog parameters
"JFROG":
diff --git a/tests/conftest.py b/tests/conftest.py
index 4abc7afb8..5ee729576 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -76,6 +76,41 @@ def pytest_addoption(parser):
default=False,
help="Stop using Testrails"
)
+ parser.addoption(
+ "--exit-on-fail",
+ action="store_true",
+ default=False,
+ help="use to stop execution if failure"
+ )
+
+ # Perfecto Parameters
+ parser.addini("perfectoURL", "Cloud URL")
+ parser.addini("securityToken", "Security Token")
+ parser.addini("platformName-iOS", "iOS Platform")
+ parser.addini("platformName-android", "Android Platform")
+ parser.addini("model-iOS", "iOS Devices")
+ parser.addini("model-android", "Android Devices")
+ parser.addini("bundleId-iOS", "iOS Devices")
+ parser.addini("bundleId-iOS-Settings", "iOS Settings App")
+ parser.addini("appPackage-android", "Android Devices")
+ parser.addini("wifi-SSID-5gl-Pwd", "Wifi 5g Password")
+ parser.addini("wifi-SSID-2g-Pwd", "Wifi 2g Password")
+ parser.addini("Default-SSID-5gl-perfecto-b", "Wifi 5g AP Name")
+ parser.addini("Default-SSID-2g-perfecto-b", "Wifi 2g AP Name")
+ parser.addini("Default-SSID-perfecto-b", "Wifi AP Name")
+ parser.addini("bundleId-iOS-Ping", "Ping Bundle ID")
+ parser.addini("browserType-iOS", "Mobile Browser Name")
+ parser.addini("projectName", "Project Name")
+ parser.addini("projectVersion", "Project Version")
+ parser.addini("jobName", "CI Job Name")
+ parser.addini("jobNumber", "CI Job Number")
+ parser.addini("reportTags", "Report Tags")
+ parser.addoption(
+ "--access-points-perfecto",
+ # nargs="+",
+ default=["Perfecto"],
+ help="list of access points to test"
+ )
#Perfecto Parameters
parser.addini("perfectoURL", "Cloud URL")
@@ -136,6 +171,7 @@ Instantiate Objects for Test session
def instantiate_controller(request, testbed):
try:
sdk_client = Controller(controller_data=CONFIGURATION[testbed]["controller"])
+
def teardown_session():
print("\nTest session Completed")
sdk_client.disconnect_Controller()
@@ -158,8 +194,10 @@ def instantiate_testrail(request):
@pytest.fixture(scope="session")
-def instantiate_firmware(instantiate_controller, instantiate_jFrog):
- firmware_client = FirmwareUtility(jfrog_credentials=instantiate_jFrog, sdk_client=instantiate_controller)
+def instantiate_firmware(instantiate_controller, instantiate_jFrog, testbed):
+ firmware_client = FirmwareUtility(jfrog_credentials=instantiate_jFrog, sdk_client=instantiate_controller,
+ model=CONFIGURATION[testbed]["access_point"][0]["model"],
+ version=CONFIGURATION[testbed]["access_point"][0]["version"])
yield firmware_client
@@ -188,6 +226,11 @@ def setup_lanforge():
yield True
+@pytest.fixture(scope="session")
+def exit_on_fail(request):
+ yield request.config.getoption("--exit-on-fail")
+
+
@pytest.fixture(scope="session")
def setup_perfecto_devices(request):
yield True
@@ -198,15 +241,42 @@ def test_cases():
yield TEST_CASES
+@pytest.fixture(scope="session")
+def apnos_obj(get_configuration, testbed):
+ yield APNOS(get_configuration[testbed]['access_point'][0])
+
+
+@pytest.fixture(scope="session")
+def instantiate_access_point(testbed):
+ APNOS(CONFIGURATION[testbed]['access_point'][0], pwd="../libs/apnos/")
+ yield True
+
+
@pytest.fixture(scope="function")
-def test_access_point(testbed):
- ap_ssh = APNOS(CONFIGURATION[testbed]['access_point'][0])
- status = ap_ssh.get_manager_state()
+def access_point_connectivity(apnos_obj, get_configuration, testbed):
+ ap_conn = {}
+ config_serial = get_configuration[testbed]['access_point'][0]['serial']
+ ap_serial = apnos_obj.get_serial_number()
+ ap_conn["serial"] = True
+ if ap_serial != config_serial:
+ ap_conn["serial"] = False
+
+ ap_conn["redir"] = False
+ ap_redir = apnos_obj.get_redirector()
+
+ # Compare with something ...
+
+ ap_conn["mgr"] = False
+ status = apnos_obj.get_manager_state()
if "ACTIVE" not in status:
+ apnos_obj.run_generic_command(cmd="service opensync restart")
time.sleep(30)
- ap_ssh = APNOS(CONFIGURATION[testbed]['access_point'][0])
- status = ap_ssh.get_manager_state()
- yield status
+ status = apnos_obj.get_manager_state()
+ if "ACTIVE" in status:
+ ap_conn["mgr"] = True
+ else:
+ ap_conn["mgr"] = True
+ yield ap_conn
@pytest.fixture(scope="session")
@@ -219,10 +289,11 @@ def setup_profile_data(testbed):
profile_data[mode][security] = {}
for radio in "2G", "5G":
profile_data[mode][security][radio] = {}
- name_string = f"{'Sanity'}-{model}-{radio}_{security}_{mode}"
+ name_string = f"{'Sanity'}-{testbed}-{model}-{radio}_{security}_{mode}"
+ ssid_name = f"{'Sanity'}-{model}-{radio}_{security}_{mode}"
passkey_string = f"{radio}-{security}_{mode}"
profile_data[mode][security][radio]["profile_name"] = name_string
- profile_data[mode][security][radio]["ssid_name"] = name_string
+ profile_data[mode][security][radio]["ssid_name"] = ssid_name
if mode == "VLAN":
profile_data[mode][security][radio]["vlan"] = 100
else:
@@ -264,10 +335,11 @@ def get_markers(request, get_security_flags):
@pytest.fixture(scope="session")
-def get_latest_firmware(testbed, instantiate_firmware):
+def get_latest_firmware(instantiate_firmware):
try:
- latest_firmware = instantiate_firmware.get_latest_fw_version(CONFIGURATION[testbed]["access_point"][0]["model"])
- except:
+ latest_firmware = instantiate_firmware.get_fw_version()
+ except Exception as e:
+ print(e)
latest_firmware = False
yield latest_firmware
@@ -287,3 +359,8 @@ def check_ap_firmware_ssh(testbed):
@pytest.fixture(scope="session")
def radius_info():
yield RADIUS_SERVER_DATA
+
+
+@pytest.fixture(scope="session")
+def get_configuration():
+ yield CONFIGURATION
diff --git a/tests/controller_tests/test_api_login.py b/tests/controller_tests/test_api_login.py
index 2ba85b095..b5e337701 100644
--- a/tests/controller_tests/test_api_login.py
+++ b/tests/controller_tests/test_api_login.py
@@ -8,9 +8,9 @@ from configuration import CONFIGURATION
@pytest.mark.sanity
@pytest.mark.sdk_version_check
-def test_cloud_sdk_version(instantiate_cloudsdk, testbed, test_cases, instantiate_testrail, instantiate_project):
+def test_cloud_sdk_version(instantiate_controller, testbed, test_cases, instantiate_testrail, instantiate_project):
try:
- response = instantiate_cloudsdk.portal_ping()
+ response = instantiate_controller.portal_ping()
if CONFIGURATION[testbed]['controller']['version'] == response._project_version:
PASS = True
instantiate_testrail.update_testrail(case_id=test_cases["cloud_ver"], run_id=instantiate_project,
diff --git a/tests/e2e/basic/README.md b/tests/e2e/basic/README.md
index 8d9bff67f..0209a09a9 100644
--- a/tests/e2e/basic/README.md
+++ b/tests/e2e/basic/README.md
@@ -30,7 +30,7 @@ import pytest
pytestmark = [pytest.mark.test_featureA, pytest.mark.bridge]
-@pytest.mark.wifi_capacity_test
+@pytest.mark.test_featureA
@pytest.mark.wifi5
@pytest.mark.wifi6
@pytest.mark.parametrize(
@@ -53,7 +53,7 @@ class TestFeatureABridge(object):
security_key = profile_data["security_key"]
security = "wpa"
upstream = get_lanforge_data["lanforge_bridge_port"]
- radio = get_lanforge_data["lanforge_2g"]
+ radio = get_lanforge_data["lanforge_2dot4g"]
# Write Your test case Here
PASS = True
assert PASS
@@ -84,7 +84,7 @@ class TestFeatureABridge(object):
security_key = profile_data["security_key"]
security = "wpa2"
upstream = get_lanforge_data["lanforge_bridge_port"]
- radio = get_lanforge_data["lanforge_2g"]
+ radio = get_lanforge_data["lanforge_2dot4g"]
# Write Your test case Here
PASS = True
assert PASS
diff --git a/tests/e2e/basic/_basic_test_setup/setup__e2e_resources.py b/tests/e2e/basic/_basic_test_setup/setup__e2e_resources.py
index bcf20b0e1..8af8183d6 100644
--- a/tests/e2e/basic/_basic_test_setup/setup__e2e_resources.py
+++ b/tests/e2e/basic/_basic_test_setup/setup__e2e_resources.py
@@ -5,6 +5,11 @@
"""
import pytest
+@pytest.mark.configure_lanforge
+def test_configure_lanforge(configure_lanforge):
+
+ assert True
+
@pytest.mark.sanity
@pytest.mark.bridge
diff --git a/tests/e2e/basic/_basic_test_setup/setup_e2e_bridge.py b/tests/e2e/basic/_basic_test_setup/setup_e2e_bridge.py
index e54844471..044e72cb5 100644
--- a/tests/e2e/basic/_basic_test_setup/setup_e2e_bridge.py
+++ b/tests/e2e/basic/_basic_test_setup/setup_e2e_bridge.py
@@ -3,6 +3,8 @@
Details: bridge mode setup
"""
+import time
+
import pytest
@@ -162,9 +164,11 @@ class TestSetupBridge:
instantiate_testrail.update_testrail(case_id=test_cases["bridge_vifs"], run_id=instantiate_project,
status_id=1,
msg='profile pushed successfully')
+ time.sleep(100)
assert setup_profiles['bridge_vifs']
else:
instantiate_testrail.update_testrail(case_id=test_cases["bridge_vifs"], run_id=instantiate_project,
status_id=5,
msg='Failed to push profile')
+ time.sleep(100)
assert False
diff --git a/tests/e2e/basic/_basic_test_setup/setup_e2e_nat.py b/tests/e2e/basic/_basic_test_setup/setup_e2e_nat.py
index dd6230154..9f60ca2f4 100644
--- a/tests/e2e/basic/_basic_test_setup/setup_e2e_nat.py
+++ b/tests/e2e/basic/_basic_test_setup/setup_e2e_nat.py
@@ -3,6 +3,7 @@
Details: nat mode setup
"""
+import time
import pytest
@@ -163,9 +164,11 @@ class TestSetupnat:
instantiate_testrail.update_testrail(case_id=test_cases["nat_vifs"], run_id=instantiate_project,
status_id=1,
msg='profile pushed successfully')
+ time.sleep(100)
assert setup_profiles['nat_vifs']
else:
instantiate_testrail.update_testrail(case_id=test_cases["nat_vifs"], run_id=instantiate_project,
status_id=5,
msg='Failed to push profile')
+ time.sleep(100)
assert False
diff --git a/tests/e2e/basic/_basic_test_setup/setup_e2e_vlan.py b/tests/e2e/basic/_basic_test_setup/setup_e2e_vlan.py
index 71c3be24e..e3a814b28 100644
--- a/tests/e2e/basic/_basic_test_setup/setup_e2e_vlan.py
+++ b/tests/e2e/basic/_basic_test_setup/setup_e2e_vlan.py
@@ -3,6 +3,7 @@
Details: vlan mode setup
"""
+import time
import pytest
@@ -164,9 +165,13 @@ class TestSetupvlan:
instantiate_testrail.update_testrail(case_id=test_cases["vlan_vifs"], run_id=instantiate_project,
status_id=1,
msg='profile pushed successfully')
+ time.sleep(100)
+
assert setup_profiles['vlan_vifs']
+
else:
instantiate_testrail.update_testrail(case_id=test_cases["vlan_vifs"], run_id=instantiate_project,
status_id=5,
msg='Failed to push profile')
+ time.sleep(100)
assert False
diff --git a/tests/e2e/basic/client_connectivity_test/test_bridge_mode.py b/tests/e2e/basic/client_connectivity_test/test_bridge_mode.py
index ae2973982..1797b9c8e 100644
--- a/tests/e2e/basic/client_connectivity_test/test_bridge_mode.py
+++ b/tests/e2e/basic/client_connectivity_test/test_bridge_mode.py
@@ -48,7 +48,7 @@ class TestBridgeModeClientConnectivity(object):
@pytest.mark.wpa
@pytest.mark.twog
def test_client_wpa_2g(self, request, get_lanforge_data, setup_profile_data, instantiate_testrail,
- instantiate_project, test_cases):
+ instantiate_controller, instantiate_project, test_cases, exit_on_fail):
profile_data = setup_profile_data["BRIDGE"]["WPA"]["2G"]
station_names = []
for i in range(0, int(request.config.getini("num_stations"))):
@@ -66,7 +66,7 @@ class TestBridgeModeClientConnectivity(object):
staConnect.dut_security = "wpa"
staConnect.station_names = station_names
staConnect.sta_prefix = get_lanforge_data["lanforge_2dot4g_prefix"]
- staConnect.runtime_secs = 10
+ staConnect.runtime_secs = 40
staConnect.bringup_time_sec = 60
staConnect.cleanup_on_exit = True
# staConnect.cleanup()
@@ -74,6 +74,8 @@ class TestBridgeModeClientConnectivity(object):
staConnect.start()
print("napping %f sec" % staConnect.runtime_secs)
time.sleep(staConnect.runtime_secs)
+ if exit_on_fail and staConnect.passes():
+ pytest.exit("Test Case Failed! exit_on_fail Exit")
staConnect.stop()
staConnect.cleanup()
run_results = staConnect.get_result_list()
@@ -84,18 +86,20 @@ class TestBridgeModeClientConnectivity(object):
if staConnect.passes():
instantiate_testrail.update_testrail(case_id=test_cases["2g_wpa_bridge"], run_id=instantiate_project,
status_id=1,
- msg='2G WPA Client Connectivity Passed successfully - bridge mode')
+ msg='2G WPA Client Connectivity Passed successfully - bridge mode' + str(run_results))
else:
instantiate_testrail.update_testrail(case_id=test_cases["2g_wpa_bridge"], run_id=instantiate_project,
status_id=5,
- msg='2G WPA Client Connectivity Failed - bridge mode')
+ msg='2G WPA Client Connectivity Failed - bridge mode' + str(run_results))
+
+ instantiate_controller.refresh_instance()
assert staConnect.passes()
# C2420
@pytest.mark.wpa
@pytest.mark.fiveg
def test_client_wpa_5g(self, request, get_lanforge_data, setup_profile_data, instantiate_project,
- instantiate_testrail, test_cases):
+ instantiate_controller, instantiate_testrail, test_cases, exit_on_fail):
profile_data = setup_profile_data["BRIDGE"]["WPA"]["5G"]
station_names = []
for i in range(0, int(request.config.getini("num_stations"))):
@@ -112,7 +116,7 @@ class TestBridgeModeClientConnectivity(object):
staConnect.dut_security = "wpa"
staConnect.station_names = station_names
staConnect.sta_prefix = get_lanforge_data["lanforge_5g_prefix"]
- staConnect.runtime_secs = 10
+ staConnect.runtime_secs = 40
staConnect.bringup_time_sec = 60
staConnect.cleanup_on_exit = True
# staConnect.cleanup()
@@ -120,6 +124,8 @@ class TestBridgeModeClientConnectivity(object):
staConnect.start()
print("napping %f sec" % staConnect.runtime_secs)
time.sleep(staConnect.runtime_secs)
+ if exit_on_fail and staConnect.passes():
+ pytest.exit("Test Case Failed! exit_on_fail Exit")
staConnect.stop()
staConnect.cleanup()
run_results = staConnect.get_result_list()
@@ -130,18 +136,20 @@ class TestBridgeModeClientConnectivity(object):
if staConnect.passes():
instantiate_testrail.update_testrail(case_id=test_cases["5g_wpa_bridge"], run_id=instantiate_project,
status_id=1,
- msg='5G WPA Client Connectivity Passed successfully - bridge mode')
+ msg='5G WPA Client Connectivity Passed successfully - bridge mode' + str(run_results))
else:
instantiate_testrail.update_testrail(case_id=test_cases["5g_wpa_bridge"], run_id=instantiate_project,
status_id=5,
- msg='5G WPA Client Connectivity Failed - bridge mode')
+ msg='5G WPA Client Connectivity Failed - bridge mode' + str(run_results))
+
+ instantiate_controller.refresh_instance()
assert staConnect.passes()
# C2419
@pytest.mark.wpa2_personal
@pytest.mark.twog
def test_client_wpa2_personal_2g(self, request, get_lanforge_data, setup_profile_data, instantiate_project,
- instantiate_testrail, test_cases):
+ instantiate_controller, instantiate_testrail, test_cases, exit_on_fail):
profile_data = setup_profile_data["BRIDGE"]["WPA2_P"]["2G"]
station_names = []
for i in range(0, int(request.config.getini("num_stations"))):
@@ -158,7 +166,7 @@ class TestBridgeModeClientConnectivity(object):
staConnect.dut_security = "wpa2"
staConnect.station_names = station_names
staConnect.sta_prefix = get_lanforge_data["lanforge_2dot4g_prefix"]
- staConnect.runtime_secs = 10
+ staConnect.runtime_secs = 40
staConnect.bringup_time_sec = 60
staConnect.cleanup_on_exit = True
# staConnect.cleanup()
@@ -166,6 +174,8 @@ class TestBridgeModeClientConnectivity(object):
staConnect.start()
print("napping %f sec" % staConnect.runtime_secs)
time.sleep(staConnect.runtime_secs)
+ if exit_on_fail and staConnect.passes():
+ pytest.exit("Test Case Failed! exit_on_fail Exit")
staConnect.stop()
staConnect.cleanup()
run_results = staConnect.get_result_list()
@@ -176,18 +186,20 @@ class TestBridgeModeClientConnectivity(object):
if staConnect.passes():
instantiate_testrail.update_testrail(case_id=test_cases["2g_wpa2_bridge"], run_id=instantiate_project,
status_id=1,
- msg='2G WPA2 Client Connectivity Passed successfully - bridge mode')
+ msg='2G WPA2 Client Connectivity Passed successfully - bridge mode' + str(run_results))
else:
instantiate_testrail.update_testrail(case_id=test_cases["2g_wpa2_bridge"], run_id=instantiate_project,
status_id=5,
- msg='2G WPA2 Client Connectivity Failed - bridge mode')
+ msg='2G WPA2 Client Connectivity Failed - bridge mode' + str(run_results))
+
+ instantiate_controller.refresh_instance()
assert staConnect.passes()
# C2237
@pytest.mark.wpa2_personal
@pytest.mark.fiveg
def test_client_wpa2_personal_5g(self, request, get_lanforge_data, setup_profile_data, instantiate_project,
- instantiate_testrail, test_cases):
+ instantiate_controller, instantiate_testrail, test_cases, exit_on_fail):
profile_data = setup_profile_data["BRIDGE"]["WPA2_P"]["5G"]
station_names = []
for i in range(0, int(request.config.getini("num_stations"))):
@@ -204,7 +216,7 @@ class TestBridgeModeClientConnectivity(object):
staConnect.dut_security = "wpa2"
staConnect.station_names = station_names
staConnect.sta_prefix = get_lanforge_data["lanforge_5g_prefix"]
- staConnect.runtime_secs = 10
+ staConnect.runtime_secs = 40
staConnect.bringup_time_sec = 60
staConnect.cleanup_on_exit = True
# staConnect.cleanup()
@@ -212,6 +224,8 @@ class TestBridgeModeClientConnectivity(object):
staConnect.start()
print("napping %f sec" % staConnect.runtime_secs)
time.sleep(staConnect.runtime_secs)
+ if exit_on_fail and staConnect.passes():
+ pytest.exit("Test Case Failed! exit_on_fail Exit")
staConnect.stop()
staConnect.cleanup()
run_results = staConnect.get_result_list()
@@ -222,11 +236,12 @@ class TestBridgeModeClientConnectivity(object):
if staConnect.passes():
instantiate_testrail.update_testrail(case_id=test_cases["5g_wpa2_bridge"], run_id=instantiate_project,
status_id=1,
- msg='5G WPA2 Client Connectivity Passed successfully - bridge mode')
+ msg='5G WPA2 Client Connectivity Passed successfully - bridge mode' + str(run_results))
else:
instantiate_testrail.update_testrail(case_id=test_cases["5g_wpa2_bridge"], run_id=instantiate_project,
status_id=5,
- msg='5G WPA2 Client Connectivity Failed - bridge mode')
+ msg='5G WPA2 Client Connectivity Failed - bridge mode' + str(run_results))
+ instantiate_controller.refresh_instance()
assert staConnect.passes()
# C2236
@@ -234,7 +249,7 @@ class TestBridgeModeClientConnectivity(object):
@pytest.mark.twog
@pytest.mark.radius
def test_client_wpa2_enterprise_2g(self, request, get_lanforge_data, setup_profile_data, instantiate_project,
- instantiate_testrail, radius_info, test_cases):
+ instantiate_controller, instantiate_testrail, radius_info, test_cases, exit_on_fail):
profile_data = setup_profile_data["BRIDGE"]["WPA2_E"]["2G"]
station_names = []
for i in range(0, int(request.config.getini("num_stations"))):
@@ -251,15 +266,16 @@ class TestBridgeModeClientConnectivity(object):
eap_connect.eap = "TTLS"
eap_connect.identity = radius_info["user"]
eap_connect.ttls_passwd = radius_info["password"]
- eap_connect.runtime_secs = 10
+ eap_connect.runtime_secs = 40
eap_connect.setup()
eap_connect.start()
print("napping %f sec" % eap_connect.runtime_secs)
time.sleep(eap_connect.runtime_secs)
+ if exit_on_fail and eap_connect.passes():
+ pytest.exit("Test Case Failed! exit_on_fail Exit")
eap_connect.stop()
try:
eap_connect.cleanup()
- eap_connect.cleanup()
except:
pass
run_results = eap_connect.get_result_list()
@@ -271,11 +287,12 @@ class TestBridgeModeClientConnectivity(object):
instantiate_testrail.update_testrail(case_id=test_cases["2g_eap_bridge"], run_id=instantiate_project,
status_id=1,
msg='5G WPA2 ENTERPRISE Client Connectivity Passed successfully - '
- 'bridge mode')
+ 'bridge mode' + str(run_results))
else:
instantiate_testrail.update_testrail(case_id=test_cases["2g_eap_bridge"], run_id=instantiate_project,
status_id=5,
- msg='5G WPA2 ENTERPRISE Client Connectivity Failed - bridge mode')
+ msg='5G WPA2 ENTERPRISE Client Connectivity Failed - bridge mode' + str(run_results))
+ instantiate_controller.refresh_instance()
assert eap_connect.passes()
# C5214
@@ -283,7 +300,7 @@ class TestBridgeModeClientConnectivity(object):
@pytest.mark.fiveg
@pytest.mark.radius
def test_client_wpa2_enterprise_5g(self, request, get_lanforge_data, setup_profile_data, instantiate_project,
- instantiate_testrail, radius_info, test_cases):
+ instantiate_controller, instantiate_testrail, radius_info, test_cases, exit_on_fail):
profile_data = setup_profile_data["BRIDGE"]["WPA2_E"]["5G"]
station_names = []
for i in range(0, int(request.config.getini("num_stations"))):
@@ -300,14 +317,16 @@ class TestBridgeModeClientConnectivity(object):
eap_connect.eap = "TTLS"
eap_connect.identity = radius_info["user"]
eap_connect.ttls_passwd = radius_info["password"]
- eap_connect.runtime_secs = 10
+ eap_connect.runtime_secs = 40
eap_connect.setup()
eap_connect.start()
print("napping %f sec" % eap_connect.runtime_secs)
time.sleep(eap_connect.runtime_secs)
+ if exit_on_fail and eap_connect.passes():
+ pytest.exit("Test Case Failed! exit_on_fail Exit")
eap_connect.stop()
try:
- eap_connect.cleanup()
+
eap_connect.cleanup()
except:
pass
@@ -320,11 +339,13 @@ class TestBridgeModeClientConnectivity(object):
instantiate_testrail.update_testrail(case_id=test_cases["5g_eap_bridge"], run_id=instantiate_project,
status_id=1,
msg='5G WPA2 ENTERPRISE Client Connectivity Passed successfully - '
- 'bridge mode')
+ 'bridge mode' + str(run_results))
else:
instantiate_testrail.update_testrail(case_id=test_cases["5g_eap_bridge"], run_id=instantiate_project,
status_id=5,
- msg='5G WPA2 ENTERPRISE Client Connectivity Failed - bridge mode')
+ msg='5G WPA2 ENTERPRISE Client Connectivity Failed - bridge mode' + str(run_results))
+
+ instantiate_controller.refresh_instance()
assert eap_connect.passes()
@pytest.mark.modify_ssid
@@ -334,7 +355,7 @@ class TestBridgeModeClientConnectivity(object):
indirect=True
)
def test_modify_ssid(self, request, update_ssid, get_lanforge_data, setup_profile_data, instantiate_testrail,
- instantiate_project, test_cases):
+ instantiate_project, test_cases, instantiate_controller, exit_on_fail):
profile_data = setup_profile_data["BRIDGE"]["WPA"]["5G"]
station_names = []
for i in range(0, int(request.config.getini("num_stations"))):
@@ -351,7 +372,7 @@ class TestBridgeModeClientConnectivity(object):
staConnect.dut_security = "wpa"
staConnect.station_names = station_names
staConnect.sta_prefix = get_lanforge_data["lanforge_5g_prefix"]
- staConnect.runtime_secs = 10
+ staConnect.runtime_secs = 40
staConnect.bringup_time_sec = 60
staConnect.cleanup_on_exit = True
# staConnect.cleanup()
@@ -359,6 +380,8 @@ class TestBridgeModeClientConnectivity(object):
staConnect.start()
print("napping %f sec" % staConnect.runtime_secs)
time.sleep(staConnect.runtime_secs)
+ if exit_on_fail and staConnect.passes():
+ pytest.exit("Test Case Failed! exit_on_fail Exit")
staConnect.stop()
staConnect.cleanup()
run_results = staConnect.get_result_list()
@@ -370,9 +393,11 @@ class TestBridgeModeClientConnectivity(object):
instantiate_testrail.update_testrail(case_id=test_cases["bridge_ssid_update"], run_id=instantiate_project,
status_id=1,
msg='5G WPA Client Connectivity Passed successfully - bridge mode '
- 'updated ssid')
+ 'updated ssid' + str(run_results))
else:
instantiate_testrail.update_testrail(case_id=test_cases["bridge_ssid_update"], run_id=instantiate_project,
status_id=5,
- msg='5G WPA Client Connectivity Failed - bridge mode updated ssid')
+ msg='5G WPA Client Connectivity Failed - bridge mode updated ssid' + str(run_results))
+
+ instantiate_controller.refresh_instance()
assert staConnect.passes()
diff --git a/tests/e2e/basic/client_connectivity_test/test_nat_mode.py b/tests/e2e/basic/client_connectivity_test/test_nat_mode.py
index c09d97be5..c2b0ee7df 100644
--- a/tests/e2e/basic/client_connectivity_test/test_nat_mode.py
+++ b/tests/e2e/basic/client_connectivity_test/test_nat_mode.py
@@ -47,7 +47,7 @@ class TestNatModeClientConnectivity(object):
@pytest.mark.wpa
@pytest.mark.twog
def test_client_wpa_2g(self, request, get_lanforge_data, setup_profile_data, instantiate_testrail,
- instantiate_project, test_cases):
+ instantiate_controller, instantiate_project, test_cases, exit_on_fail):
profile_data = setup_profile_data["NAT"]["WPA"]["2G"]
station_names = []
for i in range(0, int(request.config.getini("num_stations"))):
@@ -65,7 +65,7 @@ class TestNatModeClientConnectivity(object):
staConnect.dut_security = "wpa"
staConnect.station_names = station_names
staConnect.sta_prefix = get_lanforge_data["lanforge_2dot4g_prefix"]
- staConnect.runtime_secs = 10
+ staConnect.runtime_secs = 40
staConnect.bringup_time_sec = 60
staConnect.cleanup_on_exit = True
# staConnect.cleanup()
@@ -83,18 +83,21 @@ class TestNatModeClientConnectivity(object):
if staConnect.passes():
instantiate_testrail.update_testrail(case_id=test_cases["2g_wpa_nat"], run_id=instantiate_project,
status_id=1,
- msg='2G WPA Client Connectivity Passed successfully - nat mode')
+ msg='2G WPA Client Connectivity Passed successfully - nat mode' + str(run_results))
else:
instantiate_testrail.update_testrail(case_id=test_cases["2g_wpa_nat"], run_id=instantiate_project,
status_id=5,
- msg='2G WPA Client Connectivity Failed - nat mode')
+ msg='2G WPA Client Connectivity Failed - nat mode' + str(run_results))
+ if exit_on_fail:
+ pytest.exit("Test Case Failed! exit_on_fail Exit")
+ instantiate_controller.refresh_instance()
assert staConnect.passes()
# C2420
@pytest.mark.wpa
@pytest.mark.fiveg
def test_client_wpa_5g(self, request, get_lanforge_data, setup_profile_data, instantiate_project,
- instantiate_testrail, test_cases):
+ instantiate_controller, instantiate_testrail, test_cases, exit_on_fail):
profile_data = setup_profile_data["NAT"]["WPA"]["5G"]
station_names = []
for i in range(0, int(request.config.getini("num_stations"))):
@@ -111,7 +114,7 @@ class TestNatModeClientConnectivity(object):
staConnect.dut_security = "wpa"
staConnect.station_names = station_names
staConnect.sta_prefix = get_lanforge_data["lanforge_5g_prefix"]
- staConnect.runtime_secs = 10
+ staConnect.runtime_secs = 40
staConnect.bringup_time_sec = 60
staConnect.cleanup_on_exit = True
# staConnect.cleanup()
@@ -129,18 +132,21 @@ class TestNatModeClientConnectivity(object):
if staConnect.passes():
instantiate_testrail.update_testrail(case_id=test_cases["5g_wpa_nat"], run_id=instantiate_project,
status_id=1,
- msg='5G WPA Client Connectivity Passed successfully - nat mode')
+ msg='5G WPA Client Connectivity Passed successfully - nat mode' + str(run_results))
else:
instantiate_testrail.update_testrail(case_id=test_cases["5g_wpa_nat"], run_id=instantiate_project,
status_id=5,
- msg='5G WPA Client Connectivity Failed - nat mode')
+ msg='5G WPA Client Connectivity Failed - nat mode' + str(run_results))
+ if exit_on_fail:
+ pytest.exit("Test Case Failed! exit_on_fail Exit")
+ instantiate_controller.refresh_instance()
assert staConnect.passes()
# C2419
@pytest.mark.wpa2_personal
@pytest.mark.twog
def test_client_wpa2_personal_2g(self, request, get_lanforge_data, setup_profile_data, instantiate_project,
- instantiate_testrail, test_cases):
+ instantiate_controller, instantiate_testrail, test_cases, exit_on_fail):
profile_data = setup_profile_data["NAT"]["WPA2_P"]["2G"]
station_names = []
for i in range(0, int(request.config.getini("num_stations"))):
@@ -157,7 +163,7 @@ class TestNatModeClientConnectivity(object):
staConnect.dut_security = "wpa2"
staConnect.station_names = station_names
staConnect.sta_prefix = get_lanforge_data["lanforge_2dot4g_prefix"]
- staConnect.runtime_secs = 10
+ staConnect.runtime_secs = 40
staConnect.bringup_time_sec = 60
staConnect.cleanup_on_exit = True
# staConnect.cleanup()
@@ -175,18 +181,21 @@ class TestNatModeClientConnectivity(object):
if staConnect.passes():
instantiate_testrail.update_testrail(case_id=test_cases["2g_wpa2_nat"], run_id=instantiate_project,
status_id=1,
- msg='2G WPA2 Client Connectivity Passed successfully - nat mode')
+ msg='2G WPA2 Client Connectivity Passed successfully - nat mode' + str(run_results))
else:
instantiate_testrail.update_testrail(case_id=test_cases["2g_wpa2_nat"], run_id=instantiate_project,
status_id=5,
- msg='2G WPA2 Client Connectivity Failed - nat mode')
+ msg='2G WPA2 Client Connectivity Failed - nat mode' + str(run_results))
+ if exit_on_fail:
+ pytest.exit("Test Case Failed! exit_on_fail Exit")
+ instantiate_controller.refresh_instance()
assert staConnect.passes()
# C2237
@pytest.mark.wpa2_personal
@pytest.mark.fiveg
def test_client_wpa2_personal_5g(self, request, get_lanforge_data, setup_profile_data, instantiate_project,
- instantiate_testrail, test_cases):
+ instantiate_controller, instantiate_testrail, test_cases, exit_on_fail):
profile_data = setup_profile_data["NAT"]["WPA2_P"]["5G"]
station_names = []
for i in range(0, int(request.config.getini("num_stations"))):
@@ -203,7 +212,7 @@ class TestNatModeClientConnectivity(object):
staConnect.dut_security = "wpa2"
staConnect.station_names = station_names
staConnect.sta_prefix = get_lanforge_data["lanforge_5g_prefix"]
- staConnect.runtime_secs = 10
+ staConnect.runtime_secs = 40
staConnect.bringup_time_sec = 60
staConnect.cleanup_on_exit = True
# staConnect.cleanup()
@@ -221,11 +230,14 @@ class TestNatModeClientConnectivity(object):
if staConnect.passes():
instantiate_testrail.update_testrail(case_id=test_cases["5g_wpa2_nat"], run_id=instantiate_project,
status_id=1,
- msg='5G WPA2 Client Connectivity Passed successfully - nat mode')
+ msg='5G WPA2 Client Connectivity Passed successfully - nat mode' + str(run_results))
else:
instantiate_testrail.update_testrail(case_id=test_cases["5g_wpa2_nat"], run_id=instantiate_project,
status_id=5,
- msg='5G WPA2 Client Connectivity Failed - nat mode')
+ msg='5G WPA2 Client Connectivity Failed - nat mode' + str(run_results))
+ if exit_on_fail:
+ pytest.exit("Test Case Failed! exit_on_fail Exit")
+ instantiate_controller.refresh_instance()
assert staConnect.passes()
# C2236
@@ -233,7 +245,7 @@ class TestNatModeClientConnectivity(object):
@pytest.mark.twog
@pytest.mark.radius
def test_client_wpa2_enterprise_2g(self, request, get_lanforge_data, setup_profile_data, instantiate_project,
- instantiate_testrail, radius_info, test_cases):
+ instantiate_controller, instantiate_testrail, radius_info, test_cases, exit_on_fail):
profile_data = setup_profile_data["NAT"]["WPA2_E"]["2G"]
station_names = []
for i in range(0, int(request.config.getini("num_stations"))):
@@ -250,7 +262,7 @@ class TestNatModeClientConnectivity(object):
eap_connect.eap = "TTLS"
eap_connect.identity = radius_info["user"]
eap_connect.ttls_passwd = radius_info["password"]
- eap_connect.runtime_secs = 10
+ eap_connect.runtime_secs = 40
eap_connect.setup()
eap_connect.start()
print("napping %f sec" % eap_connect.runtime_secs)
@@ -270,11 +282,14 @@ class TestNatModeClientConnectivity(object):
instantiate_testrail.update_testrail(case_id=test_cases["2g_eap_nat"], run_id=instantiate_project,
status_id=1,
msg='5G WPA2 ENTERPRISE Client Connectivity Passed successfully - '
- 'nat mode')
+ 'nat mode' + str(run_results))
else:
instantiate_testrail.update_testrail(case_id=test_cases["2g_eap_nat"], run_id=instantiate_project,
status_id=5,
- msg='5G WPA2 ENTERPRISE Client Connectivity Failed - nat mode')
+ msg='5G WPA2 ENTERPRISE Client Connectivity Failed - nat mode' + str(run_results))
+ if exit_on_fail:
+ pytest.exit("Test Case Failed! exit_on_fail Exit")
+ instantiate_controller.refresh_instance()
assert eap_connect.passes()
# C5214
@@ -282,7 +297,7 @@ class TestNatModeClientConnectivity(object):
@pytest.mark.fiveg
@pytest.mark.radius
def test_client_wpa2_enterprise_5g(self, request, get_lanforge_data, setup_profile_data, instantiate_project,
- instantiate_testrail, radius_info, test_cases):
+ instantiate_controller, instantiate_testrail, radius_info, test_cases, exit_on_fail):
profile_data = setup_profile_data["NAT"]["WPA2_E"]["5G"]
station_names = []
for i in range(0, int(request.config.getini("num_stations"))):
@@ -299,7 +314,7 @@ class TestNatModeClientConnectivity(object):
eap_connect.eap = "TTLS"
eap_connect.identity = radius_info["user"]
eap_connect.ttls_passwd = radius_info["password"]
- eap_connect.runtime_secs = 10
+ eap_connect.runtime_secs = 40
eap_connect.setup()
eap_connect.start()
print("napping %f sec" % eap_connect.runtime_secs)
@@ -319,11 +334,14 @@ class TestNatModeClientConnectivity(object):
instantiate_testrail.update_testrail(case_id=test_cases["5g_eap_nat"], run_id=instantiate_project,
status_id=1,
msg='5G WPA2 ENTERPRISE Client Connectivity Passed successfully - '
- 'nat mode')
+ 'nat mode' + str(run_results))
else:
instantiate_testrail.update_testrail(case_id=test_cases["5g_eap_nat"], run_id=instantiate_project,
status_id=5,
- msg='5G WPA2 ENTERPRISE Client Connectivity Failed - nat mode')
+ msg='5G WPA2 ENTERPRISE Client Connectivity Failed - nat mode' + str(run_results))
+ if exit_on_fail:
+ pytest.exit("Test Case Failed! exit_on_fail Exit")
+ instantiate_controller.refresh_instance()
assert eap_connect.passes()
@pytest.mark.modify_ssid
@@ -333,7 +351,7 @@ class TestNatModeClientConnectivity(object):
indirect=True
)
def test_modify_ssid(self, request, update_ssid, get_lanforge_data, setup_profile_data, instantiate_testrail,
- instantiate_project, test_cases):
+ instantiate_project, test_cases, instantiate_controller, exit_on_fail):
profile_data = setup_profile_data["NAT"]["WPA"]["5G"]
station_names = []
for i in range(0, int(request.config.getini("num_stations"))):
@@ -350,7 +368,7 @@ class TestNatModeClientConnectivity(object):
staConnect.dut_security = "wpa"
staConnect.station_names = station_names
staConnect.sta_prefix = get_lanforge_data["lanforge_5g_prefix"]
- staConnect.runtime_secs = 10
+ staConnect.runtime_secs = 40
staConnect.bringup_time_sec = 60
staConnect.cleanup_on_exit = True
# staConnect.cleanup()
@@ -369,9 +387,12 @@ class TestNatModeClientConnectivity(object):
instantiate_testrail.update_testrail(case_id=test_cases["nat_ssid_update"], run_id=instantiate_project,
status_id=1,
msg='5G WPA Client Connectivity Passed successfully - nat mode '
- 'updated ssid')
+ 'updated ssid' + str(run_results))
else:
instantiate_testrail.update_testrail(case_id=test_cases["nat_ssid_update"], run_id=instantiate_project,
status_id=5,
- msg='5G WPA Client Connectivity Failed - nat mode updated ssid')
+ msg='5G WPA Client Connectivity Failed - nat mode updated ssid' + str(run_results))
+ if exit_on_fail:
+ pytest.exit("Test Case Failed! exit_on_fail Exit")
+ instantiate_controller.refresh_instance()
assert staConnect.passes()
diff --git a/tests/e2e/basic/client_connectivity_test/test_vlan_mode.py b/tests/e2e/basic/client_connectivity_test/test_vlan_mode.py
index 8889ec4dc..27c01dfc3 100644
--- a/tests/e2e/basic/client_connectivity_test/test_vlan_mode.py
+++ b/tests/e2e/basic/client_connectivity_test/test_vlan_mode.py
@@ -46,7 +46,7 @@ class TestVlanModeClientConnectivity(object):
@pytest.mark.wpa
@pytest.mark.twog
def test_client_wpa_2g(self, request, get_lanforge_data, setup_profile_data,
- instantiate_testrail, instantiate_project, test_cases):
+ instantiate_controller, instantiate_testrail, instantiate_project, test_cases):
profile_data = setup_profile_data["VLAN"]["WPA"]["2G"]
station_names = []
for i in range(0, int(request.config.getini("num_stations"))):
@@ -64,7 +64,7 @@ class TestVlanModeClientConnectivity(object):
staConnect.dut_security = "wpa"
staConnect.station_names = station_names
staConnect.sta_prefix = get_lanforge_data["lanforge_2dot4g_prefix"]
- staConnect.runtime_secs = 10
+ staConnect.runtime_secs = 40
staConnect.bringup_time_sec = 60
staConnect.cleanup_on_exit = True
staConnect.setup()
@@ -81,18 +81,19 @@ class TestVlanModeClientConnectivity(object):
if staConnect.passes():
instantiate_testrail.update_testrail(case_id=test_cases["2g_wpa_vlan"], run_id=instantiate_project,
status_id=1,
- msg='2G WPA Client Connectivity Passed successfully - vlan mode')
+ msg='2G WPA Client Connectivity Passed successfully - vlan mode' + str(run_results))
else:
instantiate_testrail.update_testrail(case_id=test_cases["2g_wpa_vlan"], run_id=instantiate_project,
status_id=5,
- msg='2G WPA Client Connectivity Failed - vlan mode')
+ msg='2G WPA Client Connectivity Failed - vlan mode' + str(run_results))
+ instantiate_controller.refresh_instance()
assert staConnect.passes()
# C2420
@pytest.mark.wpa
@pytest.mark.fiveg
def test_client_wpa_5g(self, request, get_lanforge_data, setup_profile_data,
- instantiate_project, instantiate_testrail, test_cases):
+ instantiate_controller, instantiate_project, instantiate_testrail, test_cases):
profile_data = setup_profile_data["VLAN"]["WPA"]["5G"]
station_names = []
for i in range(0, int(request.config.getini("num_stations"))):
@@ -109,7 +110,7 @@ class TestVlanModeClientConnectivity(object):
staConnect.dut_security = "wpa"
staConnect.station_names = station_names
staConnect.sta_prefix = get_lanforge_data["lanforge_5g_prefix"]
- staConnect.runtime_secs = 10
+ staConnect.runtime_secs = 40
staConnect.bringup_time_sec = 60
staConnect.cleanup_on_exit = True
# staConnect.cleanup()
@@ -127,18 +128,19 @@ class TestVlanModeClientConnectivity(object):
if staConnect.passes():
instantiate_testrail.update_testrail(case_id=test_cases["5g_wpa_vlan"], run_id=instantiate_project,
status_id=1,
- msg='5G WPA Client Connectivity Passed successfully - vlan mode')
+ msg='5G WPA Client Connectivity Passed successfully - vlan mode' + str(run_results))
else:
instantiate_testrail.update_testrail(case_id=test_cases["5g_wpa_vlan"], run_id=instantiate_project,
status_id=5,
- msg='5G WPA Client Connectivity Failed - vlan mode')
+ msg='5G WPA Client Connectivity Failed - vlan mode' + str(run_results))
+ instantiate_controller.refresh_instance()
assert staConnect.passes()
# C2419
@pytest.mark.wpa2_personal
@pytest.mark.twog
def test_client_wpa2_personal_2g(self, request, get_lanforge_data, setup_profile_data,
- instantiate_project, instantiate_testrail, test_cases):
+ instantiate_controller, instantiate_project, instantiate_testrail, test_cases):
profile_data = setup_profile_data["VLAN"]["WPA2_P"]["2G"]
station_names = []
for i in range(0, int(request.config.getini("num_stations"))):
@@ -155,7 +157,7 @@ class TestVlanModeClientConnectivity(object):
staConnect.dut_security = "wpa2"
staConnect.station_names = station_names
staConnect.sta_prefix = get_lanforge_data["lanforge_2dot4g_prefix"]
- staConnect.runtime_secs = 10
+ staConnect.runtime_secs = 40
staConnect.bringup_time_sec = 60
staConnect.cleanup_on_exit = True
# staConnect.cleanup()
@@ -173,18 +175,19 @@ class TestVlanModeClientConnectivity(object):
if staConnect.passes():
instantiate_testrail.update_testrail(case_id=test_cases["2g_wpa2_vlan"], run_id=instantiate_project,
status_id=1,
- msg='2G WPA2 Client Connectivity Passed successfully - vlan mode')
+ msg='2G WPA2 Client Connectivity Passed successfully - vlan mode' + str(run_results))
else:
instantiate_testrail.update_testrail(case_id=test_cases["2g_wpa2_vlan"], run_id=instantiate_project,
status_id=5,
- msg='2G WPA2 Client Connectivity Failed - vlan mode')
+ msg='2G WPA2 Client Connectivity Failed - vlan mode' + str(run_results))
+ instantiate_controller.refresh_instance()
assert staConnect.passes()
# C2237
@pytest.mark.wpa2_personal
@pytest.mark.fiveg
def test_client_wpa2_personal_5g(self, request, get_lanforge_data, setup_profile_data,
- instantiate_project, instantiate_testrail, test_cases):
+ instantiate_controller, instantiate_project, instantiate_testrail, test_cases):
profile_data = setup_profile_data["VLAN"]["WPA2_P"]["5G"]
station_names = []
for i in range(0, int(request.config.getini("num_stations"))):
@@ -201,7 +204,7 @@ class TestVlanModeClientConnectivity(object):
staConnect.dut_security = "wpa2"
staConnect.station_names = station_names
staConnect.sta_prefix = get_lanforge_data["lanforge_5g_prefix"]
- staConnect.runtime_secs = 10
+ staConnect.runtime_secs = 40
staConnect.bringup_time_sec = 60
staConnect.cleanup_on_exit = True
# staConnect.cleanup()
@@ -219,11 +222,12 @@ class TestVlanModeClientConnectivity(object):
if staConnect.passes():
instantiate_testrail.update_testrail(case_id=test_cases["5g_wpa2_vlan"], run_id=instantiate_project,
status_id=1,
- msg='5G WPA2 Client Connectivity Passed successfully - vlan mode')
+ msg='5G WPA2 Client Connectivity Passed successfully - vlan mode' + str(run_results))
else:
instantiate_testrail.update_testrail(case_id=test_cases["5g_wpa2_vlan"], run_id=instantiate_project,
status_id=5,
- msg='5G WPA2 Client Connectivity Failed - vlan mode')
+ msg='5G WPA2 Client Connectivity Failed - vlan mode' + str(run_results))
+ instantiate_controller.refresh_instance()
assert staConnect.passes()
# C2236
@@ -231,7 +235,7 @@ class TestVlanModeClientConnectivity(object):
@pytest.mark.twog
@pytest.mark.radius
def test_client_wpa2_enterprise_2g(self, request, get_lanforge_data, setup_profile_data, instantiate_project,
- instantiate_testrail, radius_info, test_cases):
+ instantiate_controller, instantiate_testrail, radius_info, test_cases):
profile_data = setup_profile_data["VLAN"]["WPA2_E"]["2G"]
station_names = []
for i in range(0, int(request.config.getini("num_stations"))):
@@ -248,7 +252,7 @@ class TestVlanModeClientConnectivity(object):
eap_connect.eap = "TTLS"
eap_connect.identity = radius_info["user"]
eap_connect.ttls_passwd = radius_info["password"]
- eap_connect.runtime_secs = 10
+ eap_connect.runtime_secs = 40
eap_connect.setup()
eap_connect.start()
print("napping %f sec" % eap_connect.runtime_secs)
@@ -268,11 +272,12 @@ class TestVlanModeClientConnectivity(object):
instantiate_testrail.update_testrail(case_id=test_cases["2g_eap_vlan"], run_id=instantiate_project,
status_id=1,
msg='5G WPA2 ENTERPRISE Client Connectivity Passed successfully - '
- 'vlan mode')
+ 'vlan mode' + str(run_results))
else:
instantiate_testrail.update_testrail(case_id=test_cases["2g_eap_vlan"], run_id=instantiate_project,
status_id=5,
- msg='5G WPA2 ENTERPRISE Client Connectivity Failed - vlan mode')
+ msg='5G WPA2 ENTERPRISE Client Connectivity Failed - vlan mode' + str(run_results))
+ instantiate_controller.refresh_instance()
assert eap_connect.passes()
# C5214
@@ -280,7 +285,7 @@ class TestVlanModeClientConnectivity(object):
@pytest.mark.fiveg
@pytest.mark.radius
def test_client_wpa2_enterprise_5g(self, request, get_lanforge_data, setup_profile_data, instantiate_project,
- instantiate_testrail, radius_info, test_cases):
+ instantiate_controller, instantiate_testrail, radius_info, test_cases):
profile_data = setup_profile_data["VLAN"]["WPA2_E"]["5G"]
station_names = []
for i in range(0, int(request.config.getini("num_stations"))):
@@ -297,7 +302,7 @@ class TestVlanModeClientConnectivity(object):
eap_connect.eap = "TTLS"
eap_connect.identity = radius_info["user"]
eap_connect.ttls_passwd = radius_info["password"]
- eap_connect.runtime_secs = 10
+ eap_connect.runtime_secs = 40
eap_connect.setup()
eap_connect.start()
print("napping %f sec" % eap_connect.runtime_secs)
@@ -317,11 +322,12 @@ class TestVlanModeClientConnectivity(object):
instantiate_testrail.update_testrail(case_id=test_cases["5g_eap_vlan"], run_id=instantiate_project,
status_id=1,
msg='5G WPA2 ENTERPRISE Client Connectivity Passed successfully - '
- 'vlan mode')
+ 'vlan mode' + str(run_results))
else:
instantiate_testrail.update_testrail(case_id=test_cases["5g_eap_vlan"], run_id=instantiate_project,
status_id=5,
- msg='5G WPA2 ENTERPRISE Client Connectivity Failed - vlan mode')
+ msg='5G WPA2 ENTERPRISE Client Connectivity Failed - vlan mode' + str(run_results))
+ instantiate_controller.refresh_instance()
assert eap_connect.passes()
@pytest.mark.modify_ssid
@@ -331,7 +337,7 @@ class TestVlanModeClientConnectivity(object):
indirect=True
)
def test_modify_ssid(self, request, update_ssid, get_lanforge_data, setup_profile_data, instantiate_testrail,
- instantiate_project, test_cases):
+ instantiate_project, test_cases, instantiate_controller):
profile_data = setup_profile_data["VLAN"]["WPA"]["5G"]
station_names = []
for i in range(0, int(request.config.getini("num_stations"))):
@@ -340,7 +346,7 @@ class TestVlanModeClientConnectivity(object):
debug_=False)
staConnect.sta_mode = 0
staConnect.upstream_resource = 1
- staConnect.upstream_port = get_lanforge_data["lanforge_bridge_port"]
+ staConnect.upstream_port = get_lanforge_data["lanforge_vlan_port"]
staConnect.radio = get_lanforge_data["lanforge_5g"]
staConnect.resource = 1
staConnect.dut_ssid = profile_data["ssid_name"]
@@ -348,7 +354,7 @@ class TestVlanModeClientConnectivity(object):
staConnect.dut_security = "wpa"
staConnect.station_names = station_names
staConnect.sta_prefix = get_lanforge_data["lanforge_5g_prefix"]
- staConnect.runtime_secs = 10
+ staConnect.runtime_secs = 40
staConnect.bringup_time_sec = 60
staConnect.cleanup_on_exit = True
# staConnect.cleanup()
@@ -367,9 +373,10 @@ class TestVlanModeClientConnectivity(object):
instantiate_testrail.update_testrail(case_id=test_cases["vlan_ssid_update"], run_id=instantiate_project,
status_id=1,
msg='5G WPA Client Connectivity Passed successfully - vlan mode '
- 'updated ssid')
+ 'updated ssid' + str(run_results))
else:
instantiate_testrail.update_testrail(case_id=test_cases["vlan_ssid_update"], run_id=instantiate_project,
status_id=5,
- msg='5G WPA Client Connectivity Failed - vlan mode updated ssid')
+ msg='5G WPA Client Connectivity Failed - vlan mode updated ssid' + str(run_results))
+ instantiate_controller.refresh_instance()
assert staConnect.passes()
diff --git a/tests/e2e/basic/conftest.py b/tests/e2e/basic/conftest.py
index 40026e9db..f847fe39f 100644
--- a/tests/e2e/basic/conftest.py
+++ b/tests/e2e/basic/conftest.py
@@ -23,8 +23,6 @@ Information:
"""
-
-
import sys
import os
import time
@@ -50,7 +48,6 @@ from configuration import FIRMWARE
from testrails.testrail_api import APIClient
from testrails.reporting import Reporting
-
"""
Basic Setup Collector
"""
@@ -80,7 +77,7 @@ def get_lanforge_data(testbed):
def instantiate_profile(instantiate_controller):
try:
profile_object = ProfileUtility(sdk_client=instantiate_controller)
- except:
+ except Exception as e:
profile_object = False
yield profile_object
@@ -91,6 +88,7 @@ def get_equipment_id(instantiate_controller, testbed):
if len(CONFIGURATION[testbed]['access_point']) == 1:
equipment_id = instantiate_controller.get_equipment_id(
serial_number=CONFIGURATION[testbed]['access_point'][0]['serial'])
+ print(equipment_id)
yield equipment_id
@@ -101,7 +99,7 @@ def upload_firmware(should_upload_firmware, instantiate_firmware, get_latest_fir
yield firmware_id
-@pytest.fixture(scope="session")
+@pytest.fixture(scope="function")
def upgrade_firmware(request, instantiate_firmware, get_equipment_id, check_ap_firmware_cloud, get_latest_firmware,
should_upgrade_firmware):
if get_latest_firmware != check_ap_firmware_cloud:
@@ -119,7 +117,7 @@ def upgrade_firmware(request, instantiate_firmware, get_equipment_id, check_ap_f
yield status
-@pytest.fixture(scope="session")
+@pytest.fixture(scope="function")
def check_ap_firmware_cloud(instantiate_controller, get_equipment_id):
yield instantiate_controller.get_ap_firmware_old_method(equipment_id=get_equipment_id)
@@ -145,9 +143,10 @@ def setup_profiles(request, create_profiles, instantiate_profile, get_equipment_
mode = str(request.param[0]).lower()
try:
instantiate_profile.push_profile_old_method(equipment_id=get_equipment_id)
- except:
- print("failed to create AP Profile")
- ap_ssh = APNOS(CONFIGURATION[testbed]['access_point'][0])
+ except Exception as e:
+ print(e)
+ print("failed to Push Profile")
+ ap_ssh = APNOS(CONFIGURATION[testbed]['access_point'][0], pwd="../libs/apnos/")
get_current_profile_cloud.sort()
# This loop will check the VIF Config with cloud profile
for i in range(0, 18):
@@ -159,7 +158,7 @@ def setup_profiles(request, create_profiles, instantiate_profile, get_equipment_
test_cases[mode + '_vifc'] = True
break
time.sleep(10)
- ap_ssh = APNOS(CONFIGURATION[testbed]['access_point'][0])
+ ap_ssh = APNOS(CONFIGURATION[testbed]['access_point'][0], pwd="../libs/apnos/")
# This loop will check the VIF Config with VIF State
for i in range(0, 18):
vif_state = list(ap_ssh.get_vif_state_ssids())
@@ -172,40 +171,40 @@ def setup_profiles(request, create_profiles, instantiate_profile, get_equipment_
test_cases[mode + '_vifs'] = True
break
time.sleep(10)
+ #
yield test_cases
@pytest.fixture(scope="module")
-def create_profiles(request, get_security_flags, get_markers, instantiate_profile, setup_profile_data):
+def create_profiles(request, testbed, get_security_flags, get_markers, instantiate_profile, setup_profile_data):
profile_id = {"ssid": [], "rf": None, "radius": None, "equipment_ap": None}
mode = str(request.param[0])
test_cases = {}
if mode not in ["BRIDGE", "NAT", "VLAN"]:
print("Invalid Mode: ", mode)
yield False
- instantiate_profile.delete_profile_by_name(profile_name="Equipment-AP-" + mode)
+ instantiate_profile.delete_profile_by_name(profile_name=testbed + "-Equipment-AP-" + mode)
for i in setup_profile_data[mode]:
for j in setup_profile_data[mode][i]:
instantiate_profile.delete_profile_by_name(
profile_name=setup_profile_data[mode][i][j]['profile_name'])
- instantiate_profile.delete_profile_by_name(profile_name="Automation-Radius-Profile-" + mode)
+ instantiate_profile.delete_profile_by_name(profile_name=testbed + "-Automation-Radius-Profile-" + mode)
instantiate_profile.get_default_profiles()
- # if get_markers["wifi5"]:
- # # Create RF Profile
- # pass
- # if get_markers["wifi6"]:
- # # Create RF Profile
- # pass
-
+ profile_data = {
+ "name": "RF-Profile-" + CONFIGURATION[testbed]['access_point'][0]['mode'] +
+ CONFIGURATION[testbed]['access_point'][0]['model'] + "_" + mode + "_" + testbed
+ }
+ instantiate_profile.delete_profile_by_name(profile_name=profile_data['name'])
+ instantiate_profile.set_rf_profile(profile_data=profile_data,
+ mode=CONFIGURATION[testbed]['access_point'][0]['mode'])
# Create RF Profile Here
- instantiate_profile.set_rf_profile()
if get_markers["radius"]:
radius_info = RADIUS_SERVER_DATA
- radius_info["name"] = "Automation-Radius-Profile-" + mode
+ radius_info["name"] = testbed + "-Automation-Radius-Profile-" + mode
try:
instantiate_profile.create_radius_profile(radius_info=radius_info)
test_cases['radius_profile'] = True
- except:
+ except Exception as e:
test_cases['radius_profile'] = False
for i in get_security_flags:
if get_markers[i] and i == "open":
@@ -216,7 +215,7 @@ def create_profiles(request, get_security_flags, get_markers, instantiate_profil
profile_data=profile_data)
profile_id["ssid"].append(profile_data['ssid_name'])
test_cases['ssid_2g_open_' + mode.lower()] = True
- except:
+ except Exception as e:
test_cases['ssid_2g_open_' + mode.lower()] = False
if get_markers["fiveg"]:
profile_data = setup_profile_data[mode]["OPEN"]["5G"]
@@ -225,7 +224,7 @@ def create_profiles(request, get_security_flags, get_markers, instantiate_profil
profile_data=profile_data)
profile_id["ssid"].append(profile_data['ssid_name'])
test_cases['ssid_5g_open_' + mode.lower()] = True
- except:
+ except Exception as e:
test_cases['ssid_5g_open_' + mode.lower()] = False
if get_markers[i] and i == "wpa":
if get_markers["twog"]:
@@ -234,7 +233,7 @@ def create_profiles(request, get_security_flags, get_markers, instantiate_profil
id = instantiate_profile.create_wpa_ssid_profile(two4g=True, fiveg=False, profile_data=profile_data)
profile_id["ssid"].append(profile_data['ssid_name'])
test_cases['ssid_2g_wpa_' + mode.lower()] = True
- except:
+ except Exception as e:
test_cases['ssid_5g_wpa_' + mode.lower()] = False
if get_markers["fiveg"]:
profile_data = setup_profile_data[mode]["WPA"]["5G"]
@@ -242,7 +241,7 @@ def create_profiles(request, get_security_flags, get_markers, instantiate_profil
id = instantiate_profile.create_wpa_ssid_profile(two4g=False, fiveg=True, profile_data=profile_data)
profile_id["ssid"].append(profile_data['ssid_name'])
test_cases['ssid_5g_wpa_' + mode.lower()] = True
- except:
+ except Exception as e:
test_cases['ssid_5g_wpa_' + mode.lower()] = False
if get_markers[i] and i == "wpa2_personal":
if get_markers["twog"]:
@@ -252,7 +251,7 @@ def create_profiles(request, get_security_flags, get_markers, instantiate_profil
profile_data=profile_data)
profile_id["ssid"].append(profile_data['ssid_name'])
test_cases['ssid_2g_wpa2_' + mode.lower()] = True
- except:
+ except Exception as e:
test_cases['ssid_2g_wpa2_' + mode.lower()] = False
if get_markers["fiveg"]:
profile_data = setup_profile_data[mode]["WPA2_P"]["5G"]
@@ -261,7 +260,7 @@ def create_profiles(request, get_security_flags, get_markers, instantiate_profil
profile_data=profile_data)
profile_id["ssid"].append(profile_data['ssid_name'])
test_cases['ssid_5g_wpa2_' + mode.lower()] = True
- except:
+ except Exception as e:
test_cases['ssid_5g_wpa2_' + mode.lower()] = False
if get_markers[i] and i == "wpa2_enterprise":
if get_markers["twog"]:
@@ -271,7 +270,7 @@ def create_profiles(request, get_security_flags, get_markers, instantiate_profil
profile_data=profile_data)
profile_id["ssid"].append(profile_data['ssid_name'])
test_cases['ssid_2g_eap_' + mode.lower()] = True
- except:
+ except Exception as e:
test_cases['ssid_2g_eap_' + mode.lower()] = False
if get_markers["fiveg"]:
profile_data = setup_profile_data[mode]["WPA2_E"]["5G"]
@@ -280,18 +279,26 @@ def create_profiles(request, get_security_flags, get_markers, instantiate_profil
profile_data=profile_data)
profile_id["ssid"].append(profile_data['ssid_name'])
test_cases['ssid_5g_eap_' + mode.lower()] = True
- except:
+ except Exception as e:
test_cases['ssid_5g_eap_' + mode.lower()] = False
# Create Equipment AP Profile Here
profile_data = {
- "profile_name": "Equipment-AP-" + mode
+ "profile_name": testbed + "-Equipment-AP-" + mode
}
try:
instantiate_profile.set_ap_profile(profile_data=profile_data)
test_cases['ap_' + mode.lower()] = True
- except:
+ except Exception as e:
+ print(e)
test_cases['ap_' + mode.lower()] = False
+
+ def teardown_profiles():
+ print("\nRemoving Profiles")
+ instantiate_profile.delete_profile_by_name(profile_name=profile_data['profile_name'])
+ time.sleep(20)
+
+ request.addfinalizer(teardown_profiles)
yield test_cases
@@ -307,3 +314,4 @@ def update_ssid(request, instantiate_profile, setup_profile_data):
requested_profile[3]
time.sleep(90)
yield status
+
diff --git a/tests/e2e/basic/wifi_capacity_test/test_bridge_mode.py b/tests/e2e/basic/wifi_capacity_test/test_bridge_mode.py
index 367734efd..b8e72ec12 100644
--- a/tests/e2e/basic/wifi_capacity_test/test_bridge_mode.py
+++ b/tests/e2e/basic/wifi_capacity_test/test_bridge_mode.py
@@ -6,14 +6,23 @@ import pytest
@pytest.mark.wifi5
@pytest.mark.wifi6
@pytest.mark.parametrize(
- 'setup_profiles',
- (["BRIDGE"]),
- indirect=True
+ 'setup_profiles, create_profiles',
+ [(["NAT"], ["NAT"])],
+ indirect=True,
+ scope="class"
)
@pytest.mark.usefixtures("setup_profiles")
+@pytest.mark.usefixtures("create_profiles")
class TestBridgeModeClientConnectivity(object):
@pytest.mark.wpa
@pytest.mark.twog
- def test_client_wpa_2g(self, request, get_lanforge_data, setup_profile_data):
+ @pytest.mark.fiveg
+ def test_client_wpa_2g(self, get_lanforge_data, setup_profile_data):
+ assert True
+
+ @pytest.mark.wpa
+ @pytest.mark.twog
+ @pytest.mark.fiveg
+ def test_client_wpa_2g(self, get_lanforge_data, setup_profile_data):
assert True
diff --git a/tests/pytest.ini b/tests/pytest.ini
index 151947900..c67e359bf 100644
--- a/tests/pytest.ini
+++ b/tests/pytest.ini
@@ -2,7 +2,8 @@
python_files = test_*.py setup_*.py
norecursedirs = .svn _build tmp*
addopts= --junitxml=test_everything.xml
-
+log_format = %(asctime)s %(levelname)s %(message)s
+log_date_format = %Y-%m-%d %H:%M:%S
num_stations=1
diff --git a/tests/test_connectivity.py b/tests/test_connectivity.py
index 091e8abf6..7e5b282db 100644
--- a/tests/test_connectivity.py
+++ b/tests/test_connectivity.py
@@ -6,10 +6,13 @@
import pytest
import sys
-pytestmark = [pytest.mark.test_connection]
+pytestmark = [pytest.mark.test_connectivity]
@pytest.mark.sanity
+@pytest.mark.bridge
+@pytest.mark.nat
+@pytest.mark.vlan
@pytest.mark.test_controller_connectivity
def test_controller_connectivity(instantiate_controller, instantiate_testrail, instantiate_project, test_cases):
try:
@@ -24,27 +27,39 @@ def test_controller_connectivity(instantiate_controller, instantiate_testrail, i
@pytest.mark.sanity
+@pytest.mark.bridge
+@pytest.mark.nat
+@pytest.mark.vlan
@pytest.mark.test_access_points_connectivity
-def test_access_points_connectivity(test_access_point, instantiate_testrail, instantiate_project, test_cases):
- if "ACTIVE" not in test_access_point:
+def test_access_points_connectivity(access_point_connectivity, instantiate_testrail, instantiate_project, test_cases, exit_on_fail):
+ if not access_point_connectivity["serial"] and not access_point_connectivity["mgr"]:
instantiate_testrail.update_testrail(case_id=test_cases["cloud_connection"], run_id=instantiate_project,
status_id=5,
msg='CloudSDK connectivity failed')
status = False
+ pytest.exit("Access Point is not Properly Connected: Sanity Failed")
else:
instantiate_testrail.update_testrail(case_id=test_cases["cloud_connection"], run_id=instantiate_project,
status_id=1,
msg='Manager status is Active')
status = True
- sys.exit()
+
assert status
-@pytest.mark.test_lanforge_connectivity
-def test_lanforge_connectivity(setup_lanforge):
- assert "instantiate_cloudsdk"
-
-
-@pytest.mark.test_perfecto_connectivity
-def test_perfecto_connectivity(setup_perfecto_devices):
- assert "instantiate_cloudsdk"
+# @pytest.mark.sanity
+# @pytest.mark.bridge
+# @pytest.mark.nat
+# @pytest.mark.vlan
+# @pytest.mark.test_lanforge_connectivity
+# def test_lanforge_connectivity(check_lanforge_connectivity):
+# assert "instantiate_cloudsdk"
+#
+#
+# @pytest.mark.sanity
+# @pytest.mark.bridge
+# @pytest.mark.nat
+# @pytest.mark.vlan
+# @pytest.mark.test_perfecto_connectivity
+# def test_perfecto_connectivity(setup_perfecto_devices):
+# assert "instantiate_cloudsdk"