From 7e2c734da1f8132462153fa5e055b9c9b804d959 Mon Sep 17 00:00:00 2001 From: "guillaume.lambert" Date: Thu, 28 Oct 2021 10:50:05 +0200 Subject: [PATCH] Run autopep8 tox profile Signed-off-by: guillaume.lambert Change-Id: I27d64e316a00de3ec21936f3427f1c63405e62da --- .../test04_renderer_service_path_nominal.py | 4 +- tests/transportpce_tests/1.2.1/test05_olm.py | 39 ++++--- .../1.2.1/test06_end2end.py | 27 +++-- .../2.2.1/test03_otn_topology.py | 13 ++- .../2.2.1/test04_flex_grid.py | 108 +++++++++--------- .../test05_renderer_service_path_nominal.py | 3 +- tests/transportpce_tests/2.2.1/test08_olm.py | 27 +++-- tests/transportpce_tests/2.2.1/test09_tapi.py | 26 +++-- .../2.2.1/test11_end2end.py | 28 +++-- .../7.1/test01_portmapping.py | 16 +-- tests/transportpce_tests/common/simulators.py | 2 +- tests/transportpce_tests/common/test_utils.py | 3 + .../test01_device_change_notifications.py | 101 ++++++++-------- .../hybrid/test02_B100G_end2end.py | 55 +++++---- tests/transportpce_tests/pce/test01_pce.py | 7 +- .../transportpce_tests/pce/test02_pce_400G.py | 22 ++-- .../with_docker/test01_gnpy.py | 4 +- .../with_docker/test02_nbinotifications.py | 1 - 18 files changed, 257 insertions(+), 229 deletions(-) diff --git a/tests/transportpce_tests/1.2.1/test04_renderer_service_path_nominal.py b/tests/transportpce_tests/1.2.1/test04_renderer_service_path_nominal.py index 305c2ec89..3a3368fe4 100644 --- a/tests/transportpce_tests/1.2.1/test04_renderer_service_path_nominal.py +++ b/tests/transportpce_tests/1.2.1/test04_renderer_service_path_nominal.py @@ -19,6 +19,7 @@ import sys sys.path.append('transportpce_tests/common/') import test_utils + class TransportPCERendererTesting(unittest.TestCase): processes = None @@ -289,7 +290,8 @@ class TransportPCERendererTesting(unittest.TestCase): res['errors']['error']) def test_17_service_path_delete_rdm_check(self): - response = test_utils.check_netconf_node_request("ROADMA01", "roadm-connections/SRG1-PP7-TXRX-DEG1-TTP-TXRX-713:720") + response = test_utils.check_netconf_node_request( + "ROADMA01", "roadm-connections/SRG1-PP7-TXRX-DEG1-TTP-TXRX-713:720") self.assertEqual(response.status_code, requests.codes.conflict) res = response.json() self.assertIn( diff --git a/tests/transportpce_tests/1.2.1/test05_olm.py b/tests/transportpce_tests/1.2.1/test05_olm.py index 96778f5ac..0443b5ca1 100644 --- a/tests/transportpce_tests/1.2.1/test05_olm.py +++ b/tests/transportpce_tests/1.2.1/test05_olm.py @@ -227,8 +227,8 @@ class TransportOlmTesting(unittest.TestCase): "dest-tp": "SRG1-PP1-TXRX", "src-tp": "DEG2-TTP-TXRX"}, {"node-id": "XPDRC01", "dest-tp": "XPDR1-CLIENT1", "src-tp": "XPDR1-NETWORK1"}], - 196.1, 40, 196.075, 196.125, 761, - 768) + 196.1, 40, 196.075, 196.125, 761, + 768) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn('Roadm-connection successfully created for nodes', res["output"]["result"]) @@ -245,8 +245,8 @@ class TransportOlmTesting(unittest.TestCase): "src-tp": "DEG1-TTP-TXRX", "dest-tp": "SRG1-PP1-TXRX"}, {"node-id": "XPDRA01", "src-tp": "XPDR1-NETWORK1", "dest-tp": "XPDR1-CLIENT1"}], - 196.1, 40, 196.075, 196.125, 761, - 768) + 196.1, 40, 196.075, 196.125, 761, + 768) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn('Roadm-connection successfully created for nodes', res["output"]["result"]) @@ -304,14 +304,16 @@ class TransportOlmTesting(unittest.TestCase): self.assertEqual(1, res['org-openroadm-optical-channel-interfaces:och']['wavelength-number']) def test_21_get_roadmconnection_ROADMA(self): - response = test_utils.check_netconf_node_request("ROADMA01", "roadm-connections/SRG1-PP1-TXRX-DEG1-TTP-TXRX-761:768") + response = test_utils.check_netconf_node_request( + "ROADMA01", "roadm-connections/SRG1-PP1-TXRX-DEG1-TTP-TXRX-761:768") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertEqual("gainLoss", res['roadm-connections'][0]['opticalControlMode']) self.assertEqual(-3.3, res['roadm-connections'][0]['target-output-power']) def test_22_get_roadmconnection_ROADMC(self): - response = test_utils.check_netconf_node_request("ROADMC01", "roadm-connections/DEG2-TTP-TXRX-SRG1-PP1-TXRX-761:768") + response = test_utils.check_netconf_node_request( + "ROADMC01", "roadm-connections/DEG2-TTP-TXRX-SRG1-PP1-TXRX-761:768") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertEqual("power", res['roadm-connections'][0]['opticalControlMode']) @@ -367,7 +369,8 @@ class TransportOlmTesting(unittest.TestCase): self.assertEqual(1, res['org-openroadm-optical-channel-interfaces:och']['wavelength-number']) def test_25_get_roadmconnection_ROADMC(self): - response = test_utils.check_netconf_node_request("ROADMC01", "roadm-connections/SRG1-PP1-TXRX-DEG2-TTP-TXRX-761:768") + response = test_utils.check_netconf_node_request( + "ROADMC01", "roadm-connections/SRG1-PP1-TXRX-DEG2-TTP-TXRX-761:768") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertEqual("gainLoss", res['roadm-connections'][0]['opticalControlMode']) @@ -416,14 +419,16 @@ class TransportOlmTesting(unittest.TestCase): self.assertIn('Success', res["output"]["result"]) def test_27_get_roadmconnection_ROADMA(self): - response = test_utils.check_netconf_node_request("ROADMA01", "roadm-connections/SRG1-PP1-TXRX-DEG1-TTP-TXRX-761:768") + response = test_utils.check_netconf_node_request( + "ROADMA01", "roadm-connections/SRG1-PP1-TXRX-DEG1-TTP-TXRX-761:768") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertEqual("off", res['roadm-connections'][0]['opticalControlMode']) self.assertEqual(-60, res['roadm-connections'][0]['target-output-power']) def test_28_get_roadmconnection_ROADMC(self): - response = test_utils.check_netconf_node_request("ROADMC01", "roadm-connections/DEG2-TTP-TXRX-SRG1-PP1-TXRX-761:768") + response = test_utils.check_netconf_node_request( + "ROADMC01", "roadm-connections/DEG2-TTP-TXRX-SRG1-PP1-TXRX-761:768") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertEqual("off", res['roadm-connections'][0]['opticalControlMode']) @@ -438,8 +443,8 @@ class TransportOlmTesting(unittest.TestCase): "dest-tp": "SRG1-PP1-TXRX", "src-tp": "DEG2-TTP-TXRX"}, {"node-id": "XPDRC01", "dest-tp": "XPDR1-CLIENT1", "src-tp": "XPDR1-NETWORK1"}], - 196.1, 40, 196.075, 196.125, 761, - 768) + 196.1, 40, 196.075, 196.125, 761, + 768) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn('Request processed', res["output"]["result"]) @@ -455,8 +460,8 @@ class TransportOlmTesting(unittest.TestCase): "src-tp": "DEG1-TTP-TXRX", "dest-tp": "SRG1-PP1-TXRX"}, {"node-id": "XPDRA01", "src-tp": "XPDR1-NETWORK1", "dest-tp": "XPDR1-CLIENT1"}], - 196.1, 40, 196.075, 196.125, 761, - 768) + 196.1, 40, 196.075, 196.125, 761, + 768) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn('Request processed', res["output"]["result"]) @@ -484,8 +489,8 @@ class TransportOlmTesting(unittest.TestCase): "dest-tp": "XPDR1-NETWORK2", "src-tp": "XPDR1-CLIENT2"}, {"node-id": "ROADMA01", "dest-tp": "DEG1-TTP-TXRX", "src-tp": "SRG1-PP2-TXRX"}], - 196.05, 40, 196.025, 196.075, 753, - 760) + 196.05, 40, 196.025, 196.075, 753, + 760) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn('Roadm-connection successfully created for nodes', res["output"]["result"]) @@ -507,8 +512,8 @@ class TransportOlmTesting(unittest.TestCase): "dest-tp": "XPDR1-NETWORK2", "src-tp": "XPDR1-CLIENT2"}, {"node-id": "ROADMA01", "dest-tp": "DEG1-TTP-TXRX", "src-tp": "SRG1-PP2-TXRX"}], - 196.1, 40, 196.075, 196.125, 761, - 768) + 196.1, 40, 196.075, 196.125, 761, + 768) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn('Request processed', res["output"]["result"]) diff --git a/tests/transportpce_tests/1.2.1/test06_end2end.py b/tests/transportpce_tests/1.2.1/test06_end2end.py index 3f7ca3b52..aba898bf9 100644 --- a/tests/transportpce_tests/1.2.1/test06_end2end.py +++ b/tests/transportpce_tests/1.2.1/test06_end2end.py @@ -19,6 +19,7 @@ import sys sys.path.append('transportpce_tests/common/') import test_utils + class TransportPCEFulltesting(unittest.TestCase): cr_serv_sample_data = {"input": { "sdnc-request-header": { @@ -249,7 +250,8 @@ class TransportPCEFulltesting(unittest.TestCase): time.sleep(2) def test_13_check_xc1_ROADMA(self): - response = test_utils.check_netconf_node_request("ROADMA01", "roadm-connections/SRG1-PP1-TXRX-DEG1-TTP-TXRX-761:768") + response = test_utils.check_netconf_node_request( + "ROADMA01", "roadm-connections/SRG1-PP1-TXRX-DEG1-TTP-TXRX-761:768") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2 @@ -271,7 +273,8 @@ class TransportPCEFulltesting(unittest.TestCase): time.sleep(5) def test_14_check_xc1_ROADMC(self): - response = test_utils.check_netconf_node_request("ROADMC01", "roadm-connections/SRG1-PP1-TXRX-DEG2-TTP-TXRX-761:768") + response = test_utils.check_netconf_node_request( + "ROADMC01", "roadm-connections/SRG1-PP1-TXRX-DEG2-TTP-TXRX-761:768") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2 @@ -343,10 +346,10 @@ 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") if 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[95], 0, "Lambda 1 should not be available") + 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[95], 0, "Lambda 1 should not be available") time.sleep(3) def test_18_connect_xprdA_N2_to_roadmA_PP2(self): @@ -411,7 +414,8 @@ class TransportPCEFulltesting(unittest.TestCase): time.sleep(1) def test_24_check_xc2_ROADMA(self): - response = test_utils.check_netconf_node_request("ROADMA01", "roadm-connections/DEG1-TTP-TXRX-SRG1-PP2-TXRX-753:760") + response = test_utils.check_netconf_node_request( + "ROADMA01", "roadm-connections/DEG1-TTP-TXRX-SRG1-PP2-TXRX-753:760") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2 @@ -660,7 +664,8 @@ class TransportPCEFulltesting(unittest.TestCase): time.sleep(1) def test_38_check_xc1_ROADMA(self): - response = test_utils.check_netconf_node_request("ROADMA01", "roadm-connections/SRG1-PP1-TXRX-DEG1-TTP-TXRX-761:768") + response = test_utils.check_netconf_node_request( + "ROADMA01", "roadm-connections/SRG1-PP1-TXRX-DEG1-TTP-TXRX-761:768") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2 @@ -682,7 +687,8 @@ class TransportPCEFulltesting(unittest.TestCase): time.sleep(7) def test_39_check_xc1_ROADMC(self): - response = test_utils.check_netconf_node_request("ROADMC01", "roadm-connections/SRG1-PP1-TXRX-DEG2-TTP-TXRX-761:768") + response = test_utils.check_netconf_node_request( + "ROADMC01", "roadm-connections/SRG1-PP1-TXRX-DEG2-TTP-TXRX-761:768") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2 @@ -734,7 +740,8 @@ class TransportPCEFulltesting(unittest.TestCase): time.sleep(2) def test_42_check_xc2_ROADMA(self): - response = test_utils.check_netconf_node_request("ROADMA01", "roadm-connections/SRG1-PP2-TXRX-DEG1-TTP-TXRX-753:760") + response = test_utils.check_netconf_node_request( + "ROADMA01", "roadm-connections/SRG1-PP2-TXRX-DEG1-TTP-TXRX-753:760") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2 diff --git a/tests/transportpce_tests/2.2.1/test03_otn_topology.py b/tests/transportpce_tests/2.2.1/test03_otn_topology.py index 84e3652c7..4ca1e226b 100644 --- a/tests/transportpce_tests/2.2.1/test03_otn_topology.py +++ b/tests/transportpce_tests/2.2.1/test03_otn_topology.py @@ -183,15 +183,16 @@ class TransportPCEtesting(unittest.TestCase): if tpType == 'XPONDER-CLIENT': client += 1 print("tpId = {}".format(tpId)) - print("tp= {}".format(res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j])) + 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]) + ['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']) + ['org-openroadm-otn-network-topology:tp-supported-interfaces'] + ['supported-interface-capability'][0]['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] diff --git a/tests/transportpce_tests/2.2.1/test04_flex_grid.py b/tests/transportpce_tests/2.2.1/test04_flex_grid.py index b48c9aacc..31b64cabe 100644 --- a/tests/transportpce_tests/2.2.1/test04_flex_grid.py +++ b/tests/transportpce_tests/2.2.1/test04_flex_grid.py @@ -71,13 +71,13 @@ class TransportPCEPortMappingTesting(unittest.TestCase): res = response.json() self.assertEqual( { - "mc-capabilities": [ - { - "mc-node-name": "DEG1-TTP", - "center-freq-granularity": 6.25, - "slot-width-granularity": 12.5 - } - ] + "mc-capabilities": [ + { + "mc-node-name": "DEG1-TTP", + "center-freq-granularity": 6.25, + "slot-width-granularity": 12.5 + } + ] }, res) time.sleep(3) @@ -87,13 +87,13 @@ class TransportPCEPortMappingTesting(unittest.TestCase): res = response.json() self.assertEqual( { - "mc-capabilities": [ - { - "mc-node-name": "DEG2-TTP", - "center-freq-granularity": 6.25, - "slot-width-granularity": 12.5 - } - ] + "mc-capabilities": [ + { + "mc-node-name": "DEG2-TTP", + "center-freq-granularity": 6.25, + "slot-width-granularity": 12.5 + } + ] }, res) time.sleep(3) @@ -103,13 +103,13 @@ class TransportPCEPortMappingTesting(unittest.TestCase): res = response.json() self.assertEqual( { - "mc-capabilities": [ - { - "mc-node-name": "SRG1-PP", - "center-freq-granularity": 6.25, - "slot-width-granularity": 12.5 - } - ] + "mc-capabilities": [ + { + "mc-node-name": "SRG1-PP", + "center-freq-granularity": 6.25, + "slot-width-granularity": 12.5 + } + ] }, res) time.sleep(3) @@ -117,27 +117,27 @@ class TransportPCEPortMappingTesting(unittest.TestCase): def test_07_device_renderer(self): data = { "transportpce-device-renderer:input": { - "transportpce-device-renderer:modulation-format": "dp-qpsk", - "transportpce-device-renderer:operation": "create", - "transportpce-device-renderer:service-name": "testNMC-MC", - "transportpce-device-renderer:wave-number": "0", - "transportpce-device-renderer:center-freq": "196.05", - "transportpce-device-renderer:width": "80", - "transportpce-device-renderer:nodes": [ - { - "transportpce-device-renderer:node-id": "ROADM-D1", - "transportpce-device-renderer:src-tp": "SRG1-PP1-TXRX", - "transportpce-device-renderer:dest-tp": "DEG1-TTP-TXRX" - } - ], - "transportpce-device-renderer:min-freq": 196.00625, - "transportpce-device-renderer:max-freq": 196.09375, - "transportpce-device-renderer:lower-spectral-slot-number": 749, - "transportpce-device-renderer:higher-spectral-slot-number": 763 - } + "transportpce-device-renderer:modulation-format": "dp-qpsk", + "transportpce-device-renderer:operation": "create", + "transportpce-device-renderer:service-name": "testNMC-MC", + "transportpce-device-renderer:wave-number": "0", + "transportpce-device-renderer:center-freq": "196.05", + "transportpce-device-renderer:width": "80", + "transportpce-device-renderer:nodes": [ + { + "transportpce-device-renderer:node-id": "ROADM-D1", + "transportpce-device-renderer:src-tp": "SRG1-PP1-TXRX", + "transportpce-device-renderer:dest-tp": "DEG1-TTP-TXRX" + } + ], + "transportpce-device-renderer:min-freq": 196.00625, + "transportpce-device-renderer:max-freq": 196.09375, + "transportpce-device-renderer:lower-spectral-slot-number": 749, + "transportpce-device-renderer:higher-spectral-slot-number": 763 + } } url = test_utils.RESTCONF_BASE_URL + \ - "/operations/transportpce-device-renderer:service-path" + "/operations/transportpce-device-renderer:service-path" response = test_utils.post_request(url, data) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() @@ -159,14 +159,14 @@ class TransportPCEPortMappingTesting(unittest.TestCase): "description": "TBD", "supporting-port": "L1", "type": "org-openroadm-interfaces:mediaChannelTrailTerminationPoint"}, - **res['interface'][0]), + **res['interface'][0]), res['interface'][0]) # Check the mc-ttp max and min-freq self.assertEqual({ - "min-freq": 196.00625, - "max-freq": 196.09375 - }, + "min-freq": 196.00625, + "max-freq": 196.09375 + }, res['interface'][0]['org-openroadm-media-channel-interfaces:mc-ttp']) time.sleep(3) @@ -189,10 +189,10 @@ class TransportPCEPortMappingTesting(unittest.TestCase): # Check the mc-ttp max and min-freq self.assertEqual({ - "frequency": 196.05, - "width": 80 + "frequency": 196.05, + "width": 80 }, - res['interface'][0]['org-openroadm-network-media-channel-interfaces:nmc-ctp']) + res['interface'][0]['org-openroadm-network-media-channel-interfaces:nmc-ctp']) time.sleep(3) # get SRG-NMC interface @@ -216,7 +216,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase): # Create ROADM-connection def test_11_roadm_connection(self): response = test_utils.check_netconf_node_request("ROADM-D1", - "roadm-connections/SRG1-PP1-TXRX-DEG1-TTP-TXRX-749:763") + "roadm-connections/SRG1-PP1-TXRX-DEG1-TTP-TXRX-749:763") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertEqual("SRG1-PP1-TXRX-DEG1-TTP-TXRX-749:763", @@ -232,18 +232,18 @@ class TransportPCEPortMappingTesting(unittest.TestCase): # delete ROADM connection def test_12_delete_roadm_connection(self): response = test_utils.delete_request(test_utils.URL_CONFIG_NETCONF_TOPO + - "node/ROADM-D1/yang-ext:mount/" + - "org-openroadm-device:org-openroadm-device/" + - "roadm-connections/SRG1-PP1-TXRX-DEG1-TTP-TXRX-749:763") + "node/ROADM-D1/yang-ext:mount/" + + "org-openroadm-device:org-openroadm-device/" + + "roadm-connections/SRG1-PP1-TXRX-DEG1-TTP-TXRX-749:763") self.assertEqual(response.status_code, requests.codes.ok) time.sleep(3) # Delete NMC SRG interface def test_13_delete_srg_interface(self): response = test_utils.delete_request(test_utils.URL_CONFIG_NETCONF_TOPO + - "node/ROADM-D1/yang-ext:mount/" + - "org-openroadm-device:org-openroadm-device/" + - "interface/SRG1-PP1-TXRX-nmc-749:763") + "node/ROADM-D1/yang-ext:mount/" + + "org-openroadm-device:org-openroadm-device/" + + "interface/SRG1-PP1-TXRX-nmc-749:763") self.assertEqual(response.status_code, requests.codes.ok) time.sleep(3) diff --git a/tests/transportpce_tests/2.2.1/test05_renderer_service_path_nominal.py b/tests/transportpce_tests/2.2.1/test05_renderer_service_path_nominal.py index 0c2ab278a..c4f263da4 100644 --- a/tests/transportpce_tests/2.2.1/test05_renderer_service_path_nominal.py +++ b/tests/transportpce_tests/2.2.1/test05_renderer_service_path_nominal.py @@ -164,7 +164,8 @@ class TransportPCERendererTesting(unittest.TestCase): res['errors']['error']) def test_10_service_path_create_rdm_check(self): - response = test_utils.check_netconf_node_request("ROADM-A1", "roadm-connections/SRG1-PP3-TXRX-DEG1-TTP-TXRX-713:720") + response = test_utils.check_netconf_node_request( + "ROADM-A1", "roadm-connections/SRG1-PP3-TXRX-DEG1-TTP-TXRX-713:720") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2 diff --git a/tests/transportpce_tests/2.2.1/test08_olm.py b/tests/transportpce_tests/2.2.1/test08_olm.py index 67e609fa7..4c5316963 100644 --- a/tests/transportpce_tests/2.2.1/test08_olm.py +++ b/tests/transportpce_tests/2.2.1/test08_olm.py @@ -229,7 +229,7 @@ class TransportOlmTesting(unittest.TestCase): {"node-id": "XPDR-C1", "dest-tp": "XPDR1-CLIENT1", "src-tp": "XPDR1-NETWORK1"}], 196.1, 40, 196.075, 196.125, 761, - 768) + 768) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn('Roadm-connection successfully created for nodes', res["output"]["result"]) @@ -247,7 +247,7 @@ class TransportOlmTesting(unittest.TestCase): {"node-id": "XPDR-A1", "src-tp": "XPDR1-NETWORK1", "dest-tp": "XPDR1-CLIENT1"}], 196.1, 40, 196.075, 196.125, 761, - 768) + 768) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn('Roadm-connection successfully created for nodes', res["output"]["result"]) @@ -301,14 +301,16 @@ class TransportOlmTesting(unittest.TestCase): self.assertEqual(196.1, res['org-openroadm-optical-channel-interfaces:och']['frequency']) def test_21_get_roadmconnection_ROADMA(self): - response = test_utils.check_netconf_node_request("ROADM-A1", "roadm-connections/SRG1-PP1-TXRX-DEG2-TTP-TXRX-761:768") + response = test_utils.check_netconf_node_request( + "ROADM-A1", "roadm-connections/SRG1-PP1-TXRX-DEG2-TTP-TXRX-761:768") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertEqual("gainLoss", res['roadm-connections'][0]['opticalControlMode']) self.assertEqual(2.0, res['roadm-connections'][0]['target-output-power']) def test_22_get_roadmconnection_ROADMC(self): - response = test_utils.check_netconf_node_request("ROADM-C1", "roadm-connections/DEG1-TTP-TXRX-SRG1-PP1-TXRX-761:768") + response = test_utils.check_netconf_node_request( + "ROADM-C1", "roadm-connections/DEG1-TTP-TXRX-SRG1-PP1-TXRX-761:768") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertEqual("power", res['roadm-connections'][0]['opticalControlMode']) @@ -360,7 +362,8 @@ class TransportOlmTesting(unittest.TestCase): self.assertEqual(196.1, res['org-openroadm-optical-channel-interfaces:och']['frequency']) def test_25_get_roadmconnection_ROADMC(self): - response = test_utils.check_netconf_node_request("ROADM-C1", "roadm-connections/SRG1-PP1-TXRX-DEG1-TTP-TXRX-761:768") + response = test_utils.check_netconf_node_request( + "ROADM-C1", "roadm-connections/SRG1-PP1-TXRX-DEG1-TTP-TXRX-761:768") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertEqual("gainLoss", res['roadm-connections'][0]['opticalControlMode']) @@ -404,14 +407,16 @@ class TransportOlmTesting(unittest.TestCase): self.assertIn('Success', res["output"]["result"]) def test_27_get_roadmconnection_ROADMA(self): - response = test_utils.check_netconf_node_request("ROADM-A1", "roadm-connections/SRG1-PP1-TXRX-DEG2-TTP-TXRX-761:768") + response = test_utils.check_netconf_node_request( + "ROADM-A1", "roadm-connections/SRG1-PP1-TXRX-DEG2-TTP-TXRX-761:768") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertEqual("off", res['roadm-connections'][0]['opticalControlMode']) self.assertEqual(-60, res['roadm-connections'][0]['target-output-power']) def test_28_get_roadmconnection_ROADMC(self): - response = test_utils.check_netconf_node_request("ROADM-C1", "roadm-connections/DEG1-TTP-TXRX-SRG1-PP1-TXRX-761:768") + response = test_utils.check_netconf_node_request( + "ROADM-C1", "roadm-connections/DEG1-TTP-TXRX-SRG1-PP1-TXRX-761:768") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertEqual("off", res['roadm-connections'][0]['opticalControlMode']) @@ -427,7 +432,7 @@ class TransportOlmTesting(unittest.TestCase): {"node-id": "XPDR-C1", "dest-tp": "XPDR1-CLIENT1", "src-tp": "XPDR1-NETWORK1"}], 196.1, 40, 196.075, 196.125, 761, - 768) + 768) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn('Request processed', res["output"]["result"]) @@ -444,7 +449,7 @@ class TransportOlmTesting(unittest.TestCase): {"node-id": "XPDR-A1", "src-tp": "XPDR1-NETWORK1", "dest-tp": "XPDR1-CLIENT1"}], 196.053125, 40, 196.025, 196.08125, 761, - 768) + 768) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn('Request processed', res["output"]["result"]) @@ -473,7 +478,7 @@ class TransportOlmTesting(unittest.TestCase): {"node-id": "ROADM-A1", "dest-tp": "DEG2-TTP-TXRX", "src-tp": "SRG1-PP2-TXRX"}], 196.1, 40, 196.075, 196.125, 753, - 760) + 760) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn('Roadm-connection successfully created for nodes', res["output"]["result"]) @@ -496,7 +501,7 @@ class TransportOlmTesting(unittest.TestCase): {"node-id": "ROADM-A1", "dest-tp": "DEG2-TTP-TXRX", "src-tp": "SRG1-PP2-TXRX"}], 196.053125, 40, 196.025, 196.08125, 761, - 768) + 768) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn('Request processed', res["output"]["result"]) diff --git a/tests/transportpce_tests/2.2.1/test09_tapi.py b/tests/transportpce_tests/2.2.1/test09_tapi.py index 150a4c1af..25a09c2b2 100644 --- a/tests/transportpce_tests/2.2.1/test09_tapi.py +++ b/tests/transportpce_tests/2.2.1/test09_tapi.py @@ -25,6 +25,7 @@ import test_utils CREATED_SUCCESSFULLY = 'Result message should contain Xponder Roadm Link created successfully' + class TransportTapitesting(unittest.TestCase): processes = None @@ -191,7 +192,7 @@ class TransportTapitesting(unittest.TestCase): self.assertEqual("Tpdr100g over WDM node", res["output"]["topology"]["node"][0]["name"][0]["value"], 'node name should be: Tpdr100g over WDM node') self.assertIn("ETH", res["output"]["topology"]["node"][0]["layer-protocol-name"], - 'Node layer protocol should contain ETH') + 'Node layer protocol should contain ETH') self.assertEqual(1, len(res["output"]["topology"]["node"][0]["node-rule-group"]), 'node should contain 1 node rule group') @@ -278,7 +279,7 @@ class TransportTapitesting(unittest.TestCase): self.assertEqual("ROADM-infra", res["output"]["topology"]["node"][0]["name"][0]["value"], 'node name should be: ROADM-infra') self.assertIn("PHOTONIC_MEDIA", res["output"]["topology"]["node"][0]["layer-protocol-name"], - 'Node layer protocol should contain PHOTONIC_MEDIA') + 'Node layer protocol should contain PHOTONIC_MEDIA') self.assertEqual(1, len(res["output"]["topology"]["node"][0]["node-rule-group"]), 'node should contain 1 node rule group') @@ -543,10 +544,10 @@ class TransportTapitesting(unittest.TestCase): for link in links: if link["name"][0]["value"] == "OTU4-SPDR-SA1-XPDR1-XPDR1-NETWORK1toSPDR-SC1-XPDR1-XPDR1-NETWORK1": self.assertEqual(100000, link["available-capacity"]["total-size"]["value"], - 'OTU4 link should have an available capacity of 100 000 Mbps') + 'OTU4 link should have an available capacity of 100 000 Mbps') elif link["name"][0]["value-name"] == "transitional link name": self.assertEqual(100, link["available-capacity"]["total-size"]["value"], - 'link should have an available capacity of 100 Gbps') + 'link should have an available capacity of 100 Gbps') self.assertEqual(2, len(link["node-edge-point"]), 'link should have 2 neps') def test_33_create_ODU4_service(self): @@ -588,13 +589,13 @@ class TransportTapitesting(unittest.TestCase): for link in links: if link["name"][0]["value"] == "OTU4-SPDR-SA1-XPDR1-XPDR1-NETWORK1toSPDR-SC1-XPDR1-XPDR1-NETWORK1": self.assertEqual(0, link["available-capacity"]["total-size"]["value"], - 'OTU4 link should have an available capacity of 0 Mbps') + 'OTU4 link should have an available capacity of 0 Mbps') elif link["name"][0]["value"] == "ODU4-SPDR-SA1-XPDR1-XPDR1-NETWORK1toSPDR-SC1-XPDR1-XPDR1-NETWORK1": self.assertEqual(100000, link["available-capacity"]["total-size"]["value"], - 'ODU4 link should have an available capacity of 100 000 Mbps') + 'ODU4 link should have an available capacity of 100 000 Mbps') elif link["name"][0]["value-name"] == "transitional link name": self.assertEqual(100, link["available-capacity"]["total-size"]["value"], - 'link should have an available capacity of 100 Gbps') + 'link should have an available capacity of 100 Gbps') self.assertEqual(2, len(link["node-edge-point"]), 'link should have 2 neps') def test_35_connect_sprda_2_n2_to_roadma_pp3(self): @@ -681,10 +682,10 @@ class TransportTapitesting(unittest.TestCase): 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"): - link_name = link["link-id"] - response = test_utils.delete_request(url+link_name) - self.assertEqual(response.status_code, requests.codes.ok) + link["org-openroadm-common-network:link-type"] == "XPONDER-INPUT"): + link_name = link["link-id"] + response = test_utils.delete_request(url+link_name) + self.assertEqual(response.status_code, requests.codes.ok) def test_42_check_tapi_topology_T0(self): url = "{}/operations/tapi-topology:get-topology-details" @@ -757,6 +758,7 @@ def find_object_with_key(list_dicts, key, value): return dict_ return None + def count_object_with_double_key(list_dicts, key1, key2, value): nb = 0 for dict in list_dicts: @@ -766,4 +768,4 @@ def count_object_with_double_key(list_dicts, key1, key2, value): if __name__ == "__main__": - unittest.main(verbosity=2) \ No newline at end of file + unittest.main(verbosity=2) diff --git a/tests/transportpce_tests/2.2.1/test11_end2end.py b/tests/transportpce_tests/2.2.1/test11_end2end.py index e68f79847..422f52596 100644 --- a/tests/transportpce_tests/2.2.1/test11_end2end.py +++ b/tests/transportpce_tests/2.2.1/test11_end2end.py @@ -10,6 +10,7 @@ # pylint: disable=no-member # pylint: disable=too-many-public-methods + import base64 import unittest import time @@ -18,7 +19,6 @@ import sys sys.path.append('transportpce_tests/common/') import test_utils - class TransportPCEFulltesting(unittest.TestCase): processes = None @@ -235,7 +235,8 @@ class TransportPCEFulltesting(unittest.TestCase): time.sleep(2) def test_13_check_xc1_ROADMA(self): - response = test_utils.check_netconf_node_request("ROADM-A1", "roadm-connections/SRG1-PP1-TXRX-DEG2-TTP-TXRX-761:768") + response = test_utils.check_netconf_node_request( + "ROADM-A1", "roadm-connections/SRG1-PP1-TXRX-DEG2-TTP-TXRX-761:768") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2 @@ -256,7 +257,8 @@ class TransportPCEFulltesting(unittest.TestCase): time.sleep(5) def test_14_check_xc1_ROADMC(self): - response = test_utils.check_netconf_node_request("ROADM-C1", "roadm-connections/SRG1-PP1-TXRX-DEG1-TTP-TXRX-761:768") + response = test_utils.check_netconf_node_request( + "ROADM-C1", "roadm-connections/SRG1-PP1-TXRX-DEG1-TTP-TXRX-761:768") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2 @@ -327,10 +329,10 @@ class TransportPCEFulltesting(unittest.TestCase): 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': - 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[95], 0, "Index 1 should not be available") + 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[95], 0, "Index 1 should not be available") time.sleep(3) def test_18_connect_xprdA_N2_to_roadmA_PP2(self): @@ -390,7 +392,8 @@ class TransportPCEFulltesting(unittest.TestCase): time.sleep(1) def test_24_check_xc2_ROADMA(self): - response = test_utils.check_netconf_node_request("ROADM-A1", "roadm-connections/DEG2-TTP-TXRX-SRG1-PP2-TXRX-753:760") + response = test_utils.check_netconf_node_request( + "ROADM-A1", "roadm-connections/DEG2-TTP-TXRX-SRG1-PP2-TXRX-753:760") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2 @@ -620,7 +623,8 @@ class TransportPCEFulltesting(unittest.TestCase): time.sleep(1) def test_38_check_xc1_ROADMA(self): - response = test_utils.check_netconf_node_request("ROADM-A1", "roadm-connections/SRG1-PP1-TXRX-DEG2-TTP-TXRX-761:768") + response = test_utils.check_netconf_node_request( + "ROADM-A1", "roadm-connections/SRG1-PP1-TXRX-DEG2-TTP-TXRX-761:768") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2 @@ -641,7 +645,8 @@ class TransportPCEFulltesting(unittest.TestCase): time.sleep(7) def test_39_check_xc1_ROADMC(self): - response = test_utils.check_netconf_node_request("ROADM-C1", "roadm-connections/SRG1-PP1-TXRX-DEG1-TTP-TXRX-761:768") + response = test_utils.check_netconf_node_request( + "ROADM-C1", "roadm-connections/SRG1-PP1-TXRX-DEG1-TTP-TXRX-761:768") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2 @@ -691,7 +696,8 @@ class TransportPCEFulltesting(unittest.TestCase): time.sleep(2) def test_42_check_xc2_ROADMA(self): - response = test_utils.check_netconf_node_request("ROADM-A1", "roadm-connections/SRG1-PP2-TXRX-DEG2-TTP-TXRX-753:760") + response = test_utils.check_netconf_node_request( + "ROADM-A1", "roadm-connections/SRG1-PP2-TXRX-DEG2-TTP-TXRX-753:760") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2 diff --git a/tests/transportpce_tests/7.1/test01_portmapping.py b/tests/transportpce_tests/7.1/test01_portmapping.py index 22d121dae..34632ee2e 100644 --- a/tests/transportpce_tests/7.1/test01_portmapping.py +++ b/tests/transportpce_tests/7.1/test01_portmapping.py @@ -78,7 +78,7 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase): res = response.json() self.assertIn( {'supported-interface-capability': - ['org-openroadm-port-types:if-otsi-otsigroup'], + ['org-openroadm-port-types:if-otsi-otsigroup'], 'supporting-port': 'L1', 'supporting-circuit-pack-name': '1/1/2-PLUG-NET', 'logical-connection-point': 'XPDR1-NETWORK1', @@ -113,7 +113,7 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase): res = response.json() self.assertIn( {'supported-interface-capability': - ['org-openroadm-port-types:if-otsi-otsigroup'], + ['org-openroadm-port-types:if-otsi-otsigroup'], 'supporting-port': 'L1', 'supporting-circuit-pack-name': '1/2/2-PLUG-NET', 'logical-connection-point': 'XPDR2-NETWORK1', @@ -131,9 +131,9 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase): self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn('org-openroadm-port-types:if-100GE-ODU4', - res['mapping'][0]['supported-interface-capability']) + res['mapping'][0]['supported-interface-capability']) self.assertIn('org-openroadm-port-types:if-OCH-OTU4-ODU4', - res['mapping'][0]['supported-interface-capability']) + res['mapping'][0]['supported-interface-capability']) self.assertEqual('C1', res['mapping'][0]['supporting-port']) self.assertEqual('1/2/1/1-PLUG-CLIENT', res['mapping'][0]['supporting-circuit-pack-name']) self.assertEqual('XPDR2-CLIENT1', res['mapping'][0]['logical-connection-point']) @@ -143,9 +143,9 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase): self.assertEqual('InService', res['mapping'][0]['port-admin-state']) self.assertEqual('InService', res['mapping'][0]['port-oper-state']) self.assertEqual({ - "min-trib-slot": "1.1", - "max-trib-slot": "1.20" - }, res['mapping'][0]['mpdr-restrictions']) + "min-trib-slot": "1.1", + "max-trib-slot": "1.20" + }, res['mapping'][0]['mpdr-restrictions']) # Added test to check mc-capability-profile for a transponder def test_08_check_mccapprofile(self): @@ -182,4 +182,4 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase): if __name__ == '__main__': - unittest.main(verbosity=2) + unittest.main(verbosity=2) diff --git a/tests/transportpce_tests/common/simulators.py b/tests/transportpce_tests/common/simulators.py index c1da758b3..e361ea53f 100644 --- a/tests/transportpce_tests/common/simulators.py +++ b/tests/transportpce_tests/common/simulators.py @@ -31,4 +31,4 @@ SIMS = { ('xpdrc', '7.1'): {'port': '17854', 'configfile': 'oper-XPDRC.xml', 'logfile': 'xpdrc-71.log'}, ('xpdra2', '7.1'): {'port': '17857', 'configfile': 'oper-XPDRA2.xml', 'logfile': 'xpdra2-71.log'}, ('xpdrc2', '7.1'): {'port': '17858', 'configfile': 'oper-XPDRC2.xml', 'logfile': 'xpdrc2-71.log'} -} \ No newline at end of file +} diff --git a/tests/transportpce_tests/common/test_utils.py b/tests/transportpce_tests/common/test_utils.py index b70a4eeb5..136991f44 100644 --- a/tests/transportpce_tests/common/test_utils.py +++ b/tests/transportpce_tests/common/test_utils.py @@ -199,6 +199,7 @@ def put_xmlrequest(url, data): headers=TYPE_APPLICATION_XML, auth=(ODL_LOGIN, ODL_PWD)) + def put_jsonrequest(url, data): return requests.request( "PUT", url.format(RESTCONF_BASE_URL), @@ -206,6 +207,7 @@ def put_jsonrequest(url, data): headers=TYPE_APPLICATION_JSON, auth=(ODL_LOGIN, ODL_PWD)) + def rawput_request(url, data): return requests.request( "PUT", url.format(RESTCONF_BASE_URL), @@ -213,6 +215,7 @@ def rawput_request(url, data): headers=TYPE_APPLICATION_JSON, auth=(ODL_LOGIN, ODL_PWD)) + def rawpost_request(url, data): return requests.request( "POST", url.format(RESTCONF_BASE_URL), diff --git a/tests/transportpce_tests/hybrid/test01_device_change_notifications.py b/tests/transportpce_tests/hybrid/test01_device_change_notifications.py index c57316b20..10a309ef7 100644 --- a/tests/transportpce_tests/hybrid/test01_device_change_notifications.py +++ b/tests/transportpce_tests/hybrid/test01_device_change_notifications.py @@ -18,7 +18,6 @@ import sys sys.path.append('transportpce_tests/common/') import test_utils - class TransportPCEFulltesting(unittest.TestCase): processes = None @@ -240,12 +239,12 @@ class TransportPCEFulltesting(unittest.TestCase): def test_13_change_status_line_port_xpdra(self): url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/1%2F0%2F1-PLUG-NET/ports/1" body = {"ports": [{ - "port-name": "1", - "logical-connection-point": "XPDR1-NETWORK1", - "port-type": "CFP2", - "circuit-id": "XPDRA-NETWORK", - "administrative-state": "outOfService", - "port-qual": "xpdr-network"}]} + "port-name": "1", + "logical-connection-point": "XPDR1-NETWORK1", + "port-type": "CFP2", + "circuit-id": "XPDRA-NETWORK", + "administrative-state": "outOfService", + "port-qual": "xpdr-network"}]} response = requests.request("PUT", url.format("http://127.0.0.1:8130/restconf"), data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD)) @@ -317,12 +316,12 @@ class TransportPCEFulltesting(unittest.TestCase): def test_17_restore_status_line_port_xpdra(self): url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/1%2F0%2F1-PLUG-NET/ports/1" body = {"ports": [{ - "port-name": "1", - "logical-connection-point": "XPDR1-NETWORK1", - "port-type": "CFP2", - "circuit-id": "XPDRA-NETWORK", - "administrative-state": "inService", - "port-qual": "xpdr-network"}]} + "port-name": "1", + "logical-connection-point": "XPDR1-NETWORK1", + "port-type": "CFP2", + "circuit-id": "XPDRA-NETWORK", + "administrative-state": "inService", + "port-qual": "xpdr-network"}]} response = requests.request("PUT", url.format("http://127.0.0.1:8130/restconf"), data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD)) @@ -367,12 +366,12 @@ class TransportPCEFulltesting(unittest.TestCase): def test_21_change_status_port_roadma_srg(self): url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/3%2F0/ports/C1" body = {"ports": [{ - "port-name": "C1", - "logical-connection-point": "SRG1-PP1", - "port-type": "client", - "circuit-id": "SRG1", - "administrative-state": "outOfService", - "port-qual": "roadm-external"}]} + "port-name": "C1", + "logical-connection-point": "SRG1-PP1", + "port-type": "client", + "circuit-id": "SRG1", + "administrative-state": "outOfService", + "port-qual": "roadm-external"}]} response = requests.request("PUT", url.format("http://127.0.0.1:8141/restconf"), data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD)) @@ -436,12 +435,12 @@ class TransportPCEFulltesting(unittest.TestCase): def test_24_restore_status_port_roadma_srg(self): url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/3%2F0/ports/C1" body = {"ports": [{ - "port-name": "C1", - "logical-connection-point": "SRG1-PP1", - "port-type": "client", - "circuit-id": "SRG1", - "administrative-state": "inService", - "port-qual": "roadm-external"}]} + "port-name": "C1", + "logical-connection-point": "SRG1-PP1", + "port-type": "client", + "circuit-id": "SRG1", + "administrative-state": "inService", + "port-qual": "roadm-external"}]} response = requests.request("PUT", url.format("http://127.0.0.1:8141/restconf"), data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD)) @@ -460,12 +459,12 @@ class TransportPCEFulltesting(unittest.TestCase): def test_28_change_status_line_port_roadma_deg(self): url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/2%2F0/ports/L1" body = {"ports": [{ - "port-name": "L1", - "logical-connection-point": "DEG2-TTP-TXRX", - "port-type": "LINE", - "circuit-id": "1", - "administrative-state": "outOfService", - "port-qual": "roadm-external"}]} + "port-name": "L1", + "logical-connection-point": "DEG2-TTP-TXRX", + "port-type": "LINE", + "circuit-id": "1", + "administrative-state": "outOfService", + "port-qual": "roadm-external"}]} response = requests.request("PUT", url.format("http://127.0.0.1:8141/restconf"), data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD)) @@ -529,12 +528,12 @@ class TransportPCEFulltesting(unittest.TestCase): def test_31_restore_status_line_port_roadma_srg(self): url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/2%2F0/ports/L1" body = {"ports": [{ - "port-name": "L1", - "logical-connection-point": "DEG2-TTP-TXRX", - "port-type": "LINE", - "circuit-id": "1", - "administrative-state": "inService", - "port-qual": "roadm-external"}]} + "port-name": "L1", + "logical-connection-point": "DEG2-TTP-TXRX", + "port-type": "LINE", + "circuit-id": "1", + "administrative-state": "inService", + "port-qual": "roadm-external"}]} response = requests.request("PUT", url.format("http://127.0.0.1:8141/restconf"), data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD)) @@ -553,10 +552,10 @@ class TransportPCEFulltesting(unittest.TestCase): def test_35_change_status_line_port_xpdrc(self): url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/1%2F0%2F1-PLUG-NET/ports/1" body = {"ports": [{ - "port-name": "1", - "port-type": "CFP2", - "administrative-state": "outOfService", - "port-qual": "xpdr-network"}]} + "port-name": "1", + "port-type": "CFP2", + "administrative-state": "outOfService", + "port-qual": "xpdr-network"}]} response = requests.request("PUT", url.format("http://127.0.0.1:8154/restconf"), data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD)) @@ -620,10 +619,10 @@ class TransportPCEFulltesting(unittest.TestCase): def test_38_restore_status_line_port_xpdrc(self): url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/1%2F0%2F1-PLUG-NET/ports/1" body = {"ports": [{ - "port-name": "1", - "port-type": "CFP2", - "administrative-state": "inService", - "port-qual": "xpdr-network"}]} + "port-name": "1", + "port-type": "CFP2", + "administrative-state": "inService", + "port-qual": "xpdr-network"}]} response = requests.request("PUT", url.format("http://127.0.0.1:8154/restconf"), data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD)) @@ -642,12 +641,12 @@ class TransportPCEFulltesting(unittest.TestCase): def test_42_change_status_port_roadma_srg(self): url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/3%2F0/ports/C2" body = {"ports": [{ - "port-name": "C2", - "logical-connection-point": "SRG1-PP2", - "port-type": "client", - "circuit-id": "SRG1", - "administrative-state": "outOfService", - "port-qual": "roadm-external"}]} + "port-name": "C2", + "logical-connection-point": "SRG1-PP2", + "port-type": "client", + "circuit-id": "SRG1", + "administrative-state": "outOfService", + "port-qual": "roadm-external"}]} response = requests.request("PUT", url.format("http://127.0.0.1:8141/restconf"), data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD)) diff --git a/tests/transportpce_tests/hybrid/test02_B100G_end2end.py b/tests/transportpce_tests/hybrid/test02_B100G_end2end.py index b5e8c70eb..f6e4e3a98 100644 --- a/tests/transportpce_tests/hybrid/test02_B100G_end2end.py +++ b/tests/transportpce_tests/hybrid/test02_B100G_end2end.py @@ -20,7 +20,6 @@ import sys sys.path.append('transportpce_tests/common/') import test_utils - class TransportPCEtesting(unittest.TestCase): processes = None @@ -293,8 +292,8 @@ class TransportPCEtesting(unittest.TestCase): self.assertDictEqual( dict({u'frequency': 196.0812, u'otsi-rate': u'org-openroadm-common-optical-channel-types:R400G-otsi', - u'transmit-power': -5, u'modulation-format': 'dp-qam16'}, - **res['interface'][0]['org-openroadm-optical-tributary-signal-interfaces:otsi']), + u'transmit-power': -5, u'modulation-format': 'dp-qam16'}, + **res['interface'][0]['org-openroadm-optical-tributary-signal-interfaces:otsi']), res['interface'][0]['org-openroadm-optical-tributary-signal-interfaces:otsi']) def test_15_check_interface_OTSI_GROUP_xpdra2(self): @@ -359,8 +358,8 @@ class TransportPCEtesting(unittest.TestCase): self.assertDictEqual( dict({u'frequency': 196.0812, u'otsi-rate': u'org-openroadm-common-optical-channel-types:R400G-otsi', - u'transmit-power': -5, u'modulation-format': 'dp-qam16'}, - **res['interface'][0]['org-openroadm-optical-tributary-signal-interfaces:otsi']), + u'transmit-power': -5, u'modulation-format': 'dp-qam16'}, + **res['interface'][0]['org-openroadm-optical-tributary-signal-interfaces:otsi']), res['interface'][0]['org-openroadm-optical-tributary-signal-interfaces:otsi']) def test_18_check_interface_OTSI_GROUP_xpdrc2(self): @@ -637,11 +636,11 @@ class TransportPCEtesting(unittest.TestCase): self.assertEqual(response.status_code, requests.codes.ok) res = response.json() input_dict_1 = {'name': 'XPDR2-CLIENT1-ETHERNET100G', - 'administrative-state': 'inService', - 'supporting-circuit-pack-name': '1/2/1/1-PLUG-CLIENT', - 'type': 'org-openroadm-interfaces:ethernetCsmacd', - 'supporting-port': 'C1' - } + 'administrative-state': 'inService', + 'supporting-circuit-pack-name': '1/2/1/1-PLUG-CLIENT', + 'type': 'org-openroadm-interfaces:ethernetCsmacd', + 'supporting-port': 'C1' + } input_dict_2 = {u'speed': 100000} self.assertDictEqual(dict(input_dict_1, **res['interface'][0]), res['interface'][0]) @@ -729,16 +728,16 @@ class TransportPCEtesting(unittest.TestCase): self.assertEqual(response.status_code, requests.codes.ok) res = response.json() input_dict_1 = {'name': 'XPDR2-CLIENT1-ETHERNET100G', - 'administrative-state': 'inService', - 'supporting-circuit-pack-name': '1/2/1/1-PLUG-CLIENT', - 'type': 'org-openroadm-interfaces:ethernetCsmacd', - 'supporting-port': 'C1' - } + 'administrative-state': 'inService', + 'supporting-circuit-pack-name': '1/2/1/1-PLUG-CLIENT', + 'type': 'org-openroadm-interfaces:ethernetCsmacd', + 'supporting-port': 'C1' + } input_dict_2 = {u'speed': 100000} self.assertDictEqual(dict(input_dict_1, **res['interface'][0]), res['interface'][0]) self.assertDictEqual(dict(input_dict_2, **res['interface'][0]['org-openroadm-ethernet-interfaces:ethernet']), - res['interface'][0]['org-openroadm-ethernet-interfaces:ethernet']) + res['interface'][0]['org-openroadm-ethernet-interfaces:ethernet']) def test_36_check_interface_ODU4_CLIENT_xpdrc2(self): response = test_utils.check_netconf_node_request( @@ -1033,7 +1032,7 @@ class TransportPCEtesting(unittest.TestCase): for tp in tpList: if tp['tp-id'] == 'XPDR2-NETWORK1': self.assertNotIn('org-openroadm-otn-network-topology:xpdr-tp-port-connection-attributes', - dict.keys(tp)) + dict.keys(tp)) def test_60_delete_OTUC4_service(self): response = test_utils.service_delete_request("service1-OTUC4") @@ -1159,7 +1158,8 @@ class TransportPCEtesting(unittest.TestCase): time.sleep(2) def test_74_check_xc1_roadma(self): - response = test_utils.check_netconf_node_request("ROADM-A1", "roadm-connections/SRG1-PP1-TXRX-DEG2-TTP-TXRX-755:768") + response = test_utils.check_netconf_node_request( + "ROADM-A1", "roadm-connections/SRG1-PP1-TXRX-DEG2-TTP-TXRX-755:768") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2 @@ -1228,24 +1228,23 @@ class TransportPCEtesting(unittest.TestCase): 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': - 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[95], 0, "Index 1 should not be available") + 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[95], 0, "Index 1 should not be available") time.sleep(3) - def test_78_check_interface_100GE_CLIENT_xpdra2(self): response = test_utils.check_netconf_node_request( "XPDR-A2", "interface/XPDR1-CLIENT1-ETHERNET") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() input_dict_1 = {'name': 'XPDR1-CLIENT1-ETHERNET', - 'administrative-state': 'inService', - 'supporting-circuit-pack-name': '1/1/1-PLUG-CLIENT', - 'type': 'org-openroadm-interfaces:ethernetCsmacd', - 'supporting-port': 'C1' - } + 'administrative-state': 'inService', + 'supporting-circuit-pack-name': '1/1/1-PLUG-CLIENT', + 'type': 'org-openroadm-interfaces:ethernetCsmacd', + 'supporting-port': 'C1' + } input_dict_2 = {u'speed': 400000} self.assertDictEqual(dict(input_dict_1, **res['interface'][0]), res['interface'][0]) diff --git a/tests/transportpce_tests/pce/test01_pce.py b/tests/transportpce_tests/pce/test01_pce.py index bbed914aa..f54a5caaf 100644 --- a/tests/transportpce_tests/pce/test01_pce.py +++ b/tests/transportpce_tests/pce/test01_pce.py @@ -19,7 +19,6 @@ import requests sys.path.append('transportpce_tests/common/') import test_utils - class TransportPCEtesting(unittest.TestCase): simple_topo_bi_dir_data = None @@ -49,7 +48,7 @@ class TransportPCEtesting(unittest.TestCase): with open(TOPO_UNI_DIR_COMPLEX_FILE, 'r') as topo_uni_dir_complex: cls.complex_topo_uni_dir_data = topo_uni_dir_complex.read() PORT_MAPPING_FILE = os.path.join(os.path.dirname(os.path.realpath(__file__)), - "..", "..", "sample_configs", "pce_portmapping_121.json") + "..", "..", "sample_configs", "pce_portmapping_121.json") with open(PORT_MAPPING_FILE, 'r') as port_mapping: cls.port_mapping_data = port_mapping.read() sample_files_parsed = True @@ -354,7 +353,7 @@ class TransportPCEtesting(unittest.TestCase): nbElmPath = len(res['output']['response-parameters']['path-description'] ['aToZ-direction']['aToZ']) self.assertEqual(31, nbElmPath) - link = {"link-id": "OpenROADM-1-3-DEG2-to-OpenROADM-1-2-DEG2", "state":"inService"} + link = {"link-id": "OpenROADM-1-3-DEG2-to-OpenROADM-1-2-DEG2", "state": "inService"} find = False for i in range(0, nbElmPath): resource_i = (res['output']['response-parameters']['path-description']['aToZ-direction']['aToZ'][i] @@ -384,7 +383,7 @@ class TransportPCEtesting(unittest.TestCase): nbElmPath = len(res['output']['response-parameters']['path-description'] ['aToZ-direction']['aToZ']) self.assertEqual(47, nbElmPath) - link = {"link-id": "OpenROADM-1-3-DEG2-to-OpenROADM-1-2-DEG2", "state":"inService"} + link = {"link-id": "OpenROADM-1-3-DEG2-to-OpenROADM-1-2-DEG2", "state": "inService"} find = False for i in range(0, nbElmPath): resource_i = (res['output']['response-parameters']['path-description']['aToZ-direction']['aToZ'][i] diff --git a/tests/transportpce_tests/pce/test02_pce_400G.py b/tests/transportpce_tests/pce/test02_pce_400G.py index 2979105ec..d852d5bc7 100644 --- a/tests/transportpce_tests/pce/test02_pce_400G.py +++ b/tests/transportpce_tests/pce/test02_pce_400G.py @@ -19,7 +19,6 @@ import requests sys.path.append('transportpce_tests/common/') import test_utils - class TransportPCE400Gtesting(unittest.TestCase): simple_topo_bi_dir_data = None @@ -36,22 +35,22 @@ class TransportPCE400Gtesting(unittest.TestCase): cls.topo_bi_dir_data = topo_bi_dir.read() OTN_TOPO_BI_DIR_FILE = os.path.join(os.path.dirname(os.path.realpath(__file__)), - "..", "..", "sample_configs", "honeynode-otntopo400G.json") + "..", "..", "sample_configs", "honeynode-otntopo400G.json") with open(OTN_TOPO_BI_DIR_FILE, 'r') as otn_topo_bi_dir: cls.otn_topo_bi_dir_data = otn_topo_bi_dir.read() OTUC4_OTN_TOPO_BI_DIR_FILE = os.path.join(os.path.dirname(os.path.realpath(__file__)), - "..", "..", "sample_configs", "honeynode-otntopo400GwithOTUC4.json") + "..", "..", "sample_configs", "honeynode-otntopo400GwithOTUC4.json") with open(OTUC4_OTN_TOPO_BI_DIR_FILE, 'r') as otuc4_otn_topo_bi_dir: cls.otuc4_otn_topo_bi_dir_data = otuc4_otn_topo_bi_dir.read() ODUC4_OTN_TOPO_BI_DIR_FILE = os.path.join(os.path.dirname(os.path.realpath(__file__)), - "..", "..", "sample_configs", "honeynode-otntopo400GwithODUC4.json") + "..", "..", "sample_configs", "honeynode-otntopo400GwithODUC4.json") with open(ODUC4_OTN_TOPO_BI_DIR_FILE, 'r') as oduc4_otn_topo_bi_dir: cls.oduc4_otn_topo_bi_dir_data = oduc4_otn_topo_bi_dir.read() PORT_MAPPING_FILE = os.path.join(os.path.dirname(os.path.realpath(__file__)), - "..", "..", "sample_configs", "pce_portmapping_71.json") + "..", "..", "sample_configs", "pce_portmapping_71.json") with open(PORT_MAPPING_FILE, 'r') as port_mapping: cls.port_mapping_data = port_mapping.read() sample_files_parsed = True @@ -139,11 +138,11 @@ class TransportPCE400Gtesting(unittest.TestCase): {"service-rate": "400", "clli": "NodeA", "service-format": "OTU", "node-id": "XPDR-A2", "rx-direction": {"port": {"port-device-name": "XPDR-A2-XPDR2"}} - }, + }, {"service-rate": "400", "clli": "NodeC", "service-format": "OTU", "node-id": "XPDR-C2", "rx-direction": {"port": {"port-device-name": "XPDR-C2-XPDR2"}} - }) + }) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn('Path is calculated', @@ -184,11 +183,11 @@ class TransportPCE400Gtesting(unittest.TestCase): {"service-rate": "400", "clli": "NodeA", "service-format": "ODU", "node-id": "XPDR-A2", "tx-direction": {"port": {"port-device-name": "XPDR-A2-XPDR2"}} - }, + }, {"service-rate": "400", "clli": "NodeC", "service-format": "ODU", "node-id": "XPDR-C2", "tx-direction": {"port": {"port-device-name": "XPDR-C2-XPDR2"}} - }) + }) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn('Path is calculated', @@ -217,11 +216,11 @@ class TransportPCE400Gtesting(unittest.TestCase): {"service-rate": "100", "clli": "NodeA", "service-format": "Ethernet", "node-id": "XPDR-A2", "tx-direction": {"port": {"port-device-name": "XPDR-A2-XPDR2", - "port-name": "XPDR2-CLIENT1"}}}, + "port-name": "XPDR2-CLIENT1"}}}, {"service-rate": "100", "clli": "NodeC", "service-format": "Ethernet", "node-id": "XPDR-C2", "tx-direction": {"port": {"port-device-name": "XPDR-C2-XPDR2", - "port-name": "XPDR2-CLIENT1"}}}) + "port-name": "XPDR2-CLIENT1"}}}) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() @@ -247,5 +246,6 @@ class TransportPCE400Gtesting(unittest.TestCase): ['zToA-direction']['modulation-format']) time.sleep(5) + if __name__ == "__main__": unittest.main(verbosity=2) diff --git a/tests/transportpce_tests/with_docker/test01_gnpy.py b/tests/transportpce_tests/with_docker/test01_gnpy.py index 103e06f08..c9fceb4d3 100644 --- a/tests/transportpce_tests/with_docker/test01_gnpy.py +++ b/tests/transportpce_tests/with_docker/test01_gnpy.py @@ -20,7 +20,6 @@ import requests sys.path.append('transportpce_tests/common/') import test_utils - class TransportGNPYtesting(unittest.TestCase): topo_cllinet_data = None @@ -49,7 +48,7 @@ class TransportGNPYtesting(unittest.TestCase): with open(TOPO_ORDTOPO_FILE, 'r') as topo_ordtopo: cls.topo_ordtopo_data = topo_ordtopo.read() PORT_MAPPING_FILE = os.path.join(os.path.dirname(os.path.realpath(__file__)), - "..", "..", "sample_configs", "gnpy", "gnpy_portmapping_121.json") + "..", "..", "sample_configs", "gnpy", "gnpy_portmapping_121.json") with open(PORT_MAPPING_FILE, 'r') as port_mapping: cls.port_mapping_data = port_mapping.read() sample_files_parsed = True @@ -228,6 +227,7 @@ class TransportGNPYtesting(unittest.TestCase): self.assertEqual(response.status_code, requests.codes.ok) time.sleep(2) + if __name__ == "__main__": # logging.basicConfig(filename='./transportpce_tests/log/response.log',filemode='w',level=logging.DEBUG) unittest.main(verbosity=2) diff --git a/tests/transportpce_tests/with_docker/test02_nbinotifications.py b/tests/transportpce_tests/with_docker/test02_nbinotifications.py index 14b8bd888..45681f684 100644 --- a/tests/transportpce_tests/with_docker/test02_nbinotifications.py +++ b/tests/transportpce_tests/with_docker/test02_nbinotifications.py @@ -20,7 +20,6 @@ import requests sys.path.append('transportpce_tests/common/') import test_utils - class TransportNbiNotificationstesting(unittest.TestCase): processes = None cr_serv_sample_data = {"input": { -- 2.36.6