Add unit tests for osism/tasks/netbox.py

osism/tasks/netbox.py

@@ -79,8 +79,10 @@ def _matches_netbox_filter(nb, netbox_filter, is_primary=False):
 
     filter_lower = netbox_filter.lower()
 
-    # Check if primary NetBox matches 'primary' filter
-    if is_primary and "primary" in filter_lower:
+    # Check if primary NetBox matches the 'primary' keyword. Mirror the field
+    # checks below (filter as a substring of the value) so that filters which
+    # merely contain "primary" (e.g. a site "primary-region") do not select it.
+    if is_primary and filter_lower in "primary":
         return True
 
     # Check URL
@@ -129,7 +131,8 @@ def set_maintenance(
                           If not provided, uses utils.secondary_nb_list.
 
     Returns:
-        bool: True if lock was acquired and operation succeeded, False if lock could not be acquired.
+        bool: True if the lock was acquired and all matching updates succeeded;
+              False if the lock could not be acquired or an update failed.
     """
     # Check if tasks are locked before execution
     utils.check_task_lock_and_exit()
@@ -139,6 +142,7 @@ def set_maintenance(
         auto_release_time=300,
     )
     if lock.acquire(timeout=120):
+        success = True
         try:
             # Process primary NetBox
             if _matches_netbox_filter(utils.nb, netbox_filter, is_primary=True):
@@ -148,6 +152,7 @@ def set_maintenance(
                 if not _update_netbox_device_field(
                     utils.nb, device_name, "maintenance", state
                 ):
+                    success = False
                     logger.error(
                         f"Could not set maintenance for {device_name} on {utils.nb.base_url}"
                     )
@@ -175,12 +180,13 @@ def set_maintenance(
                 if not _update_netbox_device_field(
                     nb, device_name, "maintenance", state
                 ):
+                    success = False
                     logger.error(
                         f"Could not set maintenance for {device_name} on {nb.base_url}"
                     )
         finally:
             lock.release()
-        return True
+        return success
     else:
         logger.error(f"Could not acquire lock for node {device_name}")
         return False
@@ -202,7 +208,8 @@ def set_provision_state(
                           If not provided, uses utils.secondary_nb_list.
 
     Returns:
-        bool: True if lock was acquired and operation succeeded, False if lock could not be acquired.
+        bool: True if the lock was acquired and all matching updates succeeded;
+              False if the lock could not be acquired or an update failed.
     """
     # Check if tasks are locked before execution
     utils.check_task_lock_and_exit()
@@ -212,6 +219,7 @@ def set_provision_state(
         auto_release_time=300,
     )
     if lock.acquire(timeout=120):
+        success = True
         try:
             # Process primary NetBox
             if _matches_netbox_filter(utils.nb, netbox_filter, is_primary=True):
@@ -221,6 +229,7 @@ def set_provision_state(
                 if not _update_netbox_device_field(
                     utils.nb, device_name, "provision_state", state
                 ):
+                    success = False
                     logger.error(
                         f"Could not set provision state for {device_name} on {utils.nb.base_url}"
                     )
@@ -248,12 +257,13 @@ def set_provision_state(
                 if not _update_netbox_device_field(
                     nb, device_name, "provision_state", state
                 ):
+                    success = False
                     logger.error(
                         f"Could not set provision state for {device_name} on {nb.base_url}"
                     )
         finally:
             lock.release()
-        return True
+        return success
     else:
         logger.error(f"Could not acquire lock for node {device_name}")
         return False
@@ -275,7 +285,8 @@ def set_power_state(
                           If not provided, uses utils.secondary_nb_list.
 
     Returns:
-        bool: True if lock was acquired and operation succeeded, False if lock could not be acquired.
+        bool: True if the lock was acquired and all matching updates succeeded;
+              False if the lock could not be acquired or an update failed.
     """
     # Convert None to "n/a" for clearer user feedback
     if state is None:
@@ -289,6 +300,7 @@ def set_power_state(
         auto_release_time=300,
     )
     if lock.acquire(timeout=120):
+        success = True
         try:
             # Process primary NetBox
             if _matches_netbox_filter(utils.nb, netbox_filter, is_primary=True):
@@ -298,6 +310,7 @@ def set_power_state(
                 if not _update_netbox_device_field(
                     utils.nb, device_name, "power_state", state
                 ):
+                    success = False
                     logger.error(
                         f"Could not set power state for {device_name} on {utils.nb.base_url}"
                     )
@@ -325,12 +338,13 @@ def set_power_state(
                 if not _update_netbox_device_field(
                     nb, device_name, "power_state", state
                 ):
+                    success = False
                     logger.error(
                         f"Could not set power state for {device_name} on {nb.base_url}"
                     )
         finally:
             lock.release()
-        return True
+        return success
     else:
         logger.error(f"Could not acquire lock for node {device_name}")
         return False
@@ -377,7 +391,9 @@ def get_interfaces_by_device(self, device_name):
 
 @app.task(bind=True, name="osism.tasks.netbox.get_addresses_by_device_and_interface")
 def get_addresses_by_device_and_interface(self, device_name, interface_name):
-    return utils.nb.dcim.addresses.filter(device=device_name, interface=interface_name)
+    return utils.nb.ipam.ip_addresses.filter(
+        device=device_name, interface=interface_name
+    )
 
 
 @app.task(bind=True, name="osism.tasks.netbox.manage")