Fixed several ruff and ty checks.

This commit is contained in:
TheNetworkGuy
2026-02-18 14:10:59 +00:00
parent dfba6f4714
commit c00ec4de31
3 changed files with 23 additions and 24 deletions
+7 -6
View File
@@ -2,6 +2,7 @@
import ssl
from os import environ
from typing import Any
from pynetbox import api as nbapi
from requests.exceptions import ConnectionError as RequestsConnectionError
@@ -27,7 +28,7 @@ class Sync:
This class is used to connect to NetBox and Zabbix and run the sync process.
"""
def __init__(self, config=None):
def __init__(self, config: dict[str, Any] | None = None):
"""
Docstring for __init__
@@ -36,7 +37,7 @@ class Sync:
"""
self.netbox = None
self.zabbix = None
self.config = config
self.config: dict[str, Any] = config if config else {}
self.nb_version = None
self._config()
@@ -182,9 +183,9 @@ class Sync:
# Set empty list for proxy processing Zabbix <= 6
zabbix_proxygroups = []
if str(self.zabbix.version).startswith("7"):
zabbix_proxygroups = self.zabbix.proxygroup.get(
zabbix_proxygroups = self.zabbix.proxygroup.get( # type: ignore[attr-defined]
output=["proxy_groupid", "name"]
) # type: ignore[attr-defined]
)
# Sanitize proxy data
if proxy_name == "host":
for proxy in zabbix_proxies:
@@ -220,7 +221,7 @@ class Sync:
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()
@@ -296,7 +297,7 @@ class Sync:
continue
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]
device.site = convert_recordset(
self.netbox.dcim.sites.filter(id=nb_device.site.id)
)
device.set_inventory(nb_device)
+1
View File
@@ -55,6 +55,7 @@ class PhysicalDevice:
self.hostgroups = []
self.hostgroup_type = "dev"
self.tenant = nb.tenant
self.site = nb.site
self.config_context = nb.config_context
self.zbxproxy = None
self.zabbix_state = 0
+15 -18
View File
@@ -81,8 +81,7 @@ class MockNetboxDevice:
# Setup device type with proper structure
if device_type is None:
self.device_type = MagicMock()
self.device_type.custom_fields = {
"zabbix_template": "TestTemplate"}
self.device_type.custom_fields = {"zabbix_template": "TestTemplate"}
self.device_type.manufacturer = MagicMock()
self.device_type.manufacturer.name = "TestManufacturer"
self.device_type.display = "Test Device Type"
@@ -107,7 +106,6 @@ class MockNetboxDevice:
def save(self):
"""Mock save method for NetBox device."""
pass
class MockNetboxVM:
@@ -308,8 +306,7 @@ 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"}
]
@@ -435,8 +432,7 @@ 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"}]
syncer = Sync()
syncer.connect(
@@ -463,8 +459,7 @@ 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 = []
syncer = Sync()
@@ -659,8 +654,7 @@ class TestDeviceHandeling(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"}
]
@@ -675,19 +669,22 @@ class TestDeviceHandeling(unittest.TestCase):
@patch("netbox_zabbix_sync.modules.core.ZabbixAPI")
@patch("netbox_zabbix_sync.modules.core.nbapi")
def test_sync_cluster_where_device_is_primary(
self, mock_api, mock_zabbix_api
):
def test_sync_cluster_where_device_is_primary(self, mock_api, mock_zabbix_api):
"""Test that sync properly handles a device that is the primary in a virtual chassis."""
# Create a device that is part of a virtual chassis and is the primary
# Setup virtual chassis mock
vc_master = MagicMock()
vc_master.id = 1 # Same as device ID - device is primary
virtual_chassis = MagicMock()
virtual_chassis.master = vc_master
virtual_chassis.name = "SW01"
device = MockNetboxDevice(
device_id=1,
name="SW01N0",
virtual_chassis=MagicMock(),
virtual_chassis=virtual_chassis,
)
device.virtual_chassis.master = MagicMock()
device.virtual_chassis.master.id = 1 # Same as device ID - device is primary
device.virtual_chassis.name = "SW01"
# Setup NetBox mock with a site for hostgroup
mock_netbox = self._setup_netbox_mock(mock_api)