Fixed issue #49

aos/blueprint.py

@@ -553,6 +553,8 @@ def is_task_active(self, bp_id: str, task_id: str) -> bool:
             ]:
                 return True
             else:
+                if task["status"] in TaskStatus.failed.value:
+                    print(json.dumps(task["detailed_status"]['errors'], indent=2))
                 return False
 
         else:
@@ -833,6 +835,9 @@ def get_bp_nodes(self, bp_id: str, node_type: str = None):
     def get_bp_node_by_id(self, bp_id: str, node_id: str):
         return self.rest.json_resp_get(f"/api/blueprints/{bp_id}/nodes/{node_id}")
 
+    def get_bp_node_asn(self, bp_id: str, node_id: str):
+        return self.rest.json_resp_get(f"/api/blueprints/{bp_id}/systems/{node_id}/domain")
+
     def get_bp_system_nodes(self, bp_id: str):
 
         return self.get_bp_nodes(bp_id, "system")
@@ -969,6 +974,10 @@ def get_all_tor_nodes(self, bp_id):
 
         return nodes
 
+    def get_device_rendered_context(self, bp_id: str, node_id: str):
+        uri = f"/api/blueprints/{bp_id}/nodes/{node_id}/config-context"
+        return self.rest.json_resp_get(uri)
+
     def create_switch_system_links(self, bp_id: str, data: dict):
         uri = f"/api/blueprints/{bp_id}/switch-system-links"
 
@@ -1300,6 +1309,71 @@ def get_routing_policies(self, bp_id: str, bp_type="staging") -> Dict:
             f"/api/blueprints/{bp_id}/routing-policies?type={bp_type}"
         )
 
+    def get_routing_policy(self, bp_id: str, rp_name: str) -> Dict:
+        for rp in self.get_routing_policies(bp_id)['items']:
+            if rp.get('label') == rp_name:
+                return rp
+
+    def get_floating_ips(self, bp_id: str, vn_id: str) -> Dict:
+        all_floating_ips = self.rest.json_resp_get(
+            f"/api/blueprints/{bp_id}/floating-ips"
+        )
+        filtered_results = list()
+        for i in all_floating_ips.get('items'):
+            print(i)
+            if i.get('virtual_network_id') == "WH2Qb8Yi2Dyot5A4qQc":
+                filtered_results.append(i)
+        return filtered_results
+
+    def create_routing_policy(self,
+                                bp_id: str,
+                                label : str,
+                                aggregate_prefixes: list = [],
+                                description: str = "",
+                                expect_default_ipv4_route: bool = False,
+                                expect_default_ipv6_route: bool = False,
+                                l2edge_subnets: bool = False,
+                                l3edge_server_links: bool = False,
+                                loopbacks: bool = False,
+                                spine_leaf_links: bool = False,
+                                spine_superspine_links: bool = False,
+                                static_routes: bool = False,
+                                extra_export_routes: list = [],
+                                extra_import_routes: list = [],
+                                import_policy: str = "default_only"
+                              ):
+        #Example of extra_import_routes or extra_export_routes
+        # export_routes = [{'action': 'permit',
+        #                         'ge_mask': None,
+        #                         'le_mask': 32,
+        #                         'prefix': '172.29.2.0/24'}]
+
+        if import_policy not in ("extra_only", "default_only", "all"):
+            raise AosAPIResourceNotFound("Import policy not allowed")
+
+        data = {
+        'aggregate_prefixes': aggregate_prefixes,
+        'description': description,
+        'expect_default_ipv4_route': expect_default_ipv4_route,
+        'expect_default_ipv6_route': expect_default_ipv6_route,
+        'export_policy': {
+                'l2edge_subnets': l2edge_subnets,
+                'l3edge_server_links': l3edge_server_links,
+                'loopbacks': loopbacks,
+                'spine_leaf_links': spine_leaf_links,
+                'spine_superspine_links': spine_superspine_links,
+                'static_routes': static_routes,
+                },
+        'extra_export_routes': extra_export_routes,
+        'extra_import_routes': extra_import_routes,
+        'import_policy': import_policy,
+        'label': label,
+        'policy_type': 'user_defined'}
+
+        rp_path = f"/api/blueprints/{bp_id}/routing-policies"
+        return self.rest.json_resp_post(rp_path, data=data)
+
+
     # External Routers
     def get_external_routers_all(self, bp_id: str):
         """
@@ -1991,6 +2065,7 @@ def create_virtual_network(
         ipv6_gateway: str = None,
         tagged_ct: bool = False,
         untagged_ct: bool = False,
+        dhcp_service: bool = False,
         timeout: int = 60,
     ):
         """
@@ -2042,6 +2117,8 @@ def create_virtual_network(
         untagged_ct
             (bool) - (optional) Create untagged connectivity template for the given
             virtual-network.
+        dhcp_service
+            (bool) - (optimal) Enables DHCP service on virtual network
         timeout
             (int) - time (seconds) to wait for creation
 
@@ -2063,12 +2140,15 @@ def create_virtual_network(
             "vn_id": vn_id,
             "bound_to": bound_to,
             "ipv4_enabled": True,
-            "dhcp_service": "dhcpServiceEnabled",
             "ipv4_subnet": ipv4_subnet,
             "ipv4_gateway": ipv4_gateway,
         }
 
+        if dhcp_service:
+            virt_net['dhcp_service'] = "dhcpServiceEnabled"
+
         if ipv6_enabled:
+            virt_net["ipv6_enabled"] = True
             virt_net["ipv6_subnet"] = ipv6_subnet
             virt_net["ipv6_gateway"] = ipv6_gateway
 
@@ -2270,3 +2350,7 @@ def get_node_relationships(
         }
 
         return self.rest.json_resp_get(url, params=params)['relationships']
+
+    def get_protocol_sessions(self, bp_id: str):
+        url = f'/api/blueprints/{bp_id}/protocol-sessions'
+        return self.rest.json_resp_get(url)['protocol-sessions']

aos/external_systems.py

@@ -77,3 +77,4 @@ def iter_all(self) -> Generator[ExternalRouter, None, None]:
 
     def find_by_name(self, rtr_name: str) -> List[ExternalRouter]:
         return [r for r in self.iter_all() if r.display_name == rtr_name]
+

docs/example-scripts/blueprint/routing_zone.md

@@ -46,6 +46,7 @@ aos.blueprint.create_security_zone(
     bp_id=bp.id,
     name=sz_name,
     leaf_loopback_ip_pools=[leaf_lo_pool.id],
+    leaf_loopback_ipv6_pools=[leaf_lo_pool.id],
     dhcp_servers=[dhcp_rely_addr],
 )
 ```

tools/ct-template-generator.py

@@ -0,0 +1,60 @@
+#
+# 2023 Per Abildgaard Toft <[email protected]
+# Tool to parse and replace uuids from apstra ct template
+#
+import json
+import re
+import pprint
+
+TEMPLATE = "templates/k666_ct.json"
+
+uuid_index = 0
+id_table = dict()
+path = []
+
+def get_uuid_index(id):
+    global uuid_index
+    global id_ta
+    if not id in id_table:
+        id_table[id] = "{{" + f"uuid_idx{uuid_index}" + "}}"
+        #Increment counter
+        uuid_index += 1
+    return id_table[id]
+
+
+def find_id(json):
+    if isinstance(json,dict):
+        for k,v in json.items():
+            path.append(k)
+            if k == "id":
+                #print(f"Found an ID json path: {str(path)} value: {v}")
+                json[k] = get_uuid_index(v)
+            if k == "vn_node_id":
+                json[k] = "{{vn_node_id}}"
+            if k == "rp_to_attach":
+                json[k] = "{{rp_to_attach}}"
+            find_id(json[k])
+            path.pop()
+    elif isinstance(json,list):
+        for i in json:
+            find_id(i)
+    # elif isinstance(json, (str,bool,int) ) :
+    #     print(f'\t {json}')
+
+if __name__ == '__main__':
+    fp = open(TEMPLATE,"r")
+    data_json = json.load(fp)
+    find_id(data_json)
+    fp.close()
+    data_str = json.dumps(data_json,indent=2)
+
+    for i in id_table:
+        data_str = re.sub(str(i),id_table[i],data_str,count=0, flags=0)
+
+    print("IDs have been replaced with the following uuid tokens")
+    for j in id_table.values():
+        print(j)
+
+    fp_out = open(TEMPLATE+"_new","w")
+    fp_out.write(data_str)
+    fp_out.close()

tools/decode-nginx-request-body.py

@@ -0,0 +1,3 @@
+
+line = "{\x22application_points\x22:[{\x22id\x22:\x22fK731ak0Z73B5vZfSSA\x22,\x22policies\x22:[{\x22policy\x22:\x22d6d35349-cf19-4e39-9390-f11601f7a7c6\x22,\x22used\x22:true}]}]}"
+print(bytes(line, 'utf-8').decode('unicode_escape'))