From b331561294efd379bc9ff45f559a7e0735d60f54 Mon Sep 17 00:00:00 2001
From: Fahrudin Halilovic <fahrudin.halilovic@ess.eu>
Date: Tue, 8 Oct 2024 14:00:13 +0200
Subject: [PATCH] add tests for synchronization - INFRA-10732
---
netbox_awx_plugin/synchronization.py | 10 +-
.../tests/test_synchronization.py | 300 ++++++++++++++++++
2 files changed, 305 insertions(+), 5 deletions(-)
create mode 100644 netbox_awx_plugin/tests/test_synchronization.py
diff --git a/netbox_awx_plugin/synchronization.py b/netbox_awx_plugin/synchronization.py
index 17a72a0..4cacfe0 100644
--- a/netbox_awx_plugin/synchronization.py
+++ b/netbox_awx_plugin/synchronization.py
@@ -92,7 +92,7 @@ class DeviceSerializer(serializers.BaseSerializer):
serializer = InterfaceSerializer(interface)
variables["netbox_interfaces"].append(serializer.data)
return {
- "name": instance.primary_ip4.dns_name,
+ "name": getattr(instance.primary_ip4, 'dns_name', instance.name),
"description": instance.description,
"enabled": instance.status == DeviceStatusChoices.STATUS_ACTIVE,
"variables": json.dumps(variables),
@@ -116,16 +116,16 @@ class VMSerializer(serializers.BaseSerializer):
def to_representation(self, instance):
variables = {
"netbox_virtualmachine_name": instance.name,
- "netbox_virtualmachine_vcpus": float(instance.vcpus),
- "netbox_virtualmachine_memory": instance.memory,
- "netbox_virtualmachine_disk": instance.disk,
+ "netbox_virtualmachine_vcpus": float(instance.vcpus) if instance.vcpus is not None else 0.0,
+ "netbox_virtualmachine_memory": instance.memory or 0,
+ "netbox_virtualmachine_disk": instance.disk or 0,
}
variables["netbox_interfaces"] = []
for interface in instance.interfaces.all():
serializer = VMInterfaceSerializer(interface)
variables["netbox_interfaces"].append(serializer.data)
return {
- "name": instance.primary_ip4.dns_name,
+ "name": getattr(instance.primary_ip4, 'dns_name', instance.name),
"description": instance.description,
"enabled": instance.status == VirtualMachineStatusChoices.STATUS_ACTIVE,
"variables": json.dumps(variables),
diff --git a/netbox_awx_plugin/tests/test_synchronization.py b/netbox_awx_plugin/tests/test_synchronization.py
new file mode 100644
index 0000000..8688323
--- /dev/null
+++ b/netbox_awx_plugin/tests/test_synchronization.py
@@ -0,0 +1,300 @@
+# tests/test_synchronization.py
+
+from django.test import TestCase
+from unittest.mock import patch, Mock, ANY
+from netbox_awx_plugin.models import AWX, AWXInventory
+from dcim.models import Device, DeviceRole, DeviceType, Manufacturer, Site
+from virtualization.models import VirtualMachine
+from ipam.models import IPAddress
+from extras.models import Tag
+from netbox_awx_plugin.synchronization import (
+ sync_host,
+ sync_group,
+ delete_host,
+ delete_group,
+ sync_all,
+ sync_host_group_association,
+ disassociate_removed_groups,
+)
+from netbox_awx_plugin.synchronization import serializers, group_prefixes
+from django.contrib.contenttypes.models import ContentType
+
+
+class SynchronizationTestCase(TestCase):
+
+ def setUp(self):
+ # Create an AWX instance and inventory
+ self.awx = AWX.objects.create(
+ name='Test AWX',
+ url='https://awx.example.com',
+ token='token123'
+ )
+ self.awx_inventory = AWXInventory.objects.create(
+ awx=self.awx,
+ inventory_id=1,
+ enabled=True,
+ )
+ # Create related objects
+ self.manufacturer = Manufacturer.objects.create(
+ name='Test Manufacturer',
+ slug='test-manufacturer'
+ )
+ self.device_role = DeviceRole.objects.create(name='Switch', slug='switch')
+ self.device_type = DeviceType.objects.create(
+ model='TestModel',
+ slug='testmodel',
+ manufacturer=self.manufacturer,
+ )
+ self.site = Site.objects.create(name='Test Site', slug='test-site', status='active')
+ self.device = Device.objects.create(
+ name='Test Device',
+ device_role=self.device_role,
+ device_type=self.device_type,
+ site=self.site,
+ status='active',
+ )
+ self.ip_address = IPAddress.objects.create(
+ address='192.0.2.1/24',
+ dns_name='test-device.example.com'
+ )
+ self.device.primary_ip4 = self.ip_address
+ self.device.save()
+
+ @patch('netbox_awx_plugin.models.AWXInventory.disassociate_host_group')
+ @patch('netbox_awx_plugin.models.AWXInventory.associate_host_group')
+ @patch('netbox_awx_plugin.models.AWXInventory.get_group')
+ @patch('netbox_awx_plugin.models.AWXInventory.get_host')
+ @patch('netbox_awx_plugin.models.AWXInventory.create_host')
+ @patch('netbox_awx_plugin.models.AWXInventory.update_host')
+ def test_sync_host_create(self, mock_update_host, mock_create_host, mock_get_host, mock_get_group, mock_associate_host_group, mock_disassociate_host_group):
+ # Simulate that the host does not exist on first call, but exists after creation
+ mock_get_host.side_effect = [
+ None, # First call returns None
+ {
+ 'id': 1,
+ 'name': 'test-device.example.com',
+ 'summary_fields': {'groups': {'results': []}}
+ } # Second call returns a mock host
+ ]
+ # Simulate that the group exists
+ mock_get_group.return_value = {'id': 2, 'name': 'site_test_site'}
+ sync_host(self.awx_inventory, Device, self.device)
+ mock_create_host.assert_called()
+ mock_update_host.assert_not_called()
+ mock_associate_host_group.assert_called_with(1, 2)
+
+ @patch('netbox_awx_plugin.models.AWXInventory.disassociate_host_group')
+ @patch('netbox_awx_plugin.models.AWXInventory.associate_host_group')
+ @patch('netbox_awx_plugin.models.AWXInventory.get_group')
+ @patch('netbox_awx_plugin.models.AWXInventory.get_host')
+ @patch('netbox_awx_plugin.models.AWXInventory.create_host')
+ @patch('netbox_awx_plugin.models.AWXInventory.update_host')
+ def test_sync_host_update(self, mock_update_host, mock_create_host, mock_get_host, mock_get_group, mock_associate_host_group, mock_disassociate_host_group):
+ # Simulate that the host exists
+ mock_get_host.return_value = {
+ 'id': 1,
+ 'name': 'test-device.example.com',
+ 'summary_fields': {'groups': {'results': []}}
+ }
+ # Simulate that the group exists
+ mock_get_group.return_value = {'id': 2, 'name': 'site_test_site'}
+ sync_host(self.awx_inventory, Device, self.device)
+ mock_update_host.assert_called()
+ mock_create_host.assert_not_called()
+ mock_associate_host_group.assert_called_with(1, 2)
+
+ @patch('netbox_awx_plugin.models.AWXInventory.delete_host')
+ def test_delete_host(self, mock_delete_host):
+ delete_host(self.awx_inventory, Device, self.device)
+ mock_delete_host.assert_called_with('test-device.example.com')
+
+ @patch('netbox_awx_plugin.models.AWXInventory.get_group')
+ @patch('netbox_awx_plugin.models.AWXInventory.create_group')
+ @patch('netbox_awx_plugin.models.AWXInventory.update_group')
+ def test_sync_group_create(self, mock_update_group, mock_create_group, mock_get_group):
+ # Simulate that the group does not exist
+ mock_get_group.return_value = None
+ sync_group(self.awx_inventory, Site, self.site)
+ mock_create_group.assert_called()
+ mock_update_group.assert_not_called()
+
+ @patch('netbox_awx_plugin.models.AWXInventory.get_group')
+ @patch('netbox_awx_plugin.models.AWXInventory.create_group')
+ @patch('netbox_awx_plugin.models.AWXInventory.update_group')
+ def test_sync_group_update(self, mock_update_group, mock_create_group, mock_get_group):
+ # Simulate that the group exists but needs updating
+ mock_get_group.return_value = {'id': 1, 'name': 'site_test_site'}
+ sync_group(self.awx_inventory, Site, self.site)
+ mock_update_group.assert_called()
+ mock_create_group.assert_not_called()
+
+ @patch('netbox_awx_plugin.models.AWXInventory.delete_group')
+ def test_delete_group(self, mock_delete_group):
+ delete_group(self.awx_inventory, Site, self.site)
+ mock_delete_group.assert_called_with('site_test_site')
+
+ # Additional Tests
+
+ @patch('netbox_awx_plugin.models.AWXInventory.update_host')
+ @patch('netbox_awx_plugin.models.AWXInventory.create_host')
+ @patch('netbox_awx_plugin.models.AWXInventory.get_host')
+ @patch('netbox_awx_plugin.models.AWXInventory.associate_host_group')
+ @patch('netbox_awx_plugin.models.AWXInventory.get_group')
+ def test_sync_host_with_tags(self, mock_get_group, mock_associate_host_group, mock_get_host, mock_create_host, mock_update_host):
+ # Add a tag to the device
+ tag = Tag.objects.create(name='Test Tag', slug='test-tag')
+ self.device.tags.add(tag)
+ self.device.save()
+ # Simulate that the group for the tag exists
+ mock_get_group.return_value = {'id': 3, 'name': 'tag_test_tag'}
+ # Simulate that the host exists
+ mock_get_host.return_value = {
+ 'id': 1,
+ 'name': 'test-device.example.com',
+ 'summary_fields': {'groups': {'results': []}}
+ }
+ sync_host(self.awx_inventory, Device, self.device)
+ # Check that the host is associated with the tag group
+ mock_associate_host_group.assert_any_call(1, 3)
+
+ @patch('netbox_awx_plugin.models.AWXInventory.disassociate_host_group')
+ @patch('netbox_awx_plugin.models.AWXInventory.get_host')
+ @patch('netbox_awx_plugin.models.AWXInventory.get_group')
+ def test_disassociate_removed_groups(self, mock_get_group, mock_get_host, mock_disassociate_host_group):
+ # Simulate that the host is associated with groups that are no longer valid
+ mock_get_host.return_value = {
+ 'id': 1,
+ 'name': 'test-device.example.com',
+ 'summary_fields': {
+ 'groups': {
+ 'results': [
+ {'id': 2, 'name': 'site_old_site'},
+ {'id': 3, 'name': 'devicerole_old_role'},
+ {'id': 4, 'name': 'tag_old_tag'},
+ ]
+ }
+ }
+ }
+ # Simulate that the group exists for the current site
+ mock_get_group.return_value = {'id': 5, 'name': 'site_test_site'}
+ # Assume the device only has site_test_site group now
+ sync_host(self.awx_inventory, Device, self.device)
+ # Check that the host is disassociated from 'site_old_site', 'devicerole_old_role', and 'tag_old_tag'
+ mock_disassociate_host_group.assert_any_call(1, 2)
+ mock_disassociate_host_group.assert_any_call(1, 3)
+ mock_disassociate_host_group.assert_any_call(1, 4)
+
+ @patch('netbox_awx_plugin.models.AWXInventory.create_host')
+ @patch('netbox_awx_plugin.models.AWXInventory.get_host')
+ @patch('netbox_awx_plugin.models.AWXInventory.associate_host_group')
+ @patch('netbox_awx_plugin.models.AWXInventory.get_group')
+ def test_sync_virtual_machine(self, mock_get_group, mock_associate_host_group, mock_get_host, mock_create_host):
+ # Create a role for the virtual machine
+ vm_content_type = ContentType.objects.get_for_model(VirtualMachine)
+ vm_role = DeviceRole.objects.create(
+ name='Web Server',
+ slug='web-server'
+ )
+ # Create a virtual machine with a primary IP and a role
+ vm = VirtualMachine.objects.create(
+ name='Test VM',
+ status='active',
+ role=vm_role
+ )
+ ip_address = IPAddress.objects.create(
+ address='192.0.2.2/24',
+ dns_name='test-vm.example.com'
+ )
+ vm.primary_ip4 = ip_address
+ vm.save()
+ # Simulate that the host does not exist on first call, but exists after creation
+ mock_get_host.side_effect = [
+ None, # First call returns None
+ {
+ 'id': 1,
+ 'name': 'test-vm.example.com',
+ 'summary_fields': {'groups': {'results': []}}
+ } # Second call returns a mock host
+ ]
+ # Simulate that the group exists for the VM role
+ mock_get_group.return_value = {'id': 2, 'name': 'devicerole_web_server'}
+ # Run the sync_host function
+ sync_host(self.awx_inventory, VirtualMachine, vm)
+ # Ensure that create_host was called
+ mock_create_host.assert_called_with(ANY)
+ # Ensure that the host is associated with the group
+ mock_associate_host_group.assert_called_with(1, 2)
+
+ @patch('netbox_awx_plugin.models.AWXInventory.update_host')
+ @patch('netbox_awx_plugin.models.AWXInventory.get_host')
+ @patch('netbox_awx_plugin.models.AWXInventory.get_group')
+ @patch('netbox_awx_plugin.models.AWXInventory.associate_host_group')
+ @patch('netbox_awx_plugin.models.AWXInventory.disassociate_host_group')
+ def test_sync_host_missing_primary_ip(self, mock_disassociate_host_group, mock_associate_host_group, mock_get_group, mock_get_host, mock_update_host):
+ # Remove the primary IP from the device
+ self.device.primary_ip4 = None
+ self.device.save()
+ # Mock get_host to return a mock host
+ mock_get_host.return_value = {
+ 'id': 1,
+ 'name': self.device.name,
+ 'summary_fields': {'groups': {'results': []}}
+ }
+ # Mock get_group to prevent network calls
+ mock_get_group.return_value = {'id': 2, 'name': 'site_test_site'}
+ # Run the sync_host function
+ sync_host(self.awx_inventory, Device, self.device)
+ # Ensure that update_host was called
+ mock_update_host.assert_called_with(1, ANY)
+ # Ensure that associate_host_group was called
+ mock_associate_host_group.assert_called_with(1, 2)
+
+ @patch('netbox_awx_plugin.synchronization.sync_host')
+ @patch('netbox_awx_plugin.synchronization.sync_group')
+ def test_sync_all(self, mock_sync_group, mock_sync_host):
+ job = Mock()
+ job.object = self.awx_inventory
+ sync_all(job)
+ # Ensure that sync_host and sync_group are called appropriately
+ self.assertTrue(mock_sync_group.called)
+ self.assertTrue(mock_sync_host.called)
+
+ # Additional tests for helper functions
+ def test_sync_host_group_association(self):
+ # Create a mock inventory
+ inventory = Mock()
+ # Create a mock host
+ host = {'id': 1, 'name': 'test-device.example.com', 'summary_fields': {'groups': {'results': []}}}
+ # Create a mock group
+ group = {'id': 2, 'name': 'site_test_site'}
+ inventory.get_group.return_value = group
+ # Call the function
+ instance = self.device.site
+ sync_host_group_association(inventory, host, Site, instance, host['summary_fields']['groups']['results'])
+ # Check that associate_host_group was called
+ inventory.associate_host_group.assert_called_with(1, 2)
+
+ def test_disassociate_removed_groups(self):
+ # Create a mock inventory
+ inventory = Mock()
+ # Create a mock host
+ host = {
+ 'id': 1,
+ 'name': 'test-device.example.com',
+ 'summary_fields': {
+ 'groups': {
+ 'results': [
+ {'id': 2, 'name': 'site_old_site'},
+ {'id': 3, 'name': 'devicerole_old_role'},
+ {'id': 4, 'name': 'tag_old_tag'},
+ ]
+ }
+ }
+ }
+ # Create an instance with current attributes
+ self.device.tags.clear()
+ disassociate_removed_groups(inventory, host, self.device, host['summary_fields']['groups']['results'])
+ # Check that disassociate_host_group was called for each group
+ inventory.disassociate_host_group.assert_any_call(1, 2)
+ inventory.disassociate_host_group.assert_any_call(1, 3)
+ inventory.disassociate_host_group.assert_any_call(1, 4)
--
GitLab