diff --git a/cosmo/features.py b/cosmo/features.py index cb5fc18..403542c 100644 --- a/cosmo/features.py +++ b/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} +) diff --git a/cosmo/routerbgpcpevisitor.py b/cosmo/routerbgpcpevisitor.py index 8ed0ea9..b1662ac 100644 --- a/cosmo/routerbgpcpevisitor.py +++ b/cosmo/routerbgpcpevisitor.py @@ -1,5 +1,5 @@ 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 @@ -7,6 +7,7 @@ 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 ) diff --git a/cosmo/routervisitor.py b/cosmo/routervisitor.py index ae522b8..5ad0ce8 100644 --- a/cosmo/routervisitor.py +++ b/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 diff --git a/cosmo/tests/test_case_bgpcpe.yml b/cosmo/tests/test_case_bgpcpe.yml index dec4ce8..02c6546 100644 --- a/cosmo/tests/test_case_bgpcpe.yml +++ b/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: diff --git a/cosmo/tests/test_features.py b/cosmo/tests/test_features.py index f47722c..642fc82 100644 --- a/cosmo/tests/test_features.py +++ b/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}) diff --git a/cosmo/tests/test_serializer.py b/cosmo/tests/test_serializer.py index 71428ac..a713925 100644 --- a/cosmo/tests/test_serializer.py +++ b/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"]