def test_02_compare_Openroadm_topology_portmapping_rdm(self):
responseTopo = test_utils_rfc8040.get_request(test_utils_rfc8040.URL_CONFIG_ORDM_TOPO)
resTopo = responseTopo.json()
- firstEntry = resTopo['ietf-network:network'][0]['node']
- for i in range(0, len(firstEntry)):
- nodeId = firstEntry[i]['node-id']
+ for val in resTopo['ietf-network:network'][0]['node']:
+ nodeId = val['node-id']
nodeMapId = nodeId.split("-")[0]
response = test_utils_rfc8040.get_portmapping_node_info(nodeMapId)
self.assertEqual(response['status_code'], requests.codes.ok)
- nbTp = len(firstEntry[i]['ietf-network-topology:termination-point'])
- for j in range(0, nbTp):
- tpId = firstEntry[i]['ietf-network-topology:termination-point'][j]['tp-id']
- if((not "CP" in tpId) and (not "CTP" in tpId)):
+ for val2 in val['ietf-network-topology:termination-point']:
+ tpId = val2['tp-id']
+ if (not "CP" in tpId) and (not "CTP" in tpId):
response2 = test_utils_rfc8040.portmapping_request(nodeMapId, tpId)
self.assertEqual(response2['status_code'], requests.codes.ok)
self.assertEqual({'frequency': 196.1,
'width': 40},
ele['org-openroadm-network-topology:xpdr-network-attributes']['wavelength'])
- if ele['tp-id'] == 'XPDR1-CLIENT2' or ele['tp-id'] == 'XPDR1-CLIENT1':
+ elif ele['tp-id'] in ('XPDR1-CLIENT2', 'XPDR1-CLIENT1'):
self.assertNotIn('org-openroadm-network-topology:xpdr-client-attributes', dict.keys(ele))
- if ele['tp-id'] == 'XPDR1-NETWORK2':
+ elif ele['tp-id'] == 'XPDR1-NETWORK2':
self.assertNotIn('org-openroadm-network-topology:xpdr-network-attributes', dict.keys(ele))
time.sleep(3)
ele['org-openroadm-network-topology:pp-attributes']['avail-freq-maps'][0]['freq-map'])
freq_map_array = [int(x) for x in freq_map]
self.assertEqual(freq_map_array[95], 0, "Lambda 1 should not be available")
- if ele['tp-id'] == 'SRG1-PP2-TXRX':
+ elif ele['tp-id'] == 'SRG1-PP2-TXRX':
self.assertNotIn('avail-freq-maps', dict.keys(ele))
time.sleep(3)
ele['org-openroadm-network-topology:ctp-attributes']['avail-freq-maps'][0]['freq-map'])
freq_map_array = [int(x) for x in freq_map]
self.assertEqual(freq_map_array[95], 0, "Lambda 1 should not be available")
- if ele['tp-id'] == 'DEG2-TTP-TXRX':
+ elif ele['tp-id'] == 'DEG2-TTP-TXRX':
freq_map = base64.b64decode(
ele['org-openroadm-network-topology:tx-ttp-attributes']['avail-freq-maps'][0]['freq-map'])
freq_map_array = [int(x) for x in freq_map]
self.assertEqual({'frequency': 196.1,
'width': 40},
ele['org-openroadm-network-topology:xpdr-network-attributes']['wavelength'])
- if ele['tp-id'] == 'XPDR1-NETWORK2':
+ elif ele['tp-id'] == 'XPDR1-NETWORK2':
self.assertEqual({'frequency': 196.05,
'width': 40},
ele['org-openroadm-network-topology:xpdr-network-attributes']['wavelength'])
- if ele['tp-id'] == 'XPDR1-CLIENT1' or ele['tp-id'] == 'XPDR1-CLIENT2':
+ elif ele['tp-id'] in ('XPDR1-CLIENT1', 'XPDR1-CLIENT2'):
self.assertNotIn('org-openroadm-network-topology:xpdr-client-attributes', dict.keys(ele))
time.sleep(10)
freq_map_array = [int(x) for x in freq_map]
self.assertEqual(freq_map_array[95], 0, "Lambda 1 should not be available")
self.assertEqual(freq_map_array[94], 255, "Lambda 2 should be available")
- if ele['tp-id'] == 'SRG1-PP2-TXRX':
+ elif ele['tp-id'] == 'SRG1-PP2-TXRX':
freq_map = base64.b64decode(
ele['org-openroadm-network-topology:pp-attributes']['avail-freq-maps'][0]['freq-map'])
freq_map_array = [int(x) for x in freq_map]
self.assertEqual(freq_map_array[95], 255, "Lambda 1 should be available")
self.assertEqual(freq_map_array[94], 0, "Lambda 2 should not be available")
- if ele['tp-id'] == 'SRG1-PP3-TXRX':
+ elif ele['tp-id'] == 'SRG1-PP3-TXRX':
self.assertNotIn('org-openroadm-network-topology:pp-attributes', dict.keys(ele))
time.sleep(10)
freq_map_array = [int(x) for x in freq_map]
self.assertEqual(freq_map_array[95], 0, "Lambda 1 should not be available")
self.assertEqual(freq_map_array[94], 0, "Lambda 2 should not be available")
- if ele['tp-id'] == 'DEG2-TTP-TXRX':
+ elif ele['tp-id'] == 'DEG2-TTP-TXRX':
freq_map = base64.b64decode(
ele['org-openroadm-network-topology:tx-ttp-attributes']['avail-freq-maps'][0]['freq-map'])
freq_map_array = [int(x) for x in freq_map]
res = response.json()
liste_tp = res['node'][0]['ietf-network-topology:termination-point']
for ele in liste_tp:
- if ((ele['org-openroadm-common-network:tp-type'] ==
- 'XPONDER-CLIENT')
- and (ele['tp-id'] == 'XPDR1-CLIENT1' or ele[
- 'tp-id'] == 'XPDR1-CLIENT3')):
+ if (ele['org-openroadm-common-network:tp-type'] == 'XPONDER-CLIENT'
+ and ele['tp-id'] in ('XPDR1-CLIENT1', 'XPDR1-CLIENT3')):
self.assertNotIn(
'org-openroadm-network-topology:xpdr-client-attributes',
dict.keys(ele))
- elif (ele['org-openroadm-common-network:tp-type'] ==
- 'XPONDER-NETWORK'):
+ elif ele['org-openroadm-common-network:tp-type'] == 'XPONDER-NETWORK':
self.assertIn('tail-equipment-id', dict.keys(
ele['org-openroadm-network-topology:'
'xpdr-network-attributes']))
freq_map_array = [int(x) for x in freq_map]
self.assertEqual(freq_map_array[95], 255, "Lambda 1 should be available")
self.assertEqual(freq_map_array[94], 255, "Lambda 2 should be available")
- if ele['tp-id'] == 'DEG2-TTP-TXRX':
+ elif ele['tp-id'] == 'DEG2-TTP-TXRX':
freq_map = base64.b64decode(
ele['org-openroadm-network-topology:tx-ttp-attributes']['avail-freq-maps'][0]['freq-map'])
freq_map_array = [int(x) for x in freq_map]
def test_02_compareOpenroadmTopologyPortMapping_rdm(self):
responseTopo = test_utils_rfc8040.get_request(test_utils_rfc8040.URL_CONFIG_ORDM_TOPO)
resTopo = responseTopo.json()
- firstEntry = resTopo['ietf-network:network'][0]['node']
nbMapCumul = 0
nbMappings = 0
- for i in range(0, len(firstEntry)):
- nodeId = firstEntry[i]['node-id']
+ for val in resTopo['ietf-network:network'][0]['node']:
+ nodeId = val['node-id']
# pylint: disable=consider-using-f-string
print("nodeId={}".format(nodeId))
nodeMapId = nodeId.split("-")[0] + "-" + nodeId.split("-")[1]
self.assertEqual(response['status_code'], requests.codes.ok)
responseMapList = test_utils_rfc8040.get_portmapping(nodeMapId)
nbMappings = len(responseMapList['nodes'][0]['mapping']) - nbMapCumul
- nbTp = len(firstEntry[i]['ietf-network-topology:termination-point'])
nbMapCurrent = 0
- for j in range(0, nbTp):
- tpId = firstEntry[i]['ietf-network-topology:termination-point'][j]['tp-id']
+ for val2 in val['ietf-network-topology:termination-point']:
+ tpId = val2['tp-id']
if (not "CP" in tpId) and (not "CTP" in tpId):
responseMap = test_utils_rfc8040.portmapping_request(nodeMapId, tpId)
self.assertEqual(responseMap['status_code'], requests.codes.ok)
elif tpType == 'XPONDER-NETWORK':
network += 1
self.assertTrue(client == 0)
- if nodeId == 'SPDR-SA1-XPDR1' or nodeId == 'SPDR-SA1-XPDR3':
+ if nodeId in ('SPDR-SA1-XPDR1', 'SPDR-SA1-XPDR3'):
self.assertTrue(network == 1)
else:
# elif nodeId == 'SPDR-SA1-XPDR2':
response = test_utils.get_otn_topo_request()
res = response.json()
self.assertEqual(response.status_code, requests.codes.ok)
- nbNode = len(res['network'][0]['node'])
- self.assertEqual(nbNode, 3)
+ self.assertEqual(len(res['network'][0]['node']), 3)
listNode = ['SPDR-SA1-XPDR1', 'SPDR-SA1-XPDR2', 'SPDR-SA1-XPDR3']
CHECK_LIST = {
'SPDR-SA1-XPDR1': {
'tp-unchecklist': []
}
}
- for i in range(0, nbNode):
- nodeId = res['network'][0]['node'][i]['node-id']
+ for val in res['network'][0]['node']:
+ nodeId = val['node-id']
if nodeId in CHECK_LIST:
- nodeType = res['network'][0]['node'][i]['org-openroadm-common-network:node-type']
- self.assertEqual(nodeType, CHECK_LIST[nodeId]['node-type'])
+ self.assertEqual(val['org-openroadm-common-network:node-type'], CHECK_LIST[nodeId]['node-type'])
self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'SPDR-SA1'},
- res['network'][0]['node'][i]['supporting-node'])
+ val['supporting-node'])
self.assertIn({'network-ref': 'openroadm-topology', 'node-ref': nodeId},
- res['network'][0]['node'][i]['supporting-node'])
+ val['supporting-node'])
self.assertIn({'network-ref': 'clli-network', 'node-ref': 'NodeSA'},
- res['network'][0]['node'][i]['supporting-node'])
- self.assertEqual(res['network'][0]['node'][i]
- ['org-openroadm-otn-network-topology:xpdr-attributes']['xpdr-number'],
+ val['supporting-node'])
+ self.assertEqual(val['org-openroadm-otn-network-topology:xpdr-attributes']['xpdr-number'],
CHECK_LIST[nodeId]['xpdr-number'])
- nbTps = len(res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
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'])
- tpId = res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['tp-id']
+ for val2 in val['ietf-network-topology:termination-point']:
+ tpType = val2['org-openroadm-common-network:tp-type']
+ tpId = val2['tp-id']
if tpType == 'XPONDER-CLIENT':
client += 1
# pylint: disable=consider-using-f-string
print("tpId = {}".format(tpId))
- print("tp= {}".format(res['network'][0]['node'][i]
- ['ietf-network-topology:termination-point'][j]))
- nbIfCapType = len(res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
- ['org-openroadm-otn-network-topology:tp-supported-interfaces']
- ['supported-interface-capability'][0])
- for k in range(0, nbIfCapType):
- self.assertIn((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']),
- CHECK_LIST[nodeId]['port-types'])
+ print("tp= {}".format(val2))
+ # pylint: disable=line-too-long
+ for val3 in val2['org-openroadm-otn-network-topology:tp-supported-interfaces']['supported-interface-capability']:
+ self.assertIn((val3['if-cap-type']), CHECK_LIST[nodeId]['port-types'])
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']),
+ self.assertEqual((val2['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']),
+ self.assertEqual((val2['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']), nodeId)
- self.assertEqual((res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
- ['supporting-termination-point'][0]['tp-ref']), tpId)
+ self.assertEqual((val2['supporting-termination-point'][0]['network-ref']), 'openroadm-topology')
+ self.assertEqual((val2['supporting-termination-point'][0]['node-ref']), nodeId)
+ self.assertEqual((val2['supporting-termination-point'][0]['tp-ref']), tpId)
self.assertTrue(client == 4)
self.assertTrue(network == CHECK_LIST[nodeId]['network_nb'])
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, CHECK_LIST[nodeId]['nbl_nb'])
- 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])
+ self.assertEqual(
+ len(val['org-openroadm-otn-network-topology:switching-pools']
+ ['odu-switching-pools'][0]['non-blocking-list']),
+ CHECK_LIST[nodeId]['nbl_nb'])
+ # pylint: disable=line-too-long
+ for nbl in val['org-openroadm-otn-network-topology:switching-pools']['odu-switching-pools'][0]['non-blocking-list']:
if nbl['nbl-number'] == 1:
if nodeId == 'SPDR-SA1-XPDR1':
self.assertEqual(nbl['available-interconnect-bandwidth'], 10)
res = response.json()
links = res['network'][0]['ietf-network-topology:link']
for link in links:
- if (link["org-openroadm-common-network:link-type"] == "XPONDER-OUTPUT" or
- link["org-openroadm-common-network:link-type"] == "XPONDER-INPUT"):
+ if link["org-openroadm-common-network:link-type"] in ('XPONDER-OUTPUT', 'XPONDER-INPUT'):
link_name = link["link-id"]
response = test_utils.delete_request(url+link_name)
self.assertEqual(response.status_code, requests.codes.ok)
response = test_utils.get_otn_topo_request()
res = response.json()
for node in res['network'][0]['node']:
- if node['node-id'] == 'SPDR-SA1-XPDR1' or node['node-id'] == 'SPDR-SC1-XPDR1':
+ if node['node-id'] in ('SPDR-SA1-XPDR1', 'SPDR-SC1-XPDR1'):
tpList = node['ietf-network-topology:termination-point']
for tp in tpList:
if tp['tp-id'] == 'XPDR1-NETWORK1':
response = test_utils.get_otn_topo_request()
res = response.json()
for node in res['network'][0]['node']:
- if node['node-id'] == 'SPDR-SA1-XPDR1' or node['node-id'] == 'SPDR-SC1-XPDR1':
+ if node['node-id'] in ('SPDR-SA1-XPDR1', 'SPDR-SC1-XPDR1'):
tpList = node['ietf-network-topology:termination-point']
for tp in tpList:
if tp['tp-id'] == 'XPDR1-NETWORK1':
response = test_utils.get_otn_topo_request()
res = response.json()
for node in res['network'][0]['node']:
- if node['node-id'] == 'SPDR-SA1-XPDR1' or node['node-id'] == 'SPDR-SC1-XPDR1':
+ if node['node-id'] in ('SPDR-SA1-XPDR1', 'SPDR-SC1-XPDR1'):
tpList = node['ietf-network-topology:termination-point']
for tp in tpList:
if tp['tp-id'] == 'XPDR1-NETWORK1':
response = test_utils.get_otn_topo_request()
res = response.json()
for node in res['network'][0]['node']:
- if node['node-id'] == 'SPDR-SA1-XPDR1' or node['node-id'] == 'SPDR-SC1-XPDR1':
+ if node['node-id'] in ('SPDR-SA1-XPDR1', 'SPDR-SC1-XPDR1'):
tpList = node['ietf-network-topology:termination-point']
for tp in tpList:
if tp['tp-id'] == 'XPDR1-NETWORK1':
response = test_utils.get_otn_topo_request()
res = response.json()
for node in res['network'][0]['node']:
- if node['node-id'] == 'SPDR-SA1-XPDR3' or node['node-id'] == 'SPDR-SC1-XPDR3':
+ if node['node-id'] in ('SPDR-SA1-XPDR3', 'SPDR-SC1-XPDR3'):
tpList = node['ietf-network-topology:termination-point']
for tp in tpList:
if tp['tp-id'] == 'XPDR3-NETWORK1':
response = test_utils.get_otn_topo_request()
res = response.json()
for node in res['network'][0]['node']:
- if node['node-id'] == 'SPDR-SA1-XPDR3' or node['node-id'] == 'SPDR-SC1-XPDR3':
+ if node['node-id'] in ('SPDR-SA1-XPDR3', 'SPDR-SC1-XPDR3'):
tpList = node['ietf-network-topology:termination-point']
for tp in tpList:
if tp['tp-id'] == 'XPDR3-NETWORK1':
res = response.json()
links = res['network'][0]['ietf-network-topology:link']
for link in links:
- if (link["org-openroadm-common-network:link-type"] == "XPONDER-OUTPUT" or
- link["org-openroadm-common-network:link-type"] == "XPONDER-INPUT"):
+ if link["org-openroadm-common-network:link-type"] in ('XPONDER-OUTPUT', 'XPONDER-INPUT'):
link_name = link["link-id"]
response = test_utils.delete_request(url+link_name)
self.assertEqual(response.status_code, requests.codes.ok)
self.assertEqual({'frequency': 196.1,
'width': 40},
ele['org-openroadm-network-topology:xpdr-network-attributes']['wavelength'])
- if ele['tp-id'] == 'XPDR1-CLIENT2' or ele['tp-id'] == 'XPDR1-CLIENT1':
+ elif ele['tp-id'] in ('XPDR1-CLIENT2', 'XPDR1-CLIENT1'):
self.assertNotIn('org-openroadm-network-topology:xpdr-client-attributes', dict.keys(ele))
- if ele['tp-id'] == 'XPDR1-NETWORK2':
+ elif ele['tp-id'] == 'XPDR1-NETWORK2':
self.assertNotIn('org-openroadm-network-topology:xpdr-network-attributes', dict.keys(ele))
time.sleep(3)
ele['org-openroadm-network-topology:pp-attributes']['avail-freq-maps'][0]['freq-map'])
freq_map_array = [int(x) for x in freq_map]
self.assertEqual(freq_map_array[95], 0, "Index 1 should not be available")
- if ele['tp-id'] == 'SRG1-PP2-TXRX':
+ elif ele['tp-id'] == 'SRG1-PP2-TXRX':
self.assertNotIn('avail-freq-maps', dict.keys(ele))
time.sleep(3)
ele['org-openroadm-network-topology:ctp-attributes']['avail-freq-maps'][0]['freq-map'])
freq_map_array = [int(x) for x in freq_map]
self.assertEqual(freq_map_array[95], 0, "Index 1 should not be available")
- if ele['tp-id'] == 'DEG2-TTP-TXRX':
+ elif ele['tp-id'] == 'DEG2-TTP-TXRX':
freq_map = base64.b64decode(
ele['org-openroadm-network-topology:tx-ttp-attributes']['avail-freq-maps'][0]['freq-map'])
freq_map_array = [int(x) for x in freq_map]
self.assertEqual({'frequency': 196.1,
'width': 40},
ele['org-openroadm-network-topology:xpdr-network-attributes']['wavelength'])
- if ele['tp-id'] == 'XPDR1-NETWORK2':
+ elif ele['tp-id'] == 'XPDR1-NETWORK2':
self.assertEqual({'frequency': 196.05,
'width': 40},
ele['org-openroadm-network-topology:xpdr-network-attributes']['wavelength'])
- if ele['tp-id'] == 'XPDR1-CLIENT1' or ele['tp-id'] == 'XPDR1-CLIENT2':
+ elif ele['tp-id'] in ('XPDR1-CLIENT1', 'XPDR1-CLIENT2'):
self.assertNotIn('org-openroadm-network-topology:xpdr-client-attributes', dict.keys(ele))
time.sleep(10)
freq_map_array = [int(x) for x in freq_map]
self.assertEqual(freq_map_array[95], 0, "Lambda 1 should not be available")
self.assertEqual(freq_map_array[94], 255, "Lambda 2 should be available")
- if ele['tp-id'] == 'SRG1-PP2-TXRX':
+ elif ele['tp-id'] == 'SRG1-PP2-TXRX':
freq_map = base64.b64decode(
ele['org-openroadm-network-topology:pp-attributes']['avail-freq-maps'][0]['freq-map'])
freq_map_array = [int(x) for x in freq_map]
self.assertEqual(freq_map_array[95], 255, "Lambda 1 should be available")
self.assertEqual(freq_map_array[94], 0, "Lambda 2 should not be available")
- if ele['tp-id'] == 'SRG1-PP3-TXRX':
+ elif ele['tp-id'] == 'SRG1-PP3-TXRX':
self.assertNotIn('org-openroadm-network-topology:pp-attributes', dict.keys(ele))
time.sleep(10)
freq_map_array = [int(x) for x in freq_map]
self.assertEqual(freq_map_array[95], 0, "Lambda 1 should not be available")
self.assertEqual(freq_map_array[94], 0, "Lambda 2 should not be available")
- if ele['tp-id'] == 'DEG2-TTP-TXRX':
+ elif ele['tp-id'] == 'DEG2-TTP-TXRX':
freq_map = base64.b64decode(
ele['org-openroadm-network-topology:tx-ttp-attributes']['avail-freq-maps'][0]['freq-map'])
freq_map_array = [int(x) for x in freq_map]
self.assertEqual(freq_map_array[94], 255, "Index 2 should be available")
liste_tp = res['node'][0]['ietf-network-topology:termination-point']
for ele in liste_tp:
- if ele['tp-id'] == 'SRG1-PP1-TXRX' or ele['tp-id'] == 'SRG1-PP2-TXRX':
+ if ele['tp-id'] in ('SRG1-PP1-TXRX', 'SRG1-PP2-TXRX'):
freq_map = base64.b64decode(
ele['org-openroadm-network-topology:pp-attributes']['avail-freq-maps'][0]['freq-map'])
freq_map_array = [int(x) for x in freq_map]
freq_map_array = [int(x) for x in freq_map]
self.assertEqual(freq_map_array[95], 255, "Lambda 1 should be available")
self.assertEqual(freq_map_array[94], 255, "Lambda 2 should be available")
- if ele['tp-id'] == 'DEG2-TTP-TXRX':
+ elif ele['tp-id'] == 'DEG2-TTP-TXRX':
freq_map = base64.b64decode(
ele['org-openroadm-network-topology:tx-ttp-attributes']['avail-freq-maps'][0]['freq-map'])
freq_map_array = [int(x) for x in freq_map]
import test_utils # nopep8
+# pylint: disable=too-few-public-methods
+class UuidServices:
+ def __init__(self):
+ # pylint: disable=invalid-name
+ self.pm = None
+ self.odu = None
+ self.dsr = None
+
+
class TransportPCEtesting(unittest.TestCase):
- processes = None
+ processes = []
WAITING = 20 # nominal value is 300
NODE_VERSION = '2.2.1'
+ uuid_services = UuidServices()
cr_serv_sample_data = {
"input": {
time.sleep(self.WAITING)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
- global SERVICE_PM_UUID
- SERVICE_PM_UUID = res['output']['service']['uuid']
+ self.uuid_services.pm = res['output']['service']['uuid']
# pylint: disable=consider-using-f-string
- print("photonic media service uuid : {}".format(res['output']['service']['uuid']))
+ print("photonic media service uuid : {}".format(self.uuid_services.pm))
input_dict_1 = {'administrative-state': 'LOCKED',
'lifecycle-state': 'PLANNED',
def test_16_get_service_PhotonicMedia(self):
response = test_utils.get_service_list_request(
- "services/" + str(SERVICE_PM_UUID))
+ "services/" + str(self.uuid_services.pm))
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(
res['services'][0]['administrative-state'], 'inService')
self.assertEqual(
- res['services'][0]['service-name'], SERVICE_PM_UUID)
+ res['services'][0]['service-name'], self.uuid_services.pm)
self.assertEqual(
res['services'][0]['connection-type'], 'infrastructure')
self.assertEqual(
time.sleep(self.WAITING)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
- global SERVICE_ODU_UUID
- SERVICE_ODU_UUID = res['output']['service']['uuid']
+ self.uuid_services.odu = res['output']['service']['uuid']
# pylint: disable=consider-using-f-string
- print("odu service uuid : {}".format(res['output']['service']['uuid']))
+ print("odu service uuid : {}".format(self.uuid_services.odu))
input_dict_1 = {'administrative-state': 'LOCKED',
'lifecycle-state': 'PLANNED',
def test_18_get_service_ODU(self):
response = test_utils.get_service_list_request(
- "services/" + str(SERVICE_ODU_UUID))
+ "services/" + str(self.uuid_services.odu))
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(
res['services'][0]['administrative-state'], 'inService')
self.assertEqual(
- res['services'][0]['service-name'], SERVICE_ODU_UUID)
+ res['services'][0]['service-name'], self.uuid_services.odu)
self.assertEqual(
res['services'][0]['connection-type'], 'infrastructure')
self.assertEqual(
time.sleep(self.WAITING)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
- global SERVICE_DSR_UUID
- SERVICE_DSR_UUID = res['output']['service']['uuid']
+ self.uuid_services.dsr = res['output']['service']['uuid']
# pylint: disable=consider-using-f-string
- print("dsr service uuid : {}".format(res['output']['service']['uuid']))
+ print("dsr service uuid : {}".format(self.uuid_services.dsr))
input_dict_1 = {'administrative-state': 'LOCKED',
'lifecycle-state': 'PLANNED',
def test_20_get_service_DSR(self):
response = test_utils.get_service_list_request(
- "services/" + str(SERVICE_DSR_UUID))
+ "services/" + str(self.uuid_services.dsr))
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(
res['services'][0]['administrative-state'], 'inService')
self.assertEqual(
- res['services'][0]['service-name'], SERVICE_DSR_UUID)
+ res['services'][0]['service-name'], self.uuid_services.dsr)
self.assertEqual(
res['services'][0]['connection-type'], 'service')
self.assertEqual(
res = response.json()
liste_service = res['output']['service']
for ele in liste_service:
- if ele['uuid'] == SERVICE_PM_UUID:
+ if ele['uuid'] == self.uuid_services.pm:
self.assertEqual(ele['operational-state'], 'ENABLED')
self.assertEqual(ele['service-layer'], 'PHOTONIC_MEDIA')
nbconnection = len(ele['connection'])
self.assertEqual(nbconnection, 9, 'There should be 9 connections')
- elif ele['uuid'] == SERVICE_ODU_UUID:
+ elif ele['uuid'] == self.uuid_services.odu:
self.assertEqual(ele['operational-state'], 'ENABLED')
self.assertEqual(ele['service-layer'], 'ODU')
nbconnection = len(ele['connection'])
self.assertEqual(nbconnection, 3, 'There should be 3 connections')
- elif ele['uuid'] == SERVICE_DSR_UUID:
+ elif ele['uuid'] == self.uuid_services.dsr:
self.assertEqual(ele['operational-state'], 'ENABLED')
self.assertEqual(ele['service-layer'], 'DSR')
nbconnection = len(ele['connection'])
time.sleep(2)
def test_22_delete_connectivity_service_DSR(self):
- response = test_utils.tapi_delete_connectivity_request(SERVICE_DSR_UUID)
+ response = test_utils.tapi_delete_connectivity_request(self.uuid_services.dsr)
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)
+ response = test_utils.tapi_delete_connectivity_request(self.uuid_services.odu)
self.assertEqual(response.status_code, requests.codes.no_content)
time.sleep(self.WAITING)
def test_24_delete_connectivity_service_PhotonicMedia(self):
- response = test_utils.tapi_delete_connectivity_request(SERVICE_PM_UUID)
+ response = test_utils.tapi_delete_connectivity_request(self.uuid_services.pm)
self.assertEqual(response.status_code, requests.codes.no_content)
time.sleep(self.WAITING)
res = response.json()
links = res['network'][0]['ietf-network-topology:link']
for link in links:
- if ((link["org-openroadm-common-network:link-type"] == "XPONDER-OUTPUT" or
- link["org-openroadm-common-network:link-type"] == "XPONDER-INPUT")
+ if (link["org-openroadm-common-network:link-type"] in ('XPONDER-OUTPUT', 'XPONDER-INPUT')
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)
res = response.json()
links = res['network'][0]['ietf-network-topology:link']
for link in links:
- if (link["org-openroadm-common-network:link-type"] == "XPONDER-OUTPUT" or
- link["org-openroadm-common-network:link-type"] == "XPONDER-INPUT"):
+ if link["org-openroadm-common-network:link-type"] in ('XPONDER-OUTPUT', 'XPONDER-INPUT'):
link_name = link["link-id"]
response = test_utils.delete_request(url+link_name)
self.assertEqual(response.status_code, requests.codes.ok)
##############################################################################
# pylint: disable=no-member
+# pylint: disable=too-many-arguments
import json
import os
def wait_until_log_contains(log_file, regexp, time_to_wait=60):
# pylint: disable=lost-exception
+ # pylint: disable=consider-using-with
stringfound = False
filefound = False
line = None
def wait_until_log_contains(log_file, regexp, time_to_wait=60):
# pylint: disable=lost-exception
+ # pylint: disable=consider-using-with
stringfound = False
filefound = False
line = None
res = response.json()
links = res['network'][0]['ietf-network-topology:link']
for link in links:
- if (link["org-openroadm-common-network:link-type"] == "XPONDER-OUTPUT" or
- link["org-openroadm-common-network:link-type"] == "XPONDER-INPUT"):
+ if link["org-openroadm-common-network:link-type"] in ('XPONDER-OUTPUT', 'XPONDER-INPUT'):
link_name = link["link-id"]
response = test_utils.delete_request(url+link_name)
self.assertEqual(response.status_code, requests.codes.ok)
response = test_utils.get_otn_topo_request()
res = response.json()
for node in res['network'][0]['node']:
- if node['node-id'] == 'XPDR-A2-XPDR2' or node['node-id'] == 'XPDR-C2-XPDR2':
+ if node['node-id'] in ('XPDR-A2-XPDR2', 'XPDR-C2-XPDR2'):
tpList = node['ietf-network-topology:termination-point']
for tp in tpList:
if tp['tp-id'] == 'XPDR2-NETWORK1':
response = test_utils.get_otn_topo_request()
res = response.json()
for node in res['network'][0]['node']:
- if node['node-id'] == 'XPDR-A2-XPDR2' or node['node-id'] == 'XPDR-C2-XPDR2':
+ if node['node-id'] in ('XPDR-A2-XPDR2', 'XPDR-C2-XPDR2'):
tpList = node['ietf-network-topology:termination-point']
for tp in tpList:
if tp['tp-id'] == 'XPDR2-NETWORK1':
response = test_utils.get_otn_topo_request()
res = response.json()
for node in res['network'][0]['node']:
- if node['node-id'] == 'XPDR-A2-XPDR2' or node['node-id'] == 'XPDR-C2-XPDR2':
+ if node['node-id'] in ('XPDR-A2-XPDR2', 'XPDR-C2-XPDR2'):
tpList = node['ietf-network-topology:termination-point']
for tp in tpList:
if tp['tp-id'] == 'XPDR2-NETWORK1':
response = test_utils.get_otn_topo_request()
res = response.json()
for node in res['network'][0]['node']:
- if node['node-id'] == 'XPDR-A2-XPDR2' or node['node-id'] == 'XPDR-C2-XPDR2':
+ if node['node-id'] in ('XPDR-A2-XPDR2', 'XPDR-C2-XPDR2'):
tpList = node['ietf-network-topology:termination-point']
for tp in tpList:
if tp['tp-id'] == 'XPDR2-NETWORK1':
response = test_utils.get_otn_topo_request()
res = response.json()
for node in res['network'][0]['node']:
- if node['node-id'] == 'XPDR-A2-XPDR2' or node['node-id'] == 'XPDR-C2-XPDR2':
+ if node['node-id'] in ('XPDR-A2-XPDR2', 'XPDR-C2-XPDR2'):
tpList = node['ietf-network-topology:termination-point']
for tp in tpList:
if tp['tp-id'] == 'XPDR2-NETWORK1':
res = response.json()
links = res['network'][0]['ietf-network-topology:link']
for link in links:
- if (link["org-openroadm-common-network:link-type"] == "XPONDER-OUTPUT" or
- link["org-openroadm-common-network:link-type"] == "XPONDER-INPUT"):
+ if link["org-openroadm-common-network:link-type"] in ('XPONDER-OUTPUT', 'XPONDER-INPUT'):
link_name = link["link-id"]
response = test_utils.delete_request(url+link_name)
self.assertEqual(response.status_code, requests.codes.ok)
class TransportNbiNotificationstesting(unittest.TestCase):
- processes = None
+ processes = []
cr_serv_sample_data = {"input": {
"sdnc-request-header": {
"request-id": "e3028bae-a90f-4ddd-a83f-cf224eba0e58",
deps = pylint
whitelist_externals = find
commands =
- find transportpce_tests/ -name *.py -exec pylint --max-line-length=120 --disable=missing-docstring --method-rgx="(([a-z_][a-zA-Z0-9_]{2,})|(_[a-z0-9_]*)|(__[a-zA-Z][a-zA-Z0-9_]+__))$" --variable-rgx="[a-zA-Z_][a-zA-Z0-9_]{1,30}$" --reports=y --score=y --output-format=colorized \{\} +
+ find transportpce_tests/ -name *.py -exec pylint --max-line-length=120 --disable=missing-docstring --module-rgx="([a-z0-9_]+$)|([0-9.]{1,30}$)" --method-rgx="(([a-z_][a-zA-Z0-9_]{2,})|(_[a-z0-9_]*)|(__[a-zA-Z][a-zA-Z0-9_]+__))$" --variable-rgx="[a-zA-Z_][a-zA-Z0-9_]{1,30}$" --reports=y --score=y --output-format=colorized \{\} +
[testenv:pyang]
basepython = python3