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()