From aefa270425afc4cea830935902d6b6643d46dd91 Mon Sep 17 00:00:00 2001 From: Gopi Raga <117985788+GopiRaga7@users.noreply.github.com> Date: Fri, 17 Feb 2023 13:13:58 +0530 Subject: [PATCH] Wifi 12235 (#784) * Added ASB two test cases and functionality Signed-off-by: GopiRaga7 * Resolved chunk error in the ASB API Signed-off-by: GopiRaga7 * adding tar file in the tests folder Signed-off-by: GopiRaga7 * sleeping after creating file Signed-off-by: GopiRaga7 * sleeping after creating file2 Signed-off-by: GopiRaga7 * more info Signed-off-by: GopiRaga7 * more info2 Signed-off-by: GopiRaga7 * adding full path Signed-off-by: GopiRaga7 * changed the path Signed-off-by: GopiRaga7 * added another file Signed-off-by: GopiRaga7 * modified and added another file Signed-off-by: GopiRaga7 * Added more logs Signed-off-by: GopiRaga7 * Fixed issue of attaching tar file to allure ASB Signed-off-by: GopiRaga7 * Asb script two testcases added Signed-off-by: GopiRaga7 --------- Signed-off-by: GopiRaga7 --- libs/tip_2x/ap_lib.py | 60 ++++++++-- libs/tip_2x/controller.py | 76 ++++++++++--- .../ucentral_gateway/test_gatewayservice.py | 105 +++++++++++++++--- 3 files changed, 206 insertions(+), 35 deletions(-) diff --git a/libs/tip_2x/ap_lib.py b/libs/tip_2x/ap_lib.py index 669e03f52..892126597 100644 --- a/libs/tip_2x/ap_lib.py +++ b/libs/tip_2x/ap_lib.py @@ -135,8 +135,8 @@ class APLIBS: logging.error("error in converting the ubus call ucentral status output to json" + output) data = {} if (data.keys().__contains__("connected") or data.keys().__contains__("disconnected")) and \ - data.keys().__contains__("latest") and \ - data.keys().__contains__("active"): + data.keys().__contains__("latest") and \ + data.keys().__contains__("active"): break else: logging.error("Error in ubus call ucentral status: " + str(output)) @@ -263,7 +263,7 @@ class APLIBS: "Please add valid certificates on AP") def run_generic_command(self, cmd="", idx=0, print_log=True, attach_allure=False, - expected_attachment_type=allure.attachment_type.TEXT): + expected_attachment_type=allure.attachment_type.TEXT, restrictions=False): input_command = cmd logging.info("Executing Command on AP: " + cmd) try: @@ -279,8 +279,12 @@ class APLIBS: if self.device_under_tests_data[idx]["method"] == "serial": owrt_args = "--prompt root@" + self.device_under_tests_data[idx][ "identifier"] + " -s serial --log stdout --user root --passwd openwifi" - cmd = f"cd ~/cicd-git/ && ./openwrt_ctl.py {owrt_args} -t {self.device_under_tests_data[idx]['serial_tty']} --action " \ - f"cmd --value \"{cmd}\" " + if not restrictions: + cmd = f"cd ~/cicd-git/ && ./openwrt_ctl.py {owrt_args} -t {self.device_under_tests_data[idx]['serial_tty']} --action " \ + f"cmd --value \"{cmd}\" " + if restrictions: + cmd = f"cd ~/cicd-git/ && ./openwrt_ctl.py {owrt_args} -t {self.device_under_tests_data[idx]['serial_tty']} --action " \ + f"cmd --value \'{cmd}\' " if print_log: logging.info(cmd) stdin, stdout, stderr = client.exec_command(cmd) @@ -474,9 +478,51 @@ class APLIBS: logging.info("Simulate radar logs: " + str(logs)) return logs + def factory_reset(self, idx=0, print_log=True, attach_allure=False): + logging.info("started factory reset") + output = self.run_generic_command(cmd="jffs2reset -y -r", + idx=idx, + print_log=print_log, + attach_allure=attach_allure) + logging.info("Done Factory reset") + # Please wait 1min after doing factory reset + return output + + def remove_restrictions(self): + self.run_generic_command(cmd="rm /certificates/restrictions.json", + idx=0, print_log=True, + attach_allure=False) + self.run_generic_command(cmd="rm /etc/ucentral/restrictions.json", + idx=0, print_log=True, + attach_allure=False) + self.factory_reset(print_log=False) + time.sleep(120) + output = self.run_generic_command(cmd='[ -f /etc/ucentral/restrictions.json ] && echo "True" || echo "False"', + idx=0, + print_log=True, + attach_allure=False, + expected_attachment_type=allure.attachment_type.TEXT) + return output + + def add_restrictions(self, restrictions_file, developer_mode): + output = self.run_generic_command(cmd=developer_mode, + idx=0, + print_log=True, + attach_allure=False, + expected_attachment_type=allure.attachment_type.TEXT) + output = self.run_generic_command(cmd=restrictions_file, + idx=0, + print_log=True, + attach_allure=False, + expected_attachment_type=allure.attachment_type.TEXT, + restrictions=True) + self.factory_reset(print_log=False) + time.sleep(120) + return output + if __name__ == '__main__': - basic= { + basic = { "target": "tip_2x", "controller": { "url": "https://sec-qa01.cicd.lab.wlan.tip.build:16001", @@ -557,4 +603,4 @@ if __name__ == '__main__': # a = obj.get_active_config() # if a == l: # print("a = l") - # print(obj.get_ap_version()) \ No newline at end of file + # print(obj.get_ap_version()) diff --git a/libs/tip_2x/controller.py b/libs/tip_2x/controller.py index 1691a77a3..6ec344101 100644 --- a/libs/tip_2x/controller.py +++ b/libs/tip_2x/controller.py @@ -6,6 +6,7 @@ import datetime import json import logging +import os import sys import time from operator import itemgetter @@ -33,13 +34,13 @@ class ConfigureController: # self.session = requests.Session() self.login_resp = self.login() self.gw_host, self.fms_host, \ - self.prov_host, self.owrrm_host, \ - self.owanalytics_host, self.owsub_host = self.get_gw_endpoint() + self.prov_host, self.owrrm_host, \ + self.owanalytics_host, self.owsub_host = self.get_gw_endpoint() if self.gw_host == "" or self.fms_host == "" or self.prov_host == "": time.sleep(60) self.gw_host, self.fms_host, \ - self.prov_host, self.owrrm_host, \ - self.owanalytics_host, self.owsub_host = self.get_gw_endpoint() + self.prov_host, self.owrrm_host, \ + self.owanalytics_host, self.owsub_host = self.get_gw_endpoint() if self.gw_host == "" or self.fms_host == "" or self.prov_host == "": self.logout() logging.info(self.gw_host, self.fms_host + self.prov_host) @@ -190,6 +191,7 @@ class ConfigureController: "Connection": "keep-alive", "Content-Type": "application/json", "Keep-Alive": "timeout=10, max=1000" + "conte" } return headers @@ -743,6 +745,49 @@ class Controller(ConfigureController): self.check_response("PUT", resp, self.make_headers(), payload, uri) return resp + def check_restrictions(self, serial_number): + uri = self.build_uri("device/" + serial_number + "/capabilities") + resp = requests.get(uri, headers=self.make_headers(), verify=False, timeout=120) + resp = resp.json() + if 'restrictions' in resp['capabilities'].keys(): + return True + else: + return False + + def asb_script(self, serial_number, payload): + uri = self.build_uri("device/" + serial_number + "/script") + payload = json.dumps(payload) + resp = requests.post(uri, data=payload, headers=self.make_headers(), verify=False, timeout=120) + resp = resp.json() + resp = resp['UUID'] + return resp + + def get_file(self, serial_number, uuid): + uri = self.build_uri("file/" + uuid + "?serialNumber=" + serial_number) + logging.info("Sending Command: " + "\n" + + "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" + + "URI: " + str(uri) + "\n" + + "Headers: " + str(self.make_headers())) + allure.attach(name="Sending Command:", body="Sending Command: " + "\n" + + "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" + + "URI: " + str(uri) + "\n" + + "Headers: " + str(self.make_headers())) + time.sleep(10) + resp = requests.get(uri, headers=self.make_headers(), verify=False, timeout=120) + self.check_response("GET", resp, self.make_headers(), "", uri) + if resp.headers.get("Transfer-Encoding") == "chunked": + file = resp.content + with open("gopi.tar.gz", "wb") as f: + for chunk in resp.iter_content(chunk_size=1024): + f.write(chunk) + else: + file = resp.content + logging.info(file) + with open("gopi.tar.gz", "wb") as f: + f.write(file) + allure.attach.file(name="file", source="gopi.tar.gz", extension=".tar") + return resp + class FMSUtils: @@ -2035,18 +2080,18 @@ class UProfileUtility: def set_radio_config(self, radio_config={}, open_roaming=False): if open_roaming: base_radio_config_2g = { - "band": "2G", - "country": "CA", - "channel-mode": "HT", - "channel-width": 20, - "channel": "auto" + "band": "2G", + "country": "CA", + "channel-mode": "HT", + "channel-width": 20, + "channel": "auto" } base_radio_config_5g = { - "band": "5G", - "country": "CA", - "channel-mode": "VHT", - "channel-width": 80, - "channel": "auto" + "band": "5G", + "country": "CA", + "channel-mode": "VHT", + "channel-width": 80, + "channel": "auto" } for band in radio_config: @@ -2201,7 +2246,8 @@ class UProfileUtility: ssid_info["pass-point"] = pass_point_data else: - ssid_info = {'name': ssid_data["ssid_name"], "bss-mode": "ap", "wifi-bands": [], "services": ["wifi-frames"]} + ssid_info = {'name': ssid_data["ssid_name"], "bss-mode": "ap", "wifi-bands": [], + "services": ["wifi-frames"]} for options in ssid_data: if options == "multi-psk": ssid_info[options] = ssid_data[options] diff --git a/tests/controller_tests/ucentral_gateway/test_gatewayservice.py b/tests/controller_tests/ucentral_gateway/test_gatewayservice.py index f2e672a51..4edc69f0f 100644 --- a/tests/controller_tests/ucentral_gateway/test_gatewayservice.py +++ b/tests/controller_tests/ucentral_gateway/test_gatewayservice.py @@ -5,7 +5,9 @@ """ import json +import logging import random +from time import sleep import allure import pytest @@ -278,6 +280,7 @@ class TestUcentralGatewayService(object): # print(resp.json()) # allure.attach(name="Device capabilities", body=str(resp.json()), #attachment_type=#allure.#attachment_type.JSON) assert resp.status_code == 200 + return resp @pytest.mark.gw_device_statistics @allure.title("Get Statistics") @@ -474,17 +477,93 @@ class TestUcentralGatewayService(object): # attachment_type=#allure.#attachment_type.JSON) assert resp.status_code == 200 + @pytest.mark.gw_rtty + @pytest.mark.ow_sdk_load_tests + @pytest.mark.owgw_api_tests + @allure.title("RTTY API") + def test_gw_service_get_rtty(self, get_target_object, get_testbed_details): + """ + Test the device rtty parameters in Gateway UI + """ + device_name = get_testbed_details['device_under_tests'][0]['identifier'] + resp = get_target_object.controller_library_object.get_rtty_params(device_name) + # print(resp.json()) + # allure.attach(name="Device RTTY parameters", body=str(resp.json()), #attachment_type=#allure.#attachment_type.JSON) + assert resp.status_code == 200 + + @pytest.mark.gw_asb_non_restricted + @allure.title("Asb script on non restricted AP") + @allure.testcase(name="WIFI-12235", url="https://telecominfraproject.atlassian.net/browse/WIFI-12235") + def test_asb_on_non_restricted_ap(self, get_target_object, get_testbed_details): + device_name = get_testbed_details['device_under_tests'][0]['identifier'] + payload = { + "serialNumber": device_name, + "timeout": 30, + "type": "diagnostic", + "script": "", + "scriptId": "", + "when": 0, + "signature": "", + "deferred": True, + "uri": "" + } + resp = resp = get_target_object.controller_library_object.check_restrictions(device_name) + if not resp: + logging.info("AP is in not restricted and we can trigger script") + uuid = get_target_object.controller_library_object.asb_script(device_name, payload) + resp = get_target_object.controller_library_object.get_file(device_name, uuid) + assert resp.status_code == 200 + else: + logging.info("AP is restricted, Removing Restrictions") + output = get_target_object.get_dut_library_object().remove_restrictions() + resp = resp = get_target_object.controller_library_object.check_restrictions(device_name) + if not resp: + logging.info("Removed Restrictions") + uuid = get_target_object.controller_library_object.asb_script(device_name, payload) + resp = get_target_object.controller_library_object.get_file(device_name, uuid) + assert resp.status_code == 200 + else: + logging.info("Unable to remove restrictions") + assert False + + @pytest.mark.gw_asb_restricted + @allure.title("Asb script on restricted AP") + @allure.testcase(name="WIFI-12236", url="https://telecominfraproject.atlassian.net/browse/WIFI-12236") + def test_asb_on_restricted_ap(self, get_target_object, get_testbed_details): + device_name = get_testbed_details['device_under_tests'][0]['identifier'] + payload = { + "serialNumber": device_name, + "timeout": 30, + "type": "diagnostic", + "script": "", + "scriptId": "", + "when": 0, + "signature": "", + "deferred": True, + "uri": "" + } + restrictions_file = 'echo \"{\\"country\\":[\\"US\\", \\"CA\\"],\\"dfs\\": true,\\"rtty\\": true,\\"tty\\": ' \ + 'true,\\"developer\\": true,\\"sysupgrade\\": true,\\"commands\\": true,\\"key_info\\": {' \ + '\\"vendor\\": \\"dummy\\",\\"algo\\": \\"static\\"}}\" > /certificates/restrictions.json ' \ + '&& echo \"True\"' + developer_mode = "fw_setenv developer 0" + output = get_target_object.get_dut_library_object().add_restrictions(restrictions_file=restrictions_file, + developer_mode=developer_mode) + resp = resp = get_target_object.controller_library_object.check_restrictions(device_name) + if resp: + logging.info("From GW it's confirmed that AP is restricted now") + uuid = get_target_object.controller_library_object.asb_script(device_name, payload) + resp = get_target_object.controller_library_object.get_file(device_name, uuid) + assert resp.status_code == 200 + else: + assert False + output = get_target_object.get_dut_library_object().remove_restrictions() + resp = resp = get_target_object.controller_library_object.check_restrictions(device_name) + if resp: + logging.info("Unable to remove restrictions in the AP") + assert False + else: + logging.info("Removed Restrictions") + + -@pytest.mark.gw_rtty -@pytest.mark.ow_sdk_load_tests -@pytest.mark.owgw_api_tests -@allure.title("RTTY API") -def test_gw_service_get_rtty(self, get_target_object, get_testbed_details): - """ - Test the device rtty parameters in Gateway UI - """ - device_name = get_testbed_details['device_under_tests'][0]['identifier'] - resp = get_target_object.controller_library_object.get_rtty_params(device_name) - # print(resp.json()) - # allure.attach(name="Device RTTY parameters", body=str(resp.json()), #attachment_type=#allure.#attachment_type.JSON) - assert resp.status_code == 200 \ No newline at end of file