import netaddr
from django.contrib.contenttypes.models import ContentType
from django.core.exceptions import ValidationError
from django.db.backends.postgresql.psycopg_any import NumericRange
from django.test import TestCase, override_settings
from netaddr import IPNetwork, IPSet

from dcim.models import Location, Region, Site, SiteGroup
from ipam.choices import *
from ipam.constants import SERVICE_PORT_MAX, SERVICE_PORT_MIN
from ipam.models import *
from ipam.utils import rebuild_prefixes
from utilities.data import string_to_ranges
from virtualization.models import VirtualMachine


class AggregateTestCase(TestCase):

    def test_family_string(self):
        # Test property when prefix is a string
        agg = Aggregate(prefix='10.0.0.0/8')
        self.assertEqual(agg.family, 4)
        agg_v6 = Aggregate(prefix='2001:db8::/32')
        self.assertEqual(agg_v6.family, 6)

    def test_get_utilization(self):
        rir = RIR.objects.create(name='RIR 1', slug='rir-1')
        aggregate = Aggregate(prefix=IPNetwork('10.0.0.0/8'), rir=rir)
        aggregate.save()

        # 25% utilization
        Prefix.objects.bulk_create((
            Prefix(prefix=IPNetwork('10.0.0.0/12')),
            Prefix(prefix=IPNetwork('10.16.0.0/12')),
            Prefix(prefix=IPNetwork('10.32.0.0/12')),
            Prefix(prefix=IPNetwork('10.48.0.0/12')),
        ))
        self.assertEqual(aggregate.get_utilization(), 25)

        # 50% utilization
        Prefix.objects.bulk_create((
            Prefix(prefix=IPNetwork('10.64.0.0/10')),
        ))
        self.assertEqual(aggregate.get_utilization(), 50)

        # 100% utilization
        Prefix.objects.bulk_create((
            Prefix(prefix=IPNetwork('10.128.0.0/9')),
        ))
        self.assertEqual(aggregate.get_utilization(), 100)


class IPRangeTestCase(TestCase):

    def test_family_string(self):
        # Test property when start_address is a string
        ip_range = IPRange(start_address='10.0.0.1/24', end_address='10.0.0.254/24')
        self.assertEqual(ip_range.family, 4)
        ip_range_v6 = IPRange(start_address='2001:db8::1/64', end_address='2001:db8::ffff/64')
        self.assertEqual(ip_range_v6.family, 6)

    def test_overlapping_range(self):
        iprange_192_168 = IPRange.objects.create(
            start_address=IPNetwork('192.168.0.1/22'), end_address=IPNetwork('192.168.0.49/22')
        )
        iprange_192_168.clean()
        iprange_3_1_99 = IPRange.objects.create(
            start_address=IPNetwork('1.2.3.1/24'), end_address=IPNetwork('1.2.3.99/24')
        )
        iprange_3_1_99.clean()
        iprange_3_100_199 = IPRange.objects.create(
            start_address=IPNetwork('1.2.3.100/24'), end_address=IPNetwork('1.2.3.199/24')
        )
        iprange_3_100_199.clean()
        iprange_3_200_255 = IPRange.objects.create(
            start_address=IPNetwork('1.2.3.200/24'), end_address=IPNetwork('1.2.3.255/24')
        )
        iprange_3_200_255.clean()
        iprange_4_1_99 = IPRange.objects.create(
            start_address=IPNetwork('1.2.4.1/24'), end_address=IPNetwork('1.2.4.99/24')
        )
        iprange_4_1_99.clean()
        iprange_4_200 = IPRange.objects.create(
            start_address=IPNetwork('1.2.4.200/24'), end_address=IPNetwork('1.2.4.255/24')
        )
        iprange_4_200.clean()

        # Overlapping range entirely within existing
        with self.assertRaises(ValidationError):
            iprange_3_123_124 = IPRange.objects.create(
                start_address=IPNetwork('1.2.3.123/26'), end_address=IPNetwork('1.2.3.124/26')
            )
            iprange_3_123_124.clean()

        # Overlapping range starting within existing
        with self.assertRaises(ValidationError):
            iprange_4_98_101 = IPRange.objects.create(
                start_address=IPNetwork('1.2.4.98/24'), end_address=IPNetwork('1.2.4.101/24')
            )
            iprange_4_98_101.clean()

        # Overlapping range ending within existing
        with self.assertRaises(ValidationError):
            iprange_4_198_201 = IPRange.objects.create(
                start_address=IPNetwork('1.2.4.198/24'), end_address=IPNetwork('1.2.4.201/24')
            )
            iprange_4_198_201.clean()

    def test_single_address_range(self):
        iprange = IPRange(
            start_address=IPNetwork('192.0.2.10/24'),
            end_address=IPNetwork('192.0.2.10/24'),
        )

        iprange.clean()
        iprange.save()

        self.assertEqual(iprange.size, 1)
        self.assertEqual(str(iprange), '192.0.2.10-192.0.2.10/24')
        self.assertEqual(iprange.get_first_available_ip(), '192.0.2.10/24')

    def test_first_available_ip_consumed_single_address_range(self):
        iprange = IPRange.objects.create(
            start_address=IPNetwork('192.0.2.10/24'),
            end_address=IPNetwork('192.0.2.10/24'),
        )
        IPAddress.objects.create(address=IPNetwork('192.0.2.10/24'))

        # The sole address in the range is now assigned, so no IPs remain available.
        self.assertIsNone(iprange.get_first_available_ip())

    def test_single_address_range_ipv6(self):
        # IPRange.name has IPv4/IPv6-specific formatting; exercise the IPv6 branch
        # for a single-address range too.
        iprange = IPRange(
            start_address=IPNetwork('2001:db8::10/64'),
            end_address=IPNetwork('2001:db8::10/64'),
        )

        iprange.clean()
        iprange.save()

        self.assertEqual(iprange.size, 1)
        self.assertEqual(str(iprange), '2001:db8::10-2001:db8::10/64')
        self.assertEqual(iprange.get_first_available_ip(), '2001:db8::10/64')

    def test_reversed_range(self):
        iprange = IPRange(
            start_address=IPNetwork('192.0.2.10/24'),
            end_address=IPNetwork('192.0.2.9/24'),
        )

        with self.assertRaises(ValidationError):
            iprange.clean()

    def test_overlapping_single_address_range(self):
        IPRange.objects.create(
            start_address=IPNetwork('192.0.2.10/24'),
            end_address=IPNetwork('192.0.2.10/24'),
        )

        iprange = IPRange(
            start_address=IPNetwork('192.0.2.10/24'),
            end_address=IPNetwork('192.0.2.10/24'),
        )

        # Assert the overlap-specific error message so this test cannot pass on a
        # regression where start_address == end_address is rejected earlier.
        with self.assertRaisesMessage(ValidationError, 'Defined addresses overlap'):
            iprange.clean()

    def test_get_child_ips_host_portion(self):
        iprange = IPRange.objects.create(
            start_address=IPNetwork('10.0.0.2/24'),
            end_address=IPNetwork('10.0.0.254/24'),
        )

        ip1 = IPAddress.objects.create(address=IPNetwork('10.0.0.2/32'))
        ip2 = IPAddress.objects.create(address=IPNetwork('10.0.0.3/24'))

        self.assertEqual(set(iprange.get_child_ips()), {ip1, ip2})

    def test_get_available_ips(self):
        """
        Tests that occupied hosts are deduplicated and excluded from the available set.
        """
        iprange = IPRange.objects.create(
            start_address=IPNetwork('192.0.2.10/24'),
            end_address=IPNetwork('192.0.2.13/24'),
        )
        IPAddress.objects.bulk_create((
            IPAddress(address=IPNetwork('192.0.2.10/24')),
            IPAddress(address=IPNetwork('192.0.2.10/32')),
        ))

        self.assertEqual(iprange.get_available_ips(), IPSet(['192.0.2.11/32', '192.0.2.12/31']))

    def test_get_available_ips_mark_populated(self):
        """
        Tests that a populated range reports no available IPs.
        """
        iprange = IPRange.objects.create(
            start_address=IPNetwork('192.0.2.10/24'),
            end_address=IPNetwork('192.0.2.13/24'),
            mark_populated=True,
        )

        self.assertEqual(iprange.get_available_ips(), IPSet())

    def test_get_available_ips_vrf(self):
        """
        Tests that IPs in other VRFs do not consume range space.
        """
        vrf1 = VRF.objects.create(name='VRF 1')
        vrf2 = VRF.objects.create(name='VRF 2')
        iprange = IPRange.objects.create(
            start_address=IPNetwork('192.0.2.10/24'),
            end_address=IPNetwork('192.0.2.11/24'),
            vrf=vrf1,
        )
        IPAddress.objects.create(address=IPNetwork('192.0.2.10/24'), vrf=vrf2)

        self.assertEqual(iprange.get_available_ips(), IPSet(['192.0.2.10/31']))

    def test_iter_available_ips(self):
        """
        Tests that iter_available_ips() yields the same addresses as get_available_ips() in order.
        """
        iprange = IPRange.objects.create(
            start_address=IPNetwork('192.0.2.10/24'),
            end_address=IPNetwork('192.0.2.13/24'),
        )
        IPAddress.objects.create(address=IPNetwork('192.0.2.11/24'))

        self.assertEqual(list(iprange.iter_available_ips()), sorted(iprange.get_available_ips()))

    def test_available_ip_count(self):
        iprange = IPRange.objects.create(
            start_address=IPNetwork('192.0.2.10/24'),
            end_address=IPNetwork('192.0.2.19/24'),
        )

        IPAddress.objects.create(address=IPNetwork('192.0.2.12/24'))

        self.assertEqual(iprange.get_available_ip_count(), 9)

    def test_available_ip_count_distinct_hosts(self):
        iprange = IPRange.objects.create(
            start_address=IPNetwork('192.0.2.10/24'),
            end_address=IPNetwork('192.0.2.19/24'),
        )

        # Two rows for .10 (different masks) must dedupe to a single occupied host.
        IPAddress.objects.bulk_create((
            IPAddress(address=IPNetwork('192.0.2.10/24')),
            IPAddress(address=IPNetwork('192.0.2.10/32')),
            IPAddress(address=IPNetwork('192.0.2.11/24')),
        ))

        self.assertEqual(iprange.get_available_ip_count(), 8)

    def test_available_ip_count_vrf(self):
        vrf1 = VRF.objects.create(name='VRF 1')
        vrf2 = VRF.objects.create(name='VRF 2')

        iprange = IPRange.objects.create(
            start_address=IPNetwork('192.0.2.10/24'),
            end_address=IPNetwork('192.0.2.19/24'),
            vrf=vrf1,
        )

        IPAddress.objects.create(address=IPNetwork('192.0.2.12/24'), vrf=vrf1)
        IPAddress.objects.create(address=IPNetwork('192.0.2.13/24'), vrf=vrf2)

        # Only the VRF 1 IP should count.
        self.assertEqual(iprange.get_available_ip_count(), 9)

    def test_available_ip_count_populated(self):
        iprange = IPRange.objects.create(
            start_address=IPNetwork('192.0.2.10/24'),
            end_address=IPNetwork('192.0.2.19/24'),
            mark_populated=True,
        )

        self.assertEqual(iprange.get_available_ip_count(), 0)

    def test_first_available_ip_full(self):
        iprange = IPRange.objects.create(
            start_address=IPNetwork('192.0.2.10/24'),
            end_address=IPNetwork('192.0.2.11/24'),
        )

        IPAddress.objects.create(address=IPNetwork('192.0.2.10/24'))
        IPAddress.objects.create(address=IPNetwork('192.0.2.11/24'))

        self.assertIsNone(iprange.get_first_available_ip())

    def test_first_available_ip_populated(self):
        iprange = IPRange.objects.create(
            start_address=IPNetwork('192.0.2.10/24'),
            end_address=IPNetwork('192.0.2.19/24'),
            mark_populated=True,
        )

        self.assertIsNone(iprange.get_first_available_ip())

    def test_first_available_ip_ipv6(self):
        iprange = IPRange.objects.create(
            start_address=IPNetwork('::/126'),
            end_address=IPNetwork('::3/126'),
        )

        self.assertEqual(iprange.get_first_available_ip(), '::/126')

    def test_utilization_distinct_hosts(self):
        iprange = IPRange.objects.create(
            start_address=IPNetwork('192.0.2.10/24'),
            end_address=IPNetwork('192.0.2.19/24'),
        )

        IPAddress.objects.bulk_create((
            IPAddress(address=IPNetwork('192.0.2.10/24')),
            IPAddress(address=IPNetwork('192.0.2.10/32')),
            IPAddress(address=IPNetwork('192.0.2.11/24')),
        ))

        # Two distinct hosts in a 10-address range.
        self.assertEqual(iprange.utilization, 2 / 10 * 100)

    def test_utilization_vrf(self):
        vrf1 = VRF.objects.create(name='VRF 1')
        vrf2 = VRF.objects.create(name='VRF 2')

        iprange = IPRange.objects.create(
            start_address=IPNetwork('192.0.2.10/24'),
            end_address=IPNetwork('192.0.2.19/24'),
            vrf=vrf1,
        )

        IPAddress.objects.create(address=IPNetwork('192.0.2.12/24'), vrf=vrf1)
        IPAddress.objects.create(address=IPNetwork('192.0.2.13/24'), vrf=vrf2)

        # Only the VRF 1 IP counts toward utilization.
        self.assertEqual(iprange.utilization, 1 / 10 * 100)

    def test_utilization_duplicate_ips_vrf(self):
        """
        Tests that identical IPs in a non-unique VRF count once toward range utilization.
        """
        vrf = VRF.objects.create(name='VRF 1', enforce_unique=False)
        iprange = IPRange.objects.create(
            start_address=IPNetwork('192.0.2.10/24'),
            end_address=IPNetwork('192.0.2.19/24'),
            vrf=vrf,
        )

        IPAddress.objects.bulk_create((
            IPAddress(address=IPNetwork('192.0.2.12/24'), vrf=vrf),
            IPAddress(address=IPNetwork('192.0.2.12/24'), vrf=vrf),
        ))

        self.assertEqual(iprange.utilization, 1 / 10 * 100)


class PrefixTestCase(TestCase):

    def assertAvailableIPCountMatchesIPSet(self, prefix):
        """
        Confirm that get_available_ip_count() matches get_available_ips().size for the supplied prefix.
        """
        self.assertEqual(prefix.get_available_ip_count(), prefix.get_available_ips().size)

    def test_family_string(self):
        # Test property when prefix is a string
        prefix = Prefix(prefix='10.0.0.0/8')
        self.assertEqual(prefix.family, 4)
        prefix_v6 = Prefix(prefix='2001:db8::/32')
        self.assertEqual(prefix_v6.family, 6)

    def test_mask_length_string(self):
        # Test property when prefix is a string
        prefix = Prefix(prefix='10.0.0.0/8')
        self.assertEqual(prefix.mask_length, 8)
        prefix_v6 = Prefix(prefix='2001:db8::/32')
        self.assertEqual(prefix_v6.mask_length, 32)

    def test_get_duplicates(self):
        prefixes = Prefix.objects.bulk_create((
            Prefix(prefix=IPNetwork('192.0.2.0/24')),
            Prefix(prefix=IPNetwork('192.0.2.0/24')),
            Prefix(prefix=IPNetwork('192.0.2.0/24')),
        ))
        duplicate_prefix_pks = [p.pk for p in prefixes[0].get_duplicates()]

        self.assertSetEqual(set(duplicate_prefix_pks), {prefixes[1].pk, prefixes[2].pk})

    def test_get_child_prefixes(self):
        vrfs = VRF.objects.bulk_create((
            VRF(name='VRF 1'),
            VRF(name='VRF 2'),
            VRF(name='VRF 3'),
        ))
        prefixes = Prefix.objects.bulk_create((
            Prefix(prefix=IPNetwork('10.0.0.0/16'), status=PrefixStatusChoices.STATUS_CONTAINER),
            Prefix(prefix=IPNetwork('10.0.0.0/24'), vrf=None),
            Prefix(prefix=IPNetwork('10.0.1.0/24'), vrf=vrfs[0]),
            Prefix(prefix=IPNetwork('10.0.2.0/24'), vrf=vrfs[1]),
            Prefix(prefix=IPNetwork('10.0.3.0/24'), vrf=vrfs[2]),
        ))
        child_prefix_pks = {p.pk for p in prefixes[0].get_child_prefixes()}

        # Global container should return all children
        self.assertSetEqual(child_prefix_pks, {prefixes[1].pk, prefixes[2].pk, prefixes[3].pk, prefixes[4].pk})

        prefixes[0].vrf = vrfs[0]
        prefixes[0].save()
        child_prefix_pks = {p.pk for p in prefixes[0].get_child_prefixes()}

        # VRF container is limited to its own VRF
        self.assertSetEqual(child_prefix_pks, {prefixes[2].pk})

    def test_get_child_ranges(self):
        prefix = Prefix(prefix='192.168.0.16/28')
        prefix.save()
        ranges = IPRange.objects.bulk_create(
            (
                # No overlap
                IPRange(
                    start_address=IPNetwork('192.168.0.1/24'), end_address=IPNetwork('192.168.0.10/24'), size=10
                ),
                # Partial overlap
                IPRange(
                    start_address=IPNetwork('192.168.0.11/24'), end_address=IPNetwork('192.168.0.17/24'), size=7
                ),
                # Full overlap
                IPRange(
                    start_address=IPNetwork('192.168.0.18/24'), end_address=IPNetwork('192.168.0.23/24'), size=6
                ),
                # Full overlap
                IPRange(
                    start_address=IPNetwork('192.168.0.24/24'), end_address=IPNetwork('192.168.0.30/24'), size=7
                ),
                # Partial overlap
                IPRange(
                    start_address=IPNetwork('192.168.0.31/24'), end_address=IPNetwork('192.168.0.40/24'), size=10
                ),
            )
        )

        child_ranges = prefix.get_child_ranges()

        self.assertEqual(len(child_ranges), 2)
        self.assertEqual(child_ranges[0], ranges[2])
        self.assertEqual(child_ranges[1], ranges[3])

    def test_get_child_ranges_other_family(self):
        """
        Tests that ranges of a different address family are not returned.
        """
        prefix = Prefix.objects.create(prefix=IPNetwork('192.168.0.16/28'))
        IPRange.objects.bulk_create((
            IPRange(
                start_address=IPNetwork('192.168.0.18/28'), end_address=IPNetwork('192.168.0.20/28'), size=3
            ),
            IPRange(start_address=IPNetwork('::1/64'), end_address=IPNetwork('::2/64'), size=2),
        ))

        child_ranges = prefix.get_child_ranges()

        self.assertEqual(len(child_ranges), 1)
        self.assertEqual(child_ranges[0].start_address, IPNetwork('192.168.0.18/28'))

    def test_get_child_ips(self):
        vrfs = VRF.objects.bulk_create((
            VRF(name='VRF 1'),
            VRF(name='VRF 2'),
            VRF(name='VRF 3'),
        ))
        parent_prefix = Prefix.objects.create(
            prefix=IPNetwork('10.0.0.0/16'), status=PrefixStatusChoices.STATUS_CONTAINER
        )
        ips = IPAddress.objects.bulk_create((
            IPAddress(address=IPNetwork('10.0.0.1/24'), vrf=None),
            IPAddress(address=IPNetwork('10.0.1.1/24'), vrf=vrfs[0]),
            IPAddress(address=IPNetwork('10.0.2.1/24'), vrf=vrfs[1]),
            IPAddress(address=IPNetwork('10.0.3.1/24'), vrf=vrfs[2]),
        ))
        child_ip_pks = {p.pk for p in parent_prefix.get_child_ips()}

        # Global container should return all children
        self.assertSetEqual(child_ip_pks, {ips[0].pk, ips[1].pk, ips[2].pk, ips[3].pk})

        parent_prefix.vrf = vrfs[0]
        parent_prefix.save()
        child_ip_pks = {p.pk for p in parent_prefix.get_child_ips()}

        # VRF container is limited to its own VRF
        self.assertSetEqual(child_ip_pks, {ips[1].pk})

    def test_get_available_prefixes(self):

        prefixes = Prefix.objects.bulk_create((
            Prefix(prefix=IPNetwork('10.0.0.0/16')),  # Parent prefix
            Prefix(prefix=IPNetwork('10.0.0.0/20')),
            Prefix(prefix=IPNetwork('10.0.32.0/20')),
            Prefix(prefix=IPNetwork('10.0.128.0/18')),
        ))
        missing_prefixes = IPSet([
            IPNetwork('10.0.16.0/20'),
            IPNetwork('10.0.48.0/20'),
            IPNetwork('10.0.64.0/18'),
            IPNetwork('10.0.192.0/18'),
        ])
        available_prefixes = prefixes[0].get_available_prefixes()

        self.assertEqual(available_prefixes, missing_prefixes)

    def test_get_available_ips(self):

        parent_prefix = Prefix.objects.create(prefix=IPNetwork('10.0.0.0/28'))
        IPAddress.objects.bulk_create((
            IPAddress(address=IPNetwork('10.0.0.1/26')),
            IPAddress(address=IPNetwork('10.0.0.3/26')),
            IPAddress(address=IPNetwork('10.0.0.5/26')),
            IPAddress(address=IPNetwork('10.0.0.7/26')),
        ))
        # Range is not marked as populated, so it doesn't count against available IP space
        IPRange.objects.create(
            start_address=IPNetwork('10.0.0.9/26'),
            end_address=IPNetwork('10.0.0.10/26')
        )
        # Populated range reduces available IP space
        IPRange.objects.create(
            start_address=IPNetwork('10.0.0.12/26'),
            end_address=IPNetwork('10.0.0.13/26'),
            mark_populated=True
        )
        missing_ips = IPSet([
            '10.0.0.2/32',
            '10.0.0.4/32',
            '10.0.0.6/32',
            '10.0.0.8/32',
            '10.0.0.9/32',
            '10.0.0.10/32',
            '10.0.0.11/32',
            '10.0.0.14/32',
        ])
        available_ips = parent_prefix.get_available_ips()

        self.assertEqual(available_ips, missing_ips)

    def test_iter_available_ips(self):
        """
        Tests that iter_available_ips() yields the same addresses as get_available_ips() in order.
        """
        parent_prefix = Prefix.objects.create(prefix=IPNetwork('10.0.0.0/28'))
        IPAddress.objects.bulk_create((
            IPAddress(address=IPNetwork('10.0.0.1/28')),
            IPAddress(address=IPNetwork('10.0.0.5/28')),
        ))
        IPRange.objects.create(
            start_address=IPNetwork('10.0.0.8/28'),
            end_address=IPNetwork('10.0.0.9/28'),
            mark_populated=True,
        )

        available_ips = list(parent_prefix.iter_available_ips())

        self.assertEqual(available_ips, sorted(parent_prefix.get_available_ips()))
        self.assertEqual(available_ips[0], netaddr.IPAddress('10.0.0.2'))
        self.assertEqual(available_ips[-1], netaddr.IPAddress('10.0.0.14'))

    def test_get_available_ips_ipv6(self):
        """
        Tests that the subnet-router anycast address is excluded and the last address included.
        """
        parent_prefix = Prefix.objects.create(prefix=IPNetwork('2001:db8::/126'))
        IPAddress.objects.create(address=IPNetwork('2001:db8::1/126'))

        self.assertEqual(parent_prefix.get_available_ips(), IPSet(['2001:db8::2/127']))

    def test_get_available_ips_pool(self):
        """
        Tests that pool prefixes include the network and broadcast addresses.
        """
        parent_prefix = Prefix.objects.create(prefix=IPNetwork('192.0.2.0/30'), is_pool=True)
        IPAddress.objects.create(address=IPNetwork('192.0.2.1/30'))

        self.assertEqual(parent_prefix.get_available_ips(), IPSet(['192.0.2.0/32', '192.0.2.2/31']))

    def test_available_ip_count_distinct_hosts(self):
        prefix = Prefix.objects.create(
            prefix=IPNetwork('192.0.2.0/29'),
            status=PrefixStatusChoices.STATUS_ACTIVE,
        )

        IPAddress.objects.bulk_create((
            IPAddress(address=IPNetwork('192.0.2.1/29')),
            IPAddress(address=IPNetwork('192.0.2.1/32')),
            IPAddress(address=IPNetwork('192.0.2.3/29')),
        ))

        # Usable hosts in /29: 6. Two unique hosts occupy .1 and .3.
        self.assertEqual(prefix.get_available_ip_count(), 4)
        self.assertAvailableIPCountMatchesIPSet(prefix)

    def test_available_ip_count_populated_ranges(self):
        prefix = Prefix.objects.create(
            prefix=IPNetwork('192.0.2.0/29'),
            status=PrefixStatusChoices.STATUS_ACTIVE,
        )

        IPAddress.objects.bulk_create((
            IPAddress(address=IPNetwork('192.0.2.1/29')),
            IPAddress(address=IPNetwork('192.0.2.3/29')),  # Inside the populated range; not double-counted.
        ))

        IPRange.objects.create(
            start_address=IPNetwork('192.0.2.3/29'),
            end_address=IPNetwork('192.0.2.4/29'),
            mark_populated=True,
        )

        # Usable 6, one IP outside the range at .1, populated range covers .3-.4.
        # Available: .2, .5, .6.
        self.assertEqual(prefix.get_available_ip_count(), 3)
        self.assertAvailableIPCountMatchesIPSet(prefix)

    def test_available_ip_count_ipv4_pool(self):
        prefix = Prefix.objects.create(
            prefix=IPNetwork('192.0.2.0/30'),
            status=PrefixStatusChoices.STATUS_ACTIVE,
            is_pool=True,
        )

        self.assertEqual(prefix.get_available_ip_count(), 4)
        self.assertAvailableIPCountMatchesIPSet(prefix)

    def test_available_ip_count_ipv4_non_pool(self):
        prefix = Prefix.objects.create(
            prefix=IPNetwork('192.0.2.0/30'),
            status=PrefixStatusChoices.STATUS_ACTIVE,
            is_pool=False,
        )

        self.assertEqual(prefix.get_available_ip_count(), 2)
        self.assertAvailableIPCountMatchesIPSet(prefix)

    def test_available_ip_count_ipv4_non_pool_ignores_unusable_ips(self):
        prefix = Prefix.objects.create(
            prefix=IPNetwork('192.0.2.0/30'),
            status=PrefixStatusChoices.STATUS_ACTIVE,
        )

        # Network and broadcast addresses are unusable for non-pool IPv4 prefixes;
        # an IP assigned to either must not reduce the available count.
        IPAddress.objects.create(address=IPNetwork('192.0.2.0/30'))
        IPAddress.objects.create(address=IPNetwork('192.0.2.3/30'))

        self.assertEqual(prefix.get_available_ip_count(), 2)
        self.assertEqual(prefix.get_first_available_ip(), '192.0.2.1/30')
        self.assertAvailableIPCountMatchesIPSet(prefix)

    def test_available_ip_count_ipv6(self):
        prefix = Prefix.objects.create(
            prefix=IPNetwork('2001:db8::/126'),
            status=PrefixStatusChoices.STATUS_ACTIVE,
        )

        # /126 has 4 addresses; normal IPv6 prefix excludes the first.
        self.assertEqual(prefix.get_available_ip_count(), 3)
        self.assertAvailableIPCountMatchesIPSet(prefix)

    def test_available_ip_count_ipv6_ignores_subnet_router_anycast(self):
        prefix = Prefix.objects.create(
            prefix=IPNetwork('2001:db8::/126'),
            status=PrefixStatusChoices.STATUS_ACTIVE,
        )

        # The subnet-router anycast (::) address is unusable for normal IPv6 prefixes;
        # an IP assigned there must not reduce the available count.
        IPAddress.objects.create(address=IPNetwork('2001:db8::/126'))

        self.assertEqual(prefix.get_available_ip_count(), 3)
        self.assertEqual(prefix.get_first_available_ip(), '2001:db8::1/126')
        self.assertAvailableIPCountMatchesIPSet(prefix)

    def test_available_ip_count_ipv6_127(self):
        prefix = Prefix.objects.create(
            prefix=IPNetwork('2001:db8::/127'),
            status=PrefixStatusChoices.STATUS_ACTIVE,
        )

        self.assertEqual(prefix.get_available_ip_count(), 2)
        self.assertAvailableIPCountMatchesIPSet(prefix)

    def test_available_ip_count_ipv6_populated_range(self):
        prefix = Prefix.objects.create(
            prefix=IPNetwork('2001:db8::/126'),
            status=PrefixStatusChoices.STATUS_ACTIVE,
        )

        IPRange.objects.create(
            start_address=IPNetwork('2001:db8::1/126'),
            end_address=IPNetwork('2001:db8::2/126'),
            mark_populated=True,
        )

        # Usable IPv6 hosts in /126: ::1, ::2, ::3. Populated: ::1-::2.
        self.assertEqual(prefix.get_available_ip_count(), 1)
        self.assertEqual(prefix.get_first_available_ip(), '2001:db8::3/126')
        self.assertAvailableIPCountMatchesIPSet(prefix)

    def test_available_ip_count_overlapping_ranges(self):
        prefix = Prefix.objects.create(
            prefix=IPNetwork('192.0.2.0/29'),
            status=PrefixStatusChoices.STATUS_ACTIVE,
        )

        IPRange.objects.create(
            start_address=IPNetwork('192.0.2.1/29'),
            end_address=IPNetwork('192.0.2.3/29'),
            mark_populated=True,
        )
        IPRange.objects.create(
            start_address=IPNetwork('192.0.2.2/29'),
            end_address=IPNetwork('192.0.2.4/29'),
            mark_populated=True,
        )

        # Usable hosts: .1-.6 => 6. Populated union: .1-.4 => 4. Available: .5-.6 => 2.
        self.assertEqual(prefix.get_available_ip_count(), 2)
        self.assertEqual(prefix.get_first_available_ip(), '192.0.2.5/29')
        self.assertAvailableIPCountMatchesIPSet(prefix)

    def test_available_ip_count_vrf(self):
        vrf1 = VRF.objects.create(name='VRF 1')
        vrf2 = VRF.objects.create(name='VRF 2')

        prefix = Prefix.objects.create(
            prefix=IPNetwork('192.0.2.0/29'),
            vrf=vrf1,
            status=PrefixStatusChoices.STATUS_ACTIVE,
        )

        IPAddress.objects.create(address=IPNetwork('192.0.2.1/29'), vrf=vrf1)
        IPAddress.objects.create(address=IPNetwork('192.0.2.2/29'), vrf=vrf2)

        # Usable .1-.6 => 6. Only the VRF 1 IP should count.
        self.assertEqual(prefix.get_available_ip_count(), 5)
        self.assertAvailableIPCountMatchesIPSet(prefix)

    def test_available_ip_count_vrf_ranges(self):
        vrf1 = VRF.objects.create(name='VRF 1')
        vrf2 = VRF.objects.create(name='VRF 2')

        prefix = Prefix.objects.create(
            prefix=IPNetwork('192.0.2.0/29'),
            vrf=vrf1,
            status=PrefixStatusChoices.STATUS_ACTIVE,
        )

        # Covers every usable host, but in a different VRF.
        IPRange.objects.create(
            start_address=IPNetwork('192.0.2.1/29'),
            end_address=IPNetwork('192.0.2.6/29'),
            vrf=vrf2,
            mark_populated=True,
        )

        self.assertEqual(prefix.get_available_ip_count(), 6)
        self.assertEqual(prefix.get_first_available_ip(), '192.0.2.1/29')
        self.assertAvailableIPCountMatchesIPSet(prefix)

    def test_available_ip_count_fully_populated(self):
        prefix = Prefix.objects.create(
            prefix=IPNetwork('192.0.2.0/30'),
            status=PrefixStatusChoices.STATUS_ACTIVE,
        )

        # Populated range covers every usable address (.1-.2 in a non-pool /30).
        IPRange.objects.create(
            start_address=IPNetwork('192.0.2.1/30'),
            end_address=IPNetwork('192.0.2.2/30'),
            mark_populated=True,
        )

        # Exercises the early-return paths that skip the child-IP count and
        # the host-stream iterator entirely.
        self.assertEqual(prefix.get_available_ip_count(), 0)
        self.assertIsNone(prefix.get_first_available_ip())
        self.assertAvailableIPCountMatchesIPSet(prefix)

    def test_available_ip_count_query_count(self):
        """
        Tests that the count runs one interval query plus exactly one host scan.
        """
        prefix = Prefix.objects.create(prefix=IPNetwork('192.0.2.0/24'))

        IPAddress.objects.bulk_create(
            IPAddress(address=IPNetwork(f'192.0.2.{i}/24')) for i in range(1, 11)
        )

        IPRange.objects.create(
            start_address=IPNetwork('192.0.2.20/24'),
            end_address=IPNetwork('192.0.2.29/24'),
            mark_populated=True,
        )

        with self.assertNumQueries(2):
            prefix.get_available_ip_count()

    def test_available_ip_count_container(self):
        prefix = Prefix.objects.create(
            prefix=IPNetwork('192.0.2.0/24'),
            status=PrefixStatusChoices.STATUS_CONTAINER,
        )

        # A child prefix exists but does not reduce the available IP count.
        Prefix.objects.create(
            prefix=IPNetwork('192.0.2.0/26'),
            status=PrefixStatusChoices.STATUS_ACTIVE,
        )

        self.assertEqual(prefix.get_available_ip_count(), 254)
        self.assertAvailableIPCountMatchesIPSet(prefix)

    def test_available_ip_count_container_vrf_duplicate_hosts(self):
        vrf1 = VRF.objects.create(name='VRF 1')
        vrf2 = VRF.objects.create(name='VRF 2')

        prefix = Prefix.objects.create(
            prefix=IPNetwork('192.0.2.0/24'),
            status=PrefixStatusChoices.STATUS_CONTAINER,
        )

        IPAddress.objects.create(address=IPNetwork('192.0.2.1/24'), vrf=vrf1)
        IPAddress.objects.create(address=IPNetwork('192.0.2.1/24'), vrf=vrf2)
        IPAddress.objects.create(address=IPNetwork('192.0.2.2/24'), vrf=vrf2)

        # A global container counts child IPs from all VRFs; the duplicate host
        # counts once. 254 usable - 2 distinct hosts.
        self.assertEqual(prefix.get_available_ip_count(), 252)
        self.assertAvailableIPCountMatchesIPSet(prefix)

    def test_available_ip_count_container_vrf_ip_in_populated_range(self):
        vrf1 = VRF.objects.create(name='VRF 1')

        prefix = Prefix.objects.create(
            prefix=IPNetwork('192.0.2.0/24'),
            status=PrefixStatusChoices.STATUS_CONTAINER,
        )

        IPRange.objects.create(
            start_address=IPNetwork('192.0.2.10/24'),
            end_address=IPNetwork('192.0.2.19/24'),
            mark_populated=True,
        )
        IPAddress.objects.create(address=IPNetwork('192.0.2.15/24'), vrf=vrf1)

        # The range covers 10 hosts; the VRF 1 IP inside it is not counted again.
        self.assertEqual(prefix.get_available_ip_count(), 244)
        self.assertAvailableIPCountMatchesIPSet(prefix)

    def test_get_first_available_prefix(self):

        prefixes = Prefix.objects.bulk_create((
            Prefix(prefix=IPNetwork('10.0.0.0/16')),  # Parent prefix
            Prefix(prefix=IPNetwork('10.0.0.0/24')),
            Prefix(prefix=IPNetwork('10.0.1.0/24')),
            Prefix(prefix=IPNetwork('10.0.2.0/24')),
        ))
        self.assertEqual(prefixes[0].get_first_available_prefix(), IPNetwork('10.0.3.0/24'))

        Prefix.objects.create(prefix=IPNetwork('10.0.3.0/24'))
        self.assertEqual(prefixes[0].get_first_available_prefix(), IPNetwork('10.0.4.0/22'))

    def test_get_first_available_ip(self):

        parent_prefix = Prefix.objects.create(prefix=IPNetwork('10.0.0.0/24'))
        IPAddress.objects.bulk_create((
            IPAddress(address=IPNetwork('10.0.0.1/24')),
            IPAddress(address=IPNetwork('10.0.0.2/24')),
            IPAddress(address=IPNetwork('10.0.0.3/24')),
        ))
        self.assertEqual(parent_prefix.get_first_available_ip(), '10.0.0.4/24')

        IPAddress.objects.create(address=IPNetwork('10.0.0.4/24'))
        self.assertEqual(parent_prefix.get_first_available_ip(), '10.0.0.5/24')

    def test_get_first_available_ip_ipv6(self):
        parent_prefix = Prefix.objects.create(prefix=IPNetwork('2001:db8:500::/64'))
        self.assertEqual(parent_prefix.get_first_available_ip(), '2001:db8:500::1/64')

    def test_get_first_available_ip_ipv6_rfc3627(self):
        parent_prefix = Prefix.objects.create(prefix=IPNetwork('2001:db8:500:4::/126'))
        self.assertEqual(parent_prefix.get_first_available_ip(), '2001:db8:500:4::1/126')

    def test_get_first_available_ip_ipv6_rfc6164(self):
        parent_prefix = Prefix.objects.create(prefix=IPNetwork('2001:db8:500:5::/127'))
        self.assertEqual(parent_prefix.get_first_available_ip(), '2001:db8:500:5::/127')

    def test_get_first_available_ip_ipv6_zero_address(self):
        prefix = Prefix.objects.create(
            prefix=IPNetwork('::/126'),
            status=PrefixStatusChoices.STATUS_ACTIVE,
        )

        # Normal IPv6 prefixes exclude the subnet-router anycast address ::.
        self.assertEqual(prefix.get_first_available_ip(), '::1/126')

    def test_get_first_available_ip_populated_ranges(self):
        prefix = Prefix.objects.create(
            prefix=IPNetwork('192.0.2.0/29'),
            status=PrefixStatusChoices.STATUS_ACTIVE,
        )

        IPAddress.objects.create(address=IPNetwork('192.0.2.1/29'))

        IPRange.objects.create(
            start_address=IPNetwork('192.0.2.2/29'),
            end_address=IPNetwork('192.0.2.3/29'),
            mark_populated=True,
        )

        self.assertEqual(prefix.get_first_available_ip(), '192.0.2.4/29')

    def test_get_first_available_ip_full(self):
        prefix = Prefix.objects.create(
            prefix=IPNetwork('192.0.2.0/30'),
            status=PrefixStatusChoices.STATUS_ACTIVE,
        )

        IPAddress.objects.create(address=IPNetwork('192.0.2.1/30'))
        IPAddress.objects.create(address=IPNetwork('192.0.2.2/30'))

        self.assertIsNone(prefix.get_first_available_ip())

    def test_get_utilization_container(self):
        prefixes = (
            Prefix(prefix=IPNetwork('10.0.0.0/24'), status=PrefixStatusChoices.STATUS_CONTAINER),
            Prefix(prefix=IPNetwork('10.0.0.0/26')),
            Prefix(prefix=IPNetwork('10.0.0.128/26')),
        )
        Prefix.objects.bulk_create(prefixes)
        self.assertEqual(prefixes[0].get_utilization(), 50)  # 50% utilization

    def test_get_utilization_noncontainer(self):
        prefix = Prefix.objects.create(
            prefix=IPNetwork('10.0.0.0/24'),
            status=PrefixStatusChoices.STATUS_ACTIVE
        )

        # Create 32 child IPs
        IPAddress.objects.bulk_create([
            IPAddress(address=IPNetwork(f'10.0.0.{i}/24')) for i in range(1, 33)
        ])
        self.assertEqual(prefix.get_utilization(), 32 / 254 * 100)  # ~12.5% utilization

        # Create a utilized child range with 32 additional IPs
        IPRange.objects.create(
            start_address=IPNetwork('10.0.0.33/24'),
            end_address=IPNetwork('10.0.0.64/24'),
            mark_utilized=True
        )
        self.assertEqual(prefix.get_utilization(), 64 / 254 * 100)  # ~25% utilization

    def test_get_utilization_distinct_hosts(self):
        prefix = Prefix.objects.create(
            prefix=IPNetwork('192.0.2.0/24'),
            status=PrefixStatusChoices.STATUS_ACTIVE,
        )

        IPAddress.objects.bulk_create((
            IPAddress(address=IPNetwork('192.0.2.10/24')),
            IPAddress(address=IPNetwork('192.0.2.10/32')),
            IPAddress(address=IPNetwork('192.0.2.11/24')),
        ))

        # Two unique occupied hosts over 254 usable IPv4 addresses.
        self.assertEqual(prefix.get_utilization(), 2 / 254 * 100)

    @override_settings(ENFORCE_GLOBAL_UNIQUE=False)
    def test_get_utilization_duplicate_ips_global(self):
        """
        Tests that identical global IPs permitted by disabled uniqueness count as one host.
        """
        prefix = Prefix.objects.create(
            prefix=IPNetwork('192.0.2.0/24'),
            status=PrefixStatusChoices.STATUS_ACTIVE,
        )

        IPAddress.objects.create(address=IPNetwork('192.0.2.10/24'))
        duplicate_ip = IPAddress(address=IPNetwork('192.0.2.10/24'))
        self.assertIsNone(duplicate_ip.clean())
        duplicate_ip.save()

        self.assertEqual(prefix.get_utilization(), 1 / 254 * 100)

    def test_get_utilization_duplicate_ips_vrf(self):
        """
        Tests that identical IPs in a non-unique VRF count as one host.
        """
        vrf = VRF.objects.create(name='VRF 1', enforce_unique=False)
        prefix = Prefix.objects.create(
            prefix=IPNetwork('192.0.2.0/24'),
            vrf=vrf,
            status=PrefixStatusChoices.STATUS_ACTIVE,
        )

        IPAddress.objects.create(address=IPNetwork('192.0.2.10/24'), vrf=vrf)
        duplicate_ip = IPAddress(address=IPNetwork('192.0.2.10/24'), vrf=vrf)
        self.assertIsNone(duplicate_ip.clean())
        duplicate_ip.save()

        self.assertEqual(prefix.get_utilization(), 1 / 254 * 100)

    def test_available_ip_count_duplicate_ips_vrf(self):
        """
        Tests that identical IPs in a non-unique VRF reduce availability once.
        """
        vrf = VRF.objects.create(name='VRF 1', enforce_unique=False)
        prefix = Prefix.objects.create(
            prefix=IPNetwork('192.0.2.0/29'),
            vrf=vrf,
            status=PrefixStatusChoices.STATUS_ACTIVE,
        )

        IPAddress.objects.bulk_create((
            IPAddress(address=IPNetwork('192.0.2.1/29'), vrf=vrf),
            IPAddress(address=IPNetwork('192.0.2.1/29'), vrf=vrf),
        ))

        # Usable hosts in /29: 6. The duplicate occupies a single host.
        self.assertEqual(prefix.get_available_ip_count(), 5)
        self.assertAvailableIPCountMatchesIPSet(prefix)

    @override_settings(ENFORCE_GLOBAL_UNIQUE=False)
    def test_get_ip_usage_summary_duplicate_ips_global(self):
        """
        Tests that the usage summary deduplicates identical global IPs in both values.
        """
        prefix = Prefix.objects.create(
            prefix=IPNetwork('192.0.2.0/24'),
            status=PrefixStatusChoices.STATUS_ACTIVE,
        )

        IPAddress.objects.bulk_create((
            IPAddress(address=IPNetwork('192.0.2.10/24')),
            IPAddress(address=IPNetwork('192.0.2.10/24')),
        ))

        summary = prefix.get_ip_usage_summary()

        self.assertEqual(summary['available_ip_count'], 253)
        self.assertEqual(summary['utilization'], 1 / 254 * 100)

    def test_get_utilization_utilized_ranges(self):
        prefix = Prefix.objects.create(
            prefix=IPNetwork('192.0.2.0/24'),
            status=PrefixStatusChoices.STATUS_ACTIVE,
        )

        IPRange.objects.create(
            start_address=IPNetwork('192.0.2.10/24'),
            end_address=IPNetwork('192.0.2.19/24'),
            mark_utilized=True,
        )

        IPAddress.objects.bulk_create((
            IPAddress(address=IPNetwork('192.0.2.1/24')),
            IPAddress(address=IPNetwork('192.0.2.10/24')),
            IPAddress(address=IPNetwork('192.0.2.11/24')),
            IPAddress(address=IPNetwork('192.0.2.20/24')),
        ))

        # Utilized range contributes 10 hosts; IPs inside the range are not double-counted.
        # Outside IPs: .1 and .20 => 2 more.
        self.assertEqual(prefix.get_utilization(), 12 / 254 * 100)

    def test_get_utilization_overlapping_utilized_ranges(self):
        prefix = Prefix.objects.create(
            prefix=IPNetwork('192.0.2.0/24'),
            status=PrefixStatusChoices.STATUS_ACTIVE,
        )

        IPRange.objects.create(
            start_address=IPNetwork('192.0.2.10/24'),
            end_address=IPNetwork('192.0.2.19/24'),
            mark_utilized=True,
        )
        IPRange.objects.create(
            start_address=IPNetwork('192.0.2.15/24'),
            end_address=IPNetwork('192.0.2.24/24'),
            mark_utilized=True,
        )

        # Union is .10-.24 => 15 hosts, not 20.
        self.assertEqual(prefix.get_utilization(), 15 / 254 * 100)

    def test_get_utilization_fully_utilized_range(self):
        prefix = Prefix.objects.create(
            prefix=IPNetwork('192.0.2.0/24'),
            status=PrefixStatusChoices.STATUS_ACTIVE,
        )

        # Utilized range covers every usable host (.1-.254 in a non-pool /24).
        IPRange.objects.create(
            start_address=IPNetwork('192.0.2.1/24'),
            end_address=IPNetwork('192.0.2.254/24'),
            mark_utilized=True,
        )

        # Exercises the early-return path that skips the child-IP count entirely.
        self.assertEqual(prefix.get_utilization(), 100)

    def test_get_utilization_ipv6_utilized_range(self):
        prefix = Prefix.objects.create(
            prefix=IPNetwork('2001:db8::/126'),
            status=PrefixStatusChoices.STATUS_ACTIVE,
        )

        IPRange.objects.create(
            start_address=IPNetwork('2001:db8::1/126'),
            end_address=IPNetwork('2001:db8::2/126'),
            mark_utilized=True,
        )

        self.assertEqual(prefix.get_utilization(), 2 / 4 * 100)

    def test_get_utilization_vrf(self):
        vrf1 = VRF.objects.create(name='VRF 1')
        vrf2 = VRF.objects.create(name='VRF 2')

        prefix = Prefix.objects.create(
            prefix=IPNetwork('192.0.2.0/24'),
            vrf=vrf1,
            status=PrefixStatusChoices.STATUS_ACTIVE,
        )

        IPAddress.objects.create(address=IPNetwork('192.0.2.1/24'), vrf=vrf1)
        IPAddress.objects.create(address=IPNetwork('192.0.2.15/24'), vrf=vrf1)
        IPAddress.objects.create(address=IPNetwork('192.0.2.2/24'), vrf=vrf2)
        IPRange.objects.create(
            start_address=IPNetwork('192.0.2.10/24'),
            end_address=IPNetwork('192.0.2.19/24'),
            vrf=vrf2,
            mark_utilized=True,
        )

        # VRF 2 objects are ignored entirely; the VRF 1 IP at .15 still counts even
        # though it falls inside the VRF 2 range's host span (exclusion intervals are
        # built only from same-VRF ranges).
        self.assertEqual(prefix.get_utilization(), 2 / 254 * 100)

    def test_get_utilization_query_count(self):
        """
        Tests that utilization for a non-container prefix uses two queries.
        """
        prefix = Prefix.objects.create(prefix=IPNetwork('192.0.2.0/24'))

        with self.assertNumQueries(2):
            prefix.get_utilization()

    def test_get_ip_usage_summary(self):
        """
        Tests that the combined summary matches the independent methods.
        """
        prefix = Prefix.objects.create(prefix=IPNetwork('192.0.2.0/24'))
        IPAddress.objects.bulk_create((
            IPAddress(address=IPNetwork('192.0.2.1/24')),
            IPAddress(address=IPNetwork('192.0.2.2/24')),
        ))
        IPRange.objects.create(
            start_address=IPNetwork('192.0.2.10/24'),
            end_address=IPNetwork('192.0.2.19/24'),
            mark_utilized=True,
        )
        IPRange.objects.create(
            start_address=IPNetwork('192.0.2.30/24'),
            end_address=IPNetwork('192.0.2.39/24'),
            mark_populated=True,
        )

        summary = prefix.get_ip_usage_summary()

        self.assertEqual(summary['available_ip_count'], prefix.get_available_ip_count())
        self.assertEqual(summary['utilization'], prefix.get_utilization())

    def test_get_ip_usage_summary_query_count(self):
        """
        Tests that the combined summary uses a single distinct-host scan (three queries).
        """
        prefix = Prefix.objects.create(prefix=IPNetwork('192.0.2.0/24'))

        with self.assertNumQueries(3):
            prefix.get_ip_usage_summary()

    def test_get_ip_usage_summary_container(self):
        """
        Tests that the summary delegates to the independent methods for containers.
        """
        container = Prefix.objects.create(
            prefix=IPNetwork('192.0.2.0/24'),
            status=PrefixStatusChoices.STATUS_CONTAINER,
        )

        summary = container.get_ip_usage_summary()

        self.assertEqual(summary['available_ip_count'], container.get_available_ip_count())
        self.assertEqual(summary['utilization'], container.get_utilization())

    def test_get_ip_usage_summary_mark_utilized(self):
        """
        Tests that a marked-utilized prefix reports 100% utilization in the summary.
        """
        prefix = Prefix.objects.create(prefix=IPNetwork('192.0.2.0/24'), mark_utilized=True)

        summary = prefix.get_ip_usage_summary()

        self.assertEqual(summary['utilization'], 100)
        self.assertEqual(summary['available_ip_count'], prefix.get_available_ip_count())

    def test_usable_size(self):
        self.assertEqual(Prefix(prefix=IPNetwork('192.0.2.0/24')).usable_size, 254)
        self.assertEqual(Prefix(prefix=IPNetwork('192.0.2.0/24'), is_pool=True).usable_size, 256)
        self.assertEqual(Prefix(prefix=IPNetwork('2001:db8::/126')).usable_size, 3)

    def test_usable_ip_bounds_string_prefix(self):
        """
        Tests that usable bounds are computed for a string-assigned prefix.
        """
        first_ip, last_ip = Prefix(prefix='192.0.2.0/24').usable_ip_bounds

        self.assertEqual(first_ip, netaddr.IPAddress('192.0.2.1'))
        self.assertEqual(last_ip, netaddr.IPAddress('192.0.2.254'))

    #
    # Uniqueness enforcement tests
    #

    @override_settings(ENFORCE_GLOBAL_UNIQUE=False)
    def test_duplicate_global(self):
        Prefix.objects.create(prefix=IPNetwork('192.0.2.0/24'))
        duplicate_prefix = Prefix(prefix=IPNetwork('192.0.2.0/24'))
        self.assertIsNone(duplicate_prefix.clean())

    def test_duplicate_global_unique(self):
        Prefix.objects.create(prefix=IPNetwork('192.0.2.0/24'))
        duplicate_prefix = Prefix(prefix=IPNetwork('192.0.2.0/24'))
        self.assertRaises(ValidationError, duplicate_prefix.clean)

    def test_duplicate_vrf(self):
        vrf = VRF.objects.create(name='Test', rd='1:1', enforce_unique=False)
        Prefix.objects.create(vrf=vrf, prefix=IPNetwork('192.0.2.0/24'))
        duplicate_prefix = Prefix(vrf=vrf, prefix=IPNetwork('192.0.2.0/24'))
        self.assertIsNone(duplicate_prefix.clean())

    def test_duplicate_vrf_unique(self):
        vrf = VRF.objects.create(name='Test', rd='1:1', enforce_unique=True)
        Prefix.objects.create(vrf=vrf, prefix=IPNetwork('192.0.2.0/24'))
        duplicate_prefix = Prefix(vrf=vrf, prefix=IPNetwork('192.0.2.0/24'))
        self.assertRaises(ValidationError, duplicate_prefix.clean)

    # Regression test for #22682
    def test_deleting_site_group_does_not_delete_prefix_scoped_to_member_site(self):
        sitegroup = SiteGroup.objects.create(name='Site Group 1', slug='site-group-1')
        site = Site.objects.create(name='Site 1', slug='site-1', group=sitegroup)
        prefix = Prefix.objects.create(prefix=IPNetwork('10.0.0.0/24'), scope=site)

        sitegroup.delete()

        site.refresh_from_db()
        prefix.refresh_from_db()
        self.assertIsNone(site.group)
        self.assertEqual(prefix.scope, site)
        self.assertIsNone(prefix._site_group_id)

    # Regression test for #22682
    def test_deleting_region_does_not_delete_prefix_scoped_to_member_site(self):
        region = Region.objects.create(name='Region 1', slug='region-1')
        site = Site.objects.create(name='Site 2', slug='site-2', region=region)
        prefix = Prefix.objects.create(prefix=IPNetwork('10.0.1.0/24'), scope=site)

        region.delete()

        site.refresh_from_db()
        prefix.refresh_from_db()
        self.assertIsNone(site.region)
        self.assertEqual(prefix.scope, site)
        self.assertIsNone(prefix._region_id)

    # Regression test for #22682
    def test_deleting_site_group_does_not_delete_prefix_scoped_to_member_location(self):
        sitegroup = SiteGroup.objects.create(name='Site Group 3', slug='site-group-3')
        site = Site.objects.create(name='Site 3', slug='site-3', group=sitegroup)
        location = Location.objects.create(name='Location 1', slug='location-1', site=site)
        prefix = Prefix.objects.create(prefix=IPNetwork('10.0.4.0/24'), scope=location)

        sitegroup.delete()

        site.refresh_from_db()
        prefix.refresh_from_db()
        self.assertIsNone(site.group)
        self.assertEqual(prefix.scope, location)
        self.assertIsNone(prefix._site_group_id)

    # Regression test for #22682
    def test_deleting_region_does_not_delete_prefix_scoped_to_member_location(self):
        region = Region.objects.create(name='Region 3', slug='region-3')
        site = Site.objects.create(name='Site 4', slug='site-4', region=region)
        location = Location.objects.create(name='Location 2', slug='location-2', site=site)
        prefix = Prefix.objects.create(prefix=IPNetwork('10.0.5.0/24'), scope=location)

        region.delete()

        site.refresh_from_db()
        prefix.refresh_from_db()
        self.assertIsNone(site.region)
        self.assertEqual(prefix.scope, location)
        self.assertIsNone(prefix._region_id)

    def test_deleting_site_group_scoped_to_it_directly_still_deletes_prefix(self):
        sitegroup = SiteGroup.objects.create(name='Site Group 2', slug='site-group-2')
        prefix = Prefix.objects.create(prefix=IPNetwork('10.0.2.0/24'), scope=sitegroup)

        sitegroup.delete()

        self.assertFalse(Prefix.objects.filter(pk=prefix.pk).exists())

    def test_deleting_region_scoped_to_it_directly_still_deletes_prefix(self):
        region = Region.objects.create(name='Region 2', slug='region-2')
        prefix = Prefix.objects.create(prefix=IPNetwork('10.0.3.0/24'), scope=region)

        region.delete()

        self.assertFalse(Prefix.objects.filter(pk=prefix.pk).exists())


class PrefixHierarchyTestCase(TestCase):
    """
    Test the automatic updating of depth and child count in response to changes made within
    the prefix hierarchy.
    """
    @classmethod
    def setUpTestData(cls):

        prefixes = (

            # IPv4
            Prefix(prefix='10.0.0.0/8', _depth=0, _children=2),
            Prefix(prefix='10.0.0.0/16', _depth=1, _children=1),
            Prefix(prefix='10.0.0.0/24', _depth=2, _children=0),

            # IPv6
            Prefix(prefix='2001:db8::/32', _depth=0, _children=2),
            Prefix(prefix='2001:db8::/40', _depth=1, _children=1),
            Prefix(prefix='2001:db8::/48', _depth=2, _children=0),

        )
        Prefix.objects.bulk_create(prefixes)

    def test_create_prefix4(self):
        # Create 10.0.0.0/12
        Prefix(prefix='10.0.0.0/12').save()

        prefixes = Prefix.objects.filter(prefix__family=4)
        self.assertEqual(prefixes[0].prefix, IPNetwork('10.0.0.0/8'))
        self.assertEqual(prefixes[0]._depth, 0)
        self.assertEqual(prefixes[0]._children, 3)
        self.assertEqual(prefixes[1].prefix, IPNetwork('10.0.0.0/12'))
        self.assertEqual(prefixes[1]._depth, 1)
        self.assertEqual(prefixes[1]._children, 2)
        self.assertEqual(prefixes[2].prefix, IPNetwork('10.0.0.0/16'))
        self.assertEqual(prefixes[2]._depth, 2)
        self.assertEqual(prefixes[2]._children, 1)
        self.assertEqual(prefixes[3].prefix, IPNetwork('10.0.0.0/24'))
        self.assertEqual(prefixes[3]._depth, 3)
        self.assertEqual(prefixes[3]._children, 0)

    def test_create_prefix6(self):
        # Create 2001:db8::/36
        Prefix(prefix='2001:db8::/36').save()

        prefixes = Prefix.objects.filter(prefix__family=6)
        self.assertEqual(prefixes[0].prefix, IPNetwork('2001:db8::/32'))
        self.assertEqual(prefixes[0]._depth, 0)
        self.assertEqual(prefixes[0]._children, 3)
        self.assertEqual(prefixes[1].prefix, IPNetwork('2001:db8::/36'))
        self.assertEqual(prefixes[1]._depth, 1)
        self.assertEqual(prefixes[1]._children, 2)
        self.assertEqual(prefixes[2].prefix, IPNetwork('2001:db8::/40'))
        self.assertEqual(prefixes[2]._depth, 2)
        self.assertEqual(prefixes[2]._children, 1)
        self.assertEqual(prefixes[3].prefix, IPNetwork('2001:db8::/48'))
        self.assertEqual(prefixes[3]._depth, 3)
        self.assertEqual(prefixes[3]._children, 0)

    def test_update_prefix4(self):
        # Change 10.0.0.0/24 to 10.0.0.0/12
        p = Prefix.objects.get(prefix='10.0.0.0/24')
        p.prefix = '10.0.0.0/12'
        p.save()

        prefixes = Prefix.objects.filter(prefix__family=4)
        self.assertEqual(prefixes[0].prefix, IPNetwork('10.0.0.0/8'))
        self.assertEqual(prefixes[0]._depth, 0)
        self.assertEqual(prefixes[0]._children, 2)
        self.assertEqual(prefixes[1].prefix, IPNetwork('10.0.0.0/12'))
        self.assertEqual(prefixes[1]._depth, 1)
        self.assertEqual(prefixes[1]._children, 1)
        self.assertEqual(prefixes[2].prefix, IPNetwork('10.0.0.0/16'))
        self.assertEqual(prefixes[2]._depth, 2)
        self.assertEqual(prefixes[2]._children, 0)

    def test_update_prefix6(self):
        # Change 2001:db8::/48 to 2001:db8::/36
        p = Prefix.objects.get(prefix='2001:db8::/48')
        p.prefix = '2001:db8::/36'
        p.save()

        prefixes = Prefix.objects.filter(prefix__family=6)
        self.assertEqual(prefixes[0].prefix, IPNetwork('2001:db8::/32'))
        self.assertEqual(prefixes[0]._depth, 0)
        self.assertEqual(prefixes[0]._children, 2)
        self.assertEqual(prefixes[1].prefix, IPNetwork('2001:db8::/36'))
        self.assertEqual(prefixes[1]._depth, 1)
        self.assertEqual(prefixes[1]._children, 1)
        self.assertEqual(prefixes[2].prefix, IPNetwork('2001:db8::/40'))
        self.assertEqual(prefixes[2]._depth, 2)
        self.assertEqual(prefixes[2]._children, 0)

    def test_update_prefix_vrf4(self):
        vrf = VRF(name='VRF A')
        vrf.save()

        # Move 10.0.0.0/16 to a VRF
        p = Prefix.objects.get(prefix='10.0.0.0/16')
        p.vrf = vrf
        p.save()

        prefixes = Prefix.objects.filter(vrf__isnull=True, prefix__family=4)
        self.assertEqual(prefixes[0].prefix, IPNetwork('10.0.0.0/8'))
        self.assertEqual(prefixes[0]._depth, 0)
        self.assertEqual(prefixes[0]._children, 1)
        self.assertEqual(prefixes[1].prefix, IPNetwork('10.0.0.0/24'))
        self.assertEqual(prefixes[1]._depth, 1)
        self.assertEqual(prefixes[1]._children, 0)

        prefixes = Prefix.objects.filter(vrf=vrf)
        self.assertEqual(prefixes[0].prefix, IPNetwork('10.0.0.0/16'))
        self.assertEqual(prefixes[0]._depth, 0)
        self.assertEqual(prefixes[0]._children, 0)

    def test_update_prefix_vrf6(self):
        vrf = VRF(name='VRF A')
        vrf.save()

        # Move 2001:db8::/40 to a VRF
        p = Prefix.objects.get(prefix='2001:db8::/40')
        p.vrf = vrf
        p.save()

        prefixes = Prefix.objects.filter(vrf__isnull=True, prefix__family=6)
        self.assertEqual(prefixes[0].prefix, IPNetwork('2001:db8::/32'))
        self.assertEqual(prefixes[0]._depth, 0)
        self.assertEqual(prefixes[0]._children, 1)
        self.assertEqual(prefixes[1].prefix, IPNetwork('2001:db8::/48'))
        self.assertEqual(prefixes[1]._depth, 1)
        self.assertEqual(prefixes[1]._children, 0)

        prefixes = Prefix.objects.filter(vrf=vrf)
        self.assertEqual(prefixes[0].prefix, IPNetwork('2001:db8::/40'))
        self.assertEqual(prefixes[0]._depth, 0)
        self.assertEqual(prefixes[0]._children, 0)

    def test_delete_prefix4(self):
        # Delete 10.0.0.0/16
        Prefix.objects.filter(prefix='10.0.0.0/16').delete()

        prefixes = Prefix.objects.filter(prefix__family=4)
        self.assertEqual(prefixes[0].prefix, IPNetwork('10.0.0.0/8'))
        self.assertEqual(prefixes[0]._depth, 0)
        self.assertEqual(prefixes[0]._children, 1)
        self.assertEqual(prefixes[1].prefix, IPNetwork('10.0.0.0/24'))
        self.assertEqual(prefixes[1]._depth, 1)
        self.assertEqual(prefixes[1]._children, 0)

    def test_delete_prefix6(self):
        # Delete 2001:db8::/40
        Prefix.objects.filter(prefix='2001:db8::/40').delete()

        prefixes = Prefix.objects.filter(prefix__family=6)
        self.assertEqual(prefixes[0].prefix, IPNetwork('2001:db8::/32'))
        self.assertEqual(prefixes[0]._depth, 0)
        self.assertEqual(prefixes[0]._children, 1)
        self.assertEqual(prefixes[1].prefix, IPNetwork('2001:db8::/48'))
        self.assertEqual(prefixes[1]._depth, 1)
        self.assertEqual(prefixes[1]._children, 0)

    def test_duplicate_prefix4(self):
        # Duplicate 10.0.0.0/16
        Prefix(prefix='10.0.0.0/16').save()

        prefixes = Prefix.objects.filter(prefix__family=4)
        self.assertEqual(prefixes[0].prefix, IPNetwork('10.0.0.0/8'))
        self.assertEqual(prefixes[0]._depth, 0)
        self.assertEqual(prefixes[0]._children, 3)
        self.assertEqual(prefixes[1].prefix, IPNetwork('10.0.0.0/16'))
        self.assertEqual(prefixes[1]._depth, 1)
        self.assertEqual(prefixes[1]._children, 1)
        self.assertEqual(prefixes[2].prefix, IPNetwork('10.0.0.0/16'))
        self.assertEqual(prefixes[2]._depth, 1)
        self.assertEqual(prefixes[2]._children, 1)
        self.assertEqual(prefixes[3].prefix, IPNetwork('10.0.0.0/24'))
        self.assertEqual(prefixes[3]._depth, 2)
        self.assertEqual(prefixes[3]._children, 0)

    def test_duplicate_prefix6(self):
        # Duplicate 2001:db8::/40
        Prefix(prefix='2001:db8::/40').save()

        prefixes = Prefix.objects.filter(prefix__family=6)
        self.assertEqual(prefixes[0].prefix, IPNetwork('2001:db8::/32'))
        self.assertEqual(prefixes[0]._depth, 0)
        self.assertEqual(prefixes[0]._children, 3)
        self.assertEqual(prefixes[1].prefix, IPNetwork('2001:db8::/40'))
        self.assertEqual(prefixes[1]._depth, 1)
        self.assertEqual(prefixes[1]._children, 1)
        self.assertEqual(prefixes[2].prefix, IPNetwork('2001:db8::/40'))
        self.assertEqual(prefixes[2]._depth, 1)
        self.assertEqual(prefixes[2]._children, 1)
        self.assertEqual(prefixes[3].prefix, IPNetwork('2001:db8::/48'))
        self.assertEqual(prefixes[3]._depth, 2)
        self.assertEqual(prefixes[3]._children, 0)

    def test_rebuild_prefixes_accepts_vrf_identifier(self):
        # None means "global table". Wipe the precomputed hierarchy so the rebuild is observable.
        Prefix.objects.update(_depth=0, _children=0)

        rebuild_prefixes(None)

        top = Prefix.objects.get(prefix='10.0.0.0/8')
        mid = Prefix.objects.get(prefix='10.0.0.0/16')
        leaf = Prefix.objects.get(prefix='10.0.0.0/24')
        self.assertEqual((top._depth, top._children), (0, 2))
        self.assertEqual((mid._depth, mid._children), (1, 1))
        self.assertEqual((leaf._depth, leaf._children), (2, 0))

    def test_rebuild_prefixes_accepts_vrf_pk(self):
        # A VRF pk filters to that VRF's prefixes.
        vrf = VRF.objects.create(name='VRF 1')
        Prefix.objects.create(prefix=IPNetwork('192.0.2.0/24'), vrf=vrf)
        Prefix.objects.create(prefix=IPNetwork('192.0.2.0/25'), vrf=vrf)

        # Reset depth/children so the rebuild has something to restore.
        Prefix.objects.filter(vrf=vrf).update(_depth=0, _children=0)

        rebuild_prefixes(vrf.pk)

        parent = Prefix.objects.get(prefix='192.0.2.0/24', vrf=vrf)
        child = Prefix.objects.get(prefix='192.0.2.0/25', vrf=vrf)
        self.assertEqual((parent._depth, parent._children), (0, 1))
        self.assertEqual((child._depth, child._children), (1, 0))


class IPAddressTestCase(TestCase):

    def test_family_string(self):
        # Test property when address is a string
        ip = IPAddress(address='10.0.0.1/24')
        self.assertEqual(ip.family, 4)
        ip_v6 = IPAddress(address='2001:db8::1/64')
        self.assertEqual(ip_v6.family, 6)

    def test_get_duplicates(self):
        ips = IPAddress.objects.bulk_create((
            IPAddress(address=IPNetwork('192.0.2.1/24')),
            IPAddress(address=IPNetwork('192.0.2.1/24')),
            IPAddress(address=IPNetwork('192.0.2.1/24')),
        ))
        duplicate_ip_pks = [p.pk for p in ips[0].get_duplicates()]

        self.assertSetEqual(set(duplicate_ip_pks), {ips[1].pk, ips[2].pk})

    #
    # Uniqueness enforcement tests
    #

    @override_settings(ENFORCE_GLOBAL_UNIQUE=False)
    def test_duplicate_global(self):
        IPAddress.objects.create(address=IPNetwork('192.0.2.1/24'))
        duplicate_ip = IPAddress(address=IPNetwork('192.0.2.1/24'))
        self.assertIsNone(duplicate_ip.clean())

    def test_duplicate_global_unique(self):
        IPAddress.objects.create(address=IPNetwork('192.0.2.1/24'))
        duplicate_ip = IPAddress(address=IPNetwork('192.0.2.1/24'))
        self.assertRaises(ValidationError, duplicate_ip.clean)

    def test_duplicate_vrf(self):
        vrf = VRF.objects.create(name='Test', rd='1:1', enforce_unique=False)
        IPAddress.objects.create(vrf=vrf, address=IPNetwork('192.0.2.1/24'))
        duplicate_ip = IPAddress(vrf=vrf, address=IPNetwork('192.0.2.1/24'))
        self.assertIsNone(duplicate_ip.clean())

    def test_duplicate_vrf_unique(self):
        vrf = VRF.objects.create(name='Test', rd='1:1', enforce_unique=True)
        IPAddress.objects.create(vrf=vrf, address=IPNetwork('192.0.2.1/24'))
        duplicate_ip = IPAddress(vrf=vrf, address=IPNetwork('192.0.2.1/24'))
        self.assertRaises(ValidationError, duplicate_ip.clean)

    def test_duplicate_nonunique_nonrole_role(self):
        IPAddress.objects.create(address=IPNetwork('192.0.2.1/24'))
        duplicate_ip = IPAddress(address=IPNetwork('192.0.2.1/24'), role=IPAddressRoleChoices.ROLE_VIP)
        self.assertRaises(ValidationError, duplicate_ip.clean)

    def test_duplicate_nonunique_role_nonrole(self):
        IPAddress.objects.create(address=IPNetwork('192.0.2.1/24'), role=IPAddressRoleChoices.ROLE_VIP)
        duplicate_ip = IPAddress(address=IPNetwork('192.0.2.1/24'))
        self.assertRaises(ValidationError, duplicate_ip.clean)

    def test_duplicate_nonunique_role(self):
        IPAddress.objects.create(address=IPNetwork('192.0.2.1/24'), role=IPAddressRoleChoices.ROLE_VIP)
        IPAddress.objects.create(address=IPNetwork('192.0.2.1/24'), role=IPAddressRoleChoices.ROLE_VIP)

    #
    # Range validation
    #

    def test_create_ip_in_unpopulated_range(self):
        IPRange.objects.create(
            start_address=IPNetwork('192.0.2.1/24'),
            end_address=IPNetwork('192.0.2.100/24')
        )
        ip = IPAddress(address=IPNetwork('192.0.2.10/24'))
        ip.full_clean()

    def test_create_ip_in_populated_range(self):
        IPRange.objects.create(
            start_address=IPNetwork('192.0.2.1/24'),
            end_address=IPNetwork('192.0.2.100/24'),
            mark_populated=True
        )
        ip = IPAddress(address=IPNetwork('192.0.2.10/24'))
        self.assertRaises(ValidationError, ip.full_clean)

    def test_mark_populated_single_address_range_blocks_ip(self):
        # A single-address range with mark_populated=True must still block creation
        # of an IPAddress at the same host with the same mask.
        IPRange.objects.create(
            start_address=IPNetwork('192.0.2.10/24'),
            end_address=IPNetwork('192.0.2.10/24'),
            mark_populated=True,
        )
        ipaddress = IPAddress(address=IPNetwork('192.0.2.10/24'))

        with self.assertRaisesMessage(ValidationError, 'Cannot create IP address'):
            ipaddress.clean()

    def test_populated_range_blocks_ip_with_different_mask(self):
        # The populated-range check compares by host portion, so a different mask
        # must not let an IPAddress slip past validation.
        IPRange.objects.create(
            start_address=IPNetwork('10.0.0.2/24'),
            end_address=IPNetwork('10.0.0.254/24'),
            mark_populated=True,
        )

        ip = IPAddress(address=IPNetwork('10.0.0.2/32'))

        with self.assertRaises(ValidationError):
            ip.full_clean()


class VLANGroupTestCase(TestCase):

    @classmethod
    def setUpTestData(cls):
        vlangroup = VLANGroup.objects.create(
            name='VLAN Group 1',
            slug='vlan-group-1',
            vid_ranges=string_to_ranges('100-199'),
        )
        VLAN.objects.bulk_create((
            VLAN(name='VLAN 100', vid=100, group=vlangroup),
            VLAN(name='VLAN 101', vid=101, group=vlangroup),
            VLAN(name='VLAN 102', vid=102, group=vlangroup),
            VLAN(name='VLAN 103', vid=103, group=vlangroup),
        ))

    def test_get_available_vids(self):
        vlangroup = VLANGroup.objects.first()
        child_vids = VLAN.objects.filter(group=vlangroup).values_list('vid', flat=True)
        self.assertEqual(len(child_vids), 4)

        available_vids = vlangroup.get_available_vids()
        self.assertListEqual(available_vids, list(range(104, 200)))

    def test_get_available_vids_with_inclusive_ranges(self):
        vlangroup = VLANGroup.objects.create(
            name='VLAN Group 2',
            slug='vlan-group-2',
            vid_ranges=[NumericRange(200, 204, bounds='[]')],
        )
        VLAN.objects.create(name='VLAN 200', vid=200, group=vlangroup)

        # Re-assign with non-canonical bounds to exercise the unsaved in-memory path.
        vlangroup.vid_ranges = [NumericRange(200, 204, bounds='[]')]
        self.assertListEqual(vlangroup.get_available_vids(), [201, 202, 203, 204])

    def test_get_next_available_vid(self):
        vlangroup = VLANGroup.objects.first()
        self.assertEqual(vlangroup.get_next_available_vid(), 104)

        VLAN.objects.create(name='VLAN 104', vid=104, group=vlangroup)
        self.assertEqual(vlangroup.get_next_available_vid(), 105)

    def test_vid_validation(self):
        vlangroup = VLANGroup.objects.first()

        vlan = VLAN(vid=1, name='VLAN 1', group=vlangroup)
        with self.assertRaises(ValidationError):
            vlan.full_clean()

        vlan = VLAN(vid=109, name='VLAN 109', group=vlangroup)
        vlan.full_clean()

    def test_overlapping_vlan(self):
        vlangroup = VLANGroup(
            name='VLAN Group 1',
            slug='vlan-group-1',
            vid_ranges=string_to_ranges('2-4,3-5'),
        )
        with self.assertRaises(ValidationError):
            vlangroup.full_clean()

        # make sure single vlan range works
        vlangroup.vid_ranges = string_to_ranges('2-2')
        vlangroup.full_clean()
        vlangroup.save()

    def test_total_vlan_ids(self):
        vlangroup = VLANGroup.objects.first()
        self.assertEqual(vlangroup.total_vlan_ids, 100)

    def test_total_vlan_ids_with_inclusive_ranges(self):
        test_cases = (
            (
                NumericRange(100, 199, bounds='[]'),
                100,
                NumericRange(100, 200, bounds='[)'),
            ),
            (
                NumericRange(100, 100, bounds='[]'),
                1,
                NumericRange(100, 101, bounds='[)'),
            ),
            (
                NumericRange(99, 199, bounds='(]'),
                100,
                NumericRange(100, 200, bounds='[)'),
            ),
        )

        for i, (vid_range, total_vlan_ids, normalized_range) in enumerate(test_cases, start=1):
            vlangroup = VLANGroup(
                name=f'VLAN Group Inclusive {i}',
                slug=f'vlan-group-inclusive-{i}',
                vid_ranges=[vid_range],
            )

            vlangroup.full_clean()
            vlangroup.save()
            self.assertEqual(vlangroup.vid_ranges, [normalized_range])
            self.assertEqual(vlangroup.total_vlan_ids, total_vlan_ids)

    def test_total_vlan_ids_with_inclusive_ranges_without_full_clean(self):
        vlangroup = VLANGroup.objects.create(
            name='VLAN Group Inclusive Save',
            slug='vlan-group-inclusive-save',
            vid_ranges=[NumericRange(100, 100, bounds='[]')],
        )

        self.assertEqual(vlangroup.vid_ranges, [NumericRange(100, 101, bounds='[)')])
        self.assertEqual(vlangroup.total_vlan_ids, 1)

    def test_total_vlan_ids_with_update_fields(self):
        vlangroup = VLANGroup.objects.create(
            name='VLAN Group Update Fields',
            slug='vlan-group-update-fields',
            vid_ranges=[NumericRange(100, 200, bounds='[)')],
        )

        vlangroup.vid_ranges = [NumericRange(100, 100, bounds='[]')]
        vlangroup.save(update_fields=['vid_ranges'])
        vlangroup.refresh_from_db()

        self.assertEqual(vlangroup.vid_ranges, [NumericRange(100, 101, bounds='[)')])
        self.assertEqual(vlangroup.total_vlan_ids, 1)

    def test_annotate_utilization_with_zero_total_vlan_ids(self):
        vlangroup = VLANGroup.objects.create(
            name='VLAN Group Zero Total',
            slug='vlan-group-zero-total',
            vid_ranges=[NumericRange(100, 101)],
        )
        VLANGroup.objects.filter(pk=vlangroup.pk).update(total_vlan_ids=0)

        vlangroup = VLANGroup.objects.annotate_utilization().get(pk=vlangroup.pk)
        self.assertIsNone(vlangroup.utilization)


class VLANTestCase(TestCase):

    @classmethod
    def setUpTestData(cls):
        VLAN.objects.bulk_create((
            VLAN(name='VLAN 1', vid=1, qinq_role=VLANQinQRoleChoices.ROLE_SERVICE),
        ))

    def test_qinq_role(self):
        svlan = VLAN.objects.filter(qinq_role=VLANQinQRoleChoices.ROLE_SERVICE).first()

        vlan = VLAN(
            name='VLAN X',
            vid=999,
            qinq_role=VLANQinQRoleChoices.ROLE_SERVICE,
            qinq_svlan=svlan
        )
        with self.assertRaises(ValidationError):
            vlan.full_clean()

    def test_vlan_group_site_validation(self):
        sitegroup = SiteGroup.objects.create(
            name='Site Group 1',
            slug='site-group-1',
        )
        sites = Site.objects.bulk_create((
            Site(
                name='Site 1',
                slug='site-1',
            ),
            Site(
                name='Site 2',
                slug='site-2',
            ),
        ))
        sitegroup.sites.add(sites[0])
        vlangroups = VLANGroup.objects.bulk_create((
            VLANGroup(
                name='VLAN Group 1',
                slug='vlan-group-1',
                scope=sitegroup,
                scope_type=ContentType.objects.get_for_model(SiteGroup),
            ),
            VLANGroup(
                name='VLAN Group 2',
                slug='vlan-group-2',
                scope=sites[0],
                scope_type=ContentType.objects.get_for_model(Site),
            ),
            VLANGroup(
                name='VLAN Group 2',
                slug='vlan-group-2',
                scope=sites[1],
                scope_type=ContentType.objects.get_for_model(Site),
            ),
        ))
        vlan = VLAN(
            name='VLAN 1',
            vid=1,
            group=vlangroups[0],
            site=sites[0],
        )

        # VLAN Group 1 and 2 should be valid
        vlan.full_clean()
        vlan.group = vlangroups[1]
        vlan.full_clean()
        vlan.group = vlangroups[2]
        with self.assertRaises(ValidationError):
            vlan.full_clean()

    def test_vlan_group_vid_validation_with_null_vid(self):
        """A missing VID on a grouped VLAN raises a ValidationError, not a TypeError."""
        group = VLANGroup.objects.create(name='VLAN Group 1', slug='vlan-group-1')
        vlan = VLAN(name='VLAN X', vid=None, group=group)
        with self.assertRaises(ValidationError):
            vlan.full_clean()


class PrefixGetChildIPsTestCase(TestCase):
    @classmethod
    def setUpTestData(cls):
        cls.prefix = Prefix.objects.create(prefix='10.0.0.0/24')
        IPAddress.objects.bulk_create((
            IPAddress(address='10.0.0.0/24'),    # Network address (inside containment)
            IPAddress(address='10.0.0.1/24'),
            IPAddress(address='10.0.0.255/24'),  # Broadcast address (inside containment)
            IPAddress(address='10.0.1.1/24'),    # Outside the prefix
        ))

    def test_get_child_ips_matches_net_host_contained(self):
        """get_child_ips returns the same IPs as the net_host_contained containment lookup."""
        expected = set(
            IPAddress.objects.filter(
                address__net_host_contained=str(self.prefix.prefix), vrf=None
            ).values_list('pk', flat=True)
        )
        actual = set(self.prefix.get_child_ips().values_list('pk', flat=True))
        self.assertEqual(actual, expected)
        self.assertEqual(len(actual), 3)

    def test_get_child_ips_sql_avoids_containment_recheck(self):
        """get_child_ips filters on an inet host range, not the <<= containment operator."""
        sql = str(self.prefix.get_child_ips().query)
        self.assertNotIn('<<=', sql)

    def test_get_child_ips_container_in_global_table_spans_vrfs(self):
        """A container prefix in the global table returns child IPs from any VRF."""
        vrf = VRF.objects.create(name='VRF 1')
        container = Prefix.objects.create(
            prefix='10.1.0.0/24', status=PrefixStatusChoices.STATUS_CONTAINER,
        )
        in_vrf = IPAddress.objects.create(address='10.1.0.5/24', vrf=vrf)
        in_global = IPAddress.objects.create(address='10.1.0.6/24')
        child_pks = set(container.get_child_ips().values_list('pk', flat=True))
        self.assertEqual(child_pks, {in_vrf.pk, in_global.pk})


class ServiceTemplateTestCase(TestCase):

    def test_servicetemplate_lowest_port(self):
        """
        Test lowest port setting for servicetemplate
        """
        template = ServiceTemplate(
            name='Template 1',
            protocol=ServiceProtocolChoices.PROTOCOL_TCP,
            ports=[80, 443, 22, 8080],  # small test list
        )
        template.full_clean()
        template.save()
        self.assertEqual(template._ports_lowest, 22)

    def test_servicetemplate_single_port(self):
        """
        Test with a single port
        """
        template = ServiceTemplate(
            name='Template 2',
            protocol=ServiceProtocolChoices.PROTOCOL_UDP,
            ports=[53],
        )
        template.full_clean()
        template.save()
        self.assertEqual(template._ports_lowest, 53)

    def test_servicetemplate_empty_ports(self):
        """
        Test with empty ports list
        """
        template = ServiceTemplate(
            name='Template 3',
            protocol=ServiceProtocolChoices.PROTOCOL_TCP,
            ports=[],
        )
        self.assertRaises(ValidationError, template.full_clean)


class ServiceTestCase(TestCase):

    @classmethod
    def setUpTestData(cls):
        site = Site.objects.create(
            name='Site 1',
            slug='site-1',
        )
        VirtualMachine.objects.create(
            name='virtual machine 1',
            site=site,
        )

    def test_large_service(self):
        """
        Test creation of service with large number of ports.
        Related to issue #22273
        """
        service = Service(
            name='Service 1',
            protocol=ServiceProtocolChoices.PROTOCOL_TCP,
            ports=list(range(SERVICE_PORT_MIN, SERVICE_PORT_MAX)),
            parent=VirtualMachine.objects.first(),
        )
        service.full_clean()
        # Testing .save() is the important part, to check for database problems
        service.save()
        self.assertEqual(service._ports_lowest, SERVICE_PORT_MIN)
