netbox/netbox/virtualization/tests/test_api.py

import copy
import logging
from decimal import Decimal
from unittest.mock import patch

from django.db import connection
from django.db.models.signals import post_init
from django.test import override_settings, tag
from django.test.utils import CaptureQueriesContext
from django.urls import reverse
from netaddr import IPNetwork
from rest_framework import status

from core.models import ObjectType
from dcim.choices import InterfaceModeChoices
from dcim.models import Interface, Platform, Site
from extras.choices import CustomFieldTypeChoices
from extras.models import ConfigContext, ConfigTemplate, CustomField, ExportTemplate
from ipam.choices import FHRPGroupProtocolChoices, VLANQinQRoleChoices
from ipam.models import VLAN, VRF, FHRPGroup, FHRPGroupAssignment, IPAddress, Prefix
from users.constants import TOKEN_PREFIX
from users.models import Token
from utilities.testing import (
    APITestCase,
    APIViewTestCases,
    create_test_device,
    create_test_nat_ip_pair,
    create_test_virtualmachine,
    disable_logging,
)
from virtualization.choices import *
from virtualization.models import *


class AppTestCase(APITestCase):

    def test_root(self):

        url = reverse('virtualization-api:api-root')
        response = self.client.get('{}?format=api'.format(url), **self.header)

        self.assertEqual(response.status_code, 200)


class ClusterTypeTestCase(APIViewTestCases.APIViewTestCase):
    model = ClusterType
    brief_fields = ['cluster_count', 'description', 'display', 'id', 'name', 'slug', 'url']
    create_data = [
        {
            'name': 'Cluster Type 4',
            'slug': 'cluster-type-4',
        },
        {
            'name': 'Cluster Type 5',
            'slug': 'cluster-type-5',
        },
        {
            'name': 'Cluster Type 6',
            'slug': 'cluster-type-6',
        },
    ]
    bulk_update_data = {
        'description': 'New description',
    }
    bulk_update_invalid_data = {
        'owner': 99999,
    }

    @classmethod
    def setUpTestData(cls):

        cluster_types = (
            ClusterType(name='Cluster Type 1', slug='cluster-type-1'),
            ClusterType(name='Cluster Type 2', slug='cluster-type-2'),
            ClusterType(name='Cluster Type 3', slug='cluster-type-3'),
        )
        ClusterType.objects.bulk_create(cluster_types)


class ClusterGroupTestCase(APIViewTestCases.APIViewTestCase):
    model = ClusterGroup
    brief_fields = ['cluster_count', 'description', 'display', 'id', 'name', 'slug', 'url']
    create_data = [
        {
            'name': 'Cluster Group 4',
            'slug': 'cluster-type-4',
        },
        {
            'name': 'Cluster Group 5',
            'slug': 'cluster-type-5',
        },
        {
            'name': 'Cluster Group 6',
            'slug': 'cluster-type-6',
        },
    ]
    bulk_update_data = {
        'description': 'New description',
    }
    bulk_update_invalid_data = {
        'owner': 99999,
    }

    @classmethod
    def setUpTestData(cls):

        cluster_Groups = (
            ClusterGroup(name='Cluster Group 1', slug='cluster-type-1'),
            ClusterGroup(name='Cluster Group 2', slug='cluster-type-2'),
            ClusterGroup(name='Cluster Group 3', slug='cluster-type-3'),
        )
        ClusterGroup.objects.bulk_create(cluster_Groups)


class ClusterTestCase(APIViewTestCases.APIViewTestCase):
    model = Cluster
    brief_fields = ['description', 'display', 'id', 'name', 'url', 'virtualmachine_count']
    bulk_update_data = {
        'status': 'offline',
        'comments': 'New comment',
    }
    bulk_update_invalid_data = {
        'status': 'not-a-valid-status',
    }

    @classmethod
    def setUpTestData(cls):

        cluster_types = (
            ClusterType(name='Cluster Type 1', slug='cluster-type-1'),
            ClusterType(name='Cluster Type 2', slug='cluster-type-2'),
        )
        ClusterType.objects.bulk_create(cluster_types)

        cluster_groups = (
            ClusterGroup(name='Cluster Group 1', slug='cluster-group-1'),
            ClusterGroup(name='Cluster Group 2', slug='cluster-group-2'),
        )
        ClusterGroup.objects.bulk_create(cluster_groups)

        clusters = (
            Cluster(
                name='Cluster 1',
                type=cluster_types[0],
                group=cluster_groups[0],
                status=ClusterStatusChoices.STATUS_PLANNED,
            ),
            Cluster(
                name='Cluster 2',
                type=cluster_types[0],
                group=cluster_groups[0],
                status=ClusterStatusChoices.STATUS_PLANNED,
            ),
            Cluster(
                name='Cluster 3',
                type=cluster_types[0],
                group=cluster_groups[0],
                status=ClusterStatusChoices.STATUS_PLANNED,
            ),
        )
        for cluster in clusters:
            cluster.save()

        cls.create_data = [
            {
                'name': 'Cluster 4',
                'type': cluster_types[1].pk,
                'group': cluster_groups[1].pk,
                'status': ClusterStatusChoices.STATUS_STAGING,
            },
            {
                'name': 'Cluster 5',
                'type': cluster_types[1].pk,
                'group': cluster_groups[1].pk,
                'status': ClusterStatusChoices.STATUS_STAGING,
            },
            {
                'name': 'Cluster 6',
                'type': cluster_types[1].pk,
                'group': cluster_groups[1].pk,
                'status': ClusterStatusChoices.STATUS_STAGING,
            },
        ]

    def test_list_vm_totals_without_loading_virtual_machines(self):
        """
        VM counts and resource totals come from annotations without loading any virtual machine.
        """
        cluster1 = Cluster.objects.get(name='Cluster 1')
        cluster2 = Cluster.objects.get(name='Cluster 2')
        VirtualMachine.objects.bulk_create((
            VirtualMachine(name='Virtual Machine 1', cluster=cluster1, vcpus=Decimal('1.5'), memory=1024, disk=10000),
            VirtualMachine(name='Virtual Machine 2', cluster=cluster1, vcpus=Decimal('2.25'), memory=2048, disk=20000),
            VirtualMachine(name='Virtual Machine 3', cluster=cluster1),
            VirtualMachine(name='Virtual Machine 4', cluster=cluster2),
        ))
        self.add_permissions('virtualization.view_cluster')
        loaded = []

        def record(sender, instance, **kwargs):
            loaded.append(instance)

        post_init.connect(record, sender=VirtualMachine)
        try:
            response = self.client.get(self._get_list_url(), **self.header)
        finally:
            post_init.disconnect(record, sender=VirtualMachine)
        self.assertHttpStatus(response, status.HTTP_200_OK)
        fields = ('virtualmachine_count', 'allocated_vcpus', 'allocated_memory', 'allocated_disk')
        totals = {row['name']: tuple(row[field] for field in fields) for row in response.data['results']}
        self.assertEqual(totals, {
            'Cluster 1': (3, Decimal('3.75'), 3072, 30000),
            'Cluster 2': (1, None, None, None),
            'Cluster 3': (0, None, None, None),
        })
        self.assertEqual(loaded, [])


class VirtualMachineTypeTestCase(APIViewTestCases.APIViewTestCase):
    model = VirtualMachineType
    brief_fields = ['description', 'display', 'id', 'name', 'slug', 'url']
    user_permissions = ('dcim.view_platform', 'virtualization.view_virtualmachine')

    @classmethod
    def setUpTestData(cls):
        cls.platforms = (
            Platform.objects.create(name='Platform 1', slug='platform-1'),
            Platform.objects.create(name='Platform 2', slug='platform-2'),
            Platform.objects.create(name='Platform 3', slug='platform-3'),
        )

        cls.virtual_machine_types = (
            VirtualMachineType.objects.create(
                name='Virtual Machine Type 1',
                slug='virtual-machine-type-1',
                default_platform=cls.platforms[0],
                default_vcpus=1,
                default_memory=1024,
            ),
            VirtualMachineType.objects.create(
                name='Virtual Machine Type 2',
                slug='virtual-machine-type-2',
                default_platform=cls.platforms[1],
                default_vcpus=2,
                default_memory=2048,
            ),
            VirtualMachineType.objects.create(
                name='Virtual Machine Type 3',
                slug='virtual-machine-type-3',
                default_platform=cls.platforms[2],
                default_vcpus=4,
                default_memory=4096,
            ),
        )

        cls.create_data = [
            {
                'name': 'Virtual Machine Type 4',
                'slug': 'virtual-machine-type-4',
                'default_platform': cls.platforms[0].pk,
                'default_vcpus': 1,
                'default_memory': 1024,
            },
            {
                'name': 'Virtual Machine Type 5',
                'slug': 'virtual-machine-type-5',
                'default_platform': cls.platforms[1].pk,
                'default_vcpus': 2,
                'default_memory': 2048,
            },
            {
                'name': 'Virtual Machine Type 6',
                'slug': 'virtual-machine-type-6',
                'default_platform': cls.platforms[2].pk,
                'default_vcpus': 4,
                'default_memory': 4096,
            },
        ]

        cls.bulk_update_data = {
            'default_platform': cls.platforms[2].pk,
            'default_vcpus': 8,
            'default_memory': 8192,
            'description': 'New description',
        }
        cls.bulk_update_invalid_data = {
            'owner': 99999,
        }


class VirtualMachineTestCase(APIViewTestCases.APIViewTestCase):
    model = VirtualMachine
    brief_fields = ['description', 'display', 'id', 'name', 'url']
    bulk_update_data = {
        'status': 'staged',
    }
    bulk_update_invalid_data = {
        'status': 'not-a-valid-status',
    }
    user_permissions = ('dcim.view_platform', 'virtualization.view_virtualmachinetype')

    @classmethod
    def setUpTestData(cls):
        clustertype = ClusterType.objects.create(name='Cluster Type 1', slug='cluster-type-1')
        clustergroup = ClusterGroup.objects.create(name='Cluster Group 1', slug='cluster-group-1')

        cls.sites = (
            Site(name='Site 1', slug='site-1'),
            Site(name='Site 2', slug='site-2'),
            Site(name='Site 3', slug='site-3'),
        )
        Site.objects.bulk_create(cls.sites)

        cls.clusters = (
            Cluster(name='Cluster 1', type=clustertype, scope=cls.sites[0], group=clustergroup),
            Cluster(name='Cluster 2', type=clustertype, scope=cls.sites[1], group=clustergroup),
            Cluster(name='Cluster 3', type=clustertype),
        )
        for cluster in cls.clusters:
            cluster.save()

        cls.devices = (
            create_test_device('device1', site=cls.sites[0], cluster=cls.clusters[0]),
            create_test_device('device2', site=cls.sites[1], cluster=cls.clusters[1]),
        )

        cls.platforms = (
            Platform.objects.create(name='Platform 1', slug='platform-1'),
            Platform.objects.create(name='Platform 2', slug='platform-2'),
            Platform.objects.create(name='Platform 3', slug='platform-3'),
        )

        cls.vm_types = (
            VirtualMachineType.objects.create(
                name='Virtual Machine Type 1',
                slug='virtual-machine-type-1',
                default_platform=cls.platforms[0],
                default_vcpus=2,
                default_memory=4096,
            ),
            VirtualMachineType.objects.create(
                name='Virtual Machine Type 2',
                slug='virtual-machine-type-2',
                default_platform=cls.platforms[1],
                default_vcpus=4,
                default_memory=8192,
            ),
        )

        virtual_machines = (
            VirtualMachine(
                name='Virtual Machine 1',
                virtual_machine_type=cls.vm_types[0],
                site=cls.sites[0],
                cluster=cls.clusters[0],
                device=cls.devices[0],
                platform=cls.platforms[0],
                vcpus=2,
                memory=4096,
                local_context_data={'A': 1},
            ),
            VirtualMachine(
                name='Virtual Machine 2',
                site=cls.sites[0],
                cluster=cls.clusters[0],
                local_context_data={'B': 2},
            ),
            VirtualMachine(
                name='Virtual Machine 3',
                site=cls.sites[0],
                cluster=cls.clusters[0],
                local_context_data={'C': 3},
                start_on_boot=VirtualMachineStartOnBootChoices.STATUS_ON,
            ),
        )
        VirtualMachine.objects.bulk_create(virtual_machines)

        cls.create_data = [
            {
                'name': 'Virtual Machine 4',
                'site': cls.sites[1].pk,
                'cluster': cls.clusters[1].pk,
                'device': cls.devices[1].pk,
                'virtual_machine_type': cls.vm_types[0].pk,
            },
            {
                'name': 'Virtual Machine 5',
                'site': cls.sites[1].pk,
                'cluster': cls.clusters[1].pk,
                'virtual_machine_type': cls.vm_types[1].pk,
            },
            {
                'name': 'Virtual Machine 6',
                'site': cls.sites[1].pk,
            },
            {
                'name': 'Virtual Machine 7',
                'cluster': cls.clusters[2].pk,
                'virtual_machine_type': cls.vm_types[0].pk,
                'start_on_boot': VirtualMachineStartOnBootChoices.STATUS_ON,
            },
        ]

    def test_virtual_machine_type_defaults_applied_on_create(self):
        data = {
            'name': 'Virtual Machine With Defaults',
            'site': self.sites[1].pk,
            'cluster': self.clusters[1].pk,
            'virtual_machine_type': self.vm_types[0].pk,
            'platform': None,
            'vcpus': None,
            'memory': None,
        }
        self.add_permissions('virtualization.add_virtualmachine')

        response = self.client.post(self._get_list_url(), data, format='json', **self.header)
        self.assertHttpStatus(response, status.HTTP_201_CREATED)

        vm = VirtualMachine.objects.get(pk=response.data['id'])
        self.assertEqual(vm.virtual_machine_type, self.vm_types[0])
        self.assertEqual(vm.platform, self.vm_types[0].default_platform)
        self.assertEqual(vm.vcpus, self.vm_types[0].default_vcpus)
        self.assertEqual(vm.memory, self.vm_types[0].default_memory)

    def test_virtual_machine_type_defaults_do_not_override_explicit_values(self):
        data = {
            'name': 'Virtual Machine With Explicit Values',
            'site': self.sites[1].pk,
            'cluster': self.clusters[1].pk,
            'virtual_machine_type': self.vm_types[0].pk,
            'platform': self.platforms[2].pk,
            'vcpus': 6,
            'memory': 12288,
        }
        self.add_permissions('virtualization.add_virtualmachine')

        response = self.client.post(self._get_list_url(), data, format='json', **self.header)
        self.assertHttpStatus(response, status.HTTP_201_CREATED)

        vm = VirtualMachine.objects.get(pk=response.data['id'])
        self.assertEqual(vm.virtual_machine_type, self.vm_types[0])
        self.assertEqual(vm.platform, self.platforms[2])
        self.assertEqual(vm.vcpus, 6)
        self.assertEqual(vm.memory, 12288)

    def test_setting_virtual_machine_type_on_existing_vm_does_not_backfill_defaults(self):
        vm = VirtualMachine.objects.get(name='Virtual Machine 2')
        self.add_permissions('virtualization.change_virtualmachine')

        response = self.client.patch(
            self._get_detail_url(vm),
            {'virtual_machine_type': self.vm_types[1].pk},
            format='json',
            **self.header,
        )
        self.assertHttpStatus(response, status.HTTP_200_OK)

        vm.refresh_from_db()
        self.assertEqual(vm.virtual_machine_type, self.vm_types[1])
        self.assertIsNone(vm.platform)
        self.assertIsNone(vm.vcpus)
        self.assertIsNone(vm.memory)

    def test_config_context_read_without_copy(self):
        """
        REST and GraphQL read cached config context without copying it.
        """
        self.add_permissions('virtualization.view_virtualmachine')
        VirtualMachine.objects.filter(name='Virtual Machine 1').update(_config_context_data={'A': 'cached'})
        VirtualMachine.objects.filter(name='Virtual Machine 2').update(_config_context_data={})
        # Virtual Machine 3 has no cache and renders its local context data
        expected = {'Virtual Machine 1': {'A': 'cached'}, 'Virtual Machine 2': {}, 'Virtual Machine 3': {'C': 3}}
        query = '{ virtual_machine_list { name config_context } }'

        with patch('extras.models.configs.copy', wraps=copy) as copy_module:
            rest = self.client.get(self._get_list_url(), **self.header)
            graphql = self.client.post(reverse('graphql'), data={'query': query}, format='json', **self.header)
        copy_module.deepcopy.assert_not_called()

        self.assertHttpStatus(rest, status.HTTP_200_OK)
        self.assertEqual({row['name']: row['config_context'] for row in rest.data['results']}, expected)
        self.assertHttpStatus(graphql, status.HTTP_200_OK)
        data = graphql.json()
        self.assertNotIn('errors', data)
        self.assertEqual(
            {row['name']: row['config_context'] for row in data['data']['virtual_machine_list']}, expected
        )

    @tag('regression')  # Issue #23448
    def test_config_context_graphql_selection(self):
        """
        GraphQL annotates config context for virtual machines with no cache however config_context is selected,
        including through fragments, without a query per virtual machine.
        """
        self.add_permissions('virtualization.view_virtualmachine')
        ConfigContext.objects.create(name='Config Context 1', weight=100, data={'foo': 123})
        VirtualMachine.objects.update(_config_context_data=None)
        expected = {vm.name: vm.render_config_context() for vm in VirtualMachine.objects.all()}

        queries = {
            'field': '{ virtual_machine_list { name config_context } }',
            'inline fragment': '{ virtual_machine_list { ... on VirtualMachineType { name config_context } } }',
            'named fragment': (
                'fragment F on VirtualMachineType { name config_context } query { virtual_machine_list { ...F } }'
            ),
        }
        for shape, query in queries.items():
            with self.subTest(shape=shape):
                with CaptureQueriesContext(connection) as captured:
                    response = self.client.post(
                        reverse('graphql'), data={'query': query}, format='json', **self.header
                    )
                self.assertHttpStatus(response, status.HTTP_200_OK)
                data = response.json()
                self.assertNotIn('errors', data)
                rows = data['data']['virtual_machine_list']
                self.assertEqual({row['name']: row['config_context'] for row in rows}, expected)
                # Rendered by the annotation in a single query, not one fallback query per virtual machine
                sql = [captured_query['sql'] for captured_query in captured.captured_queries]
                self.assertEqual(len([s for s in sql if '"extras_configcontext"' in s]), 1)

    def test_config_context_included_by_default_in_list_view(self):
        """
        Check that config context data is included by default in the virtual machines list.
        """
        virtualmachine = VirtualMachine.objects.first()
        url = '{}?id={}'.format(reverse('virtualization-api:virtualmachine-list'), virtualmachine.pk)
        self.add_permissions('virtualization.view_virtualmachine')

        response = self.client.get(url, **self.header)
        self.assertEqual(response.data['results'][0].get('config_context', {}).get('A'), 1)

    def test_config_context_not_loaded_when_omitted(self):
        """
        List responses which omit config_context neither annotate it nor load the cached context.
        """
        self.add_permissions('virtualization.view_virtualmachine')
        for params in ({'fields': 'id,name'}, {'omit': 'config_context'}, {'brief': 1}):
            with self.subTest(params=params):
                with CaptureQueriesContext(connection) as queries:
                    response = self.client.get(self._get_list_url(), params, **self.header)
                self.assertHttpStatus(response, status.HTTP_200_OK)
                sql = '\n'.join(query['sql'] for query in queries.captured_queries)
                self.assertNotIn('"config_context_data"', sql)
                # The nested device still loads its own cache
                self.assertNotIn('"virtualization_virtualmachine"."_config_context_data"', sql)

    def test_config_context_loaded_when_requested(self):
        """
        Responses which include config_context read warm virtual machines from the cache and cold ones from the
        annotation, without a query per virtual machine.
        """
        self.add_permissions('virtualization.view_virtualmachine')
        ConfigContext.objects.create(name='Config Context 1', weight=100, data={'foo': 123})
        VirtualMachine.objects.filter(name='Virtual Machine 1').update(_config_context_data={'foo': 'cached'})
        cache_refresh = (
            'SELECT "virtualization_virtualmachine"."id", "virtualization_virtualmachine"."_config_context_data" FROM'
        )
        # fields takes precedence over omit, and both over brief
        for params in (
            {},
            {'fields': 'name,config_context'},
            {'fields': 'name,config_context', 'omit': 'config_context'},
            {'brief': 1, 'fields': 'name,config_context'},
            {'brief': 1, 'omit': 'comments'},
        ):
            with self.subTest(params=params):
                with CaptureQueriesContext(connection) as queries:
                    response = self.client.get(self._get_list_url(), params, **self.header)
                self.assertHttpStatus(response, status.HTTP_200_OK)
                contexts = {row['name']: row['config_context'] for row in response.data['results']}
                self.assertEqual(contexts, {
                    'Virtual Machine 1': {'foo': 'cached'},
                    'Virtual Machine 2': {'foo': 123, 'B': 2},
                    'Virtual Machine 3': {'foo': 123, 'C': 3},
                })
                statements = [
                    query['sql'] for query in queries.captured_queries if '"extras_configcontext"' in query['sql']
                ]
                self.assertEqual(len(statements), 1)
                self.assertTrue(statements[0].startswith('SELECT "virtualization_virtualmachine"'))
                refreshes = [
                    query['sql'] for query in queries.captured_queries if query['sql'].startswith(cache_refresh)
                ]
                self.assertEqual(refreshes, [])

    def test_unique_name_per_cluster_constraint(self):
        """
        Check that creating a virtual machine with a duplicate name fails.
        """
        data = {
            'name': 'Virtual Machine 1',
            'cluster': Cluster.objects.first().pk,
        }
        url = reverse('virtualization-api:virtualmachine-list')
        self.add_permissions('virtualization.add_virtualmachine')

        response = self.client.post(url, data, format='json', **self.header)
        self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)

    def test_render_config(self):
        configtemplate = ConfigTemplate.objects.create(
            name='Config Template 1',
            template_code='Config for virtual machine {{ virtualmachine.name }}'
        )

        vm = VirtualMachine.objects.first()
        vm.config_template = configtemplate
        vm.save()

        self.add_permissions(
            'virtualization.render_config_virtualmachine', 'virtualization.view_virtualmachine'
        )
        url = reverse('virtualization-api:virtualmachine-render-config', kwargs={'pk': vm.pk})
        response = self.client.post(url, {}, format='json', **self.header)
        self.assertHttpStatus(response, status.HTTP_200_OK)
        self.assertEqual(response.data['content'], f'Config for virtual machine {vm.name}')

    def test_render_config_without_permission(self):
        configtemplate = ConfigTemplate.objects.create(
            name='Config Template 1',
            template_code='Config for virtual machine {{ virtualmachine.name }}'
        )

        vm = VirtualMachine.objects.first()
        vm.config_template = configtemplate
        vm.save()

        # No permissions added - user has no render_config permission
        url = reverse('virtualization-api:virtualmachine-render-config', kwargs={'pk': vm.pk})
        response = self.client.post(url, {}, format='json', **self.header)
        self.assertHttpStatus(response, status.HTTP_404_NOT_FOUND)

    def test_render_config_token_write_enabled(self):
        configtemplate = ConfigTemplate.objects.create(
            name='Config Template 1',
            template_code='Config for virtual machine {{ virtualmachine.name }}'
        )

        vm = VirtualMachine.objects.first()
        vm.config_template = configtemplate
        vm.save()

        self.add_permissions('virtualization.render_config_virtualmachine', 'virtualization.view_virtualmachine')
        url = reverse('virtualization-api:virtualmachine-render-config', kwargs={'pk': vm.pk})

        # Request without token auth should fail with PermissionDenied
        response = self.client.post(url, {}, format='json')
        self.assertHttpStatus(response, status.HTTP_403_FORBIDDEN)

        # Create token with write_enabled=False
        token = Token.objects.create(version=2, user=self.user, write_enabled=False)
        token_header = f'Bearer {TOKEN_PREFIX}{token.key}.{token.token}'

        # Request with write-disabled token should fail
        response = self.client.post(url, {}, format='json', HTTP_AUTHORIZATION=token_header)
        self.assertHttpStatus(response, status.HTTP_403_FORBIDDEN)

        # Enable write and retry
        token.write_enabled = True
        token.save()
        response = self.client.post(url, {}, format='json', HTTP_AUTHORIZATION=token_header)
        self.assertHttpStatus(response, status.HTTP_200_OK)

    def test_list_object_includes_nat_inside_on_primary_ip(self):
        virtualmachine = create_test_virtualmachine('natted-vm')
        interface = VMInterface.objects.create(virtual_machine=virtualmachine, name='eth0')

        real_ip, nat_ip = create_test_nat_ip_pair(
            real_address='10.0.1.10/32',
            nat_address='198.51.100.20/32',
            inside_interface=interface,
        )

        virtualmachine.primary_ip4 = nat_ip
        virtualmachine.save()

        self.add_permissions('virtualization.view_virtualmachine', 'ipam.view_ipaddress')
        response = self.client.get(f'{self._get_list_url()}?id={virtualmachine.pk}', **self.header)
        self.assertHttpStatus(response, status.HTTP_200_OK)

        result = response.data['results'][0]
        for field in ('primary_ip', 'primary_ip4'):
            self.assertEqual(result[field]['address'], str(nat_ip.address))
            self.assertEqual(result[field]['nat_inside']['address'], str(real_ip.address))
            self.assertEqual(result[field]['nat_outside'], [])

    def test_get_object_includes_nat_outside_on_primary_ip(self):
        virtualmachine = create_test_virtualmachine('real-ip-vm')
        interface = VMInterface.objects.create(virtual_machine=virtualmachine, name='eth0')

        real_ip, nat_ip = create_test_nat_ip_pair(
            real_address='10.0.1.11/32',
            nat_address='198.51.100.21/32',
            inside_interface=interface,
        )

        virtualmachine.primary_ip4 = real_ip
        virtualmachine.save()

        self.add_permissions('virtualization.view_virtualmachine', 'ipam.view_ipaddress')
        response = self.client.get(
            self._get_detail_url(virtualmachine),
            **self.header,
        )
        self.assertHttpStatus(response, status.HTTP_200_OK)

        for field in ('primary_ip', 'primary_ip4'):
            self.assertEqual(response.data[field]['address'], str(real_ip.address))
            self.assertIsNone(response.data[field]['nat_inside'])
            self.assertCountEqual(
                [ip['address'] for ip in response.data[field]['nat_outside']],
                [str(nat_ip.address)],
            )

    def test_get_object_includes_dns_name_on_primary_ip(self):
        virtualmachine = create_test_virtualmachine('dns-vm')
        interfaces = (
            VMInterface.objects.create(virtual_machine=virtualmachine, name='eth0'),
            VMInterface.objects.create(virtual_machine=virtualmachine, name='eth1'),
        )

        ip4 = IPAddress(address='192.0.2.40/32', dns_name='vm4.example.com')
        ip4.assigned_object = interfaces[0]
        ip4.save()
        ip6 = IPAddress(address='2001:db8::40/128', dns_name='vm6.example.com')
        ip6.assigned_object = interfaces[1]
        ip6.save()

        virtualmachine.primary_ip4 = ip4
        virtualmachine.primary_ip6 = ip6
        virtualmachine.save()

        self.add_permissions('virtualization.view_virtualmachine', 'ipam.view_ipaddress')
        response = self.client.get(
            f'{self._get_detail_url(virtualmachine)}?exclude=config_context',
            **self.header,
        )
        self.assertHttpStatus(response, status.HTTP_200_OK)

        self.assertEqual(response.data['primary_ip4']['dns_name'], 'vm4.example.com')
        self.assertEqual(response.data['primary_ip6']['dns_name'], 'vm6.example.com')
        self.assertIn(
            response.data['primary_ip']['dns_name'],
            ('vm4.example.com', 'vm6.example.com'),
        )

    @tag('regression')  # Ref: #23278
    def test_create_with_unassigned_primary_ip(self):
        """Creating a VM with an unassigned primary IP returns HTTP 400."""
        ip4 = IPAddress.objects.create(address='192.0.2.10/24')
        ip6 = IPAddress.objects.create(address='2001:db8::10/64')

        self.add_permissions('virtualization.add_virtualmachine')
        for field, ip in (('primary_ip4', ip4), ('primary_ip6', ip6)):
            with self.subTest(field=field):
                data = {
                    'name': 'vm-new',
                    'site': self.sites[0].pk,
                    field: ip.pk,
                }
                response = self.client.post(self._get_list_url(), data, format='json', **self.header)
                self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
                self.assertIn(field, response.data)

    def test_render_config_with_config_template_id(self):
        default_template = ConfigTemplate.objects.create(
            name='Default Template',
            template_code='Default config for {{ virtualmachine.name }}'
        )
        override_template = ConfigTemplate.objects.create(
            name='Override Template',
            template_code='Override config for {{ virtualmachine.name }}'
        )

        vm = VirtualMachine.objects.first()
        vm.config_template = default_template
        vm.save()

        self.add_permissions(
            'virtualization.render_config_virtualmachine', 'virtualization.view_virtualmachine',
            'extras.view_configtemplate'
        )
        url = reverse('virtualization-api:virtualmachine-render-config', kwargs={'pk': vm.pk})

        # Render with override template
        response = self.client.post(url, {'config_template_id': override_template.pk}, format='json', **self.header)
        self.assertHttpStatus(response, status.HTTP_200_OK)
        self.assertEqual(response.data['content'], f'Override config for {vm.name}')

        # Render with nonexistent config_template_id
        response = self.client.post(url, {'config_template_id': 999999}, format='json', **self.header)
        self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)

        # Render with non-integer config_template_id
        response = self.client.post(url, {'config_template_id': 'abc'}, format='json', **self.header)
        self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)

        # Without view_configtemplate permission, override template should not be accessible
        self.remove_permissions('extras.view_configtemplate')
        response = self.client.post(url, {'config_template_id': override_template.pk}, format='json', **self.header)
        self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)


class VMInterfaceTestCase(APIViewTestCases.APIViewTestCase):
    model = VMInterface
    brief_fields = ['description', 'display', 'id', 'name', 'url', 'virtual_machine']
    bulk_update_data = {
        'description': 'New description',
    }
    bulk_update_invalid_data = {
        'mode': 'not-a-valid-mode',
    }
    graphql_base_name = 'vm_interface'
    user_permissions = ('virtualization.view_virtualmachine', )

    @classmethod
    def setUpTestData(cls):
        virtualmachine = create_test_virtualmachine('Virtual Machine 1')

        interfaces = (
            VMInterface(virtual_machine=virtualmachine, name='Interface 1'),
            VMInterface(virtual_machine=virtualmachine, name='Interface 2'),
            VMInterface(virtual_machine=virtualmachine, name='Interface 3'),
        )
        VMInterface.objects.bulk_create(interfaces)

        vlans = (
            VLAN(name='VLAN 1', vid=1),
            VLAN(name='VLAN 2', vid=2),
            VLAN(name='VLAN 3', vid=3),
            VLAN(name='SVLAN 1', vid=1001, qinq_role=VLANQinQRoleChoices.ROLE_SERVICE),
        )
        VLAN.objects.bulk_create(vlans)

        vrfs = (
            VRF(name='VRF 1'),
            VRF(name='VRF 2'),
            VRF(name='VRF 3'),
        )
        VRF.objects.bulk_create(vrfs)

        cls.create_data = [
            {
                'virtual_machine': virtualmachine.pk,
                'name': 'Interface 4',
                'mode': InterfaceModeChoices.MODE_TAGGED,
                'tagged_vlans': [vlans[0].pk, vlans[1].pk],
                'untagged_vlan': vlans[2].pk,
                'vrf': vrfs[0].pk,
            },
            {
                'virtual_machine': virtualmachine.pk,
                'name': 'Interface 5',
                'mode': InterfaceModeChoices.MODE_TAGGED,
                'bridge': interfaces[0].pk,
                'tagged_vlans': [vlans[0].pk, vlans[1].pk],
                'untagged_vlan': vlans[2].pk,
                'vrf': vrfs[1].pk,
            },
            {
                'virtual_machine': virtualmachine.pk,
                'name': 'Interface 6',
                'mode': InterfaceModeChoices.MODE_TAGGED,
                'parent': interfaces[1].pk,
                'tagged_vlans': [vlans[0].pk, vlans[1].pk],
                'untagged_vlan': vlans[2].pk,
                'vrf': vrfs[2].pk,
            },
            {
                'virtual_machine': virtualmachine.pk,
                'name': 'Interface 7',
                'mode': InterfaceModeChoices.MODE_Q_IN_Q,
                'qinq_svlan': vlans[3].pk,
            },
        ]

    @tag('regression')
    def test_set_vminterface_as_object_in_custom_field(self):
        cf = CustomField.objects.create(
            name='associated_interface',
            type=CustomFieldTypeChoices.TYPE_OBJECT,
            related_object_type=ObjectType.objects.get_for_model(VMInterface),
            required=False
        )
        cf.object_types.set([ObjectType.objects.get_for_model(Prefix)])
        cf.save()

        prefix = Prefix.objects.create(prefix=IPNetwork('10.0.0.0/12'))
        vmi = VMInterface.objects.first()

        url = reverse('ipam-api:prefix-detail', kwargs={'pk': prefix.pk})
        data = {
            'custom_fields': {
                'associated_interface': vmi.id,
            },
        }

        self.add_permissions('ipam.change_prefix')

        response = self.client.patch(url, data, format='json', **self.header)
        self.assertEqual(response.status_code, 200)

        prefix_data = response.json()
        self.assertEqual(prefix_data['custom_fields']['associated_interface']['id'], vmi.id)

        reloaded_prefix = Prefix.objects.get(pk=prefix.pk)
        self.assertEqual(prefix.pk, reloaded_prefix.pk)
        self.assertNotEqual(reloaded_prefix.cf['associated_interface'], None)

    def test_bulk_delete_child_interfaces(self):
        interface1 = VMInterface.objects.get(name='Interface 1')
        virtual_machine = interface1.virtual_machine
        self.add_permissions('virtualization.delete_vminterface')

        # Create a child interface
        child = VMInterface.objects.create(
            virtual_machine=virtual_machine,
            name='Interface 1A',
            parent=interface1
        )
        self.assertEqual(virtual_machine.interfaces.count(), 4)

        # Attempt to delete only the parent interface
        url = self._get_detail_url(interface1)
        with disable_logging(level=logging.WARNING):
            self.client.delete(url, **self.header)
        self.assertEqual(virtual_machine.interfaces.count(), 4)  # Parent was not deleted

        # Attempt to bulk delete parent & child together
        data = [
            {"id": interface1.pk},
            {"id": child.pk},
        ]
        self.client.delete(self._get_list_url(), data, format='json', **self.header)
        self.assertEqual(virtual_machine.interfaces.count(), 2)  # Child & parent were both deleted

    def test_mac_address_create(self):
        """
        Creating a VMInterface with mac_address creates the primary MACAddress in one request.
        """
        self.add_permissions('virtualization.add_vminterface', 'dcim.add_macaddress')
        vm = VMInterface.objects.first().virtual_machine
        data = {
            'virtual_machine': vm.pk,
            'name': 'Interface MAC Create',
            'mac_address': 'AA:BB:CC:DD:EE:FF',
        }
        response = self.client.post(self._get_list_url(), data, format='json', **self.header)
        self.assertHttpStatus(response, status.HTTP_201_CREATED)
        iface = VMInterface.objects.get(pk=response.data['id'])
        self.assertIsNotNone(iface.primary_mac_address)
        self.assertEqual(str(iface.primary_mac_address.mac_address).upper(), 'AA:BB:CC:DD:EE:FF')
        self.assertEqual(iface.primary_mac_address.assigned_object, iface)

    def test_mac_address_update(self):
        """
        Patching mac_address creates/updates the primary MACAddress in one request.
        """
        self.add_permissions('virtualization.change_vminterface', 'dcim.add_macaddress', 'dcim.change_macaddress')
        iface = VMInterface.objects.first()
        url = self._get_detail_url(iface)

        # Set a new primary MAC via mac_address shortcut
        response = self.client.patch(url, {'mac_address': '11:22:33:44:55:66'}, format='json', **self.header)
        self.assertHttpStatus(response, status.HTTP_200_OK)
        iface.refresh_from_db()
        self.assertIsNotNone(iface.primary_mac_address)
        self.assertEqual(str(iface.primary_mac_address.mac_address).upper(), '11:22:33:44:55:66')

        # Update the MAC to a new value
        response = self.client.patch(url, {'mac_address': 'AA:BB:CC:DD:EE:FF'}, format='json', **self.header)
        self.assertHttpStatus(response, status.HTTP_200_OK)
        iface.refresh_from_db()
        self.assertEqual(str(iface.primary_mac_address.mac_address).upper(), 'AA:BB:CC:DD:EE:FF')

        # Clear the primary MAC by sending null
        response = self.client.patch(url, {'mac_address': None}, format='json', **self.header)
        self.assertHttpStatus(response, status.HTTP_200_OK)
        iface.refresh_from_db()
        self.assertIsNone(iface.primary_mac_address)

    def test_mac_address_invalid(self):
        """
        Sending an invalid MAC address string returns a 400 error.
        """
        self.add_permissions('virtualization.add_vminterface', 'dcim.add_macaddress')
        vm = VMInterface.objects.first().virtual_machine
        data = {
            'virtual_machine': vm.pk,
            'name': 'Interface MAC Bad',
            'mac_address': 'not-a-mac',
        }
        response = self.client.post(self._get_list_url(), data, format='json', **self.header)
        self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
        self.assertIn('mac_address', response.data)

    def test_mac_address_find_or_create(self):
        """
        Patching mac_address with a MAC that already exists on the VMInterface promotes it to
        primary without creating a duplicate MACAddress record.
        """
        from dcim.models import MACAddress
        self.add_permissions('virtualization.change_vminterface', 'dcim.add_macaddress', 'dcim.change_macaddress')
        iface = VMInterface.objects.first()

        mac1 = MACAddress.objects.create(mac_address='CC:DD:EE:FF:00:01', assigned_object=iface)
        mac2 = MACAddress.objects.create(mac_address='CC:DD:EE:FF:00:02', assigned_object=iface)
        iface.primary_mac_address = mac1
        iface.save()

        mac_count_before = iface.mac_addresses.count()
        url = self._get_detail_url(iface)

        response = self.client.patch(url, {'mac_address': 'CC:DD:EE:FF:00:02'}, format='json', **self.header)
        self.assertHttpStatus(response, status.HTTP_200_OK)

        iface.refresh_from_db()
        self.assertEqual(iface.primary_mac_address.pk, mac2.pk)
        self.assertEqual(iface.mac_addresses.count(), mac_count_before)

    def test_mac_address_conflicts_with_primary_mac_address(self):
        """
        Supplying both mac_address and primary_mac_address in one request is rejected on VMInterface
        too (the shared shortcut mixin applies to both interface types).
        """
        from dcim.models import MACAddress

        self.add_permissions(
            'virtualization.change_vminterface', 'dcim.add_macaddress', 'dcim.change_macaddress'
        )
        iface = VMInterface.objects.first()
        mac = MACAddress.objects.create(mac_address='DD:EE:FF:00:11:22', assigned_object=iface)
        url = self._get_detail_url(iface)

        response = self.client.patch(
            url,
            {'mac_address': 'DD:EE:FF:00:11:33', 'primary_mac_address': {'mac_address': str(mac.mac_address)}},
            format='json',
            **self.header
        )
        self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)

    def test_mac_address_conflicts_with_explicit_null_primary(self):
        """
        The presence-based conflict guard applies to VMInterface too: mac_address shortcut plus an
        explicit primary_mac_address=null is rejected.
        """
        self.add_permissions('virtualization.change_vminterface', 'dcim.add_macaddress')
        iface = VMInterface.objects.first()
        url = self._get_detail_url(iface)

        response = self.client.patch(
            url,
            {'mac_address': 'DD:EE:FF:00:11:44', 'primary_mac_address': None},
            format='json',
            **self.header
        )
        self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)

    def test_primary_mac_address_must_belong_to_interface(self):
        """
        Setting primary_mac_address to a MAC not assigned to this VMInterface is rejected on update,
        so the primary MAC can't dangle outside the interface's own MAC set.
        """
        from dcim.models import MACAddress

        self.add_permissions('virtualization.change_vminterface', 'dcim.change_macaddress')
        iface = VMInterface.objects.first()
        unassigned = MACAddress.objects.create(mac_address='DD:EE:FF:00:11:55')
        url = self._get_detail_url(iface)

        response = self.client.patch(
            url,
            {'primary_mac_address': {'mac_address': str(unassigned.mac_address)}},
            format='json',
            **self.header
        )
        self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
        iface.refresh_from_db()
        self.assertIsNone(iface.primary_mac_address)

    @override_settings(CUSTOM_VALIDATORS={'dcim.macaddress': [{'mac_address': {'regex': '^AA:'}}]})
    def test_mac_address_custom_validation_returns_400(self):
        """
        A MAC that fails a custom validator on VMInterface creation returns 400, not 500.
        """
        self.add_permissions('virtualization.add_vminterface', 'dcim.add_macaddress')
        vm = VirtualMachine.objects.first()
        data = {
            'virtual_machine': vm.pk,
            'name': 'VMInterface Custom Validation',
            'mac_address': 'BB:CC:DD:EE:FF:00',
        }
        response = self.client.post(self._get_list_url(), data, format='json', **self.header)
        self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)

    def test_count_ipaddresses_and_fhrp_groups(self):
        """Counts come from annotations of the VM interface's object type, never from loaded related objects."""
        self.add_permissions('virtualization.view_vminterface')
        vm_interface = VMInterface.objects.get(name='Interface 1')
        group = FHRPGroup.objects.create(
            pk=vm_interface.pk, protocol=FHRPGroupProtocolChoices.PROTOCOL_VRRP2, group_id=1
        )
        other_group = FHRPGroup.objects.create(
            pk=vm_interface.pk + 1, protocol=FHRPGroupProtocolChoices.PROTOCOL_VRRP2, group_id=2
        )
        interface = Interface.objects.create(
            pk=vm_interface.pk, device=create_test_device('Device 1'), name='eth0', type='1000base-t'
        )
        IPAddress.objects.bulk_create((
            IPAddress(address='192.0.2.1/24', assigned_object=vm_interface),
            IPAddress(address='2001:db8::1/64', assigned_object=vm_interface),
            IPAddress(address='192.0.2.2/24', assigned_object=interface),
            IPAddress(address='192.0.2.254/24', assigned_object=group),
        ))
        FHRPGroupAssignment.objects.bulk_create((
            FHRPGroupAssignment(group=group, interface=vm_interface, priority=100),
            # Two on the dcim interface, so a count of the wrong object type cannot match
            FHRPGroupAssignment(group=group, interface=interface, priority=100),
            FHRPGroupAssignment(group=other_group, interface=interface, priority=100),
        ))

        with CaptureQueriesContext(connection) as queries:
            response = self.client.get(self._get_list_url(), **self.header)
        self.assertHttpStatus(response, status.HTTP_200_OK)
        counts = {
            row['name']: (row['count_ipaddresses'], row['count_fhrp_groups']) for row in response.data['results']
        }
        self.assertEqual(counts, {'Interface 1': (2, 1), 'Interface 2': (0, 0), 'Interface 3': (0, 0)})
        # Only the VM interface statement reads the related tables, through its two count subqueries
        statements = [
            query['sql'] for query in queries.captured_queries
            if '"ipam_ipaddress"' in query['sql'] or '"ipam_fhrpgroupassignment"' in query['sql']
        ]
        self.assertEqual(len(statements), 1)
        self.assertTrue(statements[0].startswith('SELECT "virtualization_vminterface"'))
        # Unannotated instances count through the relations
        self.assertEqual((vm_interface.count_ipaddresses, vm_interface.count_fhrp_groups), (2, 1))

    def test_ordering_by_count_annotation(self):
        """Ordering by a count annotation sorts the response whether or not it renders the count."""
        self.add_permissions('virtualization.view_vminterface')
        groups = FHRPGroup.objects.bulk_create((
            FHRPGroup(protocol=FHRPGroupProtocolChoices.PROTOCOL_VRRP2, group_id=1),
            FHRPGroup(protocol=FHRPGroupProtocolChoices.PROTOCOL_VRRP2, group_id=2),
        ))
        interfaces = {name: VMInterface.objects.get(name=name) for name in ('Interface 2', 'Interface 3')}
        FHRPGroupAssignment.objects.bulk_create((
            FHRPGroupAssignment(group=groups[0], interface=interfaces['Interface 2'], priority=100),
            FHRPGroupAssignment(group=groups[1], interface=interfaces['Interface 2'], priority=100),
            FHRPGroupAssignment(group=groups[0], interface=interfaces['Interface 3'], priority=100),
        ))
        url = f'{self._get_list_url()}?ordering=-count_fhrp_groups'

        # The count's source isn't a valid ordering field
        response = self.client.get(f'{self._get_list_url()}?ordering=-_count_fhrp_groups', **self.header)
        self.assertHttpStatus(response, status.HTTP_200_OK)
        self.assertEqual(
            [row['name'] for row in response.data['results']], ['Interface 1', 'Interface 2', 'Interface 3']
        )

        # Responses without the count annotate it for the ordering alone
        for params in ('', '&brief=1', '&fields=id,name'):
            with self.subTest(params=params):
                response = self.client.get(url + params, **self.header)
                self.assertHttpStatus(response, status.HTTP_200_OK)
                self.assertEqual(
                    [row['name'] for row in response.data['results']], ['Interface 2', 'Interface 3', 'Interface 1']
                )

    def test_export_template_prefetches_related_objects(self):
        """An API export renders model instances, so it prefetches each counted relation once."""
        self.add_permissions('virtualization.view_vminterface', 'extras.view_exporttemplate')
        vm_interface = VMInterface.objects.get(name='Interface 1')
        IPAddress.objects.create(address='192.0.2.1/24', assigned_object=vm_interface)
        group = FHRPGroup.objects.create(protocol=FHRPGroupProtocolChoices.PROTOCOL_VRRP2, group_id=1)
        FHRPGroupAssignment.objects.create(group=group, interface=vm_interface, priority=100)
        export_template = ExportTemplate.objects.create(
            name='VM Interface Relations',
            template_code=(
                '{% for interface in queryset %}{{ interface.name }}={{ interface.ip_addresses.all()|length }}/'
                '{{ interface.fhrp_group_assignments.all()|length }},{% endfor %}'
            ),
        )
        export_template.object_types.set([ObjectType.objects.get_for_model(VMInterface)])

        with CaptureQueriesContext(connection) as queries:
            response = self.client.get(self._get_list_url(), {'export': export_template.name}, **self.header)
        self.assertHttpStatus(response, status.HTTP_200_OK)
        self.assertEqual(response.content.decode(), 'Interface 1=1/1,Interface 2=0/0,Interface 3=0/0,')
        for table in ('ipam_ipaddress', 'ipam_fhrpgroupassignment'):
            statements = [query['sql'] for query in queries.captured_queries if f'FROM "{table}"' in query['sql']]
            self.assertEqual(len(statements), 1, table)


class VirtualDiskTestCase(APIViewTestCases.APIViewTestCase):
    model = VirtualDisk
    brief_fields = ['description', 'display', 'id', 'name', 'size', 'url', 'virtual_machine']
    bulk_update_data = {
        'size': 888,
    }
    bulk_update_invalid_data = {
        'virtual_machine': 99999,
    }
    graphql_base_name = 'virtual_disk'
    user_permissions = ('virtualization.view_virtualmachine', )

    @classmethod
    def setUpTestData(cls):
        virtualmachine = create_test_virtualmachine('Virtual Machine 1')

        disks = (
            VirtualDisk(virtual_machine=virtualmachine, name='Disk 1', size=10),
            VirtualDisk(virtual_machine=virtualmachine, name='Disk 2', size=20),
            VirtualDisk(virtual_machine=virtualmachine, name='Disk 3', size=30),
        )
        VirtualDisk.objects.bulk_create(disks)

        cls.create_data = [
            {
                'virtual_machine': virtualmachine.pk,
                'name': 'Disk 4',
                'size': 10,
            },
            {
                'virtual_machine': virtualmachine.pk,
                'name': 'Disk 5',
                'size': 20,
            },
            {
                'virtual_machine': virtualmachine.pk,
                'name': 'Disk 6',
                'size': 30,
            },
        ]