diff --git a/module/netbox/config.py b/module/netbox/config.py index 04df50b..318d727 100644 --- a/module/netbox/config.py +++ b/module/netbox/config.py @@ -79,7 +79,9 @@ def __init__(self): ConfigOption("prune_enabled", bool, description="""Whether items which were created by this program but - can't be found in any source anymore will be deleted or not + can't be found in any source anymore will be deleted or not. + With pruning disabled no DELETE request is sent to NetBox at all, + i.e. virtual disks which are gone from the source stay in NetBox """, default_value=False), diff --git a/module/netbox/connection.py b/module/netbox/connection.py index e32ebf6..4c77acd 100644 --- a/module/netbox/connection.py +++ b/module/netbox/connection.py @@ -730,7 +730,11 @@ def update_object(self, nb_object_sub_class, unset=False, last_run=False): this_object.resolve_relations() if last_run is True and getattr(this_object, "deleted", False) is True: - self.request(nb_object_sub_class, req_type="DELETE", nb_id=this_object.nb_id) + if self.settings.prune_enabled is True: + self.request(nb_object_sub_class, req_type="DELETE", nb_id=this_object.nb_id) + else: + log.info(f"{nb_object_sub_class.name} '{this_object.get_display_name()}' is gone from its " + f"source but pruning is disabled. Keeping it in NetBox.") # add class to resolved dependencies self.resolved_dependencies.add(nb_object_sub_class) @@ -942,6 +946,10 @@ def delete_unused_tags(self): deletes all tags with primary tag in description and the attribute 'tagged_items' returned 0 """ + if self.settings.prune_enabled is False: + log.debug("Pruning disabled. Not deleting unused tags") + return + self.query_current_data([NBTag]) for this_tag in self.inventory.get_all_items(NBTag): diff --git a/settings-example.ini b/settings-example.ini index 9a2b965..bf8ab5a 100644 --- a/settings-example.ini +++ b/settings-example.ini @@ -73,7 +73,8 @@ host_fqdn = netbox.example.com ;client_cert_key = client.key ; Whether items which were created by this program but can't be found in any source -; anymore will be deleted or not +; anymore will be deleted or not. With pruning disabled no DELETE request is sent to +; NetBox at all, i.e. virtual disks which are gone from the source stay in NetBox ;prune_enabled = False ; Orphaned objects will first be tagged before they get deleted. Once the amount of days diff --git a/tests/test_prune_disabled_no_delete.py b/tests/test_prune_disabled_no_delete.py new file mode 100644 index 0000000..b577d3d --- /dev/null +++ b/tests/test_prune_disabled_no_delete.py @@ -0,0 +1,85 @@ +# -*- coding: utf-8 -*- +# Copyright (c) 2020 - 2026 netbox-sync team. All rights reserved. +# +# netbox-sync.py +# +# This work is licensed under the terms of the MIT license. +# For a copy, see file LICENSE.txt included in this +# repository or visit: . + +"""With prune_enabled = False netbox-sync must not send a single DELETE request to NetBox. + +A token without the permission to delete is the sane setup for that case, and NetBox answers every +DELETE with a 403 which ends the whole run. +""" + +import types +from unittest.mock import Mock + +import pytest + +from module.netbox.connection import NetBoxHandler +from module.netbox.object_classes import NBCluster, NBClusterType, NBTag, NBVM, NBVirtualDisk + + +def make_handler(inventory, prune_enabled): + handler = object.__new__(NetBoxHandler) + handler.inventory = inventory + handler.settings = types.SimpleNamespace(prune_enabled=prune_enabled) + handler.resolved_dependencies = set() + handler.query_current_data = Mock() + handler.request = Mock(return_value=True) + return handler + + +def deletes(handler): + return [call for call in handler.request.call_args_list if call.kwargs.get("req_type") == "DELETE"] + + +def gone_disk(inventory): + """A virtual disk which exists in NetBox and which the source does not report anymore.""" + + cluster_type = inventory.add_object(NBClusterType, data={"id": 1, "name": "vmware"}, read_from_netbox=True) + cluster = inventory.add_object(NBCluster, data={"id": 1, "name": "cluster1", "type": cluster_type}, + read_from_netbox=True) + vm = inventory.add_object(NBVM, data={"id": 1, "name": "vm01", "cluster": cluster}, read_from_netbox=True) + disk = inventory.add_object(NBVirtualDisk, data={"id": 5, "name": "Hard disk 2", "virtual_machine": vm, + "size": 1024}, read_from_netbox=True) + disk.deleted = True + return disk + + +def unused_tag(inventory): + """A tag of this program which no object carries anymore.""" + + return inventory.add_object(NBTag, data={"id": 7, "name": "Source: gone", + "description": f"{NetBoxHandler.primary_tag}: unused", + "tagged_items": 0}, read_from_netbox=True) + + +def test_disk_gone_from_the_source_is_deleted_if_pruning_is_enabled(inventory): + gone_disk(inventory) + handler = make_handler(inventory, prune_enabled=True) + + handler.update_object(NBVirtualDisk, last_run=True) + + assert [call.kwargs["nb_id"] for call in deletes(handler)] == [5] + + +def test_disk_gone_from_the_source_is_kept_if_pruning_is_disabled(inventory): + gone_disk(inventory) + handler = make_handler(inventory, prune_enabled=False) + + handler.update_object(NBVirtualDisk, last_run=True) + + assert deletes(handler) == [] + + +@pytest.mark.parametrize("prune_enabled, expected", [(True, [7]), (False, [])]) +def test_unused_tags_are_only_deleted_if_pruning_is_enabled(inventory, prune_enabled, expected): + unused_tag(inventory) + handler = make_handler(inventory, prune_enabled=prune_enabled) + + handler.delete_unused_tags() + + assert [call.kwargs["nb_id"] for call in deletes(handler)] == expected