add peer as support in numbered mode + tests

Author: loulecrivain

cosmo/clients/queries/connected_devices.graphql

@@ -11,6 +11,7 @@ query {
           name
           device {
             name
+            custom_fields
             __typename
             primary_ip4 {
               __typename

cosmo/netbox_types.py

@@ -289,6 +289,13 @@ def getISISIdentifier(self) -> str | None | Never:
             )
         return str(sys_id)
 
+    def getAssignedDeviceASN(self) -> int | None:
+        asn: Any | None = self.getCustomFields().get("ASN")
+        if asn:
+            return int(asn)
+        else:
+            return None
+
 
 class DeviceTypeType(AbstractNetboxType):
     def getBasePath(self):

cosmo/routerbgpcpevisitor.py

@@ -50,6 +50,11 @@ def processNumberedBGP(
         v6_neighbors = set()
 
         t_cpe = cpe["device"]
+        asn_constraint = {"any_as": True}
+        assigned_asn = t_cpe.getAssignedDeviceASN()
+        if assigned_asn:
+            asn_constraint = {"peer_as": assigned_asn}
+
         for item in iter(t_cpe):
             other_ipa = CpeRouterIPVisitor(other_ip_networks).accept(item)
             if not other_ipa:
@@ -58,9 +63,9 @@ def processNumberedBGP(
                 v4_neighbors.add(str(other_ipa.ip))
             elif type(other_ipa) is IPv6Interface:
                 v6_neighbors.add(str(other_ipa.ip))
+
         if v4_neighbors:
-            groups[f"{base_group_name}_V4"] = {
-                "any_as": True,
+            groups[f"{base_group_name}_V4"] = asn_constraint | {
                 "local_address": str(own_ipv4_address.ip),
                 "neighbors": list(map(lambda n: {"peer": n}, v4_neighbors)),
                 "family": {
@@ -71,8 +76,7 @@ def processNumberedBGP(
                 },
             }
         if v6_neighbors:
-            groups[f"{base_group_name}_V6"] = {
-                "any_as": True,
+            groups[f"{base_group_name}_V6"] = asn_constraint | {
                 "local_address": str(own_ipv6_address.ip),
                 "neighbors": list(map(lambda n: {"peer": n}, v6_neighbors)),
                 "family": {

cosmo/tests/test_case_bgpcpe.yml

@@ -4,6 +4,93 @@ device_list:
     slug: S9600-72XC
   id: '17799'
   interfaces:
+  - custom_fields:
+      bpdufilter: false
+      inner_tag: null
+      outer_tag: null
+      storm_control__broadcast: null
+      storm_control__multicast: null
+      storm_control__unknown_unicast: null
+    __typename: InterfaceType
+    description: ''
+    enabled: true
+    id: '191980'
+    ip_addresses: [ ]
+    lag: null
+    mac_address: null
+    mode: null
+    mtu: null
+    name: ifp-0/1/3
+    tagged_vlans: [ ]
+    tags:
+      - name: speed:1g
+        slug: speed1g
+        __typename: TagType
+    type: A_25GBASE_X_SFP28
+    untagged_vlan: null
+    vrf: null
+    parent: null
+    connected_endpoints:
+      - name: "xe-0/1/2.3"
+        __typename: InterfaceType
+        device:
+          __typename: DeviceType
+          name: 'CPE4000'
+          custom_fields:
+            ASN: 65086
+          primary_ip4: null
+          interfaces:
+            - ip_addresses: [ ]
+              id: ''
+              name: 'xe-0/1/2.0'
+              __typename: InterfaceType
+            - ip_addresses:
+                - address: 10.129.6.12/29
+                  __typename: IPAddressType
+                - address: 2a0e:b941:3::21/122
+                  __typename: IPAddressType
+              id: ''
+              name: 'xe-0/1/2.3'
+              __typename: InterfaceType
+            - ip_addresses: []
+              id: ''
+              name: 'mgmt0'
+              __typename: InterfaceType
+  - custom_fields:
+      bpdufilter: false
+      inner_tag: null
+      storm_control__broadcast: null
+      storm_control__multicast: null
+      storm_control__unknown_unicast: null
+    __typename: InterfaceType
+    description: ''
+    enabled: true
+    id: '191990'
+    ip_addresses:
+      - address: 10.129.6.11/29
+        __typename: IPAddressType
+    lag: null
+    mac_address: null
+    mode: ACCESS
+    mtu: 1500
+    name: ifp-0/1/3.3
+    tagged_vlans: []
+    tags:
+    - name: bgp:cpe
+      slug: bgpcpe
+      __typename: TagType
+    type: VIRTUAL
+    untagged_vlan:
+      __typename: VLANType
+      id: '1003'
+      name: TEST-VLAN3
+      vid: 3
+    vrf: null
+    parent:
+      __typename: InterfaceType
+      id: '191980'
+      name: "ifp-0/1/3"
+      mtu: null
   - custom_fields:
       bpdufilter: false
       inner_tag: null

cosmo/tests/test_serializer.py

@@ -641,17 +641,20 @@ def test_router_case_local_bgpcpe():
     assert 3 in d["interfaces"]["ifp-0/1/2"]["units"]
     assert 4 in d["interfaces"]["ifp-0/1/2"]["units"]
     assert 5 in d["interfaces"]["ifp-0/1/2"]["units"]
+    assert "ifp-0/1/3" in d["interfaces"]
+    assert 3 in d["interfaces"]["ifp-0/1/3"]["units"]
     assert "lo-0/0/0" in d["interfaces"]
-    assert len(d["interfaces"]) == 2
+    assert len(d["interfaces"]) == 3
 
     assert "protocols" in d["routing_instances"]["default"]
     assert "bgp" in d["routing_instances"]["default"]["protocols"]
 
     groups_default = d["routing_instances"]["default"]["protocols"]["bgp"]["groups"]
-    assert len(groups_default) == 1
+    assert len(groups_default) == 2
     assert (
         "CUST_cl390287" in groups_default
     )  #  parent interface has tobago line attached
+    assert groups_default["CUST_cl390287"]["any_as"] == True
     assert groups_default["CUST_cl390287"]["neighbors"][0]["interface"] == "ifp-0/1/2.3"
     assert groups_default["CUST_cl390287"]["family"]["ipv4_unicast"]["policy"][
         "export"
@@ -666,6 +669,12 @@ def test_router_case_local_bgpcpe():
         "import_list"
     ] == ["2a0e:b941:2:42::/64", "2a0e:b941:2::/122"]
 
+    assert (
+        "CPE_ifp-0-1-3-3_V4" in groups_default
+    )  # no tobago line attached, so no new naming
+    assert groups_default["CPE_ifp-0-1-3-3_V4"]["peer_as"] == 65086
+    assert groups_default["CPE_ifp-0-1-3-3_V4"]["neighbors"][0]["peer"] == "10.129.6.12"
+
     groups_L3VPN = d["routing_instances"]["L3VPN"]["protocols"]["bgp"]["groups"]
 
     assert "CPE_ifp-0-1-2-4" in groups_L3VPN  #  sub interface using legacy naming tag