test_api.py 35 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993
  1. from __future__ import unicode_literals
  2. from django.urls import reverse
  3. from netaddr import IPNetwork
  4. from rest_framework import status
  5. from dcim.models import Device, DeviceRole, DeviceType, Manufacturer, Site
  6. from ipam.constants import IP_PROTOCOL_TCP, IP_PROTOCOL_UDP
  7. from ipam.models import Aggregate, IPAddress, Prefix, RIR, Role, Service, VLAN, VLANGroup, VRF
  8. from utilities.testing import APITestCase
  9. class VRFTest(APITestCase):
  10. def setUp(self):
  11. super(VRFTest, self).setUp()
  12. self.vrf1 = VRF.objects.create(name='Test VRF 1', rd='65000:1')
  13. self.vrf2 = VRF.objects.create(name='Test VRF 2', rd='65000:2')
  14. self.vrf3 = VRF.objects.create(name='Test VRF 3', rd='65000:3')
  15. def test_get_vrf(self):
  16. url = reverse('ipam-api:vrf-detail', kwargs={'pk': self.vrf1.pk})
  17. response = self.client.get(url, **self.header)
  18. self.assertEqual(response.data['name'], self.vrf1.name)
  19. def test_list_vrfs(self):
  20. url = reverse('ipam-api:vrf-list')
  21. response = self.client.get(url, **self.header)
  22. self.assertEqual(response.data['count'], 3)
  23. def test_create_vrf(self):
  24. data = {
  25. 'name': 'Test VRF 4',
  26. 'rd': '65000:4',
  27. }
  28. url = reverse('ipam-api:vrf-list')
  29. response = self.client.post(url, data, format='json', **self.header)
  30. self.assertHttpStatus(response, status.HTTP_201_CREATED)
  31. self.assertEqual(VRF.objects.count(), 4)
  32. vrf4 = VRF.objects.get(pk=response.data['id'])
  33. self.assertEqual(vrf4.name, data['name'])
  34. self.assertEqual(vrf4.rd, data['rd'])
  35. def test_create_vrf_bulk(self):
  36. data = [
  37. {
  38. 'name': 'Test VRF 4',
  39. 'rd': '65000:4',
  40. },
  41. {
  42. 'name': 'Test VRF 5',
  43. 'rd': '65000:5',
  44. },
  45. {
  46. 'name': 'Test VRF 6',
  47. 'rd': '65000:6',
  48. },
  49. ]
  50. url = reverse('ipam-api:vrf-list')
  51. response = self.client.post(url, data, format='json', **self.header)
  52. self.assertHttpStatus(response, status.HTTP_201_CREATED)
  53. self.assertEqual(VRF.objects.count(), 6)
  54. self.assertEqual(response.data[0]['name'], data[0]['name'])
  55. self.assertEqual(response.data[1]['name'], data[1]['name'])
  56. self.assertEqual(response.data[2]['name'], data[2]['name'])
  57. def test_update_vrf(self):
  58. data = {
  59. 'name': 'Test VRF X',
  60. 'rd': '65000:99',
  61. }
  62. url = reverse('ipam-api:vrf-detail', kwargs={'pk': self.vrf1.pk})
  63. response = self.client.put(url, data, format='json', **self.header)
  64. self.assertHttpStatus(response, status.HTTP_200_OK)
  65. self.assertEqual(VRF.objects.count(), 3)
  66. vrf1 = VRF.objects.get(pk=response.data['id'])
  67. self.assertEqual(vrf1.name, data['name'])
  68. self.assertEqual(vrf1.rd, data['rd'])
  69. def test_delete_vrf(self):
  70. url = reverse('ipam-api:vrf-detail', kwargs={'pk': self.vrf1.pk})
  71. response = self.client.delete(url, **self.header)
  72. self.assertHttpStatus(response, status.HTTP_204_NO_CONTENT)
  73. self.assertEqual(VRF.objects.count(), 2)
  74. class RIRTest(APITestCase):
  75. def setUp(self):
  76. super(RIRTest, self).setUp()
  77. self.rir1 = RIR.objects.create(name='Test RIR 1', slug='test-rir-1')
  78. self.rir2 = RIR.objects.create(name='Test RIR 2', slug='test-rir-2')
  79. self.rir3 = RIR.objects.create(name='Test RIR 3', slug='test-rir-3')
  80. def test_get_rir(self):
  81. url = reverse('ipam-api:rir-detail', kwargs={'pk': self.rir1.pk})
  82. response = self.client.get(url, **self.header)
  83. self.assertEqual(response.data['name'], self.rir1.name)
  84. def test_list_rirs(self):
  85. url = reverse('ipam-api:rir-list')
  86. response = self.client.get(url, **self.header)
  87. self.assertEqual(response.data['count'], 3)
  88. def test_create_rir(self):
  89. data = {
  90. 'name': 'Test RIR 4',
  91. 'slug': 'test-rir-4',
  92. }
  93. url = reverse('ipam-api:rir-list')
  94. response = self.client.post(url, data, format='json', **self.header)
  95. self.assertHttpStatus(response, status.HTTP_201_CREATED)
  96. self.assertEqual(RIR.objects.count(), 4)
  97. rir4 = RIR.objects.get(pk=response.data['id'])
  98. self.assertEqual(rir4.name, data['name'])
  99. self.assertEqual(rir4.slug, data['slug'])
  100. def test_create_rir_bulk(self):
  101. data = [
  102. {
  103. 'name': 'Test RIR 4',
  104. 'slug': 'test-rir-4',
  105. },
  106. {
  107. 'name': 'Test RIR 5',
  108. 'slug': 'test-rir-5',
  109. },
  110. {
  111. 'name': 'Test RIR 6',
  112. 'slug': 'test-rir-6',
  113. },
  114. ]
  115. url = reverse('ipam-api:rir-list')
  116. response = self.client.post(url, data, format='json', **self.header)
  117. self.assertHttpStatus(response, status.HTTP_201_CREATED)
  118. self.assertEqual(RIR.objects.count(), 6)
  119. self.assertEqual(response.data[0]['name'], data[0]['name'])
  120. self.assertEqual(response.data[1]['name'], data[1]['name'])
  121. self.assertEqual(response.data[2]['name'], data[2]['name'])
  122. def test_update_rir(self):
  123. data = {
  124. 'name': 'Test RIR X',
  125. 'slug': 'test-rir-x',
  126. }
  127. url = reverse('ipam-api:rir-detail', kwargs={'pk': self.rir1.pk})
  128. response = self.client.put(url, data, format='json', **self.header)
  129. self.assertHttpStatus(response, status.HTTP_200_OK)
  130. self.assertEqual(RIR.objects.count(), 3)
  131. rir1 = RIR.objects.get(pk=response.data['id'])
  132. self.assertEqual(rir1.name, data['name'])
  133. self.assertEqual(rir1.slug, data['slug'])
  134. def test_delete_rir(self):
  135. url = reverse('ipam-api:rir-detail', kwargs={'pk': self.rir1.pk})
  136. response = self.client.delete(url, **self.header)
  137. self.assertHttpStatus(response, status.HTTP_204_NO_CONTENT)
  138. self.assertEqual(RIR.objects.count(), 2)
  139. class AggregateTest(APITestCase):
  140. def setUp(self):
  141. super(AggregateTest, self).setUp()
  142. self.rir1 = RIR.objects.create(name='Test RIR 1', slug='test-rir-1')
  143. self.rir2 = RIR.objects.create(name='Test RIR 2', slug='test-rir-2')
  144. self.aggregate1 = Aggregate.objects.create(prefix=IPNetwork('10.0.0.0/8'), rir=self.rir1)
  145. self.aggregate2 = Aggregate.objects.create(prefix=IPNetwork('172.16.0.0/12'), rir=self.rir1)
  146. self.aggregate3 = Aggregate.objects.create(prefix=IPNetwork('192.168.0.0/16'), rir=self.rir1)
  147. def test_get_aggregate(self):
  148. url = reverse('ipam-api:aggregate-detail', kwargs={'pk': self.aggregate1.pk})
  149. response = self.client.get(url, **self.header)
  150. self.assertEqual(response.data['prefix'], str(self.aggregate1.prefix))
  151. def test_list_aggregates(self):
  152. url = reverse('ipam-api:aggregate-list')
  153. response = self.client.get(url, **self.header)
  154. self.assertEqual(response.data['count'], 3)
  155. def test_create_aggregate(self):
  156. data = {
  157. 'prefix': '192.0.2.0/24',
  158. 'rir': self.rir1.pk,
  159. }
  160. url = reverse('ipam-api:aggregate-list')
  161. response = self.client.post(url, data, format='json', **self.header)
  162. self.assertHttpStatus(response, status.HTTP_201_CREATED)
  163. self.assertEqual(Aggregate.objects.count(), 4)
  164. aggregate4 = Aggregate.objects.get(pk=response.data['id'])
  165. self.assertEqual(str(aggregate4.prefix), data['prefix'])
  166. self.assertEqual(aggregate4.rir_id, data['rir'])
  167. def test_create_aggregate_bulk(self):
  168. data = [
  169. {
  170. 'prefix': '100.0.0.0/8',
  171. 'rir': self.rir1.pk,
  172. },
  173. {
  174. 'prefix': '101.0.0.0/8',
  175. 'rir': self.rir1.pk,
  176. },
  177. {
  178. 'prefix': '102.0.0.0/8',
  179. 'rir': self.rir1.pk,
  180. },
  181. ]
  182. url = reverse('ipam-api:aggregate-list')
  183. response = self.client.post(url, data, format='json', **self.header)
  184. self.assertHttpStatus(response, status.HTTP_201_CREATED)
  185. self.assertEqual(Aggregate.objects.count(), 6)
  186. self.assertEqual(response.data[0]['prefix'], data[0]['prefix'])
  187. self.assertEqual(response.data[1]['prefix'], data[1]['prefix'])
  188. self.assertEqual(response.data[2]['prefix'], data[2]['prefix'])
  189. def test_update_aggregate(self):
  190. data = {
  191. 'prefix': '11.0.0.0/8',
  192. 'rir': self.rir2.pk,
  193. }
  194. url = reverse('ipam-api:aggregate-detail', kwargs={'pk': self.aggregate1.pk})
  195. response = self.client.put(url, data, format='json', **self.header)
  196. self.assertHttpStatus(response, status.HTTP_200_OK)
  197. self.assertEqual(Aggregate.objects.count(), 3)
  198. aggregate1 = Aggregate.objects.get(pk=response.data['id'])
  199. self.assertEqual(str(aggregate1.prefix), data['prefix'])
  200. self.assertEqual(aggregate1.rir_id, data['rir'])
  201. def test_delete_aggregate(self):
  202. url = reverse('ipam-api:aggregate-detail', kwargs={'pk': self.aggregate1.pk})
  203. response = self.client.delete(url, **self.header)
  204. self.assertHttpStatus(response, status.HTTP_204_NO_CONTENT)
  205. self.assertEqual(Aggregate.objects.count(), 2)
  206. class RoleTest(APITestCase):
  207. def setUp(self):
  208. super(RoleTest, self).setUp()
  209. self.role1 = Role.objects.create(name='Test Role 1', slug='test-role-1')
  210. self.role2 = Role.objects.create(name='Test Role 2', slug='test-role-2')
  211. self.role3 = Role.objects.create(name='Test Role 3', slug='test-role-3')
  212. def test_get_role(self):
  213. url = reverse('ipam-api:role-detail', kwargs={'pk': self.role1.pk})
  214. response = self.client.get(url, **self.header)
  215. self.assertEqual(response.data['name'], self.role1.name)
  216. def test_list_roles(self):
  217. url = reverse('ipam-api:role-list')
  218. response = self.client.get(url, **self.header)
  219. self.assertEqual(response.data['count'], 3)
  220. def test_create_role(self):
  221. data = {
  222. 'name': 'Test Role 4',
  223. 'slug': 'test-role-4',
  224. }
  225. url = reverse('ipam-api:role-list')
  226. response = self.client.post(url, data, format='json', **self.header)
  227. self.assertHttpStatus(response, status.HTTP_201_CREATED)
  228. self.assertEqual(Role.objects.count(), 4)
  229. role4 = Role.objects.get(pk=response.data['id'])
  230. self.assertEqual(role4.name, data['name'])
  231. self.assertEqual(role4.slug, data['slug'])
  232. def test_create_role_bulk(self):
  233. data = [
  234. {
  235. 'name': 'Test Role 4',
  236. 'slug': 'test-role-4',
  237. },
  238. {
  239. 'name': 'Test Role 5',
  240. 'slug': 'test-role-5',
  241. },
  242. {
  243. 'name': 'Test Role 6',
  244. 'slug': 'test-role-6',
  245. },
  246. ]
  247. url = reverse('ipam-api:role-list')
  248. response = self.client.post(url, data, format='json', **self.header)
  249. self.assertHttpStatus(response, status.HTTP_201_CREATED)
  250. self.assertEqual(Role.objects.count(), 6)
  251. self.assertEqual(response.data[0]['name'], data[0]['name'])
  252. self.assertEqual(response.data[1]['name'], data[1]['name'])
  253. self.assertEqual(response.data[2]['name'], data[2]['name'])
  254. def test_update_role(self):
  255. data = {
  256. 'name': 'Test Role X',
  257. 'slug': 'test-role-x',
  258. }
  259. url = reverse('ipam-api:role-detail', kwargs={'pk': self.role1.pk})
  260. response = self.client.put(url, data, format='json', **self.header)
  261. self.assertHttpStatus(response, status.HTTP_200_OK)
  262. self.assertEqual(Role.objects.count(), 3)
  263. role1 = Role.objects.get(pk=response.data['id'])
  264. self.assertEqual(role1.name, data['name'])
  265. self.assertEqual(role1.slug, data['slug'])
  266. def test_delete_role(self):
  267. url = reverse('ipam-api:role-detail', kwargs={'pk': self.role1.pk})
  268. response = self.client.delete(url, **self.header)
  269. self.assertHttpStatus(response, status.HTTP_204_NO_CONTENT)
  270. self.assertEqual(Role.objects.count(), 2)
  271. class PrefixTest(APITestCase):
  272. def setUp(self):
  273. super(PrefixTest, self).setUp()
  274. self.site1 = Site.objects.create(name='Test Site 1', slug='test-site-1')
  275. self.vrf1 = VRF.objects.create(name='Test VRF 1', rd='65000:1')
  276. self.vlan1 = VLAN.objects.create(vid=1, name='Test VLAN 1')
  277. self.role1 = Role.objects.create(name='Test Role 1', slug='test-role-1')
  278. self.prefix1 = Prefix.objects.create(prefix=IPNetwork('192.168.1.0/24'))
  279. self.prefix2 = Prefix.objects.create(prefix=IPNetwork('192.168.2.0/24'))
  280. self.prefix3 = Prefix.objects.create(prefix=IPNetwork('192.168.3.0/24'))
  281. def test_get_prefix(self):
  282. url = reverse('ipam-api:prefix-detail', kwargs={'pk': self.prefix1.pk})
  283. response = self.client.get(url, **self.header)
  284. self.assertEqual(response.data['prefix'], str(self.prefix1.prefix))
  285. def test_list_prefixs(self):
  286. url = reverse('ipam-api:prefix-list')
  287. response = self.client.get(url, **self.header)
  288. self.assertEqual(response.data['count'], 3)
  289. def test_create_prefix(self):
  290. data = {
  291. 'prefix': '192.168.4.0/24',
  292. 'site': self.site1.pk,
  293. 'vrf': self.vrf1.pk,
  294. 'vlan': self.vlan1.pk,
  295. 'role': self.role1.pk,
  296. }
  297. url = reverse('ipam-api:prefix-list')
  298. response = self.client.post(url, data, format='json', **self.header)
  299. self.assertHttpStatus(response, status.HTTP_201_CREATED)
  300. self.assertEqual(Prefix.objects.count(), 4)
  301. prefix4 = Prefix.objects.get(pk=response.data['id'])
  302. self.assertEqual(str(prefix4.prefix), data['prefix'])
  303. self.assertEqual(prefix4.site_id, data['site'])
  304. self.assertEqual(prefix4.vrf_id, data['vrf'])
  305. self.assertEqual(prefix4.vlan_id, data['vlan'])
  306. self.assertEqual(prefix4.role_id, data['role'])
  307. def test_create_prefix_bulk(self):
  308. data = [
  309. {
  310. 'prefix': '10.0.1.0/24',
  311. },
  312. {
  313. 'prefix': '10.0.2.0/24',
  314. },
  315. {
  316. 'prefix': '10.0.3.0/24',
  317. },
  318. ]
  319. url = reverse('ipam-api:prefix-list')
  320. response = self.client.post(url, data, format='json', **self.header)
  321. self.assertHttpStatus(response, status.HTTP_201_CREATED)
  322. self.assertEqual(Prefix.objects.count(), 6)
  323. self.assertEqual(response.data[0]['prefix'], data[0]['prefix'])
  324. self.assertEqual(response.data[1]['prefix'], data[1]['prefix'])
  325. self.assertEqual(response.data[2]['prefix'], data[2]['prefix'])
  326. def test_update_prefix(self):
  327. data = {
  328. 'prefix': '192.168.99.0/24',
  329. 'site': self.site1.pk,
  330. 'vrf': self.vrf1.pk,
  331. 'vlan': self.vlan1.pk,
  332. 'role': self.role1.pk,
  333. }
  334. url = reverse('ipam-api:prefix-detail', kwargs={'pk': self.prefix1.pk})
  335. response = self.client.put(url, data, format='json', **self.header)
  336. self.assertHttpStatus(response, status.HTTP_200_OK)
  337. self.assertEqual(Prefix.objects.count(), 3)
  338. prefix1 = Prefix.objects.get(pk=response.data['id'])
  339. self.assertEqual(str(prefix1.prefix), data['prefix'])
  340. self.assertEqual(prefix1.site_id, data['site'])
  341. self.assertEqual(prefix1.vrf_id, data['vrf'])
  342. self.assertEqual(prefix1.vlan_id, data['vlan'])
  343. self.assertEqual(prefix1.role_id, data['role'])
  344. def test_delete_prefix(self):
  345. url = reverse('ipam-api:prefix-detail', kwargs={'pk': self.prefix1.pk})
  346. response = self.client.delete(url, **self.header)
  347. self.assertHttpStatus(response, status.HTTP_204_NO_CONTENT)
  348. self.assertEqual(Prefix.objects.count(), 2)
  349. def test_list_available_prefixes(self):
  350. prefix = Prefix.objects.create(prefix=IPNetwork('192.0.2.0/24'))
  351. Prefix.objects.create(prefix=IPNetwork('192.0.2.64/26'))
  352. Prefix.objects.create(prefix=IPNetwork('192.0.2.192/27'))
  353. url = reverse('ipam-api:prefix-available-prefixes', kwargs={'pk': prefix.pk})
  354. # Retrieve all available IPs
  355. response = self.client.get(url, **self.header)
  356. available_prefixes = ['192.0.2.0/26', '192.0.2.128/26', '192.0.2.224/27']
  357. for i, p in enumerate(response.data):
  358. self.assertEqual(p['prefix'], available_prefixes[i])
  359. def test_create_single_available_prefix(self):
  360. prefix = Prefix.objects.create(prefix=IPNetwork('192.0.2.0/28'), is_pool=True)
  361. url = reverse('ipam-api:prefix-available-prefixes', kwargs={'pk': prefix.pk})
  362. # Create four available prefixes with individual requests
  363. prefixes_to_be_created = [
  364. '192.0.2.0/30',
  365. '192.0.2.4/30',
  366. '192.0.2.8/30',
  367. '192.0.2.12/30',
  368. ]
  369. for i in range(4):
  370. data = {
  371. 'prefix_length': 30,
  372. 'description': 'Test Prefix {}'.format(i + 1)
  373. }
  374. response = self.client.post(url, data, format='json', **self.header)
  375. self.assertHttpStatus(response, status.HTTP_201_CREATED)
  376. self.assertEqual(response.data['prefix'], prefixes_to_be_created[i])
  377. self.assertEqual(response.data['description'], data['description'])
  378. # Try to create one more prefix
  379. response = self.client.post(url, {'prefix_length': 30}, **self.header)
  380. self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
  381. self.assertIn('detail', response.data)
  382. def test_create_multiple_available_prefixes(self):
  383. prefix = Prefix.objects.create(prefix=IPNetwork('192.0.2.0/28'), is_pool=True)
  384. url = reverse('ipam-api:prefix-available-prefixes', kwargs={'pk': prefix.pk})
  385. # Try to create five /30s (only four are available)
  386. data = [
  387. {'prefix_length': 30, 'description': 'Test Prefix 1'},
  388. {'prefix_length': 30, 'description': 'Test Prefix 2'},
  389. {'prefix_length': 30, 'description': 'Test Prefix 3'},
  390. {'prefix_length': 30, 'description': 'Test Prefix 4'},
  391. {'prefix_length': 30, 'description': 'Test Prefix 5'},
  392. ]
  393. response = self.client.post(url, data, format='json', **self.header)
  394. self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
  395. self.assertIn('detail', response.data)
  396. # Verify that no prefixes were created (the entire /28 is still available)
  397. response = self.client.get(url, **self.header)
  398. self.assertEqual(response.data[0]['prefix'], '192.0.2.0/28')
  399. # Create four /30s in a single request
  400. response = self.client.post(url, data[:4], format='json', **self.header)
  401. self.assertHttpStatus(response, status.HTTP_201_CREATED)
  402. self.assertEqual(len(response.data), 4)
  403. def test_list_available_ips(self):
  404. prefix = Prefix.objects.create(prefix=IPNetwork('192.0.2.0/29'), is_pool=True)
  405. url = reverse('ipam-api:prefix-available-ips', kwargs={'pk': prefix.pk})
  406. # Retrieve all available IPs
  407. response = self.client.get(url, **self.header)
  408. self.assertEqual(len(response.data), 8) # 8 because prefix.is_pool = True
  409. # Change the prefix to not be a pool and try again
  410. prefix.is_pool = False
  411. prefix.save()
  412. response = self.client.get(url, **self.header)
  413. self.assertEqual(len(response.data), 6) # 8 - 2 because prefix.is_pool = False
  414. def test_create_single_available_ip(self):
  415. prefix = Prefix.objects.create(prefix=IPNetwork('192.0.2.0/30'), is_pool=True)
  416. url = reverse('ipam-api:prefix-available-ips', kwargs={'pk': prefix.pk})
  417. # Create all four available IPs with individual requests
  418. for i in range(1, 5):
  419. data = {
  420. 'description': 'Test IP {}'.format(i)
  421. }
  422. response = self.client.post(url, data, format='json', **self.header)
  423. self.assertHttpStatus(response, status.HTTP_201_CREATED)
  424. self.assertEqual(response.data['description'], data['description'])
  425. # Try to create one more IP
  426. response = self.client.post(url, {}, **self.header)
  427. self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
  428. self.assertIn('detail', response.data)
  429. def test_create_multiple_available_ips(self):
  430. prefix = Prefix.objects.create(prefix=IPNetwork('192.0.2.0/29'), is_pool=True)
  431. url = reverse('ipam-api:prefix-available-ips', kwargs={'pk': prefix.pk})
  432. # Try to create nine IPs (only eight are available)
  433. data = [{'description': 'Test IP {}'.format(i)} for i in range(1, 10)] # 9 IPs
  434. response = self.client.post(url, data, format='json', **self.header)
  435. self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
  436. self.assertIn('detail', response.data)
  437. # Verify that no IPs were created (eight are still available)
  438. response = self.client.get(url, **self.header)
  439. self.assertEqual(len(response.data), 8)
  440. # Create all eight available IPs in a single request
  441. data = [{'description': 'Test IP {}'.format(i)} for i in range(1, 9)] # 8 IPs
  442. response = self.client.post(url, data, format='json', **self.header)
  443. self.assertHttpStatus(response, status.HTTP_201_CREATED)
  444. self.assertEqual(len(response.data), 8)
  445. class IPAddressTest(APITestCase):
  446. def setUp(self):
  447. super(IPAddressTest, self).setUp()
  448. self.vrf1 = VRF.objects.create(name='Test VRF 1', rd='65000:1')
  449. self.ipaddress1 = IPAddress.objects.create(address=IPNetwork('192.168.0.1/24'))
  450. self.ipaddress2 = IPAddress.objects.create(address=IPNetwork('192.168.0.2/24'))
  451. self.ipaddress3 = IPAddress.objects.create(address=IPNetwork('192.168.0.3/24'))
  452. def test_get_ipaddress(self):
  453. url = reverse('ipam-api:ipaddress-detail', kwargs={'pk': self.ipaddress1.pk})
  454. response = self.client.get(url, **self.header)
  455. self.assertEqual(response.data['address'], str(self.ipaddress1.address))
  456. def test_list_ipaddresss(self):
  457. url = reverse('ipam-api:ipaddress-list')
  458. response = self.client.get(url, **self.header)
  459. self.assertEqual(response.data['count'], 3)
  460. def test_create_ipaddress(self):
  461. data = {
  462. 'address': '192.168.0.4/24',
  463. 'vrf': self.vrf1.pk,
  464. }
  465. url = reverse('ipam-api:ipaddress-list')
  466. response = self.client.post(url, data, format='json', **self.header)
  467. self.assertHttpStatus(response, status.HTTP_201_CREATED)
  468. self.assertEqual(IPAddress.objects.count(), 4)
  469. ipaddress4 = IPAddress.objects.get(pk=response.data['id'])
  470. self.assertEqual(str(ipaddress4.address), data['address'])
  471. self.assertEqual(ipaddress4.vrf_id, data['vrf'])
  472. def test_create_ipaddress_bulk(self):
  473. data = [
  474. {
  475. 'address': '192.168.0.4/24',
  476. },
  477. {
  478. 'address': '192.168.0.5/24',
  479. },
  480. {
  481. 'address': '192.168.0.6/24',
  482. },
  483. ]
  484. url = reverse('ipam-api:ipaddress-list')
  485. response = self.client.post(url, data, format='json', **self.header)
  486. self.assertHttpStatus(response, status.HTTP_201_CREATED)
  487. self.assertEqual(IPAddress.objects.count(), 6)
  488. self.assertEqual(response.data[0]['address'], data[0]['address'])
  489. self.assertEqual(response.data[1]['address'], data[1]['address'])
  490. self.assertEqual(response.data[2]['address'], data[2]['address'])
  491. def test_update_ipaddress(self):
  492. data = {
  493. 'address': '192.168.0.99/24',
  494. 'vrf': self.vrf1.pk,
  495. }
  496. url = reverse('ipam-api:ipaddress-detail', kwargs={'pk': self.ipaddress1.pk})
  497. response = self.client.put(url, data, format='json', **self.header)
  498. self.assertHttpStatus(response, status.HTTP_200_OK)
  499. self.assertEqual(IPAddress.objects.count(), 3)
  500. ipaddress1 = IPAddress.objects.get(pk=response.data['id'])
  501. self.assertEqual(str(ipaddress1.address), data['address'])
  502. self.assertEqual(ipaddress1.vrf_id, data['vrf'])
  503. def test_delete_ipaddress(self):
  504. url = reverse('ipam-api:ipaddress-detail', kwargs={'pk': self.ipaddress1.pk})
  505. response = self.client.delete(url, **self.header)
  506. self.assertHttpStatus(response, status.HTTP_204_NO_CONTENT)
  507. self.assertEqual(IPAddress.objects.count(), 2)
  508. class VLANGroupTest(APITestCase):
  509. def setUp(self):
  510. super(VLANGroupTest, self).setUp()
  511. self.vlangroup1 = VLANGroup.objects.create(name='Test VLAN Group 1', slug='test-vlan-group-1')
  512. self.vlangroup2 = VLANGroup.objects.create(name='Test VLAN Group 2', slug='test-vlan-group-2')
  513. self.vlangroup3 = VLANGroup.objects.create(name='Test VLAN Group 3', slug='test-vlan-group-3')
  514. def test_get_vlangroup(self):
  515. url = reverse('ipam-api:vlangroup-detail', kwargs={'pk': self.vlangroup1.pk})
  516. response = self.client.get(url, **self.header)
  517. self.assertEqual(response.data['name'], self.vlangroup1.name)
  518. def test_list_vlangroups(self):
  519. url = reverse('ipam-api:vlangroup-list')
  520. response = self.client.get(url, **self.header)
  521. self.assertEqual(response.data['count'], 3)
  522. def test_create_vlangroup(self):
  523. data = {
  524. 'name': 'Test VLAN Group 4',
  525. 'slug': 'test-vlan-group-4',
  526. }
  527. url = reverse('ipam-api:vlangroup-list')
  528. response = self.client.post(url, data, format='json', **self.header)
  529. self.assertHttpStatus(response, status.HTTP_201_CREATED)
  530. self.assertEqual(VLANGroup.objects.count(), 4)
  531. vlangroup4 = VLANGroup.objects.get(pk=response.data['id'])
  532. self.assertEqual(vlangroup4.name, data['name'])
  533. self.assertEqual(vlangroup4.slug, data['slug'])
  534. def test_create_vlangroup_bulk(self):
  535. data = [
  536. {
  537. 'name': 'Test VLAN Group 4',
  538. 'slug': 'test-vlan-group-4',
  539. },
  540. {
  541. 'name': 'Test VLAN Group 5',
  542. 'slug': 'test-vlan-group-5',
  543. },
  544. {
  545. 'name': 'Test VLAN Group 6',
  546. 'slug': 'test-vlan-group-6',
  547. },
  548. ]
  549. url = reverse('ipam-api:vlangroup-list')
  550. response = self.client.post(url, data, format='json', **self.header)
  551. self.assertHttpStatus(response, status.HTTP_201_CREATED)
  552. self.assertEqual(VLANGroup.objects.count(), 6)
  553. self.assertEqual(response.data[0]['name'], data[0]['name'])
  554. self.assertEqual(response.data[1]['name'], data[1]['name'])
  555. self.assertEqual(response.data[2]['name'], data[2]['name'])
  556. def test_update_vlangroup(self):
  557. data = {
  558. 'name': 'Test VLAN Group X',
  559. 'slug': 'test-vlan-group-x',
  560. }
  561. url = reverse('ipam-api:vlangroup-detail', kwargs={'pk': self.vlangroup1.pk})
  562. response = self.client.put(url, data, format='json', **self.header)
  563. self.assertHttpStatus(response, status.HTTP_200_OK)
  564. self.assertEqual(VLANGroup.objects.count(), 3)
  565. vlangroup1 = VLANGroup.objects.get(pk=response.data['id'])
  566. self.assertEqual(vlangroup1.name, data['name'])
  567. self.assertEqual(vlangroup1.slug, data['slug'])
  568. def test_delete_vlangroup(self):
  569. url = reverse('ipam-api:vlangroup-detail', kwargs={'pk': self.vlangroup1.pk})
  570. response = self.client.delete(url, **self.header)
  571. self.assertHttpStatus(response, status.HTTP_204_NO_CONTENT)
  572. self.assertEqual(VLANGroup.objects.count(), 2)
  573. class VLANTest(APITestCase):
  574. def setUp(self):
  575. super(VLANTest, self).setUp()
  576. self.vlan1 = VLAN.objects.create(vid=1, name='Test VLAN 1')
  577. self.vlan2 = VLAN.objects.create(vid=2, name='Test VLAN 2')
  578. self.vlan3 = VLAN.objects.create(vid=3, name='Test VLAN 3')
  579. def test_get_vlan(self):
  580. url = reverse('ipam-api:vlan-detail', kwargs={'pk': self.vlan1.pk})
  581. response = self.client.get(url, **self.header)
  582. self.assertEqual(response.data['name'], self.vlan1.name)
  583. def test_list_vlans(self):
  584. url = reverse('ipam-api:vlan-list')
  585. response = self.client.get(url, **self.header)
  586. self.assertEqual(response.data['count'], 3)
  587. def test_create_vlan(self):
  588. data = {
  589. 'vid': 4,
  590. 'name': 'Test VLAN 4',
  591. }
  592. url = reverse('ipam-api:vlan-list')
  593. response = self.client.post(url, data, format='json', **self.header)
  594. self.assertHttpStatus(response, status.HTTP_201_CREATED)
  595. self.assertEqual(VLAN.objects.count(), 4)
  596. vlan4 = VLAN.objects.get(pk=response.data['id'])
  597. self.assertEqual(vlan4.vid, data['vid'])
  598. self.assertEqual(vlan4.name, data['name'])
  599. def test_create_vlan_bulk(self):
  600. data = [
  601. {
  602. 'vid': 4,
  603. 'name': 'Test VLAN 4',
  604. },
  605. {
  606. 'vid': 5,
  607. 'name': 'Test VLAN 5',
  608. },
  609. {
  610. 'vid': 6,
  611. 'name': 'Test VLAN 6',
  612. },
  613. ]
  614. url = reverse('ipam-api:vlan-list')
  615. response = self.client.post(url, data, format='json', **self.header)
  616. self.assertHttpStatus(response, status.HTTP_201_CREATED)
  617. self.assertEqual(VLAN.objects.count(), 6)
  618. self.assertEqual(response.data[0]['name'], data[0]['name'])
  619. self.assertEqual(response.data[1]['name'], data[1]['name'])
  620. self.assertEqual(response.data[2]['name'], data[2]['name'])
  621. def test_update_vlan(self):
  622. data = {
  623. 'vid': 99,
  624. 'name': 'Test VLAN X',
  625. }
  626. url = reverse('ipam-api:vlan-detail', kwargs={'pk': self.vlan1.pk})
  627. response = self.client.put(url, data, format='json', **self.header)
  628. self.assertHttpStatus(response, status.HTTP_200_OK)
  629. self.assertEqual(VLAN.objects.count(), 3)
  630. vlan1 = VLAN.objects.get(pk=response.data['id'])
  631. self.assertEqual(vlan1.vid, data['vid'])
  632. self.assertEqual(vlan1.name, data['name'])
  633. def test_delete_vlan(self):
  634. url = reverse('ipam-api:vlan-detail', kwargs={'pk': self.vlan1.pk})
  635. response = self.client.delete(url, **self.header)
  636. self.assertHttpStatus(response, status.HTTP_204_NO_CONTENT)
  637. self.assertEqual(VLAN.objects.count(), 2)
  638. class ServiceTest(APITestCase):
  639. def setUp(self):
  640. super(ServiceTest, self).setUp()
  641. site = Site.objects.create(name='Test Site 1', slug='test-site-1')
  642. manufacturer = Manufacturer.objects.create(name='Test Manufacturer 1', slug='test-manufacturer-1')
  643. devicetype = DeviceType.objects.create(manufacturer=manufacturer, model='Test Device Type 1')
  644. devicerole = DeviceRole.objects.create(name='Test Device Role 1', slug='test-device-role-1')
  645. self.device1 = Device.objects.create(
  646. name='Test Device 1', site=site, device_type=devicetype, device_role=devicerole
  647. )
  648. self.device2 = Device.objects.create(
  649. name='Test Device 2', site=site, device_type=devicetype, device_role=devicerole
  650. )
  651. self.service1 = Service.objects.create(
  652. device=self.device1, name='Test Service 1', protocol=IP_PROTOCOL_TCP, port=1
  653. )
  654. self.service1 = Service.objects.create(
  655. device=self.device1, name='Test Service 2', protocol=IP_PROTOCOL_TCP, port=2
  656. )
  657. self.service1 = Service.objects.create(
  658. device=self.device1, name='Test Service 3', protocol=IP_PROTOCOL_TCP, port=3
  659. )
  660. def test_get_service(self):
  661. url = reverse('ipam-api:service-detail', kwargs={'pk': self.service1.pk})
  662. response = self.client.get(url, **self.header)
  663. self.assertEqual(response.data['name'], self.service1.name)
  664. def test_list_services(self):
  665. url = reverse('ipam-api:service-list')
  666. response = self.client.get(url, **self.header)
  667. self.assertEqual(response.data['count'], 3)
  668. def test_create_service(self):
  669. data = {
  670. 'device': self.device1.pk,
  671. 'name': 'Test Service 4',
  672. 'protocol': IP_PROTOCOL_TCP,
  673. 'port': 4,
  674. }
  675. url = reverse('ipam-api:service-list')
  676. response = self.client.post(url, data, format='json', **self.header)
  677. self.assertHttpStatus(response, status.HTTP_201_CREATED)
  678. self.assertEqual(Service.objects.count(), 4)
  679. service4 = Service.objects.get(pk=response.data['id'])
  680. self.assertEqual(service4.device_id, data['device'])
  681. self.assertEqual(service4.name, data['name'])
  682. self.assertEqual(service4.protocol, data['protocol'])
  683. self.assertEqual(service4.port, data['port'])
  684. def test_create_service_bulk(self):
  685. data = [
  686. {
  687. 'device': self.device1.pk,
  688. 'name': 'Test Service 4',
  689. 'protocol': IP_PROTOCOL_TCP,
  690. 'port': 4,
  691. },
  692. {
  693. 'device': self.device1.pk,
  694. 'name': 'Test Service 5',
  695. 'protocol': IP_PROTOCOL_TCP,
  696. 'port': 5,
  697. },
  698. {
  699. 'device': self.device1.pk,
  700. 'name': 'Test Service 6',
  701. 'protocol': IP_PROTOCOL_TCP,
  702. 'port': 6,
  703. },
  704. ]
  705. url = reverse('ipam-api:service-list')
  706. response = self.client.post(url, data, format='json', **self.header)
  707. self.assertHttpStatus(response, status.HTTP_201_CREATED)
  708. self.assertEqual(Service.objects.count(), 6)
  709. self.assertEqual(response.data[0]['name'], data[0]['name'])
  710. self.assertEqual(response.data[1]['name'], data[1]['name'])
  711. self.assertEqual(response.data[2]['name'], data[2]['name'])
  712. def test_update_service(self):
  713. data = {
  714. 'device': self.device2.pk,
  715. 'name': 'Test Service X',
  716. 'protocol': IP_PROTOCOL_UDP,
  717. 'port': 99,
  718. }
  719. url = reverse('ipam-api:service-detail', kwargs={'pk': self.service1.pk})
  720. response = self.client.put(url, data, format='json', **self.header)
  721. self.assertHttpStatus(response, status.HTTP_200_OK)
  722. self.assertEqual(Service.objects.count(), 3)
  723. service1 = Service.objects.get(pk=response.data['id'])
  724. self.assertEqual(service1.device_id, data['device'])
  725. self.assertEqual(service1.name, data['name'])
  726. self.assertEqual(service1.protocol, data['protocol'])
  727. self.assertEqual(service1.port, data['port'])
  728. def test_delete_service(self):
  729. url = reverse('ipam-api:service-detail', kwargs={'pk': self.service1.pk})
  730. response = self.client.delete(url, **self.header)
  731. self.assertHttpStatus(response, status.HTTP_204_NO_CONTENT)
  732. self.assertEqual(Service.objects.count(), 2)