2
0

test_models.py 21 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514
  1. from netaddr import IPNetwork, IPSet
  2. from django.core.exceptions import ValidationError
  3. from django.test import TestCase, override_settings
  4. from ipam.choices import IPAddressRoleChoices, PrefixStatusChoices
  5. from ipam.models import Aggregate, IPAddress, IPRange, Prefix, RIR, VLAN, VLANGroup, VRF
  6. class TestAggregate(TestCase):
  7. def test_get_utilization(self):
  8. rir = RIR.objects.create(name='RIR 1', slug='rir-1')
  9. aggregate = Aggregate(prefix=IPNetwork('10.0.0.0/8'), rir=rir)
  10. aggregate.save()
  11. # 25% utilization
  12. Prefix.objects.bulk_create((
  13. Prefix(prefix=IPNetwork('10.0.0.0/12')),
  14. Prefix(prefix=IPNetwork('10.16.0.0/12')),
  15. Prefix(prefix=IPNetwork('10.32.0.0/12')),
  16. Prefix(prefix=IPNetwork('10.48.0.0/12')),
  17. ))
  18. self.assertEqual(aggregate.get_utilization(), 25)
  19. # 50% utilization
  20. Prefix.objects.bulk_create((
  21. Prefix(prefix=IPNetwork('10.64.0.0/10')),
  22. ))
  23. self.assertEqual(aggregate.get_utilization(), 50)
  24. # 100% utilization
  25. Prefix.objects.bulk_create((
  26. Prefix(prefix=IPNetwork('10.128.0.0/9')),
  27. ))
  28. self.assertEqual(aggregate.get_utilization(), 100)
  29. class TestPrefix(TestCase):
  30. def test_get_duplicates(self):
  31. prefixes = Prefix.objects.bulk_create((
  32. Prefix(prefix=IPNetwork('192.0.2.0/24')),
  33. Prefix(prefix=IPNetwork('192.0.2.0/24')),
  34. Prefix(prefix=IPNetwork('192.0.2.0/24')),
  35. ))
  36. duplicate_prefix_pks = [p.pk for p in prefixes[0].get_duplicates()]
  37. self.assertSetEqual(set(duplicate_prefix_pks), {prefixes[1].pk, prefixes[2].pk})
  38. def test_get_child_prefixes(self):
  39. vrfs = VRF.objects.bulk_create((
  40. VRF(name='VRF 1'),
  41. VRF(name='VRF 2'),
  42. VRF(name='VRF 3'),
  43. ))
  44. prefixes = Prefix.objects.bulk_create((
  45. Prefix(prefix=IPNetwork('10.0.0.0/16'), status=PrefixStatusChoices.STATUS_CONTAINER),
  46. Prefix(prefix=IPNetwork('10.0.0.0/24'), vrf=None),
  47. Prefix(prefix=IPNetwork('10.0.1.0/24'), vrf=vrfs[0]),
  48. Prefix(prefix=IPNetwork('10.0.2.0/24'), vrf=vrfs[1]),
  49. Prefix(prefix=IPNetwork('10.0.3.0/24'), vrf=vrfs[2]),
  50. ))
  51. child_prefix_pks = {p.pk for p in prefixes[0].get_child_prefixes()}
  52. # Global container should return all children
  53. self.assertSetEqual(child_prefix_pks, {prefixes[1].pk, prefixes[2].pk, prefixes[3].pk, prefixes[4].pk})
  54. prefixes[0].vrf = vrfs[0]
  55. prefixes[0].save()
  56. child_prefix_pks = {p.pk for p in prefixes[0].get_child_prefixes()}
  57. # VRF container is limited to its own VRF
  58. self.assertSetEqual(child_prefix_pks, {prefixes[2].pk})
  59. def test_get_child_ranges(self):
  60. prefix = Prefix(prefix='192.168.0.16/28')
  61. prefix.save()
  62. ranges = IPRange.objects.bulk_create((
  63. IPRange(start_address=IPNetwork('192.168.0.1/24'), end_address=IPNetwork('192.168.0.10/24'), size=10), # No overlap
  64. IPRange(start_address=IPNetwork('192.168.0.11/24'), end_address=IPNetwork('192.168.0.17/24'), size=7), # Partial overlap
  65. IPRange(start_address=IPNetwork('192.168.0.18/24'), end_address=IPNetwork('192.168.0.23/24'), size=6), # Full overlap
  66. IPRange(start_address=IPNetwork('192.168.0.24/24'), end_address=IPNetwork('192.168.0.30/24'), size=7), # Full overlap
  67. IPRange(start_address=IPNetwork('192.168.0.31/24'), end_address=IPNetwork('192.168.0.40/24'), size=10), # Partial overlap
  68. ))
  69. child_ranges = prefix.get_child_ranges()
  70. self.assertEqual(len(child_ranges), 2)
  71. self.assertEqual(child_ranges[0], ranges[2])
  72. self.assertEqual(child_ranges[1], ranges[3])
  73. def test_get_child_ips(self):
  74. vrfs = VRF.objects.bulk_create((
  75. VRF(name='VRF 1'),
  76. VRF(name='VRF 2'),
  77. VRF(name='VRF 3'),
  78. ))
  79. parent_prefix = Prefix.objects.create(
  80. prefix=IPNetwork('10.0.0.0/16'), status=PrefixStatusChoices.STATUS_CONTAINER
  81. )
  82. ips = IPAddress.objects.bulk_create((
  83. IPAddress(address=IPNetwork('10.0.0.1/24'), vrf=None),
  84. IPAddress(address=IPNetwork('10.0.1.1/24'), vrf=vrfs[0]),
  85. IPAddress(address=IPNetwork('10.0.2.1/24'), vrf=vrfs[1]),
  86. IPAddress(address=IPNetwork('10.0.3.1/24'), vrf=vrfs[2]),
  87. ))
  88. child_ip_pks = {p.pk for p in parent_prefix.get_child_ips()}
  89. # Global container should return all children
  90. self.assertSetEqual(child_ip_pks, {ips[0].pk, ips[1].pk, ips[2].pk, ips[3].pk})
  91. parent_prefix.vrf = vrfs[0]
  92. parent_prefix.save()
  93. child_ip_pks = {p.pk for p in parent_prefix.get_child_ips()}
  94. # VRF container is limited to its own VRF
  95. self.assertSetEqual(child_ip_pks, {ips[1].pk})
  96. def test_get_available_prefixes(self):
  97. prefixes = Prefix.objects.bulk_create((
  98. Prefix(prefix=IPNetwork('10.0.0.0/16')), # Parent prefix
  99. Prefix(prefix=IPNetwork('10.0.0.0/20')),
  100. Prefix(prefix=IPNetwork('10.0.32.0/20')),
  101. Prefix(prefix=IPNetwork('10.0.128.0/18')),
  102. ))
  103. missing_prefixes = IPSet([
  104. IPNetwork('10.0.16.0/20'),
  105. IPNetwork('10.0.48.0/20'),
  106. IPNetwork('10.0.64.0/18'),
  107. IPNetwork('10.0.192.0/18'),
  108. ])
  109. available_prefixes = prefixes[0].get_available_prefixes()
  110. self.assertEqual(available_prefixes, missing_prefixes)
  111. def test_get_available_ips(self):
  112. parent_prefix = Prefix.objects.create(prefix=IPNetwork('10.0.0.0/28'))
  113. IPAddress.objects.bulk_create((
  114. IPAddress(address=IPNetwork('10.0.0.1/26')),
  115. IPAddress(address=IPNetwork('10.0.0.3/26')),
  116. IPAddress(address=IPNetwork('10.0.0.5/26')),
  117. IPAddress(address=IPNetwork('10.0.0.7/26')),
  118. ))
  119. IPRange.objects.create(
  120. start_address=IPNetwork('10.0.0.9/26'),
  121. end_address=IPNetwork('10.0.0.12/26')
  122. )
  123. missing_ips = IPSet([
  124. '10.0.0.2/32',
  125. '10.0.0.4/32',
  126. '10.0.0.6/32',
  127. '10.0.0.8/32',
  128. '10.0.0.13/32',
  129. '10.0.0.14/32',
  130. ])
  131. available_ips = parent_prefix.get_available_ips()
  132. self.assertEqual(available_ips, missing_ips)
  133. def test_get_first_available_prefix(self):
  134. prefixes = Prefix.objects.bulk_create((
  135. Prefix(prefix=IPNetwork('10.0.0.0/16')), # Parent prefix
  136. Prefix(prefix=IPNetwork('10.0.0.0/24')),
  137. Prefix(prefix=IPNetwork('10.0.1.0/24')),
  138. Prefix(prefix=IPNetwork('10.0.2.0/24')),
  139. ))
  140. self.assertEqual(prefixes[0].get_first_available_prefix(), IPNetwork('10.0.3.0/24'))
  141. Prefix.objects.create(prefix=IPNetwork('10.0.3.0/24'))
  142. self.assertEqual(prefixes[0].get_first_available_prefix(), IPNetwork('10.0.4.0/22'))
  143. def test_get_first_available_ip(self):
  144. parent_prefix = Prefix.objects.create(prefix=IPNetwork('10.0.0.0/24'))
  145. IPAddress.objects.bulk_create((
  146. IPAddress(address=IPNetwork('10.0.0.1/24')),
  147. IPAddress(address=IPNetwork('10.0.0.2/24')),
  148. IPAddress(address=IPNetwork('10.0.0.3/24')),
  149. ))
  150. self.assertEqual(parent_prefix.get_first_available_ip(), '10.0.0.4/24')
  151. IPAddress.objects.create(address=IPNetwork('10.0.0.4/24'))
  152. self.assertEqual(parent_prefix.get_first_available_ip(), '10.0.0.5/24')
  153. def test_get_utilization_container(self):
  154. prefixes = (
  155. Prefix(prefix=IPNetwork('10.0.0.0/24'), status=PrefixStatusChoices.STATUS_CONTAINER),
  156. Prefix(prefix=IPNetwork('10.0.0.0/26')),
  157. Prefix(prefix=IPNetwork('10.0.0.128/26')),
  158. )
  159. Prefix.objects.bulk_create(prefixes)
  160. self.assertEqual(prefixes[0].get_utilization(), 50) # 50% utilization
  161. def test_get_utilization_noncontainer(self):
  162. prefix = Prefix.objects.create(
  163. prefix=IPNetwork('10.0.0.0/24'),
  164. status=PrefixStatusChoices.STATUS_ACTIVE
  165. )
  166. # Create 32 child IPs
  167. IPAddress.objects.bulk_create([
  168. IPAddress(address=IPNetwork(f'10.0.0.{i}/24')) for i in range(1, 33)
  169. ])
  170. self.assertEqual(prefix.get_utilization(), 12) # 12.5% utilization
  171. # Create a child range with 32 additional IPs
  172. IPRange.objects.create(start_address=IPNetwork('10.0.0.33/24'), end_address=IPNetwork('10.0.0.64/24'))
  173. self.assertEqual(prefix.get_utilization(), 25) # 25% utilization
  174. #
  175. # Uniqueness enforcement tests
  176. #
  177. @override_settings(ENFORCE_GLOBAL_UNIQUE=False)
  178. def test_duplicate_global(self):
  179. Prefix.objects.create(prefix=IPNetwork('192.0.2.0/24'))
  180. duplicate_prefix = Prefix(prefix=IPNetwork('192.0.2.0/24'))
  181. self.assertIsNone(duplicate_prefix.clean())
  182. @override_settings(ENFORCE_GLOBAL_UNIQUE=True)
  183. def test_duplicate_global_unique(self):
  184. Prefix.objects.create(prefix=IPNetwork('192.0.2.0/24'))
  185. duplicate_prefix = Prefix(prefix=IPNetwork('192.0.2.0/24'))
  186. self.assertRaises(ValidationError, duplicate_prefix.clean)
  187. def test_duplicate_vrf(self):
  188. vrf = VRF.objects.create(name='Test', rd='1:1', enforce_unique=False)
  189. Prefix.objects.create(vrf=vrf, prefix=IPNetwork('192.0.2.0/24'))
  190. duplicate_prefix = Prefix(vrf=vrf, prefix=IPNetwork('192.0.2.0/24'))
  191. self.assertIsNone(duplicate_prefix.clean())
  192. def test_duplicate_vrf_unique(self):
  193. vrf = VRF.objects.create(name='Test', rd='1:1', enforce_unique=True)
  194. Prefix.objects.create(vrf=vrf, prefix=IPNetwork('192.0.2.0/24'))
  195. duplicate_prefix = Prefix(vrf=vrf, prefix=IPNetwork('192.0.2.0/24'))
  196. self.assertRaises(ValidationError, duplicate_prefix.clean)
  197. class TestPrefixHierarchy(TestCase):
  198. """
  199. Test the automatic updating of depth and child count in response to changes made within
  200. the prefix hierarchy.
  201. """
  202. @classmethod
  203. def setUpTestData(cls):
  204. prefixes = (
  205. # IPv4
  206. Prefix(prefix='10.0.0.0/8', _depth=0, _children=2),
  207. Prefix(prefix='10.0.0.0/16', _depth=1, _children=1),
  208. Prefix(prefix='10.0.0.0/24', _depth=2, _children=0),
  209. # IPv6
  210. Prefix(prefix='2001:db8::/32', _depth=0, _children=2),
  211. Prefix(prefix='2001:db8::/40', _depth=1, _children=1),
  212. Prefix(prefix='2001:db8::/48', _depth=2, _children=0),
  213. )
  214. Prefix.objects.bulk_create(prefixes)
  215. def test_create_prefix4(self):
  216. # Create 10.0.0.0/12
  217. Prefix(prefix='10.0.0.0/12').save()
  218. prefixes = Prefix.objects.filter(prefix__family=4)
  219. self.assertEqual(prefixes[0].prefix, IPNetwork('10.0.0.0/8'))
  220. self.assertEqual(prefixes[0]._depth, 0)
  221. self.assertEqual(prefixes[0]._children, 3)
  222. self.assertEqual(prefixes[1].prefix, IPNetwork('10.0.0.0/12'))
  223. self.assertEqual(prefixes[1]._depth, 1)
  224. self.assertEqual(prefixes[1]._children, 2)
  225. self.assertEqual(prefixes[2].prefix, IPNetwork('10.0.0.0/16'))
  226. self.assertEqual(prefixes[2]._depth, 2)
  227. self.assertEqual(prefixes[2]._children, 1)
  228. self.assertEqual(prefixes[3].prefix, IPNetwork('10.0.0.0/24'))
  229. self.assertEqual(prefixes[3]._depth, 3)
  230. self.assertEqual(prefixes[3]._children, 0)
  231. def test_create_prefix6(self):
  232. # Create 2001:db8::/36
  233. Prefix(prefix='2001:db8::/36').save()
  234. prefixes = Prefix.objects.filter(prefix__family=6)
  235. self.assertEqual(prefixes[0].prefix, IPNetwork('2001:db8::/32'))
  236. self.assertEqual(prefixes[0]._depth, 0)
  237. self.assertEqual(prefixes[0]._children, 3)
  238. self.assertEqual(prefixes[1].prefix, IPNetwork('2001:db8::/36'))
  239. self.assertEqual(prefixes[1]._depth, 1)
  240. self.assertEqual(prefixes[1]._children, 2)
  241. self.assertEqual(prefixes[2].prefix, IPNetwork('2001:db8::/40'))
  242. self.assertEqual(prefixes[2]._depth, 2)
  243. self.assertEqual(prefixes[2]._children, 1)
  244. self.assertEqual(prefixes[3].prefix, IPNetwork('2001:db8::/48'))
  245. self.assertEqual(prefixes[3]._depth, 3)
  246. self.assertEqual(prefixes[3]._children, 0)
  247. def test_update_prefix4(self):
  248. # Change 10.0.0.0/24 to 10.0.0.0/12
  249. p = Prefix.objects.get(prefix='10.0.0.0/24')
  250. p.prefix = '10.0.0.0/12'
  251. p.save()
  252. prefixes = Prefix.objects.filter(prefix__family=4)
  253. self.assertEqual(prefixes[0].prefix, IPNetwork('10.0.0.0/8'))
  254. self.assertEqual(prefixes[0]._depth, 0)
  255. self.assertEqual(prefixes[0]._children, 2)
  256. self.assertEqual(prefixes[1].prefix, IPNetwork('10.0.0.0/12'))
  257. self.assertEqual(prefixes[1]._depth, 1)
  258. self.assertEqual(prefixes[1]._children, 1)
  259. self.assertEqual(prefixes[2].prefix, IPNetwork('10.0.0.0/16'))
  260. self.assertEqual(prefixes[2]._depth, 2)
  261. self.assertEqual(prefixes[2]._children, 0)
  262. def test_update_prefix6(self):
  263. # Change 2001:db8::/48 to 2001:db8::/36
  264. p = Prefix.objects.get(prefix='2001:db8::/48')
  265. p.prefix = '2001:db8::/36'
  266. p.save()
  267. prefixes = Prefix.objects.filter(prefix__family=6)
  268. self.assertEqual(prefixes[0].prefix, IPNetwork('2001:db8::/32'))
  269. self.assertEqual(prefixes[0]._depth, 0)
  270. self.assertEqual(prefixes[0]._children, 2)
  271. self.assertEqual(prefixes[1].prefix, IPNetwork('2001:db8::/36'))
  272. self.assertEqual(prefixes[1]._depth, 1)
  273. self.assertEqual(prefixes[1]._children, 1)
  274. self.assertEqual(prefixes[2].prefix, IPNetwork('2001:db8::/40'))
  275. self.assertEqual(prefixes[2]._depth, 2)
  276. self.assertEqual(prefixes[2]._children, 0)
  277. def test_update_prefix_vrf4(self):
  278. vrf = VRF(name='VRF A')
  279. vrf.save()
  280. # Move 10.0.0.0/16 to a VRF
  281. p = Prefix.objects.get(prefix='10.0.0.0/16')
  282. p.vrf = vrf
  283. p.save()
  284. prefixes = Prefix.objects.filter(vrf__isnull=True, prefix__family=4)
  285. self.assertEqual(prefixes[0].prefix, IPNetwork('10.0.0.0/8'))
  286. self.assertEqual(prefixes[0]._depth, 0)
  287. self.assertEqual(prefixes[0]._children, 1)
  288. self.assertEqual(prefixes[1].prefix, IPNetwork('10.0.0.0/24'))
  289. self.assertEqual(prefixes[1]._depth, 1)
  290. self.assertEqual(prefixes[1]._children, 0)
  291. prefixes = Prefix.objects.filter(vrf=vrf)
  292. self.assertEqual(prefixes[0].prefix, IPNetwork('10.0.0.0/16'))
  293. self.assertEqual(prefixes[0]._depth, 0)
  294. self.assertEqual(prefixes[0]._children, 0)
  295. def test_update_prefix_vrf6(self):
  296. vrf = VRF(name='VRF A')
  297. vrf.save()
  298. # Move 2001:db8::/40 to a VRF
  299. p = Prefix.objects.get(prefix='2001:db8::/40')
  300. p.vrf = vrf
  301. p.save()
  302. prefixes = Prefix.objects.filter(vrf__isnull=True, prefix__family=6)
  303. self.assertEqual(prefixes[0].prefix, IPNetwork('2001:db8::/32'))
  304. self.assertEqual(prefixes[0]._depth, 0)
  305. self.assertEqual(prefixes[0]._children, 1)
  306. self.assertEqual(prefixes[1].prefix, IPNetwork('2001:db8::/48'))
  307. self.assertEqual(prefixes[1]._depth, 1)
  308. self.assertEqual(prefixes[1]._children, 0)
  309. prefixes = Prefix.objects.filter(vrf=vrf)
  310. self.assertEqual(prefixes[0].prefix, IPNetwork('2001:db8::/40'))
  311. self.assertEqual(prefixes[0]._depth, 0)
  312. self.assertEqual(prefixes[0]._children, 0)
  313. def test_delete_prefix4(self):
  314. # Delete 10.0.0.0/16
  315. Prefix.objects.filter(prefix='10.0.0.0/16').delete()
  316. prefixes = Prefix.objects.filter(prefix__family=4)
  317. self.assertEqual(prefixes[0].prefix, IPNetwork('10.0.0.0/8'))
  318. self.assertEqual(prefixes[0]._depth, 0)
  319. self.assertEqual(prefixes[0]._children, 1)
  320. self.assertEqual(prefixes[1].prefix, IPNetwork('10.0.0.0/24'))
  321. self.assertEqual(prefixes[1]._depth, 1)
  322. self.assertEqual(prefixes[1]._children, 0)
  323. def test_delete_prefix6(self):
  324. # Delete 2001:db8::/40
  325. Prefix.objects.filter(prefix='2001:db8::/40').delete()
  326. prefixes = Prefix.objects.filter(prefix__family=6)
  327. self.assertEqual(prefixes[0].prefix, IPNetwork('2001:db8::/32'))
  328. self.assertEqual(prefixes[0]._depth, 0)
  329. self.assertEqual(prefixes[0]._children, 1)
  330. self.assertEqual(prefixes[1].prefix, IPNetwork('2001:db8::/48'))
  331. self.assertEqual(prefixes[1]._depth, 1)
  332. self.assertEqual(prefixes[1]._children, 0)
  333. def test_duplicate_prefix4(self):
  334. # Duplicate 10.0.0.0/16
  335. Prefix(prefix='10.0.0.0/16').save()
  336. prefixes = Prefix.objects.filter(prefix__family=4)
  337. self.assertEqual(prefixes[0].prefix, IPNetwork('10.0.0.0/8'))
  338. self.assertEqual(prefixes[0]._depth, 0)
  339. self.assertEqual(prefixes[0]._children, 3)
  340. self.assertEqual(prefixes[1].prefix, IPNetwork('10.0.0.0/16'))
  341. self.assertEqual(prefixes[1]._depth, 1)
  342. self.assertEqual(prefixes[1]._children, 1)
  343. self.assertEqual(prefixes[2].prefix, IPNetwork('10.0.0.0/16'))
  344. self.assertEqual(prefixes[2]._depth, 1)
  345. self.assertEqual(prefixes[2]._children, 1)
  346. self.assertEqual(prefixes[3].prefix, IPNetwork('10.0.0.0/24'))
  347. self.assertEqual(prefixes[3]._depth, 2)
  348. self.assertEqual(prefixes[3]._children, 0)
  349. def test_duplicate_prefix6(self):
  350. # Duplicate 2001:db8::/40
  351. Prefix(prefix='2001:db8::/40').save()
  352. prefixes = Prefix.objects.filter(prefix__family=6)
  353. self.assertEqual(prefixes[0].prefix, IPNetwork('2001:db8::/32'))
  354. self.assertEqual(prefixes[0]._depth, 0)
  355. self.assertEqual(prefixes[0]._children, 3)
  356. self.assertEqual(prefixes[1].prefix, IPNetwork('2001:db8::/40'))
  357. self.assertEqual(prefixes[1]._depth, 1)
  358. self.assertEqual(prefixes[1]._children, 1)
  359. self.assertEqual(prefixes[2].prefix, IPNetwork('2001:db8::/40'))
  360. self.assertEqual(prefixes[2]._depth, 1)
  361. self.assertEqual(prefixes[2]._children, 1)
  362. self.assertEqual(prefixes[3].prefix, IPNetwork('2001:db8::/48'))
  363. self.assertEqual(prefixes[3]._depth, 2)
  364. self.assertEqual(prefixes[3]._children, 0)
  365. class TestIPAddress(TestCase):
  366. def test_get_duplicates(self):
  367. ips = IPAddress.objects.bulk_create((
  368. IPAddress(address=IPNetwork('192.0.2.1/24')),
  369. IPAddress(address=IPNetwork('192.0.2.1/24')),
  370. IPAddress(address=IPNetwork('192.0.2.1/24')),
  371. ))
  372. duplicate_ip_pks = [p.pk for p in ips[0].get_duplicates()]
  373. self.assertSetEqual(set(duplicate_ip_pks), {ips[1].pk, ips[2].pk})
  374. #
  375. # Uniqueness enforcement tests
  376. #
  377. @override_settings(ENFORCE_GLOBAL_UNIQUE=False)
  378. def test_duplicate_global(self):
  379. IPAddress.objects.create(address=IPNetwork('192.0.2.1/24'))
  380. duplicate_ip = IPAddress(address=IPNetwork('192.0.2.1/24'))
  381. self.assertIsNone(duplicate_ip.clean())
  382. @override_settings(ENFORCE_GLOBAL_UNIQUE=True)
  383. def test_duplicate_global_unique(self):
  384. IPAddress.objects.create(address=IPNetwork('192.0.2.1/24'))
  385. duplicate_ip = IPAddress(address=IPNetwork('192.0.2.1/24'))
  386. self.assertRaises(ValidationError, duplicate_ip.clean)
  387. def test_duplicate_vrf(self):
  388. vrf = VRF.objects.create(name='Test', rd='1:1', enforce_unique=False)
  389. IPAddress.objects.create(vrf=vrf, address=IPNetwork('192.0.2.1/24'))
  390. duplicate_ip = IPAddress(vrf=vrf, address=IPNetwork('192.0.2.1/24'))
  391. self.assertIsNone(duplicate_ip.clean())
  392. def test_duplicate_vrf_unique(self):
  393. vrf = VRF.objects.create(name='Test', rd='1:1', enforce_unique=True)
  394. IPAddress.objects.create(vrf=vrf, address=IPNetwork('192.0.2.1/24'))
  395. duplicate_ip = IPAddress(vrf=vrf, address=IPNetwork('192.0.2.1/24'))
  396. self.assertRaises(ValidationError, duplicate_ip.clean)
  397. @override_settings(ENFORCE_GLOBAL_UNIQUE=True)
  398. def test_duplicate_nonunique_nonrole_role(self):
  399. IPAddress.objects.create(address=IPNetwork('192.0.2.1/24'))
  400. duplicate_ip = IPAddress(address=IPNetwork('192.0.2.1/24'), role=IPAddressRoleChoices.ROLE_VIP)
  401. self.assertRaises(ValidationError, duplicate_ip.clean)
  402. @override_settings(ENFORCE_GLOBAL_UNIQUE=True)
  403. def test_duplicate_nonunique_role_nonrole(self):
  404. IPAddress.objects.create(address=IPNetwork('192.0.2.1/24'), role=IPAddressRoleChoices.ROLE_VIP)
  405. duplicate_ip = IPAddress(address=IPNetwork('192.0.2.1/24'))
  406. self.assertRaises(ValidationError, duplicate_ip.clean)
  407. @override_settings(ENFORCE_GLOBAL_UNIQUE=True)
  408. def test_duplicate_nonunique_role(self):
  409. IPAddress.objects.create(address=IPNetwork('192.0.2.1/24'), role=IPAddressRoleChoices.ROLE_VIP)
  410. IPAddress.objects.create(address=IPNetwork('192.0.2.1/24'), role=IPAddressRoleChoices.ROLE_VIP)
  411. class TestVLANGroup(TestCase):
  412. def test_get_next_available_vid(self):
  413. vlangroup = VLANGroup.objects.create(name='VLAN Group 1', slug='vlan-group-1')
  414. VLAN.objects.bulk_create((
  415. VLAN(name='VLAN 1', vid=1, group=vlangroup),
  416. VLAN(name='VLAN 2', vid=2, group=vlangroup),
  417. VLAN(name='VLAN 3', vid=3, group=vlangroup),
  418. VLAN(name='VLAN 5', vid=5, group=vlangroup),
  419. ))
  420. self.assertEqual(vlangroup.get_next_available_vid(), 4)
  421. VLAN.objects.bulk_create((
  422. VLAN(name='VLAN 4', vid=4, group=vlangroup),
  423. ))
  424. self.assertEqual(vlangroup.get_next_available_vid(), 6)