FMS initial Library and firmware upgrade logic

Signed-off-by: shivamcandela <shivam.thakur@candelatech.com>

libs/controller/controller_2x/controller.py

@@ -34,13 +34,18 @@ class ConfigureController:
         self.access_token = ""
         # self.session = requests.Session()
         self.login_resp = self.login()
-        self.gw_host = self.get_gw_endpoint()
+        self.gw_host, self.fms_host = self.get_gw_endpoint()
 
     def build_uri_sec(self, path):
         new_uri = 'https://%s:%d/api/v1/%s' % (self.host.hostname, self.host.port, path)
         print(new_uri)
         return new_uri
 
+    def build_url_fms(self, path):
+        new_uri = 'https://%s:%d/api/v1/%s' % (self.fms_host.hostname, self.fms_host.port, path)
+        print(new_uri)
+        return new_uri
+
     def build_uri(self, path):
 
         new_uri = 'https://%s:%d/api/v1/%s' % (self.gw_host.hostname, self.gw_host.port, path)
@@ -92,9 +97,11 @@ class ConfigureController:
         print(resp)
         self.check_response("GET", resp, self.make_headers(), "", uri)
         services = resp.json()
-        print(services["endpoints"][0]["uri"])
+        print("REST ENDPOINT: ", services["endpoints"][0]["uri"])
+        print("FMS ENDPOINT: ", services["endpoints"][1]["uri"])
         gw_host = urlparse(services["endpoints"][0]["uri"])
-        return gw_host
+        fms_host = urlparse(services["endpoints"][1]["uri"])
+        return gw_host, fms_host
 
     def logout(self):
         uri = self.build_uri_sec('oauth2/%s' % self.access_token)
@@ -155,6 +162,73 @@ class Controller(ConfigureController):
         return device_info["UUID"]
 
 
+class FMSUtils:
+
+    def __init__(self, sdk_client=None, controller_data=None):
+        if sdk_client is None:
+            self.sdk_client = Controller(controller_data=controller_data)
+        self.sdk_client = sdk_client
+
+    def upgrade_firmware(self, serial="", url=""):
+        response = self.sdk_client.request(service="gw", command="device/" + serial + "upgrade/",
+                                           method="POST", params="revisionSet=true",
+                                           payload="{ \"serialNumber\" : " + serial + " , \"uri\" : " + url + " }")
+        print(response)
+        pass
+
+    def ap_model_lookup(self, model=""):
+        devices = self.get_device_set()
+        model_name = ""
+        for device in devices['deviceTypes']:
+            if str(device).__contains__(model):
+                model_name = device
+        return model_name
+
+    def get_revisions(self):
+        response = self.sdk_client.request(service="fms", command="firmwares/", method="GET", params="revisionSet=true",
+                                           payload="")
+        if response.status_code == 200:
+            return response.json()
+        else:
+            return {}
+
+    def get_latest_fw(self, model=""):
+
+        device_type = self.ap_model_lookup(model=model)
+
+        response = self.sdk_client.request(service="fms", command="firmwares/", method="GET",
+                                           params="latestOnly=true&deviceType=" + device_type,
+                                           payload="")
+        if response.status_code == 200:
+            return response.json()
+        else:
+            return {}
+
+    def get_device_set(self):
+        response = self.sdk_client.request(service="fms", command="firmwares/", method="GET", params="deviceSet=true",
+                                           payload="")
+        if response.status_code == 200:
+            return response.json()
+        else:
+            return {}
+
+    def get_firmwares(self, limit="", deviceType="", latestonly=""):
+
+        params = "limit=" + limit + "&deviceType=" + deviceType + "&latestonly=" + latestonly
+
+        response = self.sdk_client.request(service="fms", command="firmwares/", method="GET", params=params, payload="")
+        print(response)
+        if response.status_code == 200:
+            data = response.json()
+            print(data)
+
+        # resp = requests.get(uri, headers=self.sdk_client.make_headers(), verify=False, timeout=100)
+        # self.sdk_client.check_response("GET", resp, self.sdk_client.make_headers(), "", uri)
+        # devices = resp.json()
+        # # resp.close()()
+        return "devices"
+
+
 class UProfileUtility:
 
     def __init__(self, sdk_client=None, controller_data=None):
@@ -435,13 +509,17 @@ if __name__ == '__main__':
         'password': 'openwifi',
     }
     obj = Controller(controller_data=controller)
-    profile = UProfileUtility(sdk_client=obj)
-    profile.set_mode(mode="BRIDGE")
-    profile.set_radio_config()
-    ssid = {"ssid_name": "ssid_wpa2_2g", "appliedRadios": ["2G", "5G"], "security": "psk", "security_key": "something",
-            "vlan": 100}
-    profile.add_ssid(ssid_data=ssid, radius=False)
-    profile.push_config(serial_number="903cb39d6918")
-    print(profile.get_ssid_info())
-    # print(obj.get_devices())
+    fms = FMSUtils(sdk_client=obj)
+    # fms.get_device_set()
+    model = fms.get_latest_fw(model="eap102")
+    print(model)
+    # profile = UProfileUtility(sdk_client=obj)
+    # profile.set_mode(mode="BRIDGE")
+    # profile.set_radio_config()
+    # ssid = {"ssid_name": "ssid_wpa2_2g", "appliedRadios": ["2G", "5G"], "security": "psk", "security_key": "something",
+    #         "vlan": 100}
+    # profile.add_ssid(ssid_data=ssid, radius=False)
+    # profile.push_config(serial_number="903cb39d6918")
+    # print(profile.get_ssid_info())
+    # # print(obj.get_devices())
     obj.logout()

tests/configuration.py

@@ -169,7 +169,7 @@ CONFIGURATION = {
                 'password': "pumpkin77",
                 'port': 22,
                 'jumphost_tty': "/dev/ttyAP5",
-                'version': "https://tip.jfrog.io/artifactory/tip-wlan-ap-firmware/uCentral/edgecore_eap102/20210625-edgecore_eap102-uCentral-trunk-4225122-upgrade.bin"
+                'version': "latest"
             }
         ],
         "traffic_generator": {

tests/conftest.py

@@ -654,3 +654,9 @@ def fixtures_ver(request, get_configuration):
         print("1.x")
         obj = Fixtures_1x(configuration=get_configuration)
     yield obj
+
+
+@pytest.fixture(scope="session")
+def firmware_upgrade(fixtures_ver, get_apnos, get_configuration):
+    fixtures_ver.setup_firmware(get_apnos, get_configuration)
+    yield True

tests/fixtures_2x.py

@@ -29,6 +29,7 @@ if 'py-json' not in sys.path:
     sys.path.append('../py-scripts')
 from apnos.apnos import APNOS
 from controller.controller_2x.controller import Controller
+from controller.controller_2x.controller import FMSUtils
 from configuration import CONFIGURATION
 from configuration import RADIUS_SERVER_DATA
 from configuration import RADIUS_ACCOUNTING_DATA
@@ -42,6 +43,7 @@ class Fixtures_2x:
         print("2.X")
         try:
             self.controller_obj = Controller(controller_data=self.lab_info["controller"])
+            self.fw_client = FMSUtils(sdk_client=self.controller_obj)
         except Exception as e:
             print(e)
             allure.attach(body=str(e), name="Controller Instantiation Failed: ")
@@ -51,7 +53,35 @@ class Fixtures_2x:
     def disconnect(self):
         self.controller_obj.logout()
 
-    def setup_firmware(self):
+    def setup_firmware(self, get_apnos, get_configuration):
+        # Query AP Firmware
+        for ap in get_configuration['access_point']:
+            firmware_url = ""
+            ap_ssh = get_apnos(ap, pwd="../libs/apnos/", sdk="2.x")
+            ap_version = ap_ssh.get_ap_version_ucentral()
+            if ap['version'] == "latest":
+                response = self.fw_client.get_latest_fw(model=ap["model"])
+                if 'revision' in list(response.keys()):
+                    version_latest = response['revision']
+                    print(ap_version, version_latest)
+                    print(str(ap_version).replace(" ", ""), str(version_latest).replace(" ", ""),
+                          str(version_latest).replace(" ", "") == 'OpenWrt21.02-SNAPSHOTr16011+75-6fd65c6573/TIP-v2.1.0-rc2-d0a0715')
+                    if str(ap_version).replace(" ", "") == str(version_latest).replace(" ", ""):
+                        continue
+                    else:
+                        if 'uri' in list(response.keys()):
+                            firmware_url = response['uri']
+
+                        else:
+                            continue
+                        pass
+                else:
+                    continue
+
+
+
+        # Compare with the specified one
+        # if 'latest'
         pass
 
     def get_ap_version(self, get_apnos, get_configuration):

tests/test_connectivity.py

@@ -9,6 +9,26 @@ pytestmark = [pytest.mark.test_resources, pytest.mark.sanity,
               pytest.mark.sanity_55]
 
 
+
+
+@pytest.mark.fw
+def test_firmware(firmware_upgrade):
+    assert True
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
 @allure.testcase(name="Test Resources", url="")
 class TestResources(object):
     """Test Case Class: Test cases to cover resource Connectivity"""