X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=tests%2Ftransportpce_tests%2F2.2.1%2Ftest_otn_topology.py;h=bc2c28a901014c22e419add735a2d36a0cdca4ec;hb=6fe500a1bf557eb44d22584ac6bf7c924f12c47f;hp=7505306cc77dd58a9f4e3f0470c577364ca44cd2;hpb=38788ce8eedd77f7b1dc089df00d9e2a9bcfc1f8;p=transportpce.git diff --git a/tests/transportpce_tests/2.2.1/test_otn_topology.py b/tests/transportpce_tests/2.2.1/test_otn_topology.py index 7505306cc..bc2c28a90 100644 --- a/tests/transportpce_tests/2.2.1/test_otn_topology.py +++ b/tests/transportpce_tests/2.2.1/test_otn_topology.py @@ -9,69 +9,41 @@ # http://www.apache.org/licenses/LICENSE-2.0 ############################################################################## -import json -import os -import psutil -import requests -import signal -import shutil -import subprocess -import time +# pylint: disable=no-member +# pylint: disable=too-many-public-methods + import unittest +import time import logging -import test_utils +import requests +from common import test_utils class TransportPCEtesting(unittest.TestCase): - sim_process1 = None - odl_process = None - restconf_baseurl = "http://localhost:8181/restconf" + processes = None @classmethod def setUpClass(cls): - cls.odl_process = test_utils.start_tpce() - cls.sim_process1 = test_utils.start_sim('spdrav2') + cls.processes = test_utils.start_tpce() + cls.processes = test_utils.start_sims(['spdra']) @classmethod def tearDownClass(cls): - for child in psutil.Process(cls.odl_process.pid).children(): - child.send_signal(signal.SIGINT) - child.wait() - cls.odl_process.send_signal(signal.SIGINT) - cls.odl_process.wait() - for child in psutil.Process(cls.sim_process1.pid).children(): - child.send_signal(signal.SIGINT) - child.wait() - cls.sim_process1.send_signal(signal.SIGINT) - cls.sim_process1.wait() + # pylint: disable=not-an-iterable + for process in cls.processes: + test_utils.shutdown_process(process) + print("all processes killed") def setUp(self): time.sleep(5) def test_01_connect_SPDR_SA1(self): - url = ("{}/config/network-topology:" - "network-topology/topology/topology-netconf/node/SPDR-SA1" - .format(self.restconf_baseurl)) - data = {"node": [{ - "node-id": "SPDR-SA1", - "netconf-node-topology:username": "admin", - "netconf-node-topology:password": "admin", - "netconf-node-topology:host": "127.0.0.1", - "netconf-node-topology:port": test_utils.sims['spdrav2']['port'], - "netconf-node-topology:tcp-only": "false", - "netconf-node-topology:pass-through": {}}]} - headers = {'content-type': 'application/json'} - response = requests.request( - "PUT", url, data=json.dumps(data), headers=headers, - auth=('admin', 'admin')) - self.assertEqual(response.status_code, requests.codes.created) + response = test_utils.mount_device("SPDR-SA1", 'spdra') + self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201) time.sleep(10) - url = ("{}/operational/network-topology:" - "network-topology/topology/topology-netconf/node/SPDR-SA1" - .format(self.restconf_baseurl)) - response = requests.request( - "GET", url, headers=headers, auth=('admin', 'admin')) + + response = test_utils.get_netconf_oper_request("SPDR-SA1") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertEqual( @@ -79,11 +51,7 @@ class TransportPCEtesting(unittest.TestCase): 'connected') def test_02_getClliNetwork(self): - url = ("{}/config/ietf-network:networks/network/clli-network" - .format(self.restconf_baseurl)) - headers = {'content-type': 'application/json'} - response = requests.request( - "GET", url, headers=headers, auth=('admin', 'admin')) + response = test_utils.get_clli_net_request() self.assertEqual(response.status_code, requests.codes.ok) res = response.json() logging.info(res) @@ -91,11 +59,7 @@ class TransportPCEtesting(unittest.TestCase): self.assertEqual(res['network'][0]['node'][0]['org-openroadm-clli-network:clli'], 'NodeSA') def test_03_getOpenRoadmNetwork(self): - url = ("{}/config/ietf-network:networks/network/openroadm-network" - .format(self.restconf_baseurl)) - headers = {'content-type': 'application/json'} - response = requests.request( - "GET", url, headers=headers, auth=('admin', 'admin')) + response = test_utils.get_ordm_net_request() self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertEqual(res['network'][0]['node'][0]['node-id'], 'SPDR-SA1') @@ -107,34 +71,27 @@ class TransportPCEtesting(unittest.TestCase): self.assertEqual(res['network'][0]['node'][0]['org-openroadm-network:ip'], '1.2.3.4') def test_04_getLinks_OpenroadmTopology(self): - url = ("{}/config/ietf-network:networks/network/openroadm-topology" - .format(self.restconf_baseurl)) - headers = {'content-type': 'application/json'} - response = requests.request( - "GET", url, headers=headers, auth=('admin', 'admin')) + response = test_utils.get_ordm_topo_request("") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() # Tests related to links self.assertNotIn('ietf-network-topology:link', res['network'][0]) def test_05_getNodes_OpenRoadmTopology(self): - url = ("{}/config/ietf-network:networks/network/openroadm-topology" - .format(self.restconf_baseurl)) - headers = {'content-type': 'application/json'} - response = requests.request( - "GET", url, headers=headers, auth=('admin', 'admin')) + # pylint: disable=redundant-unittest-assert + response = test_utils.get_ordm_topo_request("") res = response.json() # Tests related to nodes self.assertEqual(response.status_code, requests.codes.ok) self.assertIn('node', res['network'][0]) - if ('node' in res['network'][0]): + if 'node' in res['network'][0]: nbNode = len(res['network'][0]['node']) self.assertEqual(nbNode, 2) listNode = ['SPDR-SA1-XPDR1', 'SPDR-SA1-XPDR2'] for i in range(0, nbNode): nodeType = res['network'][0]['node'][i]['org-openroadm-common-network:node-type'] nodeId = res['network'][0]['node'][i]['node-id'] - if (nodeId == 'SPDR-SA1-XPDR1'): + if nodeId == 'SPDR-SA1-XPDR1': self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'SPDR-SA1'}, res['network'][0]['node'][i]['supporting-node']) self.assertIn({'network-ref': 'clli-network', 'node-ref': 'NodeSA'}, @@ -144,16 +101,16 @@ class TransportPCEtesting(unittest.TestCase): 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'] - if (tpType == 'XPONDER-CLIENT'): + tpType = (res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j] + ['org-openroadm-common-network:tp-type']) + if tpType == 'XPONDER-CLIENT': client += 1 - elif (tpType == 'XPONDER-NETWORK'): + elif tpType == 'XPONDER-NETWORK': network += 1 self.assertTrue(client == 0) self.assertTrue(network == 1) listNode.remove(nodeId) - elif (nodeId == 'SPDR-SA1-XPDR2'): + elif nodeId == 'SPDR-SA1-XPDR2': self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'SPDR-SA1'}, res['network'][0]['node'][i]['supporting-node']) self.assertIn({'network-ref': 'clli-network', 'node-ref': 'NodeSA'}, @@ -163,11 +120,11 @@ class TransportPCEtesting(unittest.TestCase): 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'] - if (tpType == 'XPONDER-CLIENT'): + tpType = (res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j] + ['org-openroadm-common-network:tp-type']) + if tpType == 'XPONDER-CLIENT': client += 1 - elif (tpType == 'XPONDER-NETWORK'): + elif tpType == 'XPONDER-NETWORK': network += 1 self.assertTrue(client == 0) self.assertTrue(network == 4) @@ -177,21 +134,14 @@ class TransportPCEtesting(unittest.TestCase): self.assertEqual(len(listNode), 0) def test_06_getLinks_OtnTopology(self): - url = ("{}/config/ietf-network:networks/network/otn-topology" - .format(self.restconf_baseurl)) - headers = {'content-type': 'application/json'} - response = requests.request( - "GET", url, headers=headers, auth=('admin', 'admin')) + response = test_utils.get_otn_topo_request() self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertNotIn('ietf-network-topology:link', res['network'][0]) def test_07_getNodes_OtnTopology(self): - url = ("{}/config/ietf-network:networks/network/otn-topology" - .format(self.restconf_baseurl)) - headers = {'content-type': 'application/json'} - response = requests.request( - "GET", url, headers=headers, auth=('admin', 'admin')) + # pylint: disable=redundant-unittest-assert + response = test_utils.get_otn_topo_request() res = response.json() self.assertEqual(response.status_code, requests.codes.ok) nbNode = len(res['network'][0]['node']) @@ -199,7 +149,7 @@ class TransportPCEtesting(unittest.TestCase): listNode = ['SPDR-SA1-XPDR1', 'SPDR-SA1-XPDR2'] for i in range(0, nbNode): nodeId = res['network'][0]['node'][i]['node-id'] - if(nodeId == 'SPDR-SA1-XPDR1'): + if nodeId == 'SPDR-SA1-XPDR1': nodeType = res['network'][0]['node'][i]['org-openroadm-common-network:node-type'] self.assertEqual(nodeType, 'MUXPDR') self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'SPDR-SA1'}, @@ -214,26 +164,35 @@ class TransportPCEtesting(unittest.TestCase): client = 0 network = 0 for j in range(0, nbTps): - tpType = res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['org-openroadm-common-network:tp-type'] + tpType = (res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j] + ['org-openroadm-common-network:tp-type']) tpId = res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['tp-id'] - if (tpType == 'XPONDER-CLIENT'): + if tpType == 'XPONDER-CLIENT': client += 1 - self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['org-openroadm-otn-network-topology:tp-supported-interfaces'] - ['supported-interface-capability'][0]['if-cap-type'], 'org-openroadm-port-types:if-10GE-ODU2e') - self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j] - ['org-openroadm-otn-network-topology:xpdr-tp-port-connection-attributes']['rate'], 'org-openroadm-otn-common-types:ODU2e') - elif (tpType == 'XPONDER-NETWORK'): + self.assertEqual((res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j] + ['org-openroadm-otn-network-topology:tp-supported-interfaces'] + ['supported-interface-capability'][0]['if-cap-type']), + 'org-openroadm-port-types:if-10GE-ODU2e') + self.assertEqual((res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j] + ['org-openroadm-otn-network-topology:xpdr-tp-port-connection-attributes'] + ['rate']), + 'org-openroadm-otn-common-types:ODU2e') + elif tpType == 'XPONDER-NETWORK': network += 1 - self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['org-openroadm-otn-network-topology:tp-supported-interfaces'] - ['supported-interface-capability'][0]['if-cap-type'], 'org-openroadm-port-types:if-OCH-OTU4-ODU4') - self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j] - ['org-openroadm-otn-network-topology:xpdr-tp-port-connection-attributes']['rate'], 'org-openroadm-otn-common-types:ODU4') - self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point'] - [j]['supporting-termination-point'][0]['network-ref'], 'openroadm-topology') - self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point'] - [j]['supporting-termination-point'][0]['node-ref'], 'SPDR-SA1-XPDR1') - self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point'] - [j]['supporting-termination-point'][0]['tp-ref'], 'XPDR1-NETWORK1') + self.assertEqual((res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j] + ['org-openroadm-otn-network-topology:tp-supported-interfaces'] + ['supported-interface-capability'][0]['if-cap-type']), + 'org-openroadm-port-types:if-OCH-OTU4-ODU4') + self.assertEqual((res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j] + ['org-openroadm-otn-network-topology:xpdr-tp-port-connection-attributes'] + ['rate']), + 'org-openroadm-otn-common-types:ODU4') + self.assertEqual((res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j] + ['supporting-termination-point'][0]['network-ref']), 'openroadm-topology') + self.assertEqual((res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j] + ['supporting-termination-point'][0]['node-ref']), 'SPDR-SA1-XPDR1') + self.assertEqual((res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j] + ['supporting-termination-point'][0]['tp-ref']), 'XPDR1-NETWORK1') self.assertTrue(client == 4) self.assertTrue(network == 1) listNode.remove(nodeId) @@ -241,14 +200,15 @@ class TransportPCEtesting(unittest.TestCase): ['odu-switching-pools'][0]['non-blocking-list']) self.assertEqual(nbNbl, 4) for k in range(0, nbNbl): - nbl = res['network'][0]['node'][i]['org-openroadm-otn-network-topology:switching-pools']['odu-switching-pools'][0]['non-blocking-list'][k] - if (nbl['nbl-number'] == 1): + nbl = (res['network'][0]['node'][i]['org-openroadm-otn-network-topology:switching-pools'] + ['odu-switching-pools'][0]['non-blocking-list'][k]) + if nbl['nbl-number'] == 1: self.assertEqual(nbl['available-interconnect-bandwidth'], 10) self.assertEqual(nbl['interconnect-bandwidth-unit'], 1000000000) self.assertIn('XPDR1-NETWORK1', nbl['tp-list']) self.assertIn('XPDR1-CLIENT1', nbl['tp-list']) self.assertNotIn('XPDR1-CLIENT2', nbl['tp-list']) - elif(nodeId == 'SPDR-SA1-XPDR2'): + elif nodeId == 'SPDR-SA1-XPDR2': nodeType = res['network'][0]['node'][i]['org-openroadm-common-network:node-type'] self.assertEqual(nodeType, 'SWITCH') self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'SPDR-SA1'}, @@ -263,33 +223,43 @@ class TransportPCEtesting(unittest.TestCase): client = 0 network = 0 for j in range(0, nbTps): - tpType = res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['org-openroadm-common-network:tp-type'] + tpType = (res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j] + ['org-openroadm-common-network:tp-type']) tpId = res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['tp-id'] - if (tpType == 'XPONDER-CLIENT'): + if tpType == 'XPONDER-CLIENT': client += 1 - self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['org-openroadm-otn-network-topology:tp-supported-interfaces'] - ['supported-interface-capability'][0]['if-cap-type'], 'org-openroadm-port-types:if-100GE-ODU4') - self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j] - ['org-openroadm-otn-network-topology:xpdr-tp-port-connection-attributes']['rate'], 'org-openroadm-otn-common-types:ODU4') - elif (tpType == 'XPONDER-NETWORK'): + self.assertEqual((res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j] + ['org-openroadm-otn-network-topology:tp-supported-interfaces'] + ['supported-interface-capability'][0]['if-cap-type']), + 'org-openroadm-port-types:if-100GE-ODU4') + self.assertEqual((res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j] + ['org-openroadm-otn-network-topology:xpdr-tp-port-connection-attributes'] + ['rate']), + 'org-openroadm-otn-common-types:ODU4') + elif tpType == 'XPONDER-NETWORK': network += 1 - self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['org-openroadm-otn-network-topology:tp-supported-interfaces'] - ['supported-interface-capability'][0]['if-cap-type'], 'org-openroadm-port-types:if-OCH-OTU4-ODU4') - self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j] - ['org-openroadm-otn-network-topology:xpdr-tp-port-connection-attributes']['rate'], 'org-openroadm-otn-common-types:ODU4') - self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point'] - [j]['supporting-termination-point'][0]['network-ref'], 'openroadm-topology') - self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point'] - [j]['supporting-termination-point'][0]['node-ref'], 'SPDR-SA1-XPDR2') - self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point'] - [j]['supporting-termination-point'][0]['tp-ref'], tpId) + self.assertEqual((res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j] + ['org-openroadm-otn-network-topology:tp-supported-interfaces'] + ['supported-interface-capability'][0]['if-cap-type']), + 'org-openroadm-port-types:if-OCH-OTU4-ODU4') + self.assertEqual((res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j] + ['org-openroadm-otn-network-topology:xpdr-tp-port-connection-attributes'] + ['rate']), + 'org-openroadm-otn-common-types:ODU4') + self.assertEqual((res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j] + ['supporting-termination-point'][0]['network-ref']), 'openroadm-topology') + self.assertEqual((res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j] + ['supporting-termination-point'][0]['node-ref']), 'SPDR-SA1-XPDR2') + self.assertEqual((res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j] + ['supporting-termination-point'][0]['tp-ref']), tpId) self.assertTrue(client == 4) self.assertTrue(network == 4) listNode.remove(nodeId) nbNbl = len(res['network'][0]['node'][i]['org-openroadm-otn-network-topology:switching-pools'] ['odu-switching-pools'][0]['non-blocking-list']) self.assertEqual(nbNbl, 1) - nbl = res['network'][0]['node'][i]['org-openroadm-otn-network-topology:switching-pools']['odu-switching-pools'][0]['non-blocking-list'][0] + nbl = (res['network'][0]['node'][i]['org-openroadm-otn-network-topology:switching-pools'] + ['odu-switching-pools'][0]['non-blocking-list'][0]) self.assertIn('XPDR2-NETWORK1', nbl['tp-list']) self.assertIn('XPDR2-CLIENT1', nbl['tp-list']) self.assertIn('XPDR2-NETWORK2', nbl['tp-list']) @@ -303,22 +273,11 @@ class TransportPCEtesting(unittest.TestCase): self.assertEqual(len(listNode), 0) def test_08_disconnect_SPDR_SA1(self): - url = ("{}/config/network-topology:" - "network-topology/topology/topology-netconf/node/SPDR-SA1" - .format(self.restconf_baseurl)) - data = {} - headers = {'content-type': 'application/json'} - response = requests.request( - "DELETE", url, data=json.dumps(data), headers=headers, - auth=('admin', 'admin')) - self.assertEqual(response.status_code, requests.codes.ok) + response = test_utils.unmount_device("SPDR-SA1") + self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200) def test_09_getClliNetwork(self): - url = ("{}/config/ietf-network:networks/network/clli-network" - .format(self.restconf_baseurl)) - headers = {'content-type': 'application/json'} - response = requests.request( - "GET", url, headers=headers, auth=('admin', 'admin')) + response = test_utils.get_clli_net_request() self.assertEqual(response.status_code, requests.codes.ok) res = response.json() nbNode = len(res['network'][0]['node']) @@ -326,30 +285,18 @@ class TransportPCEtesting(unittest.TestCase): self.assertEqual(res['network'][0]['node'][0]['org-openroadm-clli-network:clli'], 'NodeSA') def test_10_getOpenRoadmNetwork(self): - url = ("{}/config/ietf-network:networks/network/openroadm-network" - .format(self.restconf_baseurl)) - headers = {'content-type': 'application/json'} - response = requests.request( - "GET", url, headers=headers, auth=('admin', 'admin')) + response = test_utils.get_ordm_net_request() self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertNotIn('node', res['network'][0]) def test_11_getNodes_OpenRoadmTopology(self): - url = ("{}/config/ietf-network:networks/network/openroadm-topology" - .format(self.restconf_baseurl)) - headers = {'content-type': 'application/json'} - response = requests.request( - "GET", url, headers=headers, auth=('admin', 'admin')) + response = test_utils.get_ordm_topo_request("") res = response.json() self.assertNotIn('node', res['network'][0]) def test_12_getNodes_OtnTopology(self): - url = ("{}/config/ietf-network:networks/network/otn-topology" - .format(self.restconf_baseurl)) - headers = {'content-type': 'application/json'} - response = requests.request( - "GET", url, headers=headers, auth=('admin', 'admin')) + response = test_utils.get_otn_topo_request() res = response.json() self.assertNotIn('node', res['network'][0])