add tests for signals - INFRA-10664

Closes INFRA-10376

netbox_awx_plugin/tests/test_signals.py

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