NE-1007: change naming of auto bgp sessions

cosmo/features.py

@@ -1,7 +1,7 @@
 # implementation guide
 # https://martinfowler.com/articles/feature-toggles.html
 from argparse import Action, ArgumentParser
-from typing import Never, Self, Optional, TextIO, Sequence, Any
+from typing import Never, Self, Optional, TextIO, Sequence, Any, Callable
 
 import yaml
 
@@ -83,4 +83,19 @@ def __str__(self):
         return ", ".join(features_desc)
 
 
-features = FeatureToggle({"interface-auto-descriptions": True})
+def with_feature(instance: FeatureToggle, feature_name: str):
+    def decorator_with_feature(func: Callable):
+        def exe_with_feature(*args, **kwargs):
+            previous_state = instance.featureIsEnabled(feature_name)
+            instance.setFeature(feature_name, True)
+            func(*args, **kwargs)
+            instance.setFeature(feature_name, previous_state)
+
+        return exe_with_feature
+
+    return decorator_with_feature
+
+
+features = FeatureToggle(
+    {"interface-auto-descriptions": True, "new-bgp-cpe-group-naming": False}
+)

cosmo/routerbgpcpevisitor.py

@@ -1,12 +1,13 @@
 from abc import ABCMeta, abstractmethod
-from typing import List, NoReturn
+from typing import List, NoReturn, TypeGuard
 
 from multimethod import multimethod as singledispatchmethod
 from ipaddress import IPv4Interface, IPv6Interface
 
 from cosmo.common import head, CosmoOutputType, InterfaceSerializationError
 from cosmo.cperoutervisitor import CpeRouterExporterVisitor, CpeRouterIPVisitor
 from cosmo.abstractroutervisitor import AbstractRouterExporterVisitor
+from cosmo.features import features
 from cosmo.log import warn
 from cosmo.netbox_types import (
     TagType,
@@ -119,6 +120,33 @@ def acceptVRFNameOrFailOn(
     ) -> None | NoReturn:
         pass
 
+    @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() == "deprecated_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 features.featureIsEnabled("new-bgp-cpe-group-naming")
+            or 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():
@@ -142,9 +170,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 = self._default_vrf_name
         # make the type checker happy, since it cannot reliably infer
         # type from default values of policy_v4 and policy_v6
@@ -178,7 +204,6 @@ def processBgpCpeTag(self, o: TagType):
             elif af and af is IPv6Interface:
                 v6_import.add(prefix)
 
-        # import pdb; pdb.set_trace()
         policy_v4["import_list"], policy_v6["import_list"] = self.processImportLists(
             v4_import, v6_import
         )

cosmo/routervisitor.py

@@ -815,6 +815,8 @@ def _(self, o: TagType):
                 return self.processAccessTag(o)
             case "unnumbered":
                 return self.processBgpUnnumberedTag(o)
+            case "deprecated_naming":
+                pass  # ignore, as it is treated in bgp cpe visitor
             case "bgp":
                 if o.getTagValue() == "cpe":
                     pass  # ignore, treated with whole tag list

cosmo/tests/test_case_bgpcpe.yml

@@ -30,6 +30,78 @@ 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'
+        tenant:
+          id: 43876
+          url: https://netbox.example.com/api/plugins/tobago/tenants/43876/
+          display: Contoso Ltd.
+          name: Contoso Ltd.
+          slug: contoso-ltd
+        line:
+          display: cl390287
+          id: 9834
+          name: '390287'
+          name_long: cl390287
+          url: https://netbox.example.com/api/plugins/tobago/lines/9834/
+          service:
+            id: 4893794
+            url: https://netbox.example.com/api/plugins/tobago/services/4893794/
+            display: "#9823 (cl390287)"
+            business_service: null
+        status: current
     connected_endpoints:
     - name: "xe-0/1/2.3"
       __typename: InterfaceType
@@ -117,6 +189,9 @@ device_list:
     - name: bgp:cpe
       slug: bgpcpe
       __typename: TagType
+    - name: deprecated_naming:cpe
+      slug: deprecatednamingcpe
+      __typename: TagType
     type: VIRTUAL
     untagged_vlan: null
     vrf:
@@ -210,6 +285,9 @@ device_list:
     - name: max-prefixes:100
       slug: maxprefixes100
       __typename: TagType
+    - name: deprecated_naming:cpe
+      slug: deprecatednamingcpe
+      __typename: TagType
     type: VIRTUAL
     untagged_vlan: null
     vrf:
@@ -254,6 +332,9 @@ device_list:
     - name: max-prefixes:100
       slug: maxprefixes100
       __typename: TagType
+    - name: deprecated_naming:cpe
+      slug: deprecatednamingcpe
+      __typename: TagType
     type: VIRTUAL
     untagged_vlan: null
     vrf:

cosmo/tests/test_features.py

@@ -3,7 +3,11 @@
 
 import pytest
 
-from cosmo.features import NonExistingFeatureToggleException, FeatureToggle
+from cosmo.features import (
+    NonExistingFeatureToggleException,
+    FeatureToggle,
+    with_feature,
+)
 
 
 def test_set_get():
@@ -57,6 +61,17 @@ def test_non_existing_features():
         ft.setFeature("i-do-not-exist", True)
 
 
+def test_with_feature_decorator():
+    ft = FeatureToggle({"feature_a": False})
+
+    @with_feature(ft, "feature_a")
+    def execute_with_decorator():
+        assert ft.featureIsEnabled("feature_a")
+
+    execute_with_decorator()
+    assert not ft.featureIsEnabled("feature_a")
+
+
 def test_argparse_integration():
     ft = FeatureToggle({"feature_a": False, "feature_b": False, "feature_c": True})
 

cosmo/tests/test_serializer.py

@@ -5,6 +5,7 @@
 import copy
 
 from cosmo.common import DeviceSerializationError
+from cosmo.features import with_feature, features
 from cosmo.manufacturers import ManufacturerFactoryFromDevice
 from cosmo.netbox_types import DeviceType
 
@@ -516,6 +517,7 @@ def test_router_case_local_l3vpn():
     assert ri["routing_options"] == {}
 
 
+@with_feature(features, "new-bgp-cpe-group-naming")
 def test_router_case_local_bgpcpe():
     [d] = get_router_sd_from_path("./test_case_bgpcpe.yml")
 
@@ -532,26 +534,26 @@ 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"
-    )
-    assert groups_default["CPE_ifp-0-1-2-3"]["family"]["ipv4_unicast"]["policy"][
+        "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["CUST_cl390287"]["family"]["ipv4_unicast"]["policy"][
         "export"
     ] == ["DEFAULT_V4"]
-    assert groups_default["CPE_ifp-0-1-2-3"]["family"]["ipv6_unicast"]["policy"][
+    assert 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"
@@ -568,25 +570,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"]