fix: assign addresses with a non-unique role to every interface

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:

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