diff --git a/netbox_zabbix_sync/__init__.py b/netbox_zabbix_sync/__init__.py index 08f1ffd..7517e89 100644 --- a/netbox_zabbix_sync/__init__.py +++ b/netbox_zabbix_sync/__init__.py @@ -2,4 +2,4 @@ Makes core module sync function available at package level for easier imports. """ -from netbox_zabbix_sync.modules.core import sync as sync +from netbox_zabbix_sync.modules.core import Sync as Sync diff --git a/netbox_zabbix_sync/modules/cli.py b/netbox_zabbix_sync/modules/cli.py index 48a50e9..016e28a 100644 --- a/netbox_zabbix_sync/modules/cli.py +++ b/netbox_zabbix_sync/modules/cli.py @@ -2,10 +2,12 @@ import argparse import logging from os import environ -from netbox_zabbix_sync.modules.core import sync +from netbox_zabbix_sync.modules.core import Sync from netbox_zabbix_sync.modules.exceptions import EnvironmentVarError -from netbox_zabbix_sync.modules.logging import get_logger, set_log_levels +from netbox_zabbix_sync.modules.logging import get_logger, set_log_levels, setup_logger +# Set logging +setup_logger() logger = get_logger() @@ -47,7 +49,8 @@ def main(arguments): netbox_token = environ.get("NETBOX_TOKEN") # Run main sync process - sync( + syncer = Sync() + syncer.connect( nb_host=netbox_host, nb_token=netbox_token, zbx_host=zabbix_host, @@ -55,6 +58,7 @@ def main(arguments): zbx_pass=zabbix_pass, zbx_token=zabbix_token, ) + syncer.start() def parse_cli(): diff --git a/netbox_zabbix_sync/modules/core.py b/netbox_zabbix_sync/modules/core.py index 111dad7..d53c305 100644 --- a/netbox_zabbix_sync/modules/core.py +++ b/netbox_zabbix_sync/modules/core.py @@ -5,14 +5,13 @@ import sys from os import environ from pynetbox import api -from pynetbox.core.query import RequestError as NBRequestError from requests.exceptions import ConnectionError as RequestsConnectionError from zabbix_utils import APIRequestError, ProcessingError, ZabbixAPI from netbox_zabbix_sync.modules.device import PhysicalDevice from netbox_zabbix_sync.modules.exceptions import SyncError -from netbox_zabbix_sync.modules.logging import get_logger, setup_logger -from netbox_zabbix_sync.modules.settings import load_config +from netbox_zabbix_sync.modules.logging import get_logger +from netbox_zabbix_sync.modules.settings import DEFAULT_CONFIG, load_config from netbox_zabbix_sync.modules.tools import ( convert_recordset, proxy_prepper, @@ -20,259 +19,336 @@ from netbox_zabbix_sync.modules.tools import ( ) from netbox_zabbix_sync.modules.virtual_machine import VirtualMachine -# Import configuration settings -config = load_config() - - -setup_logger() logger = get_logger() -def sync(nb_host, nb_token, zbx_host, zbx_user, zbx_pass, zbx_token): +class Sync: """ - Run the NetBox to Zabbix sync process. + Class that hosts the main sync process. + This class is used to connect to NetBox and Zabbix and run the sync process. """ - # Set NetBox API - netbox = api(nb_host, token=nb_token, threading=True) - # Create API call to get all custom fields which are on the device objects - try: - # Get NetBox version - nb_version = netbox.version - logger.debug("NetBox version is %s.", nb_version) - except RequestsConnectionError: - logger.error( - "Unable to connect to NetBox with URL %s. Please check the URL and status of NetBox.", - nb_host, - ) - sys.exit(1) - except NBRequestError as e: - logger.error("NetBox error: %s", e) - sys.exit(1) - # Check if the provided Hostgroup layout is valid - device_cfs = [] - vm_cfs = [] - device_cfs = list( - netbox.extras.custom_fields.filter( - type=["text", "object", "select"], content_types="dcim.device" - ) - ) - verify_hg_format( - config["hostgroup_format"], device_cfs=device_cfs, hg_type="dev", logger=logger - ) - if config["sync_vms"]: - vm_cfs = list( - netbox.extras.custom_fields.filter( - type=["text", "object", "select"], - content_types="virtualization.virtualmachine", + + def __init__(self, config=None): + """ + Docstring for __init__ + + :param self: Description + :param config: Description + """ + self.netbox = None + self.zabbix = None + self.config = config + self.nb_version = None + self._config() + + def _config(self): + """ + Load config and check if provided config is valid. + """ + if not self.config: + self.config = load_config() + return True + # Check if provided config is valid + if not isinstance(self.config, dict): + e = "Provided config is not a dictionary." + logger.error(e) + raise SyncError(e) + # Combine default options and provided config + default_config = DEFAULT_CONFIG.copy() + for key in self.config: + # Check if the user provided an invalid option parameter + if key not in default_config: + e = f"Provided config contains invalid key: {key}." + logger.error(e) + raise SyncError(e) + # Remove keys from default config to keep track of missing keys + default_config.pop(key) + # Add missing options with default values + for key in default_config: + self.config[key] = default_config[key] + return True + + def connect( + self, nb_host, nb_token, zbx_host, zbx_user=None, zbx_pass=None, zbx_token=None + ): + """ + Docstring for connect + + :param self: Description + :param nb_host: Description + :param nb_token: Description + :param zbx_host: Description + :param zbx_user: Description + :param zbx_pass: Description + :param zbx_token: Description + """ + # Initialize Netbox API connection + netbox = api(nb_host, token=nb_token, threading=True) + try: + # Get NetBox version + nb_version = netbox.version + logger.debug("NetBox version is %s.", nb_version) + self.netbox = netbox + self.nb_version = nb_version + except RequestsConnectionError: + logger.error( + "Unable to connect to NetBox with URL %s. Please check the URL and status of NetBox.", + nb_host, + ) + # Set Zabbix API + try: + ssl_ctx = ssl.create_default_context() + + # If a custom CA bundle is set for pynetbox (requests), also use it for the Zabbix API + if environ.get("REQUESTS_CA_BUNDLE", None): + ssl_ctx.load_verify_locations(environ["REQUESTS_CA_BUNDLE"]) + + if not zbx_token: + self.zabbix = ZabbixAPI( + zbx_host, user=zbx_user, password=zbx_pass, ssl_context=ssl_ctx + ) + else: + self.zabbix = ZabbixAPI(zbx_host, token=zbx_token, ssl_context=ssl_ctx) + self.zabbix.check_auth() + except (APIRequestError, ProcessingError) as zbx_error: + e = f"Zabbix returned the following error: {zbx_error}." + logger.error(e) + sys.exit(1) + + def start(self): + """ + Run the NetBox to Zabbix sync process. + """ + if not self.netbox or not self.zabbix: + e = "Not connected to NetBox or Zabbix. Please run the connect function first." + logger.error(e) + raise SyncError(e) + device_cfs = [] + vm_cfs = [] + # Create API call to get all custom fields which are on the device objects + device_cfs = list( + self.netbox.extras.custom_fields.filter( + type=["text", "object", "select"], content_types="dcim.device" ) ) + # Check if the provided Hostgroup layout is valid verify_hg_format( - config["vm_hostgroup_format"], vm_cfs=vm_cfs, hg_type="vm", logger=logger + self.config["hostgroup_format"], + device_cfs=device_cfs, + hg_type="dev", + logger=logger, ) - # Set Zabbix API - try: - ssl_ctx = ssl.create_default_context() - - # If a custom CA bundle is set for pynetbox (requests), also use it for the Zabbix API - if environ.get("REQUESTS_CA_BUNDLE", None): - ssl_ctx.load_verify_locations(environ["REQUESTS_CA_BUNDLE"]) - - if not zbx_token: - zabbix = ZabbixAPI( - zbx_host, user=zbx_user, password=zbx_pass, ssl_context=ssl_ctx + if self.config["sync_vms"]: + vm_cfs = list( + self.netbox.extras.custom_fields.filter( + type=["text", "object", "select"], + content_types="virtualization.virtualmachine", + ) ) - else: - zabbix = ZabbixAPI(zbx_host, token=zbx_token, ssl_context=ssl_ctx) - zabbix.check_auth() - except (APIRequestError, ProcessingError) as zbx_error: - e = f"Zabbix returned the following error: {zbx_error}." - logger.error(e) - sys.exit(1) - # Set API parameter mapping based on API version - proxy_name = "host" if not str(zabbix.version).startswith("7") else "name" - # Get all Zabbix and NetBox data - netbox_devices = list(netbox.dcim.devices.filter(**config["nb_device_filter"])) - netbox_vms = [] - if config["sync_vms"]: - netbox_vms = list( - netbox.virtualization.virtual_machines.filter(**config["nb_vm_filter"]) + verify_hg_format( + self.config["vm_hostgroup_format"], + vm_cfs=vm_cfs, + hg_type="vm", + logger=logger, + ) + # Set API parameter mapping based on API version + proxy_name = "host" if not str(self.zabbix.version).startswith("7") else "name" + # Get all Zabbix and NetBox data + netbox_devices = list( + self.netbox.dcim.devices.filter(**self.config["nb_device_filter"]) ) - netbox_site_groups = convert_recordset(netbox.dcim.site_groups.all()) - netbox_regions = convert_recordset(netbox.dcim.regions.all()) - netbox_journals = netbox.extras.journal_entries - zabbix_groups = zabbix.hostgroup.get(output=["groupid", "name"]) # type: ignore[attr-defined] - zabbix_templates = zabbix.template.get(output=["templateid", "name"]) # type: ignore[attr-defined] - zabbix_proxies = zabbix.proxy.get(output=["proxyid", proxy_name]) # type: ignore[attr-defined] - # Set empty list for proxy processing Zabbix <= 6 - zabbix_proxygroups = [] - if str(zabbix.version).startswith("7"): - zabbix_proxygroups = zabbix.proxygroup.get(output=["proxy_groupid", "name"]) # type: ignore[attr-defined] - # Sanitize proxy data - if proxy_name == "host": - for proxy in zabbix_proxies: - proxy["name"] = proxy.pop("host") - # Prepare list of all proxy and proxy_groups - zabbix_proxy_list = proxy_prepper(zabbix_proxies, zabbix_proxygroups) + netbox_vms = [] + if self.config["sync_vms"]: + netbox_vms = list( + self.netbox.virtualization.virtual_machines.filter( + **self.config["nb_vm_filter"] + ) + ) + netbox_site_groups = convert_recordset(self.netbox.dcim.site_groups.all()) + netbox_regions = convert_recordset(self.netbox.dcim.regions.all()) + netbox_journals = self.netbox.extras.journal_entries + zabbix_groups = self.zabbix.hostgroup.get(output=["groupid", "name"]) # type: ignore[attr-defined] + zabbix_templates = self.zabbix.template.get(output=["templateid", "name"]) # type: ignore[attr-defined] + zabbix_proxies = self.zabbix.proxy.get(output=["proxyid", proxy_name]) # type: ignore[attr-defined] + # Set empty list for proxy processing Zabbix <= 6 + zabbix_proxygroups = [] + if str(self.zabbix.version).startswith("7"): + zabbix_proxygroups = self.zabbix.proxygroup.get( + output=["proxy_groupid", "name"] + ) # type: ignore[attr-defined] + # Sanitize proxy data + if proxy_name == "host": + for proxy in zabbix_proxies: + proxy["name"] = proxy.pop("host") + # Prepare list of all proxy and proxy_groups + zabbix_proxy_list = proxy_prepper(zabbix_proxies, zabbix_proxygroups) - # Go through all NetBox devices - for nb_vm in netbox_vms: - try: - vm = VirtualMachine( - nb_vm, - zabbix, - netbox_journals, - nb_version, - config["create_journal"], - logger, - ) - logger.debug("Host %s: Started operations on VM.", vm.name) - vm.set_vm_template() - # Check if a valid template has been found for this VM. - if not vm.zbx_template_names: - continue - vm.set_hostgroup( - config["vm_hostgroup_format"], netbox_site_groups, netbox_regions - ) - # Check if a valid hostgroup has been found for this VM. - if not vm.hostgroups: - continue - if config["extended_site_properties"] and nb_vm.site: - logger.debug("VM %s: extending site information.", vm.name) - vm.site = convert_recordset(netbox.dcim.sites.filter(id=nb_vm.site.id)) # type: ignore[attr-defined] - vm.set_inventory(nb_vm) - vm.set_usermacros() - vm.set_tags() - # Checks if device is in cleanup state - if vm.status in config["zabbix_device_removal"]: - if vm.zabbix_id: - # Delete device from Zabbix - # and remove hostID from NetBox. - vm.cleanup() - logger.info("VM %s: cleanup complete", vm.name) + # Go through all NetBox devices + for nb_vm in netbox_vms: + try: + vm = VirtualMachine( + nb_vm, + self.zabbix, + netbox_journals, + self.nb_version, + self.config["create_journal"], + logger, + ) + logger.debug("Host %s: Started operations on VM.", vm.name) + vm.set_vm_template() + # Check if a valid template has been found for this VM. + if not vm.zbx_template_names: continue - # Device has been added to NetBox - # but is not in Activate state - logger.info( - "VM %s: Skipping since this VM is not in the active state.", vm.name + vm.set_hostgroup( + self.config["vm_hostgroup_format"], + netbox_site_groups, + netbox_regions, ) - continue - # Check if the VM is in the disabled state - if vm.status in config["zabbix_device_disable"]: - vm.zabbix_state = 1 - # Add hostgroup if config is set - if config["create_hostgroups"]: - # Create new hostgroup. Potentially multiple groups if nested - hostgroups = vm.create_zbx_hostgroup(zabbix_groups) - # go through all newly created hostgroups - for group in hostgroups: - # Add new hostgroups to zabbix group list - zabbix_groups.append(group) - # Check if VM is already in Zabbix - if vm.zabbix_id: - vm.consistency_check( - zabbix_groups, - zabbix_templates, - zabbix_proxy_list, - config["full_proxy_sync"], - config["create_hostgroups"], - ) - continue - # Add VM to Zabbix - vm.create_in_zabbix(zabbix_groups, zabbix_templates, zabbix_proxy_list) - except SyncError: - pass - - for nb_device in netbox_devices: - try: - # Set device instance set data such as hostgroup and template information. - device = PhysicalDevice( - nb_device, - zabbix, - netbox_journals, - nb_version, - config["create_journal"], - logger, - ) - logger.debug("Host %s: Started operations on device.", device.name) - device.set_template( - config["templates_config_context"], - config["templates_config_context_overrule"], - ) - # Check if a valid template has been found for this VM. - if not device.zbx_template_names: - continue - device.set_hostgroup( - config["hostgroup_format"], netbox_site_groups, netbox_regions - ) - # Check if a valid hostgroup has been found for this VM. - if not device.hostgroups: - logger.warning( - "Host %s: Host has no valid hostgroups, Skipping this host...", - device.name, - ) - continue - if config["extended_site_properties"] and nb_device.site: - logger.debug("Device %s: extending site information.", device.name) - device.site = convert_recordset( # type: ignore[attr-defined] - netbox.dcim.sites.filter(id=nb_device.site.id) - ) - device.set_inventory(nb_device) - device.set_usermacros() - device.set_tags() - # Checks if device is part of cluster. - # Requires clustering variable - if device.is_cluster() and config["clustering"]: - # Check if device is primary or secondary - if device.promote_primary_device(): + # Check if a valid hostgroup has been found for this VM. + if not vm.hostgroups: + continue + if self.config["extended_site_properties"] and nb_vm.site: + logger.debug("VM %s: extending site information.", vm.name) + vm.site = convert_recordset( + self.netbox.dcim.sites.filter(id=nb_vm.site.id) + ) # type: ignore[attr-defined] + vm.set_inventory(nb_vm) + vm.set_usermacros() + vm.set_tags() + # Checks if device is in cleanup state + if vm.status in self.config["zabbix_device_removal"]: + if vm.zabbix_id: + # Delete device from Zabbix + # and remove hostID from self.netbox. + vm.cleanup() + logger.info("VM %s: cleanup complete", vm.name) + continue + # Device has been added to NetBox + # but is not in Activate state logger.info( - "Device %s: is part of cluster and primary.", device.name + "VM %s: Skipping since this VM is not in the active state.", + vm.name, ) - else: - # Device is secondary in cluster. - # Don't continue with this device. - logger.info( - "Device %s: Is part of cluster but not primary. Skipping this host...", + continue + # Check if the VM is in the disabled state + if vm.status in self.config["zabbix_device_disable"]: + vm.zabbix_state = 1 + # Add hostgroup if config is set + if self.config["create_hostgroups"]: + # Create new hostgroup. Potentially multiple groups if nested + hostgroups = vm.create_zbx_hostgroup(zabbix_groups) + # go through all newly created hostgroups + for group in hostgroups: + # Add new hostgroups to zabbix group list + zabbix_groups.append(group) + # Check if VM is already in Zabbix + if vm.zabbix_id: + vm.consistency_check( + zabbix_groups, + zabbix_templates, + zabbix_proxy_list, + self.config["full_proxy_sync"], + self.config["create_hostgroups"], + ) + continue + # Add VM to Zabbix + vm.create_in_zabbix(zabbix_groups, zabbix_templates, zabbix_proxy_list) + except SyncError: + pass + + for nb_device in netbox_devices: + try: + # Set device instance set data such as hostgroup and template information. + device = PhysicalDevice( + nb_device, + self.zabbix, + netbox_journals, + self.nb_version, + self.config["create_journal"], + logger, + ) + logger.debug("Host %s: Started operations on device.", device.name) + device.set_template( + self.config["templates_config_context"], + self.config["templates_config_context_overrule"], + ) + # Check if a valid template has been found for this VM. + if not device.zbx_template_names: + continue + device.set_hostgroup( + self.config["hostgroup_format"], netbox_site_groups, netbox_regions + ) + # Check if a valid hostgroup has been found for this VM. + if not device.hostgroups: + logger.warning( + "Host %s: Host has no valid hostgroups, Skipping this host...", device.name, ) continue - # Checks if device is in cleanup state - if device.status in config["zabbix_device_removal"]: - if device.zabbix_id: - # Delete device from Zabbix - # and remove hostID from NetBox. - device.cleanup() - logger.info("Device %s: cleanup complete", device.name) + if self.config["extended_site_properties"] and nb_device.site: + logger.debug("Device %s: extending site information.", device.name) + device.site = convert_recordset( # type: ignore[attr-defined] + self.netbox.dcim.sites.filter(id=nb_device.site.id) + ) + device.set_inventory(nb_device) + device.set_usermacros() + device.set_tags() + # Checks if device is part of cluster. + # Requires clustering variable + if device.is_cluster() and self.config["clustering"]: + # Check if device is primary or secondary + if device.promote_primary_device(): + logger.info( + "Device %s: is part of cluster and primary.", device.name + ) + else: + # Device is secondary in cluster. + # Don't continue with this device. + logger.info( + "Device %s: Is part of cluster but not primary. Skipping this host...", + device.name, + ) + continue + # Checks if device is in cleanup state + if device.status in self.config["zabbix_device_removal"]: + if device.zabbix_id: + # Delete device from Zabbix + # and remove hostID from NetBox. + device.cleanup() + logger.info("Device %s: cleanup complete", device.name) + continue + # Device has been added to NetBox + # but is not in Activate state + logger.info( + "Device %s: Skipping since this device is not in the active state.", + device.name, + ) continue - # Device has been added to NetBox - # but is not in Activate state - logger.info( - "Device %s: Skipping since this device is not in the active state.", - device.name, + # Check if the device is in the disabled state + if device.status in self.config["zabbix_device_disable"]: + device.zabbix_state = 1 + # Add hostgroup is config is set + if self.config["create_hostgroups"]: + # Create new hostgroup. Potentially multiple groups if nested + hostgroups = device.create_zbx_hostgroup(zabbix_groups) + # go through all newly created hostgroups + for group in hostgroups: + # Add new hostgroups to zabbix group list + zabbix_groups.append(group) + # Check if device is already in Zabbix + if device.zabbix_id: + device.consistency_check( + zabbix_groups, + zabbix_templates, + zabbix_proxy_list, + self.config["full_proxy_sync"], + self.config["create_hostgroups"], + ) + continue + # Add device to Zabbix + device.create_in_zabbix( + zabbix_groups, zabbix_templates, zabbix_proxy_list ) - continue - # Check if the device is in the disabled state - if device.status in config["zabbix_device_disable"]: - device.zabbix_state = 1 - # Add hostgroup is config is set - if config["create_hostgroups"]: - # Create new hostgroup. Potentially multiple groups if nested - hostgroups = device.create_zbx_hostgroup(zabbix_groups) - # go through all newly created hostgroups - for group in hostgroups: - # Add new hostgroups to zabbix group list - zabbix_groups.append(group) - # Check if device is already in Zabbix - if device.zabbix_id: - device.consistency_check( - zabbix_groups, - zabbix_templates, - zabbix_proxy_list, - config["full_proxy_sync"], - config["create_hostgroups"], - ) - continue - # Add device to Zabbix - device.create_in_zabbix(zabbix_groups, zabbix_templates, zabbix_proxy_list) - except SyncError: - pass - zabbix.logout() + except SyncError: + pass + self.zabbix.logout() diff --git a/pyproject.toml b/pyproject.toml index 640fdf8..df55916 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -80,6 +80,8 @@ select = [ "tests/*" = [ # Ignore use of assert "S101", + # Ignore hardcoded passwords / tokens + "S106", ] diff --git a/tests/test_core.py b/tests/test_core.py index 77dca09..e45c2dd 100644 --- a/tests/test_core.py +++ b/tests/test_core.py @@ -7,7 +7,7 @@ from pynetbox.core.query import RequestError as NBRequestError from requests.exceptions import ConnectionError as RequestsConnectionError from zabbix_utils import APIRequestError, ProcessingError -from netbox_zabbix_sync.modules.core import sync +from netbox_zabbix_sync.modules.core import Sync # Minimal config for testing - includes all keys used by sync() TEST_CONFIG = { @@ -88,14 +88,16 @@ class TestSyncNetboxConnection(unittest.TestCase): ) with self.assertRaises(SystemExit) as context: - sync( - "http://netbox.local", - "token", - "http://zabbix.local", - "user", - "pass", - None, + syncer = Sync() + syncer.connect( + nb_host="http://netbox.local", + nb_token="token", + zbx_host="http://zabbix.local", + zbx_user="user", + zbx_pass="pass", + zbx_token=None, ) + syncer.start() self.assertEqual(context.exception.code, 1) @@ -110,7 +112,7 @@ class TestSyncNetboxConnection(unittest.TestCase): ) with self.assertRaises(SystemExit) as context: - sync( + Sync( "http://netbox.local", "token", "http://zabbix.local", @@ -150,7 +152,7 @@ class TestSyncZabbixConnection(unittest.TestCase): ) with self.assertRaises(SystemExit) as context: - sync( + Sync( "http://netbox.local", "token", "http://zabbix.local", @@ -173,7 +175,7 @@ class TestSyncZabbixConnection(unittest.TestCase): ) with self.assertRaises(SystemExit) as context: - sync( + Sync( "http://netbox.local", "token", "http://zabbix.local", @@ -219,7 +221,7 @@ class TestSyncZabbixAuthentication(unittest.TestCase): self._setup_netbox_mock(mock_api) self._setup_zabbix_mock(mock_zabbix_api) - sync( + Sync( "http://netbox.local", "nb_token", "http://zabbix.local", @@ -243,7 +245,7 @@ class TestSyncZabbixAuthentication(unittest.TestCase): self._setup_netbox_mock(mock_api) self._setup_zabbix_mock(mock_zabbix_api) - sync( + Sync( "http://netbox.local", "nb_token", "http://zabbix.local", @@ -281,7 +283,8 @@ class TestSyncDeviceProcessing(unittest.TestCase): mock_zabbix = MagicMock() mock_zabbix_api.return_value = mock_zabbix mock_zabbix.version = version - mock_zabbix.hostgroup.get.return_value = [{"groupid": "1", "name": "TestGroup"}] + mock_zabbix.hostgroup.get.return_value = [ + {"groupid": "1", "name": "TestGroup"}] mock_zabbix.template.get.return_value = [ {"templateid": "1", "name": "TestTemplate"} ] @@ -308,7 +311,7 @@ class TestSyncDeviceProcessing(unittest.TestCase): mock_device_instance.zbx_template_names = [] mock_physical_device.return_value = mock_device_instance - sync( + Sync( "http://netbox.local", "nb_token", "http://zabbix.local", @@ -340,7 +343,7 @@ class TestSyncDeviceProcessing(unittest.TestCase): mock_vm_instance.zbx_template_names = [] mock_virtual_machine.return_value = mock_vm_instance - sync( + Sync( "http://netbox.local", "nb_token", "http://zabbix.local", @@ -365,7 +368,7 @@ class TestSyncDeviceProcessing(unittest.TestCase): self._setup_netbox_mock(mock_api, vms=[vm1]) self._setup_zabbix_mock(mock_zabbix_api) - sync( + Sync( "http://netbox.local", "nb_token", "http://zabbix.local", @@ -405,9 +408,10 @@ class TestSyncZabbixVersionHandling(unittest.TestCase): mock_zabbix.version = "6.0" mock_zabbix.hostgroup.get.return_value = [] mock_zabbix.template.get.return_value = [] - mock_zabbix.proxy.get.return_value = [{"proxyid": "1", "host": "proxy1"}] + mock_zabbix.proxy.get.return_value = [ + {"proxyid": "1", "host": "proxy1"}] - sync( + Sync( "http://netbox.local", "nb_token", "http://zabbix.local", @@ -431,10 +435,11 @@ class TestSyncZabbixVersionHandling(unittest.TestCase): mock_zabbix.version = "7.0" mock_zabbix.hostgroup.get.return_value = [] mock_zabbix.template.get.return_value = [] - mock_zabbix.proxy.get.return_value = [{"proxyid": "1", "name": "proxy1"}] + mock_zabbix.proxy.get.return_value = [ + {"proxyid": "1", "name": "proxy1"}] mock_zabbix.proxygroup.get.return_value = [] - sync( + Sync( "http://netbox.local", "nb_token", "http://zabbix.local", @@ -461,7 +466,7 @@ class TestSyncZabbixVersionHandling(unittest.TestCase): mock_zabbix.proxy.get.return_value = [] mock_zabbix.proxygroup.get.return_value = [] - sync( + Sync( "http://netbox.local", "nb_token", "http://zabbix.local", @@ -487,7 +492,7 @@ class TestSyncZabbixVersionHandling(unittest.TestCase): mock_zabbix.template.get.return_value = [] mock_zabbix.proxy.get.return_value = [] - sync( + Sync( "http://netbox.local", "nb_token", "http://zabbix.local", @@ -529,7 +534,7 @@ class TestSyncLogout(unittest.TestCase): mock_zabbix.template.get.return_value = [] mock_zabbix.proxy.get.return_value = [] - sync( + Sync( "http://netbox.local", "nb_token", "http://zabbix.local", @@ -579,7 +584,7 @@ class TestSyncProxyNameSanitization(unittest.TestCase): ] mock_proxy_prepper.return_value = [] - sync( + Sync( "http://netbox.local", "nb_token", "http://zabbix.local",