Adjusted Gitignore, added config module, adjusted requirements for YAML support, added first unittests

.gitignore

@@ -1,6 +1,6 @@
 *.log
 .venv
-config.py
+/config.py
 # Byte-compiled / optimized / DLL files
 __pycache__/
 *.py[cod]

config.yaml

@@ -0,0 +1,27 @@
+# Required: Custom Field name for Zabbix templates
+template_cf: "zabbix_templates"
+
+# Required: Custom Field name for Zabbix device
+device_cf: "zabbix_hostid"
+
+# Optional: Traverse site groups and assign Zabbix hostgroups based on site groups
+traverse_site_groups: false
+
+# Optional: Traverse regions and assign Zabbix hostgroups based on region hierarchy
+traverse_regions: false
+
+# Optional: Enable inventory syncing for host metadata
+inventory_sync: true
+
+# Optional: Choose which inventory fields to sync ("enabled", "manual", "disabled")
+inventory_mode: "manual"
+
+# Optional: Mapping of NetBox device fields to Zabbix inventory fields
+# See: https://www.zabbix.com/documentation/current/en/manual/api/reference/host/object#host_inventory
+inventory_map:
+  serial: "serial"
+  asset_tag: "asset_tag"
+  description: "comment"
+  location: "location"
+  contact: "contact"
+  site: "site"

modules/config.py

@@ -0,0 +1,37 @@
+"""
+Module for parsing configuration from the top level config.yaml file
+"""
+from pathlib import Path
+import yaml
+
+DEFAULT_CONFIG = {
+    "templates_config_context": False,
+    "templates_config_context_overrule": False,
+    "template_cf": "zabbix_template",
+    "device_cf": "zabbix_hostid",
+    "clustering": False,
+    "create_hostgroups": True,
+    "create_journal": False,
+    "sync_vms": False,
+    "zabbix_device_removal": ["Decommissioning", "Inventory"],
+    "zabbix_device_disable": ["Offline", "Planned", "Staged", "Failed"]
+}
+
+
+def load_config(config_path="config.yaml"):
+    """Loads config from YAML file and combines it with default config"""
+    # Get data from default config.
+    config = DEFAULT_CONFIG.copy()
+    # Set config path
+    config_file = Path(config_path)
+    # Check if file exists
+    if config_file.exists():
+        try:
+            with open(config_file, "r", encoding="utf-8") as f:
+                user_config = yaml.safe_load(f) or {}
+                config.update(user_config)
+        except OSError:
+            # Probably some I/O error with user permissions etc.
+            # Ignore for now and return default config
+            pass
+    return config

requirements.txt

@@ -1,2 +1,3 @@
 pynetbox
 zabbix-utils==2.0.1
+pyyaml

tests/test_device_deletion.py

@@ -0,0 +1,144 @@
+"""Testing device creation"""
+from unittest.mock import MagicMock, patch, call
+from modules.device import PhysicalDevice
+from modules.config import load_config
+
+config = load_config()
+
+
+def mock_nb_device():
+    mock = MagicMock()
+    mock.id = 1
+    mock.url = "http://netbox:8000/api/dcim/devices/1/"
+    mock.display_url = "http://netbox:8000/dcim/devices/1/"
+    mock.display = "SW01"
+    mock.name = "SW01"
+
+    mock.device_type = MagicMock()
+    mock.device_type.id = 1
+    mock.device_type.url = "http://netbox:8000/api/dcim/device-types/1/"
+    mock.device_type.display = "Catalyst 3750G-48TS-S"
+    mock.device_type.manufacturer = MagicMock()
+    mock.device_type.manufacturer.id = 1
+    mock.device_type.manufacturer.url = "http://netbox:8000/api/dcim/manufacturers/1/"
+    mock.device_type.manufacturer.display = "Cisco"
+    mock.device_type.manufacturer.name = "Cisco"
+    mock.device_type.manufacturer.slug = "cisco"
+    mock.device_type.manufacturer.description = ""
+    mock.device_type.model = "Catalyst 3750G-48TS-S"
+    mock.device_type.slug = "cisco-ws-c3750g-48ts-s"
+    mock.device_type.description = ""
+
+    mock.role = MagicMock()
+    mock.role.id = 1
+    mock.role.url = "http://netbox:8000/api/dcim/device-roles/1/"
+    mock.role.display = "Switch"
+    mock.role.name = "Switch"
+    mock.role.slug = "switch"
+    mock.role.description = ""
+
+    mock.tenant = None
+    mock.platform = None
+    mock.serial = "0031876"
+    mock.asset_tag = None
+
+    mock.site = MagicMock()
+    mock.site.id = 2
+    mock.site.url = "http://netbox:8000/api/dcim/sites/2/"
+    mock.site.display = "AMS01"
+    mock.site.name = "AMS01"
+    mock.site.slug = "ams01"
+    mock.site.description = ""
+
+    mock.location = None
+    mock.rack = None
+    mock.position = None
+    mock.face = None
+    mock.latitude = None
+    mock.longitude = None
+    mock.parent_device = None
+
+    mock.status = MagicMock()
+    mock.status.value = "decommissioning"
+    mock.status.label = "Decommissioning"
+
+    mock.cluster = None
+    mock.virtual_chassis = None
+    mock.vc_position = None
+    mock.vc_priority = None
+    mock.description = ""
+    mock.comments = ""
+    mock.config_template = None
+    mock.config_context = {}
+    mock.local_context_data = None
+    mock.tags = []
+
+    mock.custom_fields = {"zabbix_hostid": 1956}
+
+    def save(self):
+        pass
+
+    return mock
+
+def mock_zabbix():
+    mock = MagicMock()
+    mock.host.get.return_value = [{}]
+    mock.host.delete.return_value = True
+
+    return mock
+
+netbox_journals = MagicMock()
+nb_version = '4.2'
+create_journal = MagicMock()
+logger = MagicMock()
+
+def test_check_cluster_status():
+    """Checks if the isCluster function is functioning properly"""
+    nb_device = mock_nb_device()
+    zabbix = mock_zabbix()
+    device = PhysicalDevice(nb_device, zabbix, None, None,
+                            None, logger)
+    assert device.isCluster() == False
+
+
+def test_device_deletion_host_exists():
+    """Checks device deletion process"""
+    nb_device = mock_nb_device()
+    zabbix = mock_zabbix()
+    with patch.object(PhysicalDevice, 'create_journal_entry') as mock_journal:
+        # Create device
+        device = PhysicalDevice(nb_device, zabbix, netbox_journals, nb_version,
+                                create_journal, logger)
+        device.cleanup()
+        # Check if Zabbix HostID is empty
+        assert device.nb.custom_fields[config["device_cf"]] is None
+        # Check if API calls are executed
+        device.zabbix.host.get.assert_called_once_with(filter={'hostid': 1956}, output=[])
+        device.zabbix.host.delete.assert_called_once_with(1956)
+        # check logger
+        mock_journal.assert_called_once_with("warning", "Deleted host from Zabbix")
+        device.logger.info.assert_called_once_with("Host SW01: Deleted host from Zabbix.")
+
+
+def test_device_deletion_host_notExists():
+    nb_device = mock_nb_device()
+    zabbix = mock_zabbix()
+    zabbix.host.get.return_value = None
+
+    with patch.object(PhysicalDevice, 'create_journal_entry') as mock_journal:
+        # Create new device
+        device = PhysicalDevice(nb_device, zabbix, netbox_journals, nb_version,
+                                create_journal, logger)
+        # Try to clean the device up in Zabbix
+        device.cleanup()
+        # Confirm that a call was issued to Zabbix to check if the host exists
+        device.zabbix.host.get.assert_called_once_with(filter={'hostid': 1956}, output=[])
+        # Confirm that no device was deleted in Zabbix
+        device.zabbix.host.delete.assert_not_called()
+        # Test logging
+        log_calls = [
+            call('Host SW01: Deleted host from Zabbix.'),
+            call('Host SW01: was already deleted from Zabbix. Removed link in NetBox.')
+        ]
+        logger.info.assert_has_calls(log_calls)
+        assert logger.info.call_count == 2