# Path computed by PCE is feasible according to Gnpy
def test_04_path_computation_FeasibleWithPCE(self):
response = test_utils.path_computation_request("request-1", "service-1",
- {"node-id": "XPONDER-1", "service-rate": "100", "service-format": "Ethernet", "clli": "Node1"},
- {"node-id": "XPONDER-5", "service-rate": "100", "service-format": "Ethernet", "clli": "Node5"})
+ {"node-id": "XPONDER-1", "service-rate": "100",
+ "service-format": "Ethernet", "clli": "Node1"},
+ {"node-id": "XPONDER-5", "service-rate": "100", "service-format": "Ethernet", "clli": "Node5"})
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(res['output']['configuration-response-common'][
# another one (low SNR)
def test_05_path_computation_FoundByPCE_NotFeasibleByGnpy(self):
response = test_utils.path_computation_request("request-2", "service-2",
- {"node-id": "XPONDER-1", "service-rate": "100", "service-format": "Ethernet", "clli": "Node1"},
- {"node-id": "XPONDER-5", "service-rate": "100", "service-format": "Ethernet", "clli": "Node5"},
- {"include_": { "ordered-hops": [
- {"hop-number": "0", "hop-type": {"node-id": "OpenROADM-2"}},
- {"hop-number": "1", "hop-type": {"node-id": "OpenROADM-3"}},
- {"hop-number": "2", "hop-type": {"node-id": "OpenROADM-4"}}]}})
+ {"node-id": "XPONDER-1", "service-rate": "100",
+ "service-format": "Ethernet", "clli": "Node1"},
+ {"node-id": "XPONDER-5", "service-rate": "100",
+ "service-format": "Ethernet", "clli": "Node5"},
+ {"include_": {"ordered-hops": [
+ {"hop-number": "0", "hop-type": {"node-id": "OpenROADM-2"}},
+ {"hop-number": "1", "hop-type": {"node-id": "OpenROADM-3"}},
+ {"hop-number": "2", "hop-type": {"node-id": "OpenROADM-4"}}]}})
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(res['output']['configuration-response-common'][
# #PCE cannot find a path while GNPy finds a feasible one
def test_06_path_computation_NotFoundByPCE_FoundByGNPy(self):
response = test_utils.path_computation_request("request-3", "service-3",
- {"node-id": "XPONDER-1", "service-rate": "100", "service-format": "Ethernet", "clli": "Node1"},
- {"node-id": "XPONDER-4", "service-rate": "100", "service-format": "Ethernet", "clli": "Node5"},
- {"include_": {"ordered-hops": [
- {"hop-number": "0", "hop-type": {"node-id": "OpenROADM-2"}},
- {"hop-number": "1", "hop-type": {"node-id": "OpenROADM-3"}}]}})
+ {"node-id": "XPONDER-1", "service-rate": "100",
+ "service-format": "Ethernet", "clli": "Node1"},
+ {"node-id": "XPONDER-4", "service-rate": "100",
+ "service-format": "Ethernet", "clli": "Node5"},
+ {"include_": {"ordered-hops": [
+ {"hop-number": "0", "hop-type": {"node-id": "OpenROADM-2"}},
+ {"hop-number": "1", "hop-type": {"node-id": "OpenROADM-3"}}]}})
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(res['output']['configuration-response-common'][
# Not found path by PCE and GNPy cannot find another one
def test_07_path_computation_FoundByPCE_NotFeasibleByGnpy(self):
response = test_utils.path_computation_request("request-4", "service-4",
- {"node-id": "XPONDER-1", "service-rate": "100", "service-format": "Ethernet", "clli": "Node1"},
- {"node-id": "XPONDER-4", "service-rate": "100", "service-format": "Ethernet", "clli": "Node5"},
- {"include_": {"ordered-hops": [
- {"hop-number": "0","hop-type": {"node-id": "OpenROADM-2"}},
- {"hop-number": "1","hop-type": {"node-id": "OpenROADM-3"}},
- {"hop-number": "2","hop-type": {"node-id": "OpenROADM-4"}},
- {"hop-number": "3","hop-type": {"node-id": "OpenROADM-3"}}]}})
+ {"node-id": "XPONDER-1", "service-rate": "100",
+ "service-format": "Ethernet", "clli": "Node1"},
+ {"node-id": "XPONDER-4", "service-rate": "100",
+ "service-format": "Ethernet", "clli": "Node5"},
+ {"include_": {"ordered-hops": [
+ {"hop-number": "0", "hop-type": {"node-id": "OpenROADM-2"}},
+ {"hop-number": "1", "hop-type": {"node-id": "OpenROADM-3"}},
+ {"hop-number": "2", "hop-type": {"node-id": "OpenROADM-4"}},
+ {"hop-number": "3", "hop-type": {"node-id": "OpenROADM-3"}}]}})
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(res['output']['configuration-response-common'][
# Path Computation success
def test_04_path_computation_xpdr_bi(self):
response = test_utils.path_computation_request("request-1", "service-1",
- {"node-id": "XPDRA01", "service-rate": "100", "service-format": "Ethernet", "clli": "nodeA"},
- {"node-id": "XPDRC01", "service-rate": "100", "service-format": "Ethernet", "clli": "nodeC"})
+ {"node-id": "XPDRA01", "service-rate": "100",
+ "service-format": "Ethernet", "clli": "nodeA"},
+ {"node-id": "XPDRC01", "service-rate": "100", "service-format": "Ethernet", "clli": "nodeC"})
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Path is calculated',
# Path Computation success
def test_05_path_computation_rdm_bi(self):
response = test_utils.path_computation_request("request-1", "service-1",
- {"node-id": "ROADMA01", "service-rate": "100", "service-format": "Ethernet", "clli": "NodeA"},
- {"node-id": "ROADMC01", "service-rate": "100", "service-format": "Ethernet", "clli": "NodeC"})
+ {"node-id": "ROADMA01", "service-rate": "100",
+ "service-format": "Ethernet", "clli": "NodeA"},
+ {"node-id": "ROADMC01", "service-rate": "100", "service-format": "Ethernet", "clli": "NodeC"})
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Path is calculated',
# Path Computation success
def test_11_path_computation_xpdr_uni(self):
response = test_utils.path_computation_request("request-1", "service-1",
- {"node-id": "XPONDER-1-2", "service-rate": "100", "service-format": "Ethernet", "clli": "ORANGE1"},
- {"node-id": "XPONDER-3-2", "service-rate": "100", "service-format": "Ethernet", "clli": "ORANGE3"})
+ {"node-id": "XPONDER-1-2", "service-rate": "100",
+ "service-format": "Ethernet", "clli": "ORANGE1"},
+ {"node-id": "XPONDER-3-2", "service-rate": "100", "service-format": "Ethernet", "clli": "ORANGE3"})
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Path is calculated',
# Path Computation success
def test_12_path_computation_rdm_uni(self):
response = test_utils.path_computation_request("request1", "service1",
- {"service-rate": "100", "service-format": "Ethernet", "clli": "cll21", "node-id": "OpenROADM-2-1"},
- {"service-rate": "100", "service-format": "Ethernet", "clli": "ncli22", "node-id": "OpenROADM-2-2"})
+ {"service-rate": "100", "service-format": "Ethernet",
+ "clli": "cll21", "node-id": "OpenROADM-2-1"},
+ {"service-rate": "100", "service-format": "Ethernet", "clli": "ncli22", "node-id": "OpenROADM-2-2"})
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Path is calculated',
# Test failed path computation
def test_17_fail_path_computation(self):
response = test_utils.post_request(test_utils.URL_PATH_COMPUTATION_REQUEST,
- {"input": { "service-handler-header": { "request-id": "request-1" }}})
+ {"input": {"service-handler-header": {"request-id": "request-1"}}})
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Service Name is not set',
# Test1 success path computation
def test_18_success1_path_computation(self):
response = test_utils.path_computation_request("request1", "service1",
- {"service-format": "Ethernet","service-rate": "100", "clli": "ORANGE2", "node-id": "XPONDER-2-2",
- "tx-direction": {"port": {
- "port-device-name": "Some port-device-name",
- "port-type": "Some port-type",
- "port-name": "Some port-name",
- "port-rack": "Some port-rack",
- "port-shelf": "Some port-shelf",
- "port-slot": "Some port-slot",
- "port-sub-slot": "Some port-sub-slot"
- }},
- "rx-direction": {"port": {
- "port-device-name": "Some port-device-name",
- "port-type": "Some port-type",
- "port-name": "Some port-name",
- "port-rack": "Some port-rack",
- "port-shelf": "Some port-shelf",
- "port-slot": "Some port-slot",
- "port-sub-slot": "Some port-sub-slot"
- }}},
- {"service-format": "Ethernet", "service-rate": "100", "clli": "ORANGE1", "node-id": "XPONDER-1-2",
- "tx-direction": {"port": {
- "port-device-name": "Some port-device-name",
- "port-type": "Some port-type",
- "port-name": "Some port-name",
- "port-rack": "Some port-rack",
- "port-shelf": "Some port-shelf",
- "port-slot": "Some port-slot",
- "port-sub-slot": "Some port-sub-slot"
- }},
- "rx-direction": {"port": {
- "port-device-name": "Some port-device-name",
- "port-type": "Some port-type",
- "port-name": "Some port-name",
- "port-rack": "Some port-rack",
- "port-shelf": "Some port-shelf",
- "port-slot": "Some port-slot",
- "port-sub-slot": "Some port-sub-slot"
- }}},
- {"customer-code": ["Some customer-code"], "co-routing": {"existing-service": ["Some existing-service"]}},
- {"customer-code": ["Some customer-code"], "co-routing": {"existing-service": ["Some existing-service"]}},
- "hop-count", {"locally-protected-links": "true"})
+ {"service-format": "Ethernet", "service-rate": "100", "clli": "ORANGE2", "node-id": "XPONDER-2-2",
+ "tx-direction": {"port": {
+ "port-device-name": "Some port-device-name",
+ "port-type": "Some port-type",
+ "port-name": "Some port-name",
+ "port-rack": "Some port-rack",
+ "port-shelf": "Some port-shelf",
+ "port-slot": "Some port-slot",
+ "port-sub-slot": "Some port-sub-slot"
+ }},
+ "rx-direction": {"port": {
+ "port-device-name": "Some port-device-name",
+ "port-type": "Some port-type",
+ "port-name": "Some port-name",
+ "port-rack": "Some port-rack",
+ "port-shelf": "Some port-shelf",
+ "port-slot": "Some port-slot",
+ "port-sub-slot": "Some port-sub-slot"
+ }}},
+ {"service-format": "Ethernet", "service-rate": "100", "clli": "ORANGE1", "node-id": "XPONDER-1-2",
+ "tx-direction": {"port": {
+ "port-device-name": "Some port-device-name",
+ "port-type": "Some port-type",
+ "port-name": "Some port-name",
+ "port-rack": "Some port-rack",
+ "port-shelf": "Some port-shelf",
+ "port-slot": "Some port-slot",
+ "port-sub-slot": "Some port-sub-slot"
+ }},
+ "rx-direction": {"port": {
+ "port-device-name": "Some port-device-name",
+ "port-type": "Some port-type",
+ "port-name": "Some port-name",
+ "port-rack": "Some port-rack",
+ "port-shelf": "Some port-shelf",
+ "port-slot": "Some port-slot",
+ "port-sub-slot": "Some port-sub-slot"
+ }}},
+ {"customer-code": ["Some customer-code"],
+ "co-routing": {"existing-service": ["Some existing-service"]}},
+ {"customer-code": ["Some customer-code"],
+ "co-routing": {"existing-service": ["Some existing-service"]}},
+ "hop-count", {"locally-protected-links": "true"})
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Path is calculated',
# Test2 success path computation with path description
def test_19_success2_path_computation(self):
response = test_utils.path_computation_request("request 1", "service 1",
- {"service-rate": "100", "service-format": "Ethernet", "node-id": "XPONDER-1-2", "clli": "ORANGE1"},
- {"service-rate": "100", "service-format": "Ethernet", "node-id": "XPONDER-3-2", "clli": "ORANGE3"})
+ {"service-rate": "100", "service-format": "Ethernet",
+ "node-id": "XPONDER-1-2", "clli": "ORANGE1"},
+ {"service-rate": "100", "service-format": "Ethernet", "node-id": "XPONDER-3-2", "clli": "ORANGE3"})
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Path is calculated',
# Test3 success path computation with hard-constraints exclude
def test_20_success3_path_computation(self):
response = test_utils.path_computation_request("request 1", "service 1",
- {"service-rate": "100", "service-format": "Ethernet", "node-id": "XPONDER-1-2", "clli": "ORANGE1"},
- {"service-rate": "100", "service-format": "Ethernet", "node-id": "XPONDER-3-2", "clli": "ORANGE3"},
- {"exclude_": { "node-id": ["OpenROADM-2-1", "OpenROADM-2-2"] }})
+ {"service-rate": "100", "service-format": "Ethernet",
+ "node-id": "XPONDER-1-2", "clli": "ORANGE1"},
+ {"service-rate": "100", "service-format": "Ethernet",
+ "node-id": "XPONDER-3-2", "clli": "ORANGE3"},
+ {"exclude_": {"node-id": ["OpenROADM-2-1", "OpenROADM-2-2"]}})
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Path is calculated',
# Path computation before deleting oms-attribute of the link :openroadm1-3 to openroadm1-2
def test_21_path_computation_before_oms_attribute_deletion(self):
response = test_utils.path_computation_request("request 1", "service 1",
- {"service-rate": "100", "service-format": "Ethernet", "node-id": "XPONDER-2-2", "clli": "ORANGE2"},
- {"service-rate": "100", "service-format": "Ethernet", "node-id": "XPONDER-1-2", "clli": "ORANGE1"})
+ {"service-rate": "100", "service-format": "Ethernet",
+ "node-id": "XPONDER-2-2", "clli": "ORANGE2"},
+ {"service-rate": "100", "service-format": "Ethernet", "node-id": "XPONDER-1-2", "clli": "ORANGE1"})
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Path is calculated',
link = {"link-id": "OpenROADM-1-3-DEG2-to-OpenROADM-1-2-DEG2"}
find = False
for i in range(0, nbElmPath):
- resource_i = res['output']['response-parameters']['path-description']['aToZ-direction']['aToZ'][i]['resource']
+ resource_i = (res['output']['response-parameters']['path-description']['aToZ-direction']['aToZ'][i]
+ ['resource'])
if(resource_i == link):
find = True
self.assertEqual(find, True)
# Path computation after deleting oms-attribute of the link :openroadm1-3 to openroadm1-2
def test_23_path_computation_after_oms_attribute_deletion(self):
response = test_utils.path_computation_request("request 1", "service 1",
- {"service-rate": "100", "service-format": "Ethernet", "node-id": "XPONDER-2-2", "clli": "ORANGE2"},
- {"service-rate": "100", "service-format": "Ethernet", "node-id": "XPONDER-1-2", "clli": "ORANGE1"})
+ {"service-rate": "100", "service-format": "Ethernet",
+ "node-id": "XPONDER-2-2", "clli": "ORANGE2"},
+ {"service-rate": "100", "service-format": "Ethernet", "node-id": "XPONDER-1-2", "clli": "ORANGE1"})
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Path is calculated',
link = {"link-id": "OpenROADM-1-3-DEG2-to-OpenROADM-1-2-DEG2"}
find = False
for i in range(0, nbElmPath):
- resource_i = res['output']['response-parameters']['path-description']['aToZ-direction']['aToZ'][i]['resource']
+ resource_i = (res['output']['response-parameters']['path-description']['aToZ-direction']['aToZ'][i]
+ ['resource'])
if (resource_i == link):
find = True
self.assertNotEqual(find, True)
def test_05_service_path_create(self):
response = test_utils.service_path_request("create", "service_test", "7",
[{"renderer:node-id": "ROADMA01",
- "renderer:src-tp": "SRG1-PP7-TXRX", "renderer:dest-tp": "DEG1-TTP-TXRX"},
+ "renderer:src-tp": "SRG1-PP7-TXRX",
+ "renderer:dest-tp": "DEG1-TTP-TXRX"},
{"renderer:node-id": "XPDRA01",
- "renderer:src-tp": "XPDR1-CLIENT1", "renderer:dest-tp": "XPDR1-NETWORK1"}])
+ "renderer:src-tp": "XPDR1-CLIENT1",
+ "renderer:dest-tp": "XPDR1-NETWORK1"}])
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Roadm-connection successfully created for nodes: ROADMA01', res["output"]["result"])
def test_14_service_path_delete(self):
response = test_utils.service_path_request("delete", "service_test", "7",
[{"renderer:node-id": "ROADMA01",
- "renderer:src-tp": "SRG1-PP7-TXRX", "renderer:dest-tp": "DEG1-TTP-TXRX"},
+ "renderer:src-tp": "SRG1-PP7-TXRX",
+ "renderer:dest-tp": "DEG1-TTP-TXRX"},
{"renderer:node-id": "XPDRA01",
- "renderer:src-tp": "XPDR1-CLIENT1", "renderer:dest-tp": "XPDR1-NETWORK1"}])
+ "renderer:src-tp": "XPDR1-CLIENT1",
+ "renderer:dest-tp": "XPDR1-NETWORK1"}])
self.assertEqual(response.status_code, requests.codes.ok)
self.assertEqual(response.json(), {
'output': {'result': 'Request processed', 'success': True}})
res = response.json()
self.assertIn(
{"error-type": "application", "error-tag": "data-missing",
- "error-message": "Request could not be completed because the relevant data model content does not exist"},
+ "error-message":
+ "Request could not be completed because the relevant data model content does not exist"},
res['errors']['error'])
def test_17_service_path_delete_rdm_check(self):
res = response.json()
self.assertIn(
{"error-type": "application", "error-tag": "data-missing",
- "error-message": "Request could not be completed because the relevant data model content does not exist"},
+ "error-message":
+ "Request could not be completed because the relevant data model content does not exist"},
res['errors']['error'])
def test_18_service_path_delete_xpdr_check(self):
res = response.json()
self.assertIn(
{"error-type": "application", "error-tag": "data-missing",
- "error-message": "Request could not be completed because the relevant data model content does not exist"},
+ "error-message":
+ "Request could not be completed because the relevant data model content does not exist"},
res['errors']['error'])
def test_19_service_path_delete_xpdr_check(self):
res = response.json()
self.assertIn(
{"error-type": "application", "error-tag": "data-missing",
- "error-message": "Request could not be completed because the relevant data model content does not exist"},
+ "error-message":
+ "Request could not be completed because the relevant data model content does not exist"},
res['errors']['error'])
def test_20_service_path_delete_xpdr_check(self):
res = response.json()
self.assertIn(
{"error-type": "application", "error-tag": "data-missing",
- "error-message": "Request could not be completed because the relevant data model content does not exist"},
+ "error-message":
+ "Request could not be completed because the relevant data model content does not exist"},
res['errors']['error'])
def test_21_service_path_delete_xpdr_check(self):
res = response.json()
self.assertIn(
{"error-type": "application", "error-tag": "data-missing",
- "error-message": "Request could not be completed because the relevant data model content does not exist"},
+ "error-message":
+ "Request could not be completed because the relevant data model content does not exist"},
res['errors']['error'])
def test_22_service_path_delete_xpdr_check(self):
self.assertEqual(nbLink, 10)
expressLink = ['ROADMA01-DEG2-DEG2-CTP-TXRXtoROADMA01-DEG1-DEG1-CTP-TXRX',
'ROADMA01-DEG1-DEG1-CTP-TXRXtoROADMA01-DEG2-DEG2-CTP-TXRX']
- addLink = ['ROADMA01-SRG1-SRG1-CP-TXRXtoROADMA01-DEG2-DEG2-CTP-TXRX', 'ROADMA01-SRG1-SRG1-CP-TXRXtoROADMA01-DEG1-DEG1-CTP-TXRX',
- 'ROADMA01-SRG3-SRG3-CP-TXRXtoROADMA01-DEG2-DEG2-CTP-TXRX', 'ROADMA01-SRG3-SRG3-CP-TXRXtoROADMA01-DEG1-DEG1-CTP-TXRX']
- dropLink = ['ROADMA01-DEG1-DEG1-CTP-TXRXtoROADMA01-SRG1-SRG1-CP-TXRX', 'ROADMA01-DEG2-DEG2-CTP-TXRXtoROADMA01-SRG1-SRG1-CP-TXRX',
- 'ROADMA01-DEG1-DEG1-CTP-TXRXtoROADMA01-SRG3-SRG3-CP-TXRX', 'ROADMA01-DEG2-DEG2-CTP-TXRXtoROADMA01-SRG3-SRG3-CP-TXRX']
+ addLink = ['ROADMA01-SRG1-SRG1-CP-TXRXtoROADMA01-DEG2-DEG2-CTP-TXRX',
+ 'ROADMA01-SRG1-SRG1-CP-TXRXtoROADMA01-DEG1-DEG1-CTP-TXRX',
+ 'ROADMA01-SRG3-SRG3-CP-TXRXtoROADMA01-DEG2-DEG2-CTP-TXRX',
+ 'ROADMA01-SRG3-SRG3-CP-TXRXtoROADMA01-DEG1-DEG1-CTP-TXRX']
+ dropLink = ['ROADMA01-DEG1-DEG1-CTP-TXRXtoROADMA01-SRG1-SRG1-CP-TXRX',
+ 'ROADMA01-DEG2-DEG2-CTP-TXRXtoROADMA01-SRG1-SRG1-CP-TXRX',
+ 'ROADMA01-DEG1-DEG1-CTP-TXRXtoROADMA01-SRG3-SRG3-CP-TXRX',
+ 'ROADMA01-DEG2-DEG2-CTP-TXRXtoROADMA01-SRG3-SRG3-CP-TXRX']
for i in range(0, nbLink):
linkId = res['network'][0]['ietf-network-topology:link'][i]['link-id']
- if res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-common-network:link-type'] == 'EXPRESS-LINK':
+ if (res['network'][0]['ietf-network-topology:link'][i]
+ ['org-openroadm-common-network:link-type'] == 'EXPRESS-LINK'):
find = linkId in expressLink
self.assertEqual(find, True)
expressLink.remove(linkId)
- elif res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-common-network:link-type'] == 'ADD-LINK':
+ elif (res['network'][0]['ietf-network-topology:link'][i]
+ ['org-openroadm-common-network:link-type'] == 'ADD-LINK'):
find = linkId in addLink
self.assertEqual(find, True)
addLink.remove(linkId)
- elif res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-common-network:link-type'] == 'DROP-LINK':
+ elif (res['network'][0]['ietf-network-topology:link'][i]
+ ['org-openroadm-common-network:link-type'] == 'DROP-LINK'):
find = linkId in dropLink
self.assertEqual(find, True)
dropLink.remove(linkId)
client = 0
network = 0
for j in range(0, nbTps):
- tpType = res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['org-openroadm-common-network:tp-type']
+ tpType = res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j][
+ 'org-openroadm-common-network:tp-type']
tpId = res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['tp-id']
if (tpType == 'XPONDER-CLIENT'):
client += 1
self.assertEqual(nbLink, 12)
expressLink = ['ROADMA01-DEG2-DEG2-CTP-TXRXtoROADMA01-DEG1-DEG1-CTP-TXRX',
'ROADMA01-DEG1-DEG1-CTP-TXRXtoROADMA01-DEG2-DEG2-CTP-TXRX']
- addLink = ['ROADMA01-SRG1-SRG1-CP-TXRXtoROADMA01-DEG2-DEG2-CTP-TXRX', 'ROADMA01-SRG1-SRG1-CP-TXRXtoROADMA01-DEG1-DEG1-CTP-TXRX',
- 'ROADMA01-SRG3-SRG3-CP-TXRXtoROADMA01-DEG2-DEG2-CTP-TXRX', 'ROADMA01-SRG3-SRG3-CP-TXRXtoROADMA01-DEG1-DEG1-CTP-TXRX']
- dropLink = ['ROADMA01-DEG1-DEG1-CTP-TXRXtoROADMA01-SRG1-SRG1-CP-TXRX', 'ROADMA01-DEG2-DEG2-CTP-TXRXtoROADMA01-SRG1-SRG1-CP-TXRX',
- 'ROADMA01-DEG1-DEG1-CTP-TXRXtoROADMA01-SRG3-SRG3-CP-TXRX', 'ROADMA01-DEG2-DEG2-CTP-TXRXtoROADMA01-SRG3-SRG3-CP-TXRX']
+ addLink = ['ROADMA01-SRG1-SRG1-CP-TXRXtoROADMA01-DEG2-DEG2-CTP-TXRX',
+ 'ROADMA01-SRG1-SRG1-CP-TXRXtoROADMA01-DEG1-DEG1-CTP-TXRX',
+ 'ROADMA01-SRG3-SRG3-CP-TXRXtoROADMA01-DEG2-DEG2-CTP-TXRX',
+ 'ROADMA01-SRG3-SRG3-CP-TXRXtoROADMA01-DEG1-DEG1-CTP-TXRX']
+ dropLink = ['ROADMA01-DEG1-DEG1-CTP-TXRXtoROADMA01-SRG1-SRG1-CP-TXRX',
+ 'ROADMA01-DEG2-DEG2-CTP-TXRXtoROADMA01-SRG1-SRG1-CP-TXRX',
+ 'ROADMA01-DEG1-DEG1-CTP-TXRXtoROADMA01-SRG3-SRG3-CP-TXRX',
+ 'ROADMA01-DEG2-DEG2-CTP-TXRXtoROADMA01-SRG3-SRG3-CP-TXRX']
XPDR_IN = ['ROADMA01-SRG1-SRG1-PP1-TXRXtoXPDRA01-XPDR1-XPDR1-NETWORK1']
XPDR_OUT = ['XPDRA01-XPDR1-XPDR1-NETWORK1toROADMA01-SRG1-SRG1-PP1-TXRX']
for i in range(0, nbLink):
# Tests related to links
nbLink = len(res['network'][0]['ietf-network-topology:link'])
self.assertEqual(nbLink, 20)
- expressLink = ['ROADMA01-DEG2-DEG2-CTP-TXRXtoROADMA01-DEG1-DEG1-CTP-TXRX', 'ROADMA01-DEG1-DEG1-CTP-TXRXtoROADMA01-DEG2-DEG2-CTP-TXRX',
- 'ROADMC01-DEG2-DEG2-CTP-TXRXtoROADMC01-DEG1-DEG1-CTP-TXRX', 'ROADMC01-DEG1-DEG1-CTP-TXRXtoROADMC01-DEG2-DEG2-CTP-TXRX']
- addLink = ['ROADMA01-SRG1-SRG1-CP-TXRXtoROADMA01-DEG2-DEG2-CTP-TXRX', 'ROADMA01-SRG1-SRG1-CP-TXRXtoROADMA01-DEG1-DEG1-CTP-TXRX',
- 'ROADMA01-SRG3-SRG3-CP-TXRXtoROADMA01-DEG2-DEG2-CTP-TXRX', 'ROADMA01-SRG3-SRG3-CP-TXRXtoROADMA01-DEG1-DEG1-CTP-TXRX',
- 'ROADMC01-SRG1-SRG1-CP-TXRXtoROADMC01-DEG2-DEG2-CTP-TXRX', 'ROADMC01-SRG1-SRG1-CP-TXRXtoROADMC01-DEG1-DEG1-CTP-TXRX']
- dropLink = ['ROADMA01-DEG1-DEG1-CTP-TXRXtoROADMA01-SRG1-SRG1-CP-TXRX', 'ROADMA01-DEG2-DEG2-CTP-TXRXtoROADMA01-SRG1-SRG1-CP-TXRX',
- 'ROADMA01-DEG1-DEG1-CTP-TXRXtoROADMA01-SRG3-SRG3-CP-TXRX', 'ROADMA01-DEG2-DEG2-CTP-TXRXtoROADMA01-SRG3-SRG3-CP-TXRX',
- 'ROADMC01-DEG1-DEG1-CTP-TXRXtoROADMC01-SRG1-SRG1-CP-TXRX', 'ROADMC01-DEG2-DEG2-CTP-TXRXtoROADMC01-SRG1-SRG1-CP-TXRX']
+ expressLink = ['ROADMA01-DEG2-DEG2-CTP-TXRXtoROADMA01-DEG1-DEG1-CTP-TXRX',
+ 'ROADMA01-DEG1-DEG1-CTP-TXRXtoROADMA01-DEG2-DEG2-CTP-TXRX',
+ 'ROADMC01-DEG2-DEG2-CTP-TXRXtoROADMC01-DEG1-DEG1-CTP-TXRX',
+ 'ROADMC01-DEG1-DEG1-CTP-TXRXtoROADMC01-DEG2-DEG2-CTP-TXRX']
+ addLink = ['ROADMA01-SRG1-SRG1-CP-TXRXtoROADMA01-DEG2-DEG2-CTP-TXRX',
+ 'ROADMA01-SRG1-SRG1-CP-TXRXtoROADMA01-DEG1-DEG1-CTP-TXRX',
+ 'ROADMA01-SRG3-SRG3-CP-TXRXtoROADMA01-DEG2-DEG2-CTP-TXRX',
+ 'ROADMA01-SRG3-SRG3-CP-TXRXtoROADMA01-DEG1-DEG1-CTP-TXRX',
+ 'ROADMC01-SRG1-SRG1-CP-TXRXtoROADMC01-DEG2-DEG2-CTP-TXRX',
+ 'ROADMC01-SRG1-SRG1-CP-TXRXtoROADMC01-DEG1-DEG1-CTP-TXRX']
+ dropLink = ['ROADMA01-DEG1-DEG1-CTP-TXRXtoROADMA01-SRG1-SRG1-CP-TXRX',
+ 'ROADMA01-DEG2-DEG2-CTP-TXRXtoROADMA01-SRG1-SRG1-CP-TXRX',
+ 'ROADMA01-DEG1-DEG1-CTP-TXRXtoROADMA01-SRG3-SRG3-CP-TXRX',
+ 'ROADMA01-DEG2-DEG2-CTP-TXRXtoROADMA01-SRG3-SRG3-CP-TXRX',
+ 'ROADMC01-DEG1-DEG1-CTP-TXRXtoROADMC01-SRG1-SRG1-CP-TXRX',
+ 'ROADMC01-DEG2-DEG2-CTP-TXRXtoROADMC01-SRG1-SRG1-CP-TXRX']
R2RLink = ['ROADMA01-DEG1-DEG1-TTP-TXRXtoROADMC01-DEG2-DEG2-TTP-TXRX',
'ROADMC01-DEG2-DEG2-TTP-TXRXtoROADMA01-DEG1-DEG1-TTP-TXRX']
XPDR_IN = ['ROADMA01-SRG1-SRG1-PP1-TXRXtoXPDRA01-XPDR1-XPDR1-NETWORK1']
link_id = res['network'][0]['ietf-network-topology:link'][i]['link-id']
if(link_id in R2RLink):
find = False
- spanLoss = res['network'][0]['ietf-network-topology:link'][i][
- 'org-openroadm-network-topology:OMS-attributes']['span']["engineered-spanloss"]
- length = res['network'][0]['ietf-network-topology:link'][i][
- 'org-openroadm-network-topology:OMS-attributes']['span']['link-concatenation'][0]['SRLG-length']
+ spanLoss = (res['network'][0]['ietf-network-topology:link'][i]
+ ['org-openroadm-network-topology:OMS-attributes']['span']["engineered-spanloss"])
+ length = (res['network'][0]['ietf-network-topology:link'][i]
+ ['org-openroadm-network-topology:OMS-attributes']['span']['link-concatenation'][0]
+ ['SRLG-length'])
if((spanLoss != None) & (length != None)):
find = True
self.assertTrue(find)
client = 0
network = 0
for j in range(0, nbTps):
- tpType = res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['org-openroadm-common-network:tp-type']
+ tpType = res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j][
+ 'org-openroadm-common-network:tp-type']
if (tpType == 'XPONDER-CLIENT'):
client += 1
elif (tpType == 'XPONDER-NETWORK'):
res = response.json()
# Tests related to links
nbLink = len(res['network'][0]['ietf-network-topology:link'])
- listR2RLink = ['ROADMA01-DEG1-DEG1-TTP-TXRXtoROADMC01-DEG2-DEG2-TTP-TXRX', 'ROADMC01-DEG2-DEG2-TTP-TXRXtoROADMA01-DEG1-DEG1-TTP-TXRX',
- 'ROADMA01-DEG2-DEG2-TTP-TXRXtoROADMB01-DEG1-DEG1-TTP-TXRX', 'ROADMC01-DEG1-DEG1-TTP-TXRXtoROADMB01-DEG2-DEG2-TTP-TXRX',
- 'ROADMB01-DEG1-DEG1-TTP-TXRXtoROADMA01-DEG2-DEG2-TTP-TXRX', 'ROADMB01-DEG2-DEG2-TTP-TXRXtoROADMC01-DEG1-DEG1-TTP-TXRX']
+ listR2RLink = ['ROADMA01-DEG1-DEG1-TTP-TXRXtoROADMC01-DEG2-DEG2-TTP-TXRX',
+ 'ROADMC01-DEG2-DEG2-TTP-TXRXtoROADMA01-DEG1-DEG1-TTP-TXRX',
+ 'ROADMA01-DEG2-DEG2-TTP-TXRXtoROADMB01-DEG1-DEG1-TTP-TXRX',
+ 'ROADMC01-DEG1-DEG1-TTP-TXRXtoROADMB01-DEG2-DEG2-TTP-TXRX',
+ 'ROADMB01-DEG1-DEG1-TTP-TXRXtoROADMA01-DEG2-DEG2-TTP-TXRX',
+ 'ROADMB01-DEG2-DEG2-TTP-TXRXtoROADMC01-DEG1-DEG1-TTP-TXRX']
for i in range(0, nbLink):
- if res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-common-network:link-type'] == 'ROADM-TO-ROADM':
+ if (res['network'][0]['ietf-network-topology:link'][i]
+ ['org-openroadm-common-network:link-type'] == 'ROADM-TO-ROADM'):
link_id = res['network'][0]['ietf-network-topology:link'][i]['link-id']
find = link_id in listR2RLink
self.assertEqual(find, True)
link_type = res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-common-network:link-type']
link_src = res['network'][0]['ietf-network-topology:link'][i]['source']['source-node']
link_dest = res['network'][0]['ietf-network-topology:link'][i]['destination']['dest-node']
- oppLink_id = res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-common-network:opposite-link']
+ oppLink_id = (res['network'][0]['ietf-network-topology:link'][i]
+ ['org-openroadm-common-network:opposite-link'])
# Find the opposite link
response_oppLink = test_utils.get_ordm_topo_request("ietf-network-topology:link/"+oppLink_id)
self.assertEqual(response_oppLink.status_code, requests.codes.ok)
self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
['org-openroadm-common-network:tp-type'], 'XPONDER-NETWORK')
self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
- ['org-openroadm-network-topology:xpdr-network-attributes']['tail-equipment-id'],
+ ['org-openroadm-network-topology:xpdr-network-attributes']
+ ['tail-equipment-id'],
'ROADMA01-SRG1--SRG1-PP1-TXRX')
self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'XPDRA01'},
res['network'][0]['node'][i]['supporting-node'])
self.assertEqual(nbLink, 16)
expressLink = ['ROADMA01-DEG2-DEG2-CTP-TXRXtoROADMA01-DEG1-DEG1-CTP-TXRX',
'ROADMA01-DEG1-DEG1-CTP-TXRXtoROADMA01-DEG2-DEG2-CTP-TXRX']
- addLink = ['ROADMA01-SRG1-SRG1-CP-TXRXtoROADMA01-DEG2-DEG2-CTP-TXRX', 'ROADMA01-SRG1-SRG1-CP-TXRXtoROADMA01-DEG1-DEG1-CTP-TXRX',
- 'ROADMA01-SRG3-SRG3-CP-TXRXtoROADMA01-DEG2-DEG2-CTP-TXRX', 'ROADMA01-SRG3-SRG3-CP-TXRXtoROADMA01-DEG1-DEG1-CTP-TXRX']
- dropLink = ['ROADMA01-DEG1-DEG1-CTP-TXRXtoROADMA01-SRG1-SRG1-CP-TXRX', 'ROADMA01-DEG2-DEG2-CTP-TXRXtoROADMA01-SRG1-SRG1-CP-TXRX',
- 'ROADMA01-DEG1-DEG1-CTP-TXRXtoROADMA01-SRG3-SRG3-CP-TXRX', 'ROADMA01-DEG2-DEG2-CTP-TXRXtoROADMA01-SRG3-SRG3-CP-TXRX']
+ addLink = ['ROADMA01-SRG1-SRG1-CP-TXRXtoROADMA01-DEG2-DEG2-CTP-TXRX',
+ 'ROADMA01-SRG1-SRG1-CP-TXRXtoROADMA01-DEG1-DEG1-CTP-TXRX',
+ 'ROADMA01-SRG3-SRG3-CP-TXRXtoROADMA01-DEG2-DEG2-CTP-TXRX',
+ 'ROADMA01-SRG3-SRG3-CP-TXRXtoROADMA01-DEG1-DEG1-CTP-TXRX']
+ dropLink = ['ROADMA01-DEG1-DEG1-CTP-TXRXtoROADMA01-SRG1-SRG1-CP-TXRX',
+ 'ROADMA01-DEG2-DEG2-CTP-TXRXtoROADMA01-SRG1-SRG1-CP-TXRX',
+ 'ROADMA01-DEG1-DEG1-CTP-TXRXtoROADMA01-SRG3-SRG3-CP-TXRX',
+ 'ROADMA01-DEG2-DEG2-CTP-TXRXtoROADMA01-SRG3-SRG3-CP-TXRX']
roadmtoroadmLink = 0
for i in range(0, nbLink):
- if (res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-common-network:link-type'] == 'EXPRESS-LINK'):
+ if (res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-common-network:link-type']
+ == 'EXPRESS-LINK'):
link_id = res['network'][0]['ietf-network-topology:link'][i]['link-id']
find = link_id in expressLink
self.assertEqual(find, True)
expressLink.remove(link_id)
- elif (res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-common-network:link-type'] == 'ADD-LINK'):
+ elif (res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-common-network:link-type']
+ == 'ADD-LINK'):
link_id = res['network'][0]['ietf-network-topology:link'][i]['link-id']
find = link_id in addLink
self.assertEqual(find, True)
addLink.remove(link_id)
- elif (res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-common-network:link-type'] == 'DROP-LINK'):
+ elif (res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-common-network:link-type']
+ == 'DROP-LINK'):
link_id = res['network'][0]['ietf-network-topology:link'][i]['link-id']
find = link_id in dropLink
self.assertEqual(find, True)
self.assertEqual(response.status_code, requests.codes.ok)
res_mapping = (response.json())['mapping'][0]
self.assertSetEqual({'org-openroadm-port-types:if-10GE-ODU2e',
- 'org-openroadm-port-types:if-10GE-ODU2',
- 'org-openroadm-port-types:if-10GE'},
- set(res_mapping['supported-interface-capability']))
+ 'org-openroadm-port-types:if-10GE-ODU2',
+ 'org-openroadm-port-types:if-10GE'},
+ set(res_mapping['supported-interface-capability']))
self.assertDictEqual(
dict(
{'supporting-port': 'CP1-SFP4-P1',
response = test_utils.otn_service_path_request("create", "service1", "10G", "Ethernet",
[{"node-id": "SPDR-SA1", "client-tp": "XPDR1-CLIENT1",
"network-tp": "XPDR1-NETWORK1"}],
- {"ethernet-encoding": "eth encode", "trib-slot": ["1"], "trib-port-number": "1"})
+ {"ethernet-encoding": "eth encode",
+ "trib-slot": ["1"], "trib-port-number": "1"})
time.sleep(3)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
response = test_utils.otn_service_path_request("delete", "service1", "10G", "Ethernet",
[{"node-id": "SPDR-SA1", "client-tp": "XPDR1-CLIENT1",
"network-tp": "XPDR1-NETWORK1"}],
- {"ethernet-encoding": "eth encode", "trib-slot": ["1"], "trib-port-number": "1"})
+ {"ethernet-encoding": "eth encode",
+ "trib-slot": ["1"], "trib-port-number": "1"})
time.sleep(3)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
client = 0
network = 0
for j in range(0, nbTps):
- tpType = res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['org-openroadm-common-network:tp-type']
+ tpType = (res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
+ ['org-openroadm-common-network:tp-type'])
if (tpType == 'XPONDER-CLIENT'):
client += 1
elif (tpType == 'XPONDER-NETWORK'):
client = 0
network = 0
for j in range(0, nbTps):
- tpType = res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['org-openroadm-common-network:tp-type']
+ tpType = (res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
+ ['org-openroadm-common-network:tp-type'])
if (tpType == 'XPONDER-CLIENT'):
client += 1
elif (tpType == 'XPONDER-NETWORK'):
client = 0
network = 0
for j in range(0, nbTps):
- tpType = res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['org-openroadm-common-network:tp-type']
+ tpType = (res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
+ ['org-openroadm-common-network:tp-type'])
tpId = res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['tp-id']
if (tpType == 'XPONDER-CLIENT'):
client += 1
- self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['org-openroadm-otn-network-topology:tp-supported-interfaces']
- ['supported-interface-capability'][0]['if-cap-type'], 'org-openroadm-port-types:if-10GE-ODU2e')
- self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
- ['org-openroadm-otn-network-topology:xpdr-tp-port-connection-attributes']['rate'], 'org-openroadm-otn-common-types:ODU2e')
+ self.assertEqual((res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
+ ['org-openroadm-otn-network-topology:tp-supported-interfaces']
+ ['supported-interface-capability'][0]['if-cap-type']),
+ 'org-openroadm-port-types:if-10GE-ODU2e')
+ self.assertEqual((res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
+ ['org-openroadm-otn-network-topology:xpdr-tp-port-connection-attributes']
+ ['rate']),
+ 'org-openroadm-otn-common-types:ODU2e')
elif (tpType == 'XPONDER-NETWORK'):
network += 1
- self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['org-openroadm-otn-network-topology:tp-supported-interfaces']
- ['supported-interface-capability'][0]['if-cap-type'], 'org-openroadm-port-types:if-OCH-OTU4-ODU4')
- self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
- ['org-openroadm-otn-network-topology:xpdr-tp-port-connection-attributes']['rate'], 'org-openroadm-otn-common-types:ODU4')
- self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point']
- [j]['supporting-termination-point'][0]['network-ref'], 'openroadm-topology')
- self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point']
- [j]['supporting-termination-point'][0]['node-ref'], 'SPDR-SA1-XPDR1')
- self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point']
- [j]['supporting-termination-point'][0]['tp-ref'], 'XPDR1-NETWORK1')
+ self.assertEqual((res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
+ ['org-openroadm-otn-network-topology:tp-supported-interfaces']
+ ['supported-interface-capability'][0]['if-cap-type']),
+ 'org-openroadm-port-types:if-OCH-OTU4-ODU4')
+ self.assertEqual((res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
+ ['org-openroadm-otn-network-topology:xpdr-tp-port-connection-attributes']
+ ['rate']),
+ 'org-openroadm-otn-common-types:ODU4')
+ self.assertEqual((res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
+ ['supporting-termination-point'][0]['network-ref']), 'openroadm-topology')
+ self.assertEqual((res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
+ ['supporting-termination-point'][0]['node-ref']), 'SPDR-SA1-XPDR1')
+ self.assertEqual((res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
+ ['supporting-termination-point'][0]['tp-ref']), 'XPDR1-NETWORK1')
self.assertTrue(client == 4)
self.assertTrue(network == 1)
listNode.remove(nodeId)
['odu-switching-pools'][0]['non-blocking-list'])
self.assertEqual(nbNbl, 4)
for k in range(0, nbNbl):
- nbl = res['network'][0]['node'][i]['org-openroadm-otn-network-topology:switching-pools']['odu-switching-pools'][0]['non-blocking-list'][k]
+ nbl = (res['network'][0]['node'][i]['org-openroadm-otn-network-topology:switching-pools']
+ ['odu-switching-pools'][0]['non-blocking-list'][k])
if (nbl['nbl-number'] == 1):
self.assertEqual(nbl['available-interconnect-bandwidth'], 10)
self.assertEqual(nbl['interconnect-bandwidth-unit'], 1000000000)
client = 0
network = 0
for j in range(0, nbTps):
- tpType = res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['org-openroadm-common-network:tp-type']
+ tpType = (res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
+ ['org-openroadm-common-network:tp-type'])
tpId = res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['tp-id']
if (tpType == 'XPONDER-CLIENT'):
client += 1
- self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['org-openroadm-otn-network-topology:tp-supported-interfaces']
- ['supported-interface-capability'][0]['if-cap-type'], 'org-openroadm-port-types:if-100GE-ODU4')
- self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
- ['org-openroadm-otn-network-topology:xpdr-tp-port-connection-attributes']['rate'], 'org-openroadm-otn-common-types:ODU4')
+ self.assertEqual((res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
+ ['org-openroadm-otn-network-topology:tp-supported-interfaces']
+ ['supported-interface-capability'][0]['if-cap-type']),
+ 'org-openroadm-port-types:if-100GE-ODU4')
+ self.assertEqual((res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
+ ['org-openroadm-otn-network-topology:xpdr-tp-port-connection-attributes']
+ ['rate']),
+ 'org-openroadm-otn-common-types:ODU4')
elif (tpType == 'XPONDER-NETWORK'):
network += 1
- self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['org-openroadm-otn-network-topology:tp-supported-interfaces']
- ['supported-interface-capability'][0]['if-cap-type'], 'org-openroadm-port-types:if-OCH-OTU4-ODU4')
- self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
- ['org-openroadm-otn-network-topology:xpdr-tp-port-connection-attributes']['rate'], 'org-openroadm-otn-common-types:ODU4')
- self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point']
- [j]['supporting-termination-point'][0]['network-ref'], 'openroadm-topology')
- self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point']
- [j]['supporting-termination-point'][0]['node-ref'], 'SPDR-SA1-XPDR2')
- self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point']
- [j]['supporting-termination-point'][0]['tp-ref'], tpId)
+ self.assertEqual((res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
+ ['org-openroadm-otn-network-topology:tp-supported-interfaces']
+ ['supported-interface-capability'][0]['if-cap-type']),
+ 'org-openroadm-port-types:if-OCH-OTU4-ODU4')
+ self.assertEqual((res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
+ ['org-openroadm-otn-network-topology:xpdr-tp-port-connection-attributes']
+ ['rate']),
+ 'org-openroadm-otn-common-types:ODU4')
+ self.assertEqual((res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
+ ['supporting-termination-point'][0]['network-ref']), 'openroadm-topology')
+ self.assertEqual((res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
+ ['supporting-termination-point'][0]['node-ref']), 'SPDR-SA1-XPDR2')
+ self.assertEqual((res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
+ ['supporting-termination-point'][0]['tp-ref']), tpId)
self.assertTrue(client == 4)
self.assertTrue(network == 4)
listNode.remove(nodeId)
nbNbl = len(res['network'][0]['node'][i]['org-openroadm-otn-network-topology:switching-pools']
['odu-switching-pools'][0]['non-blocking-list'])
self.assertEqual(nbNbl, 1)
- nbl = res['network'][0]['node'][i]['org-openroadm-otn-network-topology:switching-pools']['odu-switching-pools'][0]['non-blocking-list'][0]
+ nbl = (res['network'][0]['node'][i]['org-openroadm-otn-network-topology:switching-pools']
+ ['odu-switching-pools'][0]['non-blocking-list'][0])
self.assertIn('XPDR2-NETWORK1', nbl['tp-list'])
self.assertIn('XPDR2-CLIENT1', nbl['tp-list'])
self.assertIn('XPDR2-NETWORK2', nbl['tp-list'])
def test_05_service_path_create(self):
response = test_utils.service_path_request("create", "service_test", "7",
[{"renderer:node-id": "ROADM-A1",
- "renderer:src-tp": "SRG1-PP3-TXRX", "renderer:dest-tp": "DEG1-TTP-TXRX"},
+ "renderer:src-tp": "SRG1-PP3-TXRX",
+ "renderer:dest-tp": "DEG1-TTP-TXRX"},
{"renderer:node-id": "XPDR-A1",
- "renderer:src-tp": "XPDR1-CLIENT1", "renderer:dest-tp": "XPDR1-NETWORK1"}])
+ "renderer:src-tp": "XPDR1-CLIENT1",
+ "renderer:dest-tp": "XPDR1-NETWORK1"}])
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Roadm-connection successfully created for nodes: ROADM-A1', res["output"]["result"])
def test_17_service_path_delete(self):
response = test_utils.service_path_request("delete", "service_test", "7",
[{"renderer:node-id": "ROADM-A1",
- "renderer:src-tp": "SRG1-PP3-TXRX", "renderer:dest-tp": "DEG1-TTP-TXRX"},
+ "renderer:src-tp": "SRG1-PP3-TXRX",
+ "renderer:dest-tp": "DEG1-TTP-TXRX"},
{"renderer:node-id": "XPDR-A1",
- "renderer:src-tp": "XPDR1-CLIENT1", "renderer:dest-tp": "XPDR1-NETWORK1"}])
+ "renderer:src-tp": "XPDR1-CLIENT1",
+ "renderer:dest-tp": "XPDR1-NETWORK1"}])
self.assertEqual(response.status_code, requests.codes.ok)
self.assertEqual(response.json(), {
'output': {'result': 'Request processed', 'success': True}})
'ROADM-A1-DEG2-DEG2-CTP-TXRXtoROADM-A1-SRG3-SRG3-CP-TXRX']
for i in range(0, nbLink):
linkId = res['network'][0]['ietf-network-topology:link'][i]['link-id']
- if (res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-common-network:link-type'] == 'EXPRESS-LINK'):
+ if (res['network'][0]['ietf-network-topology:link'][i]
+ ['org-openroadm-common-network:link-type'] == 'EXPRESS-LINK'):
find = linkId in expressLink
self.assertEqual(find, True)
expressLink.remove(linkId)
- elif (res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-common-network:link-type'] == 'ADD-LINK'):
+ elif (res['network'][0]['ietf-network-topology:link'][i]
+ ['org-openroadm-common-network:link-type'] == 'ADD-LINK'):
find = linkId in addLink
self.assertEqual(find, True)
addLink.remove(linkId)
- elif (res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-common-network:link-type'] == 'DROP-LINK'):
+ elif (res['network'][0]['ietf-network-topology:link'][i]
+ ['org-openroadm-common-network:link-type'] == 'DROP-LINK'):
find = linkId in dropLink
self.assertEqual(find, True)
dropLink.remove(linkId)
client = 0
network = 0
for j in range(0, nbTps):
- tpType = res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['org-openroadm-common-network:tp-type']
+ tpType = (res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
+ ['org-openroadm-common-network:tp-type'])
tpId = res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['tp-id']
if (tpType == 'XPONDER-CLIENT'):
client += 1
def test_10_connect_tail_xpdr_rdm(self):
# Connect the tail: XPDRA to ROADMA
response = test_utils.connect_xpdr_to_rdm_request("XPDR-A1", "1", "1",
- "ROADM-A1", "1", "SRG1-PP1-TXRX")
+ "ROADM-A1", "1", "SRG1-PP1-TXRX")
self.assertEqual(response.status_code, requests.codes.ok)
def test_11_connect_tail_rdm_xpdr(self):
response = test_utils.connect_rdm_to_xpdr_request("XPDR-A1", "1", "1",
- "ROADM-A1", "1", "SRG1-PP1-TXRX")
+ "ROADM-A1", "1", "SRG1-PP1-TXRX")
self.assertEqual(response.status_code, requests.codes.ok)
def test_12_getLinks_OpenRoadmTopology(self):
self.assertEqual(nbLink, 12)
expressLink = ['ROADM-A1-DEG2-DEG2-CTP-TXRXtoROADM-A1-DEG1-DEG1-CTP-TXRX',
'ROADM-A1-DEG1-DEG1-CTP-TXRXtoROADM-A1-DEG2-DEG2-CTP-TXRX']
- addLink = ['ROADM-A1-SRG1-SRG1-CP-TXRXtoROADM-A1-DEG2-DEG2-CTP-TXRX', 'ROADM-A1-SRG1-SRG1-CP-TXRXtoROADM-A1-DEG1-DEG1-CTP-TXRX',
- 'ROADM-A1-SRG3-SRG3-CP-TXRXtoROADM-A1-DEG2-DEG2-CTP-TXRX', 'ROADM-A1-SRG3-SRG3-CP-TXRXtoROADM-A1-DEG1-DEG1-CTP-TXRX']
- dropLink = ['ROADM-A1-DEG1-DEG1-CTP-TXRXtoROADM-A1-SRG1-SRG1-CP-TXRX', 'ROADM-A1-DEG2-DEG2-CTP-TXRXtoROADM-A1-SRG1-SRG1-CP-TXRX',
- 'ROADM-A1-DEG1-DEG1-CTP-TXRXtoROADM-A1-SRG3-SRG3-CP-TXRX', 'ROADM-A1-DEG2-DEG2-CTP-TXRXtoROADM-A1-SRG3-SRG3-CP-TXRX']
+ addLink = ['ROADM-A1-SRG1-SRG1-CP-TXRXtoROADM-A1-DEG2-DEG2-CTP-TXRX',
+ 'ROADM-A1-SRG1-SRG1-CP-TXRXtoROADM-A1-DEG1-DEG1-CTP-TXRX',
+ 'ROADM-A1-SRG3-SRG3-CP-TXRXtoROADM-A1-DEG2-DEG2-CTP-TXRX',
+ 'ROADM-A1-SRG3-SRG3-CP-TXRXtoROADM-A1-DEG1-DEG1-CTP-TXRX']
+ dropLink = ['ROADM-A1-DEG1-DEG1-CTP-TXRXtoROADM-A1-SRG1-SRG1-CP-TXRX',
+ 'ROADM-A1-DEG2-DEG2-CTP-TXRXtoROADM-A1-SRG1-SRG1-CP-TXRX',
+ 'ROADM-A1-DEG1-DEG1-CTP-TXRXtoROADM-A1-SRG3-SRG3-CP-TXRX',
+ 'ROADM-A1-DEG2-DEG2-CTP-TXRXtoROADM-A1-SRG3-SRG3-CP-TXRX']
XPDR_IN = ['ROADM-A1-SRG1-SRG1-PP1-TXRXtoXPDR-A1-XPDR1-XPDR1-NETWORK1']
XPDR_OUT = ['XPDR-A1-XPDR1-XPDR1-NETWORK1toROADM-A1-SRG1-SRG1-PP1-TXRX']
for i in range(0, nbLink):
# Tests related to links
nbLink = len(res['network'][0]['ietf-network-topology:link'])
self.assertEqual(nbLink, 20)
- expressLink = ['ROADM-A1-DEG2-DEG2-CTP-TXRXtoROADM-A1-DEG1-DEG1-CTP-TXRX', 'ROADM-A1-DEG1-DEG1-CTP-TXRXtoROADM-A1-DEG2-DEG2-CTP-TXRX',
- 'ROADM-C1-DEG2-DEG2-CTP-TXRXtoROADM-C1-DEG1-DEG1-CTP-TXRX', 'ROADM-C1-DEG1-DEG1-CTP-TXRXtoROADM-C1-DEG2-DEG2-CTP-TXRX']
- addLink = ['ROADM-A1-SRG1-SRG1-CP-TXRXtoROADM-A1-DEG2-DEG2-CTP-TXRX', 'ROADM-A1-SRG1-SRG1-CP-TXRXtoROADM-A1-DEG1-DEG1-CTP-TXRX',
- 'ROADM-A1-SRG3-SRG3-CP-TXRXtoROADM-A1-DEG2-DEG2-CTP-TXRX', 'ROADM-A1-SRG3-SRG3-CP-TXRXtoROADM-A1-DEG1-DEG1-CTP-TXRX',
- 'ROADM-C1-SRG1-SRG1-CP-TXRXtoROADM-C1-DEG2-DEG2-CTP-TXRX', 'ROADM-C1-SRG1-SRG1-CP-TXRXtoROADM-C1-DEG1-DEG1-CTP-TXRX']
- dropLink = ['ROADM-A1-DEG1-DEG1-CTP-TXRXtoROADM-A1-SRG1-SRG1-CP-TXRX', 'ROADM-A1-DEG2-DEG2-CTP-TXRXtoROADM-A1-SRG1-SRG1-CP-TXRX',
- 'ROADM-A1-DEG1-DEG1-CTP-TXRXtoROADM-A1-SRG3-SRG3-CP-TXRX', 'ROADM-A1-DEG2-DEG2-CTP-TXRXtoROADM-A1-SRG3-SRG3-CP-TXRX',
- 'ROADM-C1-DEG1-DEG1-CTP-TXRXtoROADM-C1-SRG1-SRG1-CP-TXRX', 'ROADM-C1-DEG2-DEG2-CTP-TXRXtoROADM-C1-SRG1-SRG1-CP-TXRX']
+ expressLink = ['ROADM-A1-DEG2-DEG2-CTP-TXRXtoROADM-A1-DEG1-DEG1-CTP-TXRX',
+ 'ROADM-A1-DEG1-DEG1-CTP-TXRXtoROADM-A1-DEG2-DEG2-CTP-TXRX',
+ 'ROADM-C1-DEG2-DEG2-CTP-TXRXtoROADM-C1-DEG1-DEG1-CTP-TXRX',
+ 'ROADM-C1-DEG1-DEG1-CTP-TXRXtoROADM-C1-DEG2-DEG2-CTP-TXRX']
+ addLink = ['ROADM-A1-SRG1-SRG1-CP-TXRXtoROADM-A1-DEG2-DEG2-CTP-TXRX',
+ 'ROADM-A1-SRG1-SRG1-CP-TXRXtoROADM-A1-DEG1-DEG1-CTP-TXRX',
+ 'ROADM-A1-SRG3-SRG3-CP-TXRXtoROADM-A1-DEG2-DEG2-CTP-TXRX',
+ 'ROADM-A1-SRG3-SRG3-CP-TXRXtoROADM-A1-DEG1-DEG1-CTP-TXRX',
+ 'ROADM-C1-SRG1-SRG1-CP-TXRXtoROADM-C1-DEG2-DEG2-CTP-TXRX',
+ 'ROADM-C1-SRG1-SRG1-CP-TXRXtoROADM-C1-DEG1-DEG1-CTP-TXRX']
+ dropLink = ['ROADM-A1-DEG1-DEG1-CTP-TXRXtoROADM-A1-SRG1-SRG1-CP-TXRX',
+ 'ROADM-A1-DEG2-DEG2-CTP-TXRXtoROADM-A1-SRG1-SRG1-CP-TXRX',
+ 'ROADM-A1-DEG1-DEG1-CTP-TXRXtoROADM-A1-SRG3-SRG3-CP-TXRX',
+ 'ROADM-A1-DEG2-DEG2-CTP-TXRXtoROADM-A1-SRG3-SRG3-CP-TXRX',
+ 'ROADM-C1-DEG1-DEG1-CTP-TXRXtoROADM-C1-SRG1-SRG1-CP-TXRX',
+ 'ROADM-C1-DEG2-DEG2-CTP-TXRXtoROADM-C1-SRG1-SRG1-CP-TXRX']
R2RLink = ['ROADM-A1-DEG2-DEG2-TTP-TXRXtoROADM-C1-DEG1-DEG1-TTP-TXRX',
'ROADM-C1-DEG1-DEG1-TTP-TXRXtoROADM-A1-DEG2-DEG2-TTP-TXRX']
XPDR_IN = ['ROADM-A1-SRG1-SRG1-PP1-TXRXtoXPDR-A1-XPDR1-XPDR1-NETWORK1']
client = 0
network = 0
for j in range(0, nbTps):
- tpType = res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['org-openroadm-common-network:tp-type']
+ tpType = (res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
+ ['org-openroadm-common-network:tp-type'])
if (tpType == 'XPONDER-CLIENT'):
client += 1
elif (tpType == 'XPONDER-NETWORK'):
res = response.json()
# Tests related to links
nbLink = len(res['network'][0]['ietf-network-topology:link'])
- listR2RLink = ['ROADM-A1-DEG2-DEG2-TTP-TXRXtoROADM-C1-DEG1-DEG1-TTP-TXRX', 'ROADM-C1-DEG1-DEG1-TTP-TXRXtoROADM-A1-DEG2-DEG2-TTP-TXRX',
- 'ROADM-A1-DEG1-DEG1-TTP-TXRXtoROADM-B1-DEG1-DEG1-TTP-TXRX', 'ROADM-C1-DEG2-DEG2-TTP-TXRXtoROADM-B1-DEG2-DEG2-TTP-TXRX',
- 'ROADM-B1-DEG1-DEG1-TTP-TXRXtoROADM-A1-DEG1-DEG1-TTP-TXRX', 'ROADM-B1-DEG2-DEG2-TTP-TXRXtoROADM-C1-DEG2-DEG2-TTP-TXRX']
+ listR2RLink = ['ROADM-A1-DEG2-DEG2-TTP-TXRXtoROADM-C1-DEG1-DEG1-TTP-TXRX',
+ 'ROADM-C1-DEG1-DEG1-TTP-TXRXtoROADM-A1-DEG2-DEG2-TTP-TXRX',
+ 'ROADM-A1-DEG1-DEG1-TTP-TXRXtoROADM-B1-DEG1-DEG1-TTP-TXRX',
+ 'ROADM-C1-DEG2-DEG2-TTP-TXRXtoROADM-B1-DEG2-DEG2-TTP-TXRX',
+ 'ROADM-B1-DEG1-DEG1-TTP-TXRXtoROADM-A1-DEG1-DEG1-TTP-TXRX',
+ 'ROADM-B1-DEG2-DEG2-TTP-TXRXtoROADM-C1-DEG2-DEG2-TTP-TXRX']
for i in range(0, nbLink):
- if res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-common-network:link-type'] == 'ROADM-TO-ROADM':
+ if (res['network'][0]['ietf-network-topology:link'][i]
+ ['org-openroadm-common-network:link-type'] == 'ROADM-TO-ROADM'):
link_id = res['network'][0]['ietf-network-topology:link'][i]['link-id']
find = link_id in listR2RLink
self.assertEqual(find, True)
link_type = res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-common-network:link-type']
link_src = res['network'][0]['ietf-network-topology:link'][i]['source']['source-node']
link_dest = res['network'][0]['ietf-network-topology:link'][i]['destination']['dest-node']
- oppLink_id = res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-common-network:opposite-link']
+ oppLink_id = (res['network'][0]['ietf-network-topology:link'][i]
+ ['org-openroadm-common-network:opposite-link'])
# Find the opposite link
response_oppLink = test_utils.get_ordm_topo_request("ietf-network-topology:link/"+oppLink_id)
self.assertEqual(response_oppLink.status_code, requests.codes.ok)
# #Tests related to links
# nbLink=len(res['network'][0]['ietf-network-topology:link'])
# self.assertEqual(nbLink,8)
-# expressLink=['ROADMA-DEG2-DEG2-CTP-TXRXtoROADMA-DEG1-DEG1-CTP-TXRX','ROADMA-DEG1-DEG1-CTP-TXRXtoROADMA-DEG2-DEG2-CTP-TXRX']
-# addLink=['ROADMA-SRG1-SRG1-CP-TXRXtoROADMA-DEG2-DEG2-CTP-TXRX','ROADMA-SRG1-SRG1-CP-TXRXtoROADMA-DEG1-DEG1-CTP-TXRX',]
-# dropLink=['ROADMA-DEG1-DEG1-CTP-TXRXtoROADMA-SRG1-SRG1-CP-TXRX','ROADMA-DEG2-DEG2-CTP-TXRXtoROADMA-SRG1-SRG1-CP-TXRX']
+# expressLink=['ROADMA-DEG2-DEG2-CTP-TXRXtoROADMA-DEG1-DEG1-CTP-TXRX',
+# 'ROADMA-DEG1-DEG1-CTP-TXRXtoROADMA-DEG2-DEG2-CTP-TXRX']
+# addLink=['ROADMA-SRG1-SRG1-CP-TXRXtoROADMA-DEG2-DEG2-CTP-TXRX',
+# 'ROADMA-SRG1-SRG1-CP-TXRXtoROADMA-DEG1-DEG1-CTP-TXRX',]
+# dropLink=['ROADMA-DEG1-DEG1-CTP-TXRXtoROADMA-SRG1-SRG1-CP-TXRX',
+# 'ROADMA-DEG2-DEG2-CTP-TXRXtoROADMA-SRG1-SRG1-CP-TXRX']
# XPDR_IN=['ROADMA-SRG1-SRG1-PP1-TXRXtoXPDRA-XPDR1-XPDR1-NETWORK1']
# XPDR_OUT=['XPDRA-XPDR1-XPDR1-NETWORK1toROADMA-SRG1-SRG1-PP1-TXRX']
# for i in range(0,nbLink):
# self.assertEqual(len(XPDR_OUT),0)
#
# for i in range(0,nbLink):
-# self.assertNotEqual(res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-common-network:link-type'],'ROADM-TO-ROADM')
-# self.assertNotEqual(res['network'][0]['ietf-network-topology:link'][i]['link-id'],'ROADMC-SRG1-SRG1-CP-TXRXtoROADMC-DEG1-DEG1-CTP-TXRX')
-# self.assertNotEqual(res['network'][0]['ietf-network-topology:link'][i]['link-id'],'ROADMC-DEG1-DEG1-CTP-TXRXtoROADMC-SRG1-SRG1-CP-TXRX')
-# self.assertNotEqual(res['network'][0]['ietf-network-topology:link'][i]['link-id'],'ROADMC-SRG1-SRG1-CP-TXRXtoROADMC-DEG2-DEG1-CTP-TXRX')
-# self.assertNotEqual(res['network'][0]['ietf-network-topology:link'][i]['link-id'],'ROADMC-DEG1-DEG2-CTP-TXRXtoROADMC-SRG1-SRG1-CP-TXRX')
-# self.assertNotEqual(res['network'][0]['ietf-network-topology:link'][i]['link-id'],'ROADMC-DEG1-DEG1-CTP-TXRXtoROADMC-DEG2-DEG2-CTP-TXRX')
-# self.assertNotEqual(res['network'][0]['ietf-network-topology:link'][i]['link-id'],'ROADMC-DEG2-DEG2-CTP-TXRXtoROADMC-DEG1-DEG1-CTP-TXRX')
+# self.assertNotEqual((res['network'][0]['ietf-network-topology:link'][i]
+# ['org-openroadm-common-network:link-type']),'ROADM-TO-ROADM')
+# self.assertNotEqual(res['network'][0]['ietf-network-topology:link'][i]['link-id'],
+# 'ROADMC-SRG1-SRG1-CP-TXRXtoROADMC-DEG1-DEG1-CTP-TXRX')
+# self.assertNotEqual(res['network'][0]['ietf-network-topology:link'][i]['link-id'],
+# 'ROADMC-DEG1-DEG1-CTP-TXRXtoROADMC-SRG1-SRG1-CP-TXRX')
+# self.assertNotEqual(res['network'][0]['ietf-network-topology:link'][i]['link-id'],
+# 'ROADMC-SRG1-SRG1-CP-TXRXtoROADMC-DEG2-DEG1-CTP-TXRX')
+# self.assertNotEqual(res['network'][0]['ietf-network-topology:link'][i]['link-id'],
+# 'ROADMC-DEG1-DEG2-CTP-TXRXtoROADMC-SRG1-SRG1-CP-TXRX')
+# self.assertNotEqual(res['network'][0]['ietf-network-topology:link'][i]['link-id'],
+# 'ROADMC-DEG1-DEG1-CTP-TXRXtoROADMC-DEG2-DEG2-CTP-TXRX')
+# self.assertNotEqual(res['network'][0]['ietf-network-topology:link'][i]['link-id'],
+# 'ROADMC-DEG2-DEG2-CTP-TXRXtoROADMC-DEG1-DEG1-CTP-TXRX')
def test_32_getNodes_OpenRoadmTopology(self):
response = test_utils.get_ordm_topo_request("")
for j in range(0, nbTp):
tpid = res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['tp-id']
if (tpid == 'XPDR1-CLIENT1'):
- self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
- ['org-openroadm-common-network:tp-type'], 'XPONDER-CLIENT')
+ self.assertEqual((res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
+ ['org-openroadm-common-network:tp-type']), 'XPONDER-CLIENT')
if (tpid == 'XPDR1-NETWORK1'):
- self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
- ['org-openroadm-common-network:tp-type'], 'XPONDER-NETWORK')
- self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
- ['org-openroadm-network-topology:xpdr-network-attributes']['tail-equipment-id'],
+ self.assertEqual((res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
+ ['org-openroadm-common-network:tp-type']), 'XPONDER-NETWORK')
+ self.assertEqual((res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
+ ['org-openroadm-network-topology:xpdr-network-attributes']
+ ['tail-equipment-id']),
'ROADM-A1-SRG1--SRG1-PP1-TXRX')
self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'XPDR-A1'},
res['network'][0]['node'][i]['supporting-node'])
self.assertEqual(nbLink, 16)
expressLink = ['ROADM-A1-DEG2-DEG2-CTP-TXRXtoROADM-A1-DEG1-DEG1-CTP-TXRX',
'ROADM-A1-DEG1-DEG1-CTP-TXRXtoROADM-A1-DEG2-DEG2-CTP-TXRX']
- addLink = ['ROADM-A1-SRG1-SRG1-CP-TXRXtoROADM-A1-DEG2-DEG2-CTP-TXRX', 'ROADM-A1-SRG1-SRG1-CP-TXRXtoROADM-A1-DEG1-DEG1-CTP-TXRX',
- 'ROADM-A1-SRG3-SRG3-CP-TXRXtoROADM-A1-DEG2-DEG2-CTP-TXRX', 'ROADM-A1-SRG3-SRG3-CP-TXRXtoROADM-A1-DEG1-DEG1-CTP-TXRX']
- dropLink = ['ROADM-A1-DEG1-DEG1-CTP-TXRXtoROADM-A1-SRG1-SRG1-CP-TXRX', 'ROADM-A1-DEG2-DEG2-CTP-TXRXtoROADM-A1-SRG1-SRG1-CP-TXRX',
- 'ROADM-A1-DEG1-DEG1-CTP-TXRXtoROADM-A1-SRG3-SRG3-CP-TXRX', 'ROADM-A1-DEG2-DEG2-CTP-TXRXtoROADM-A1-SRG3-SRG3-CP-TXRX']
+ addLink = ['ROADM-A1-SRG1-SRG1-CP-TXRXtoROADM-A1-DEG2-DEG2-CTP-TXRX',
+ 'ROADM-A1-SRG1-SRG1-CP-TXRXtoROADM-A1-DEG1-DEG1-CTP-TXRX',
+ 'ROADM-A1-SRG3-SRG3-CP-TXRXtoROADM-A1-DEG2-DEG2-CTP-TXRX',
+ 'ROADM-A1-SRG3-SRG3-CP-TXRXtoROADM-A1-DEG1-DEG1-CTP-TXRX']
+ dropLink = ['ROADM-A1-DEG1-DEG1-CTP-TXRXtoROADM-A1-SRG1-SRG1-CP-TXRX',
+ 'ROADM-A1-DEG2-DEG2-CTP-TXRXtoROADM-A1-SRG1-SRG1-CP-TXRX',
+ 'ROADM-A1-DEG1-DEG1-CTP-TXRXtoROADM-A1-SRG3-SRG3-CP-TXRX',
+ 'ROADM-A1-DEG2-DEG2-CTP-TXRXtoROADM-A1-SRG3-SRG3-CP-TXRX']
roadmtoroadmLink = 0
for i in range(0, nbLink):
- if (res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-common-network:link-type'] == 'EXPRESS-LINK'):
+ if (res['network'][0]['ietf-network-topology:link'][i]
+ ['org-openroadm-common-network:link-type'] == 'EXPRESS-LINK'):
link_id = res['network'][0]['ietf-network-topology:link'][i]['link-id']
find = link_id in expressLink
self.assertEqual(find, True)
expressLink.remove(link_id)
- elif (res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-common-network:link-type'] == 'ADD-LINK'):
+ elif (res['network'][0]['ietf-network-topology:link'][i]
+ ['org-openroadm-common-network:link-type'] == 'ADD-LINK'):
link_id = res['network'][0]['ietf-network-topology:link'][i]['link-id']
find = link_id in addLink
self.assertEqual(find, True)
addLink.remove(link_id)
- elif (res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-common-network:link-type'] == 'DROP-LINK'):
+ elif (res['network'][0]['ietf-network-topology:link'][i]
+ ['org-openroadm-common-network:link-type'] == 'DROP-LINK'):
link_id = res['network'][0]['ietf-network-topology:link'][i]['link-id']
find = link_id in dropLink
self.assertEqual(find, True)
def service_create_request(attr):
return post_request(URL_SERV_CREATE, attr)
-def service_delete_request(servicename : str,
- requestid = "e3028bae-a90f-4ddd-a83f-cf224eba0e58",
+
+def service_delete_request(servicename: str,
+ requestid="e3028bae-a90f-4ddd-a83f-cf224eba0e58",
notificationurl="http://localhost:8585/NotificationServer/notify"):
attr = {"input": {
- "sdnc-request-header": {
- "request-id": requestid,
- "rpc-action": "service-delete",
- "request-system-id": "appname",
- "notification-url": notificationurl},
- "service-delete-req-info": {
- "service-name": servicename,
- "tail-retention": "no"}}}
+ "sdnc-request-header": {
+ "request-id": requestid,
+ "rpc-action": "service-delete",
+ "request-system-id": "appname",
+ "notification-url": notificationurl},
+ "service-delete-req-info": {
+ "service-name": servicename,
+ "tail-retention": "no"}}}
return post_request(URL_SERV_DELETE, attr)
return post_request(URL_SERVICE_PATH, attr)
-def otn_service_path_request(operation: str, servicename: str, servicerate: str, servicetype: str, nodes, eth_attr=None):
+def otn_service_path_request(operation: str, servicename: str, servicerate: str, servicetype: str, nodes,
+ eth_attr=None):
attr = {"service-name": servicename,
"operation": operation,
"service-rate": servicerate,
"logical-connection-point": lcp}}
return post_request(URL_CREATE_OTS_OMS, attr)
+
def path_computation_request(requestid: str, servicename: str, serviceaend, servicezend,
hardconstraints=None, softconstraints=None, metric="hop-count", other_attr=None):
- attr = {"service-name": servicename,
- "resource-reserve": "true",
- "service-handler-header": { "request-id": requestid },
- "service-a-end": serviceaend,
- "service-z-end": servicezend,
- "pce-metric": metric}
+ attr = {"service-name": servicename,
+ "resource-reserve": "true",
+ "service-handler-header": {"request-id": requestid},
+ "service-a-end": serviceaend,
+ "service-z-end": servicezend,
+ "pce-metric": metric}
if hardconstraints:
- attr.update({ "hard-constraints": hardconstraints})
+ attr.update({"hard-constraints": hardconstraints})
if softconstraints:
- attr.update({ "soft-constraints": softconstraints})
+ attr.update({"soft-constraints": softconstraints})
if other_attr:
attr.update(other_attr)
- return post_request(URL_PATH_COMPUTATION_REQUEST, {"input": attr })
+ return post_request(URL_PATH_COMPUTATION_REQUEST, {"input": attr})
def shutdown_process(process):
Code Review / transportpce.git / commitdiff