2
0

test_models.py 26 KB

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