From 180ccec2ea6bd0c731b96b0a69420277b44c9d7c Mon Sep 17 00:00:00 2001 From: Sergey Sannikov Date: Thu, 10 Sep 2026 02:56:34 +0400 Subject: [PATCH] fix: assign addresses with a non-unique role to every interface (#344) --- module/sources/common/source_base.py | 55 ++++++++++-- tests/test_ip_non_unique_roles.py | 125 +++++++++++++++++++++++++++ 2 files changed, 171 insertions(+), 9 deletions(-) create mode 100644 tests/test_ip_non_unique_roles.py diff --git a/module/sources/common/source_base.py b/module/sources/common/source_base.py index b81b1b8..f7f2bb6 100644 --- a/module/sources/common/source_base.py +++ b/module/sources/common/source_base.py @@ -42,10 +42,37 @@ def implements(cls, source_type): return False + # NetBox exempts addresses with these roles from its uniqueness check, they are + # meant to exist on more than one interface at the same time + non_unique_ip_roles = ("anycast", "vip", "vrrp", "hsrp", "glbp", "carp") + # stub function to implement a finish call for each source def finish(self): pass + @staticmethod + def get_ip_address_role(ip_object): + """ + Returns the role of an IP address object. The NetBox API reports it as a dict, + an object this program created itself carries the plain value. + + Parameters + ---------- + ip_object: NBIPAddress + IP address object to get the role from + + Returns + ------- + (str, None): role of this IP address or None if unset + """ + + role = grab(ip_object, "data.role") + + if isinstance(role, dict): + return role.get("value") + + return role + def ip_is_primary_ip_of_object(self, ip_object, device_vm_object) -> bool: """ Check if a NBIPAddress object is currently set as primary IPv4 or IPv6 @@ -527,6 +554,7 @@ def add_update_interface(self, interface_object, device_object, interface_data, # try to find matching IP address object this_ip_object = None skip_this_ip = False + non_unique_ip_role = None for ip in self.inventory.get_all_items(NBIPAddress): # check if address matches (without prefix length) ip_address_string = grab(ip, "data.address", fallback="") @@ -571,6 +599,16 @@ def add_update_interface(self, interface_object, device_object, interface_data, this_ip_object = ip break + # an address whose role marks it as non unique belongs on several interfaces at + # the same time, so this interface gets an object of its own rather than taking + # this one over. Keep looking, this interface may already have its own object + current_ip_role = self.get_ip_address_role(ip) + if current_ip_role in self.non_unique_ip_roles: + log.debug(f"{ip.name} '{ip.get_display_name()}' is a '{current_ip_role}' address and " + f"can be assigned to multiple interfaces at the same time.") + non_unique_ip_role = current_ip_role + continue + # get current IP interface status current_nic_enabled = grab(current_ip_nic, "data.enabled", fallback=True) this_nic_enabled = grab(interface_object, "data.enabled", fallback=True) @@ -593,12 +631,6 @@ def add_update_interface(self, interface_object, device_object, interface_data, this_ip_object = ip - if grab(ip, "data.role.value") == "anycast": - log.debug(f"{ip.name} '{ip.get_display_name()}' is an Anycast address and " - f"can be assigned to multiple interfaces at the same time.") - skip_this_ip = True - break - if current_nic_enabled == this_nic_enabled: this_log_handler = log.warning @@ -664,6 +696,10 @@ def add_update_interface(self, interface_object, device_object, interface_data, nic_ip_data["tenant"] = ip_tenant if not isinstance(this_ip_object, NBIPAddress): + + if non_unique_ip_role is not None: + nic_ip_data["role"] = non_unique_ip_role + log.debug(f"No existing {NBIPAddress.name} object found. Creating a new one.") this_ip_object = self.inventory.add_object(NBIPAddress, data=nic_ip_data, source=self) @@ -686,9 +722,10 @@ def add_update_interface(self, interface_object, device_object, interface_data, if skip_ip_handling is True or skip_ip_removal is True: continue - if grab(current_ip, "data.role.value") == "anycast": - log.debug2(f"{current_ip.name} '{current_ip.get_display_name()}' is an Anycast address and will " - f"NOT be deleted from interface") + current_ip_role = self.get_ip_address_role(current_ip) + if current_ip_role in self.non_unique_ip_roles: + log.debug2(f"{current_ip.name} '{current_ip.get_display_name()}' is a '{current_ip_role}' address " + f"and will NOT be deleted from interface") continue if current_ip not in ip_address_objects: diff --git a/tests/test_ip_non_unique_roles.py b/tests/test_ip_non_unique_roles.py new file mode 100644 index 0000000..7050acf --- /dev/null +++ b/tests/test_ip_non_unique_roles.py @@ -0,0 +1,125 @@ +""" +An address whose NetBox role marks it as non-unique (anycast, vip, vrrp, hsrp, +glbp, carp) lives on more than one interface by definition. Discovering it on a +second machine has to assign it there as well instead of skipping it (issue #344). +""" +from types import SimpleNamespace + +import pytest + +from module.netbox.inventory import NetBoxInventory +from module.netbox.object_classes import ( + NBSite, NBClusterType, NBCluster, NBVM, NBVMInterface, NBIPAddress, +) +from module.sources.common.source_base import SourceBase + +NON_UNIQUE_ROLES = ["anycast", "vip", "vrrp", "hsrp", "glbp", "carp"] +ADDRESS = "10.0.0.5/24" + +# vmware object whose guest tools report as running, so IP handling is not skipped +_VM_TOOLS_RUNNING = SimpleNamespace(guest=SimpleNamespace(toolsRunningStatus="guestToolsRunning")) + + +@pytest.fixture +def inventory(): + def _reset(): + inv = NetBoxInventory() + inv.base_structure = {} + inv.source_list = [] + inv.init() + inv.netbox_api_version = "4.0.0" + return inv + + inv = _reset() + yield inv + _reset() + + +def _make_source(inventory): + src = SourceBase() + src.inventory = inventory + src.name = "test" + src.source_tag = "Source: test" + src.settings = SimpleNamespace( + skip_fhrp_group_ips=False, + preserve_primary_ips=False, + ip_tenant_inheritance_order=["disabled"], + disable_vlan_sync=True, + vlan_group_relation_by_id=None, + vlan_group_relation_by_name=None, + vlan_sync_exclude_by_id=None, + vlan_sync_exclude_by_name=None, + ) + src.return_longest_matching_prefix_for_ip = lambda *a, **k: None + return src + + +def _vm_and_nic(inv, name): + site = inv.add_object(NBSite, data={"name": "site1"}, read_from_netbox=True) + ctype = inv.add_object(NBClusterType, data={"name": "vmware"}, read_from_netbox=True) + cluster = inv.add_object(NBCluster, data={"name": "c1", "type": ctype, "scope": site}, + read_from_netbox=True) + vm = inv.add_object(NBVM, data={"name": name, "cluster": cluster, "status": "active"}, + read_from_netbox=True) + nic = inv.add_object(NBVMInterface, data={"name": "eth0", "virtual_machine": vm, "enabled": True}, + read_from_netbox=True) + return vm, nic + + +def _add_ip(inv, nic, role=None, address=ADDRESS): + data = { + "address": address, + "assigned_object_type": "virtualization.vminterface", + "assigned_object_id": nic, + } + if role is not None: + data["role"] = role + return inv.add_object(NBIPAddress, data=data, read_from_netbox=True) + + +def _sync(src, vm, nic, address=ADDRESS): + _iface, ip_objects = src.add_update_interface( + interface_object=nic, device_object=vm, + interface_data={"name": "eth0"}, interface_ips=[address], + vmware_object=_VM_TOOLS_RUNNING, + ) + return ip_objects + + +@pytest.mark.parametrize("role", NON_UNIQUE_ROLES) +def test_non_unique_address_is_assigned_to_the_second_interface(inventory, role): + _vm_a, nic_a = _vm_and_nic(inventory, "vm-a") + existing = _add_ip(inventory, nic_a, role=role) + vm_b, nic_b = _vm_and_nic(inventory, "vm-b") + + ip_objects = _sync(_make_source(inventory), vm_b, nic_b) + + assert ip_objects, f"a {role} address must be assigned to this interface as well" + assigned = ip_objects[0] + assert assigned is not existing, "the other interface must keep its own object" + assert assigned.get_interface() is nic_b + assert existing.get_interface() is nic_a, "the other interface lost its address" + assert str(assigned.data.get("role")) == role, "the new object must carry the same role" + + +def test_plain_duplicate_is_still_skipped(inventory): + # control: without a non-unique role the address stays ambiguous and is skipped + _vm_a, nic_a = _vm_and_nic(inventory, "vm-a") + _add_ip(inventory, nic_a) + vm_b, nic_b = _vm_and_nic(inventory, "vm-b") + + assert _sync(_make_source(inventory), vm_b, nic_b) == [] + + +def test_second_run_reuses_the_object_it_created(inventory): + # control: the fix must not create a new object on every run + _vm_a, nic_a = _vm_and_nic(inventory, "vm-a") + _add_ip(inventory, nic_a, role="anycast") + vm_b, nic_b = _vm_and_nic(inventory, "vm-b") + src = _make_source(inventory) + + first = _sync(src, vm_b, nic_b) + second = _sync(src, vm_b, nic_b) + + assert first and second and first[0] is second[0], "a second run created another object" + assert len(list(inventory.get_all_items(NBIPAddress))) == 2