"layer-protocol-name": "PHOTONIC_MEDIA",
"service-interface-point": {
"service-interface-point-uuid": "b1a0d883-32b8-3b0b-93d6-7ed074f6f107"
- },
+ },
"administrative-state": "UNLOCKED",
"operational-state": "ENABLED",
"direction": "BIDIRECTIONAL",
"layer-protocol-name": "PHOTONIC_MEDIA",
"service-interface-point": {
"service-interface-point-uuid": "d1d6305e-179b-346f-b02d-8260aebe1ce8"
- },
+ },
"administrative-state": "UNLOCKED",
"operational-state": "ENABLED",
"direction": "BIDIRECTIONAL",
print("photonic media service uuid : {}".format(res['output']['service']['uuid']))
input_dict_1 = {'administrative-state': 'LOCKED',
- 'lifecycle-state': 'PLANNED',
- 'operational-state': 'DISABLED',
- 'service-type': 'POINT_TO_POINT_CONNECTIVITY',
- 'service-layer': 'PHOTONIC_MEDIA',
- 'connectivity-direction': 'BIDIRECTIONAL'
+ 'lifecycle-state': 'PLANNED',
+ 'operational-state': 'DISABLED',
+ 'service-type': 'POINT_TO_POINT_CONNECTIVITY',
+ 'service-layer': 'PHOTONIC_MEDIA',
+ 'connectivity-direction': 'BIDIRECTIONAL'
}
input_dict_2 = {'value-name': 'OpenROADM node id',
'value': 'SPDR-SC1-XPDR1'}
def test_16_get_service_PhotonicMedia(self):
response = test_utils.get_service_list_request(
- "services/"+ str(service_pm_uuid))
+ "services/" + str(service_pm_uuid))
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(
print("odu service uuid : {}".format(res['output']['service']['uuid']))
input_dict_1 = {'administrative-state': 'LOCKED',
- 'lifecycle-state': 'PLANNED',
- 'operational-state': 'DISABLED',
- 'service-type': 'POINT_TO_POINT_CONNECTIVITY',
- 'service-layer': 'ODU',
- 'connectivity-direction': 'BIDIRECTIONAL'
+ 'lifecycle-state': 'PLANNED',
+ 'operational-state': 'DISABLED',
+ 'service-type': 'POINT_TO_POINT_CONNECTIVITY',
+ 'service-layer': 'ODU',
+ 'connectivity-direction': 'BIDIRECTIONAL'
}
input_dict_2 = {'value-name': 'OpenROADM node id',
'value': 'SPDR-SC1-XPDR1'}
def test_18_get_service_ODU(self):
response = test_utils.get_service_list_request(
- "services/"+ str(service_odu_uuid))
+ "services/" + str(service_odu_uuid))
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(
print("dsr service uuid : {}".format(res['output']['service']['uuid']))
input_dict_1 = {'administrative-state': 'LOCKED',
- 'lifecycle-state': 'PLANNED',
- 'operational-state': 'DISABLED',
- 'service-type': 'POINT_TO_POINT_CONNECTIVITY',
- 'service-layer': 'DSR',
- 'connectivity-direction': 'BIDIRECTIONAL'
+ 'lifecycle-state': 'PLANNED',
+ 'operational-state': 'DISABLED',
+ 'service-type': 'POINT_TO_POINT_CONNECTIVITY',
+ 'service-layer': 'DSR',
+ 'connectivity-direction': 'BIDIRECTIONAL'
}
input_dict_2 = {'value-name': 'OpenROADM node id',
'value': 'SPDR-SC1-XPDR1'}
def test_20_get_service_DSR(self):
response = test_utils.get_service_list_request(
- "services/"+ str(service_dsr_uuid))
+ "services/" + str(service_dsr_uuid))
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(
time.sleep(2)
def test_22_delete_connectivity_service_DSR(self):
- response = test_utils.tapi_delete_connectivity_request(service_dsr_uuid)
- self.assertEqual(response.status_code, requests.codes.no_content)
- time.sleep(self.WAITING)
+ response = test_utils.tapi_delete_connectivity_request(service_dsr_uuid)
+ self.assertEqual(response.status_code, requests.codes.no_content)
+ time.sleep(self.WAITING)
def test_23_delete_connectivity_service_ODU(self):
response = test_utils.tapi_delete_connectivity_request(service_odu_uuid)
self.assertEqual(nb_links, 4)
for link in res['network'][0]['ietf-network-topology:link']:
self.assertEqual(
- link['org-openroadm-otn-network-topology:available-bandwidth'], 0)
+ link['org-openroadm-otn-network-topology:available-bandwidth'], 0)
self.assertEqual(
- link['org-openroadm-otn-network-topology:used-bandwidth'], 100000)
+ link['org-openroadm-otn-network-topology:used-bandwidth'], 100000)
def test_60_delete_service_100GE_ABC(self):
response = test_utils.service_delete_request("service-100GE-ABC")
for link in links:
if ((link["org-openroadm-common-network:link-type"] == "XPONDER-OUTPUT" or
link["org-openroadm-common-network:link-type"] == "XPONDER-INPUT")
- and ('SPDR-SB1' in link['link-id'] or 'ROADM-B1' in link['link-id'])):
+ and ('SPDR-SB1' in link['link-id'] or 'ROADM-B1' in link['link-id'])):
link_name = link["link-id"]
response = test_utils.delete_request(url+link_name)
self.assertEqual(response.status_code, requests.codes.ok)
links = res['network'][0]['ietf-network-topology:link']
for link in links:
if (link["org-openroadm-common-network:link-type"] == "ROADM-TO-ROADM"
- and 'ROADM-B1' in link['link-id']):
+ and 'ROADM-B1' in link['link-id']):
link_name = link["link-id"]
response = test_utils.delete_request(url+link_name)
self.assertEqual(response.status_code, requests.codes.ok)
self.test_50_check_interface_ODU4_NETWORK_spdra()
self.test_51_check_ODU4_connection_spdra()
-
def test_88_check_configuration_spdrc(self):
self.test_52_check_interface_100GE_CLIENT_spdrc()
self.test_53_check_interface_ODU4_CLIENT_spdrc()
"tail-retention": "no"}}}
return post_request(URL_SERV_DELETE, attr)
+
def service_path_request(operation: str, servicename: str, wavenumber: str, nodes, centerfreq: str,
slotwidth: int, minfreq: float, maxfreq: float, lowerslotnumber: int,
higherslotnumber: int):
attr.update(other_attr)
return post_request(URL_PATH_COMPUTATION_REQUEST, {"input": attr})
+
def tapi_create_connectivity_request(topologyidorname):
return post_request(URL_TAPI_CREATE_CONNECTIVITY, topologyidorname)
+
def tapi_delete_connectivity_request(serviceidorname):
attr = {
"input": {
"service-id-or-name": serviceidorname}}
- return post_request(URL_TAPI_DELETE_CONNECTIVITY , attr)
+ return post_request(URL_TAPI_DELETE_CONNECTIVITY, attr)
+
def tapi_get_topology_details_request(topologyidorname):
attr = {
"input": {
"topology-id-or-name": topologyidorname}}
- return post_request(URL_TAPI_TOPOLOGY_DETAILS , attr)
+ return post_request(URL_TAPI_TOPOLOGY_DETAILS, attr)
+
def tapi_get_node_details_request(topologyidorname, nodeidorname):
attr = {
"node-id-or-name": nodeidorname}}
return post_request(URL_TAPI_NODE_DETAILS, attr)
+
def tapi_get_sip_details_request():
return post_request(URL_TAPI_SIP_LIST, "")
+
def tapi_get_service_list_request():
return post_request(URL_TAPI_SERVICE_LIST, "")
+
def shutdown_process(process):
if process is not None:
for child in psutil.Process(process.pid).children():