From a8343688872d307ada1b4521721313a8ac3a3e5e Mon Sep 17 00:00:00 2001 From: "guillaume.lambert" Date: Mon, 29 Nov 2021 14:38:38 +0100 Subject: [PATCH] Fix remaining pylint warnings apart fixme & dup JIRA: TRNSPRTPCE-320 Signed-off-by: guillaume.lambert Change-Id: I256c36f1059783a26fa0e30292de3c2c1781376b --- .../1.2.1/test02_topo_portmapping.py | 12 ++- .../1.2.1/test06_end2end.py | 29 ++++---- .../2.2.1/test02_topo_portmapping.py | 10 +-- .../2.2.1/test04_otn_topology.py | 74 ++++++++----------- tests/transportpce_tests/2.2.1/test10_tapi.py | 3 +- .../2.2.1/test11_otn_end2end.py | 15 ++-- .../2.2.1/test12_end2end.py | 22 +++--- .../2.2.1/test13_tapi_full_multi_layer.py | 51 +++++++------ .../2.2.1/test14_otn_switch_end2end.py | 6 +- tests/transportpce_tests/common/test_utils.py | 2 + .../common/test_utils_rfc8040.py | 1 + .../test01_device_change_notifications.py | 3 +- .../hybrid/test02_B100G_end2end.py | 13 ++-- .../with_docker/test02_nbinotifications.py | 2 +- tox.ini | 2 +- 15 files changed, 113 insertions(+), 132 deletions(-) diff --git a/tests/transportpce_tests/1.2.1/test02_topo_portmapping.py b/tests/transportpce_tests/1.2.1/test02_topo_portmapping.py index a2f88b5f6..f8ebc2b2f 100644 --- a/tests/transportpce_tests/1.2.1/test02_topo_portmapping.py +++ b/tests/transportpce_tests/1.2.1/test02_topo_portmapping.py @@ -54,16 +54,14 @@ class TransportPCEtesting(unittest.TestCase): 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) diff --git a/tests/transportpce_tests/1.2.1/test06_end2end.py b/tests/transportpce_tests/1.2.1/test06_end2end.py index c8f4eac61..a9f84435c 100644 --- a/tests/transportpce_tests/1.2.1/test06_end2end.py +++ b/tests/transportpce_tests/1.2.1/test06_end2end.py @@ -309,9 +309,9 @@ class TransportPCEFulltesting(unittest.TestCase): 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) @@ -330,7 +330,7 @@ class TransportPCEFulltesting(unittest.TestCase): 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) @@ -349,7 +349,7 @@ class TransportPCEFulltesting(unittest.TestCase): 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] @@ -448,11 +448,11 @@ class TransportPCEFulltesting(unittest.TestCase): 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) @@ -473,13 +473,13 @@ class TransportPCEFulltesting(unittest.TestCase): 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) @@ -500,7 +500,7 @@ class TransportPCEFulltesting(unittest.TestCase): 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] @@ -567,15 +567,12 @@ class TransportPCEFulltesting(unittest.TestCase): 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'])) @@ -628,7 +625,7 @@ class TransportPCEFulltesting(unittest.TestCase): 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] diff --git a/tests/transportpce_tests/2.2.1/test02_topo_portmapping.py b/tests/transportpce_tests/2.2.1/test02_topo_portmapping.py index 06e83473b..7e228ccb7 100644 --- a/tests/transportpce_tests/2.2.1/test02_topo_portmapping.py +++ b/tests/transportpce_tests/2.2.1/test02_topo_portmapping.py @@ -52,11 +52,10 @@ class TransportPCEtesting(unittest.TestCase): 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] @@ -65,10 +64,9 @@ class TransportPCEtesting(unittest.TestCase): 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) diff --git a/tests/transportpce_tests/2.2.1/test04_otn_topology.py b/tests/transportpce_tests/2.2.1/test04_otn_topology.py index 41b1f65b0..7a72b34a4 100644 --- a/tests/transportpce_tests/2.2.1/test04_otn_topology.py +++ b/tests/transportpce_tests/2.2.1/test04_otn_topology.py @@ -116,7 +116,7 @@ class TransportPCEtesting(unittest.TestCase): 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': @@ -135,8 +135,7 @@ class TransportPCEtesting(unittest.TestCase): 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': { @@ -162,66 +161,51 @@ class TransportPCEtesting(unittest.TestCase): '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) diff --git a/tests/transportpce_tests/2.2.1/test10_tapi.py b/tests/transportpce_tests/2.2.1/test10_tapi.py index 6b5ef4b41..f0a429fcb 100644 --- a/tests/transportpce_tests/2.2.1/test10_tapi.py +++ b/tests/transportpce_tests/2.2.1/test10_tapi.py @@ -695,8 +695,7 @@ class TransportTapitesting(unittest.TestCase): 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) diff --git a/tests/transportpce_tests/2.2.1/test11_otn_end2end.py b/tests/transportpce_tests/2.2.1/test11_otn_end2end.py index 9d6bcd84b..f448ba331 100644 --- a/tests/transportpce_tests/2.2.1/test11_otn_end2end.py +++ b/tests/transportpce_tests/2.2.1/test11_otn_end2end.py @@ -566,7 +566,7 @@ class TransportPCEtesting(unittest.TestCase): 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': @@ -816,7 +816,7 @@ class TransportPCEtesting(unittest.TestCase): 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': @@ -886,7 +886,7 @@ class TransportPCEtesting(unittest.TestCase): 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': @@ -922,7 +922,7 @@ class TransportPCEtesting(unittest.TestCase): 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': @@ -1372,7 +1372,7 @@ class TransportPCEtesting(unittest.TestCase): 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': @@ -1442,7 +1442,7 @@ class TransportPCEtesting(unittest.TestCase): 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': @@ -1474,8 +1474,7 @@ class TransportPCEtesting(unittest.TestCase): 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) diff --git a/tests/transportpce_tests/2.2.1/test12_end2end.py b/tests/transportpce_tests/2.2.1/test12_end2end.py index 9472e3892..090cd536e 100644 --- a/tests/transportpce_tests/2.2.1/test12_end2end.py +++ b/tests/transportpce_tests/2.2.1/test12_end2end.py @@ -293,9 +293,9 @@ class TransportPCEFulltesting(unittest.TestCase): 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) @@ -314,7 +314,7 @@ class TransportPCEFulltesting(unittest.TestCase): 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) @@ -333,7 +333,7 @@ class TransportPCEFulltesting(unittest.TestCase): 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] @@ -426,11 +426,11 @@ class TransportPCEFulltesting(unittest.TestCase): 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) @@ -451,13 +451,13 @@ class TransportPCEFulltesting(unittest.TestCase): 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) @@ -478,7 +478,7 @@ class TransportPCEFulltesting(unittest.TestCase): 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] @@ -556,7 +556,7 @@ class TransportPCEFulltesting(unittest.TestCase): 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] @@ -589,7 +589,7 @@ class TransportPCEFulltesting(unittest.TestCase): 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] diff --git a/tests/transportpce_tests/2.2.1/test13_tapi_full_multi_layer.py b/tests/transportpce_tests/2.2.1/test13_tapi_full_multi_layer.py index 752256110..348610107 100644 --- a/tests/transportpce_tests/2.2.1/test13_tapi_full_multi_layer.py +++ b/tests/transportpce_tests/2.2.1/test13_tapi_full_multi_layer.py @@ -24,11 +24,21 @@ sys.path.append('transportpce_tests/common/') 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": { @@ -256,10 +266,9 @@ class TransportPCEtesting(unittest.TestCase): 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', @@ -283,13 +292,13 @@ class TransportPCEtesting(unittest.TestCase): 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( @@ -309,10 +318,9 @@ class TransportPCEtesting(unittest.TestCase): 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', @@ -336,13 +344,13 @@ class TransportPCEtesting(unittest.TestCase): 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( @@ -363,10 +371,9 @@ class TransportPCEtesting(unittest.TestCase): 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', @@ -393,13 +400,13 @@ class TransportPCEtesting(unittest.TestCase): 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( @@ -412,17 +419,17 @@ class TransportPCEtesting(unittest.TestCase): 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']) @@ -432,17 +439,17 @@ class TransportPCEtesting(unittest.TestCase): 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) diff --git a/tests/transportpce_tests/2.2.1/test14_otn_switch_end2end.py b/tests/transportpce_tests/2.2.1/test14_otn_switch_end2end.py index ab5a56225..fad074f23 100644 --- a/tests/transportpce_tests/2.2.1/test14_otn_switch_end2end.py +++ b/tests/transportpce_tests/2.2.1/test14_otn_switch_end2end.py @@ -1288,8 +1288,7 @@ class TransportPCEtesting(unittest.TestCase): 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) @@ -1497,8 +1496,7 @@ class TransportPCEtesting(unittest.TestCase): 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) diff --git a/tests/transportpce_tests/common/test_utils.py b/tests/transportpce_tests/common/test_utils.py index c18a4d1f5..cfb5f92e0 100644 --- a/tests/transportpce_tests/common/test_utils.py +++ b/tests/transportpce_tests/common/test_utils.py @@ -10,6 +10,7 @@ ############################################################################## # pylint: disable=no-member +# pylint: disable=too-many-arguments import json import os @@ -507,6 +508,7 @@ def start_honeynode(log_file: str, sim): 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 diff --git a/tests/transportpce_tests/common/test_utils_rfc8040.py b/tests/transportpce_tests/common/test_utils_rfc8040.py index e1323f217..825e12102 100644 --- a/tests/transportpce_tests/common/test_utils_rfc8040.py +++ b/tests/transportpce_tests/common/test_utils_rfc8040.py @@ -193,6 +193,7 @@ def start_honeynode(log_file: str, sim): 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 diff --git a/tests/transportpce_tests/hybrid/test01_device_change_notifications.py b/tests/transportpce_tests/hybrid/test01_device_change_notifications.py index 9ab4ae2d5..275c55ce0 100644 --- a/tests/transportpce_tests/hybrid/test01_device_change_notifications.py +++ b/tests/transportpce_tests/hybrid/test01_device_change_notifications.py @@ -723,8 +723,7 @@ class TransportPCEFulltesting(unittest.TestCase): 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) diff --git a/tests/transportpce_tests/hybrid/test02_B100G_end2end.py b/tests/transportpce_tests/hybrid/test02_B100G_end2end.py index 988d42408..510303a9c 100644 --- a/tests/transportpce_tests/hybrid/test02_B100G_end2end.py +++ b/tests/transportpce_tests/hybrid/test02_B100G_end2end.py @@ -589,7 +589,7 @@ class TransportPCEtesting(unittest.TestCase): 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': @@ -845,7 +845,7 @@ class TransportPCEtesting(unittest.TestCase): 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': @@ -919,7 +919,7 @@ class TransportPCEtesting(unittest.TestCase): 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': @@ -997,7 +997,7 @@ class TransportPCEtesting(unittest.TestCase): 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': @@ -1033,7 +1033,7 @@ class TransportPCEtesting(unittest.TestCase): 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': @@ -1406,8 +1406,7 @@ class TransportPCEtesting(unittest.TestCase): 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) diff --git a/tests/transportpce_tests/with_docker/test02_nbinotifications.py b/tests/transportpce_tests/with_docker/test02_nbinotifications.py index 235f75ef7..8164f8403 100644 --- a/tests/transportpce_tests/with_docker/test02_nbinotifications.py +++ b/tests/transportpce_tests/with_docker/test02_nbinotifications.py @@ -25,7 +25,7 @@ import test_utils # nopep8 class TransportNbiNotificationstesting(unittest.TestCase): - processes = None + processes = [] cr_serv_sample_data = {"input": { "sdnc-request-header": { "request-id": "e3028bae-a90f-4ddd-a83f-cf224eba0e58", diff --git a/tox.ini b/tox.ini index b0f4896b9..8e9528a65 100644 --- a/tox.ini +++ b/tox.ini @@ -227,7 +227,7 @@ basepython = python3 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 -- 2.36.6