diff --git a/netbox_awx_plugin/synchronization.py b/netbox_awx_plugin/synchronization.py
index 17a72a00f589f49018d1d7849f6c1af148fcb1db..4cacfe0c7a00a968c46f5937da7d3204387dee4b 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_signals.py b/netbox_awx_plugin/tests/test_signals.py
new file mode 100644
index 0000000000000000000000000000000000000000..b99a41edf6acdbd710b70cfed43ab9d9fd0a4040
--- /dev/null
+++ b/netbox_awx_plugin/tests/test_signals.py
@@ -0,0 +1,341 @@
+from django.test import TestCase
+from unittest.mock import patch, call
+
+from dcim.models import Site, Device, DeviceRole, DeviceType, Manufacturer, Interface
+from ipam.models import Prefix, IPAddress
+from virtualization.models import VirtualMachine
+from extras.models import Tag
+from netbox_awx_plugin.models import AWX, AWXInventory
+from netbox_awx_plugin.signals import enqueue_task
+from django.db.models.signals import pre_save
+
+class SignalsTestCase(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 a manufacturer for device types
+ self.manufacturer = Manufacturer.objects.create(
+ name='Test Manufacturer',
+ slug='test-manufacturer'
+ )
+
+ @patch('netbox_awx_plugin.signals.enqueue_task')
+ def test_site_post_save_signal(self, mock_enqueue_task):
+ # Create a Site instance
+ site = Site.objects.create(
+ name='Test Site',
+ slug='test-site',
+ status='active',
+ )
+ # Check that enqueue_task was called
+ mock_enqueue_task.assert_called_with('sync_group', self.awx_inventory, Site, site)
+
+
+ @patch('netbox_awx_plugin.signals.enqueue_task')
+ def test_device_role_post_save_signal(self, mock_enqueue_task):
+ # Create a DeviceRole instance
+ device_role = DeviceRole.objects.create(name='Router', slug='router')
+ # Check that enqueue_task was called
+ mock_enqueue_task.assert_called_with('sync_group', self.awx_inventory, DeviceRole, device_role)
+
+
+ @patch('netbox_awx_plugin.signals.enqueue_task')
+ def test_device_type_post_save_signal(self, mock_enqueue_task):
+ # Create a DeviceType instance
+ device_type = DeviceType.objects.create(
+ model='TestModel',
+ slug='testmodel',
+ manufacturer=self.manufacturer,
+ )
+ # Check that enqueue_task was called
+ mock_enqueue_task.assert_called_with('sync_group', self.awx_inventory, DeviceType, device_type)
+
+
+ @patch('netbox_awx_plugin.signals.enqueue_task')
+ def test_prefix_post_save_signal(self, mock_enqueue_task):
+ # Create a Prefix instance
+ prefix = Prefix.objects.create(prefix='192.0.2.0/24')
+ # Check that enqueue_task was called
+ mock_enqueue_task.assert_called_with('sync_group', self.awx_inventory, Prefix, prefix)
+
+
+ @patch('netbox_awx_plugin.signals.enqueue_task')
+ def test_tag_post_save_signal(self, mock_enqueue_task):
+ # Create a Tag instance
+ tag = Tag.objects.create(name='Test Tag', slug='test-tag')
+ # Check that enqueue_task was called
+ mock_enqueue_task.assert_called_with('sync_group', self.awx_inventory, Tag, tag)
+
+ @patch('netbox_awx_plugin.signals.enqueue_task')
+ def test_device_post_save_signal(self, mock_enqueue_task):
+ # Create related objects
+ device_role = DeviceRole.objects.create(name='Switch', slug='switch')
+ device_type = DeviceType.objects.create(
+ model='TestModel',
+ slug='testmodel',
+ manufacturer=self.manufacturer,
+ )
+ site = Site.objects.create(name='Test Site', slug='test-site', status='active')
+
+ # Reset the mock after creating related objects
+ mock_enqueue_task.reset_mock()
+
+ # Create a Device without primary IP
+ device = Device.objects.create(
+ name='Test Device',
+ device_role=device_role,
+ device_type=device_type,
+ site=site,
+ status='active',
+ )
+ # Ensure enqueue_task was not called yet
+ mock_enqueue_task.assert_not_called()
+
+ # Assign primary IP with DNS name
+ ip_address = IPAddress.objects.create(
+ address='192.0.2.1/24',
+ dns_name='test-device.example.com'
+ )
+ device.primary_ip4 = ip_address
+ device.save()
+
+ # Now enqueue_task should be called
+ mock_enqueue_task.assert_called_with('sync_host', self.awx_inventory, Device, device)
+
+ @patch('netbox_awx_plugin.signals.enqueue_task')
+ def test_device_post_save_no_ip_dns(self, mock_enqueue_task):
+ # Create related objects
+ device_role = DeviceRole.objects.create(name='Switch', slug='switch')
+ device_type = DeviceType.objects.create(
+ model='TestModel',
+ slug='testmodel',
+ manufacturer=self.manufacturer,
+ )
+ site = Site.objects.create(name='Test Site', slug='test-site', status='active')
+
+ # Reset the mock after creating related objects
+ mock_enqueue_task.reset_mock()
+
+ # Create a Device without primary IP and DNS name
+ device = Device.objects.create(
+ name='Test Device',
+ device_role=device_role,
+ device_type=device_type,
+ site=site,
+ status='active',
+ )
+ # Ensure enqueue_task was not called
+ mock_enqueue_task.assert_not_called()
+
+ @patch('netbox_awx_plugin.signals.enqueue_task')
+ def test_tag_assignment(self, mock_enqueue_task):
+ # Create a Tag and assign it to a Device
+ tag = Tag.objects.create(name='Test Tag', slug='test-tag')
+ device_role = DeviceRole.objects.create(name='Switch', slug='switch')
+ device_type = DeviceType.objects.create(
+ model='TestModel',
+ slug='testmodel',
+ manufacturer=self.manufacturer,
+ )
+ site = Site.objects.create(name='Test Site', slug='test-site', status='active')
+ device = Device.objects.create(
+ name='Test Device',
+ device_role=device_role,
+ device_type=device_type,
+ site=site,
+ status='active',
+ )
+
+ # Reset the mock after creating related objects
+ mock_enqueue_task.reset_mock()
+
+ device.tags.add(tag)
+ device.save()
+
+ # enqueue_task should be called due to tag assignment
+ mock_enqueue_task.assert_called_with('sync_host', self.awx_inventory, Device, device)
+
+
+ @patch('netbox_awx_plugin.signals.enqueue_task')
+ def test_site_post_delete_signal(self, mock_enqueue_task):
+ # Create a Site instance
+ site = Site.objects.create(
+ name='Test Site',
+ slug='test-site',
+ status='active',
+ )
+ site.delete()
+ # Check that enqueue_task was called
+ mock_enqueue_task.assert_called_with('delete_group', self.awx_inventory, Site, site)
+
+
+ @patch('netbox_awx_plugin.signals.enqueue_task')
+ def test_device_role_post_delete_signal(self, mock_enqueue_task):
+ # Create a DeviceRole instance
+ device_role = DeviceRole.objects.create(name='Router', slug='router')
+ device_role.delete()
+ # Check that enqueue_task was called
+ mock_enqueue_task.assert_called_with('delete_group', self.awx_inventory, DeviceRole, device_role)
+
+
+ @patch('netbox_awx_plugin.signals.enqueue_task')
+ def test_device_type_post_delete_signal(self, mock_enqueue_task):
+ # Create a DeviceType instance
+ device_type = DeviceType.objects.create(
+ model='TestModel',
+ slug='testmodel',
+ manufacturer=self.manufacturer,
+ )
+ device_type.delete()
+ # Check that enqueue_task was called
+ mock_enqueue_task.assert_called_with('delete_group', self.awx_inventory, DeviceType, device_type)
+
+
+ @patch('netbox_awx_plugin.signals.enqueue_task')
+ def test_prefix_post_delete_signal(self, mock_enqueue_task):
+ # Create a Prefix instance
+ prefix = Prefix.objects.create(prefix='192.0.2.0/24')
+ prefix.delete()
+ # Check that enqueue_task was called
+ mock_enqueue_task.assert_called_with('delete_group', self.awx_inventory, Prefix, prefix)
+
+
+ @patch('netbox_awx_plugin.signals.enqueue_task')
+ def test_tag_post_delete_signal(self, mock_enqueue_task):
+ # Create a Tag instance
+ tag = Tag.objects.create(name='Test Tag', slug='test-tag')
+ tag.delete()
+ # Check that enqueue_task was called
+ mock_enqueue_task.assert_called_with('delete_group', self.awx_inventory, Tag, tag)
+
+
+ def test_handle_group_pre_save(self):
+ # Create a Site instance
+ site = Site.objects.create(
+ name='Original Name',
+ slug='original-slug',
+ status='active',
+ )
+ site.name = 'Updated Name'
+ # Manually send pre_save signal
+ pre_save.send(sender=Site, instance=site)
+ # Check that the original instance is stored
+ self.assertEqual(site._original.name, 'Original Name')
+
+
+ @patch('netbox_awx_plugin.signals.enqueue_task')
+ def test_virtual_machine_post_save_signal(self, mock_enqueue_task):
+ # Create a VirtualMachine instance
+ vm = VirtualMachine.objects.create(name='Test VM', status='active')
+ # Ensure enqueue_task was not called yet
+ mock_enqueue_task.assert_not_called()
+
+ # Assign primary IP with DNS name
+ ip_address = IPAddress.objects.create(
+ address='192.0.2.2/24',
+ dns_name='test-vm.example.com'
+ )
+ vm.primary_ip4 = ip_address
+ vm.save()
+
+ # Now enqueue_task should be called
+ mock_enqueue_task.assert_called_with('sync_host', self.awx_inventory, VirtualMachine, vm)
+
+
+ @patch('netbox_awx_plugin.signals.enqueue_task')
+ def test_interface_post_save_signal(self, mock_enqueue_task):
+ # Create related objects
+ device_role = DeviceRole.objects.create(name='Switch', slug='switch')
+ device_type = DeviceType.objects.create(
+ model='TestModel',
+ slug='testmodel',
+ manufacturer=self.manufacturer,
+ )
+ site = Site.objects.create(name='Test Site', slug='test-site', status='active')
+ device = Device.objects.create(
+ name='Test Device',
+ device_role=device_role,
+ device_type=device_type,
+ site=site,
+ status='active',
+ )
+
+ # Assign primary IP with DNS name to the device
+ ip_address = IPAddress.objects.create(
+ address='192.0.2.1/24',
+ dns_name='test-device.example.com'
+ )
+ device.primary_ip4 = ip_address
+ device.save()
+
+ # Reset the mock after setting up device
+ mock_enqueue_task.reset_mock()
+
+ # Create an Interface
+ interface = Interface.objects.create(device=device, name='eth0')
+
+ # Check that enqueue_task was called due to interface save
+ mock_enqueue_task.assert_called_with('sync_host', self.awx_inventory, Device, device)
+
+
+ @patch('netbox_awx_plugin.signals.enqueue_task')
+ def test_vm_tag_assignment(self, mock_enqueue_task):
+ # Create a Tag and assign it to a VirtualMachine
+ tag = Tag.objects.create(name='Test Tag', slug='test-tag')
+ vm = VirtualMachine.objects.create(name='Test VM', status='active')
+
+ # Reset the mock after creating related objects
+ mock_enqueue_task.reset_mock()
+
+ vm.tags.add(tag)
+ vm.save()
+
+ # enqueue_task should be called due to tag assignment
+ mock_enqueue_task.assert_called_with('sync_host', self.awx_inventory, VirtualMachine, vm)
+
+
+ @patch('netbox_awx_plugin.signals.enqueue_task')
+ def test_multiple_inventories(self, mock_enqueue_task):
+ # Create another AWX Inventory
+ another_awx_inventory = AWXInventory.objects.create(
+ awx=self.awx,
+ inventory_id=2,
+ enabled=True,
+ )
+ # Create a Site instance
+ site = Site.objects.create(
+ name='Test Site',
+ slug='test-site',
+ status='active',
+ )
+ # Check that enqueue_task was called for both inventories
+ calls = [
+ call('sync_group', self.awx_inventory, Site, site),
+ call('sync_group', another_awx_inventory, Site, site)
+ ]
+ mock_enqueue_task.assert_has_calls(calls, any_order=True)
+
+
+ def test_no_enabled_inventories(self):
+ # Disable the AWX Inventory
+ self.awx_inventory.enabled = False
+ self.awx_inventory.save()
+ # Create a Site instance
+ with patch('netbox_awx_plugin.signals.enqueue_task') as mock_enqueue_task:
+ site = Site.objects.create(
+ name='Test Site',
+ slug='test-site',
+ status='active',
+ )
+ # Check that enqueue_task was not called
+ mock_enqueue_task.assert_not_called()
diff --git a/netbox_awx_plugin/tests/test_synchronization.py b/netbox_awx_plugin/tests/test_synchronization.py
new file mode 100644
index 0000000000000000000000000000000000000000..8688323d400366f191331c6d4bd5de9449fb0d06
--- /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)