diff --git a/.gitignore b/.gitignore index 5fdbd95..27761cd 100644 --- a/.gitignore +++ b/.gitignore @@ -4,4 +4,5 @@ # Byte-compiled / optimized / DLL files __pycache__/ *.py[cod] -.vscode \ No newline at end of file +.vscode +.flake \ No newline at end of file diff --git a/modules/config.py b/modules/config.py index 3c74216..3eca1d8 100644 --- a/modules/config.py +++ b/modules/config.py @@ -18,7 +18,8 @@ DEFAULT_CONFIG = { "create_journal": False, "sync_vms": False, "zabbix_device_removal": ["Decommissioning", "Inventory"], - "zabbix_device_disable": ["Offline", "Planned", "Staged", "Failed"] + "zabbix_device_disable": ["Offline", "Planned", "Staged", "Failed"], + "inventory_mode": "disabled" } @@ -36,7 +37,7 @@ def load_config(): def load_env_variable(config_environvar): """Returns config from environment variable""" - prefix = "NZS_" + prefix = "NBZX_" config_environvar = prefix + config_environvar.upper() if config_environvar in environ: return environ[config_environvar] diff --git a/tests/test_configuration_parsing.py b/tests/test_configuration_parsing.py index 4f97abf..23438b4 100644 --- a/tests/test_configuration_parsing.py +++ b/tests/test_configuration_parsing.py @@ -102,15 +102,26 @@ def test_load_config_file_not_found(): def test_load_env_variable_function(): """Test the load_env_variable function directly""" - # Test when the environment variable exists - with patch.dict(os.environ, {"templates_config_context": "True"}): + # Create a real environment variable for testing with correct prefix and uppercase + test_var = "NBZX_TEMPLATES_CONFIG_CONTEXT" + original_env = os.environ.get(test_var, None) + try: + # Set the environment variable with the proper prefix and case + os.environ[test_var] = "True" + + # Test that it's properly read (using lowercase in the function call) value = load_env_variable("templates_config_context") assert value == "True" - # Test when the environment variable doesn't exist - with patch.dict(os.environ, {}, clear=True): + # Test when the environment variable doesn't exist value = load_env_variable("nonexistent_variable") assert value is None + finally: + # Clean up - restore original environment + if original_env is not None: + os.environ[test_var] = original_env + else: + os.environ.pop(test_var, None) def test_load_config_file_exception_handling(): diff --git a/tests/test_device_deletion.py b/tests/test_device_deletion.py index 41c4420..392ba1a 100644 --- a/tests/test_device_deletion.py +++ b/tests/test_device_deletion.py @@ -1,142 +1,166 @@ -"""Testing device creation""" -from unittest.mock import MagicMock, patch, call +"""Tests for device deletion functionality in the PhysicalDevice class.""" +import unittest +from unittest.mock import MagicMock, patch +from zabbix_utils import APIRequestError from modules.device import PhysicalDevice -from modules.config import load_config - -config = load_config() +from modules.exceptions import SyncExternalError -def mock_nb_device(): - """Function to mock Netbox 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" +class TestDeviceDeletion(unittest.TestCase): + """Test class for device deletion functionality.""" - mock.device_type = MagicMock() - mock.device_type.display = "Catalyst 3750G-48TS-S" - mock.device_type.manufacturer = MagicMock() - 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" + def setUp(self): + """Set up test fixtures.""" + # Create mock NetBox device + self.mock_nb_device = MagicMock() + self.mock_nb_device.id = 123 + self.mock_nb_device.name = "test-device" + self.mock_nb_device.status.label = "Decommissioning" + self.mock_nb_device.custom_fields = {"zabbix_hostid": "456"} + self.mock_nb_device.config_context = {} - mock.role = MagicMock() - mock.role.id = 1 - mock.role.display = "Switch" - mock.role.name = "Switch" - mock.role.slug = "switch" + # Set up a primary IP + primary_ip = MagicMock() + primary_ip.address = "192.168.1.1/24" + self.mock_nb_device.primary_ip = primary_ip - mock.tenant = None - mock.platform = None - mock.serial = "0031876" - mock.asset_tag = None + # Create mock Zabbix API + self.mock_zabbix = MagicMock() + self.mock_zabbix.version = "6.0" - mock.site = MagicMock() - mock.site.display = "AMS01" - mock.site.name = "AMS01" - mock.site.slug = "ams01" + # Set up mock host.get response + self.mock_zabbix.host.get.return_value = [{"hostid": "456"}] - mock.location = None - mock.rack = None - mock.position = None - mock.face = None - mock.parent_device = None + # Mock NetBox journal class + self.mock_nb_journal = MagicMock() - mock.status = MagicMock() - mock.status.value = "decommissioning" - mock.status.label = "Decommissioning" + # Create logger mock + self.mock_logger = MagicMock() - 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 + # Create PhysicalDevice instance with mocks + with patch('modules.device.config', {"device_cf": "zabbix_hostid"}): + self.device = PhysicalDevice( + self.mock_nb_device, + self.mock_zabbix, + self.mock_nb_journal, + "3.0", + journal=True, + logger=self.mock_logger + ) - mock.custom_fields = {"zabbix_hostid": 1956} - return mock + def test_cleanup_successful_deletion(self): + """Test successful device deletion from Zabbix.""" + # Setup + self.mock_zabbix.host.get.return_value = [{"hostid": "456"}] + self.mock_zabbix.host.delete.return_value = {"hostids": ["456"]} + # Execute + self.device.cleanup() -def mock_zabbix(): - """Function to mock Zabbix""" - mock = MagicMock() - mock.host.get.return_value = [{}] - mock.host.delete.return_value = True - return mock + # Verify + self.mock_zabbix.host.get.assert_called_once_with(filter={'hostid': '456'}, output=[]) + self.mock_zabbix.host.delete.assert_called_once_with('456') + self.mock_nb_device.save.assert_called_once() + self.assertIsNone(self.mock_nb_device.custom_fields["zabbix_hostid"]) + self.mock_logger.info.assert_called_with(f"Host {self.device.name}: " + "Deleted host from Zabbix.") + def test_cleanup_device_already_deleted(self): + """Test cleanup when device is already deleted from Zabbix.""" + # Setup + self.mock_zabbix.host.get.return_value = [] # Empty list means host not found -netbox_journals = MagicMock() -NB_VERSION = '4.2' -create_journal = MagicMock() -logger = MagicMock() + # Execute + self.device.cleanup() + # Verify + self.mock_zabbix.host.get.assert_called_once_with(filter={'hostid': '456'}, output=[]) + self.mock_zabbix.host.delete.assert_not_called() + self.mock_nb_device.save.assert_called_once() + self.assertIsNone(self.mock_nb_device.custom_fields["zabbix_hostid"]) + self.mock_logger.info.assert_called_with( + f"Host {self.device.name}: was already deleted from Zabbix. Removed link in NetBox.") -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() is False + def test_cleanup_api_error(self): + """Test cleanup when Zabbix API returns an error.""" + # Setup + self.mock_zabbix.host.get.return_value = [{"hostid": "456"}] + self.mock_zabbix.host.delete.side_effect = APIRequestError("API Error") + # Execute and verify + with self.assertRaises(SyncExternalError): + self.device.cleanup() -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.") + # Verify correct calls were made + self.mock_zabbix.host.get.assert_called_once_with(filter={'hostid': '456'}, output=[]) + self.mock_zabbix.host.delete.assert_called_once_with('456') + self.mock_nb_device.save.assert_not_called() + self.mock_logger.error.assert_called() + def test_zeroize_cf(self): + """Test _zeroize_cf method that clears the custom field.""" + # Execute + self.device._zeroize_cf() # pylint: disable=protected-access -def test_device_deletion_host_not_exists(): - """ - Test if device in Netbox gets unlinked - when host is not present in Zabbix - """ - nb_device = mock_nb_device() - zabbix = mock_zabbix() - zabbix.host.get.return_value = None + # Verify + self.assertIsNone(self.mock_nb_device.custom_fields["zabbix_hostid"]) + self.mock_nb_device.save.assert_called_once() - 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 - mock_journal.assert_called_once_with("warning", - "Deleted host from Zabbix") + def test_create_journal_entry(self): + """Test create_journal_entry method.""" + # Setup + test_message = "Test journal entry" + + # Execute + result = self.device.create_journal_entry("info", test_message) + + # Verify + self.assertTrue(result) + self.mock_nb_journal.create.assert_called_once() + journal_entry = self.mock_nb_journal.create.call_args[0][0] + self.assertEqual(journal_entry["assigned_object_type"], "dcim.device") + self.assertEqual(journal_entry["assigned_object_id"], 123) + self.assertEqual(journal_entry["kind"], "info") + self.assertEqual(journal_entry["comments"], test_message) + + def test_create_journal_entry_invalid_severity(self): + """Test create_journal_entry with invalid severity.""" + # Execute + result = self.device.create_journal_entry("invalid", "Test message") + + # Verify + self.assertFalse(result) + self.mock_nb_journal.create.assert_not_called() + self.mock_logger.warning.assert_called() + + def test_create_journal_entry_when_disabled(self): + """Test create_journal_entry when journaling is disabled.""" + # Setup - create device with journal=False + with patch('modules.device.config', {"device_cf": "zabbix_hostid"}): + device = PhysicalDevice( + self.mock_nb_device, + self.mock_zabbix, + self.mock_nb_journal, + "3.0", + journal=False, # Disable journaling + logger=self.mock_logger + ) + + # Execute + result = device.create_journal_entry("info", "Test message") + + # Verify + self.assertFalse(result) + self.mock_nb_journal.create.assert_not_called() + + def test_cleanup_updates_journal(self): + """Test that cleanup method creates a journal entry.""" + # Setup + self.mock_zabbix.host.get.return_value = [{"hostid": "456"}] + + # Execute + with patch.object(self.device, 'create_journal_entry') as mock_journal_entry: + self.device.cleanup() + + # Verify + mock_journal_entry.assert_called_once_with("warning", "Deleted host from Zabbix") diff --git a/tests/test_interface.py b/tests/test_interface.py new file mode 100644 index 0000000..ff55218 --- /dev/null +++ b/tests/test_interface.py @@ -0,0 +1,247 @@ +"""Tests for the ZabbixInterface class in the interface module.""" +import unittest +from modules.interface import ZabbixInterface +from modules.exceptions import InterfaceConfigError + + +class TestZabbixInterface(unittest.TestCase): + """Test class for ZabbixInterface functionality.""" + + def setUp(self): + """Set up test fixtures.""" + self.test_ip = "192.168.1.1" + self.empty_context = {} + self.default_interface = ZabbixInterface(self.empty_context, self.test_ip) + + # Create some test contexts for different scenarios + self.snmpv2_context = { + "zabbix": { + "interface_type": 2, + "interface_port": "161", + "snmp": { + "version": 2, + "community": "public", + "bulk": 1 + } + } + } + + self.snmpv3_context = { + "zabbix": { + "interface_type": 2, + "snmp": { + "version": 3, + "securityname": "snmpuser", + "securitylevel": "authPriv", + "authprotocol": "SHA", + "authpassphrase": "authpass123", + "privprotocol": "AES", + "privpassphrase": "privpass123", + "contextname": "context1" + } + } + } + + self.agent_context = { + "zabbix": { + "interface_type": 1, + "interface_port": "10050" + } + } + + def test_init(self): + """Test initialization of ZabbixInterface.""" + interface = ZabbixInterface(self.empty_context, self.test_ip) + + # Check basic properties + self.assertEqual(interface.ip, self.test_ip) + self.assertEqual(interface.context, self.empty_context) + self.assertEqual(interface.interface["ip"], self.test_ip) + self.assertEqual(interface.interface["main"], "1") + self.assertEqual(interface.interface["useip"], "1") + self.assertEqual(interface.interface["dns"], "") + + def test_get_context_empty(self): + """Test get_context with empty context.""" + interface = ZabbixInterface(self.empty_context, self.test_ip) + result = interface.get_context() + self.assertFalse(result) + + def test_get_context_with_interface_type(self): + """Test get_context with interface_type but no port.""" + context = {"zabbix": {"interface_type": 2}} + interface = ZabbixInterface(context, self.test_ip) + + # Should set type and default port + result = interface.get_context() + self.assertTrue(result) + self.assertEqual(interface.interface["type"], 2) + self.assertEqual(interface.interface["port"], "161") # Default port for SNMP + + def test_get_context_with_interface_type_and_port(self): + """Test get_context with both interface_type and port.""" + context = {"zabbix": {"interface_type": 1, "interface_port": "12345"}} + interface = ZabbixInterface(context, self.test_ip) + + # Should set type and specified port + result = interface.get_context() + self.assertTrue(result) + self.assertEqual(interface.interface["type"], 1) + self.assertEqual(interface.interface["port"], "12345") + + def test_set_default_port(self): + """Test _set_default_port for different interface types.""" + interface = ZabbixInterface(self.empty_context, self.test_ip) + + # Test for agent type (1) + interface.interface["type"] = 1 + interface._set_default_port() # pylint: disable=protected-access + self.assertEqual(interface.interface["port"], "10050") + + # Test for SNMP type (2) + interface.interface["type"] = 2 + interface._set_default_port() # pylint: disable=protected-access + self.assertEqual(interface.interface["port"], "161") + + # Test for IPMI type (3) + interface.interface["type"] = 3 + interface._set_default_port() # pylint: disable=protected-access + self.assertEqual(interface.interface["port"], "623") + + # Test for JMX type (4) + interface.interface["type"] = 4 + interface._set_default_port() # pylint: disable=protected-access + self.assertEqual(interface.interface["port"], "12345") + + # Test for unsupported type + interface.interface["type"] = 99 + result = interface._set_default_port() # pylint: disable=protected-access + self.assertFalse(result) + + def test_set_snmp_v2(self): + """Test set_snmp with SNMPv2 configuration.""" + interface = ZabbixInterface(self.snmpv2_context, self.test_ip) + interface.get_context() # Set the interface type + + # Call set_snmp + interface.set_snmp() + + # Check SNMP details + self.assertEqual(interface.interface["details"]["version"], "2") + self.assertEqual(interface.interface["details"]["community"], "public") + self.assertEqual(interface.interface["details"]["bulk"], "1") + + def test_set_snmp_v3(self): + """Test set_snmp with SNMPv3 configuration.""" + interface = ZabbixInterface(self.snmpv3_context, self.test_ip) + interface.get_context() # Set the interface type + + # Call set_snmp + interface.set_snmp() + + # Check SNMP details + self.assertEqual(interface.interface["details"]["version"], "3") + self.assertEqual(interface.interface["details"]["securityname"], "snmpuser") + self.assertEqual(interface.interface["details"]["securitylevel"], "authPriv") + self.assertEqual(interface.interface["details"]["authprotocol"], "SHA") + self.assertEqual(interface.interface["details"]["authpassphrase"], "authpass123") + self.assertEqual(interface.interface["details"]["privprotocol"], "AES") + self.assertEqual(interface.interface["details"]["privpassphrase"], "privpass123") + self.assertEqual(interface.interface["details"]["contextname"], "context1") + + def test_set_snmp_no_snmp_config(self): + """Test set_snmp with missing SNMP configuration.""" + # Create context with interface type but no SNMP config + context = {"zabbix": {"interface_type": 2}} + interface = ZabbixInterface(context, self.test_ip) + interface.get_context() # Set the interface type + + # Call set_snmp - should raise exception + with self.assertRaises(InterfaceConfigError): + interface.set_snmp() + + def test_set_snmp_unsupported_version(self): + """Test set_snmp with unsupported SNMP version.""" + # Create context with invalid SNMP version + context = { + "zabbix": { + "interface_type": 2, + "snmp": { + "version": 4 # Invalid version + } + } + } + interface = ZabbixInterface(context, self.test_ip) + interface.get_context() # Set the interface type + + # Call set_snmp - should raise exception + with self.assertRaises(InterfaceConfigError): + interface.set_snmp() + + def test_set_snmp_no_version(self): + """Test set_snmp with missing SNMP version.""" + # Create context without SNMP version + context = { + "zabbix": { + "interface_type": 2, + "snmp": { + "community": "public" # No version specified + } + } + } + interface = ZabbixInterface(context, self.test_ip) + interface.get_context() # Set the interface type + + # Call set_snmp - should raise exception + with self.assertRaises(InterfaceConfigError): + interface.set_snmp() + + def test_set_snmp_non_snmp_interface(self): + """Test set_snmp with non-SNMP interface type.""" + interface = ZabbixInterface(self.agent_context, self.test_ip) + interface.get_context() # Set the interface type + + # Call set_snmp - should raise exception + with self.assertRaises(InterfaceConfigError): + interface.set_snmp() + + def test_set_default_snmp(self): + """Test set_default_snmp method.""" + interface = ZabbixInterface(self.empty_context, self.test_ip) + interface.set_default_snmp() + + # Check interface properties + self.assertEqual(interface.interface["type"], "2") + self.assertEqual(interface.interface["port"], "161") + self.assertEqual(interface.interface["details"]["version"], "2") + self.assertEqual(interface.interface["details"]["community"], "{$SNMP_COMMUNITY}") + self.assertEqual(interface.interface["details"]["bulk"], "1") + + def test_set_default_agent(self): + """Test set_default_agent method.""" + interface = ZabbixInterface(self.empty_context, self.test_ip) + interface.set_default_agent() + + # Check interface properties + self.assertEqual(interface.interface["type"], "1") + self.assertEqual(interface.interface["port"], "10050") + + def test_snmpv2_no_community(self): + """Test SNMPv2 with no community string specified.""" + # Create context with SNMPv2 but no community + context = { + "zabbix": { + "interface_type": 2, + "snmp": { + "version": 2 + } + } + } + interface = ZabbixInterface(context, self.test_ip) + interface.get_context() # Set the interface type + + # Call set_snmp + interface.set_snmp() + + # Should use default community string + self.assertEqual(interface.interface["details"]["community"], "{$SNMP_COMMUNITY}") diff --git a/tests/test_physical_device.py b/tests/test_physical_device.py new file mode 100644 index 0000000..4fe8ce3 --- /dev/null +++ b/tests/test_physical_device.py @@ -0,0 +1,373 @@ +"""Tests for the PhysicalDevice class in the device module.""" +import unittest +from unittest.mock import MagicMock, patch +from modules.device import PhysicalDevice +from modules.exceptions import TemplateError, SyncInventoryError + + +class TestPhysicalDevice(unittest.TestCase): + """Test class for PhysicalDevice functionality.""" + + def setUp(self): + """Set up test fixtures.""" + # Create mock NetBox device + self.mock_nb_device = MagicMock() + self.mock_nb_device.id = 123 + self.mock_nb_device.name = "test-device" + self.mock_nb_device.status.label = "Active" + self.mock_nb_device.custom_fields = {"zabbix_hostid": None} + self.mock_nb_device.config_context = {} + + # Set up a primary IP + primary_ip = MagicMock() + primary_ip.address = "192.168.1.1/24" + self.mock_nb_device.primary_ip = primary_ip + + # Create mock Zabbix API + self.mock_zabbix = MagicMock() + self.mock_zabbix.version = "6.0" + + # Mock NetBox journal class + self.mock_nb_journal = MagicMock() + + # Create logger mock + self.mock_logger = MagicMock() + + # Create PhysicalDevice instance with mocks + with patch('modules.device.config', + {"device_cf": "zabbix_hostid", + "template_cf": "zabbix_template", + "templates_config_context": False, + "templates_config_context_overrule": False, + "traverse_regions": False, + "traverse_site_groups": False, + "inventory_mode": "disabled", + "inventory_sync": False, + "inventory_map": {} + }): + self.device = PhysicalDevice( + self.mock_nb_device, + self.mock_zabbix, + self.mock_nb_journal, + "3.0", + journal=True, + logger=self.mock_logger + ) + + def test_init(self): + """Test the initialization of the PhysicalDevice class.""" + # Check that basic properties are set correctly + self.assertEqual(self.device.name, "test-device") + self.assertEqual(self.device.id, 123) + self.assertEqual(self.device.status, "Active") + self.assertEqual(self.device.ip, "192.168.1.1") + self.assertEqual(self.device.cidr, "192.168.1.1/24") + + def test_init_no_primary_ip(self): + """Test initialization when device has no primary IP.""" + # Set primary_ip to None + self.mock_nb_device.primary_ip = None + + # Creating device should raise SyncInventoryError + with patch('modules.device.config', {"device_cf": "zabbix_hostid"}): + with self.assertRaises(SyncInventoryError): + PhysicalDevice( + self.mock_nb_device, + self.mock_zabbix, + self.mock_nb_journal, + "3.0", + logger=self.mock_logger + ) + + def test_set_basics_with_special_characters(self): + """Test _setBasics when device name contains special characters.""" + # Set name with special characters that + # will actually trigger the special character detection + self.mock_nb_device.name = "test-devïce" + + # We need to patch the search function to simulate finding special characters + with patch('modules.device.search') as mock_search, \ + patch('modules.device.config', {"device_cf": "zabbix_hostid"}): + # Make the search function return True to simulate special characters + mock_search.return_value = True + + device = PhysicalDevice( + self.mock_nb_device, + self.mock_zabbix, + self.mock_nb_journal, + "3.0", + logger=self.mock_logger + ) + + # With the mocked search function, the name should be changed to NETBOX_ID format + self.assertEqual(device.name, f"NETBOX_ID{self.mock_nb_device.id}") + # And visible_name should be set to the original name + self.assertEqual(device.visible_name, "test-devïce") + # use_visible_name flag should be set + self.assertTrue(device.use_visible_name) + + def test_get_templates_context(self): + """Test get_templates_context with valid config.""" + # Set up config_context with valid template data + self.mock_nb_device.config_context = { + "zabbix": { + "templates": ["Template1", "Template2"] + } + } + + # Create device with the updated mock + with patch('modules.device.config', {"device_cf": "zabbix_hostid"}): + device = PhysicalDevice( + self.mock_nb_device, + self.mock_zabbix, + self.mock_nb_journal, + "3.0", + logger=self.mock_logger + ) + + # Test that templates are returned correctly + templates = device.get_templates_context() + self.assertEqual(templates, ["Template1", "Template2"]) + + def test_get_templates_context_with_string(self): + """Test get_templates_context with a string instead of list.""" + # Set up config_context with a string template + self.mock_nb_device.config_context = { + "zabbix": { + "templates": "Template1" + } + } + + # Create device with the updated mock + with patch('modules.device.config', {"device_cf": "zabbix_hostid"}): + device = PhysicalDevice( + self.mock_nb_device, + self.mock_zabbix, + self.mock_nb_journal, + "3.0", + logger=self.mock_logger + ) + + # Test that template is wrapped in a list + templates = device.get_templates_context() + self.assertEqual(templates, ["Template1"]) + + def test_get_templates_context_no_zabbix_key(self): + """Test get_templates_context when zabbix key is missing.""" + # Set up config_context without zabbix key + self.mock_nb_device.config_context = {} + + # Create device with the updated mock + with patch('modules.device.config', {"device_cf": "zabbix_hostid"}): + device = PhysicalDevice( + self.mock_nb_device, + self.mock_zabbix, + self.mock_nb_journal, + "3.0", + logger=self.mock_logger + ) + + # Test that TemplateError is raised + with self.assertRaises(TemplateError): + device.get_templates_context() + + def test_get_templates_context_no_templates_key(self): + """Test get_templates_context when templates key is missing.""" + # Set up config_context without templates key + self.mock_nb_device.config_context = {"zabbix": {}} + + # Create device with the updated mock + with patch('modules.device.config', {"device_cf": "zabbix_hostid"}): + device = PhysicalDevice( + self.mock_nb_device, + self.mock_zabbix, + self.mock_nb_journal, + "3.0", + logger=self.mock_logger + ) + + # Test that TemplateError is raised + with self.assertRaises(TemplateError): + device.get_templates_context() + + def test_set_template_with_config_context(self): + """Test set_template with templates_config_context=True.""" + # Set up config_context with templates + self.mock_nb_device.config_context = { + "zabbix": { + "templates": ["Template1"] + } + } + + # Mock get_templates_context to return expected templates + with patch.object(PhysicalDevice, 'get_templates_context', return_value=["Template1"]): + with patch('modules.device.config', {"device_cf": "zabbix_hostid"}): + device = PhysicalDevice( + self.mock_nb_device, + self.mock_zabbix, + self.mock_nb_journal, + "3.0", + logger=self.mock_logger + ) + + # Call set_template with prefer_config_context=True + result = device.set_template(prefer_config_context=True, overrule_custom=False) + + # Check result and template names + self.assertTrue(result) + self.assertEqual(device.zbx_template_names, ["Template1"]) + + def test_set_inventory_disabled_mode(self): + """Test set_inventory with inventory_mode=disabled.""" + # Configure with disabled inventory mode + config_patch = { + "device_cf": "zabbix_hostid", + "inventory_mode": "disabled", + "inventory_sync": False + } + + with patch('modules.device.config', config_patch): + device = PhysicalDevice( + self.mock_nb_device, + self.mock_zabbix, + self.mock_nb_journal, + "3.0", + logger=self.mock_logger + ) + + # Call set_inventory with the config patch still active + with patch('modules.device.config', config_patch): + result = device.set_inventory({}) + + # Check result + self.assertTrue(result) + # Default value for disabled inventory + self.assertEqual(device.inventory_mode, -1) + + def test_set_inventory_manual_mode(self): + """Test set_inventory with inventory_mode=manual.""" + # Configure with manual inventory mode + config_patch = { + "device_cf": "zabbix_hostid", + "inventory_mode": "manual", + "inventory_sync": False + } + + with patch('modules.device.config', config_patch): + device = PhysicalDevice( + self.mock_nb_device, + self.mock_zabbix, + self.mock_nb_journal, + "3.0", + logger=self.mock_logger + ) + + # Call set_inventory with the config patch still active + with patch('modules.device.config', config_patch): + result = device.set_inventory({}) + + # Check result + self.assertTrue(result) + self.assertEqual(device.inventory_mode, 0) # Manual mode + + def test_set_inventory_automatic_mode(self): + """Test set_inventory with inventory_mode=automatic.""" + # Configure with automatic inventory mode + config_patch = { + "device_cf": "zabbix_hostid", + "inventory_mode": "automatic", + "inventory_sync": False + } + + with patch('modules.device.config', config_patch): + device = PhysicalDevice( + self.mock_nb_device, + self.mock_zabbix, + self.mock_nb_journal, + "3.0", + logger=self.mock_logger + ) + + # Call set_inventory with the config patch still active + with patch('modules.device.config', config_patch): + result = device.set_inventory({}) + + # Check result + self.assertTrue(result) + self.assertEqual(device.inventory_mode, 1) # Automatic mode + + def test_set_inventory_with_inventory_sync(self): + """Test set_inventory with inventory_sync=True.""" + # Configure with inventory sync enabled + config_patch = { + "device_cf": "zabbix_hostid", + "inventory_mode": "manual", + "inventory_sync": True, + "inventory_map": { + "name": "name", + "serial": "serialno_a" + } + } + + with patch('modules.device.config', config_patch): + device = PhysicalDevice( + self.mock_nb_device, + self.mock_zabbix, + self.mock_nb_journal, + "3.0", + logger=self.mock_logger + ) + + # Create a mock device with the required attributes + mock_device_data = { + "name": "test-device", + "serial": "ABC123" + } + + # Call set_inventory with the config patch still active + with patch('modules.device.config', config_patch): + result = device.set_inventory(mock_device_data) + + # Check result + self.assertTrue(result) + self.assertEqual(device.inventory_mode, 0) # Manual mode + self.assertEqual(device.inventory, { + "name": "test-device", + "serialno_a": "ABC123" + }) + + def test_iscluster_true(self): + """Test isCluster when device is part of a cluster.""" + # Set up virtual_chassis + self.mock_nb_device.virtual_chassis = MagicMock() + + # Create device with the updated mock + with patch('modules.device.config', {"device_cf": "zabbix_hostid"}): + device = PhysicalDevice( + self.mock_nb_device, + self.mock_zabbix, + self.mock_nb_journal, + "3.0", + logger=self.mock_logger + ) + + # Check isCluster result + self.assertTrue(device.isCluster()) + + def test_is_cluster_false(self): + """Test isCluster when device is not part of a cluster.""" + # Set virtual_chassis to None + self.mock_nb_device.virtual_chassis = None + + # Create device with the updated mock + with patch('modules.device.config', {"device_cf": "zabbix_hostid"}): + device = PhysicalDevice( + self.mock_nb_device, + self.mock_zabbix, + self.mock_nb_journal, + "3.0", + logger=self.mock_logger + ) + + # Check isCluster result + self.assertFalse(device.isCluster())