connector sample

Author: katzy687

.gitignore

@@ -6,4 +6,5 @@ Orchestration-Templates/
 Bug-exploration/
 *.pyc
 credentials.py
-cli_credentials.py
+cli_credentials.py
+__pycache__/

resource-scripts/get_port_from_connector/__main__.py

@@ -0,0 +1,11 @@
+from port_from_connector import get_port_from_connector_alias
+import os
+
+TARGET_ALIAS_PARAM = "target_alias"
+
+
+target_alias = os.environ.get(TARGET_ALIAS_PARAM)
+if not target_alias:
+    raise ValueError(f"No value received for param '{TARGET_ALIAS_PARAM}'")
+
+get_port_from_connector_alias(target_alias)

resource-scripts/get_port_from_connector/debug.py

@@ -0,0 +1,15 @@
+from cloudshell.helpers.scripts.cloudshell_dev_helpers import attach_to_cloudshell_as
+from port_from_connector import get_port_from_connector_alias
+
+LIVE_SANDBOX_ID = "87df6742-0fd6-43e4-b068-48764b97d7e5"
+TARGET_RESOURCE_NAME = "DUT mock 1"
+TARGET_ALIAS = "1"
+
+attach_to_cloudshell_as(user="admin",
+                        password="admin",
+                        domain="Global",
+                        reservation_id=LIVE_SANDBOX_ID,
+                        server_address="localhost",
+                        resource_name=TARGET_RESOURCE_NAME)
+
+get_port_from_connector_alias(TARGET_ALIAS)

resource-scripts/get_port_from_connector/port_from_connector.py

@@ -0,0 +1,38 @@
+from cloudshell.helpers.scripts.cloudshell_scripts_helpers import get_reservation_context_details, get_api_session, \
+    get_resource_context_details
+from cloudshell.api.cloudshell_api import Connector
+
+
+def _get_connected_resource(connector: Connector, root_resource_name: str):
+    if connector.Source.startswith(root_resource_name):
+        target_side = connector.Source
+    elif connector.Target.startswith(root_resource_name):
+        target_side = connector.Target
+    else:
+        raise ValueError(f"root resource {root_resource_name} not present in Source or Target of connector")
+
+    connected_port = target_side.split("/")[-1]
+    return connected_port
+
+
+def get_port_from_connector_alias(target_alias: str):
+    api = get_api_session()
+    sb_details = get_reservation_context_details()
+    sb_id = sb_details.id
+
+    resource_details = get_resource_context_details()
+    resource_name = resource_details.name
+
+    api.WriteMessageToReservationOutput(reservationId=sb_id,
+                                        message=f"Searching for port connected to alias '{target_alias}'...")
+
+    reservation_details = api.GetReservationDetails(reservationId=sb_id, disableCache=True).ReservationDescription
+    connectors = reservation_details.Connectors
+    target_connector_search = [x for x in connectors if x.Alias == target_alias]
+    if not target_connector_search:
+        raise ValueError(f"Connector with Alias {target_alias} not found")
+    target_connector = target_connector_search[0]
+    connected_resource = _get_connected_resource(target_connector, resource_name)
+
+    # printing to std_out will be the return output of resource scripts
+    print(connected_resource)

resource-scripts/get_port_from_connector/requirements.txt

@@ -0,0 +1 @@
+cloudshell-automation-api