NeoAnd/netbox-zabbix-sync
1
0
Fork 0
mirror of https://github.com/TheNetworkGuy/netbox-zabbix-sync.git synced 2026-03-21 12:08:39 -06:00
Code Issues Actions Packages Projects Releases Wiki Activity

Changed sync function to class

Browse Source
This commit is contained in:
TheNetworkGuy
2026-02-17 15:45:43 +00:00
parent d55fc0a4e7
commit 223a27f47c
5 changed files with 354 additions and 267 deletions
Download Patch File Download Diff File Expand all files Collapse all files
netbox_zabbix_sync/__init__.py
+1 -1
View File
@@ -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
netbox_zabbix_sync/modules/cli.py
+7 -3
View File
@@ -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():
netbox_zabbix_sync/modules/core.py
+314 -238
View File
@@ -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()
pyproject.toml
+2
View File
@@ -80,6 +80,8 @@ select = [
"tests/*" = [
# Ignore use of assert
"S101",
# Ignore hardcoded passwords / tokens
"S106",
]
tests/test_core.py
+30 -25
View File
@@ -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",