From 060ba43d1475f2bc4e23efde5902d6a27978783d Mon Sep 17 00:00:00 2001 From: lou lecrivain Date: Wed, 25 Mar 2026 17:08:06 +0100 Subject: [PATCH] fixes for netbox stable way of registering manufacturer / platforms (our fork had different constraints, but now we switched to netbox stable) list of changes: - extend manufacturer slug match logic - fix manufacturer tests --- cosmo/clients/queries/device.graphql | 4 +++ cosmo/manufacturers.py | 49 ++++++++++++++++++++-------- cosmo/netbox_types.py | 5 ++- cosmo/tests/test_serializer.py | 8 +++-- 4 files changed, 50 insertions(+), 16 deletions(-) diff --git a/cosmo/clients/queries/device.graphql b/cosmo/clients/queries/device.graphql index d41d861..24c1c66 100644 --- a/cosmo/clients/queries/device.graphql +++ b/cosmo/clients/queries/device.graphql @@ -9,6 +9,10 @@ query { device_type { __typename + manufacturer { + __typename + slug + } slug } platform { diff --git a/cosmo/manufacturers.py b/cosmo/manufacturers.py index 2496e6c..003023b 100644 --- a/cosmo/manufacturers.py +++ b/cosmo/manufacturers.py @@ -7,7 +7,13 @@ from cosmo.common import DeviceSerializationError -from cosmo.netbox_types import DeviceType, InterfaceType, PlatformType, VRFType +from cosmo.netbox_types import ( + DeviceType, + InterfaceType, + PlatformType, + VRFType, + DeviceTypeType, +) class AbstractManufacturer(ABC): @@ -16,22 +22,39 @@ def __init__(self, cosmo_config: "CosmoConfig"): @classmethod def isCompatibleWith(cls, device: DeviceType): - # Note: If the platform cannot be parsed, getPlatform will be a string. + if not isinstance(device.getDeviceType(), DeviceTypeType): + return False + if not isinstance(device.getPlatform(), PlatformType): return False + device_platform_match = False if device.getPlatform().getManufacturer(): - return ( + device_platform_match = ( device.getPlatform().getManufacturer().getSlug() - == cls.myManufacturerSlug() + in cls.myManufacturerSlugs() + ) + else: + device_platform_match = bool( + re.match(cls.myPlatformRE(), device.getPlatform().getSlug()) + ) + + device_manufacturer_match = False + if device.getDeviceType().getManufacturer(): + device_manufacturer_match = ( + device.getDeviceType().getManufacturer().getSlug() + in cls.myManufacturerSlugs() ) else: - # fallback in case no manufacturer is filled in for the platform - return re.match(cls.myPlatformRE(), device.getPlatform().getSlug()) + device_manufacturer_match = bool( + re.match(cls.myPlatformRE(), device.getDeviceType().getSlug()) + ) + + return device_platform_match or device_manufacturer_match @staticmethod @abstractmethod - def myManufacturerSlug(): + def myManufacturerSlugs() -> list[str]: pass @classmethod @@ -110,8 +133,8 @@ class JuniperManufacturer(AbstractJuniperRtBrickManufacturerCommon): _platform_re = re.compile(r"REPLACEME") @staticmethod - def myManufacturerSlug(): - return "juniper" + def myManufacturerSlugs(): + return ["juniper"] @classmethod def myPlatformRE(cls): @@ -138,8 +161,8 @@ class RtBrickManufacturer(AbstractJuniperRtBrickManufacturerCommon): _platform_re = re.compile(r"REPLACEME") @staticmethod - def myManufacturerSlug(): - return "rtbrick" + def myManufacturerSlugs(): + return ["rtbrick", "ufispace", "edgecore"] @classmethod def myPlatformRE(cls): @@ -167,8 +190,8 @@ class CumulusNetworksManufacturer(AbstractManufacturer): _platform_re = re.compile(r"^cumulus-linux[a-zA-Z0-9-]*") @staticmethod - def myManufacturerSlug(): - return "cumulus-networks" + def myManufacturerSlugs(): + return ["cumulus-networks"] @classmethod def myPlatformRE(cls): diff --git a/cosmo/netbox_types.py b/cosmo/netbox_types.py index 8f19efa..29550b4 100644 --- a/cosmo/netbox_types.py +++ b/cosmo/netbox_types.py @@ -301,13 +301,16 @@ class DeviceTypeType(AbstractNetboxType): def getBasePath(self): return "/dcim/device-types/" + def getManufacturer(self): + return self.get("manufacturer") + class PlatformType(AbstractNetboxType): def getBasePath(self): return "/dcim/platforms/" def getManufacturer(self): - return self["manufacturer"] + return self.get("manufacturer") class ManufacturerType(AbstractNetboxType): diff --git a/cosmo/tests/test_serializer.py b/cosmo/tests/test_serializer.py index 9c6f97a..d2798d2 100644 --- a/cosmo/tests/test_serializer.py +++ b/cosmo/tests/test_serializer.py @@ -118,7 +118,7 @@ def test_router_platforms(mock_cosmo_config_fixture, mock_global_vrf, mock_l3vpn DeviceType(juniper_s.device), mock_cosmo_config_fixture ).get() assert juniper_manufacturer.getManagementVRFName() == "mgmt_junos" - assert juniper_manufacturer.myManufacturerSlug() == "juniper" + assert juniper_manufacturer.myManufacturerSlugs() == ["juniper"] assert ( juniper_manufacturer.spitVRFPathWith(mock_global_vrf, {}) == juniper_manufacturer._spitDefaultVRFPathWith({}) @@ -146,7 +146,11 @@ def test_router_platforms(mock_cosmo_config_fixture, mock_global_vrf, mock_l3vpn DeviceType(rtbrick_s.device), mock_cosmo_config_fixture ).get() assert rtbrick_manufacturer.getManagementVRFName() == "mgmt" - assert rtbrick_manufacturer.myManufacturerSlug() == "rtbrick" + assert rtbrick_manufacturer.myManufacturerSlugs() == [ + "rtbrick", + "ufispace", + "edgecore", + ] assert ( rtbrick_manufacturer.spitVRFPathWith(mock_global_vrf, {}) == rtbrick_manufacturer._spitDefaultVRFPathWith({})