test_models.py 29 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734
  1. from django.contrib.contenttypes.models import ContentType
  2. from django.core.exceptions import ValidationError
  3. from django.test import TestCase, override_settings
  4. from netaddr import IPNetwork, IPSet
  5. from utilities.data import string_to_ranges
  6. from dcim.models import Site, SiteGroup
  7. from ipam.choices import *
  8. from ipam.models import *
  9. class TestAggregate(TestCase):
  10. def test_get_utilization(self):
  11. rir = RIR.objects.create(name='RIR 1', slug='rir-1')
  12. aggregate = Aggregate(prefix=IPNetwork('10.0.0.0/8'), rir=rir)
  13. aggregate.save()
  14. # 25% utilization
  15. Prefix.objects.bulk_create((
  16. Prefix(prefix=IPNetwork('10.0.0.0/12')),
  17. Prefix(prefix=IPNetwork('10.16.0.0/12')),
  18. Prefix(prefix=IPNetwork('10.32.0.0/12')),
  19. Prefix(prefix=IPNetwork('10.48.0.0/12')),
  20. ))
  21. self.assertEqual(aggregate.get_utilization(), 25)
  22. # 50% utilization
  23. Prefix.objects.bulk_create((
  24. Prefix(prefix=IPNetwork('10.64.0.0/10')),
  25. ))
  26. self.assertEqual(aggregate.get_utilization(), 50)
  27. # 100% utilization
  28. Prefix.objects.bulk_create((
  29. Prefix(prefix=IPNetwork('10.128.0.0/9')),
  30. ))
  31. self.assertEqual(aggregate.get_utilization(), 100)
  32. class TestIPRange(TestCase):
  33. def test_overlapping_range(self):
  34. iprange_192_168 = IPRange.objects.create(
  35. start_address=IPNetwork('192.168.0.1/22'), end_address=IPNetwork('192.168.0.49/22')
  36. )
  37. iprange_192_168.clean()
  38. iprange_3_1_99 = IPRange.objects.create(
  39. start_address=IPNetwork('1.2.3.1/24'), end_address=IPNetwork('1.2.3.99/24')
  40. )
  41. iprange_3_1_99.clean()
  42. iprange_3_100_199 = IPRange.objects.create(
  43. start_address=IPNetwork('1.2.3.100/24'), end_address=IPNetwork('1.2.3.199/24')
  44. )
  45. iprange_3_100_199.clean()
  46. iprange_3_200_255 = IPRange.objects.create(
  47. start_address=IPNetwork('1.2.3.200/24'), end_address=IPNetwork('1.2.3.255/24')
  48. )
  49. iprange_3_200_255.clean()
  50. iprange_4_1_99 = IPRange.objects.create(
  51. start_address=IPNetwork('1.2.4.1/24'), end_address=IPNetwork('1.2.4.99/24')
  52. )
  53. iprange_4_1_99.clean()
  54. iprange_4_200 = IPRange.objects.create(
  55. start_address=IPNetwork('1.2.4.200/24'), end_address=IPNetwork('1.2.4.255/24')
  56. )
  57. iprange_4_200.clean()
  58. # Overlapping range entirely within existing
  59. with self.assertRaises(ValidationError):
  60. iprange_3_123_124 = IPRange.objects.create(
  61. start_address=IPNetwork('1.2.3.123/26'), end_address=IPNetwork('1.2.3.124/26')
  62. )
  63. iprange_3_123_124.clean()
  64. # Overlapping range starting within existing
  65. with self.assertRaises(ValidationError):
  66. iprange_4_98_101 = IPRange.objects.create(
  67. start_address=IPNetwork('1.2.4.98/24'), end_address=IPNetwork('1.2.4.101/24')
  68. )
  69. iprange_4_98_101.clean()
  70. # Overlapping range ending within existing
  71. with self.assertRaises(ValidationError):
  72. iprange_4_198_201 = IPRange.objects.create(
  73. start_address=IPNetwork('1.2.4.198/24'), end_address=IPNetwork('1.2.4.201/24')
  74. )
  75. iprange_4_198_201.clean()
  76. class TestPrefix(TestCase):
  77. def test_get_duplicates(self):
  78. prefixes = Prefix.objects.bulk_create((
  79. Prefix(prefix=IPNetwork('192.0.2.0/24')),
  80. Prefix(prefix=IPNetwork('192.0.2.0/24')),
  81. Prefix(prefix=IPNetwork('192.0.2.0/24')),
  82. ))
  83. duplicate_prefix_pks = [p.pk for p in prefixes[0].get_duplicates()]
  84. self.assertSetEqual(set(duplicate_prefix_pks), {prefixes[1].pk, prefixes[2].pk})
  85. def test_get_child_prefixes(self):
  86. vrfs = VRF.objects.bulk_create((
  87. VRF(name='VRF 1'),
  88. VRF(name='VRF 2'),
  89. VRF(name='VRF 3'),
  90. ))
  91. prefixes = Prefix.objects.bulk_create((
  92. Prefix(prefix=IPNetwork('10.0.0.0/16'), status=PrefixStatusChoices.STATUS_CONTAINER),
  93. Prefix(prefix=IPNetwork('10.0.0.0/24'), vrf=None),
  94. Prefix(prefix=IPNetwork('10.0.1.0/24'), vrf=vrfs[0]),
  95. Prefix(prefix=IPNetwork('10.0.2.0/24'), vrf=vrfs[1]),
  96. Prefix(prefix=IPNetwork('10.0.3.0/24'), vrf=vrfs[2]),
  97. ))
  98. child_prefix_pks = {p.pk for p in prefixes[0].get_child_prefixes()}
  99. # Global container should return all children
  100. self.assertSetEqual(child_prefix_pks, {prefixes[1].pk, prefixes[2].pk, prefixes[3].pk, prefixes[4].pk})
  101. prefixes[0].vrf = vrfs[0]
  102. prefixes[0].save()
  103. child_prefix_pks = {p.pk for p in prefixes[0].get_child_prefixes()}
  104. # VRF container is limited to its own VRF
  105. self.assertSetEqual(child_prefix_pks, {prefixes[2].pk})
  106. def test_get_child_ranges(self):
  107. prefix = Prefix(prefix='192.168.0.16/28')
  108. prefix.save()
  109. ranges = IPRange.objects.bulk_create(
  110. (
  111. # No overlap
  112. IPRange(
  113. start_address=IPNetwork('192.168.0.1/24'), end_address=IPNetwork('192.168.0.10/24'), size=10
  114. ),
  115. # Partial overlap
  116. IPRange(
  117. start_address=IPNetwork('192.168.0.11/24'), end_address=IPNetwork('192.168.0.17/24'), size=7
  118. ),
  119. # Full overlap
  120. IPRange(
  121. start_address=IPNetwork('192.168.0.18/24'), end_address=IPNetwork('192.168.0.23/24'), size=6
  122. ),
  123. # Full overlap
  124. IPRange(
  125. start_address=IPNetwork('192.168.0.24/24'), end_address=IPNetwork('192.168.0.30/24'), size=7
  126. ),
  127. # Partial overlap
  128. IPRange(
  129. start_address=IPNetwork('192.168.0.31/24'), end_address=IPNetwork('192.168.0.40/24'), size=10
  130. ),
  131. )
  132. )
  133. child_ranges = prefix.get_child_ranges()
  134. self.assertEqual(len(child_ranges), 2)
  135. self.assertEqual(child_ranges[0], ranges[2])
  136. self.assertEqual(child_ranges[1], ranges[3])
  137. def test_get_child_ips(self):
  138. vrfs = VRF.objects.bulk_create((
  139. VRF(name='VRF 1'),
  140. VRF(name='VRF 2'),
  141. VRF(name='VRF 3'),
  142. ))
  143. parent_prefix = Prefix.objects.create(
  144. prefix=IPNetwork('10.0.0.0/16'), status=PrefixStatusChoices.STATUS_CONTAINER
  145. )
  146. ips = IPAddress.objects.bulk_create((
  147. IPAddress(address=IPNetwork('10.0.0.1/24'), vrf=None),
  148. IPAddress(address=IPNetwork('10.0.1.1/24'), vrf=vrfs[0]),
  149. IPAddress(address=IPNetwork('10.0.2.1/24'), vrf=vrfs[1]),
  150. IPAddress(address=IPNetwork('10.0.3.1/24'), vrf=vrfs[2]),
  151. ))
  152. child_ip_pks = {p.pk for p in parent_prefix.get_child_ips()}
  153. # Global container should return all children
  154. self.assertSetEqual(child_ip_pks, {ips[0].pk, ips[1].pk, ips[2].pk, ips[3].pk})
  155. parent_prefix.vrf = vrfs[0]
  156. parent_prefix.save()
  157. child_ip_pks = {p.pk for p in parent_prefix.get_child_ips()}
  158. # VRF container is limited to its own VRF
  159. self.assertSetEqual(child_ip_pks, {ips[1].pk})
  160. def test_get_available_prefixes(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/20')),
  164. Prefix(prefix=IPNetwork('10.0.32.0/20')),
  165. Prefix(prefix=IPNetwork('10.0.128.0/18')),
  166. ))
  167. missing_prefixes = IPSet([
  168. IPNetwork('10.0.16.0/20'),
  169. IPNetwork('10.0.48.0/20'),
  170. IPNetwork('10.0.64.0/18'),
  171. IPNetwork('10.0.192.0/18'),
  172. ])
  173. available_prefixes = prefixes[0].get_available_prefixes()
  174. self.assertEqual(available_prefixes, missing_prefixes)
  175. def test_get_available_ips(self):
  176. parent_prefix = Prefix.objects.create(prefix=IPNetwork('10.0.0.0/28'))
  177. IPAddress.objects.bulk_create((
  178. IPAddress(address=IPNetwork('10.0.0.1/26')),
  179. IPAddress(address=IPNetwork('10.0.0.3/26')),
  180. IPAddress(address=IPNetwork('10.0.0.5/26')),
  181. IPAddress(address=IPNetwork('10.0.0.7/26')),
  182. ))
  183. # Range is not marked as populated, so it doesn't count against available IP space
  184. IPRange.objects.create(
  185. start_address=IPNetwork('10.0.0.9/26'),
  186. end_address=IPNetwork('10.0.0.10/26')
  187. )
  188. # Populated range reduces available IP space
  189. IPRange.objects.create(
  190. start_address=IPNetwork('10.0.0.12/26'),
  191. end_address=IPNetwork('10.0.0.13/26'),
  192. mark_populated=True
  193. )
  194. missing_ips = IPSet([
  195. '10.0.0.2/32',
  196. '10.0.0.4/32',
  197. '10.0.0.6/32',
  198. '10.0.0.8/32',
  199. '10.0.0.9/32',
  200. '10.0.0.10/32',
  201. '10.0.0.11/32',
  202. '10.0.0.14/32',
  203. ])
  204. available_ips = parent_prefix.get_available_ips()
  205. self.assertEqual(available_ips, missing_ips)
  206. def test_get_first_available_prefix(self):
  207. prefixes = Prefix.objects.bulk_create((
  208. Prefix(prefix=IPNetwork('10.0.0.0/16')), # Parent prefix
  209. Prefix(prefix=IPNetwork('10.0.0.0/24')),
  210. Prefix(prefix=IPNetwork('10.0.1.0/24')),
  211. Prefix(prefix=IPNetwork('10.0.2.0/24')),
  212. ))
  213. self.assertEqual(prefixes[0].get_first_available_prefix(), IPNetwork('10.0.3.0/24'))
  214. Prefix.objects.create(prefix=IPNetwork('10.0.3.0/24'))
  215. self.assertEqual(prefixes[0].get_first_available_prefix(), IPNetwork('10.0.4.0/22'))
  216. def test_get_first_available_ip(self):
  217. parent_prefix = Prefix.objects.create(prefix=IPNetwork('10.0.0.0/24'))
  218. IPAddress.objects.bulk_create((
  219. IPAddress(address=IPNetwork('10.0.0.1/24')),
  220. IPAddress(address=IPNetwork('10.0.0.2/24')),
  221. IPAddress(address=IPNetwork('10.0.0.3/24')),
  222. ))
  223. self.assertEqual(parent_prefix.get_first_available_ip(), '10.0.0.4/24')
  224. IPAddress.objects.create(address=IPNetwork('10.0.0.4/24'))
  225. self.assertEqual(parent_prefix.get_first_available_ip(), '10.0.0.5/24')
  226. def test_get_first_available_ip_ipv6(self):
  227. parent_prefix = Prefix.objects.create(prefix=IPNetwork('2001:db8:500::/64'))
  228. self.assertEqual(parent_prefix.get_first_available_ip(), '2001:db8:500::1/64')
  229. def test_get_first_available_ip_ipv6_rfc3627(self):
  230. parent_prefix = Prefix.objects.create(prefix=IPNetwork('2001:db8:500:4::/126'))
  231. self.assertEqual(parent_prefix.get_first_available_ip(), '2001:db8:500:4::1/126')
  232. def test_get_first_available_ip_ipv6_rfc6164(self):
  233. parent_prefix = Prefix.objects.create(prefix=IPNetwork('2001:db8:500:5::/127'))
  234. self.assertEqual(parent_prefix.get_first_available_ip(), '2001:db8:500:5::/127')
  235. def test_get_utilization_container(self):
  236. prefixes = (
  237. Prefix(prefix=IPNetwork('10.0.0.0/24'), status=PrefixStatusChoices.STATUS_CONTAINER),
  238. Prefix(prefix=IPNetwork('10.0.0.0/26')),
  239. Prefix(prefix=IPNetwork('10.0.0.128/26')),
  240. )
  241. Prefix.objects.bulk_create(prefixes)
  242. self.assertEqual(prefixes[0].get_utilization(), 50) # 50% utilization
  243. def test_get_utilization_noncontainer(self):
  244. prefix = Prefix.objects.create(
  245. prefix=IPNetwork('10.0.0.0/24'),
  246. status=PrefixStatusChoices.STATUS_ACTIVE
  247. )
  248. # Create 32 child IPs
  249. IPAddress.objects.bulk_create([
  250. IPAddress(address=IPNetwork(f'10.0.0.{i}/24')) for i in range(1, 33)
  251. ])
  252. self.assertEqual(prefix.get_utilization(), 32 / 254 * 100) # ~12.5% utilization
  253. # Create a utilized child range with 32 additional IPs
  254. IPRange.objects.create(
  255. start_address=IPNetwork('10.0.0.33/24'),
  256. end_address=IPNetwork('10.0.0.64/24'),
  257. mark_utilized=True
  258. )
  259. self.assertEqual(prefix.get_utilization(), 64 / 254 * 100) # ~25% utilization
  260. #
  261. # Uniqueness enforcement tests
  262. #
  263. @override_settings(ENFORCE_GLOBAL_UNIQUE=False)
  264. def test_duplicate_global(self):
  265. Prefix.objects.create(prefix=IPNetwork('192.0.2.0/24'))
  266. duplicate_prefix = Prefix(prefix=IPNetwork('192.0.2.0/24'))
  267. self.assertIsNone(duplicate_prefix.clean())
  268. def test_duplicate_global_unique(self):
  269. Prefix.objects.create(prefix=IPNetwork('192.0.2.0/24'))
  270. duplicate_prefix = Prefix(prefix=IPNetwork('192.0.2.0/24'))
  271. self.assertRaises(ValidationError, duplicate_prefix.clean)
  272. def test_duplicate_vrf(self):
  273. vrf = VRF.objects.create(name='Test', rd='1:1', enforce_unique=False)
  274. Prefix.objects.create(vrf=vrf, prefix=IPNetwork('192.0.2.0/24'))
  275. duplicate_prefix = Prefix(vrf=vrf, prefix=IPNetwork('192.0.2.0/24'))
  276. self.assertIsNone(duplicate_prefix.clean())
  277. def test_duplicate_vrf_unique(self):
  278. vrf = VRF.objects.create(name='Test', rd='1:1', enforce_unique=True)
  279. Prefix.objects.create(vrf=vrf, prefix=IPNetwork('192.0.2.0/24'))
  280. duplicate_prefix = Prefix(vrf=vrf, prefix=IPNetwork('192.0.2.0/24'))
  281. self.assertRaises(ValidationError, duplicate_prefix.clean)
  282. class TestPrefixHierarchy(TestCase):
  283. """
  284. Test the automatic updating of depth and child count in response to changes made within
  285. the prefix hierarchy.
  286. """
  287. @classmethod
  288. def setUpTestData(cls):
  289. prefixes = (
  290. # IPv4
  291. Prefix(prefix='10.0.0.0/8', _depth=0, _children=2),
  292. Prefix(prefix='10.0.0.0/16', _depth=1, _children=1),
  293. Prefix(prefix='10.0.0.0/24', _depth=2, _children=0),
  294. # IPv6
  295. Prefix(prefix='2001:db8::/32', _depth=0, _children=2),
  296. Prefix(prefix='2001:db8::/40', _depth=1, _children=1),
  297. Prefix(prefix='2001:db8::/48', _depth=2, _children=0),
  298. )
  299. Prefix.objects.bulk_create(prefixes)
  300. def test_create_prefix4(self):
  301. # Create 10.0.0.0/12
  302. Prefix(prefix='10.0.0.0/12').save()
  303. prefixes = Prefix.objects.filter(prefix__family=4)
  304. self.assertEqual(prefixes[0].prefix, IPNetwork('10.0.0.0/8'))
  305. self.assertEqual(prefixes[0]._depth, 0)
  306. self.assertEqual(prefixes[0]._children, 3)
  307. self.assertEqual(prefixes[1].prefix, IPNetwork('10.0.0.0/12'))
  308. self.assertEqual(prefixes[1]._depth, 1)
  309. self.assertEqual(prefixes[1]._children, 2)
  310. self.assertEqual(prefixes[2].prefix, IPNetwork('10.0.0.0/16'))
  311. self.assertEqual(prefixes[2]._depth, 2)
  312. self.assertEqual(prefixes[2]._children, 1)
  313. self.assertEqual(prefixes[3].prefix, IPNetwork('10.0.0.0/24'))
  314. self.assertEqual(prefixes[3]._depth, 3)
  315. self.assertEqual(prefixes[3]._children, 0)
  316. def test_create_prefix6(self):
  317. # Create 2001:db8::/36
  318. Prefix(prefix='2001:db8::/36').save()
  319. prefixes = Prefix.objects.filter(prefix__family=6)
  320. self.assertEqual(prefixes[0].prefix, IPNetwork('2001:db8::/32'))
  321. self.assertEqual(prefixes[0]._depth, 0)
  322. self.assertEqual(prefixes[0]._children, 3)
  323. self.assertEqual(prefixes[1].prefix, IPNetwork('2001:db8::/36'))
  324. self.assertEqual(prefixes[1]._depth, 1)
  325. self.assertEqual(prefixes[1]._children, 2)
  326. self.assertEqual(prefixes[2].prefix, IPNetwork('2001:db8::/40'))
  327. self.assertEqual(prefixes[2]._depth, 2)
  328. self.assertEqual(prefixes[2]._children, 1)
  329. self.assertEqual(prefixes[3].prefix, IPNetwork('2001:db8::/48'))
  330. self.assertEqual(prefixes[3]._depth, 3)
  331. self.assertEqual(prefixes[3]._children, 0)
  332. def test_update_prefix4(self):
  333. # Change 10.0.0.0/24 to 10.0.0.0/12
  334. p = Prefix.objects.get(prefix='10.0.0.0/24')
  335. p.prefix = '10.0.0.0/12'
  336. p.save()
  337. prefixes = Prefix.objects.filter(prefix__family=4)
  338. self.assertEqual(prefixes[0].prefix, IPNetwork('10.0.0.0/8'))
  339. self.assertEqual(prefixes[0]._depth, 0)
  340. self.assertEqual(prefixes[0]._children, 2)
  341. self.assertEqual(prefixes[1].prefix, IPNetwork('10.0.0.0/12'))
  342. self.assertEqual(prefixes[1]._depth, 1)
  343. self.assertEqual(prefixes[1]._children, 1)
  344. self.assertEqual(prefixes[2].prefix, IPNetwork('10.0.0.0/16'))
  345. self.assertEqual(prefixes[2]._depth, 2)
  346. self.assertEqual(prefixes[2]._children, 0)
  347. def test_update_prefix6(self):
  348. # Change 2001:db8::/48 to 2001:db8::/36
  349. p = Prefix.objects.get(prefix='2001:db8::/48')
  350. p.prefix = '2001:db8::/36'
  351. p.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, 2)
  356. self.assertEqual(prefixes[1].prefix, IPNetwork('2001:db8::/36'))
  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, 2)
  361. self.assertEqual(prefixes[2]._children, 0)
  362. def test_update_prefix_vrf4(self):
  363. vrf = VRF(name='VRF A')
  364. vrf.save()
  365. # Move 10.0.0.0/16 to a VRF
  366. p = Prefix.objects.get(prefix='10.0.0.0/16')
  367. p.vrf = vrf
  368. p.save()
  369. prefixes = Prefix.objects.filter(vrf__isnull=True, prefix__family=4)
  370. self.assertEqual(prefixes[0].prefix, IPNetwork('10.0.0.0/8'))
  371. self.assertEqual(prefixes[0]._depth, 0)
  372. self.assertEqual(prefixes[0]._children, 1)
  373. self.assertEqual(prefixes[1].prefix, IPNetwork('10.0.0.0/24'))
  374. self.assertEqual(prefixes[1]._depth, 1)
  375. self.assertEqual(prefixes[1]._children, 0)
  376. prefixes = Prefix.objects.filter(vrf=vrf)
  377. self.assertEqual(prefixes[0].prefix, IPNetwork('10.0.0.0/16'))
  378. self.assertEqual(prefixes[0]._depth, 0)
  379. self.assertEqual(prefixes[0]._children, 0)
  380. def test_update_prefix_vrf6(self):
  381. vrf = VRF(name='VRF A')
  382. vrf.save()
  383. # Move 2001:db8::/40 to a VRF
  384. p = Prefix.objects.get(prefix='2001:db8::/40')
  385. p.vrf = vrf
  386. p.save()
  387. prefixes = Prefix.objects.filter(vrf__isnull=True, 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, 1)
  391. self.assertEqual(prefixes[1].prefix, IPNetwork('2001:db8::/48'))
  392. self.assertEqual(prefixes[1]._depth, 1)
  393. self.assertEqual(prefixes[1]._children, 0)
  394. prefixes = Prefix.objects.filter(vrf=vrf)
  395. self.assertEqual(prefixes[0].prefix, IPNetwork('2001:db8::/40'))
  396. self.assertEqual(prefixes[0]._depth, 0)
  397. self.assertEqual(prefixes[0]._children, 0)
  398. def test_delete_prefix4(self):
  399. # Delete 10.0.0.0/16
  400. Prefix.objects.filter(prefix='10.0.0.0/16').delete()
  401. prefixes = Prefix.objects.filter(prefix__family=4)
  402. self.assertEqual(prefixes[0].prefix, IPNetwork('10.0.0.0/8'))
  403. self.assertEqual(prefixes[0]._depth, 0)
  404. self.assertEqual(prefixes[0]._children, 1)
  405. self.assertEqual(prefixes[1].prefix, IPNetwork('10.0.0.0/24'))
  406. self.assertEqual(prefixes[1]._depth, 1)
  407. self.assertEqual(prefixes[1]._children, 0)
  408. def test_delete_prefix6(self):
  409. # Delete 2001:db8::/40
  410. Prefix.objects.filter(prefix='2001:db8::/40').delete()
  411. prefixes = Prefix.objects.filter(prefix__family=6)
  412. self.assertEqual(prefixes[0].prefix, IPNetwork('2001:db8::/32'))
  413. self.assertEqual(prefixes[0]._depth, 0)
  414. self.assertEqual(prefixes[0]._children, 1)
  415. self.assertEqual(prefixes[1].prefix, IPNetwork('2001:db8::/48'))
  416. self.assertEqual(prefixes[1]._depth, 1)
  417. self.assertEqual(prefixes[1]._children, 0)
  418. def test_duplicate_prefix4(self):
  419. # Duplicate 10.0.0.0/16
  420. Prefix(prefix='10.0.0.0/16').save()
  421. prefixes = Prefix.objects.filter(prefix__family=4)
  422. self.assertEqual(prefixes[0].prefix, IPNetwork('10.0.0.0/8'))
  423. self.assertEqual(prefixes[0]._depth, 0)
  424. self.assertEqual(prefixes[0]._children, 3)
  425. self.assertEqual(prefixes[1].prefix, IPNetwork('10.0.0.0/16'))
  426. self.assertEqual(prefixes[1]._depth, 1)
  427. self.assertEqual(prefixes[1]._children, 1)
  428. self.assertEqual(prefixes[2].prefix, IPNetwork('10.0.0.0/16'))
  429. self.assertEqual(prefixes[2]._depth, 1)
  430. self.assertEqual(prefixes[2]._children, 1)
  431. self.assertEqual(prefixes[3].prefix, IPNetwork('10.0.0.0/24'))
  432. self.assertEqual(prefixes[3]._depth, 2)
  433. self.assertEqual(prefixes[3]._children, 0)
  434. def test_duplicate_prefix6(self):
  435. # Duplicate 2001:db8::/40
  436. Prefix(prefix='2001:db8::/40').save()
  437. prefixes = Prefix.objects.filter(prefix__family=6)
  438. self.assertEqual(prefixes[0].prefix, IPNetwork('2001:db8::/32'))
  439. self.assertEqual(prefixes[0]._depth, 0)
  440. self.assertEqual(prefixes[0]._children, 3)
  441. self.assertEqual(prefixes[1].prefix, IPNetwork('2001:db8::/40'))
  442. self.assertEqual(prefixes[1]._depth, 1)
  443. self.assertEqual(prefixes[1]._children, 1)
  444. self.assertEqual(prefixes[2].prefix, IPNetwork('2001:db8::/40'))
  445. self.assertEqual(prefixes[2]._depth, 1)
  446. self.assertEqual(prefixes[2]._children, 1)
  447. self.assertEqual(prefixes[3].prefix, IPNetwork('2001:db8::/48'))
  448. self.assertEqual(prefixes[3]._depth, 2)
  449. self.assertEqual(prefixes[3]._children, 0)
  450. class TestIPAddress(TestCase):
  451. def test_get_duplicates(self):
  452. ips = IPAddress.objects.bulk_create((
  453. IPAddress(address=IPNetwork('192.0.2.1/24')),
  454. IPAddress(address=IPNetwork('192.0.2.1/24')),
  455. IPAddress(address=IPNetwork('192.0.2.1/24')),
  456. ))
  457. duplicate_ip_pks = [p.pk for p in ips[0].get_duplicates()]
  458. self.assertSetEqual(set(duplicate_ip_pks), {ips[1].pk, ips[2].pk})
  459. #
  460. # Uniqueness enforcement tests
  461. #
  462. @override_settings(ENFORCE_GLOBAL_UNIQUE=False)
  463. def test_duplicate_global(self):
  464. IPAddress.objects.create(address=IPNetwork('192.0.2.1/24'))
  465. duplicate_ip = IPAddress(address=IPNetwork('192.0.2.1/24'))
  466. self.assertIsNone(duplicate_ip.clean())
  467. def test_duplicate_global_unique(self):
  468. IPAddress.objects.create(address=IPNetwork('192.0.2.1/24'))
  469. duplicate_ip = IPAddress(address=IPNetwork('192.0.2.1/24'))
  470. self.assertRaises(ValidationError, duplicate_ip.clean)
  471. def test_duplicate_vrf(self):
  472. vrf = VRF.objects.create(name='Test', rd='1:1', enforce_unique=False)
  473. IPAddress.objects.create(vrf=vrf, address=IPNetwork('192.0.2.1/24'))
  474. duplicate_ip = IPAddress(vrf=vrf, address=IPNetwork('192.0.2.1/24'))
  475. self.assertIsNone(duplicate_ip.clean())
  476. def test_duplicate_vrf_unique(self):
  477. vrf = VRF.objects.create(name='Test', rd='1:1', enforce_unique=True)
  478. IPAddress.objects.create(vrf=vrf, address=IPNetwork('192.0.2.1/24'))
  479. duplicate_ip = IPAddress(vrf=vrf, address=IPNetwork('192.0.2.1/24'))
  480. self.assertRaises(ValidationError, duplicate_ip.clean)
  481. def test_duplicate_nonunique_nonrole_role(self):
  482. IPAddress.objects.create(address=IPNetwork('192.0.2.1/24'))
  483. duplicate_ip = IPAddress(address=IPNetwork('192.0.2.1/24'), role=IPAddressRoleChoices.ROLE_VIP)
  484. self.assertRaises(ValidationError, duplicate_ip.clean)
  485. def test_duplicate_nonunique_role_nonrole(self):
  486. IPAddress.objects.create(address=IPNetwork('192.0.2.1/24'), role=IPAddressRoleChoices.ROLE_VIP)
  487. duplicate_ip = IPAddress(address=IPNetwork('192.0.2.1/24'))
  488. self.assertRaises(ValidationError, duplicate_ip.clean)
  489. def test_duplicate_nonunique_role(self):
  490. IPAddress.objects.create(address=IPNetwork('192.0.2.1/24'), role=IPAddressRoleChoices.ROLE_VIP)
  491. IPAddress.objects.create(address=IPNetwork('192.0.2.1/24'), role=IPAddressRoleChoices.ROLE_VIP)
  492. #
  493. # Range validation
  494. #
  495. def test_create_ip_in_unpopulated_range(self):
  496. IPRange.objects.create(
  497. start_address=IPNetwork('192.0.2.1/24'),
  498. end_address=IPNetwork('192.0.2.100/24')
  499. )
  500. ip = IPAddress(address=IPNetwork('192.0.2.10/24'))
  501. ip.full_clean()
  502. def test_create_ip_in_populated_range(self):
  503. IPRange.objects.create(
  504. start_address=IPNetwork('192.0.2.1/24'),
  505. end_address=IPNetwork('192.0.2.100/24'),
  506. mark_populated=True
  507. )
  508. ip = IPAddress(address=IPNetwork('192.0.2.10/24'))
  509. self.assertRaises(ValidationError, ip.full_clean)
  510. class TestVLANGroup(TestCase):
  511. @classmethod
  512. def setUpTestData(cls):
  513. vlangroup = VLANGroup.objects.create(
  514. name='VLAN Group 1',
  515. slug='vlan-group-1',
  516. vid_ranges=string_to_ranges('100-199'),
  517. )
  518. VLAN.objects.bulk_create((
  519. VLAN(name='VLAN 100', vid=100, group=vlangroup),
  520. VLAN(name='VLAN 101', vid=101, group=vlangroup),
  521. VLAN(name='VLAN 102', vid=102, group=vlangroup),
  522. VLAN(name='VLAN 103', vid=103, group=vlangroup),
  523. ))
  524. def test_get_available_vids(self):
  525. vlangroup = VLANGroup.objects.first()
  526. child_vids = VLAN.objects.filter(group=vlangroup).values_list('vid', flat=True)
  527. self.assertEqual(len(child_vids), 4)
  528. available_vids = vlangroup.get_available_vids()
  529. self.assertListEqual(available_vids, list(range(104, 200)))
  530. def test_get_next_available_vid(self):
  531. vlangroup = VLANGroup.objects.first()
  532. self.assertEqual(vlangroup.get_next_available_vid(), 104)
  533. VLAN.objects.create(name='VLAN 104', vid=104, group=vlangroup)
  534. self.assertEqual(vlangroup.get_next_available_vid(), 105)
  535. def test_vid_validation(self):
  536. vlangroup = VLANGroup.objects.first()
  537. vlan = VLAN(vid=1, name='VLAN 1', group=vlangroup)
  538. with self.assertRaises(ValidationError):
  539. vlan.full_clean()
  540. vlan = VLAN(vid=109, name='VLAN 109', group=vlangroup)
  541. vlan.full_clean()
  542. def test_overlapping_vlan(self):
  543. vlangroup = VLANGroup(
  544. name='VLAN Group 1',
  545. slug='vlan-group-1',
  546. vid_ranges=string_to_ranges('2-4,3-5'),
  547. )
  548. with self.assertRaises(ValidationError):
  549. vlangroup.full_clean()
  550. # make sure single vlan range works
  551. vlangroup.vid_ranges = string_to_ranges('2-2')
  552. vlangroup.full_clean()
  553. vlangroup.save()
  554. class TestVLAN(TestCase):
  555. @classmethod
  556. def setUpTestData(cls):
  557. VLAN.objects.bulk_create((
  558. VLAN(name='VLAN 1', vid=1, qinq_role=VLANQinQRoleChoices.ROLE_SERVICE),
  559. ))
  560. def test_qinq_role(self):
  561. svlan = VLAN.objects.filter(qinq_role=VLANQinQRoleChoices.ROLE_SERVICE).first()
  562. vlan = VLAN(
  563. name='VLAN X',
  564. vid=999,
  565. qinq_role=VLANQinQRoleChoices.ROLE_SERVICE,
  566. qinq_svlan=svlan
  567. )
  568. with self.assertRaises(ValidationError):
  569. vlan.full_clean()
  570. def test_vlan_group_site_validation(self):
  571. sitegroup = SiteGroup.objects.create(
  572. name='Site Group 1',
  573. slug='site-group-1',
  574. )
  575. sites = Site.objects.bulk_create((
  576. Site(
  577. name='Site 1',
  578. slug='site-1',
  579. ),
  580. Site(
  581. name='Site 2',
  582. slug='site-2',
  583. ),
  584. ))
  585. sitegroup.sites.add(sites[0])
  586. vlangroups = VLANGroup.objects.bulk_create((
  587. VLANGroup(
  588. name='VLAN Group 1',
  589. slug='vlan-group-1',
  590. scope=sitegroup,
  591. scope_type=ContentType.objects.get_for_model(SiteGroup),
  592. ),
  593. VLANGroup(
  594. name='VLAN Group 2',
  595. slug='vlan-group-2',
  596. scope=sites[0],
  597. scope_type=ContentType.objects.get_for_model(Site),
  598. ),
  599. VLANGroup(
  600. name='VLAN Group 2',
  601. slug='vlan-group-2',
  602. scope=sites[1],
  603. scope_type=ContentType.objects.get_for_model(Site),
  604. ),
  605. ))
  606. vlan = VLAN(
  607. name='VLAN 1',
  608. vid=1,
  609. group=vlangroups[0],
  610. site=sites[0],
  611. )
  612. # VLAN Group 1 and 2 should be valid
  613. vlan.full_clean()
  614. vlan.group = vlangroups[1]
  615. vlan.full_clean()
  616. vlan.group = vlangroups[2]
  617. with self.assertRaises(ValidationError):
  618. vlan.full_clean()