change default naming for bgp cpe sessions

+ also keep old naming as a fallback when no tobago lines are attached
+ old naming still available using naming:cpe tag
+ test suite updated with all cases for new/old naming scheme switch

Author: loulecrivain

cosmo/netbox_types.py

@@ -15,7 +15,7 @@
     InterfaceSerializationError,
     head,
 )
-from typing import Self, Iterator, TypeVar, NoReturn, Never, Type
+from typing import Self, Iterator, TypeVar, NoReturn, Never, Type, Union
 
 from .tobago_types import TobagoAbstractTerminationType
 
@@ -540,9 +540,12 @@ def isLoopbackChild(self):
     def getConnectedEndpoints(self) -> list[DeviceType]:
         return self.get("connected_endpoints", [])
 
-    def hasAnAttachedTobagoLine(self):
+    def hasAnAttachedTobagoLine(self) -> bool:
         return self.get("attached_tobago_line") is not None
 
+    def getAttachedTobagoLine(self) -> Union["CosmoTobagoLine", None]:
+        return self.get("attached_tobago_line")
+
 
 class VLANType(AbstractNetboxType):
     def __repr__(self):

cosmo/routerbgpcpevisitor.py

@@ -1,5 +1,6 @@
 from functools import singledispatchmethod
 from ipaddress import IPv4Interface, IPv6Interface, ip_interface
+from typing import TypeGuard
 
 from cosmo.common import head, CosmoOutputType
 from cosmo.cperoutervisitor import CpeRouterExporterVisitor, CpeRouterIPVisitor
@@ -89,6 +90,29 @@ def processUnnumberedBGP(base_group_name, linked_interface, policy_v4, policy_v6
             }
         }
 
+    @staticmethod
+    def getGroupName(
+        linked_interface: InterfaceType, parent_interface: InterfaceType
+    ) -> str:
+        # technically a type guard, as we're narrowing on TagType. described
+        # as such to satisfy the type checker.
+        def tagFilter(t: TagType) -> TypeGuard[TagType]:
+            return t.getTagName() == "naming" and t.getTagValue() == "cpe"
+
+        attached_tobago_line = parent_interface.getAttachedTobagoLine()
+        # if legacy naming tag is present, or no tobago line is attached, we keep the old name as a fallback
+        if not attached_tobago_line or any(
+            filter(
+                tagFilter,
+                linked_interface.getTags(),
+            )
+        ):
+            return "CPE_" + linked_interface.getName().replace(".", "-").replace(
+                "/", "-"
+            )
+        else:  # use new naming scheme with tobago line name
+            return "CUST_" + attached_tobago_line.getLineNameLong()
+
     def processBgpCpeTag(self, o: TagType):
         linked_interface = o.getParent(InterfaceType)
         if not linked_interface.hasParentInterface():
@@ -112,9 +136,7 @@ def processBgpCpeTag(self, o: TagType):
             )
             return
 
-        group_name = "CPE_" + linked_interface.getName().replace(".", "-").replace(
-            "/", "-"
-        )
+        group_name = self.getGroupName(linked_interface, parent_interface)
         vrf_name = "default"
         # make the type checker happy, since it cannot reliably infer
         # type from default values of policy_v4 and policy_v6

cosmo/routervisitor.py

@@ -850,6 +850,8 @@ def _(self, o: TagType):
                 return self.processAccessTag(o)
             case "unnumbered":
                 return self.processBgpUnnumberedTag(o)
+            case "naming":
+                pass  # ignore, as it is treated in bgp cpe visitor
             case "bgp":
                 if o.getTagValue() == "cpe":
                     return self.bgpcpe_exporter.accept(o)

cosmo/tests/test_case_bgpcpe.yml

@@ -30,6 +30,67 @@ device_list:
     untagged_vlan: null
     vrf: null
     parent: null
+    attached_tobago_line:
+      __typename: CosmoTobagoLine
+      component_type: CABLE
+      element:
+        description: ''
+        display: cable 000128934
+        id: 39645
+        label: cable 000128934
+        url: https://netbox.example.com/api/dcim/cables/39645/
+      element_id: 93827
+      element_type: dcim.cable
+      id: 5110
+      index: 1
+      termination_a:
+        _occupied: true
+        cable:
+          description: ''
+          display: cable 000128934
+          id: 39645
+          label: cable 000128934
+          url: https://netbox.example.com/api/dcim/cables/39645/
+        description: ''
+        device:
+          description: ''
+          display: TEST0001
+          id: 17799
+          name: TEST0001
+          url: https://netbox.example.com/api/dcim/devices/1747/
+        display: ifp-0/1/2
+        id: 191940
+        name: ifp-0/1/2
+        url: https://netbox.example.com/api/dcim/interfaces/191940/
+      termination_a_id: 92387
+      termination_a_type: dcim.interface
+      termination_b:
+        _occupied: true
+        id: 198208
+        url: https://netbox.example.com/api/dcim/interfaces/192878/
+        display: DC10 duplex front 10b
+        device:
+          id: 39948
+          url: https://netbox.example.com/api/dcim/device/39948/
+          display: Panel C
+          name: Panel C
+          description: null
+        name: DC10 duplex front 10b
+        description: ""
+      termination_b_id: 454
+      termination_b_type: dcim.frontport
+      version:
+        created: '2025-09-03T11:34:19.643456+01:00'
+        custom_fields: { }
+        id: 2617
+        last_updated: '2025-09-03T11:34:40.819703+01:00'
+        line:
+          display: cl390287
+          id: 9834
+          name: '390287'
+          name_long: cl390287
+          url: https://netbox.example.com/api/plugins/tobago/lines/9834/
+        status: current
     connected_endpoints:
     - device:
         name: ''
@@ -114,6 +175,9 @@ device_list:
     - name: bgp:cpe
       slug: bgpcpe
       __typename: TagType
+    - name: naming:cpe
+      slug: namingcpe
+      __typename: TagType
     type: VIRTUAL
     untagged_vlan: null
     vrf:

cosmo/tests/test_serializer.py

@@ -475,28 +475,28 @@ def test_router_case_local_bgpcpe():
 
     groups_default = d["routing_instances"]["default"]["protocols"]["bgp"]["groups"]
     assert len(groups_default) == 1
-    assert "CPE_ifp-0-1-2-3" in groups_default
     assert (
-        groups_default["CPE_ifp-0-1-2-3"]["neighbors"][0]["interface"] == "ifp-0/1/2.3"
-    )
+        "CUST_cl390287" in groups_default
+    )  #  parent interface has tobago line attached
+    assert groups_default["CUST_cl390287"]["neighbors"][0]["interface"] == "ifp-0/1/2.3"
     assert (
-        groups_default["CPE_ifp-0-1-2-3"]["family"]["ipv4_unicast"]["policy"]["export"]
+        groups_default["CUST_cl390287"]["family"]["ipv4_unicast"]["policy"]["export"]
         == "DEFAULT_V4"
     )
     assert (
-        groups_default["CPE_ifp-0-1-2-3"]["family"]["ipv6_unicast"]["policy"]["export"]
+        groups_default["CUST_cl390287"]["family"]["ipv6_unicast"]["policy"]["export"]
         == "DEFAULT_V6"
     )
-    assert groups_default["CPE_ifp-0-1-2-3"]["family"]["ipv4_unicast"]["policy"][
+    assert groups_default["CUST_cl390287"]["family"]["ipv4_unicast"]["policy"][
         "import_list"
     ] == ["10.1.0.0/28"]
-    assert groups_default["CPE_ifp-0-1-2-3"]["family"]["ipv6_unicast"]["policy"][
+    assert groups_default["CUST_cl390287"]["family"]["ipv6_unicast"]["policy"][
         "import_list"
     ] == ["2a0e:b941:2:42::/64", "2a0e:b941:2::/122"]
 
     groups_L3VPN = d["routing_instances"]["L3VPN"]["protocols"]["bgp"]["groups"]
 
-    assert "CPE_ifp-0-1-2-4" in groups_L3VPN
+    assert "CPE_ifp-0-1-2-4" in groups_L3VPN  #  sub interface using legacy naming tag
     assert groups_L3VPN["CPE_ifp-0-1-2-4"]["neighbors"][0]["interface"] == "ifp-0/1/2.4"
     assert (
         not "export"
@@ -513,25 +513,23 @@ def test_router_case_local_bgpcpe():
         "import_list"
     ] == ["2a0e:b941:2:42::/64", "2a0e:b941:2::/122"]
 
-    assert "CPE_ifp-0-1-2-5_V4" in groups_L3VPN
-    assert "CPE_ifp-0-1-2-5_V6" in groups_L3VPN
-    assert groups_L3VPN["CPE_ifp-0-1-2-5_V4"]["neighbors"][0]["peer"] == "10.128.6.12"
-    assert (
-        groups_L3VPN["CPE_ifp-0-1-2-5_V6"]["neighbors"][0]["peer"] == "2a0e:b941:2::21"
-    )
+    assert "CUST_cl390287_V4" in groups_L3VPN
+    assert "CUST_cl390287_V6" in groups_L3VPN
+    assert groups_L3VPN["CUST_cl390287_V4"]["neighbors"][0]["peer"] == "10.128.6.12"
+    assert groups_L3VPN["CUST_cl390287_V6"]["neighbors"][0]["peer"] == "2a0e:b941:2::21"
     assert (
         not "export"
-        in groups_L3VPN["CPE_ifp-0-1-2-5_V4"]["family"]["ipv4_unicast"]["policy"]
+        in groups_L3VPN["CUST_cl390287_V4"]["family"]["ipv4_unicast"]["policy"]
     )
     assert (
         not "export"
-        in groups_L3VPN["CPE_ifp-0-1-2-5_V6"]["family"]["ipv6_unicast"]["policy"]
+        in groups_L3VPN["CUST_cl390287_V6"]["family"]["ipv6_unicast"]["policy"]
     )
-    assert groups_L3VPN["CPE_ifp-0-1-2-5_V4"]["family"]["ipv4_unicast"]["policy"][
+    assert groups_L3VPN["CUST_cl390287_V4"]["family"]["ipv4_unicast"]["policy"][
         "import_list"
     ] == ["10.1.0.0/28"]
     # should not be allowed to announce our transfer nets, so '2a0e:b941:2::/122' should not be there
-    assert groups_L3VPN["CPE_ifp-0-1-2-5_V6"]["family"]["ipv6_unicast"]["policy"][
+    assert groups_L3VPN["CUST_cl390287_V6"]["family"]["ipv6_unicast"]["policy"][
         "import_list"
     ] == ["2a0e:b941:2:42::/64"]