pytest initial commit

.gitignore

@@ -137,6 +137,15 @@ dmypy.json
 # Cython debug symbols
 cython_debug/
 
+# nightlies
 docker/*
 !docker/nightly*
 docker/nightly*log
+
+# nightlies with pytest
+pytest/*
+!pytest/conftest.py
+!pytest/test_*.py
+!pytest/helpers
+!pytest/pytest.ini
+pytest/nightly*log

pytest/conftest.py

@@ -0,0 +1,253 @@
+import pytest
+from time import sleep, gmtime, strftime
+
+import sys
+import os
+sys.path.append(os.path.join(os.path.dirname(__file__), 'helpers'))
+from utils import CloudSDK_Client, TestRail_Client, jFrog_Client
+
+def pytest_addoption(parser):
+    parser.addini("jfrog-base-url", "jfrog base url")
+    parser.addini("jfrog-user-id", "jfrog username")
+    parser.addini("jfrog-user-password", "jfrog password")
+    parser.addini("sdk-base-url", "cloud sdk base url")
+    parser.addini("sdk-user-id", "cloud sdk username")
+    parser.addini("sdk-user-password", "cloud sdk user password")
+    parser.addini("sdk-customer-id", "cloud sdk customer id for the access points")
+    parser.addini("testrail-base-url", "testrail base url")
+    parser.addini("testrail-project", "testrail project name to use to generate test reports")
+    parser.addini("testrail-user-id", "testrail username")
+    parser.addini("testrail-user-password", "testrail user password")
+    parser.addini("lanforge-ip-address", "LANforge ip address to connect to")
+    parser.addini("lanforge-port-number", "LANforge port number to connect to")
+    parser.addini("lanforge-radio", "LANforge radio to use")
+    parser.addini("lanforge-ethernet-port", "LANforge ethernet adapter to use")
+
+    parser.addoption(
+        "--testrail-user-password",
+        action="store",
+        default="password",
+        help="testrail user password",
+        type=str
+    )
+
+    # # Cloud SDK
+    # parser.addoption(
+    #     "--sdk-base-url",
+    #     action="store",
+    #     default="wlan-portal-svc.cicd.lab.wlan.tip.build",
+    #     help="cloudsdk base url",
+    #     type=str
+    # )
+    # parser.addoption(
+    #     "--sdk-user-id",
+    #     action="store",
+    #     default="support@example.com",
+    #     help="cloudsdk user id",
+    #     type=str
+    # )
+    # parser.addoption(
+    #     "--sdk-user-password",
+    #     action="store",
+    #     default="support",
+    #     help="cloudsdk user password",
+    #     type=str
+    # )
+
+    # # jFrog
+    # parser.addoption(
+    #     "--jfrog-base-url",
+    #     action="store",
+    #     default="tip.jFrog.io/artifactory/tip-wlan-ap-firmware",
+    #     help="jfrog base url",
+    #     type=str
+    # )
+    # parser.addoption(
+    #     "--jfrog-user-id",
+    #     action="store",
+    #     default="tip-read",
+    #     help="jfrog user id",
+    #     type=str
+    # )
+    # parser.addoption(
+    #     "--jfrog-user-password",
+    #     action="store",
+    #     default="tip-read",
+    #     help="jfrog user password",
+    #     type=str
+    # )
+
+    # # testrail
+    # parser.addoption(
+    #     "--testrail-base-url",
+    #     action="store",
+    #     default="telecominfraproject.testrail.com",
+    #     help="testrail base url",
+    #     type=str
+    # )
+    # parser.addoption(
+    #     "--testrail-project",
+    #     action="store",
+    #     default="opsfleet-wlan",
+    #     help="testrail project name",
+    #     type=str
+    # )
+    # parser.addoption(
+    #     "--testrail-user-id",
+    #     action="store",
+    #     default="gleb@opsfleet.com",
+    #     help="testrail user id",
+    #     type=str
+    # )
+    # parser.addoption(
+    #     "--testrail-user-password",
+    #     action="store",
+    #     default="password",
+    #     help="testrail user password",
+    #     type=str
+    # )
+
+    # # lanforge
+    # parser.addoption(
+    #     "--lanforge-ip-address",
+    #     action="store",
+    #     default="10.28.3.6",
+    #     help="ip address of the lanforge gui",
+    #     type=str
+    # )
+    # parser.addoption(
+    #     "--lanforge-port-number",
+    #     action="store",
+    #     default="8080",
+    #     help="port of the lanforge gui",
+    #     type=str
+    # )
+
+    # change behaviour
+    parser.addoption(
+        "--skip-update-firmware",
+        action="store_true",
+        default=False,
+        help="skip updating firmware on the AP (useful for local testing)"
+    )
+    parser.addoption(
+        "--no-testrails",
+        action="store_true",
+        default=False,
+        help="do not generate testrails tests"
+    )
+    # this has to be the last argument
+    # example: --access-points ECW5410 EA8300-EU    
+    parser.addoption(
+        "--access-points",
+        nargs="+",
+        default=[ "ECW5410" ],
+        help="list of access points to test"
+    )
+
+def pytest_generate_tests(metafunc):
+    metafunc.parametrize("access_points", metafunc.config.getoption('--access-points'), scope="session")
+
+# run something after all tests are done regardless of the outcome
+def pytest_unconfigure(config):
+    print("Tests cleanup done")
+
+@pytest.fixture(scope="session")
+def setup_testrails(request, instantiate_testrail, access_points):
+    if request.config.getoption("--no-testrails"):
+        yield -1
+        return # needed to stop fixture execution
+    if request.config.getoption("--skip-update-firmware"):
+        firmware_update_case = []
+    else:
+        firmware_update_case = [ 2831 ]
+    seen = {None}
+    test_data = []
+    session = request.node
+    for item in session.items:
+        cls = item.getparent(pytest.Class)
+        if cls not in seen:
+            if hasattr(cls.obj, "get_test_data"):
+                test_data.append(cls.obj.get_test_data())
+            seen.add(cls)
+    testrail_project_id = instantiate_testrail.get_project_id(request.config.getini("testrail-project"))
+    runId = instantiate_testrail.create_testrun(
+        name=f'Nightly_model_{access_points}_{strftime("%Y-%m-%d", gmtime())}',
+        case_ids=( [*test_data] + firmware_update_case ),
+        project_id=testrail_project_id
+    )
+    yield runId
+
+@pytest.fixture(scope="session")
+def setup_cloudsdk(request, instantiate_cloudsdk):
+    # snippet to do cleanup after all the tests are done
+    def fin():
+        print("Cloud SDK cleanup done")
+    request.addfinalizer(fin)
+    instantiate_cloudsdk.set_ap_profile(3, 6)
+    yield {
+        "LANforge": {
+            "host": request.config.getini("lanforge-ip-address"),
+            "port": request.config.getini("lanforge-port-number"),
+            "radio": request.config.getini("lanforge-radio"),
+            "eth_port": request.config.getini("lanforge-ethernet-port"),
+            "runtime_duration": 15
+        },
+        "24ghz": {
+            "ssid": "TipWlan-cloud-wifi",
+            "password": "w1r3l3ss-fr33d0m",
+            "station_names": [ "sta2237" ]
+        }
+    }
+
+@pytest.fixture(scope="session")
+def update_firmware(request, setup_testrails, instantiate_jFrog, instantiate_cloudsdk, access_points):
+    if request.config.getoption("--skip-update-firmware"):
+        return
+    latest_image = instantiate_jFrog.get_latest_image(access_points)
+    if latest_image in instantiate_cloudsdk.get_images(access_points):
+        model_firmware_id = instantiate_cloudsdk.get_firmware_id(latest_image)
+    else:
+        fw_url = instantiate_jFrog.get_latest_image_url(access_points, latest_image)
+        fw_upload_status = instantiate_cloudsdk.firwmare_upload(access_points, latest_image, fw_url)
+        model_firmware_id = fw_upload_status['id']
+
+    # Get Current AP Firmware and upgrade\run tests if needed
+    # currently the AP id is hardcoded, but it should be looked up during the tests and not hardcoded in the config files or parameters
+    ap_fw = instantiate_cloudsdk.ap_firmware(request.config.getini("sdk-customer-id"), 3)
+    if ap_fw == latest_image:
+        pytest.skip("Do not need to upgrade firmware")
+    else:
+        instantiate_cloudsdk.update_firmware(3, model_firmware_id)
+        sleep_counter = 0
+        while True:
+            sleep_counter += 1
+            if instantiate_cloudsdk.ap_firmware(2, 3) == latest_image:
+                return
+            if sleep_counter > 0:
+                return
+            sleep(60)
+
+@pytest.fixture(scope="session")
+def instantiate_cloudsdk(request):
+    yield CloudSDK_Client(
+        request.config.getini("sdk-base-url"),
+        request.config.getini("sdk-user-id"),
+        request.config.getini("sdk-user-password")
+    )
+
+@pytest.fixture(scope="session")
+def instantiate_testrail(request):
+    yield TestRail_Client(
+        request.config.getini("testrail-base-url"),
+        request.config.getini("testrail-user-id"),
+        request.config.getoption("--testrail-user-password")
+    )
+
+@pytest.fixture(scope="session")
+def instantiate_jFrog(request):
+    yield jFrog_Client(
+        request.config.getini("jfrog-base-url"),
+        request.config.getini("jfrog-user-id"),
+        request.config.getini("jfrog-user-password")
+    )

pytest/helpers/utils.py

@@ -0,0 +1,186 @@
+import re
+import requests
+import json
+
+# Class to interact with Testrail
+class TestRail_Client:
+    def __init__(self, url, user, password):
+        self.user = user
+        self.password = password
+        self.__url = f"https://{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"
+        }
+
+        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))
+
+        if response.status_code > 201:
+            try:
+                error = response.json()
+            except: # response.content not formatted as JSON
+                error = str(response.content)
+            return
+        else:
+            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 = self.send_get("get_projects")
+        for project in projects:
+            if project["name"] == project_name:
+                return project["id"]
+
+    def update_testrail(self, case_id, run_id, status_id, msg):
+        result = self.send_post(
+            f"add_result_for_case/{run_id}/{case_id}",
+            { "status_id": status_id, "comment": msg }
+        )
+
+    def create_testrun(self, name, case_ids, project_id):
+        result = self.send_post(
+            f"add_run/{project_id}",
+            {"name": name, "case_ids": case_ids, "include_all": False}
+        )
+        return result["id"]
+
+# Class for jFrog Interaction
+class jFrog_Client:
+    def __init__(self, url, user, password):
+        self.user = user
+        self.password = password
+        self.baseUrl = f"https://{url}"
+
+    def get_latest_image(self, model):
+        # todo handle auth errors
+        response = requests.get(f"{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, url, user, password):
+        self.baseUrl = f"https://{url}"
+        cloud_login_url = f"{self.baseUrl}/management/v1/oauth2/token"
+        payload = {
+            "userId": user,
+            "password": 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)
+        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)
+        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)
+        return(response.json())
+
+    def get_firmware_id(self, 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)
+
+    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)
+
+        # Add Lab Profile ID to Equipment
+        equipment_info = response.json()
+        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)

pytest/pytest.ini

@@ -0,0 +1,23 @@
+[pytest]
+addopts= --junitxml=test_everything.xml
+# jFrog parameters
+jfrog-base-url=tip.jFrog.io/artifactory/tip-wlan-ap-firmware
+jfrog-user-id=tip-read
+jfrog-user-password=tip-read
+# Cloud SDK parameters
+sdk-base-url=wlan-portal-svc.cicd.lab.wlan.tip.build
+sdk-user-id=support@example.com
+sdk-user-password=support
+# Testrails parameters
+testrail-base-url=telecominfraproject.testrail.com
+testrail-project=opsfleet-wlan
+testrail-user-id=gleb@opsfleet.com
+testrail-user-password=use_command_line_to_pass_this
+# LANforge
+lanforge-ip-address=localhost
+lanforge-port-number=8080
+lanforge-radio=wiphy4
+lanforge-ethernet-port=eth2
+
+# Cloud SDK settings
+sdk-customer-id=2

pytest/test_24ghz.py

@@ -0,0 +1,44 @@
+# https://docs.pytest.org/en/latest/example/markers.html
+# https://docs.pytest.org/en/latest/usage.html
+# http://pythontesting.net/framework/pytest/pytest-introduction/
+
+import pytest
+from time import sleep, gmtime, strftime
+from sta_connect2 import StaConnect2
+
+@pytest.mark.usefixtures('setup_testrails')
+@pytest.mark.usefixtures('setup_cloudsdk')
+@pytest.mark.usefixtures('update_firmware')
+@pytest.mark.usefixtures('instantiate_testrail')
+class Test24ghz(object):
+    @pytest.mark.featureA
+    def test_single_client_wpa2(self, setup_testrails, setup_cloudsdk, update_firmware, instantiate_testrail):
+        lf_config = setup_cloudsdk["LANforge"]
+        radio_config = setup_cloudsdk["24ghz"]
+
+        staConnect = StaConnect2(lf_config["host"], lf_config["port"], debug_ = False)
+        staConnect.sta_mode = 0
+        staConnect.upstream_resource = 1
+        staConnect.upstream_port = lf_config["eth_port"]
+        staConnect.radio = lf_config["radio"]
+        staConnect.runtime_secs = lf_config["runtime_duration"]
+        staConnect.resource = 1
+        staConnect.dut_ssid = radio_config["ssid"]
+        staConnect.dut_passwd = radio_config["password"]
+        staConnect.dut_security = "wpa2"
+        staConnect.station_names = radio_config["station_names"]
+        staConnect.bringup_time_sec = 60
+        staConnect.cleanup_on_exit = True
+        staConnect.setup()
+        staConnect.start()
+        sleep(staConnect.runtime_secs)
+        staConnect.stop()
+        staConnect.cleanup()
+
+        assert staConnect.passes()
+        if setup_testrails > 0:
+            instantiate_testrail.update_testrail(case_id=2835, run_id=setup_testrails, status_id=1, msg="testing")
+
+    @pytest.mark.featureB
+    def test_feature_b(self):
+        pass