727 lines
36 KiB
Python
727 lines
36 KiB
Python
|
# -*- coding: utf-8 -*-
|
||
|
|
|
||
|
|
from __future__ import absolute_import
|
||
|
|
|
||
|
|
from norduniclient.testing import Neo4jTestCase
|
||
|
|
from norduniclient import core
|
||
|
|
from norduniclient import exceptions
|
||
|
|
from norduniclient import models
|
||
|
|
|
||
|
|
__author__ = 'lundberg'
|
||
|
|
|
||
|
|
|
||
|
|
class ModelsTests(Neo4jTestCase):
|
||
|
|
|
||
|
|
def setUp(self):
|
||
|
|
super(ModelsTests, self).setUp()
|
||
|
|
q1 = """
|
||
|
|
// Create nodes
|
||
|
|
CREATE (router1:Node:Physical:Router{name:'Router1', handle_id:'1'}),
|
||
|
|
(port1:Node:Physical:Port{name:'Port1', handle_id:'2'}),
|
||
|
|
(unit1:Node:Logical:Unit{name:'Unit1', handle_id:'3'}),
|
||
|
|
(port6:Node:Physical:Port{name:'Port6', handle_id:'4'}),
|
||
|
|
(unit2:Node:Logical:Unit{name:'Unit2', handle_id:'5'}),
|
||
|
|
(provider1:Node:Relation:Provider{name:'Provider1', handle_id:'6'}),
|
||
|
|
(peering_group1:Node:Logical:Peering_Group{name:'Peering Group1', handle_id:'7'}),
|
||
|
|
(peering_partner1:Node:Relation:Peering_Partner{name:'Peering Partner1', handle_id:'8'}),
|
||
|
|
(service2:Node:Logical:Service{name:'Service2', handle_id:'9'}),
|
||
|
|
(service3:Node:Logical:Service{name:'Service3', handle_id:'10'}),
|
||
|
|
(site1:Node:Location:Site{name:'Site1', handle_id:'11'}),
|
||
|
|
(rack1:Node:Location:Rack{name:'Rack1', handle_id:'12'}),
|
||
|
|
(optical_node1:Node:Physical:Optical_Node{name:'Optical Node1', handle_id:'13'}),
|
||
|
|
(port2:Node:Physical:Port{name:'Port2', handle_id:'14', description:'This is a port'}),
|
||
|
|
(rack2:Node:Location:Rack{name:'Rack2', handle_id:'15'}),
|
||
|
|
(optical_node2:Node:Physical:Optical_Node{name:'Optical Node2', handle_id:'16'}),
|
||
|
|
(port3:Node:Physical:Port{name:'Port3', handle_id:'17'}),
|
||
|
|
(site2:Node:Location:Site{name:'Site2', handle_id:'18'}),
|
||
|
|
(rack3:Node:Location:Rack{name:'Rack3', handle_id:'19'}),
|
||
|
|
(optical_path1:Node:Logical:Optical_Path{name:'Optical Path1', handle_id:'20'}),
|
||
|
|
(optical_link1:Node:Logical:Optical_Link{name:'Optical Link1', handle_id:'21'}),
|
||
|
|
(optical_link2:Node:Logical:Optical_Link{name:'Optical Link2', handle_id:'22'}),
|
||
|
|
(odf1:Node:Physical:ODF{name:'ODF1', handle_id:'23'}),
|
||
|
|
(port4:Node:Physical:Port{name:'Port4', handle_id:'24'}),
|
||
|
|
(odf2:Node:Physical:ODF{name:'ODF2', handle_id:'25'}),
|
||
|
|
(port5:Node:Physical:Port{name:'Port5', handle_id:'26'}),
|
||
|
|
(port7:Node:Physical:Port{name:'Port7', handle_id:'27'}),
|
||
|
|
(cable1:Node:Physical:Cable{name:'Cable1', handle_id:'28'}),
|
||
|
|
(cable2:Node:Physical:Cable{name:'Cable2', handle_id:'29'}),
|
||
|
|
(cable3:Node:Physical:Cable{name:'Cable3', handle_id:'30'}),
|
||
|
|
(cable4:Node:Physical:Cable{name:'Cable4', handle_id:'31'}),
|
||
|
|
(host1:Node:Physical:Host{name:'Host1', handle_id:'32'}),
|
||
|
|
(host2:Node:Logical:Host{name:'Host2', handle_id:'33'}),
|
||
|
|
(customer1:Node:Relation:Customer{name:'Customer1', handle_id:'34'}),
|
||
|
|
(customer2:Node:Relation:Customer{name:'Customer2', handle_id:'35'}),
|
||
|
|
(customer3:Node:Relation:Customer{name:'Customer3', handle_id:'36'}),
|
||
|
|
(customer4:Node:Relation:Customer{name:'Customer4', handle_id:'37'}),
|
||
|
|
(service4:Node:Logical:Service{name:'Service4', handle_id:'38'}),
|
||
|
|
(provider2:Node:Relation:Provider{name:'Provider2', handle_id:'39'}),
|
||
|
|
(port8:Node:Physical:Port{name:'Port8', handle_id:'40'}),
|
||
|
|
(rack4:Node:Location:Rack{name:'Rack4', handle_id:'41'}),
|
||
|
|
(cable5:Node:Physical:Cable{name:'Cable5', handle_id:'42'}),
|
||
|
|
(peering_group2:Node:Logical:Peering_Group{name:'Peering Group2', handle_id:'44'}),
|
||
|
|
(cable6:Node:Physical:Cable{name:'Cable6', handle_id:'45'}),
|
||
|
|
(service5:Node:Logical:Service{name:'Service5', handle_id:'46'}),
|
||
|
|
(external_equipment1:Node:Physical:External_Equipment{name:'External Equipment1', handle_id:'47'}),
|
||
|
|
|
||
|
|
// Create relationships
|
||
|
|
(router1)-[:Has]->(port1),
|
||
|
|
(unit1)-[:Part_of]->(port1),
|
||
|
|
(router1)-[:Has]->(port6),
|
||
|
|
(unit2)-[:Part_of]->(port6),
|
||
|
|
(provider1)-[:Owns]->(router1),
|
||
|
|
(provider1)-[:Provides]->(peering_group1),
|
||
|
|
(peering_partner1)-[:Uses {ip_address:'127.0.0.1'}]->(peering_group1),
|
||
|
|
(peering_group1)-[:Depends_on]->(unit1),
|
||
|
|
(site1)-[:Has]->(rack1),
|
||
|
|
(router1)-[:Located_in]->(rack1),
|
||
|
|
(provider1)-[:Responsible_for]->(rack1),
|
||
|
|
(optical_node1)-[:Has]->(port2),
|
||
|
|
(site1)-[:Has]->(rack2),
|
||
|
|
(optical_node1)-[:Located_in]->(rack2),
|
||
|
|
(optical_node2)-[:Has]->(port3),
|
||
|
|
(site2)-[:Has]->(rack3),
|
||
|
|
(optical_node2)-[:Located_in]->(rack3),
|
||
|
|
(provider1)-[:Provides]->(optical_path1),
|
||
|
|
(service2)-[:Depends_on]->(optical_path1),
|
||
|
|
(service3)-[:Depends_on]->(unit2),
|
||
|
|
(odf1)-[:Located_in]->(rack2),
|
||
|
|
(odf1)-[:Has]->(port4),
|
||
|
|
(odf2)-[:Located_in]->(rack3),
|
||
|
|
(odf2)-[:Has]->(port5),
|
||
|
|
(odf2)-[:Has]->(port7),
|
||
|
|
(port4)<-[:Connected_to]-(cable1)-[:Connected_to]->(port2),
|
||
|
|
(port5)<-[:Connected_to]-(cable2)-[:Connected_to]->(port3),
|
||
|
|
(port4)<-[:Connected_to]-(cable3)-[:Connected_to]->(port5),
|
||
|
|
(port6)<-[:Connected_to]-(cable4)-[:Connected_to]->(port7),
|
||
|
|
(port7)<-[:Connected_to]-(cable5),
|
||
|
|
(optical_link1)-[:Depends_on]->(port2),
|
||
|
|
(optical_link2)-[:Depends_on]->(port3),
|
||
|
|
(optical_link1)-[:Depends_on]->(port4),
|
||
|
|
(optical_link2)-[:Depends_on]->(port5),
|
||
|
|
(optical_path1)-[:Depends_on]->(port4),
|
||
|
|
(optical_path1)-[:Depends_on]->(port5),
|
||
|
|
(optical_path1)-[:Depends_on]->(optical_link1),
|
||
|
|
(optical_path1)-[:Depends_on]->(optical_link2),
|
||
|
|
(provider1)-[:Owns]->(host1),
|
||
|
|
(host2)-[:Depends_on]->(host1),
|
||
|
|
(customer1)-[:Uses]->(host2),
|
||
|
|
(customer2)-[:Uses]->(service2),
|
||
|
|
(customer2)-[:Uses]->(service3),
|
||
|
|
(customer3)-[:Uses]->(service3),
|
||
|
|
(service5)-[:Depends_on]->(external_equipment1)
|
||
|
|
"""
|
||
|
|
|
||
|
|
q2 = """
|
||
|
|
// Create nodes
|
||
|
|
CREATE (physical1:Node:Physical:Generic{name:'Physical1', handle_id:'101'}),
|
||
|
|
(physical2:Node:Physical:Generic{name:'Physical2', handle_id:'102', description:'This is a port'}),
|
||
|
|
(logical1:Node:Logical:Generic{name:'Logical1', handle_id:'103'}),
|
||
|
|
(physical3:Node:Physical:Generic{name:'Physical3', handle_id:'104'}),
|
||
|
|
(logical2:Node:Logical:Generic{name:'Logical2', handle_id:'105'}),
|
||
|
|
(relation1:Node:Relation:Generic{name:'Relation1', handle_id:'106'}),
|
||
|
|
(logical3:Node:Logical:Generic{name:'Logical3', handle_id:'107'}),
|
||
|
|
(relation2:Node:Relation:Generic{name:'Relation2', handle_id:'108'}),
|
||
|
|
(location1:Node:Location:Generic{name:'Location1', handle_id:'109'}),
|
||
|
|
(location2:Node:Location:Generic{name:'Location2', handle_id:'110'}),
|
||
|
|
(logical4:Node:Logical:Generic{name:'Logical4', handle_id:'111'}),
|
||
|
|
(physical4:Node:Physical:Generic{name:'Physical4', handle_id:'112', description:'This is a cable'}),
|
||
|
|
|
||
|
|
// Create relationships
|
||
|
|
(physical1)-[:Has]->(physical2),
|
||
|
|
(logical1)-[:Part_of]->(physical2),
|
||
|
|
(physical1)-[:Has]->(physical3),
|
||
|
|
(logical2)-[:Part_of]->(physical3),
|
||
|
|
(relation1)-[:Owns]->(physical1),
|
||
|
|
(relation1)-[:Provides]->(logical3),
|
||
|
|
(relation2)-[:Uses]->(logical3),
|
||
|
|
(logical3)-[:Depends_on]->(logical1),
|
||
|
|
(location1)-[:Has]->(location2),
|
||
|
|
(physical1)-[:Located_in]->(location2),
|
||
|
|
(relation1)-[:Responsible_for]->(location2),
|
||
|
|
(logical4)-[:Depends_on]->(logical3),
|
||
|
|
(physical2)<-[:Connected_to]-(physical4)-[:Connected_to]->(physical3)
|
||
|
|
"""
|
||
|
|
|
||
|
|
# Insert mocked network
|
||
|
|
with self.neo4jdb.session as s:
|
||
|
|
s.run(q1)
|
||
|
|
|
||
|
|
# Insert generic models
|
||
|
|
with self.neo4jdb.session as s:
|
||
|
|
s.run(q2)
|
||
|
|
|
||
|
|
def test_base_node_model(self):
|
||
|
|
node_model_1 = core.get_node_model(self.neo4jdb, handle_id='101')
|
||
|
|
node_model_2 = core.get_node_model(self.neo4jdb, handle_id='102')
|
||
|
|
|
||
|
|
self.assertIsNotNone(str(node_model_1))
|
||
|
|
self.assertIsNotNone(repr(node_model_1))
|
||
|
|
|
||
|
|
self.assertEqual(node_model_1, node_model_1)
|
||
|
|
self.assertGreater(node_model_2, node_model_1)
|
||
|
|
self.assertLess(node_model_1, node_model_2)
|
||
|
|
|
||
|
|
self.assertEqual(node_model_1.handle_id, '101')
|
||
|
|
self.assertIn(node_model_1.meta_type, core.META_TYPES)
|
||
|
|
self.assertIsInstance(node_model_1.labels, list)
|
||
|
|
self.assertIsNotNone(node_model_1.data)
|
||
|
|
self.assertIsInstance(node_model_1.incoming, dict)
|
||
|
|
self.assertIsInstance(node_model_1.outgoing, dict)
|
||
|
|
self.assertIsInstance(node_model_1.relationships, dict)
|
||
|
|
|
||
|
|
def test_add_label(self):
|
||
|
|
node_model_1 = core.get_node_model(self.neo4jdb, handle_id='101')
|
||
|
|
initial_labels = node_model_1.labels
|
||
|
|
node_model_1.add_label('Test_Label')
|
||
|
|
node_model_1 = node_model_1.reload()
|
||
|
|
new_labels = node_model_1.labels
|
||
|
|
initial_labels.append('Test_Label')
|
||
|
|
self.assertEqual(sorted(new_labels), sorted(initial_labels))
|
||
|
|
|
||
|
|
def test_remove_label(self):
|
||
|
|
node_model_1 = core.get_node_model(self.neo4jdb, handle_id='101')
|
||
|
|
initial_labels = node_model_1.labels
|
||
|
|
node_model_1 = node_model_1.add_label('Test_Label')
|
||
|
|
new_labels = node_model_1.labels
|
||
|
|
expected_labels = initial_labels + ['Test_Label']
|
||
|
|
self.assertEqual(sorted(new_labels), sorted(expected_labels))
|
||
|
|
node_model_1 = node_model_1.remove_label('Test_Label')
|
||
|
|
new_labels = node_model_1.labels
|
||
|
|
self.assertEqual(sorted(new_labels), sorted(initial_labels))
|
||
|
|
|
||
|
|
def test_change_meta_type(self):
|
||
|
|
node_model_1 = core.get_node_model(self.neo4jdb, handle_id='101')
|
||
|
|
self.assertEqual(node_model_1.meta_type, 'Physical')
|
||
|
|
node_model_1 = node_model_1.change_meta_type('Logical')
|
||
|
|
self.assertEqual(node_model_1.meta_type, 'Logical')
|
||
|
|
|
||
|
|
def test_switch_type(self):
|
||
|
|
node_model_1 = core.get_node_model(self.neo4jdb, handle_id='101')
|
||
|
|
self.assertIn('Generic', node_model_1.labels)
|
||
|
|
node_model_1 = node_model_1.switch_type(old_type='Generic', new_type='New_Type')
|
||
|
|
self.assertNotIn('Generic', node_model_1.labels)
|
||
|
|
self.assertIn('New_Type', node_model_1.labels)
|
||
|
|
|
||
|
|
def test_delete(self):
|
||
|
|
node_model_1 = core.get_node_model(self.neo4jdb, handle_id='101')
|
||
|
|
node_model_1.delete()
|
||
|
|
self.assertRaises(exceptions.NodeNotFound, core.get_node_model, self.neo4jdb, handle_id='101')
|
||
|
|
|
||
|
|
def test_base_relationship_model(self):
|
||
|
|
node_model_1 = core.get_node_model(self.neo4jdb, handle_id='101')
|
||
|
|
outgoing_relationships = node_model_1.outgoing
|
||
|
|
self.assertGreater(len(outgoing_relationships), 0)
|
||
|
|
|
||
|
|
for rel_type, relationships in outgoing_relationships.items():
|
||
|
|
self.assertIsNotNone(rel_type)
|
||
|
|
for item in relationships:
|
||
|
|
relationship_model = core.get_relationship_model(self.neo4jdb, item['relationship_id'])
|
||
|
|
self.assertIsNotNone(str(relationship_model))
|
||
|
|
self.assertIsNotNone(repr(relationship_model))
|
||
|
|
self.assertIsNotNone(relationship_model.type)
|
||
|
|
self.assertIsInstance(relationship_model.id, int)
|
||
|
|
self.assertIsNotNone(relationship_model.data)
|
||
|
|
self.assertEqual(relationship_model.start['handle_id'], node_model_1.handle_id)
|
||
|
|
self.assertEqual(relationship_model.end['handle_id'], item['node'].handle_id)
|
||
|
|
|
||
|
|
def test_get_location_path(self):
|
||
|
|
# Model with location
|
||
|
|
physical1 = core.get_node_model(self.neo4jdb, handle_id='101')
|
||
|
|
location_path = physical1.get_location_path()
|
||
|
|
self.assertEqual(location_path['location_path'][0]['name'], 'Location1')
|
||
|
|
self.assertEqual(location_path['location_path'][1]['name'], 'Location2')
|
||
|
|
|
||
|
|
# Model without location
|
||
|
|
relation1 = core.get_node_model(self.neo4jdb, handle_id='106')
|
||
|
|
location_path = relation1.get_location_path()
|
||
|
|
self.assertEqual(location_path['location_path'], [])
|
||
|
|
|
||
|
|
def test_get_location(self):
|
||
|
|
# Model with location
|
||
|
|
physical1 = core.get_node_model(self.neo4jdb, handle_id='101')
|
||
|
|
location = physical1.get_location()
|
||
|
|
self.assertIsInstance(location['Located_in'][0]['node'], models.LocationModel)
|
||
|
|
self.assertEqual(location['Located_in'][0]['node'].data['name'], 'Location2')
|
||
|
|
self.assertIsInstance(location['Located_in'][0]['relationship_id'], int)
|
||
|
|
|
||
|
|
# Model without location
|
||
|
|
relation1 = core.get_node_model(self.neo4jdb, handle_id='106')
|
||
|
|
location = relation1.get_location()
|
||
|
|
self.assertIsNone(location.get('Located_in'))
|
||
|
|
|
||
|
|
def test_get_placement_path(self):
|
||
|
|
# Models with placement path
|
||
|
|
physical2 = core.get_node_model(self.neo4jdb, handle_id='102')
|
||
|
|
placement_path = physical2.get_placement_path()
|
||
|
|
self.assertEqual(placement_path['placement_path'][0]['name'], 'Physical1')
|
||
|
|
|
||
|
|
# Model without placement path
|
||
|
|
relation1 = core.get_node_model(self.neo4jdb, handle_id='106')
|
||
|
|
location_path = relation1.get_placement_path()
|
||
|
|
self.assertEqual(location_path['placement_path'], [])
|
||
|
|
|
||
|
|
def test_get_child_form_data(self):
|
||
|
|
physical1 = core.get_node_model(self.neo4jdb, handle_id='101')
|
||
|
|
child_form_data = physical1.get_child_form_data(node_type='Generic')
|
||
|
|
for data in child_form_data:
|
||
|
|
self.assertIn(data['handle_id'], ['102', '104'])
|
||
|
|
self.assertIn(data['name'], ['Physical2', 'Physical3'])
|
||
|
|
self.assertIn(data['description'], ['This is a port', None])
|
||
|
|
self.assertEqual(data['labels'], [u'Node', u'Physical', u'Generic'])
|
||
|
|
|
||
|
|
def test_get_relations(self):
|
||
|
|
physical1 = core.get_node_model(self.neo4jdb, handle_id='101')
|
||
|
|
relations = physical1.get_relations()
|
||
|
|
self.assertEqual(physical1.meta_type, 'Physical')
|
||
|
|
self.assertIsInstance(relations['Owns'][0]['node'], models.RelationModel)
|
||
|
|
|
||
|
|
logical3 = core.get_node_model(self.neo4jdb, handle_id='107')
|
||
|
|
relations = logical3.get_relations()
|
||
|
|
self.assertEqual(logical3.meta_type, 'Logical')
|
||
|
|
self.assertIsInstance(relations['Uses'][0]['node'], models.RelationModel)
|
||
|
|
|
||
|
|
logical3 = core.get_node_model(self.neo4jdb, handle_id='107')
|
||
|
|
relations = logical3.get_relations()
|
||
|
|
self.assertIsInstance(relations['Provides'][0]['node'], models.RelationModel)
|
||
|
|
|
||
|
|
location2 = core.get_node_model(self.neo4jdb, handle_id='110')
|
||
|
|
relations = location2.get_relations()
|
||
|
|
self.assertIsInstance(relations['Responsible_for'][0]['node'], models.RelationModel)
|
||
|
|
|
||
|
|
def test_get_dependencies(self):
|
||
|
|
logical3 = core.get_node_model(self.neo4jdb, handle_id='107')
|
||
|
|
dependencies = logical3.get_dependencies()
|
||
|
|
self.assertEqual(len(dependencies['Depends_on']), 1)
|
||
|
|
self.assertEqual(dependencies['Depends_on'][0]['node'].handle_id, '103')
|
||
|
|
self.assertIsInstance(dependencies['Depends_on'][0]['node'], models.LogicalModel)
|
||
|
|
|
||
|
|
def test_get_dependents(self):
|
||
|
|
logical1 = core.get_node_model(self.neo4jdb, handle_id='103')
|
||
|
|
dependents = logical1.get_dependents()
|
||
|
|
self.assertEqual(len(dependents['Depends_on']), 1)
|
||
|
|
self.assertEqual(dependents['Depends_on'][0]['node'].handle_id, '107')
|
||
|
|
self.assertIsInstance(dependents['Depends_on'][0]['node'], models.LogicalModel)
|
||
|
|
|
||
|
|
def test_get_dependent_as_types(self):
|
||
|
|
logical1 = core.get_node_model(self.neo4jdb, handle_id='103')
|
||
|
|
dependents = logical1.get_dependent_as_types()
|
||
|
|
self.assertEqual(dependents['direct'][0]['name'], 'Logical3')
|
||
|
|
self.assertEqual(dependents['links'], [])
|
||
|
|
self.assertEqual(dependents['oms'], [])
|
||
|
|
self.assertEqual(dependents['paths'], [])
|
||
|
|
self.assertEqual(dependents['services'], [])
|
||
|
|
|
||
|
|
def test_get_dependent_as_types_port_with_unit_services(self):
|
||
|
|
port6 = core.get_node_model(self.neo4jdb, handle_id='4')
|
||
|
|
dependent = port6.get_dependent_as_types()
|
||
|
|
self.assertEqual(dependent['direct'], [])
|
||
|
|
self.assertEqual(dependent['links'], [])
|
||
|
|
self.assertEqual(dependent['oms'], [])
|
||
|
|
self.assertEqual(dependent['paths'], [])
|
||
|
|
self.assertEqual(len(dependent['services']), 1)
|
||
|
|
self.assertEqual(dependent['services'][0]['name'], 'Service3')
|
||
|
|
|
||
|
|
def test_get_dependent_as_types_equipment(self):
|
||
|
|
external1 = core.get_node_model(self.neo4jdb, handle_id='47')
|
||
|
|
# a bit nasty just moving a port
|
||
|
|
external1.set_has('24') # port4
|
||
|
|
dependents = external1.get_dependent_as_types()
|
||
|
|
self.assertIn('Service5', [n['name'] for n in dependents['direct']])
|
||
|
|
self.assertIn('Optical Link1', [n['name'] for n in dependents['links']])
|
||
|
|
self.assertIn('Optical Link2', [n['name'] for n in dependents['links']])
|
||
|
|
self.assertEqual(dependents['oms'], [])
|
||
|
|
self.assertIn('Optical Path1', [n['name'] for n in dependents['paths']])
|
||
|
|
self.assertIn('Service2', [n['name'] for n in dependents['services']])
|
||
|
|
self.assertIn('Service5', [n['name'] for n in dependents['services']])
|
||
|
|
|
||
|
|
def test_get_dependent_as_types_equipment_only_direct(self):
|
||
|
|
external1 = core.get_node_model(self.neo4jdb, handle_id='47')
|
||
|
|
dependents = external1.get_dependent_as_types()
|
||
|
|
self.assertEqual(dependents['direct'][0]['name'], 'Service5')
|
||
|
|
self.assertEqual(dependents['links'], [])
|
||
|
|
self.assertEqual(dependents['oms'], [])
|
||
|
|
self.assertEqual(dependents['paths'], [])
|
||
|
|
self.assertEqual(dependents['services'][0]['name'], 'Service5')
|
||
|
|
|
||
|
|
def test_get_dependencies_as_types(self):
|
||
|
|
logical4 = core.get_node_model(self.neo4jdb, handle_id='111')
|
||
|
|
dependencies = logical4.get_dependencies_as_types()
|
||
|
|
self.assertEqual(dependencies['direct'][0]['name'], 'Logical3')
|
||
|
|
self.assertEqual(dependencies['links'], [])
|
||
|
|
self.assertEqual(dependencies['oms'], [])
|
||
|
|
self.assertEqual(dependencies['paths'], [])
|
||
|
|
self.assertEqual(dependencies['services'], [])
|
||
|
|
|
||
|
|
def test_get_ports(self):
|
||
|
|
physical4 = core.get_node_model(self.neo4jdb, handle_id='112')
|
||
|
|
ports = physical4.get_ports()
|
||
|
|
self.assertIsInstance(ports, list)
|
||
|
|
self.assertEqual(len(ports), 0)
|
||
|
|
|
||
|
|
def test_get_part_of_logical_model(self):
|
||
|
|
unit1 = core.get_node_model(self.neo4jdb, handle_id='3')
|
||
|
|
part_of = unit1.get_part_of()
|
||
|
|
self.assertEqual(part_of['Part_of'][0]['node'].handle_id, '2')
|
||
|
|
|
||
|
|
def test_set_user_logical_model(self):
|
||
|
|
customer4 = core.get_node_model(self.neo4jdb, handle_id='37')
|
||
|
|
service4 = core.get_node_model(self.neo4jdb, handle_id='38')
|
||
|
|
|
||
|
|
result = service4.set_user(customer4.handle_id)
|
||
|
|
self.assertEqual(result['Uses'][0]['created'], True)
|
||
|
|
relations = service4.get_relations()
|
||
|
|
self.assertEqual(len(relations['Uses']), 1)
|
||
|
|
self.assertEqual(relations['Uses'][0]['node'].handle_id, customer4.handle_id)
|
||
|
|
|
||
|
|
# Do not accept duplicates
|
||
|
|
result = service4.set_user(customer4.handle_id)
|
||
|
|
self.assertEqual(result['Uses'][0]['created'], False)
|
||
|
|
relations = service4.get_relations()
|
||
|
|
self.assertEqual(len(relations['Uses']), 1)
|
||
|
|
|
||
|
|
def test_set_provider_logical_model(self):
|
||
|
|
provider_1 = core.get_node_model(self.neo4jdb, handle_id='6')
|
||
|
|
service4 = core.get_node_model(self.neo4jdb, handle_id='38')
|
||
|
|
|
||
|
|
result = service4.set_provider(provider_1.handle_id)
|
||
|
|
self.assertEqual(result['Provides'][0]['created'], True)
|
||
|
|
relations = service4.get_relations()
|
||
|
|
self.assertEqual(len(relations['Provides']), 1)
|
||
|
|
self.assertEqual(relations['Provides'][0]['node'].handle_id, provider_1.handle_id)
|
||
|
|
|
||
|
|
# Do not accept duplicates
|
||
|
|
result = service4.set_provider(provider_1.handle_id)
|
||
|
|
self.assertEqual(result['Provides'][0]['created'], False)
|
||
|
|
relations = service4.get_relations()
|
||
|
|
self.assertEqual(len(relations['Provides']), 1)
|
||
|
|
|
||
|
|
def test_set_dependency_logical_model(self):
|
||
|
|
optical_path1 = core.get_node_model(self.neo4jdb, handle_id='20')
|
||
|
|
service4 = core.get_node_model(self.neo4jdb, handle_id='38')
|
||
|
|
|
||
|
|
result = service4.set_dependency(optical_path1.handle_id)
|
||
|
|
self.assertEqual(result['Depends_on'][0]['created'], True)
|
||
|
|
relations = service4.get_dependencies()
|
||
|
|
self.assertEqual(len(relations['Depends_on']), 1)
|
||
|
|
self.assertEqual(relations['Depends_on'][0]['node'].handle_id, optical_path1.handle_id)
|
||
|
|
|
||
|
|
# Do not accept duplicates
|
||
|
|
result = service4.set_dependency(optical_path1.handle_id)
|
||
|
|
self.assertEqual(result['Depends_on'][0]['created'], False)
|
||
|
|
relations = service4.get_dependencies()
|
||
|
|
self.assertEqual(len(relations['Depends_on']), 1)
|
||
|
|
|
||
|
|
def test_get_location_physical_model(self):
|
||
|
|
router1 = core.get_node_model(self.neo4jdb, handle_id='1')
|
||
|
|
location = router1.get_location()
|
||
|
|
self.assertIsInstance(location['Located_in'][0]['node'], models.LocationModel)
|
||
|
|
self.assertEqual(location['Located_in'][0]['node'].data['name'], 'Rack1')
|
||
|
|
self.assertIsInstance(location['Located_in'][0]['relationship_id'], int)
|
||
|
|
|
||
|
|
def test_set_owner_physical_model(self):
|
||
|
|
router1 = core.get_node_model(self.neo4jdb, handle_id='1')
|
||
|
|
customer4 = core.get_node_model(self.neo4jdb, handle_id='37')
|
||
|
|
|
||
|
|
result = router1.set_owner(customer4.handle_id)
|
||
|
|
self.assertEqual(result['Owns'][0]['created'], True)
|
||
|
|
relations = router1.get_relations()
|
||
|
|
self.assertEqual(len(relations['Owns']), 2)
|
||
|
|
|
||
|
|
# Do not accept duplicates
|
||
|
|
result = router1.set_owner(customer4.handle_id)
|
||
|
|
self.assertEqual(result['Owns'][0]['created'], False)
|
||
|
|
relations = router1.get_relations()
|
||
|
|
self.assertEqual(len(relations['Owns']), 2)
|
||
|
|
|
||
|
|
def test_set_provider_physical_model(self):
|
||
|
|
router1 = core.get_node_model(self.neo4jdb, handle_id='1')
|
||
|
|
provider_2 = core.get_node_model(self.neo4jdb, handle_id='39')
|
||
|
|
|
||
|
|
result = router1.set_provider(provider_2.handle_id)
|
||
|
|
self.assertEqual(result['Provides'][0]['created'], True)
|
||
|
|
relations = router1.get_relations()
|
||
|
|
self.assertEqual(len(relations['Provides']), 1)
|
||
|
|
self.assertEqual(relations['Provides'][0]['node'].handle_id, provider_2.handle_id)
|
||
|
|
|
||
|
|
# Do not accept duplicates
|
||
|
|
result = router1.set_provider(provider_2.handle_id)
|
||
|
|
self.assertEqual(result['Provides'][0]['created'], False)
|
||
|
|
relations = router1.get_relations()
|
||
|
|
self.assertEqual(len(relations['Provides']), 1)
|
||
|
|
|
||
|
|
def test_set_location_physical_model(self):
|
||
|
|
router1 = core.get_node_model(self.neo4jdb, handle_id='1')
|
||
|
|
rack_2 = core.get_node_model(self.neo4jdb, handle_id='15')
|
||
|
|
|
||
|
|
result = router1.set_location(rack_2.handle_id)
|
||
|
|
self.assertEqual(result['Located_in'][0]['created'], True)
|
||
|
|
location = router1.get_location()
|
||
|
|
self.assertEqual(len(location['Located_in']), 2)
|
||
|
|
|
||
|
|
# Do not accept duplicates
|
||
|
|
result = router1.set_location(rack_2.handle_id)
|
||
|
|
self.assertEqual(result['Located_in'][0]['created'], False)
|
||
|
|
location = router1.get_location()
|
||
|
|
self.assertEqual(len(location['Located_in']), 2)
|
||
|
|
|
||
|
|
def test_set_and_get_has_physical_model(self):
|
||
|
|
router1 = core.get_node_model(self.neo4jdb, handle_id='1')
|
||
|
|
port8 = core.get_node_model(self.neo4jdb, handle_id='40')
|
||
|
|
|
||
|
|
result = router1.set_has(port8.handle_id)
|
||
|
|
self.assertEqual(result['Has'][0]['created'], True)
|
||
|
|
children = router1.get_has()
|
||
|
|
self.assertEqual(len(children['Has']), 3)
|
||
|
|
|
||
|
|
# Do not accept duplicates
|
||
|
|
result = router1.set_has(port8.handle_id)
|
||
|
|
self.assertEqual(result['Has'][0]['created'], False)
|
||
|
|
children = router1.get_has()
|
||
|
|
self.assertEqual(len(children['Has']), 3)
|
||
|
|
|
||
|
|
def test_set_and_get_part_of_physical_model(self):
|
||
|
|
port8 = core.get_node_model(self.neo4jdb, handle_id='40')
|
||
|
|
unit1 = core.get_node_model(self.neo4jdb, handle_id='3')
|
||
|
|
|
||
|
|
result = port8.set_part_of(unit1.handle_id)
|
||
|
|
self.assertEqual(result['Part_of'][0]['created'], True)
|
||
|
|
children = port8.get_part_of()
|
||
|
|
self.assertEqual(len(children['Part_of']), 1)
|
||
|
|
|
||
|
|
# Do not accept duplicates
|
||
|
|
result = port8.set_part_of(unit1.handle_id)
|
||
|
|
self.assertEqual(result['Part_of'][0]['created'], False)
|
||
|
|
children = port8.get_part_of()
|
||
|
|
self.assertEqual(len(children['Part_of']), 1)
|
||
|
|
|
||
|
|
def test_get_parent_physical_model(self):
|
||
|
|
port1 = core.get_node_model(self.neo4jdb, handle_id='2')
|
||
|
|
parent = port1.get_parent()
|
||
|
|
self.assertIsInstance(parent['Has'][0]['node'], models.PhysicalModel)
|
||
|
|
self.assertEqual(parent['Has'][0]['node'].data['name'], 'Router1')
|
||
|
|
self.assertIsInstance(parent['Has'][0]['relationship_id'], int)
|
||
|
|
|
||
|
|
def test_get_location_path_location_model(self):
|
||
|
|
rack_2 = core.get_node_model(self.neo4jdb, handle_id='15')
|
||
|
|
location_path = rack_2.get_location_path()
|
||
|
|
self.assertEqual(location_path['location_path'][0]['name'], 'Site1')
|
||
|
|
|
||
|
|
def test_get_parent_location_model(self):
|
||
|
|
rack_2 = core.get_node_model(self.neo4jdb, handle_id='15')
|
||
|
|
parent = rack_2.get_parent()
|
||
|
|
self.assertEqual(parent['Has'][0]['node'].data['name'], 'Site1')
|
||
|
|
|
||
|
|
def test_get_located_in_location_model(self):
|
||
|
|
rack_2 = core.get_node_model(self.neo4jdb, handle_id='15')
|
||
|
|
located_in = rack_2.get_located_in()
|
||
|
|
self.assertEqual(len(located_in['Located_in']), 2)
|
||
|
|
optical_node = [node for node in located_in['Located_in'] if node['node'].data['name'] == 'Optical Node1'][0]
|
||
|
|
self.assertIsInstance(optical_node['node'], models.PhysicalModel)
|
||
|
|
self.assertIsInstance(optical_node['relationship_id'], int)
|
||
|
|
|
||
|
|
def test_set_and_get_has_location_model(self):
|
||
|
|
site1 = core.get_node_model(self.neo4jdb, handle_id='11')
|
||
|
|
rack_4 = core.get_node_model(self.neo4jdb, handle_id='41')
|
||
|
|
|
||
|
|
result = site1.set_has(rack_4.handle_id)
|
||
|
|
self.assertEqual(result['Has'][0]['created'], True)
|
||
|
|
children = site1.get_has()
|
||
|
|
self.assertEqual(len(children['Has']), 3)
|
||
|
|
|
||
|
|
# Do not accept duplicates
|
||
|
|
result = site1.set_has(rack_4.handle_id)
|
||
|
|
self.assertEqual(result['Has'][0]['created'], False)
|
||
|
|
children = site1.get_has()
|
||
|
|
self.assertEqual(len(children['Has']), 3)
|
||
|
|
|
||
|
|
def test_set_responsible_for_location_model(self):
|
||
|
|
rack_4 = core.get_node_model(self.neo4jdb, handle_id='41')
|
||
|
|
provider_2 = core.get_node_model(self.neo4jdb, handle_id='39')
|
||
|
|
|
||
|
|
result = rack_4.set_responsible_for(provider_2.handle_id)
|
||
|
|
self.assertEqual(result['Responsible_for'][0]['created'], True)
|
||
|
|
relations = rack_4.get_relations()
|
||
|
|
self.assertEqual(len(relations['Responsible_for']), 1)
|
||
|
|
self.assertEqual(relations['Responsible_for'][0]['node'].handle_id, provider_2.handle_id)
|
||
|
|
|
||
|
|
# Do not accept duplicates
|
||
|
|
result = rack_4.set_responsible_for(provider_2.handle_id)
|
||
|
|
self.assertEqual(result['Responsible_for'][0]['created'], False)
|
||
|
|
relations = rack_4.get_relations()
|
||
|
|
self.assertEqual(len(relations['Responsible_for']), 1)
|
||
|
|
|
||
|
|
# TODO: EquipmentModel get_ports should probably work as CommonQueries get_ports
|
||
|
|
def test_get_ports_equipment_model(self):
|
||
|
|
odf1 = core.get_node_model(self.neo4jdb, handle_id='23')
|
||
|
|
ports = odf1.get_ports()
|
||
|
|
self.assertIsInstance(ports, dict)
|
||
|
|
self.assertEqual(len(ports['Has']), 1)
|
||
|
|
for rel_type, items in ports.items():
|
||
|
|
self.assertEqual(len(items), 1)
|
||
|
|
|
||
|
|
def test_get_port_equipment_model(self):
|
||
|
|
router1 = core.get_node_model(self.neo4jdb, handle_id='1')
|
||
|
|
ports = router1.get_port('Port1')
|
||
|
|
self.assertEqual(len(ports['Has']), 1)
|
||
|
|
self.assertIsInstance(ports['Has'][0]['node'], models.PortModel)
|
||
|
|
self.assertEqual(ports['Has'][0]['node'].data['name'], 'Port1')
|
||
|
|
|
||
|
|
def test_get_dependent_as_types_equipment_model(self):
|
||
|
|
optical_node2 = core.get_node_model(self.neo4jdb, handle_id='16')
|
||
|
|
dependents = optical_node2.get_dependent_as_types()
|
||
|
|
self.assertEqual(dependents['direct'], [])
|
||
|
|
self.assertEqual(dependents['links'][0]['name'], 'Optical Link2')
|
||
|
|
self.assertEqual(dependents['oms'], [])
|
||
|
|
self.assertEqual(dependents['paths'][0]['name'], 'Optical Path1')
|
||
|
|
self.assertEqual(dependents['services'][0]['name'], 'Service2')
|
||
|
|
|
||
|
|
def test_get_connections_equipment_model(self):
|
||
|
|
odf2 = core.get_node_model(self.neo4jdb, handle_id='25')
|
||
|
|
connections = odf2.get_connections()
|
||
|
|
self.assertEqual(len(connections), 4)
|
||
|
|
for connection in connections:
|
||
|
|
self.assertIsNotNone(connection['porta'])
|
||
|
|
self.assertIsNotNone(connection['cable'])
|
||
|
|
|
||
|
|
def test_get_connections_subequipment_model(self):
|
||
|
|
port4 = core.get_node_model(self.neo4jdb, handle_id='24')
|
||
|
|
connections = port4.get_connections()
|
||
|
|
self.assertEqual(len(connections), 2)
|
||
|
|
for connection in connections:
|
||
|
|
self.assertIsNotNone(connection['porta'])
|
||
|
|
self.assertIsNotNone(connection['cable'])
|
||
|
|
|
||
|
|
def test_get_dependent_as_types_host_model(self):
|
||
|
|
host1 = core.get_node_model(self.neo4jdb, handle_id='32')
|
||
|
|
dependents = host1.get_dependent_as_types()
|
||
|
|
self.assertEqual(dependents['direct'][0]['name'], 'Host2')
|
||
|
|
self.assertEqual(dependents['links'], [])
|
||
|
|
self.assertEqual(dependents['oms'], [])
|
||
|
|
self.assertEqual(dependents['paths'], [])
|
||
|
|
self.assertEqual(dependents['services'], [])
|
||
|
|
|
||
|
|
# TODO: Fix duplicates
|
||
|
|
def test_get_units_port_model(self):
|
||
|
|
port1 = core.get_node_model(self.neo4jdb, handle_id='2')
|
||
|
|
units = port1.get_units()
|
||
|
|
self.assertEqual(units['Part_of'][0]['node'].handle_id, '3')
|
||
|
|
|
||
|
|
def test_get_unit_port_model(self):
|
||
|
|
port1 = core.get_node_model(self.neo4jdb, handle_id='2')
|
||
|
|
units = port1.get_unit('Unit1')
|
||
|
|
self.assertEqual(units['Part_of'][0]['node'].handle_id, '3')
|
||
|
|
|
||
|
|
def test_get_connected_to_port_model(self):
|
||
|
|
port4 = core.get_node_model(self.neo4jdb, handle_id='24')
|
||
|
|
connected_to = port4.get_connected_to()
|
||
|
|
self.assertIn(connected_to['Connected_to'][0]['node'].handle_id, ['28', '30'])
|
||
|
|
|
||
|
|
def test_get_connection_path_port_model(self):
|
||
|
|
port4 = core.get_node_model(self.neo4jdb, handle_id='24')
|
||
|
|
connection_path = port4.get_connection_path()
|
||
|
|
self.assertEqual(len(connection_path), 7)
|
||
|
|
|
||
|
|
def test_get_child_form_data_router_model(self):
|
||
|
|
physical1 = core.get_node_model(self.neo4jdb, handle_id='1')
|
||
|
|
child_form_data = physical1.get_child_form_data(node_type='Port')
|
||
|
|
self.assertEqual(child_form_data[0]['handle_id'], '2')
|
||
|
|
self.assertEqual(child_form_data[0]['name'], 'Port1')
|
||
|
|
self.assertEqual(child_form_data[0]['description'], None)
|
||
|
|
self.assertEqual(child_form_data[0]['labels'], [u'Node', u'Physical', u'Port'])
|
||
|
|
|
||
|
|
def test_get_peering_groups_peering_partner_model(self):
|
||
|
|
peering_partner1 = core.get_node_model(self.neo4jdb, handle_id='8')
|
||
|
|
host_services = peering_partner1.get_peering_groups()
|
||
|
|
self.assertEqual(len(host_services['Uses']), 1)
|
||
|
|
self.assertIsInstance(host_services['Uses'][0]['node'], models.PeeringGroupModel)
|
||
|
|
self.assertEqual(host_services['Uses'][0]['node'].data['name'], 'Peering Group1')
|
||
|
|
self.assertEqual(host_services['Uses'][0]['relationship']['ip_address'], '127.0.0.1')
|
||
|
|
|
||
|
|
def test_set_and_get_peering_group_peering_partner_model(self):
|
||
|
|
peering_partner1 = core.get_node_model(self.neo4jdb, handle_id='8')
|
||
|
|
peering_group2 = core.get_node_model(self.neo4jdb, handle_id='44')
|
||
|
|
|
||
|
|
peering_partner1.set_peering_group(peering_group2.handle_id, ip_address='127.0.0.2')
|
||
|
|
peering_groups = peering_partner1.get_peering_group(peering_group2.handle_id, ip_address='127.0.0.2')
|
||
|
|
self.assertEqual(len(peering_groups['Uses']), 1)
|
||
|
|
|
||
|
|
# TODO: Fix duplicates
|
||
|
|
|
||
|
|
def test_set_and_get_group_dependency_peering_group_model(self):
|
||
|
|
peering_group2 = core.get_node_model(self.neo4jdb, handle_id='44')
|
||
|
|
unit2 = core.get_node_model(self.neo4jdb, handle_id='5')
|
||
|
|
|
||
|
|
peering_group2.set_group_dependency(unit2.handle_id, ip_address='127.0.0.3')
|
||
|
|
dependencies = peering_group2.get_group_dependency(unit2.handle_id, ip_address='127.0.0.3')
|
||
|
|
self.assertEqual(len(dependencies['Depends_on']), 1)
|
||
|
|
|
||
|
|
# TODO: Fix duplicates
|
||
|
|
|
||
|
|
def test_get_connected_equipment_cable_model(self):
|
||
|
|
cable1 = core.get_node_model(self.neo4jdb, handle_id='28')
|
||
|
|
connections = cable1.get_connected_equipment()
|
||
|
|
self.assertEqual(len(connections), 2)
|
||
|
|
for connection in connections:
|
||
|
|
self.assertIsNotNone(connection['port'])
|
||
|
|
self.assertIsNotNone(connection['end'])
|
||
|
|
self.assertIsNotNone(connection['location'])
|
||
|
|
self.assertIsNotNone(connection['site'])
|
||
|
|
|
||
|
|
def test_get_dependent_as_types_cable_model(self):
|
||
|
|
cable1 = core.get_node_model(self.neo4jdb, handle_id='28')
|
||
|
|
dependents = cable1.get_dependent_as_types()
|
||
|
|
|
||
|
|
for optical_link in dependents['links']:
|
||
|
|
self.assertTrue(optical_link['name'] in ['Optical Link1', 'Optical Link2'])
|
||
|
|
self.assertEqual(dependents['oms'], [])
|
||
|
|
self.assertEqual(dependents['paths'][0]['name'], 'Optical Path1')
|
||
|
|
self.assertEqual(dependents['services'][0]['name'], 'Service2')
|
||
|
|
|
||
|
|
def test_get_services_cable_model(self):
|
||
|
|
cable1 = core.get_node_model(self.neo4jdb, handle_id='28')
|
||
|
|
services = cable1.get_services()
|
||
|
|
self.assertEqual(len(services), 1)
|
||
|
|
self.assertEqual(services[0]['service']['name'], 'Service2')
|
||
|
|
self.assertIsInstance(services[0]['users'], list)
|
||
|
|
self.assertEqual(services[0]['users'][0]['name'], 'Customer2')
|
||
|
|
|
||
|
|
def test_get_connection_path_cable_model(self):
|
||
|
|
cable1 = core.get_node_model(self.neo4jdb, handle_id='28')
|
||
|
|
connection_path = cable1.get_connection_path()
|
||
|
|
self.assertEqual(len(connection_path), 7)
|
||
|
|
|
||
|
|
def test_set_connected_to_cable_model(self):
|
||
|
|
cable6 = core.get_node_model(self.neo4jdb, handle_id='45')
|
||
|
|
port7 = core.get_node_model(self.neo4jdb, handle_id='27')
|
||
|
|
|
||
|
|
result = cable6.set_connected_to(port7.handle_id)
|
||
|
|
self.assertEqual(result['Connected_to'][0]['created'], True)
|
||
|
|
relationships = cable6.relationships
|
||
|
|
self.assertEqual(len(relationships['Connected_to']), 1)
|
||
|
|
|
||
|
|
# Do not accept duplicates
|
||
|
|
result = cable6.set_connected_to(port7.handle_id)
|
||
|
|
self.assertEqual(result['Connected_to'][0]['created'], False)
|
||
|
|
relationships = cable6.relationships
|
||
|
|
self.assertEqual(len(relationships['Connected_to']), 1)
|
||
|
|
|
||
|
|
def test_get_placement_path_unit_model(self):
|
||
|
|
unit1 = core.get_node_model(self.neo4jdb, handle_id='3')
|
||
|
|
placement_path = unit1.get_placement_path()
|
||
|
|
self.assertEqual(placement_path['placement_path'][0]['name'], 'Router1')
|
||
|
|
self.assertEqual(placement_path['placement_path'][1]['name'], 'Port1')
|
||
|
|
|
||
|
|
def test_get_location_path_unit_model(self):
|
||
|
|
unit1 = core.get_node_model(self.neo4jdb, handle_id='3')
|
||
|
|
location_path = unit1.get_location_path()
|
||
|
|
self.assertEqual(location_path['location_path'][0]['name'], 'Site1')
|
||
|
|
self.assertEqual(location_path['location_path'][1]['name'], 'Rack1')
|
||
|
|
self.assertEqual(location_path['location_path'][2]['name'], 'Router1')
|
||
|
|
self.assertEqual(location_path['location_path'][3]['name'], 'Port1')
|
||
|
|
|
||
|
|
def test_get_customers_service_model(self):
|
||
|
|
service2 = core.get_node_model(self.neo4jdb, handle_id='9')
|
||
|
|
customers = service2.get_customers()
|
||
|
|
self.assertEqual(len(customers['customers']), 1)
|
||
|
|
self.assertIsInstance(customers['customers'][0]['node'], models.CustomerModel)
|