update services script added

Author: katzy687

automation_api_scripts/__init__.py

@@ -0,0 +1,11 @@
+from cloudshell.api.cloudshell_api import CloudShellAPISession
+
+SANDBOX_ID = "6c30767e-c495-4d3a-9a1b-c25a335d8205"
+
+api = CloudShellAPISession(host="localhost", username="admin", password="admin", domain="Global")
+res = api.AddRoutesToReservation(reservationId=SANDBOX_ID,
+                                 sourceResourcesFullPath="DUT Mock 2/Port 1",
+                                 targetResourcesFullPath="DUT Mock 2/Port 1",
+                                 mappingType="bi",
+                                 routeAlias="test api")
+pass

automation_api_scripts/add_loopback_l1_route.py

@@ -0,0 +1,12 @@
+import time
+
+from cloudshell.api.cloudshell_api import CloudShellAPISession
+
+# TARGET_RESOURCE = "L1Mock1"
+TARGET_RESOURCE = "DUT Mock 2"
+api = CloudShellAPISession(host="localhost", username="admin", password="admin", domain="Global")
+res = api.GetResourceMappings(resources=[TARGET_RESOURCE])
+# res = api.MapPorts(sourcePort="L1Mock1/Blade 1/Port 004",
+#                    destinationPort="L1Mock1/Blade 1/Port 004",
+#                    mappingType="bi")
+pass

automation_api_scripts/get_mappings.py

@@ -0,0 +1,30 @@
+from cloudshell.api.cloudshell_api import CloudShellAPISession
+from quali_api import QualiAPISession
+import credentials
+from update_services import update_services_flow
+
+
+def main():
+    import argparse
+    parser = argparse.ArgumentParser()
+    parser.add_argument("target_service")
+    parser.add_argument("old_vlan")
+    parser.add_argument("new_vlan")
+    args = parser.parse_args()
+    target_service = args.target_service
+    old_vlan = args.old_vlan
+    new_vlan = args.new_vlan
+
+    my_api = CloudShellAPISession(host=credentials.HOST,
+                                  username=credentials.USER,
+                                  password=credentials.PASSWORD,
+                                  domain=credentials.DOMAIN)
+    my_quali_api = QualiAPISession(host=credentials.HOST,
+                                   username=credentials.USER,
+                                   password=credentials.PASSWORD,
+                                   domain=credentials.DOMAIN)
+    update_services_flow(my_api, my_quali_api, target_service, old_vlan, new_vlan)
+
+
+if __name__ == "__main__":
+    main()

packaging-api-samples/__init__.py

@@ -0,0 +1,2 @@
+VLAN_MANUAL_SERVICE = "VLAN Manual"
+VLAN_ID_ATTR = "VLAN ID"

packaging-api-samples/update-services/cli.py

@@ -0,0 +1,39 @@
+from cloudshell.api.cloudshell_api import CloudShellAPISession
+from typing import List
+from timeit import default_timer
+
+
+def find_blueprints_flow(api: CloudShellAPISession, target_service_name: str, old_vlan: str) -> List[str]:
+    print("starting blueprint search")
+    start = default_timer()
+    all_blueprints = api.GetTopologiesByCategory().Topologies
+    print(f"System Blueprint Count: {len(all_blueprints)}")
+
+    target_blueprints = []
+    for blueprint in all_blueprints:
+        services = api.GetTopologyDetails(blueprint).Services
+        for curr_service in services:
+            if curr_service.ServiceName == target_service_name:
+                attrs = curr_service.Attributes
+                vlan_id = [x for x in attrs if x.Name == "VLAN ID"][0].Value
+                if vlan_id == old_vlan:
+                    print(f"found target blueprint {blueprint}")
+                    target_blueprints.append(blueprint)
+                    break
+    print(f"Target blueprints found: {len(target_blueprints)}")
+    print(f"search done after {int(default_timer() - start)} seconds")
+    # need only the blueprint name for quali api, not the full path
+    return [x.split("/")[-1] for x in target_blueprints]
+
+
+if __name__ == "__main__":
+    import credentials
+    import constants
+
+    OLD_VLAN = "43"
+    api = CloudShellAPISession(host=credentials.HOST,
+                               username=credentials.USER,
+                               password=credentials.PASSWORD,
+                               domain=credentials.DOMAIN)
+    found = find_blueprints_flow(api, constants.VLAN_MANUAL_SERVICE, OLD_VLAN)
+    print(f"found: {found}")

packaging-api-samples/update-services/constants.py

@@ -0,0 +1,41 @@
+import os
+
+import update_package
+from quali_api import QualiAPISession
+from timeit import default_timer
+
+
+def process_blueprint_flow(quali_api: QualiAPISession, blueprint_name: str, target_service: str, new_vlan_id: str):
+
+    print(f"processing blueprint {blueprint_name}")
+    start = default_timer()
+
+    # download package
+    zip_name = f"{blueprint_name}.zip"
+    quali_api.export_package(blueprint_full_names=[blueprint_name],
+                             file_path=zip_name)
+
+    # remove unneeded files
+    update_package.clean_bp_archive(zip_name)
+
+    # update target attribute
+    update_package.edit_vlan_id_in_package(zip_name, target_service, new_vlan_id)
+
+    # re-upload
+    quali_api.import_package(zip_name)
+
+    # delete zip archive
+    os.remove(zip_name)
+    print(f"done processing {blueprint_name} after {int(default_timer() - start)} seconds")
+
+
+if __name__ == "__main__":
+    import credentials
+    import constants
+    TARGET_BLUEPRINT = "vlan dev"
+    NEW_VLAN_ID = "122"
+    api = QualiAPISession(host=credentials.HOST,
+                          username=credentials.USER,
+                          password=credentials.PASSWORD,
+                          domain=credentials.DOMAIN)
+    process_blueprint_flow(api, TARGET_BLUEPRINT, constants.VLAN_MANUAL_SERVICE, NEW_VLAN_ID)

packaging-api-samples/update-services/find_blueprints.py

@@ -0,0 +1,55 @@
+import requests
+from typing import List
+
+
+class QualiAPISession:
+    def __init__(self, host, username='', password='', domain='Global', token_id='', port=9000):
+        self.session = requests.Session()
+        self.base_url = f"http://{host}:{port}/Api"
+        self._json_header = {"Content-Type": "application/json"}
+        self.login(username, password, token_id, domain)
+
+    def login(self, username="", password="", token_id="", domain="Global"):
+        if token_id:
+            body = {"Token": token_id, "Domain": domain}
+        elif username and password:
+            body = {"Username": username, "Password": password, "Domain": domain}
+        else:
+            raise ValueError("Must supply Username / Password OR a token_id")
+        url = f"{self.base_url}/Auth/Login"
+        login_result = requests.put(url, json=body, headers=self._json_header)
+        if not login_result.ok:
+            raise Exception(f"Failed Quali API login. Status {login_result.status_code}: {login_result.text}")
+
+        # strip the extraneous quotes
+        token_str = login_result.text[1:-1]
+        self.session.headers.update({"Authorization": f"Basic {token_str}"})
+
+    @staticmethod
+    def _validate_response(response: requests.Response):
+        if not response.ok:
+            response.raise_for_status()
+
+    def import_package(self, package_filepath: str):
+        """
+        Push a zipped Package file into the CloudShell Server
+        provide full path to zip file
+        """
+        url = f"{self.base_url}/Package/ImportPackage"
+        with open(package_filepath, "rb") as package_file:
+            package_content = package_file.read()
+            response = self.session.post(url, files={'QualiPackage': package_content})
+        self._validate_response(response)
+
+    def export_package(self, blueprint_full_names: List[str], file_path: str):
+        """
+        Export one or more multiple blueprints from CloudShell as a Quali Package
+        pass file name to export to local dir, or full path
+        """
+        url = f"{self.base_url}/Package/ExportPackage"
+        body = {"TopologyNames": blueprint_full_names}
+        response = self.session.post(url, json=body, headers=self._json_header)
+        self._validate_response(response)
+        with open(file_path, "wb") as target_file:
+            target_file.write(response.content)
+

packaging-api-samples/update-services/process_blueprint.py

@@ -0,0 +1,4 @@
+cloudshell-automation-api
+quali-utils # from cloudshell download center
+requests
+cloudshell-scriptfoundry

packaging-api-samples/update-services/quali_api.py

@@ -0,0 +1,65 @@
+import os
+import shutil
+import zipfile
+
+from quali_utils.quali_packaging import PackageEditor
+import constants
+
+
+def _clean_bp_package(base_path: str):
+    """
+    clean out unwanted folders
+    """
+    dir_contents = os.listdir(base_path)
+    for item in dir_contents:
+        if item.endswith(".xml"):
+            continue
+        if item == "Topologies":
+            continue
+        full_path = os.path.join(base_path, item)
+        shutil.rmtree(full_path)
+
+
+def clean_bp_archive(package_path: str):
+    """
+    1. unzip the package
+    2. clean out package for everything except Topology xml
+    3. zip everything back up
+    """
+    temp_folder = "temp_bp_extracted"
+
+    # UNZIP PACKAGE TO TEMP FOLDER
+    with zipfile.ZipFile(package_path, 'r') as zip_ref:
+        zip_ref.extractall(temp_folder)
+
+    # CLEAR OUT EVERYTHING EXCEPT APP TEMPLATES AND METADATA XML
+    _clean_bp_package(temp_folder)
+
+    # REZIP TEMP FOLDER BACK TO ZIP PACKAGE
+    package_file_name = package_path.split("/")[-1].split(".zip")[0]  # handle full path and remove zip extension
+    shutil.make_archive(package_file_name, 'zip', temp_folder)
+
+    # clean up temp folder
+    shutil.rmtree(temp_folder)
+
+
+def edit_vlan_id_in_package(package_path: str, target_service: str, new_vlan_id: str):
+    p = PackageEditor()
+    p.load(package_path)
+    bp_names = p.get_topology_names()
+    if len(bp_names) > 1:
+        raise ValueError("This script assumes only one blueprint per package")
+    bp_name = bp_names[0]
+    p.set_attribute_to_service(topology_name=bp_name,
+                               service_alias=target_service,
+                               attribute_name=constants.VLAN_ID_ATTR,
+                               attribute_value=new_vlan_id,
+                               publish=False)
+    pass
+
+
+if __name__ == "__main__":
+    TARGET_PACKAGE = "vlan dev.zip"
+    TARGET_SERVICE = "VLAN Manual"
+    clean_bp_archive(TARGET_PACKAGE)
+    edit_vlan_id_in_package(TARGET_PACKAGE, TARGET_SERVICE, "42")

packaging-api-samples/update-services/requirements.txt

@@ -0,0 +1,43 @@
+from cloudshell.api.cloudshell_api import CloudShellAPISession
+from quali_api import QualiAPISession
+from find_blueprints import find_blueprints_flow
+from process_blueprint import process_blueprint_flow
+from timeit import default_timer
+
+
+def update_services_flow(api: CloudShellAPISession,
+                         quali_api: QualiAPISession,
+                         target_service: str,
+                         old_vlan: str,
+                         new_vlan: str):
+    start = default_timer()
+    print(f"Starting Flow. Changing service '{target_service}' VLAN {old_vlan} --> {new_vlan}")
+
+    # find target blueprints
+    target_blueprints = find_blueprints_flow(api, target_service, old_vlan)
+    if not target_blueprints:
+        print("No blueprints found")
+        return
+
+    # process bluperints
+    for curr_bp in target_blueprints:
+        process_blueprint_flow(quali_api, curr_bp, target_service, new_vlan)
+
+    print(f"Update flow done after total {int(default_timer() - start)} seconds")
+
+
+if __name__ == "__main__":
+    import credentials
+    import constants
+
+    OLD_VLAN_ID = "122"
+    NEW_VLAN_ID = "129"
+    my_api = CloudShellAPISession(host=credentials.HOST,
+                                  username=credentials.USER,
+                                  password=credentials.PASSWORD,
+                                  domain=credentials.DOMAIN)
+    my_quali_api = QualiAPISession(host=credentials.HOST,
+                                   username=credentials.USER,
+                                   password=credentials.PASSWORD,
+                                   domain=credentials.DOMAIN)
+    update_services_flow(my_api, my_quali_api, constants.VLAN_MANUAL_SERVICE, OLD_VLAN_ID, NEW_VLAN_ID)