diff --git a/changes/414.changed b/changes/414.changed new file mode 100644 index 000000000..5b15e8fb6 --- /dev/null +++ b/changes/414.changed @@ -0,0 +1 @@ +Changed IPFabric interface media matching to fall back on interface names. \ No newline at end of file diff --git a/nautobot_ssot/integrations/ipfabric/diffsync/adapter_ipfabric.py b/nautobot_ssot/integrations/ipfabric/diffsync/adapter_ipfabric.py index 996d93684..5e14973fe 100644 --- a/nautobot_ssot/integrations/ipfabric/diffsync/adapter_ipfabric.py +++ b/nautobot_ssot/integrations/ipfabric/diffsync/adapter_ipfabric.py @@ -96,7 +96,7 @@ def load_device_interfaces(self, device_model, interfaces, device_primary_ip, ne if iface.get("mac") else DEFAULT_INTERFACE_MAC, mtu=iface.get("mtu") if iface.get("mtu") else DEFAULT_INTERFACE_MTU, - type=ipfabric_utils.convert_media_type(iface.get("media") or ""), + type=ipfabric_utils.convert_media_type(iface.get("media"), iface_name), mgmt_only=iface.get("mgmt_only", False), ip_address=ip_address, subnet_mask=subnet_mask, diff --git a/nautobot_ssot/integrations/ipfabric/utilities/utils.py b/nautobot_ssot/integrations/ipfabric/utilities/utils.py index fad946197..7c601ffcc 100644 --- a/nautobot_ssot/integrations/ipfabric/utilities/utils.py +++ b/nautobot_ssot/integrations/ipfabric/utilities/utils.py @@ -1,4 +1,6 @@ """General utils for IPFabric.""" +import re + from nautobot_ssot.integrations.ipfabric.constants import DEFAULT_INTERFACE_TYPE @@ -37,99 +39,122 @@ def convert_media_type( # pylint: disable=too-many-return-statements,too-many-branches,too-many-statements media_type: str, + interface_name: str, ) -> str: """Convert provided `media_type` to value used by Nautobot. Args: media_type: The media type of an inteface (i.e. SFP-10GBase) + interface_name: The name of the interface with `media_type`. Returns: str: The corresponding represention of `media_type` in Nautobot. """ - media_type = media_type.lower().replace("-", "") - if VIRTUAL in media_type: - return VIRTUAL - if BRIDGE in media_type: - return BRIDGE - if LAG in media_type: - return LAG - - if TEN_GIG_BASE_T in media_type: - return "10gbase-t" - - # Going from 10Gig to lower bandwidths to allow media that supports multiple - # bandwidths to use highest supported bandwidth - if TEN_GIG in media_type: - nautobot_media_type = "10gbase-x-" - if XFP in media_type: - nautobot_media_type += "xfp" - elif X2 in media_type: - nautobot_media_type += "x2" - elif XENPAK in media_type: - nautobot_media_type += "xenpak" - else: - nautobot_media_type += "sfpp" - return nautobot_media_type - - # Flipping order of 5gig and 2.5g as both use the string 5gbase - if TWO_AND_HALF_GIG_BASE in media_type: - return "2.5gbase-t" - - if FIVE_GIG_BASE in media_type: - return "5gbase-t" - - if GIG_BASE in media_type or RJ45 in media_type or GIG in media_type: - nautobot_media_type = "1000base-" - if GBIC in media_type: - nautobot_media_type += "x-gbic" - elif SFP in media_type or SR in media_type or LR in media_type or SX in media_type or LX in media_type: - nautobot_media_type += "x-sfp" - else: - nautobot_media_type += "t" - return nautobot_media_type - - if HUNDRED_MEG_BASE in media_type or HUNDRED_MEG in media_type: - return "100base-tx" - - if TWENTY_FIVE_GIG in media_type: - return "25gbase-x-sfp28" - - if FORTY_GIG in media_type: - return "40gbase-x-qsfpp" - - if FIFTY_GIG in media_type: - return "50gbase-x-sfp56" - - if HUNDRED_GIG in media_type: - nautobot_media_type = "100gbase-x-" - if QSFP in media_type: - nautobot_media_type += "qsfp28" - else: - nautobot_media_type += "cfp" - return nautobot_media_type - - if TWO_HUNDRED_GIG in media_type: - nautobot_media_type = "200gbase-x-" - if QSFP in media_type: - nautobot_media_type += "qsfp56" - else: - nautobot_media_type += "cfp2" - return nautobot_media_type - - if FOUR_HUNDRED_GIG in media_type: - nautobot_media_type = "400gbase-x-" - if QSFP in media_type: - nautobot_media_type += "qsfp112" - else: - nautobot_media_type += "osfp" - return nautobot_media_type - - if EIGHT_HUNDRED_GIG in media_type: - nautobot_media_type = "800gbase-x-" - if QSFP in media_type: - nautobot_media_type += "qsfpdd" - else: - nautobot_media_type += "osfp" - return nautobot_media_type + if media_type: + media_type = media_type.lower().replace("-", "") + if VIRTUAL in media_type: + return VIRTUAL + if BRIDGE in media_type: + return BRIDGE + if LAG in media_type: + return LAG + + if TEN_GIG_BASE_T in media_type: + return "10gbase-t" + + # Going from 10Gig to lower bandwidths to allow media that supports multiple + # bandwidths to use highest supported bandwidth + if TEN_GIG in media_type: + nautobot_media_type = "10gbase-x-" + if XFP in media_type: + nautobot_media_type += "xfp" + elif X2 in media_type: + nautobot_media_type += "x2" + elif XENPAK in media_type: + nautobot_media_type += "xenpak" + else: + nautobot_media_type += "sfpp" + return nautobot_media_type + + # Flipping order of 5gig and 2.5g as both use the string 5gbase + if TWO_AND_HALF_GIG_BASE in media_type: + return "2.5gbase-t" + + if FIVE_GIG_BASE in media_type: + return "5gbase-t" + + if GIG_BASE in media_type or RJ45 in media_type or GIG in media_type: + nautobot_media_type = "1000base-" + if GBIC in media_type: + nautobot_media_type += "x-gbic" + elif SFP in media_type or SR in media_type or LR in media_type or SX in media_type or LX in media_type: + nautobot_media_type += "x-sfp" + else: + nautobot_media_type += "t" + return nautobot_media_type + + if HUNDRED_MEG_BASE in media_type or HUNDRED_MEG in media_type: + return "100base-tx" + + if TWENTY_FIVE_GIG in media_type: + return "25gbase-x-sfp28" + + if FORTY_GIG in media_type: + return "40gbase-x-qsfpp" + + if FIFTY_GIG in media_type: + return "50gbase-x-sfp56" + + if HUNDRED_GIG in media_type: + nautobot_media_type = "100gbase-x-" + if QSFP in media_type: + nautobot_media_type += "qsfp28" + else: + nautobot_media_type += "cfp" + return nautobot_media_type + + if TWO_HUNDRED_GIG in media_type: + nautobot_media_type = "200gbase-x-" + if QSFP in media_type: + nautobot_media_type += "qsfp56" + else: + nautobot_media_type += "cfp2" + return nautobot_media_type + + if FOUR_HUNDRED_GIG in media_type: + nautobot_media_type = "400gbase-x-" + if QSFP in media_type: + nautobot_media_type += "qsfp112" + else: + nautobot_media_type += "osfp" + return nautobot_media_type + + if EIGHT_HUNDRED_GIG in media_type: + nautobot_media_type = "800gbase-x-" + if QSFP in media_type: + nautobot_media_type += "qsfpdd" + else: + nautobot_media_type += "osfp" + return nautobot_media_type + else: + interface_name = interface_name.lower() + regex_to_type = ( + (r"po(rt-?channel)?\d", "lag"), + (r"vl(an)?\d", "virtual"), + (r"lo(opback)?\d", "virtual"), + (r"tu(nnel)?\d", "virtual"), + (r"vx(lan)?\d", "virtual"), + (r"fa(stethernet)?\d", "100base-tx"), + (r"gi(gabitethernet)?\d", "1000base-t"), + (r"te(ngigabitethernet)?\d", "10gbase-x-sfpp"), + (r"twentyfivegigabitethernet\d", "25gbase-x-sfp28"), + (r"fo(rtygigabitethernet)?\d", "40gbase-x-qsfpp"), + (r"fi(ftygigabitethernet)?\d", "50gbase-x-sfp56"), + (r"hu(ndredgigabitethernet)?\d", "100gbase-x-qsfp28"), + (r"twohundredgigabitethernet\d", "200gbase-x-qsfp56"), + ) + for regex, iface_type in regex_to_type: + if re.match(regex, interface_name): + return iface_type return DEFAULT_INTERFACE_TYPE diff --git a/nautobot_ssot/tests/ipfabric/test_utils.py b/nautobot_ssot/tests/ipfabric/test_utils.py index 1d34c3032..c0f4e8352 100644 --- a/nautobot_ssot/tests/ipfabric/test_utils.py +++ b/nautobot_ssot/tests/ipfabric/test_utils.py @@ -9,95 +9,145 @@ class TestUtils(SimpleTestCase): # pylint: disable=too-many-public-methods """Test IPFabric utilities.utils.""" def test_virtual_interface(self): - self.assertEqual("virtual", utils.convert_media_type("Virtual")) + self.assertEqual("virtual", utils.convert_media_type("Virtual", "VLAN1")) def test_bridge_interface(self): - self.assertEqual("bridge", utils.convert_media_type("Bridge")) + self.assertEqual("bridge", utils.convert_media_type("Bridge", "Bridge0")) def test_lag_interface(self): - self.assertEqual("lag", utils.convert_media_type("LAG")) + self.assertEqual("lag", utils.convert_media_type("LAG", "Po1")) def test_hunderd_meg_base_t_interface(self): - self.assertEqual("100base-tx", utils.convert_media_type("100Base-T")) + self.assertEqual("100base-tx", utils.convert_media_type("100Base-T", "Fa1")) def test_hundred_meg_interface(self): - self.assertEqual("100base-tx", utils.convert_media_type("100MegabitEthernet")) + self.assertEqual("100base-tx", utils.convert_media_type("100MegabitEthernet", "Fa1")) def test_gig_base_t_interface(self): - self.assertEqual("1000base-t", utils.convert_media_type("1000BaseT")) - self.assertEqual("1000base-t", utils.convert_media_type("10/100/1000BaseTX")) + self.assertEqual("1000base-t", utils.convert_media_type("1000BaseT", "Gi1")) + self.assertEqual("1000base-t", utils.convert_media_type("10/100/1000BaseTX", "Gi1")) def test_rj45_uses_gig_base_t_interface(self): - self.assertEqual("1000base-t", utils.convert_media_type("RJ45")) + self.assertEqual("1000base-t", utils.convert_media_type("RJ45", "Gi1")) def test_gig_default_uses_base_t_interface(self): - self.assertEqual("1000base-t", utils.convert_media_type("1GigThisUsesDefault")) + self.assertEqual("1000base-t", utils.convert_media_type("1GigThisUsesDefault", "Gi1")) def test_gig_sfp_interface(self): - self.assertEqual("1000base-x-sfp", utils.convert_media_type("10/100/1000BaseTX SFP")) + self.assertEqual("1000base-x-sfp", utils.convert_media_type("10/100/1000BaseTX SFP", "Gi1")) def test_gig_sfp_used_for_sfp_type_interface(self): - self.assertEqual("1000base-x-sfp", utils.convert_media_type("1000BaseLX")) - self.assertEqual("1000base-x-sfp", utils.convert_media_type("1000BaseSX")) - self.assertEqual("1000base-x-sfp", utils.convert_media_type("1000BaseLR")) - self.assertEqual("1000base-x-sfp", utils.convert_media_type("1000BaseSR")) + self.assertEqual("1000base-x-sfp", utils.convert_media_type("1000BaseLX", "Gi1")) + self.assertEqual("1000base-x-sfp", utils.convert_media_type("1000BaseSX", "Gi1")) + self.assertEqual("1000base-x-sfp", utils.convert_media_type("1000BaseLR", "Gi1")) + self.assertEqual("1000base-x-sfp", utils.convert_media_type("1000BaseSR", "Gi1")) def test_gig_gbic_interface(self): - self.assertEqual("1000base-x-gbic", utils.convert_media_type("10/100/1000BaseTX GBIC")) + self.assertEqual("1000base-x-gbic", utils.convert_media_type("10/100/1000BaseTX GBIC", "Gi1")) def test_two_and_half_gig_base_t_interface(self): - self.assertEqual("2.5gbase-t", utils.convert_media_type("100/1000/2.5GBaseTX")) + self.assertEqual("2.5gbase-t", utils.convert_media_type("100/1000/2.5GBaseTX", "TwoGi1")) def test_five_gig_base_t_interface(self): - self.assertEqual("5gbase-t", utils.convert_media_type("100/1000/2.5G/5GBaseTX")) + self.assertEqual("5gbase-t", utils.convert_media_type("100/1000/2.5G/5GBaseTX", "FiveGi1")) def test_ten_gig_xfp_interface(self): - self.assertEqual("10gbase-x-xfp", utils.convert_media_type("10GBase XFP")) + self.assertEqual("10gbase-x-xfp", utils.convert_media_type("10GBase XFP", "TenGi1")) def test_ten_gig_x2_interface(self): - self.assertEqual("10gbase-x-x2", utils.convert_media_type("10GBase X2")) + self.assertEqual("10gbase-x-x2", utils.convert_media_type("10GBase X2", "TenGi1")) def test_ten_gig_xenpak_interface(self): - self.assertEqual("10gbase-x-xenpak", utils.convert_media_type("10GBase XENPAK")) + self.assertEqual("10gbase-x-xenpak", utils.convert_media_type("10GBase XENPAK", "TenGi1")) def test_ten_gig_sfp_interface(self): - self.assertEqual("10gbase-x-sfpp", utils.convert_media_type("10GBase SFP")) + self.assertEqual("10gbase-x-sfpp", utils.convert_media_type("10GBase SFP", "TenGi1")) def test_ten_gig_default_uses_sfp_interface(self): - self.assertEqual("10gbase-x-sfpp", utils.convert_media_type("10G")) + self.assertEqual("10gbase-x-sfpp", utils.convert_media_type("10G", "TenGi1")) def test_twenty_five_gig_sfp_interface(self): - self.assertEqual("25gbase-x-sfp28", utils.convert_media_type("25G")) + self.assertEqual("25gbase-x-sfp28", utils.convert_media_type("25G", "TweGi1")) def test_forty_gig_sfp_interface(self): - self.assertEqual("40gbase-x-qsfpp", utils.convert_media_type("40G")) + self.assertEqual("40gbase-x-qsfpp", utils.convert_media_type("40G", "FoGi1")) def test_fifty_gig_sfp_interface(self): - self.assertEqual("50gbase-x-sfp56", utils.convert_media_type("50G")) + self.assertEqual("50gbase-x-sfp56", utils.convert_media_type("50G", "FiGi1")) def test_hundred_gig_qsfp_interface(self): - self.assertEqual("100gbase-x-qsfp28", utils.convert_media_type("100G QSFP")) + self.assertEqual("100gbase-x-qsfp28", utils.convert_media_type("100G QSFP", "HunGi1")) def test_hundred_gig_default_uses_cfp_interface(self): - self.assertEqual("100gbase-x-cfp", utils.convert_media_type("100G")) + self.assertEqual("100gbase-x-cfp", utils.convert_media_type("100G", "HunGi1")) def test_two_hundred_gig_qsfp_interface(self): - self.assertEqual("200gbase-x-qsfp56", utils.convert_media_type("200G QSFP")) + self.assertEqual("200gbase-x-qsfp56", utils.convert_media_type("200G QSFP", "TwoHunGi1")) def test_two_hundred_gig_default_uses_cfp_interface(self): - self.assertEqual("200gbase-x-cfp2", utils.convert_media_type("200G")) + self.assertEqual("200gbase-x-cfp2", utils.convert_media_type("200G", "TwoHunGi1")) def test_four_hundred_gig_qsfp_interface(self): - self.assertEqual("400gbase-x-qsfp112", utils.convert_media_type("400G QSFP")) + self.assertEqual("400gbase-x-qsfp112", utils.convert_media_type("400G QSFP", "FoHunGi1")) def test_four_hundred_gig_default_uses_osfp_interface(self): - self.assertEqual("400gbase-x-osfp", utils.convert_media_type("400G")) + self.assertEqual("400gbase-x-osfp", utils.convert_media_type("400G", "FoHunGi1")) def test_eight_hundred_gig_qsfp_interface(self): - self.assertEqual("800gbase-x-qsfpdd", utils.convert_media_type("800G QSFP")) + self.assertEqual("800gbase-x-qsfpdd", utils.convert_media_type("800G QSFP", "EiHunGi1")) def test_eight_hundred_gig_default_uses_osfp_interface(self): - self.assertEqual("800gbase-x-osfp", utils.convert_media_type("800G")) + self.assertEqual("800gbase-x-osfp", utils.convert_media_type("800G", "EiHunGi1")) def test_unknown_interface_uses_default_interface(self): - self.assertEqual(DEFAULT_INTERFACE_TYPE, utils.convert_media_type("ThisShouldGiveTheDefault")) + self.assertEqual(DEFAULT_INTERFACE_TYPE, utils.convert_media_type("ThisShouldGiveTheDefault", "")) + + def test_interface_name_lag(self): + self.assertEqual("lag", utils.convert_media_type("", "Po1")) + self.assertEqual("lag", utils.convert_media_type("", "Port-channel1")) + + def test_interface_name_vlan(self): + self.assertEqual("virtual", utils.convert_media_type("", "Vlan1")) + self.assertEqual("virtual", utils.convert_media_type("", "Vl1")) + + def test_interface_name_loopback(self): + self.assertEqual("virtual", utils.convert_media_type("", "Loopback1")) + self.assertEqual("virtual", utils.convert_media_type("", "Lo1")) + + def test_interface_name_tunnel(self): + self.assertEqual("virtual", utils.convert_media_type("", "Tu1")) + self.assertEqual("virtual", utils.convert_media_type("", "Tunnel1")) + + def test_interface_name_vxlan(self): + self.assertEqual("virtual", utils.convert_media_type("", "Vxlan1")) + self.assertEqual("virtual", utils.convert_media_type("", "Vx1")) + + def test_interface_name_fastethernet(self): + self.assertEqual("100base-tx", utils.convert_media_type("", "FastEthernet1")) + self.assertEqual("100base-tx", utils.convert_media_type("", "Fa1")) + + def test_interface_name_gigethernet(self): + self.assertEqual("1000base-t", utils.convert_media_type("", "GigabitEthernet1")) + self.assertEqual("1000base-t", utils.convert_media_type("", "Gi1")) + + def test_interface_name_tengigethernet(self): + self.assertEqual("10gbase-x-sfpp", utils.convert_media_type("", "TenGigabitEthernet1")) + self.assertEqual("10gbase-x-sfpp", utils.convert_media_type("", "Te1")) + + def test_interface_name_twentyfivegigethernet(self): + self.assertEqual("25gbase-x-sfp28", utils.convert_media_type("", "TwentyFiveGigabitEthernet1")) + + def test_interface_name_fortygigethernet(self): + self.assertEqual("40gbase-x-qsfpp", utils.convert_media_type("", "FortyGigabitEthernet1")) + self.assertEqual("40gbase-x-qsfpp", utils.convert_media_type("", "Fo1")) + + def test_interface_name_fiftygigethernet(self): + self.assertEqual("50gbase-x-sfp56", utils.convert_media_type("", "FiftyGigabitEthernet1")) + self.assertEqual("50gbase-x-sfp56", utils.convert_media_type("", "Fi1")) + + def test_interface_name_hundredgigethernet(self): + self.assertEqual("100gbase-x-qsfp28", utils.convert_media_type("", "HundredGigabitEthernet1")) + self.assertEqual("100gbase-x-qsfp28", utils.convert_media_type("", "Hu1")) + + def test_interface_name_twohundredgigethernet(self): + self.assertEqual("200gbase-x-qsfp56", utils.convert_media_type("", "TwoHundredGigabitEthernet1"))