test_models.py 25 KB

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