test_models.py 44 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110
  1. from django.core.exceptions import ValidationError
  2. from django.test import TestCase
  3. from circuits.models import *
  4. from dcim.choices import *
  5. from dcim.models import *
  6. from tenancy.models import Tenant
  7. class RackTestCase(TestCase):
  8. def setUp(self):
  9. self.site1 = Site.objects.create(
  10. name='TestSite1',
  11. slug='test-site-1'
  12. )
  13. self.site2 = Site.objects.create(
  14. name='TestSite2',
  15. slug='test-site-2'
  16. )
  17. self.group1 = RackGroup.objects.create(
  18. name='TestGroup1',
  19. slug='test-group-1',
  20. site=self.site1
  21. )
  22. self.group2 = RackGroup.objects.create(
  23. name='TestGroup2',
  24. slug='test-group-2',
  25. site=self.site2
  26. )
  27. self.rack = Rack.objects.create(
  28. name='TestRack1',
  29. facility_id='A101',
  30. site=self.site1,
  31. group=self.group1,
  32. u_height=42
  33. )
  34. self.manufacturer = Manufacturer.objects.create(
  35. name='Acme',
  36. slug='acme'
  37. )
  38. self.device_type = {
  39. 'ff2048': DeviceType.objects.create(
  40. manufacturer=self.manufacturer,
  41. model='FrameForwarder 2048',
  42. slug='ff2048'
  43. ),
  44. 'cc5000': DeviceType.objects.create(
  45. manufacturer=self.manufacturer,
  46. model='CurrentCatapult 5000',
  47. slug='cc5000',
  48. u_height=0
  49. ),
  50. }
  51. self.role = {
  52. 'Server': DeviceRole.objects.create(
  53. name='Server',
  54. slug='server',
  55. ),
  56. 'Switch': DeviceRole.objects.create(
  57. name='Switch',
  58. slug='switch',
  59. ),
  60. 'Console Server': DeviceRole.objects.create(
  61. name='Console Server',
  62. slug='console-server',
  63. ),
  64. 'PDU': DeviceRole.objects.create(
  65. name='PDU',
  66. slug='pdu',
  67. ),
  68. }
  69. def test_rack_device_outside_height(self):
  70. rack1 = Rack(
  71. name='TestRack2',
  72. facility_id='A102',
  73. site=self.site1,
  74. u_height=42
  75. )
  76. rack1.save()
  77. device1 = Device(
  78. name='TestSwitch1',
  79. device_type=DeviceType.objects.get(manufacturer__slug='acme', slug='ff2048'),
  80. device_role=DeviceRole.objects.get(slug='switch'),
  81. site=self.site1,
  82. rack=rack1,
  83. position=43,
  84. face=DeviceFaceChoices.FACE_FRONT,
  85. )
  86. device1.save()
  87. with self.assertRaises(ValidationError):
  88. rack1.clean()
  89. def test_rack_group_site(self):
  90. rack_invalid_group = Rack(
  91. name='TestRack2',
  92. facility_id='A102',
  93. site=self.site1,
  94. u_height=42,
  95. group=self.group2
  96. )
  97. rack_invalid_group.save()
  98. with self.assertRaises(ValidationError):
  99. rack_invalid_group.clean()
  100. def test_mount_single_device(self):
  101. device1 = Device(
  102. name='TestSwitch1',
  103. device_type=DeviceType.objects.get(manufacturer__slug='acme', slug='ff2048'),
  104. device_role=DeviceRole.objects.get(slug='switch'),
  105. site=self.site1,
  106. rack=self.rack,
  107. position=10,
  108. face=DeviceFaceChoices.FACE_REAR,
  109. )
  110. device1.save()
  111. # Validate rack height
  112. self.assertEqual(list(self.rack.units), list(reversed(range(1, 43))))
  113. # Validate inventory (front face)
  114. rack1_inventory_front = self.rack.get_rack_units(face=DeviceFaceChoices.FACE_FRONT)
  115. self.assertEqual(rack1_inventory_front[-10]['device'], device1)
  116. del(rack1_inventory_front[-10])
  117. for u in rack1_inventory_front:
  118. self.assertIsNone(u['device'])
  119. # Validate inventory (rear face)
  120. rack1_inventory_rear = self.rack.get_rack_units(face=DeviceFaceChoices.FACE_REAR)
  121. self.assertEqual(rack1_inventory_rear[-10]['device'], device1)
  122. del(rack1_inventory_rear[-10])
  123. for u in rack1_inventory_rear:
  124. self.assertIsNone(u['device'])
  125. def test_mount_zero_ru(self):
  126. pdu = Device.objects.create(
  127. name='TestPDU',
  128. device_role=self.role.get('PDU'),
  129. device_type=self.device_type.get('cc5000'),
  130. site=self.site1,
  131. rack=self.rack,
  132. position=None,
  133. face='',
  134. )
  135. self.assertTrue(pdu)
  136. class DeviceTestCase(TestCase):
  137. def setUp(self):
  138. self.site = Site.objects.create(name='Test Site 1', slug='test-site-1')
  139. manufacturer = Manufacturer.objects.create(name='Test Manufacturer 1', slug='test-manufacturer-1')
  140. self.device_type = DeviceType.objects.create(
  141. manufacturer=manufacturer, model='Test Device Type 1', slug='test-device-type-1'
  142. )
  143. self.device_role = DeviceRole.objects.create(
  144. name='Test Device Role 1', slug='test-device-role-1', color='ff0000'
  145. )
  146. # Create DeviceType components
  147. ConsolePortTemplate(
  148. device_type=self.device_type,
  149. name='Console Port 1'
  150. ).save()
  151. ConsoleServerPortTemplate(
  152. device_type=self.device_type,
  153. name='Console Server Port 1'
  154. ).save()
  155. ppt = PowerPortTemplate(
  156. device_type=self.device_type,
  157. name='Power Port 1',
  158. maximum_draw=1000,
  159. allocated_draw=500
  160. )
  161. ppt.save()
  162. PowerOutletTemplate(
  163. device_type=self.device_type,
  164. name='Power Outlet 1',
  165. power_port=ppt,
  166. feed_leg=PowerOutletFeedLegChoices.FEED_LEG_A
  167. ).save()
  168. InterfaceTemplate(
  169. device_type=self.device_type,
  170. name='Interface 1',
  171. type=InterfaceTypeChoices.TYPE_1GE_FIXED,
  172. mgmt_only=True
  173. ).save()
  174. rpt = RearPortTemplate(
  175. device_type=self.device_type,
  176. name='Rear Port 1',
  177. type=PortTypeChoices.TYPE_8P8C,
  178. positions=8
  179. )
  180. rpt.save()
  181. FrontPortTemplate(
  182. device_type=self.device_type,
  183. name='Front Port 1',
  184. type=PortTypeChoices.TYPE_8P8C,
  185. rear_port=rpt,
  186. rear_port_position=2
  187. ).save()
  188. DeviceBayTemplate(
  189. device_type=self.device_type,
  190. name='Device Bay 1'
  191. ).save()
  192. def test_device_creation(self):
  193. """
  194. Ensure that all Device components are copied automatically from the DeviceType.
  195. """
  196. d = Device(
  197. site=self.site,
  198. device_type=self.device_type,
  199. device_role=self.device_role,
  200. name='Test Device 1'
  201. )
  202. d.save()
  203. ConsolePort.objects.get(
  204. device=d,
  205. name='Console Port 1'
  206. )
  207. ConsoleServerPort.objects.get(
  208. device=d,
  209. name='Console Server Port 1'
  210. )
  211. pp = PowerPort.objects.get(
  212. device=d,
  213. name='Power Port 1',
  214. maximum_draw=1000,
  215. allocated_draw=500
  216. )
  217. PowerOutlet.objects.get(
  218. device=d,
  219. name='Power Outlet 1',
  220. power_port=pp,
  221. feed_leg=PowerOutletFeedLegChoices.FEED_LEG_A
  222. )
  223. Interface.objects.get(
  224. device=d,
  225. name='Interface 1',
  226. type=InterfaceTypeChoices.TYPE_1GE_FIXED,
  227. mgmt_only=True
  228. )
  229. rp = RearPort.objects.get(
  230. device=d,
  231. name='Rear Port 1',
  232. type=PortTypeChoices.TYPE_8P8C,
  233. positions=8
  234. )
  235. FrontPort.objects.get(
  236. device=d,
  237. name='Front Port 1',
  238. type=PortTypeChoices.TYPE_8P8C,
  239. rear_port=rp,
  240. rear_port_position=2
  241. )
  242. DeviceBay.objects.get(
  243. device=d,
  244. name='Device Bay 1'
  245. )
  246. def test_multiple_unnamed_devices(self):
  247. device1 = Device(
  248. site=self.site,
  249. device_type=self.device_type,
  250. device_role=self.device_role,
  251. name=''
  252. )
  253. device1.save()
  254. device2 = Device(
  255. site=device1.site,
  256. device_type=device1.device_type,
  257. device_role=device1.device_role,
  258. name=''
  259. )
  260. device2.full_clean()
  261. device2.save()
  262. self.assertEqual(Device.objects.filter(name='').count(), 2)
  263. def test_device_duplicate_names(self):
  264. device1 = Device(
  265. site=self.site,
  266. device_type=self.device_type,
  267. device_role=self.device_role,
  268. name='Test Device 1'
  269. )
  270. device1.save()
  271. device2 = Device(
  272. site=device1.site,
  273. device_type=device1.device_type,
  274. device_role=device1.device_role,
  275. name=device1.name
  276. )
  277. # Two devices assigned to the same Site and no Tenant should fail validation
  278. with self.assertRaises(ValidationError):
  279. device2.full_clean()
  280. tenant = Tenant.objects.create(name='Test Tenant 1', slug='test-tenant-1')
  281. device1.tenant = tenant
  282. device1.save()
  283. device2.tenant = tenant
  284. # Two devices assigned to the same Site and the same Tenant should fail validation
  285. with self.assertRaises(ValidationError):
  286. device2.full_clean()
  287. device2.tenant = None
  288. # Two devices assigned to the same Site and different Tenants should pass validation
  289. device2.full_clean()
  290. device2.save()
  291. class CableTestCase(TestCase):
  292. def setUp(self):
  293. site = Site.objects.create(name='Test Site 1', slug='test-site-1')
  294. manufacturer = Manufacturer.objects.create(name='Test Manufacturer 1', slug='test-manufacturer-1')
  295. devicetype = DeviceType.objects.create(
  296. manufacturer=manufacturer, model='Test Device Type 1', slug='test-device-type-1'
  297. )
  298. devicerole = DeviceRole.objects.create(
  299. name='Test Device Role 1', slug='test-device-role-1', color='ff0000'
  300. )
  301. self.device1 = Device.objects.create(
  302. device_type=devicetype, device_role=devicerole, name='TestDevice1', site=site
  303. )
  304. self.device2 = Device.objects.create(
  305. device_type=devicetype, device_role=devicerole, name='TestDevice2', site=site
  306. )
  307. self.interface1 = Interface.objects.create(device=self.device1, name='eth0')
  308. self.interface2 = Interface.objects.create(device=self.device2, name='eth0')
  309. self.cable = Cable(termination_a=self.interface1, termination_b=self.interface2)
  310. self.cable.save()
  311. self.power_port1 = PowerPort.objects.create(device=self.device2, name='psu1')
  312. self.patch_pannel = Device.objects.create(
  313. device_type=devicetype, device_role=devicerole, name='TestPatchPannel', site=site
  314. )
  315. self.rear_port = RearPort.objects.create(device=self.patch_pannel, name='R1', type=1000)
  316. self.front_port = FrontPort.objects.create(
  317. device=self.patch_pannel, name='F1', type=1000, rear_port=self.rear_port
  318. )
  319. def test_cable_creation(self):
  320. """
  321. When a new Cable is created, it must be cached on either termination point.
  322. """
  323. interface1 = Interface.objects.get(pk=self.interface1.pk)
  324. self.assertEqual(self.cable.termination_a, interface1)
  325. interface2 = Interface.objects.get(pk=self.interface2.pk)
  326. self.assertEqual(self.cable.termination_b, interface2)
  327. def test_cable_deletion(self):
  328. """
  329. When a Cable is deleted, the `cable` field on its termination points must be nullified. The str() method
  330. should still return the PK of the string even after being nullified.
  331. """
  332. self.cable.delete()
  333. self.assertIsNone(self.cable.pk)
  334. self.assertNotEqual(str(self.cable), '#None')
  335. interface1 = Interface.objects.get(pk=self.interface1.pk)
  336. self.assertIsNone(interface1.cable)
  337. interface2 = Interface.objects.get(pk=self.interface2.pk)
  338. self.assertIsNone(interface2.cable)
  339. def test_cabletermination_deletion(self):
  340. """
  341. When a CableTermination object is deleted, its attached Cable (if any) must also be deleted.
  342. """
  343. self.interface1.delete()
  344. cable = Cable.objects.filter(pk=self.cable.pk).first()
  345. self.assertIsNone(cable)
  346. def test_cable_validates_compatibale_types(self):
  347. """
  348. The clean method should have a check to ensure only compatible port types can be connected by a cable
  349. """
  350. # An interface cannot be connected to a power port
  351. cable = Cable(termination_a=self.interface1, termination_b=self.power_port1)
  352. with self.assertRaises(ValidationError):
  353. cable.clean()
  354. def test_cable_cannot_have_the_same_terminination_on_both_ends(self):
  355. """
  356. A cable cannot be made with the same A and B side terminations
  357. """
  358. cable = Cable(termination_a=self.interface1, termination_b=self.interface1)
  359. with self.assertRaises(ValidationError):
  360. cable.clean()
  361. def test_cable_front_port_cannot_connect_to_corresponding_rear_port(self):
  362. """
  363. A cable cannot connect a front port to its corresponding rear port
  364. """
  365. cable = Cable(termination_a=self.front_port, termination_b=self.rear_port)
  366. with self.assertRaises(ValidationError):
  367. cable.clean()
  368. def test_cable_cannot_terminate_to_an_existing_connection(self):
  369. """
  370. Either side of a cable cannot be terminated when that side already has a connection
  371. """
  372. # Try to create a cable with the same interface terminations
  373. cable = Cable(termination_a=self.interface2, termination_b=self.interface1)
  374. with self.assertRaises(ValidationError):
  375. cable.clean()
  376. def test_cable_cannot_terminate_to_a_virtual_inteface(self):
  377. """
  378. A cable cannot terminate to a virtual interface
  379. """
  380. virtual_interface = Interface(device=self.device1, name="V1", type=InterfaceTypeChoices.TYPE_VIRTUAL)
  381. cable = Cable(termination_a=self.interface2, termination_b=virtual_interface)
  382. with self.assertRaises(ValidationError):
  383. cable.clean()
  384. def test_cable_cannot_terminate_to_a_wireless_inteface(self):
  385. """
  386. A cable cannot terminate to a wireless interface
  387. """
  388. wireless_interface = Interface(device=self.device1, name="W1", type=InterfaceTypeChoices.TYPE_80211A)
  389. cable = Cable(termination_a=self.interface2, termination_b=wireless_interface)
  390. with self.assertRaises(ValidationError):
  391. cable.clean()
  392. class CablePathTestCase(TestCase):
  393. @classmethod
  394. def setUpTestData(cls):
  395. site = Site.objects.create(name='Site 1', slug='site-1')
  396. manufacturer = Manufacturer.objects.create(name='Manufacturer 1', slug='manufacturer-1')
  397. devicetype = DeviceType.objects.create(
  398. manufacturer=manufacturer, model='Device Type 1', slug='device-type-1'
  399. )
  400. devicerole = DeviceRole.objects.create(
  401. name='Device Role 1', slug='device-role-1', color='ff0000'
  402. )
  403. provider = Provider.objects.create(name='Provider 1', slug='provider-1')
  404. circuittype = CircuitType.objects.create(name='Circuit Type 1', slug='circuit-type-1')
  405. circuit = Circuit.objects.create(provider=provider, type=circuittype, cid='1')
  406. CircuitTermination.objects.bulk_create((
  407. CircuitTermination(circuit=circuit, site=site, term_side='A', port_speed=1000),
  408. CircuitTermination(circuit=circuit, site=site, term_side='Z', port_speed=1000),
  409. ))
  410. # Create four network devices with four interfaces each
  411. devices = (
  412. Device(device_type=devicetype, device_role=devicerole, name='Device 1', site=site),
  413. Device(device_type=devicetype, device_role=devicerole, name='Device 2', site=site),
  414. Device(device_type=devicetype, device_role=devicerole, name='Device 3', site=site),
  415. Device(device_type=devicetype, device_role=devicerole, name='Device 4', site=site),
  416. )
  417. Device.objects.bulk_create(devices)
  418. for device in devices:
  419. Interface.objects.bulk_create((
  420. Interface(device=device, name='Interface 1', type=InterfaceTypeChoices.TYPE_1GE_FIXED),
  421. Interface(device=device, name='Interface 2', type=InterfaceTypeChoices.TYPE_1GE_FIXED),
  422. Interface(device=device, name='Interface 3', type=InterfaceTypeChoices.TYPE_1GE_FIXED),
  423. Interface(device=device, name='Interface 4', type=InterfaceTypeChoices.TYPE_1GE_FIXED),
  424. ))
  425. # Create four patch panels, each with one rear port and four front ports
  426. patch_panels = (
  427. Device(device_type=devicetype, device_role=devicerole, name='Panel 1', site=site),
  428. Device(device_type=devicetype, device_role=devicerole, name='Panel 2', site=site),
  429. Device(device_type=devicetype, device_role=devicerole, name='Panel 3', site=site),
  430. Device(device_type=devicetype, device_role=devicerole, name='Panel 4', site=site),
  431. Device(device_type=devicetype, device_role=devicerole, name='Panel 5', site=site),
  432. )
  433. Device.objects.bulk_create(patch_panels)
  434. # Create patch panels with 4 positions
  435. for patch_panel in patch_panels[:4]:
  436. rearport = RearPort.objects.create(device=patch_panel, name='Rear Port 1', positions=4, type=PortTypeChoices.TYPE_8P8C)
  437. FrontPort.objects.bulk_create((
  438. FrontPort(device=patch_panel, name='Front Port 1', rear_port=rearport, rear_port_position=1, type=PortTypeChoices.TYPE_8P8C),
  439. FrontPort(device=patch_panel, name='Front Port 2', rear_port=rearport, rear_port_position=2, type=PortTypeChoices.TYPE_8P8C),
  440. FrontPort(device=patch_panel, name='Front Port 3', rear_port=rearport, rear_port_position=3, type=PortTypeChoices.TYPE_8P8C),
  441. FrontPort(device=patch_panel, name='Front Port 4', rear_port=rearport, rear_port_position=4, type=PortTypeChoices.TYPE_8P8C),
  442. ))
  443. # Create a 1-on-1 patch panel
  444. for patch_panel in patch_panels[4:]:
  445. rearport = RearPort.objects.create(device=patch_panel, name='Rear Port 1', positions=1, type=PortTypeChoices.TYPE_8P8C)
  446. FrontPort.objects.create(device=patch_panel, name='Front Port 1', rear_port=rearport, rear_port_position=1, type=PortTypeChoices.TYPE_8P8C)
  447. def test_direct_connection(self):
  448. """
  449. Test a direct connection between two interfaces.
  450. [Device 1] ----- [Device 2]
  451. Iface1 Iface1
  452. """
  453. # Create cable
  454. cable = Cable(
  455. termination_a=Interface.objects.get(device__name='Device 1', name='Interface 1'),
  456. termination_b=Interface.objects.get(device__name='Device 2', name='Interface 1')
  457. )
  458. cable.save()
  459. # Retrieve endpoints
  460. endpoint_a = Interface.objects.get(device__name='Device 1', name='Interface 1')
  461. endpoint_b = Interface.objects.get(device__name='Device 2', name='Interface 1')
  462. # Validate connections
  463. self.assertEqual(endpoint_a.connected_endpoint, endpoint_b)
  464. self.assertEqual(endpoint_b.connected_endpoint, endpoint_a)
  465. self.assertTrue(endpoint_a.connection_status)
  466. self.assertTrue(endpoint_b.connection_status)
  467. # Delete cable
  468. cable.delete()
  469. # Refresh endpoints
  470. endpoint_a.refresh_from_db()
  471. endpoint_b.refresh_from_db()
  472. # Check that connections have been nullified
  473. self.assertIsNone(endpoint_a.connected_endpoint)
  474. self.assertIsNone(endpoint_b.connected_endpoint)
  475. self.assertIsNone(endpoint_a.connection_status)
  476. self.assertIsNone(endpoint_b.connection_status)
  477. def test_connection_via_single_rear_port(self):
  478. """
  479. Test a connection which passes through a single front/rear port pair.
  480. 1 2
  481. [Device 1] ----- [Panel 5] ----- [Device 2]
  482. Iface1 FP1 RP1 Iface1
  483. """
  484. # Create cables (FP first, RP second)
  485. cable1 = Cable(
  486. termination_a=Interface.objects.get(device__name='Device 1', name='Interface 1'),
  487. termination_b=FrontPort.objects.get(device__name='Panel 5', name='Front Port 1')
  488. )
  489. cable1.save()
  490. cable2 = Cable(
  491. termination_b=RearPort.objects.get(device__name='Panel 5', name='Rear Port 1'),
  492. termination_a=Interface.objects.get(device__name='Device 2', name='Interface 1')
  493. )
  494. cable2.save()
  495. # Retrieve endpoints
  496. endpoint_a = Interface.objects.get(device__name='Device 1', name='Interface 1')
  497. endpoint_b = Interface.objects.get(device__name='Device 2', name='Interface 1')
  498. # Validate connections
  499. self.assertEqual(endpoint_a.connected_endpoint, endpoint_b)
  500. self.assertEqual(endpoint_b.connected_endpoint, endpoint_a)
  501. self.assertTrue(endpoint_a.connection_status)
  502. self.assertTrue(endpoint_b.connection_status)
  503. # Delete cable 1
  504. cable1.delete()
  505. # Refresh endpoints
  506. endpoint_a.refresh_from_db()
  507. endpoint_b.refresh_from_db()
  508. # Check that connections have been nullified
  509. self.assertIsNone(endpoint_a.connected_endpoint)
  510. self.assertIsNone(endpoint_b.connected_endpoint)
  511. self.assertIsNone(endpoint_a.connection_status)
  512. self.assertIsNone(endpoint_b.connection_status)
  513. # Recreate cable 1 to test creating the cables in reverse order (RP first, FP second)
  514. cable1.save()
  515. # Refresh endpoints
  516. endpoint_a.refresh_from_db()
  517. endpoint_b.refresh_from_db()
  518. # Validate connections
  519. self.assertEqual(endpoint_a.connected_endpoint, endpoint_b)
  520. self.assertEqual(endpoint_b.connected_endpoint, endpoint_a)
  521. self.assertTrue(endpoint_a.connection_status)
  522. self.assertTrue(endpoint_b.connection_status)
  523. # Delete cable 2
  524. cable2.delete()
  525. # Refresh endpoints
  526. endpoint_a.refresh_from_db()
  527. endpoint_b.refresh_from_db()
  528. # Check that connections have been nullified
  529. self.assertIsNone(endpoint_a.connected_endpoint)
  530. self.assertIsNone(endpoint_b.connected_endpoint)
  531. self.assertIsNone(endpoint_a.connection_status)
  532. self.assertIsNone(endpoint_b.connection_status)
  533. def test_connection_via_nested_single_rear_port(self):
  534. """
  535. Test a connection which passes through a single front/rear port pair between two multi-port MUXes.
  536. Test two connections via patched rear ports:
  537. Device 1 <---> Device 2
  538. Device 3 <---> Device 4
  539. 1 2
  540. [Device 1] -----------+ +----------- [Device 2]
  541. Iface1 | | Iface1
  542. FP1 | 3 4 | FP1
  543. [Panel 1] ----- [Panel 5] ----- [Panel 2]
  544. FP2 | RP1 RP1 FP1 RP1 | FP2
  545. Iface1 | | Iface1
  546. [Device 3] -----------+ +----------- [Device 4]
  547. 5 6
  548. """
  549. # Create cables (Panel 5 RP first, FP second)
  550. cable1 = Cable(
  551. termination_a=Interface.objects.get(device__name='Device 1', name='Interface 1'),
  552. termination_b=FrontPort.objects.get(device__name='Panel 1', name='Front Port 1')
  553. )
  554. cable1.save()
  555. cable2 = Cable(
  556. termination_b=FrontPort.objects.get(device__name='Panel 2', name='Front Port 1'),
  557. termination_a=Interface.objects.get(device__name='Device 2', name='Interface 1')
  558. )
  559. cable2.save()
  560. cable3 = Cable(
  561. termination_b=RearPort.objects.get(device__name='Panel 1', name='Rear Port 1'),
  562. termination_a=RearPort.objects.get(device__name='Panel 5', name='Rear Port 1')
  563. )
  564. cable3.save()
  565. cable4 = Cable(
  566. termination_b=FrontPort.objects.get(device__name='Panel 5', name='Front Port 1'),
  567. termination_a=RearPort.objects.get(device__name='Panel 2', name='Rear Port 1')
  568. )
  569. cable4.save()
  570. cable5 = Cable(
  571. termination_b=FrontPort.objects.get(device__name='Panel 1', name='Front Port 2'),
  572. termination_a=Interface.objects.get(device__name='Device 3', name='Interface 1')
  573. )
  574. cable5.save()
  575. cable6 = Cable(
  576. termination_b=FrontPort.objects.get(device__name='Panel 2', name='Front Port 2'),
  577. termination_a=Interface.objects.get(device__name='Device 4', name='Interface 1')
  578. )
  579. cable6.save()
  580. # Retrieve endpoints
  581. endpoint_a = Interface.objects.get(device__name='Device 1', name='Interface 1')
  582. endpoint_b = Interface.objects.get(device__name='Device 2', name='Interface 1')
  583. endpoint_c = Interface.objects.get(device__name='Device 3', name='Interface 1')
  584. endpoint_d = Interface.objects.get(device__name='Device 4', name='Interface 1')
  585. # Validate connections
  586. self.assertEqual(endpoint_a.connected_endpoint, endpoint_b)
  587. self.assertEqual(endpoint_b.connected_endpoint, endpoint_a)
  588. self.assertEqual(endpoint_c.connected_endpoint, endpoint_d)
  589. self.assertEqual(endpoint_d.connected_endpoint, endpoint_c)
  590. self.assertTrue(endpoint_a.connection_status)
  591. self.assertTrue(endpoint_b.connection_status)
  592. self.assertTrue(endpoint_c.connection_status)
  593. self.assertTrue(endpoint_d.connection_status)
  594. # Delete cable 3
  595. cable3.delete()
  596. # Refresh endpoints
  597. endpoint_a.refresh_from_db()
  598. endpoint_b.refresh_from_db()
  599. endpoint_c.refresh_from_db()
  600. endpoint_d.refresh_from_db()
  601. # Check that connections have been nullified
  602. self.assertIsNone(endpoint_a.connected_endpoint)
  603. self.assertIsNone(endpoint_b.connected_endpoint)
  604. self.assertIsNone(endpoint_c.connected_endpoint)
  605. self.assertIsNone(endpoint_d.connected_endpoint)
  606. self.assertIsNone(endpoint_a.connection_status)
  607. self.assertIsNone(endpoint_b.connection_status)
  608. self.assertIsNone(endpoint_c.connection_status)
  609. self.assertIsNone(endpoint_d.connection_status)
  610. # Recreate cable 3 to test reverse order (Panel 5 FP first, RP second)
  611. cable3.save()
  612. # Refresh endpoints
  613. endpoint_a.refresh_from_db()
  614. endpoint_b.refresh_from_db()
  615. endpoint_c.refresh_from_db()
  616. endpoint_d.refresh_from_db()
  617. # Validate connections
  618. self.assertEqual(endpoint_a.connected_endpoint, endpoint_b)
  619. self.assertEqual(endpoint_b.connected_endpoint, endpoint_a)
  620. self.assertEqual(endpoint_c.connected_endpoint, endpoint_d)
  621. self.assertEqual(endpoint_d.connected_endpoint, endpoint_c)
  622. self.assertTrue(endpoint_a.connection_status)
  623. self.assertTrue(endpoint_b.connection_status)
  624. self.assertTrue(endpoint_c.connection_status)
  625. self.assertTrue(endpoint_d.connection_status)
  626. # Delete cable 4
  627. cable4.delete()
  628. # Refresh endpoints
  629. endpoint_a.refresh_from_db()
  630. endpoint_b.refresh_from_db()
  631. endpoint_c.refresh_from_db()
  632. endpoint_d.refresh_from_db()
  633. # Check that connections have been nullified
  634. self.assertIsNone(endpoint_a.connected_endpoint)
  635. self.assertIsNone(endpoint_b.connected_endpoint)
  636. self.assertIsNone(endpoint_c.connected_endpoint)
  637. self.assertIsNone(endpoint_d.connected_endpoint)
  638. self.assertIsNone(endpoint_a.connection_status)
  639. self.assertIsNone(endpoint_b.connection_status)
  640. self.assertIsNone(endpoint_c.connection_status)
  641. self.assertIsNone(endpoint_d.connection_status)
  642. def test_connections_via_patch(self):
  643. """
  644. Test two connections via patched rear ports:
  645. Device 1 <---> Device 2
  646. Device 3 <---> Device 4
  647. 1 2
  648. [Device 1] -----------+ +----------- [Device 2]
  649. Iface1 | | Iface1
  650. FP1 | 3 | FP1
  651. [Panel 1] ----- [Panel 2]
  652. FP2 | RP1 RP1 | FP2
  653. Iface1 | | Iface1
  654. [Device 3] -----------+ +----------- [Device 4]
  655. 4 5
  656. """
  657. # Create cables
  658. cable1 = Cable(
  659. termination_a=Interface.objects.get(device__name='Device 1', name='Interface 1'),
  660. termination_b=FrontPort.objects.get(device__name='Panel 1', name='Front Port 1')
  661. )
  662. cable1.save()
  663. cable2 = Cable(
  664. termination_a=Interface.objects.get(device__name='Device 2', name='Interface 1'),
  665. termination_b=FrontPort.objects.get(device__name='Panel 2', name='Front Port 1')
  666. )
  667. cable2.save()
  668. cable3 = Cable(
  669. termination_a=RearPort.objects.get(device__name='Panel 1', name='Rear Port 1'),
  670. termination_b=RearPort.objects.get(device__name='Panel 2', name='Rear Port 1')
  671. )
  672. cable3.save()
  673. cable4 = Cable(
  674. termination_a=Interface.objects.get(device__name='Device 3', name='Interface 1'),
  675. termination_b=FrontPort.objects.get(device__name='Panel 1', name='Front Port 2')
  676. )
  677. cable4.save()
  678. cable5 = Cable(
  679. termination_a=Interface.objects.get(device__name='Device 4', name='Interface 1'),
  680. termination_b=FrontPort.objects.get(device__name='Panel 2', name='Front Port 2')
  681. )
  682. cable5.save()
  683. # Retrieve endpoints
  684. endpoint_a = Interface.objects.get(device__name='Device 1', name='Interface 1')
  685. endpoint_b = Interface.objects.get(device__name='Device 2', name='Interface 1')
  686. endpoint_c = Interface.objects.get(device__name='Device 3', name='Interface 1')
  687. endpoint_d = Interface.objects.get(device__name='Device 4', name='Interface 1')
  688. # Validate connections
  689. self.assertEqual(endpoint_a.connected_endpoint, endpoint_b)
  690. self.assertEqual(endpoint_b.connected_endpoint, endpoint_a)
  691. self.assertEqual(endpoint_c.connected_endpoint, endpoint_d)
  692. self.assertEqual(endpoint_d.connected_endpoint, endpoint_c)
  693. self.assertTrue(endpoint_a.connection_status)
  694. self.assertTrue(endpoint_b.connection_status)
  695. self.assertTrue(endpoint_c.connection_status)
  696. self.assertTrue(endpoint_d.connection_status)
  697. # Delete cable 3
  698. cable3.delete()
  699. # Refresh endpoints
  700. endpoint_a.refresh_from_db()
  701. endpoint_b.refresh_from_db()
  702. endpoint_c.refresh_from_db()
  703. endpoint_d.refresh_from_db()
  704. # Check that connections have been nullified
  705. self.assertIsNone(endpoint_a.connected_endpoint)
  706. self.assertIsNone(endpoint_b.connected_endpoint)
  707. self.assertIsNone(endpoint_c.connected_endpoint)
  708. self.assertIsNone(endpoint_d.connected_endpoint)
  709. self.assertIsNone(endpoint_a.connection_status)
  710. self.assertIsNone(endpoint_b.connection_status)
  711. self.assertIsNone(endpoint_c.connection_status)
  712. self.assertIsNone(endpoint_d.connection_status)
  713. def test_connections_via_multiple_patches(self):
  714. """
  715. Test two connections via patched rear ports:
  716. Device 1 <---> Device 2
  717. Device 3 <---> Device 4
  718. 1 2 3
  719. [Device 1] -----------+ +---------------+ +----------- [Device 2]
  720. Iface1 | | | | Iface1
  721. FP1 | 4 | FP1 FP1 | 5 | FP1
  722. [Panel 1] ----- [Panel 2] [Panel 3] ----- [Panel 4]
  723. FP2 | RP1 RP1 | FP2 FP2 | RP1 RP1 | FP2
  724. Iface1 | | | | Iface1
  725. [Device 3] -----------+ +---------------+ +----------- [Device 4]
  726. 6 7 8
  727. """
  728. # Create cables
  729. cable1 = Cable(
  730. termination_a=Interface.objects.get(device__name='Device 1', name='Interface 1'),
  731. termination_b=FrontPort.objects.get(device__name='Panel 1', name='Front Port 1')
  732. )
  733. cable1.save()
  734. cable2 = Cable(
  735. termination_a=FrontPort.objects.get(device__name='Panel 2', name='Front Port 1'),
  736. termination_b=FrontPort.objects.get(device__name='Panel 3', name='Front Port 1')
  737. )
  738. cable2.save()
  739. cable3 = Cable(
  740. termination_a=FrontPort.objects.get(device__name='Panel 4', name='Front Port 1'),
  741. termination_b=Interface.objects.get(device__name='Device 2', name='Interface 1')
  742. )
  743. cable3.save()
  744. cable4 = Cable(
  745. termination_a=RearPort.objects.get(device__name='Panel 1', name='Rear Port 1'),
  746. termination_b=RearPort.objects.get(device__name='Panel 2', name='Rear Port 1')
  747. )
  748. cable4.save()
  749. cable5 = Cable(
  750. termination_a=RearPort.objects.get(device__name='Panel 3', name='Rear Port 1'),
  751. termination_b=RearPort.objects.get(device__name='Panel 4', name='Rear Port 1')
  752. )
  753. cable5.save()
  754. cable6 = Cable(
  755. termination_a=Interface.objects.get(device__name='Device 3', name='Interface 1'),
  756. termination_b=FrontPort.objects.get(device__name='Panel 1', name='Front Port 2')
  757. )
  758. cable6.save()
  759. cable7 = Cable(
  760. termination_a=FrontPort.objects.get(device__name='Panel 2', name='Front Port 2'),
  761. termination_b=FrontPort.objects.get(device__name='Panel 3', name='Front Port 2')
  762. )
  763. cable7.save()
  764. cable8 = Cable(
  765. termination_a=FrontPort.objects.get(device__name='Panel 4', name='Front Port 2'),
  766. termination_b=Interface.objects.get(device__name='Device 4', name='Interface 1')
  767. )
  768. cable8.save()
  769. # Retrieve endpoints
  770. endpoint_a = Interface.objects.get(device__name='Device 1', name='Interface 1')
  771. endpoint_b = Interface.objects.get(device__name='Device 2', name='Interface 1')
  772. endpoint_c = Interface.objects.get(device__name='Device 3', name='Interface 1')
  773. endpoint_d = Interface.objects.get(device__name='Device 4', name='Interface 1')
  774. # Validate connections
  775. self.assertEqual(endpoint_a.connected_endpoint, endpoint_b)
  776. self.assertEqual(endpoint_b.connected_endpoint, endpoint_a)
  777. self.assertEqual(endpoint_c.connected_endpoint, endpoint_d)
  778. self.assertEqual(endpoint_d.connected_endpoint, endpoint_c)
  779. self.assertTrue(endpoint_a.connection_status)
  780. self.assertTrue(endpoint_b.connection_status)
  781. self.assertTrue(endpoint_c.connection_status)
  782. self.assertTrue(endpoint_d.connection_status)
  783. # Delete cables 4 and 5
  784. cable4.delete()
  785. cable5.delete()
  786. # Refresh endpoints
  787. endpoint_a.refresh_from_db()
  788. endpoint_b.refresh_from_db()
  789. endpoint_c.refresh_from_db()
  790. endpoint_d.refresh_from_db()
  791. # Check that connections have been nullified
  792. self.assertIsNone(endpoint_a.connected_endpoint)
  793. self.assertIsNone(endpoint_b.connected_endpoint)
  794. self.assertIsNone(endpoint_c.connected_endpoint)
  795. self.assertIsNone(endpoint_d.connected_endpoint)
  796. self.assertIsNone(endpoint_a.connection_status)
  797. self.assertIsNone(endpoint_b.connection_status)
  798. self.assertIsNone(endpoint_c.connection_status)
  799. self.assertIsNone(endpoint_d.connection_status)
  800. def test_connections_via_nested_rear_ports(self):
  801. """
  802. Test two connections via nested rear ports:
  803. Device 1 <---> Device 2
  804. Device 3 <---> Device 4
  805. 1 2
  806. [Device 1] -----------+ +----------- [Device 2]
  807. Iface1 | | Iface1
  808. FP1 | 3 4 5 | FP1
  809. [Panel 1] ----- [Panel 2] ----- [Panel 3] ----- [Panel 4]
  810. FP2 | RP1 FP1 RP1 RP1 FP1 RP1 | FP2
  811. Iface1 | | Iface1
  812. [Device 3] -----------+ +----------- [Device 4]
  813. 6 7
  814. """
  815. # Create cables
  816. cable1 = Cable(
  817. termination_a=Interface.objects.get(device__name='Device 1', name='Interface 1'),
  818. termination_b=FrontPort.objects.get(device__name='Panel 1', name='Front Port 1')
  819. )
  820. cable1.save()
  821. cable2 = Cable(
  822. termination_a=FrontPort.objects.get(device__name='Panel 4', name='Front Port 1'),
  823. termination_b=Interface.objects.get(device__name='Device 2', name='Interface 1')
  824. )
  825. cable2.save()
  826. cable3 = Cable(
  827. termination_a=RearPort.objects.get(device__name='Panel 1', name='Rear Port 1'),
  828. termination_b=FrontPort.objects.get(device__name='Panel 2', name='Front Port 1')
  829. )
  830. cable3.save()
  831. cable4 = Cable(
  832. termination_a=RearPort.objects.get(device__name='Panel 2', name='Rear Port 1'),
  833. termination_b=RearPort.objects.get(device__name='Panel 3', name='Rear Port 1')
  834. )
  835. cable4.save()
  836. cable5 = Cable(
  837. termination_a=FrontPort.objects.get(device__name='Panel 3', name='Front Port 1'),
  838. termination_b=RearPort.objects.get(device__name='Panel 4', name='Rear Port 1')
  839. )
  840. cable5.save()
  841. cable6 = Cable(
  842. termination_a=Interface.objects.get(device__name='Device 3', name='Interface 1'),
  843. termination_b=FrontPort.objects.get(device__name='Panel 1', name='Front Port 2')
  844. )
  845. cable6.save()
  846. cable7 = Cable(
  847. termination_a=FrontPort.objects.get(device__name='Panel 4', name='Front Port 2'),
  848. termination_b=Interface.objects.get(device__name='Device 4', name='Interface 1')
  849. )
  850. cable7.save()
  851. # Retrieve endpoints
  852. endpoint_a = Interface.objects.get(device__name='Device 1', name='Interface 1')
  853. endpoint_b = Interface.objects.get(device__name='Device 2', name='Interface 1')
  854. endpoint_c = Interface.objects.get(device__name='Device 3', name='Interface 1')
  855. endpoint_d = Interface.objects.get(device__name='Device 4', name='Interface 1')
  856. # Validate connections
  857. self.assertEqual(endpoint_a.connected_endpoint, endpoint_b)
  858. self.assertEqual(endpoint_b.connected_endpoint, endpoint_a)
  859. self.assertEqual(endpoint_c.connected_endpoint, endpoint_d)
  860. self.assertEqual(endpoint_d.connected_endpoint, endpoint_c)
  861. self.assertTrue(endpoint_a.connection_status)
  862. self.assertTrue(endpoint_b.connection_status)
  863. self.assertTrue(endpoint_c.connection_status)
  864. self.assertTrue(endpoint_d.connection_status)
  865. # Delete cable 4
  866. cable4.delete()
  867. # Refresh endpoints
  868. endpoint_a.refresh_from_db()
  869. endpoint_b.refresh_from_db()
  870. endpoint_c.refresh_from_db()
  871. endpoint_d.refresh_from_db()
  872. # Check that connections have been nullified
  873. self.assertIsNone(endpoint_a.connected_endpoint)
  874. self.assertIsNone(endpoint_b.connected_endpoint)
  875. self.assertIsNone(endpoint_c.connected_endpoint)
  876. self.assertIsNone(endpoint_d.connected_endpoint)
  877. self.assertIsNone(endpoint_a.connection_status)
  878. self.assertIsNone(endpoint_b.connection_status)
  879. self.assertIsNone(endpoint_c.connection_status)
  880. self.assertIsNone(endpoint_d.connection_status)
  881. def test_connection_via_circuit(self):
  882. """
  883. 1 2
  884. [Device 1] ----- [Circuit] ----- [Device 2]
  885. Iface1 A Z Iface1
  886. """
  887. # Create cables
  888. cable1 = Cable(
  889. termination_a=Interface.objects.get(device__name='Device 1', name='Interface 1'),
  890. termination_b=CircuitTermination.objects.get(term_side='A')
  891. )
  892. cable1.save()
  893. cable2 = Cable(
  894. termination_a=CircuitTermination.objects.get(term_side='Z'),
  895. termination_b=Interface.objects.get(device__name='Device 2', name='Interface 1')
  896. )
  897. cable2.save()
  898. # Retrieve endpoints
  899. endpoint_a = Interface.objects.get(device__name='Device 1', name='Interface 1')
  900. endpoint_b = Interface.objects.get(device__name='Device 2', name='Interface 1')
  901. # Validate connections
  902. self.assertEqual(endpoint_a.connected_endpoint, endpoint_b)
  903. self.assertEqual(endpoint_b.connected_endpoint, endpoint_a)
  904. self.assertTrue(endpoint_a.connection_status)
  905. self.assertTrue(endpoint_b.connection_status)
  906. # Delete circuit
  907. circuit = Circuit.objects.first().delete()
  908. # Refresh endpoints
  909. endpoint_a.refresh_from_db()
  910. endpoint_b.refresh_from_db()
  911. # Check that connections have been nullified
  912. self.assertIsNone(endpoint_a.connected_endpoint)
  913. self.assertIsNone(endpoint_b.connected_endpoint)
  914. self.assertIsNone(endpoint_a.connection_status)
  915. self.assertIsNone(endpoint_b.connection_status)
  916. def test_connection_via_patched_circuit(self):
  917. """
  918. 1 2 3 4
  919. [Device 1] ----- [Panel 1] ----- [Circuit] ----- [Panel 2] ----- [Device 2]
  920. Iface1 FP1 RP1 A Z RP1 FP1 Iface1
  921. """
  922. # Create cables
  923. cable1 = Cable(
  924. termination_a=Interface.objects.get(device__name='Device 1', name='Interface 1'),
  925. termination_b=FrontPort.objects.get(device__name='Panel 1', name='Front Port 1')
  926. )
  927. cable1.save()
  928. cable2 = Cable(
  929. termination_a=RearPort.objects.get(device__name='Panel 1', name='Rear Port 1'),
  930. termination_b=CircuitTermination.objects.get(term_side='A')
  931. )
  932. cable2.save()
  933. cable3 = Cable(
  934. termination_a=CircuitTermination.objects.get(term_side='Z'),
  935. termination_b=RearPort.objects.get(device__name='Panel 2', name='Rear Port 1')
  936. )
  937. cable3.save()
  938. cable4 = Cable(
  939. termination_a=FrontPort.objects.get(device__name='Panel 2', name='Front Port 1'),
  940. termination_b=Interface.objects.get(device__name='Device 2', name='Interface 1')
  941. )
  942. cable4.save()
  943. # Retrieve endpoints
  944. endpoint_a = Interface.objects.get(device__name='Device 1', name='Interface 1')
  945. endpoint_b = Interface.objects.get(device__name='Device 2', name='Interface 1')
  946. # Validate connections
  947. self.assertEqual(endpoint_a.connected_endpoint, endpoint_b)
  948. self.assertEqual(endpoint_b.connected_endpoint, endpoint_a)
  949. self.assertTrue(endpoint_a.connection_status)
  950. self.assertTrue(endpoint_b.connection_status)
  951. # Delete circuit
  952. circuit = Circuit.objects.first().delete()
  953. # Refresh endpoints
  954. endpoint_a.refresh_from_db()
  955. endpoint_b.refresh_from_db()
  956. # Check that connections have been nullified
  957. self.assertIsNone(endpoint_a.connected_endpoint)
  958. self.assertIsNone(endpoint_b.connected_endpoint)
  959. self.assertIsNone(endpoint_a.connection_status)
  960. self.assertIsNone(endpoint_b.connection_status)