Wifi 12235 (#784)

* Added ASB two test cases and functionality Signed-off-by: GopiRaga7 <gopi.raga@candelatech.com> * Resolved chunk error in the ASB API Signed-off-by: GopiRaga7 <gopi.raga@candelatech.com> * adding tar file in the tests folder Signed-off-by: GopiRaga7 <gopi.raga@candelatech.com> * sleeping after creating file Signed-off-by: GopiRaga7 <gopi.raga@candelatech.com> * sleeping after creating file2 Signed-off-by: GopiRaga7 <gopi.raga@candelatech.com> * more info Signed-off-by: GopiRaga7 <gopi.raga@candelatech.com> * more info2 Signed-off-by: GopiRaga7 <gopi.raga@candelatech.com> * adding full path Signed-off-by: GopiRaga7 <gopi.raga@candelatech.com> * changed the path Signed-off-by: GopiRaga7 <gopi.raga@candelatech.com> * added another file Signed-off-by: GopiRaga7 <gopi.raga@candelatech.com> * modified and added another file Signed-off-by: GopiRaga7 <gopi.raga@candelatech.com> * Added more logs Signed-off-by: GopiRaga7 <gopi.raga@candelatech.com> * Fixed issue of attaching tar file to allure ASB Signed-off-by: GopiRaga7 <gopi.raga@candelatech.com> * Asb script two testcases added Signed-off-by: GopiRaga7 <gopi.raga@candelatech.com> --------- Signed-off-by: GopiRaga7 <gopi.raga@candelatech.com>
2025-10-30 18:38:06 +00:00 · 2023-02-17 13:13:58 +05:30
parent 16c5a10ff2
commit aefa270425
3 changed files with 206 additions and 35 deletions
--- 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())
+    # print(obj.get_ap_version())
--- 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]
--- 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