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