content: import json
  import logging
  
  from django.test import tag
  from django.urls import reverse
  from netaddr import IPNetwork
  from rest_framework import status
  
  from dcim.models import Device, DeviceRole, DeviceType, Interface, Manufacturer, Site
  from ipam.choices import *
  from ipam.models import *
  from tenancy.models import Tenant
  from utilities.data import string_to_ranges
  from utilities.testing import APITestCase, APIViewTestCases, create_test_device, disable_logging
  
  
  class AppTest(APITestCase):
  
      def test_root(self):
  
          url = reverse('ipam-api:api-root')
          response = self.client.get('{}?format=api'.format(url), **self.header)
  
          self.assertEqual(response.status_code, 200)
  
  
  class ASNRangeTest(APIViewTestCases.APIViewTestCase):
      model = ASNRange
      brief_fields = ['description', 'display', 'id', 'name', 'url']
      bulk_update_data = {
          'description': 'New description',
      }
  
      @classmethod
      def setUpTestData(cls):
          rirs = (
              RIR(name='RIR 1', slug='rir-1', is_private=True),
              RIR(name='RIR 2', slug='rir-2', is_private=True),
          )
          RIR.objects.bulk_create(rirs)
  
          tenants = (
              Tenant(name='Tenant 1', slug='tenant-1'),
              Tenant(name='Tenant 2', slug='tenant-2'),
          )
          Tenant.objects.bulk_create(tenants)
  
          asn_ranges = (
              ASNRange(name='ASN Range 1', slug='asn-range-1', rir=rirs[0], tenant=tenants[0], start=100, end=199),
              ASNRange(name='ASN Range 2', slug='asn-range-2', rir=rirs[0], tenant=tenants[0], start=200, end=299),
              ASNRange(name='ASN Range 3', slug='asn-range-3', rir=rirs[0], tenant=tenants[0], start=300, end=399),
          )
          ASNRange.objects.bulk_create(asn_ranges)
  
          cls.create_data = [
              {
                  'name': 'ASN Range 4',
                  'slug': 'asn-range-4',
                  'rir': rirs[1].pk,
                  'start': 400,
                  'end': 499,
                  'tenant': tenants[1].pk,
              },
              {
                  'name': 'ASN Range 5',
                  'slug': 'asn-range-5',
                  'rir': rirs[1].pk,
                  'start': 500,
                  'end': 599,
                  'tenant': tenants[1].pk,
              },
              {
                  'name': 'ASN Range 6',
                  'slug': 'asn-range-6',
                  'rir': rirs[1].pk,
                  'start': 600,
                  'end': 699,
                  'tenant': tenants[1].pk,
              },
          ]
  
      def test_list_available_asns(self):
          """
          Test retrieval of all available ASNs within a parent range.
          """
          rir = RIR.objects.first()
          asnrange = ASNRange.objects.create(name='Range 1', slug='range-1', rir=rir, start=101, end=110)
          url = reverse('ipam-api:asnrange-available-asns', kwargs={'pk': asnrange.pk})
          self.add_permissions('ipam.view_asnrange', 'ipam.view_asn')
  
          response = self.client.get(url, **self.header)
          self.assertHttpStatus(response, status.HTTP_200_OK)
          self.assertEqual(len(response.data), 10)
  
      def test_create_single_available_asn(self):
          """
          Test creation of the first available ASN within a range.
          """
          rir = RIR.objects.first()
          asnrange = ASNRange.objects.create(name='Range 1', slug='range-1', rir=rir, start=101, end=110)
          url = reverse('ipam-api:asnrange-available-asns', kwargs={'pk': asnrange.pk})
          self.add_permissions('ipam.view_asnrange', 'ipam.add_asn')
  
          data = {
              'description': 'New ASN'
          }
          response = self.client.post(url, data, format='json', **self.header)
          self.assertHttpStatus(response, status.HTTP_201_CREATED)
          self.assertEqual(response.data['rir']['id'], asnrange.rir.pk)
          self.assertEqual(response.data['description'], data['description'])
  
      def test_create_multiple_available_asns(self):
          """
          Test the creation of several available ASNs within a parent range.
          """
          rir = RIR.objects.first()
          asnrange = ASNRange.objects.create(name='Range 1', slug='range-1', rir=rir, start=101, end=110)
          url = reverse('ipam-api:asnrange-available-asns', kwargs={'pk': asnrange.pk})
          self.add_permissions('ipam.view_asnrange', 'ipam.add_asn')
  
          # Try to create eleven ASNs (only ten are available)
          data = [
              {'description': f'New ASN {i}'}
              for i in range(1, 12)
          ]
          assert len(data) == 11
          response = self.client.post(url, data, format='json', **self.header)
          self.assertHttpStatus(response, status.HTTP_409_CONFLICT)
          self.assertIn('detail', response.data)
  
          # Create all ten available ASNs in a single request
          data.pop()
          assert len(data) == 10
          response = self.client.post(url, data, format='json', **self.header)
          self.assertHttpStatus(response, status.HTTP_201_CREATED)
          self.assertEqual(len(response.data), 10)
  
  
  class ASNTest(APIViewTestCases.APIViewTestCase):
      model = ASN
      brief_fields = ['asn', 'description', 'display', 'id', 'url']
      bulk_update_data = {
          'description': 'New description',
      }
  
      @classmethod
      def setUpTestData(cls):
          rirs = (
              RIR(name='RIR 1', slug='rir-1', is_private=True),
              RIR(name='RIR 2', slug='rir-2', is_private=True),
          )
          RIR.objects.bulk_create(rirs)
  
          sites = (
              Site(name='Site 1', slug='site-1'),
              Site(name='Site 2', slug='site-2')
          )
          Site.objects.bulk_create(sites)
  
          tenants = (
              Tenant(name='Tenant 1', slug='tenant-1'),
              Tenant(name='Tenant 2', slug='tenant-2'),
          )
          Tenant.objects.bulk_create(tenants)
  
          asns = (
              ASN(asn=65000, rir=rirs[0], tenant=tenants[0]),
              ASN(asn=65001, rir=rirs[0], tenant=tenants[1]),
              ASN(asn=4200000000, rir=rirs[1], tenant=tenants[0]),
              ASN(asn=4200000001, rir=rirs[1], tenant=tenants[1]),
          )
          ASN.objects.bulk_create(asns)
  
          asns[0].sites.set([sites[0]])
          asns[1].sites.set([sites[1]])
          asns[2].sites.set([sites[0]])
          asns[3].sites.set([sites[1]])
  
          cls.create_data = [
              {
                  'asn': 64512,
                  'rir': rirs[0].pk,
              },
              {
                  'asn': 65002,
                  'rir': rirs[0].pk,
              },
              {
                  'asn': 4200000002,
                  'rir': rirs[1].pk,
              },
          ]
  
  
  class VRFTest(APIViewTestCases.APIViewTestCase):
      model = VRF
      brief_fields = ['description', 'display', 'id', 'name', 'prefix_count', 'rd', 'url']
      create_data = [
          {
              'name': 'VRF 4',
              'rd': '65000:4',
          },
          {
              'name': 'VRF 5',
              'rd': '65000:5',
          },
          {
              'name': 'VRF 6',
              'rd': '65000:6',
          },
      ]
      bulk_update_data = {
          'description': 'New description',
      }
  
      @classmethod
      def setUpTestData(cls):
  
          vrfs = (
              VRF(name='VRF 1', rd='65000:1'),
              VRF(name='VRF 2', rd='65000:2'),
              VRF(name='VRF 3'),  # No RD
          )
          VRF.objects.bulk_create(vrfs)
  
  
  class RouteTargetTest(APIViewTestCases.APIViewTestCase):
      model = RouteTarget
      brief_fields = ['description', 'display', 'id', 'name', 'url']
      create_data = [
          {
              'name': '65000:1004',
          },
          {
              'name': '65000:1005',
          },
          {
              'name': '65000:1006',
          },
      ]
      bulk_update_data = {
          'description': 'New description',
      }
  
      @classmethod
      def setUpTestData(cls):
  
          route_targets = (
              RouteTarget(name='65000:1001'),
              RouteTarget(name='65000:1002'),
              RouteTarget(name='65000:1003'),
          )
          RouteTarget.objects.bulk_create(route_targets)
  
  
  class RIRTest(APIViewTestCases.APIViewTestCase):
      model = RIR
      brief_fields = ['aggregate_count', 'description', 'display', 'id', 'name', 'slug', 'url']
      create_data = [
          {
              'name': 'RIR 4',
              'slug': 'rir-4',
          },
          {
              'name': 'RIR 5',
              'slug': 'rir-5',
          },
          {
              'name': 'RIR 6',
              'slug': 'rir-6',
          },
      ]
      bulk_update_data = {
          'description': 'New description',
      }
  
      @classmethod
      def setUpTestData(cls):
  
          rirs = (
              RIR(name='RIR 1', slug='rir-1'),
              RIR(name='RIR 2', slug='rir-2'),
              RIR(name='RIR 3', slug='rir-3'),
          )
          RIR.objects.bulk_create(rirs)
  
  
  class AggregateTest(APIViewTestCases.APIViewTestCase):
      model = Aggregate
      brief_fields = ['description', 'display', 'family', 'id', 'prefix', 'url']
      bulk_update_data = {
          'description': 'New description',
      }
  
      @classmethod
      def setUpTestData(cls):
  
          rirs = (
              RIR(name='RIR 1', slug='rir-1'),
              RIR(name='RIR 2', slug='rir-2'),
          )
          RIR.objects.bulk_create(rirs)
  
          aggregates = (
              Aggregate(prefix=IPNetwork('10.0.0.0/8'), rir=rirs[0]),
              Aggregate(prefix=IPNetwork('172.16.0.0/12'), rir=rirs[0]),
              Aggregate(prefix=IPNetwork('192.168.0.0/16'), rir=rirs[0]),
          )
          Aggregate.objects.bulk_create(aggregates)
  
          cls.create_data = [
              {
                  'prefix': '100.0.0.0/8',
                  'rir': rirs[1].pk,
              },
              {
                  'prefix': '101.0.0.0/8',
                  'rir': rirs[1].pk,
              },
              {
                  'prefix': '102.0.0.0/8',
                  'rir': rirs[1].pk,
              },
          ]
  
      @tag('regression')
      def test_graphql_aggregate_prefix_exact(self):
          """
          Test case to verify aggregate prefix equality via field lookup in GraphQL API.
          """
  
          self.add_permissions('ipam.view_aggregate', 'ipam.view_rir')
  
          rir = RIR.objects.create(name='RFC6598', slug='rfc6598', is_private=True)
          aggregate1 = Aggregate.objects.create(prefix='100.64.0.0/10', rir=rir)
          Aggregate.objects.create(prefix='203.0.113.0/24', rir=rir)
  
          url = reverse('graphql')
          query = """{
              aggregate_list(filters: { prefix: { exact: "100.64.0.0/10" } }) { prefix }
          }"""
          response = self.client.post(url, data={'query': query}, format='json', **self.header)
          self.assertHttpStatus(response, status.HTTP_200_OK)
          data = response.json()
          self.assertNotIn('errors', data)
  
          prefixes = {row['prefix'] for row in data['data']['aggregate_list']}
          self.assertIn(str(aggregate1.prefix), prefixes)
  
      @tag('regression')
      def test_graphql_aggregate_contains_skips_invalid(self):
          """
          Test the GraphQL API Aggregate `contains` filter skips invalid input.
          """
  
          self.add_permissions('ipam.view_aggregate', 'ipam.view_rir')
  
          rir = RIR.objects.create(name='RIR 3', slug='rir-3', is_private=False)
          aggregate1 = Aggregate.objects.create(prefix='100.64.0.0/10', rir=rir)
          Aggregate.objects.create(prefix='203.0.113.0/24', rir=rir)
  
          url = reverse('graphql')
          query = """{
              aggregate_list(filters: { contains: ["100.64.16.0/24", "not-a-cidr", ""] }) { prefix }
          }"""
          response = self.client.post(url, data={'query': query}, format='json', **self.header)
          self.assertHttpStatus(response, status.HTTP_200_OK)
          data = response.json()
          self.assertNotIn('errors', data)
  
          prefixes = {row['prefix'] for row in data['data']['aggregate_list']}
          self.assertIn(str(aggregate1.prefix), prefixes)
          # No exception occurred; invalid entries were ignored
  
  
  class RoleTest(APIViewTestCases.APIViewTestCase):
      model = Role
      brief_fields = ['description', 'display', 'id', 'name', 'prefix_count', 'slug', 'url', 'vlan_count']
      create_data = [
          {
              'name': 'Role 4',
              'slug': 'role-4',
          },
          {
              'name': 'Role 5',
              'slug': 'role-5',
          },
          {
              'name': 'Role 6',
              'slug': 'role-6',
          },
      ]
      bulk_update_data = {
          'description': 'New description',
      }
  
      @classmethod
      def setUpTestData(cls):
  
          roles = (
              Role(name='Role 1', slug='role-1'),
              Role(name='Role 2', slug='role-2'),
              Role(name='Role 3', slug='role-3'),
          )
          Role.objects.bulk_create(roles)
  
  
  class PrefixTest(APIViewTestCases.APIViewTestCase):
      model = Prefix
      brief_fields = ['_depth', 'description', 'display', 'family', 'id', 'prefix', 'url']
      create_data = [
          {
              'prefix': '192.168.4.0/24',
          },
          {
              'prefix': '192.168.5.0/24',
          },
          {
              'prefix': '192.168.6.0/24',
          },
      ]
      bulk_update_data = {
          'description': 'New description',
      }
  
      @classmethod
      def setUpTestData(cls):
  
          prefixes = (
              Prefix(prefix=IPNetwork('192.168.1.0/24')),
              Prefix(prefix=IPNetwork('192.168.2.0/24')),
              Prefix(prefix=IPNetwork('192.168.3.0/24')),
          )
          Prefix.objects.bulk_create(prefixes)
  
      @tag('regression')
      def test_clean_validates_scope(self):
          prefix = Prefix.objects.first()
          site = Site.objects.create(name='Test Site', slug='test-site')
  
          data = {'scope_type': 'dcim.site', 'scope_id': site.id}
          url = reverse('ipam-api:prefix-detail', kwargs={'pk': prefix.pk})
          self.add_permissions('ipam.change_prefix')
  
          response = self.client.patch(url, data, format='json', **self.header)
          self.assertHttpStatus(response, status.HTTP_200_OK)
  
      def test_list_available_prefixes(self):
          """
          Test retrieval of all available prefixes within a parent prefix.
          """
          vrf = VRF.objects.create(name='VRF 1')
          prefix = Prefix.objects.create(prefix=IPNetwork('192.0.2.0/24'), vrf=vrf)
          Prefix.objects.create(prefix=IPNetwork('192.0.2.64/26'), vrf=vrf)
          Prefix.objects.create(prefix=IPNetwork('192.0.2.192/27'), vrf=vrf)
          url = reverse('ipam-api:prefix-available-prefixes', kwargs={'pk': prefix.pk})
          self.add_permissions('ipam.view_prefix')
  
          # Retrieve all available IPs
          response = self.client.get(url, **self.header)
          available_prefixes = ['192.0.2.0/26', '192.0.2.128/26', '192.0.2.224/27']
          for i, p in enumerate(response.data):
              self.assertEqual(p['prefix'], available_prefixes[i])
  
      def test_create_single_available_prefix(self):
          """
          Test retrieval of the first available prefix within a parent prefix.
          """
          vrf = VRF.objects.create(name='VRF 1')
          prefix = Prefix.objects.create(prefix=IPNetwork('192.0.2.0/28'), vrf=vrf, is_pool=True)
          url = reverse('ipam-api:prefix-available-prefixes', kwargs={'pk': prefix.pk})
          self.add_permissions('ipam.view_prefix', 'ipam.add_prefix')
  
          # Create four available prefixes with individual requests
          prefixes_to_be_created = [
              '192.0.2.0/30',
              '192.0.2.4/30',
              '192.0.2.8/30',
              '192.0.2.12/30',
          ]
          for i in range(4):
              data = {
                  'prefix_length': 30,
                  'description': 'Test Prefix {}'.format(i + 1)
              }
              response = self.client.post(url, data, format='json', **self.header)
              self.assertHttpStatus(response, status.HTTP_201_CREATED)
              self.assertEqual(response.data['prefix'], prefixes_to_be_created[i])
              self.assertEqual(response.data['vrf']['id'], vrf.pk)
              self.assertEqual(response.data['description'], data['description'])
  
          # Try to create one more prefix
          response = self.client.post(url, {'prefix_length': 30}, format='json', **self.header)
          self.assertHttpStatus(response, status.HTTP_409_CONFLICT)
          self.assertIn('detail', response.data)
  
          # Try to create invalid prefix type
          response = self.client.post(url, {'prefix_length': '30'}, format='json', **self.header)
          self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
          self.assertIn('prefix_length', response.data[0])
  
      def test_create_multiple_available_prefixes(self):
          """
          Test the creation of available prefixes within a parent prefix.
          """
          vrf = VRF.objects.create(name='VRF 1')
          prefix = Prefix.objects.create(prefix=IPNetwork('192.0.2.0/28'), vrf=vrf, is_pool=True)
          url = reverse('ipam-api:prefix-available-prefixes', kwargs={'pk': prefix.pk})
          self.add_permissions('ipam.view_prefix', 'ipam.add_prefix')
  
          # Try to create five /30s (only four are available)
          data = [
              {'prefix_length': 30, 'description': 'Prefix 1'},
              {'prefix_length': 30, 'description': 'Prefix 2'},
              {'prefix_length': 30, 'description': 'Prefix 3'},
              {'prefix_length': 30, 'description': 'Prefix 4'},
              {'prefix_length': 30, 'description': 'Prefix 5'},
          ]
          response = self.client.post(url, data, format='json', **self.header)
          self.assertHttpStatus(response, status.HTTP_409_CONFLICT)
          self.assertIn('detail', response.data)
  
          # Verify that no prefixes were created (the entire /28 is still available)
          response = self.client.get(url, **self.header)
          self.assertHttpStatus(response, status.HTTP_200_OK)
          self.assertEqual(response.data[0]['prefix'], '192.0.2.0/28')
  
          # Create four /30s in a single request
          response = self.client.post(url, data[:4], format='json', **self.header)
          self.assertHttpStatus(response, status.HTTP_201_CREATED)
          self.assertEqual(len(response.data), 4)
  
      def test_list_available_ips(self):
          """
          Test retrieval of all available IP addresses within a parent prefix.
          """
          vrf = VRF.objects.create(name='VRF 1')
          prefix = Prefix.objects.create(prefix=IPNetwork('192.0.2.0/29'), vrf=vrf, is_pool=True)
          url = reverse('ipam-api:prefix-available-ips', kwargs={'pk': prefix.pk})
          self.add_permissions('ipam.view_prefix', 'ipam.view_ipaddress')
  
          # Retrieve all available IPs
          response = self.client.get(url, **self.header)
          self.assertHttpStatus(response, status.HTTP_200_OK)
          self.assertEqual(len(response.data), 8)  # 8 because prefix.is_pool = True
  
          # Change the prefix to not be a pool and try again
          prefix.is_pool = False
          prefix.save()
          response = self.client.get(url, **self.header)
          self.assertEqual(len(response.data), 6)  # 8 - 2 because prefix.is_pool = False
  
      def test_create_single_available_ip(self):
          """
          Test retrieval of the first available IP address within a parent prefix.
          """
          vrf = VRF.objects.create(name='VRF 1')
          prefix = Prefix.objects.create(prefix=IPNetwork('192.0.2.0/30'), vrf=vrf, is_pool=True)
          url = reverse('ipam-api:prefix-available-ips', kwargs={'pk': prefix.pk})
          self.add_permissions('ipam.view_prefix', 'ipam.add_ipaddress')
  
          # Create all four available IPs with individual requests
          for i in range(1, 5):
              data = {
                  'description': 'Test IP {}'.format(i)
              }
              response = self.client.post(url, data, format='json', **self.header)
              self.assertHttpStatus(response, status.HTTP_201_CREATED)
              self.assertEqual(response.data['vrf']['id'], vrf.pk)
              self.assertEqual(response.data['description'], data['description'])
  
          # Try to create one more IP
          response = self.client.post(url, {}, format='json', **self.header)
          self.assertHttpStatus(response, status.HTTP_409_CONFLICT)
          self.assertIn('detail', response.data)
  
      def test_create_multiple_available_ips(self):
          """
          Test the creation of available IP addresses within a parent prefix.
          """
          vrf = VRF.objects.create(name='VRF 1')
          prefix = Prefix.objects.create(prefix=IPNetwork('192.0.2.0/29'), vrf=vrf, is_pool=True)
          url = reverse('ipam-api:prefix-available-ips', kwargs={'pk': prefix.pk})
          self.add_permissions('ipam.view_prefix', 'ipam.add_ipaddress')
  
          # Try to create nine IPs (only eight are available)
          data = [{'description': f'Test IP {i}'} for i in range(1, 10)]  # 9 IPs
          response = self.client.post(url, data, format='json', **self.header)
          self.assertHttpStatus(response, status.HTTP_409_CONFLICT)
          self.assertIn('detail', response.data)
  
          # Create all eight available IPs in a single request
          data = [{'description': 'Test IP {}'.format(i)} for i in range(1, 9)]  # 8 IPs
          response = self.client.post(url, data, format='json', **self.header)
          self.assertHttpStatus(response, status.HTTP_201_CREATED)
          self.assertEqual(len(response.data), 8)
  
      @tag('regression')
      def test_graphql_tenant_prefixes_contains_nested_skips_invalid(self):
          """
          Test the GraphQL API Tenant nested Prefix `contains` filter skips invalid input.
          """
  
          self.add_permissions('ipam.view_prefix', 'ipam.view_vrf', 'tenancy.view_tenant')
  
          tenant = Tenant.objects.create(name='Tenant 1', slug='tenant-1')
          vrf = VRF.objects.create(name='Test VRF 1', rd='64512:1')
          Prefix.objects.create(prefix='10.20.0.0/16', vrf=vrf, tenant=tenant)
          Prefix.objects.create(prefix='198.51.100.0/24', vrf=vrf)  # non-tenant
  
          url = reverse('graphql')
          query = """{
              tenant_list(filters: { prefixes: { contains: ["10.20.1.0/24", "not-a-cidr"] } }) { id }
          }"""
          response = self.client.post(url, data={'query': query}, format='json', **self.header)
          self.assertHttpStatus(response, status.HTTP_200_OK)
          data = response.json()
          self.assertNotIn('errors', data)
  
          self.assertTrue(data['data']['tenant_list'])  # tenant returned
  
  
  class IPRangeTest(APIViewTestCases.APIViewTestCase):
      model = IPRange
      brief_fields = ['description', 'display', 'end_address', 'family', 'id', 'start_address', 'url']
      create_data = [
          {
              'start_address': '192.168.4.10/24',
              'end_address': '192.168.4.50/24',
          },
          {
              'start_address': '192.168.5.10/24',
              'end_address': '192.168.5.50/24',
          },
          {
              'start_address': '192.168.6.10/24',
              'end_address': '192.168.6.50/24',
          },
      ]
      bulk_update_data = {
          'description': 'New description',
      }
  
      @classmethod
      def setUpTestData(cls):
  
          ip_ranges = (
              IPRange(start_address=IPNetwork('192.168.1.10/24'), end_address=IPNetwork('192.168.1.50/24'), size=51),
              IPRange(start_address=IPNetwork('192.168.2.10/24'), end_address=IPNetwork('192.168.2.50/24'), size=51),
              IPRange(start_address=IPNetwork('192.168.3.10/24'), end_address=IPNetwork('192.168.3.50/24'), size=51),
          )
          IPRange.objects.bulk_create(ip_ranges)
  
      def test_list_available_ips(self):
          """
          Test retrieval of all available IP addresses within a parent IP range.
          """
          iprange = IPRange.objects.create(
              start_address=IPNetwork('192.0.2.10/24'),
              end_address=IPNetwork('192.0.2.19/24')
          )
          url = reverse('ipam-api:iprange-available-ips', kwargs={'pk': iprange.pk})
          self.add_permissions('ipam.view_iprange', 'ipam.view_ipaddress')
  
          # Retrieve all available IPs
          response = self.client.get(url, **self.header)
          self.assertHttpStatus(response, status.HTTP_200_OK)
          self.assertEqual(len(response.data), 10)
  
      def test_create_single_available_ip(self):
          """
          Test retrieval of the first available IP address within a parent IP range.
          """
          vrf = VRF.objects.create(name='Test VRF 1', rd='1234')
          iprange = IPRange.objects.create(
              start_address=IPNetwork('192.0.2.1/24'),
              end_address=IPNetwork('192.0.2.3/24'),
              vrf=vrf
          )
          url = reverse('ipam-api:iprange-available-ips', kwargs={'pk': iprange.pk})
          self.add_permissions('ipam.view_iprange', 'ipam.add_ipaddress')
  
          # Create all three available IPs with individual requests
          for i in range(1, 4):
              data = {
                  'description': f'Test IP #{i}'
              }
              response = self.client.post(url, data, format='json', **self.header)
              self.assertHttpStatus(response, status.HTTP_201_CREATED)
              self.assertEqual(response.data['vrf']['id'], vrf.pk)
              self.assertEqual(response.data['description'], data['description'])
  
          # Try to create one more IP
          response = self.client.post(url, {}, format='json', **self.header)
          self.assertHttpStatus(response, status.HTTP_409_CONFLICT)
          self.assertIn('detail', response.data)
  
      def test_create_multiple_available_ips(self):
          """
          Test the creation of available IP addresses within a parent IP range.
          """
          iprange = IPRange.objects.create(
              start_address=IPNetwork('192.0.2.1/24'),
              end_address=IPNetwork('192.0.2.8/24')
          )
          url = reverse('ipam-api:iprange-available-ips', kwargs={'pk': iprange.pk})
          self.add_permissions('ipam.view_iprange', 'ipam.add_ipaddress')
  
          # Try to create nine IPs (only eight are available)
          data = [{'description': f'Test IP #{i}'} for i in range(1, 10)]  # 9 IPs
          response = self.client.post(url, data, format='json', **self.header)
          self.assertHttpStatus(response, status.HTTP_409_CONFLICT)
          self.assertIn('detail', response.data)
  
          # Create all eight available IPs in a single request
          data = [{'description': f'Test IP #{i}'} for i in range(1, 9)]  # 8 IPs
          response = self.client.post(url, data, format='json', **self.header)
          self.assertHttpStatus(response, status.HTTP_201_CREATED)
          self.assertEqual(len(response.data), 8)
  
      @tag('regression')
      def test_graphql_tenant_ip_ranges_parent_nested_skips_invalid(self):
          """
          Test the GraphQL API Tenant nested IP Range `parent` filter skips invalid input.
          """
  
          self.add_permissions('tenancy.view_tenant', 'ipam.view_iprange', 'ipam.view_vrf')
  
          tenant = Tenant.objects.create(name='Tenant 1', slug='tenant-1')
          vrf = VRF.objects.create(name='Test VRF 1', rd='64512:1')
          IPRange.objects.create(
              start_address=IPNetwork('10.30.0.1/24'), end_address=IPNetwork('10.30.0.255/24'), vrf=vrf, tenant=tenant
          )
          IPRange.objects.create(
              start_address=IPNetwork('10.31.0.1/24'), end_address=IPNetwork('10.31.0.255/24'), vrf=vrf, tenant=tenant
          )
  
          url = reverse('graphql')
          query = """{
              tenant_list(filters: {
                  name: { exact: "Tenant 1" }
                  ip_ranges: { parent: ["10.30.0.0/24", "bogus"] }
              }) { id }
          }"""
          response = self.client.post(url, data={'query': query}, format='json', **self.header)
          self.assertHttpStatus(response, status.HTTP_200_OK)
          data = response.json()
          self.assertNotIn('errors', data)
          self.assertTrue(data['data']['tenant_list'])  # tenant returned
          # No exception occurred; invalid entries were ignored
  
      @tag('regression')
      def test_graphql_tenant_ip_ranges_contains_nested_skips_invalid(self):
          """
          Test the GraphQL API Tenant nested IP Range `contains` filter skips invalid input.
          """
  
          self.add_permissions('tenancy.view_tenant', 'ipam.view_iprange', 'ipam.view_vrf')
  
          tenant = Tenant.objects.create(name='Tenant 2', slug='tenant-2')
          vrf = VRF.objects.create(name='Test VRF 1', rd='64512:2')
          IPRange.objects.create(
              start_address=IPNetwork('10.40.0.1/24'), end_address=IPNetwork('10.40.0.255/24'), vrf=vrf, tenant=tenant
          )
  
          url = reverse('graphql')
          query = """{
              tenant_list(filters: {
                  name: { exact: "Tenant 2" }
                  ip_ranges: { contains: ["10.40.0.128/25", "###"] }
              }) { id }
          }"""
          response = self.client.post(url, data={'query': query}, format='json', **self.header)
          self.assertHttpStatus(response, status.HTTP_200_OK)
          data = response.json()
          self.assertNotIn('errors', data)
          self.assertTrue(data['data']['tenant_list'])  # tenant returned
          # No exception occurred; invalid entries were ignored
  
  
  class IPAddressTest(APIViewTestCases.APIViewTestCase):
      model = IPAddress
      brief_fields = ['address', 'description', 'display', 'family', 'id', 'url']
      create_data = [
          {
              'address': '192.168.0.4/24',
          },
          {
              'address': '192.168.0.5/24',
          },
          {
              'address': '192.168.0.6/24',
          },
      ]
      bulk_update_data = {
          'description': 'New description',
      }
      graphql_filter = {
          'address': {'lookup': 'i_exact', 'value': '192.168.0.1/24'},
      }
  
      @classmethod
      def setUpTestData(cls):
  
          ip_addresses = (
              IPAddress(address=IPNetwork('192.168.0.1/24')),
              IPAddress(address=IPNetwork('192.168.0.2/24')),
              IPAddress(address=IPNetwork('192.168.0.3/24')),
          )
          IPAddress.objects.bulk_create(ip_addresses)
  
      def test_assign_object(self):
          """
          Test the creation of available IP addresses within a parent IP range.
          """
          site = Site.objects.create(name='Site 1')
          manufacturer = Manufacturer.objects.create(name='Manufacturer 1')
          device_type = DeviceType.objects.create(model='Device Type 1', manufacturer=manufacturer)
          role = DeviceRole.objects.create(name='Switch')
          device1 = Device.objects.create(
              name='Device 1',
              site=site,
              device_type=device_type,
              role=role,
              status='active'
          )
          interface1 = Interface.objects.create(name='Interface 1', device=device1, type='1000baset')
          interface2 = Interface.objects.create(name='Interface 2', device=device1, type='1000baset')
          device2 = Device.objects.create(
              name='Device 2',
              site=site,
              device_type=device_type,
              role=role,
              status='active'
          )
          interface3 = Interface.objects.create(name='Interface 3', device=device2, type='1000baset')
  
          ip_addresses = (
              IPAddress(address=IPNetwork('192.168.0.4/24'), assigned_object=interface1),
              IPAddress(address=IPNetwork('192.168.1.4/24')),
          )
          IPAddress.objects.bulk_create(ip_addresses)
  
          ip1 = ip_addresses[0]
          ip1.assigned_object = interface1
          device1.primary_ip4 = ip_addresses[0]
          device1.save()
  
          url = reverse('ipam-api:ipaddress-detail', kwargs={'pk': ip1.pk})
          self.add_permissions('ipam.change_ipaddress')
  
          # assign to same parent
          data = {
              'assigned_object_id': interface2.pk
          }
          response = self.client.patch(url, data, format='json', **self.header)
          self.assertHttpStatus(response, status.HTTP_200_OK)
  
          # assign to same different parent - should error
          data = {
              'assigned_object_id': interface3.pk
          }
          response = self.client.patch(url, data, format='json', **self.header)
          self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
  
      @tag('regression')
      def test_graphql_device_primary_ip4_assigned_nested(self):
          """
          Test the GraphQL API Device nested IP Address `primary_ip4` filter.
          """
  
          self.add_permissions('dcim.view_device', 'dcim.view_interface', 'ipam.view_ipaddress')
  
          site = Site.objects.create(name='Site 1')
          manufacturer = Manufacturer.objects.create(name='Manufacturer 1')
          device_type = DeviceType.objects.create(model='Device Type 1', manufacturer=manufacturer)
          role = DeviceRole.objects.create(name='Switch')
  
          device1 = Device.objects.create(name='Device 1', site=site, device_type=device_type, role=role, status='active')
          interface1 = Interface.objects.create(name='Interface 1', device=device1, type='1000baset')
  