From 166a4562d82e400ae1a2915d38ac6c8efeb8e56e Mon Sep 17 00:00:00 2001 From: Gilles Thouenon Date: Tue, 21 May 2019 17:47:50 +0200 Subject: [PATCH] prepare func tests for openroadm v2.2.1 support add openroamd version 2.2.1 support for the following functional tests: - portmapping - topology - renderer service path nominal - creation/deletion of interfaces and roadm-connection on devices. - add equipment-state in XPDRA configuration file (circcuit-pack). - OLM JIRA: TRNSPRTPCE-112 JIRA: TRNSPRTPCE-113 Change-Id: I746e4962e05ac9570721260a1df289f3bf974448 Signed-off-by: Gilles Thouenon Co-authored-by: Christophe Betoule --- .../openroadm/2.2.1/oper-ROADMA.xml | 2 +- .../openroadm/2.2.1/oper-XPDRA.xml | 2 + .../{ => 1.2.1}/test_end2end.py | 0 .../{ => 1.2.1}/test_olm.py | 0 .../{ => 1.2.1}/test_pce.py | 0 .../{ => 1.2.1}/test_portmapping.py | 0 .../test_renderer_service_path_nominal.py | 0 .../{ => 1.2.1}/test_topoPortMapping.py | 0 .../{ => 1.2.1}/test_topology.py | 0 tests/transportpce_tests/2.2.1/test_olm.py | 997 ++++++++++++++ .../2.2.1/test_portmapping.py | 343 +++++ .../test_renderer_service_path_nominal.py | 598 ++++++++ .../transportpce_tests/2.2.1/test_topology.py | 1213 +++++++++++++++++ tox.ini | 21 +- 14 files changed, 3167 insertions(+), 9 deletions(-) rename tests/transportpce_tests/{ => 1.2.1}/test_end2end.py (100%) rename tests/transportpce_tests/{ => 1.2.1}/test_olm.py (100%) rename tests/transportpce_tests/{ => 1.2.1}/test_pce.py (100%) rename tests/transportpce_tests/{ => 1.2.1}/test_portmapping.py (100%) rename tests/transportpce_tests/{ => 1.2.1}/test_renderer_service_path_nominal.py (100%) rename tests/transportpce_tests/{ => 1.2.1}/test_topoPortMapping.py (100%) rename tests/transportpce_tests/{ => 1.2.1}/test_topology.py (100%) create mode 100644 tests/transportpce_tests/2.2.1/test_olm.py create mode 100644 tests/transportpce_tests/2.2.1/test_portmapping.py create mode 100644 tests/transportpce_tests/2.2.1/test_renderer_service_path_nominal.py create mode 100644 tests/transportpce_tests/2.2.1/test_topology.py diff --git a/tests/sample_configs/openroadm/2.2.1/oper-ROADMA.xml b/tests/sample_configs/openroadm/2.2.1/oper-ROADMA.xml index 888369396..c466f5cfd 100644 --- a/tests/sample_configs/openroadm/2.2.1/oper-ROADMA.xml +++ b/tests/sample_configs/openroadm/2.2.1/oper-ROADMA.xml @@ -770,7 +770,7 @@ rx 15min - -45.1 + -21.1 dBm complete diff --git a/tests/sample_configs/openroadm/2.2.1/oper-XPDRA.xml b/tests/sample_configs/openroadm/2.2.1/oper-XPDRA.xml index b8fa22f19..2bd9f54b3 100644 --- a/tests/sample_configs/openroadm/2.2.1/oper-XPDRA.xml +++ b/tests/sample_configs/openroadm/2.2.1/oper-XPDRA.xml @@ -117,6 +117,7 @@ 1 1 true + not-reserved-available 1 CFP2 @@ -155,6 +156,7 @@ 1 1 true + not-reserved-available C1 QSFP28 diff --git a/tests/transportpce_tests/test_end2end.py b/tests/transportpce_tests/1.2.1/test_end2end.py similarity index 100% rename from tests/transportpce_tests/test_end2end.py rename to tests/transportpce_tests/1.2.1/test_end2end.py diff --git a/tests/transportpce_tests/test_olm.py b/tests/transportpce_tests/1.2.1/test_olm.py similarity index 100% rename from tests/transportpce_tests/test_olm.py rename to tests/transportpce_tests/1.2.1/test_olm.py diff --git a/tests/transportpce_tests/test_pce.py b/tests/transportpce_tests/1.2.1/test_pce.py similarity index 100% rename from tests/transportpce_tests/test_pce.py rename to tests/transportpce_tests/1.2.1/test_pce.py diff --git a/tests/transportpce_tests/test_portmapping.py b/tests/transportpce_tests/1.2.1/test_portmapping.py similarity index 100% rename from tests/transportpce_tests/test_portmapping.py rename to tests/transportpce_tests/1.2.1/test_portmapping.py diff --git a/tests/transportpce_tests/test_renderer_service_path_nominal.py b/tests/transportpce_tests/1.2.1/test_renderer_service_path_nominal.py similarity index 100% rename from tests/transportpce_tests/test_renderer_service_path_nominal.py rename to tests/transportpce_tests/1.2.1/test_renderer_service_path_nominal.py diff --git a/tests/transportpce_tests/test_topoPortMapping.py b/tests/transportpce_tests/1.2.1/test_topoPortMapping.py similarity index 100% rename from tests/transportpce_tests/test_topoPortMapping.py rename to tests/transportpce_tests/1.2.1/test_topoPortMapping.py diff --git a/tests/transportpce_tests/test_topology.py b/tests/transportpce_tests/1.2.1/test_topology.py similarity index 100% rename from tests/transportpce_tests/test_topology.py rename to tests/transportpce_tests/1.2.1/test_topology.py diff --git a/tests/transportpce_tests/2.2.1/test_olm.py b/tests/transportpce_tests/2.2.1/test_olm.py new file mode 100644 index 000000000..83d268ac2 --- /dev/null +++ b/tests/transportpce_tests/2.2.1/test_olm.py @@ -0,0 +1,997 @@ +#!/usr/bin/env python + +############################################################################# +# Copyright (c) 2017 Orange, Inc. and others. All rights reserved. +# +# All rights reserved. This program and the accompanying materials +# are made available under the terms of the Apache License, Version 2.0 +# which accompanies this distribution, and is available at +# http://www.apache.org/licenses/LICENSE-2.0 +############################################################################# + +import unittest +import requests +import time +import subprocess +import signal +import json +import os +import psutil +import shutil +from unittest.result import failfast + + +class TransportOlmTesting(unittest.TestCase): + + honeynode_process1 = None + honeynode_process2 = None + honeynode_process3 = None + honeynode_process4 = None + odl_process = None + restconf_baseurl = "http://localhost:8181/restconf" + +#START_IGNORE_XTESTING + + @classmethod + def __start_honeynode1(cls): + executable = ("./honeynode/2.2.1/honeynode-distribution/target/honeynode-distribution-1.18.01-hc" + "/honeynode-distribution-1.18.01/honeycomb-tpce") + if os.path.isfile(executable): + with open('honeynode1.log', 'w') as outfile: + cls.honeynode_process1 = subprocess.Popen( + [executable, "17840", "sample_configs/openroadm/2.2.1/oper-XPDRA.xml"], + stdout=outfile) + + @classmethod + def __start_honeynode2(cls): + executable = ("./honeynode/2.2.1/honeynode-distribution/target/honeynode-distribution-1.18.01-hc" + "/honeynode-distribution-1.18.01/honeycomb-tpce") + if os.path.isfile(executable): + with open('honeynode2.log', 'w') as outfile: + cls.honeynode_process2 = subprocess.Popen( + [executable, "17841", "sample_configs/openroadm/2.2.1/oper-ROADMA.xml"], + stdout=outfile) + + @classmethod + def __start_honeynode3(cls): + executable = ("./honeynode/2.2.1/honeynode-distribution/target/honeynode-distribution-1.18.01-hc" + "/honeynode-distribution-1.18.01/honeycomb-tpce") + if os.path.isfile(executable): + with open('honeynode3.log', 'w') as outfile: + cls.honeynode_process3 = subprocess.Popen( + [executable, "17843", "sample_configs/openroadm/2.2.1/oper-ROADMC.xml"], + stdout=outfile) + @classmethod + def __start_honeynode4(cls): + executable = ("./honeynode/2.2.1/honeynode-distribution/target/honeynode-distribution-1.18.01-hc" + "/honeynode-distribution-1.18.01/honeycomb-tpce") + if os.path.isfile(executable): + with open('honeynode4.log', 'w') as outfile: + cls.honeynode_process4 = subprocess.Popen( + [executable, "17844", "sample_configs/openroadm/2.2.1/oper-XPDRC.xml"], + stdout=outfile) + + @classmethod + def __start_odl(cls): + executable = "../karaf/target/assembly/bin/karaf" + with open('odl.log', 'w') as outfile: + cls.odl_process = subprocess.Popen( + ["bash", executable, "server"], stdout=outfile, + stdin=open(os.devnull)) + + @classmethod + def setUpClass(cls): + cls.__start_honeynode1() + time.sleep(30) + cls.__start_honeynode2() + time.sleep(30) + cls.__start_honeynode3() + time.sleep(30) + cls.__start_honeynode4() + time.sleep(30) + cls.__start_odl() + time.sleep(60) + + @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.honeynode_process1.pid).children(): + child.send_signal(signal.SIGINT) + child.wait() + cls.honeynode_process1.send_signal(signal.SIGINT) + cls.honeynode_process1.wait() + for child in psutil.Process(cls.honeynode_process2.pid).children(): + child.send_signal(signal.SIGINT) + child.wait() + cls.honeynode_process2.send_signal(signal.SIGINT) + cls.honeynode_process2.wait() + for child in psutil.Process(cls.honeynode_process3.pid).children(): + child.send_signal(signal.SIGINT) + child.wait() + cls.honeynode_process3.send_signal(signal.SIGINT) + cls.honeynode_process3.wait() + for child in psutil.Process(cls.honeynode_process4.pid).children(): + child.send_signal(signal.SIGINT) + child.wait() + cls.honeynode_process4.send_signal(signal.SIGINT) + cls.honeynode_process4.wait() + + def setUp(self): + print ("execution of {}".format(self.id().split(".")[-1])) + time.sleep(1) + +#END_IGNORE_XTESTING + + def test_01_xpdrA_device_connected(self): + url = ("{}/config/network-topology:" + "network-topology/topology/topology-netconf/node/XPDR-A1" + .format(self.restconf_baseurl)) + data = {"node": [{ + "node-id": "XPDR-A1", + "netconf-node-topology:username": "admin", + "netconf-node-topology:password": "admin", + "netconf-node-topology:host": "127.0.0.1", + "netconf-node-topology:port": "17840", + "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) + time.sleep(20) + + def test_02_xpdrC_device_connected(self): + url = ("{}/config/network-topology:" + "network-topology/topology/topology-netconf/node/XPDR-C1" + .format(self.restconf_baseurl)) + data = {"node": [{ + "node-id": "XPDR-C1", + "netconf-node-topology:username": "admin", + "netconf-node-topology:password": "admin", + "netconf-node-topology:host": "127.0.0.1", + "netconf-node-topology:port": "17844", + "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) + time.sleep(20) + + def test_03_rdmA_device_connected(self): + url = ("{}/config/network-topology:" + "network-topology/topology/topology-netconf/node/ROADM-A1" + .format(self.restconf_baseurl)) + data = {"node": [{ + "node-id": "ROADM-A1", + "netconf-node-topology:username": "admin", + "netconf-node-topology:password": "admin", + "netconf-node-topology:host": "127.0.0.1", + "netconf-node-topology:port": "17841", + "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) + time.sleep(20) + + def test_04_rdmC_device_connected(self): + url = ("{}/config/network-topology:" + "network-topology/topology/topology-netconf/node/ROADM-C1" + .format(self.restconf_baseurl)) + data = {"node": [{ + "node-id": "ROADM-C1", + "netconf-node-topology:username": "admin", + "netconf-node-topology:password": "admin", + "netconf-node-topology:host": "127.0.0.1", + "netconf-node-topology:port": "17843", + "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) + time.sleep(20) + + def test_05_connect_xprdA_to_roadmA(self): + url = "{}/operations/transportpce-networkutils:init-xpdr-rdm-links".format(self.restconf_baseurl) + data = { + "networkutils:input": { + "networkutils:links-input": { + "networkutils:xpdr-node": "XPDR-A1", + "networkutils:xpdr-num": "1", + "networkutils:network-num": "1", + "networkutils:rdm-node": "ROADM-A1", + "networkutils:srg-num": "1", + "networkutils:termination-point-num": "SRG1-PP1-TXRX" + } + } + } + headers = {'content-type': 'application/json'} + response = requests.request( + "POST", url, data=json.dumps(data), + headers=headers, auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"]) + + def test_06_connect_roadmA_to_xpdrA(self): + url = "{}/operations/transportpce-networkutils:init-rdm-xpdr-links".format(self.restconf_baseurl) + data = { + "networkutils:input": { + "networkutils:links-input": { + "networkutils:xpdr-node": "XPDR-A1", + "networkutils:xpdr-num": "1", + "networkutils:network-num": "1", + "networkutils:rdm-node": "ROADM-A1", + "networkutils:srg-num": "1", + "networkutils:termination-point-num": "SRG1-PP1-TXRX" + } + } + } + headers = {'content-type': 'application/json'} + response = requests.request( + "POST", url, data=json.dumps(data), + headers=headers, auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + self.assertIn('Roadm Xponder links created successfully', res["output"]["result"]) + + def test_07_connect_xprdC_to_roadmC(self): + url = "{}/operations/transportpce-networkutils:init-xpdr-rdm-links".format(self.restconf_baseurl) + data = { + "networkutils:input": { + "networkutils:links-input": { + "networkutils:xpdr-node": "XPDR-C1", + "networkutils:xpdr-num": "1", + "networkutils:network-num": "1", + "networkutils:rdm-node": "ROADM-C1", + "networkutils:srg-num": "1", + "networkutils:termination-point-num": "SRG1-PP1-TXRX" + } + } + } + headers = {'content-type': 'application/json'} + response = requests.request( + "POST", url, data=json.dumps(data), + headers=headers, auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"]) + + def test_08_connect_roadmC_to_xpdrC(self): + url = "{}/operations/transportpce-networkutils:init-rdm-xpdr-links".format(self.restconf_baseurl) + data = { + "networkutils:input": { + "networkutils:links-input": { + "networkutils:xpdr-node": "XPDR-C1", + "networkutils:xpdr-num": "1", + "networkutils:network-num": "1", + "networkutils:rdm-node": "ROADM-C1", + "networkutils:srg-num": "1", + "networkutils:termination-point-num": "SRG1-PP1-TXRX" + } + } + } + headers = {'content-type': 'application/json'} + response = requests.request( + "POST", url, data=json.dumps(data), + headers=headers, auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + self.assertIn('Roadm Xponder links created successfully', res["output"]["result"]) + + def test_09_create_OTS_ROADMA(self): + url = "{}/operations/transportpce-device-renderer:create-ots-oms".format(self.restconf_baseurl) + data = { + "input" : { + "node-id" : "ROADM-A1", + "logical-connection-point" : "DEG1-TTP-TXRX" + } + } + headers = {'content-type': 'application/json'} + response = requests.request( + "POST", url, data=json.dumps(data), + headers=headers, auth=('admin', 'admin')) + time.sleep(10) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + self.assertIn('Interfaces OTS-DEG1-TTP-TXRX - OMS-DEG1-TTP-TXRX successfully created on node ROADM-A1', + res["output"]["result"]) + + def test_10_create_OTS_ROADMC(self): + url = "{}/operations/transportpce-device-renderer:create-ots-oms".format(self.restconf_baseurl) + data = { + "input" : { + "node-id" : "ROADM-C1", + "logical-connection-point" : "DEG2-TTP-TXRX" + } + } + headers = {'content-type': 'application/json'} + response = requests.request( + "POST", url, data=json.dumps(data), + headers=headers, auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + self.assertIn('Interfaces OTS-DEG2-TTP-TXRX - OMS-DEG2-TTP-TXRX successfully created on node ROADM-C1', + res["output"]["result"]) + + def test_11_get_PM_ROADMA(self): + url = "{}/operations/transportpce-olm:get-pm".format(self.restconf_baseurl) + data = { + "input": { + "node-id": "ROADM-A1", + "resource-type": "interface", + "granularity": "15min", + "resource-identifier": { + "resource-name" : "OTS-DEG2-TTP-TXRX" + } + } + } + headers = {'content-type': 'application/json'} + response = requests.request( + "POST", url, data=json.dumps(data), + headers=headers, auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + self.assertIn({ + "pmparameter-name": "OpticalPowerOutput", + "pmparameter-value": "2.5" + }, res["output"]["measurements"]) + self.assertIn({ + "pmparameter-name": "OpticalReturnLoss", + "pmparameter-value": "40" + }, res["output"]["measurements"]) + self.assertIn({ + "pmparameter-name": "OpticalPowerInput", + "pmparameter-value": "-21.1" + }, res["output"]["measurements"]) + + def test_12_get_PM_ROADMC(self): + url = "{}/operations/transportpce-olm:get-pm".format(self.restconf_baseurl) + data = { + "input": { + "node-id": "ROADM-C1", + "resource-type": "interface", + "granularity": "15min", + "resource-identifier": { + "resource-name" : "OTS-DEG1-TTP-TXRX" + } + } + } + headers = {'content-type': 'application/json'} + response = requests.request( + "POST", url, data=json.dumps(data), + headers=headers, auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + self.assertIn({ + "pmparameter-name": "OpticalPowerOutput", + "pmparameter-value": "4.6" + }, res["output"]["measurements"]) + self.assertIn({ + "pmparameter-name": "OpticalReturnLoss", + "pmparameter-value": "49.1" + }, res["output"]["measurements"]) + self.assertIn({ + "pmparameter-name": "OpticalPowerInput", + "pmparameter-value": "-15.1" + }, res["output"]["measurements"]) + + def test_13_calculate_span_loss_base_ROADMA_ROADMC(self): + url = "{}/operations/transportpce-olm:calculate-spanloss-base".format(self.restconf_baseurl) + data = { + "input": { + "src-type": "link", + "link-id": "ROADM-A1-DEG2-DEG2-TTP-TXRXtoROADM-C1-DEG1-DEG1-TTP-TXRX" + } + } + headers = {'content-type': 'application/json'} + response = requests.request( + "POST", url, data=json.dumps(data), + headers=headers, auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + self.assertIn('Success', + res["output"]["result"]) + self.assertIn({ + "spanloss": "18", + "link-id": "ROADM-A1-DEG2-DEG2-TTP-TXRXtoROADM-C1-DEG1-DEG1-TTP-TXRX" + }, res["output"]["spans"]) + time.sleep(5) + + def test_14_calculate_span_loss_base_all(self): + url = "{}/operations/transportpce-olm:calculate-spanloss-base".format(self.restconf_baseurl) + data = { + "input": { + "src-type": "all" + } + } + headers = {'content-type': 'application/json'} + response = requests.request( + "POST", url, data=json.dumps(data), + headers=headers, auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + self.assertIn('Success', + res["output"]["result"]) + self.assertIn({ + "spanloss": "26", + "link-id": "ROADM-C1-DEG1-DEG1-TTP-TXRXtoROADM-A1-DEG2-DEG2-TTP-TXRX" + }, res["output"]["spans"]) + self.assertIn({ + "spanloss": "18", + "link-id": "ROADM-A1-DEG2-DEG2-TTP-TXRXtoROADM-C1-DEG1-DEG1-TTP-TXRX" + }, res["output"]["spans"]) + time.sleep(5) + + def test_15_get_OTS_DEG2_TTP_TXRX_ROADMA(self): + url = ("{}/config/network-topology:network-topology/topology/topology-netconf/" + "node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/interface/OTS-DEG2-TTP-TXRX/" + "org-openroadm-optical-transport-interfaces:ots".format(self.restconf_baseurl)) + headers = {'content-type': 'application/json'} + response = requests.request( + "GET", url, headers=headers, auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + self.assertEqual(18, res['org-openroadm-optical-transport-interfaces:ots']['span-loss-transmit']) + self.assertEqual(26, res['org-openroadm-optical-transport-interfaces:ots']['span-loss-receive']) + + def test_16_get_OTS_DEG1_TTP_TXRX_ROADMC(self): + url = ("{}/config/network-topology:network-topology/topology/topology-netconf/" + "node/ROADM-C1/yang-ext:mount/org-openroadm-device:org-openroadm-device/interface/OTS-DEG1-TTP-TXRX/" + "org-openroadm-optical-transport-interfaces:ots".format(self.restconf_baseurl)) + headers = {'content-type': 'application/json'} + response = requests.request( + "GET", url, headers=headers, auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + self.assertEqual(26, res['org-openroadm-optical-transport-interfaces:ots']['span-loss-transmit']) + self.assertEqual(18, res['org-openroadm-optical-transport-interfaces:ots']['span-loss-receive']) + + def test_17_servicePath_create_AToZ(self): + url = "{}/operations/transportpce-device-renderer:service-path".format(self.restconf_baseurl) + data = { + "input": { + "service-name": "test", + "wave-number": "1", + "modulation-format": "qpsk", + "operation": "create", + "nodes": [ + { + "dest-tp": "XPDR1-NETWORK1", + "src-tp": "XPDR1-CLIENT1", + "node-id": "XPDR-A1" + }, + { + "dest-tp": "DEG2-TTP-TXRX", + "src-tp": "SRG1-PP1-TXRX", + "node-id": "ROADM-A1" + }, + { + "dest-tp": "SRG1-PP1-TXRX", + "src-tp": "DEG1-TTP-TXRX", + "node-id": "ROADM-C1" + }, + { + "dest-tp": "XPDR1-CLIENT1", + "src-tp": "XPDR1-NETWORK1", + "node-id": "XPDR-C1" + } + ] + } + } + headers = {'content-type': 'application/json'} + response = requests.request( + "POST", url, data=json.dumps(data), + headers=headers, auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + self.assertIn('Roadm-connection successfully created for nodes', res["output"]["result"]) + time.sleep(40) + + def test_18_servicePath_create_ZToA(self): + url = "{}/operations/transportpce-device-renderer:service-path".format(self.restconf_baseurl) + data = { + "input": { + "service-name": "test", + "wave-number": "1", + "modulation-format": "qpsk", + "operation": "create", + "nodes": [ + { + "dest-tp": "XPDR1-NETWORK1", + "src-tp": "XPDR1-CLIENT1", + "node-id": "XPDR-C1" + }, + { + "dest-tp": "DEG1-TTP-TXRX", + "src-tp": "SRG1-PP1-TXRX", + "node-id": "ROADM-C1" + }, + { + "src-tp": "DEG2-TTP-TXRX", + "dest-tp": "SRG1-PP1-TXRX", + "node-id": "ROADM-A1" + }, + { + "src-tp": "XPDR1-NETWORK1", + "dest-tp": "XPDR1-CLIENT1", + "node-id": "XPDR-A1" + } + ] + } + } + headers = {'content-type': 'application/json'} + response = requests.request( + "POST", url, data=json.dumps(data), + headers=headers, auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + self.assertIn('Roadm-connection successfully created for nodes', res["output"]["result"]) + time.sleep(40) + + def test_19_service_power_setup_XPDRA_XPDRC(self): + url = "{}/operations/transportpce-olm:service-power-setup".format(self.restconf_baseurl) + data = { + "input": { + "service-name": "test", + "wave-number": 1, + "nodes": [ + { + "dest-tp": "XPDR1-NETWORK1", + "src-tp": "XPDR1-CLIENT1", + "node-id": "XPDR-A1" + }, + { + "dest-tp": "DEG2-TTP-TXRX", + "src-tp": "SRG1-PP1-TXRX", + "node-id": "ROADM-A1" + }, + { + "dest-tp": "SRG1-PP1-TXRX", + "src-tp": "DEG1-TTP-TXRX", + "node-id": "ROADM-C1" + }, + { + "dest-tp": "XPDR1-CLIENT1", + "src-tp": "XPDR1-NETWORK1", + "node-id": "XPDR-C1" + } + ] + } + } + headers = {'content-type': 'application/json'} + response = requests.request( + "POST", url, data=json.dumps(data), + headers=headers, auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + self.assertIn('Success', res["output"]["result"]) + + def test_20_get_interface_XPDRA_XPDR1_NETWORK1(self): + url = ("{}/config/network-topology:network-topology/topology/topology-netconf/node/XPDR-A1/yang-ext:mount/" + "org-openroadm-device:org-openroadm-device/interface/XPDR1-NETWORK1-1/" + "org-openroadm-optical-channel-interfaces:och".format(self.restconf_baseurl)) + headers = {'content-type': 'application/json'} + response = requests.request( + "GET", url, headers=headers, auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + self.assertEqual(-5, res['org-openroadm-optical-channel-interfaces:och']['transmit-power']) + self.assertEqual(196.1, res['org-openroadm-optical-channel-interfaces:och']['frequency']) + + def test_21_get_roadmconnection_ROADMA(self): + url = ("{}/config/network-topology:network-topology/topology/topology-netconf/node/ROADM-A1/yang-ext:mount/" + "org-openroadm-device:org-openroadm-device/roadm-connections/" + "SRG1-PP1-TXRX-DEG2-TTP-TXRX-1".format(self.restconf_baseurl)) + headers = {'content-type': 'application/json'} + response = requests.request( + "GET", url, headers=headers, auth=('admin', 'admin')) + 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): + url = ("{}/config/network-topology:network-topology/topology/topology-netconf/node/ROADM-C1/yang-ext:mount/" + "org-openroadm-device:org-openroadm-device/roadm-connections/" + "DEG1-TTP-TXRX-SRG1-PP1-TXRX-1".format(self.restconf_baseurl)) + headers = {'content-type': 'application/json'} + response = requests.request( + "GET", url, headers=headers, auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + self.assertEqual("power", res['roadm-connections'][0]['opticalControlMode']) + + def test_23_service_power_setup_XPDRC_XPDRA(self): + url = "{}/operations/transportpce-olm:service-power-setup".format(self.restconf_baseurl) + data = { + "input": { + "service-name": "test", + "wave-number": 1, + "nodes": [ + { + "dest-tp": "XPDR1-NETWORK1", + "src-tp": "XPDR1-CLIENT1", + "node-id": "XPDR-C1" + }, + { + "dest-tp": "DEG1-TTP-TXRX", + "src-tp": "SRG1-PP1-TXRX", + "node-id": "ROADM-C1" + }, + { + "src-tp": "DEG2-TTP-TXRX", + "dest-tp": "SRG1-PP1-TXRX", + "node-id": "ROADM-A1" + }, + { + "src-tp": "XPDR1-NETWORK1", + "dest-tp": "XPDR1-CLIENT1", + "node-id": "XPDR-A1" + } + ] + } + } + headers = {'content-type': 'application/json'} + response = requests.request( + "POST", url, data=json.dumps(data), + headers=headers, auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + self.assertIn('Success', res["output"]["result"]) + + def test_24_get_interface_XPDRC_XPDR1_NETWORK1(self): + url = ("{}/config/network-topology:network-topology/topology/topology-netconf/node/XPDR-C1/yang-ext:mount/" + "org-openroadm-device:org-openroadm-device/interface/XPDR1-NETWORK1-1/" + "org-openroadm-optical-channel-interfaces:och".format(self.restconf_baseurl)) + headers = {'content-type': 'application/json'} + response = requests.request( + "GET", url, headers=headers, auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + self.assertEqual(-5 , res['org-openroadm-optical-channel-interfaces:och']['transmit-power']) + self.assertEqual(196.1, res['org-openroadm-optical-channel-interfaces:och']['frequency']) + + def test_25_get_roadmconnection_ROADMC(self): + url = ("{}/config/network-topology:network-topology/topology/topology-netconf/node/ROADM-C1/yang-ext:mount/" + "org-openroadm-device:org-openroadm-device/roadm-connections/" + "SRG1-PP1-TXRX-DEG1-TTP-TXRX-1".format(self.restconf_baseurl)) + headers = {'content-type': 'application/json'} + response = requests.request( + "GET", url, headers=headers, auth=('admin', 'admin')) + 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_26_service_power_turndown_XPDRA_XPDRC(self): + url = "{}/operations/transportpce-olm:service-power-turndown".format(self.restconf_baseurl) + data = { + "input": { + "service-name": "test", + "wave-number": 1, + "nodes": [ + { + "dest-tp": "XPDR1-NETWORK1", + "src-tp": "XPDR1-CLIENT1", + "node-id": "XPDR-A1" + }, + { + "dest-tp": "DEG2-TTP-TXRX", + "src-tp": "SRG1-PP1-TXRX", + "node-id": "ROADM-A1" + }, + { + "dest-tp": "SRG1-PP1-TXRX", + "src-tp": "DEG1-TTP-TXRX", + "node-id": "ROADM-C1" + }, + { + "dest-tp": "XPDR1-CLIENT1", + "src-tp": "XPDR1-NETWORK1", + "node-id": "XPDR-C1" + } + ] + } + } + headers = {'content-type': 'application/json'} + response = requests.request( + "POST", url, data=json.dumps(data), + headers=headers, auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + self.assertIn('Success', res["output"]["result"]) + + def test_27_get_roadmconnection_ROADMA(self): + url = ("{}/config/network-topology:network-topology/topology/topology-netconf/node/ROADM-A1/yang-ext:mount/" + "org-openroadm-device:org-openroadm-device/roadm-connections/" + "SRG1-PP1-TXRX-DEG2-TTP-TXRX-1".format(self.restconf_baseurl)) + headers = {'content-type': 'application/json'} + response = requests.request( + "GET", url, headers=headers, auth=('admin', 'admin')) + 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): + url = ("{}/config/network-topology:network-topology/topology/topology-netconf/node/ROADM-C1/yang-ext:mount/" + "org-openroadm-device:org-openroadm-device/roadm-connections/" + "DEG1-TTP-TXRX-SRG1-PP1-TXRX-1".format(self.restconf_baseurl)) + headers = {'content-type': 'application/json'} + response = requests.request( + "GET", url, headers=headers, auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + self.assertEqual("off", res['roadm-connections'][0]['opticalControlMode']) + + def test_29_servicePath_delete_AToZ(self): + url = "{}/operations/transportpce-device-renderer:service-path".format(self.restconf_baseurl) + data = { + "input": { + "service-name": "test", + "wave-number": "1", + "modulation-format": "qpsk", + "operation": "delete", + "nodes": [ + { + "dest-tp": "XPDR1-NETWORK1", + "src-tp": "XPDR1-CLIENT1", + "node-id": "XPDR-A1" + }, + { + "dest-tp": "DEG2-TTP-TXRX", + "src-tp": "SRG1-PP1-TXRX", + "node-id": "ROADM-A1" + }, + { + "dest-tp": "SRG1-PP1-TXRX", + "src-tp": "DEG1-TTP-TXRX", + "node-id": "ROADM-C1" + }, + { + "dest-tp": "XPDR1-CLIENT1", + "src-tp": "XPDR1-NETWORK1", + "node-id": "XPDR-C1" + } + ] + } + } + headers = {'content-type': 'application/json'} + response = requests.request( + "POST", url, data=json.dumps(data), + headers=headers, auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + self.assertIn('Request processed', res["output"]["result"]) + time.sleep(10) + + def test_30_servicePath_delete_ZToA(self): + url = "{}/operations/transportpce-device-renderer:service-path".format(self.restconf_baseurl) + data = { + "input": { + "service-name": "test", + "wave-number": "1", + "modulation-format": "qpsk", + "operation": "delete", + "nodes": [ + { + "dest-tp": "XPDR1-NETWORK1", + "src-tp": "XPDR1-CLIENT1", + "node-id": "XPDR-C1" + }, + { + "dest-tp": "DEG1-TTP-TXRX", + "src-tp": "SRG1-PP1-TXRX", + "node-id": "ROADM-C1" + }, + { + "src-tp": "DEG2-TTP-TXRX", + "dest-tp": "SRG1-PP1-TXRX", + "node-id": "ROADM-A1" + }, + { + "src-tp": "XPDR1-NETWORK1", + "dest-tp": "XPDR1-CLIENT1", + "node-id": "XPDR-A1" + } + ] + } + } + headers = {'content-type': 'application/json'} + response = requests.request( + "POST", url, data=json.dumps(data), + headers=headers, auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + self.assertIn('Request processed', res["output"]["result"]) + time.sleep(10) + + """to test case where SRG where the xpdr is connected to has no optical range data""" + + def test_31_connect_xprdA_to_roadmA(self): + url = "{}/operations/transportpce-networkutils:init-xpdr-rdm-links".format(self.restconf_baseurl) + data = { + "networkutils:input": { + "networkutils:links-input": { + "networkutils:xpdr-node": "XPDR-A1", + "networkutils:xpdr-num": "1", + "networkutils:network-num": "2", + "networkutils:rdm-node": "ROADM-A1", + "networkutils:srg-num": "1", + "networkutils:termination-point-num": "SRG1-PP2-TXRX" + } + } + } + headers = {'content-type': 'application/json'} + response = requests.request( + "POST", url, data=json.dumps(data), + headers=headers, auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"]) + + def test_32_connect_roadmA_to_xpdrA(self): + url = "{}/operations/transportpce-networkutils:init-rdm-xpdr-links".format(self.restconf_baseurl) + data = { + "networkutils:input": { + "networkutils:links-input": { + "networkutils:xpdr-node": "XPDR-A1", + "networkutils:xpdr-num": "1", + "networkutils:network-num": "2", + "networkutils:rdm-node": "ROADM-A1", + "networkutils:srg-num": "1", + "networkutils:termination-point-num": "SRG1-PP2-TXRX" + } + } + } + headers = {'content-type': 'application/json'} + response = requests.request( + "POST", url, data=json.dumps(data), + headers=headers, auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + self.assertIn('Roadm Xponder links created successfully', res["output"]["result"]) + + def test_33_servicePath_create_AToZ(self): + url = "{}/operations/transportpce-device-renderer:service-path".format(self.restconf_baseurl) + data = { + "input": { + "service-name": "test2", + "wave-number": "2", + "modulation-format": "qpsk", + "operation": "create", + "nodes": [ + { + "dest-tp": "XPDR1-NETWORK2", + "src-tp": "XPDR1-CLIENT2", + "node-id": "XPDR-A1" + }, + { + "dest-tp": "DEG2-TTP-TXRX", + "src-tp": "SRG1-PP2-TXRX", + "node-id": "ROADM-A1" + } + ] + } + } + headers = {'content-type': 'application/json'} + response = requests.request( + "POST", url, data=json.dumps(data), + headers=headers, auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + self.assertIn('Roadm-connection successfully created for nodes', res["output"]["result"]) + time.sleep(40) + + def test_34_get_interface_XPDRA_XPDR1_NETWORK2(self): + url = ("{}/config/network-topology:network-topology/topology/topology-netconf/node/XPDR-A1/yang-ext:mount/" + "org-openroadm-device:org-openroadm-device/interface/XPDR1-NETWORK2-2/" + "org-openroadm-optical-channel-interfaces:och".format(self.restconf_baseurl)) + headers = {'content-type': 'application/json'} + response = requests.request( + "GET", url, headers=headers, auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + self.assertEqual(-5, res['org-openroadm-optical-channel-interfaces:och']['transmit-power']) +# self.assertEqual(2, res['org-openroadm-optical-channel-interfaces:och']['wavelength-number']) + + def test_35_servicePath_delete_AToZ(self): + url = "{}/operations/transportpce-device-renderer:service-path".format(self.restconf_baseurl) + data = { + "input": { + "service-name": "test", + "wave-number": "1", + "modulation-format": "qpsk", + "operation": "delete", + "nodes": [ + { + "dest-tp": "XPDR1-NETWORK2", + "src-tp": "XPDR1-CLIENT2", + "node-id": "XPDR-A1" + }, + { + "dest-tp": "DEG2-TTP-TXRX", + "src-tp": "SRG1-PP2-TXRX", + "node-id": "ROADM-A1" + } + ] + } + } + headers = {'content-type': 'application/json'} + response = requests.request( + "POST", url, data=json.dumps(data), + headers=headers, auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + self.assertIn('Request processed', res["output"]["result"]) + time.sleep(10) + + def test_36_xpdrA_device_disconnected(self): + url = ("{}/config/network-topology:" + "network-topology/topology/topology-netconf/node/XPDR-A1" + .format(self.restconf_baseurl)) + headers = {'content-type': 'application/json'} + response = requests.request( + "DELETE", url, headers=headers, + auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.ok) + time.sleep(10) + + def test_37_xpdrC_device_disconnected(self): + url = ("{}/config/network-topology:" + "network-topology/topology/topology-netconf/node/XPDR-C1" + .format(self.restconf_baseurl)) + headers = {'content-type': 'application/json'} + response = requests.request( + "DELETE", url, headers=headers, + auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.ok) + time.sleep(10) + + def test_38_calculate_span_loss_current(self): + url = "{}/operations/transportpce-olm:calculate-spanloss-current".format(self.restconf_baseurl) + headers = {'content-type': 'application/json'} + response = requests.request( + "POST", url, headers=headers, auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + self.assertIn('Success', + res["output"]["result"]) + time.sleep(5) + + def test_39_rdmA_device_disconnected(self): + url = ("{}/config/network-topology:" + "network-topology/topology/topology-netconf/node/ROADM-A1" + .format(self.restconf_baseurl)) + headers = {'content-type': 'application/json'} + response = requests.request( + "DELETE", url, headers=headers, + auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.ok) + time.sleep(10) + + def test_40_rdmC_device_disconnected(self): + url = ("{}/config/network-topology:" + "network-topology/topology/topology-netconf/node/ROADM-C1" + .format(self.restconf_baseurl)) + headers = {'content-type': 'application/json'} + response = requests.request( + "DELETE", url, headers=headers, + auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.ok) + time.sleep(10) + +if __name__ == "__main__": + unittest.main(verbosity=2) diff --git a/tests/transportpce_tests/2.2.1/test_portmapping.py b/tests/transportpce_tests/2.2.1/test_portmapping.py new file mode 100644 index 000000000..de1dbcfc8 --- /dev/null +++ b/tests/transportpce_tests/2.2.1/test_portmapping.py @@ -0,0 +1,343 @@ +#!/usr/bin/env python + +############################################################################## +# Copyright (c) 2019 Orange, Inc. and others. All rights reserved. +# +# All rights reserved. This program and the accompanying materials +# are made available under the terms of the Apache License, Version 2.0 +# which accompanies this distribution, and is available at +# http://www.apache.org/licenses/LICENSE-2.0 +############################################################################## + +import json +import os +import psutil +import requests +import signal +import shutil +import subprocess +import time +import unittest + + +class TransportPCEPortMappingTesting(unittest.TestCase): + + honeynode_process1 = None + honeynode_process2 = None + odl_process = None + restconf_baseurl = "http://localhost:8181/restconf" + + @classmethod + def __start_honeynode1(cls): + executable = ("./honeynode/2.2.1/honeynode-distribution/target/honeynode-distribution-1.18.01-hc" + "/honeynode-distribution-1.18.01/honeycomb-tpce") + if os.path.isfile(executable): + with open('honeynode1.log', 'w') as outfile: + cls.honeynode_process1 = subprocess.Popen( + [executable, "17840", "sample_configs/openroadm/2.2.1/oper-XPDRA.xml"], + stdout=outfile) + + @classmethod + def __start_honeynode2(cls): + executable = ("./honeynode/2.2.1/honeynode-distribution/target/honeynode-distribution-1.18.01-hc" + "/honeynode-distribution-1.18.01/honeycomb-tpce") + if os.path.isfile(executable): + with open('honeynode2.log', 'w') as outfile: + cls.honeynode_process2 = subprocess.Popen( + [executable, "17841", "sample_configs/openroadm/2.2.1/oper-ROADMA.xml"], + stdout=outfile) + + @classmethod + def __start_odl(cls): + executable = "../karaf/target/assembly/bin/karaf" + with open('odl.log', 'w') as outfile: + cls.odl_process = subprocess.Popen( + ["bash", executable, "server"], stdout=outfile, + stdin=open(os.devnull)) + + @classmethod + def setUpClass(cls): + cls.__start_honeynode1() + time.sleep(20) + cls.__start_honeynode2() + time.sleep(20) + cls.__start_odl() + time.sleep(60) + + @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.honeynode_process1.pid).children(): + child.send_signal(signal.SIGINT) + child.wait() + cls.honeynode_process1.send_signal(signal.SIGINT) + cls.honeynode_process1.wait() + for child in psutil.Process(cls.honeynode_process2.pid).children(): + child.send_signal(signal.SIGINT) + child.wait() + cls.honeynode_process2.send_signal(signal.SIGINT) + cls.honeynode_process2.wait() + + def setUp(self): + print ("execution of {}".format(self.id().split(".")[-1])) + time.sleep(10) + + def test_01_rdm_device_connected(self): + url = ("{}/config/network-topology:" + "network-topology/topology/topology-netconf/node/ROADM-A1" + .format(self.restconf_baseurl)) + data = {"node": [{ + "node-id": "ROADM-A1", + "netconf-node-topology:username": "admin", + "netconf-node-topology:password": "admin", + "netconf-node-topology:host": "127.0.0.1", + "netconf-node-topology:port": "17841", + "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) + time.sleep(20) + + def test_02_rdm_device_connected(self): + url = ("{}/operational/network-topology:" + "network-topology/topology/topology-netconf/node/ROADM-A1" + .format(self.restconf_baseurl)) + headers = {'content-type': 'application/json'} + response = requests.request( + "GET", url, headers=headers, auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + self.assertEqual( + res['node'][0]['netconf-node-topology:connection-status'], + 'connected') + time.sleep(10) + + def test_03_rdm_portmapping_DEG1_TTP_TXRX(self): + url = ("{}/config/transportpce-portmapping:network/" + "nodes/ROADM-A1/mapping/DEG1-TTP-TXRX" + .format(self.restconf_baseurl)) + headers = {'content-type': 'application/json'} + response = requests.request( + "GET", url, headers=headers, auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + self.assertIn( + {'supporting-port': 'L1', 'supporting-circuit-pack-name': '1/0', + 'logical-connection-point': 'DEG1-TTP-TXRX'}, + res['mapping']) + + def test_04_rdm_portmapping_DEG2_TTP_TXRX_with_ots_oms(self): + url = ("{}/config/transportpce-portmapping:network/" + "nodes/ROADM-A1/mapping/DEG2-TTP-TXRX" + .format(self.restconf_baseurl)) + headers = {'content-type': 'application/json'} + response = requests.request( + "GET", url, headers=headers, auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + self.assertIn( + {'supporting-port': 'L1', 'supporting-circuit-pack-name': '2/0', + 'logical-connection-point': 'DEG2-TTP-TXRX', + 'supporting-oms': 'OMS-DEG2-TTP-TXRX', 'supporting-ots': 'OTS-DEG2-TTP-TXRX'}, + res['mapping']) + + def test_05_rdm_portmapping_SRG1_PP3_TXRX(self): + url = ("{}/config/transportpce-portmapping:network/" + "nodes/ROADM-A1/mapping/SRG1-PP3-TXRX" + .format(self.restconf_baseurl)) + headers = {'content-type': 'application/json'} + response = requests.request( + "GET", url, headers=headers, auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + self.assertIn( + {'supporting-port': 'C3', 'supporting-circuit-pack-name': '3/0', + 'logical-connection-point': 'SRG1-PP3-TXRX'}, + res['mapping']) + + def test_06_rdm_portmapping_SRG3_PP1_TXRX(self): + url = ("{}/config/transportpce-portmapping:network/" + "nodes/ROADM-A1/mapping/SRG3-PP1-TXRX" + .format(self.restconf_baseurl)) + headers = {'content-type': 'application/json'} + response = requests.request( + "GET", url, headers=headers, auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + self.assertIn( + {'supporting-port': 'C1', 'supporting-circuit-pack-name': '5/0', + 'logical-connection-point': 'SRG3-PP1-TXRX'}, + res['mapping']) + + def test_07_xpdr_device_connected(self): + url = ("{}/config/network-topology:" + "network-topology/topology/topology-netconf/node/XPDR-A1" + .format(self.restconf_baseurl)) + data = {"node": [{ + "node-id": "XPDR-A1", + "netconf-node-topology:username": "admin", + "netconf-node-topology:password": "admin", + "netconf-node-topology:host": "127.0.0.1", + "netconf-node-topology:port": "17840", + "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) + time.sleep(20) + + def test_08_xpdr_device_connected(self): + url = ("{}/operational/network-topology:" + "network-topology/topology/topology-netconf/node/XPDR-A1" + .format(self.restconf_baseurl)) + headers = {'content-type': 'application/json'} + response = requests.request( + "GET", url, headers=headers, auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + self.assertEqual( + res['node'][0]['netconf-node-topology:connection-status'], + 'connected') + time.sleep(10) + + def test_09_xpdr_portmapping_NETWORK1(self): + url = ("{}/config/transportpce-portmapping:network/" + "nodes/XPDR-A1/mapping/XPDR1-NETWORK1" + .format(self.restconf_baseurl)) + headers = {'content-type': 'application/json'} + response = requests.request( + "GET", url, headers=headers, auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + self.assertIn( + {'supporting-port': '1', 'supporting-circuit-pack-name': '1/0/1-PLUG-NET', + 'logical-connection-point': 'XPDR1-NETWORK1'}, + res['mapping']) + + def test_10_xpdr_portmapping_NETWORK2(self): + url = ("{}/config/transportpce-portmapping:network/" + "nodes/XPDR-A1/mapping/XPDR1-NETWORK2" + .format(self.restconf_baseurl)) + headers = {'content-type': 'application/json'} + response = requests.request( + "GET", url, headers=headers, auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + self.assertIn( + {'supporting-port': '1', 'supporting-circuit-pack-name': '1/0/2-PLUG-NET', + 'logical-connection-point': 'XPDR1-NETWORK2'}, + res['mapping']) + + def test_11_xpdr_portmapping_CLIENT1(self): + url = ("{}/config/transportpce-portmapping:network/" + "nodes/XPDR-A1/mapping/XPDR1-CLIENT1" + .format(self.restconf_baseurl)) + headers = {'content-type': 'application/json'} + response = requests.request( + "GET", url, headers=headers, auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + self.assertIn( + {'supporting-port': 'C1', + 'supporting-circuit-pack-name': '1/0/1-PLUG-CLIENT', + 'logical-connection-point': 'XPDR1-CLIENT1'}, + res['mapping']) + + def test_12_xpdr_portmapping_CLIENT2(self): + url = ("{}/config/transportpce-portmapping:network/" + "nodes/XPDR-A1/mapping/XPDR1-CLIENT2" + .format(self.restconf_baseurl)) + headers = {'content-type': 'application/json'} + response = requests.request( + "GET", url, headers=headers, auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + self.assertIn( + {'supporting-port': 'C1', + 'supporting-circuit-pack-name': '1/0/2-PLUG-CLIENT', + 'logical-connection-point': 'XPDR1-CLIENT2'}, + res['mapping']) + + def test_13_xpdr_device_disconnected(self): + url = ("{}/config/network-topology:" + "network-topology/topology/topology-netconf/node/XPDR-A1" + .format(self.restconf_baseurl)) + headers = {'content-type': 'application/json'} + response = requests.request( + "DELETE", url, headers=headers, + auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.ok) + time.sleep(20) + + def test_14_xpdr_device_disconnected(self): + url = ("{}/operational/network-topology:network-topology/topology/" + "topology-netconf/node/XPDR-A1".format(self.restconf_baseurl)) + headers = {'content-type': 'application/json'} + response = requests.request( + "GET", url, headers=headers, auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.not_found) + res = response.json() + self.assertIn( + {"error-type":"application", "error-tag":"data-missing", + "error-message":"Request could not be completed because the relevant data model content does not exist"}, + res['errors']['error']) + + def test_15_xpdr_device_disconnected(self): + url = ("{}/config/transportpce-portmapping:network/nodes/XPDR-A1".format(self.restconf_baseurl)) + headers = {'content-type': 'application/json'} + response = requests.request( + "GET", url, headers=headers, auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.not_found) + res = response.json() + self.assertIn( + {"error-type":"application", "error-tag":"data-missing", + "error-message":"Request could not be completed because the relevant data model content does not exist"}, + res['errors']['error']) + + def test_16_rdm_device_disconnected(self): + url = ("{}/config/network-topology:network-topology/topology/topology-netconf/node/ROADM-A1" + .format(self.restconf_baseurl)) + headers = {'content-type': 'application/json'} + response = requests.request( + "DELETE", url, headers=headers, + auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.ok) + time.sleep(20) + + def test_17_rdm_device_disconnected(self): + url = ("{}/operational/network-topology:network-topology/topology/topology-netconf/node/ROADM-A1" + .format(self.restconf_baseurl)) + headers = {'content-type': 'application/json'} + response = requests.request( + "GET", url, headers=headers, auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.not_found) + res = response.json() + self.assertIn( + {"error-type":"application", "error-tag":"data-missing", + "error-message":"Request could not be completed because the relevant data model content does not exist"}, + res['errors']['error']) + + def test_18_rdm_device_disconnected(self): + url = ("{}/config/transportpce-portmapping:network/nodes/ROADM-A1".format(self.restconf_baseurl)) + headers = {'content-type': 'application/json'} + response = requests.request( + "GET", url, headers=headers, auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.not_found) + res = response.json() + self.assertIn( + {"error-type":"application", "error-tag":"data-missing", + "error-message":"Request could not be completed because the relevant data model content does not exist"}, + res['errors']['error']) + + +if __name__ == "__main__": + unittest.main(verbosity=2) diff --git a/tests/transportpce_tests/2.2.1/test_renderer_service_path_nominal.py b/tests/transportpce_tests/2.2.1/test_renderer_service_path_nominal.py new file mode 100644 index 000000000..31fe8ab67 --- /dev/null +++ b/tests/transportpce_tests/2.2.1/test_renderer_service_path_nominal.py @@ -0,0 +1,598 @@ +#!/usr/bin/env python + +############################################################################# +# Copyright (c) 2017 Orange, Inc. and others. All rights reserved. +# +# All rights reserved. This program and the accompanying materials +# are made available under the terms of the Apache License, Version 2.0 +# which accompanies this distribution, and is available at +# http://www.apache.org/licenses/LICENSE-2.0 +############################################################################# + +import unittest +import requests +import time +import subprocess +import signal +import json +import os +import psutil +import shutil +from unittest.result import failfast + + +class TransportPCERendererTesting(unittest.TestCase): + + honeynode_process1 = None + honeynode_process2 = None + odl_process = None + restconf_baseurl = "http://localhost:8181/restconf" + +#START_IGNORE_XTESTING + + @classmethod + def __start_honeynode1(cls): + executable = ("./honeynode/2.2.1/honeynode-distribution/target/honeynode-distribution-1.18.01-hc" + "/honeynode-distribution-1.18.01/honeycomb-tpce") + if os.path.isfile(executable): + with open('honeynode1.log', 'w') as outfile: + cls.honeynode_process1 = subprocess.Popen( + [executable, "17841", "sample_configs/openroadm/2.2.1/oper-ROADMA.xml"], + stdout=outfile) + + @classmethod + def __start_honeynode2(cls): + executable = ("./honeynode/2.2.1/honeynode-distribution/target/honeynode-distribution-1.18.01-hc" + "/honeynode-distribution-1.18.01/honeycomb-tpce") + if os.path.isfile(executable): + with open('honeynode2.log', 'w') as outfile: + cls.honeynode_process2 = subprocess.Popen( + [executable, "17840", "sample_configs/openroadm/2.2.1/oper-XPDRA.xml"], + stdout=outfile) + + @classmethod + def __start_odl(cls): + executable = "../karaf/target/assembly/bin/karaf" + with open('odl.log', 'w') as outfile: + cls.odl_process = subprocess.Popen( + ["bash", executable, "server"], stdout=outfile, + stdin=open(os.devnull)) + + @classmethod + def setUpClass(cls): + cls.__start_honeynode1() + time.sleep(20) + cls.__start_honeynode2() + time.sleep(20) + cls.__start_odl() + time.sleep(60) + + @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.honeynode_process1.pid).children(): + child.send_signal(signal.SIGINT) + child.wait() + cls.honeynode_process1.send_signal(signal.SIGINT) + cls.honeynode_process1.wait() + for child in psutil.Process(cls.honeynode_process2.pid).children(): + child.send_signal(signal.SIGINT) + child.wait() + cls.honeynode_process2.send_signal(signal.SIGINT) + cls.honeynode_process2.wait() + + def setUp(self): + print ("execution of {}".format(self.id().split(".")[-1])) + time.sleep(10) + +#END_IGNORE_XTESTING + + def test_01_rdm_device_connected(self): + url = ("{}/config/network-topology:" + "network-topology/topology/topology-netconf/node/ROADM-A1" + .format(self.restconf_baseurl)) + data = {"node": [{ + "node-id": "ROADM-A1", + "netconf-node-topology:username": "admin", + "netconf-node-topology:password": "admin", + "netconf-node-topology:host": "127.0.0.1", + "netconf-node-topology:port": "17841", + "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) + time.sleep(20) + + def test_02_xpdr_device_connected(self): + url = ("{}/config/network-topology:" + "network-topology/topology/topology-netconf/node/XPDR-A1" + .format(self.restconf_baseurl)) + data = {"node": [{ + "node-id": "XPDR-A1", + "netconf-node-topology:username": "admin", + "netconf-node-topology:password": "admin", + "netconf-node-topology:host": "127.0.0.1", + "netconf-node-topology:port": "17840", + "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) + time.sleep(20) + + def test_03_rdm_portmapping(self): + url = ("{}/config/transportpce-portmapping:network/" + "nodes/ROADM-A1" + .format(self.restconf_baseurl)) + headers = {'content-type': 'application/json'} + response = requests.request( + "GET", url, headers=headers, auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + self.assertIn( + {'supporting-port': 'L1', 'supporting-circuit-pack-name': '1/0', + 'logical-connection-point': 'DEG1-TTP-TXRX'}, + res['nodes'][0]['mapping']) + self.assertIn( + {'supporting-port': 'C3', 'supporting-circuit-pack-name': '3/0', + 'logical-connection-point': 'SRG1-PP3-TXRX'}, + res['nodes'][0]['mapping']) + + def test_04_xpdr_portmapping(self): + url = ("{}/config/transportpce-portmapping:network/" + "nodes/XPDR-A1" + .format(self.restconf_baseurl)) + headers = {'content-type': 'application/json'} + response = requests.request( + "GET", url, headers=headers, auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + self.assertIn( + {'supporting-port': '1', 'supporting-circuit-pack-name': '1/0/1-PLUG-NET', + 'logical-connection-point': 'XPDR1-NETWORK1'}, + res['nodes'][0]['mapping']) + self.assertIn( + {'supporting-port': 'C1', + 'supporting-circuit-pack-name': '1/0/1-PLUG-CLIENT', + 'logical-connection-point': 'XPDR1-CLIENT1'}, + res['nodes'][0]['mapping']) + + def test_05_service_path_create(self): + url = "{}/operations/transportpce-device-renderer:service-path".format(self.restconf_baseurl) + data = {"renderer:input": { + "renderer:service-name": "service_test", + "renderer:wave-number": "7", + "renderer:modulation-format": "qpsk", + "renderer:operation": "create", + "renderer:nodes": [ + {"renderer:node-id": "ROADM-A1", + "renderer:src-tp": "SRG1-PP3-TXRX", + "renderer:dest-tp": "DEG1-TTP-TXRX"}, + {"renderer:node-id": "XPDR-A1", + "renderer:src-tp": "XPDR1-CLIENT1", + "renderer:dest-tp": "XPDR1-NETWORK1"}]}} + headers = {'content-type': 'application/json'} + response = requests.request( + "POST", url, data=json.dumps(data), + headers=headers, auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + self.assertIn('Roadm-connection successfully created for nodes: ROADM-A1', res["output"]["result"]) + + def test_06_service_path_create_rdm_check(self): + url = ("{}/config/network-topology:network-topology/topology/topology-netconf/" + "node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/" + "interface/DEG1-TTP-TXRX-nmc" + .format(self.restconf_baseurl)) + headers = {'content-type': 'application/json'} + response = requests.request( + "GET", url, headers=headers, auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + self.assertDictContainsSubset({'name': 'DEG1-TTP-TXRX-nmc', 'administrative-state': 'inService', + 'supporting-circuit-pack-name': '1/0', + 'type': 'org-openroadm-interfaces:networkMediaChannelConnectionTerminationPoint', + 'supporting-port': 'L1'}, res['interface'][0]) + self.assertDictEqual( + {u'frequency': 195.79999999999998, u'width': 1531.11}, + res['interface'][0]['org-openroadm-network-media-channel-interfaces:nmc-ctp']) + + def test_07_service_path_create_rdm_check(self): + url = ("{}/config/network-topology:network-topology/topology/topology-netconf/" + "node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/" + "interface/DEG1-TTP-TXRX-mc" + .format(self.restconf_baseurl)) + headers = {'content-type': 'application/json'} + response = requests.request( + "GET", url, headers=headers, auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + self.assertDictContainsSubset({'name': 'DEG1-TTP-TXRX-mc', 'administrative-state': 'inService', + 'supporting-circuit-pack-name': '1/0', + 'type': 'org-openroadm-interfaces:mediaChannelTrailTerminationPoint', + 'supporting-port': 'L1'}, res['interface'][0]) + self.assertDictEqual( + {u'min-freq': 195.77499999999998, u'max-freq': 195.825}, + res['interface'][0]['org-openroadm-media-channel-interfaces:mc-ttp']) + + + def test_08_service_path_create_rdm_check(self): + url = ("{}/config/network-topology:network-topology/topology/topology-netconf/" + "node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/" + "interface/SRG1-PP3-TXRX-nmc" + .format(self.restconf_baseurl)) + headers = {'content-type': 'application/json'} + response = requests.request( + "GET", url, headers=headers, auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + self.assertDictContainsSubset({'name': 'SRG1-PP3-TXRX-nmc', 'administrative-state': 'inService', + 'supporting-circuit-pack-name': '3/0', + 'type': 'org-openroadm-interfaces:networkMediaChannelConnectionTerminationPoint', + 'supporting-port': 'C3'}, res['interface'][0]) + self.assertDictEqual( + {u'frequency': 195.79999999999998, u'width': 1531.11}, + res['interface'][0]['org-openroadm-network-media-channel-interfaces:nmc-ctp']) + + def test_09_service_path_create_rdm_check(self): + url = ("{}/config/network-topology:network-topology/topology/topology-netconf/" + "node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/" + "interface/SRG1-PP3-TXRX-mc" + .format(self.restconf_baseurl)) + headers = {'content-type': 'application/json'} + response = requests.request( + "GET", url, headers=headers, auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + self.assertDictContainsSubset({'name': 'SRG1-PP3-TXRX-mc', 'administrative-state': 'inService', + 'supporting-circuit-pack-name': '3/0', + 'type': 'org-openroadm-interfaces:mediaChannelTrailTerminationPoint', + 'supporting-port': 'C3'}, res['interface'][0]) + self.assertDictEqual( + {u'min-freq': 195.77499999999998, u'max-freq': 195.825}, + res['interface'][0]['org-openroadm-media-channel-interfaces:mc-ttp']) + + def test_10_service_path_create_rdm_check(self): + url = ("{}/config/network-topology:network-topology/topology/topology-netconf/" + "node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/" + "roadm-connections/SRG1-PP3-TXRX-DEG1-TTP-TXRX-7" + .format(self.restconf_baseurl)) + headers = {'content-type': 'application/json'} + response = requests.request( + "GET", url, headers=headers, auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + self.assertDictContainsSubset( + {'connection-name': 'SRG1-PP3-TXRX-DEG1-TTP-TXRX-7', + 'opticalControlMode': 'off'}, + res['roadm-connections'][0]) + self.assertDictEqual( + {'src-if': 'SRG1-PP3-TXRX-nmc'}, + res['roadm-connections'][0]['source']) + self.assertDictEqual( + {'dst-if': 'DEG1-TTP-TXRX-nmc'}, + res['roadm-connections'][0]['destination']) + + def test_11_service_path_create_xpdr_check(self): + url = ("{}/config/network-topology:network-topology/topology/topology-netconf/" + "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/" + "interface/XPDR1-NETWORK1-7" + .format(self.restconf_baseurl)) + headers = {'content-type': 'application/json'} + response = requests.request( + "GET", url, headers=headers, auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + self.assertDictContainsSubset( + {'name': 'XPDR1-NETWORK1-7', 'administrative-state': 'inService', + 'supporting-circuit-pack-name': '1/0/1-PLUG-NET', + 'type': 'org-openroadm-interfaces:opticalChannel', + 'supporting-port': '1'}, + res['interface'][0]) + self.assertDictEqual( + {u'rate': u'org-openroadm-common-types:R100G', + u'transmit-power':-5, + u'frequency': 195.79999999999998}, + res['interface'][0]['org-openroadm-optical-channel-interfaces:och']) + + def test_12_service_path_create_xpdr_check(self): + url = ("{}/config/network-topology:network-topology/topology/topology-netconf/" + "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/" + "interface/XPDR1-NETWORK1-OTU" + .format(self.restconf_baseurl)) + headers = {'content-type': 'application/json'} + response = requests.request( + "GET", url, headers=headers, auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + self.assertDictContainsSubset( + {'name': 'XPDR1-NETWORK1-OTU', 'administrative-state': 'inService', + 'supporting-circuit-pack-name': '1/0/1-PLUG-NET', + 'type': 'org-openroadm-interfaces:otnOtu', + 'supporting-port': '1', + 'supporting-interface': 'XPDR1-NETWORK1-7'}, + res['interface'][0]) + self.assertDictEqual( + {u'rate': u'org-openroadm-otn-common-types:OTU4', + u'fec': u'scfec'}, + res['interface'][0]['org-openroadm-otn-otu-interfaces:otu']) + + def test_13_service_path_create_xpdr_check(self): + url = ("{}/config/network-topology:network-topology/topology/topology-netconf/" + "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/" + "interface/XPDR1-NETWORK1-ODU" + .format(self.restconf_baseurl)) + headers = {'content-type': 'application/json'} + response = requests.request( + "GET", url, headers=headers, auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + self.assertDictContainsSubset( + {'name': 'XPDR1-NETWORK1-ODU', 'administrative-state': 'inService', + 'supporting-circuit-pack-name': '1/0/1-PLUG-NET', + 'type': 'org-openroadm-interfaces:otnOdu', + 'supporting-port': '1', + 'supporting-interface': 'XPDR1-NETWORK1-OTU'}, + res['interface'][0]) + self.assertDictContainsSubset( + {'rate': 'org-openroadm-otn-common-types:ODU4', + u'monitoring-mode': u'terminated'}, + res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']) + self.assertDictEqual({u'exp-payload-type': u'07', u'payload-type': u'07'}, + res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']['opu']) + + def test_14_service_path_create_xpdr_check(self): + url = ("{}/config/network-topology:network-topology/topology/topology-netconf/" + "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/" + "interface/XPDR1-CLIENT1-ETHERNET" + .format(self.restconf_baseurl)) + headers = {'content-type': 'application/json'} + response = requests.request( + "GET", url, headers=headers, auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + self.assertDictContainsSubset( + {'name': 'XPDR1-CLIENT1-ETHERNET', 'administrative-state': 'inService', + 'supporting-circuit-pack-name': '1/0/1-PLUG-CLIENT', + 'type': 'org-openroadm-interfaces:ethernetCsmacd', + 'supporting-port': 'C1'}, + res['interface'][0]) + self.assertDictEqual( + {'speed': 100000, + 'mtu': 9000, + 'auto-negotiation': 'enabled', + 'duplex': 'full', + 'fec': 'off'}, + res['interface'][0]['org-openroadm-ethernet-interfaces:ethernet']) + + def test_15_service_path_create_xpdr_check(self): + url = ("{}/config/network-topology:network-topology/topology/topology-netconf/" + "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/" + "circuit-packs/1%2F0%2F1-PLUG-NET" + .format(self.restconf_baseurl)) + headers = {'content-type': 'application/json'} + response = requests.request( + "GET", url, headers=headers, auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + self.assertIn('not-reserved-inuse', res['circuit-packs'][0]["equipment-state"]) + + def test_16_service_path_create_xpdr_check(self): + url = ("{}/config/network-topology:network-topology/topology/topology-netconf/" + "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/" + "circuit-packs/1%2F0%2F1-PLUG-CLIENT" + .format(self.restconf_baseurl)) + headers = {'content-type': 'application/json'} + response = requests.request( + "GET", url, headers=headers, auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + self.assertIn('not-reserved-inuse', res['circuit-packs'][0]["equipment-state"]) + + def test_17_service_path_delete(self): + url = "{}/operations/transportpce-device-renderer:service-path".format(self.restconf_baseurl) + data = {"renderer:input": { + "renderer:service-name": "service_test", + "renderer:wave-number": "7", + "renderer:operation": "delete", + "renderer:nodes": [ + {"renderer:node-id": "ROADM-A1", + "renderer:src-tp": "SRG1-PP3-TXRX", + "renderer:dest-tp": "DEG1-TTP-TXRX"}, + {"renderer:node-id": "XPDR-A1", + "renderer:src-tp": "XPDR1-CLIENT1", + "renderer:dest-tp": "XPDR1-NETWORK1"}]}} + headers = {'content-type': 'application/json'} + response = requests.request( + "POST", url, data=json.dumps(data), + headers=headers, auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.ok) + self.assertEqual(response.json(), { + 'output': {'result': 'Request processed', 'success': True}}) + + def test_18_service_path_delete_rdm_check(self): + url = ("{}/config/network-topology:network-topology/topology/topology-netconf/" + "node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/" + "interface/DEG1-TTP-TXRX-mc" + .format(self.restconf_baseurl)) + headers = {'content-type': 'application/json'} + response = requests.request( + "GET", url, headers=headers, auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.not_found) + res = response.json() + self.assertIn( + {"error-type":"application", "error-tag":"data-missing", + "error-message":"Request could not be completed because the relevant data model content does not exist"}, + res['errors']['error']) + + def test_19_service_path_delete_rdm_check(self): + url = ("{}/config/network-topology:network-topology/topology/topology-netconf/" + "node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/" + "interface/DEG1-TTP-TXRX-nmc" + .format(self.restconf_baseurl)) + headers = {'content-type': 'application/json'} + response = requests.request( + "GET", url, headers=headers, auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.not_found) + res = response.json() + self.assertIn( + {"error-type":"application", "error-tag":"data-missing", + "error-message":"Request could not be completed because the relevant data model content does not exist"}, + res['errors']['error']) + + def test_20_service_path_delete_rdm_check(self): + url = ("{}/config/network-topology:network-topology/topology/topology-netconf/" + "node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/" + "interface/SRG1-PP3-TXRX-mc" + .format(self.restconf_baseurl)) + headers = {'content-type': 'application/json'} + response = requests.request( + "GET", url, headers=headers, auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.not_found) + res = response.json() + self.assertIn( + {"error-type":"application", "error-tag":"data-missing", "error-message":"Request could not be completed because the relevant data model content does not exist"}, + res['errors']['error']) + + def test_21_service_path_delete_rdm_check(self): + url = ("{}/config/network-topology:network-topology/topology/topology-netconf/" + "node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/" + "interface/SRG1-PP3-TXRX-nmc" + .format(self.restconf_baseurl)) + headers = {'content-type': 'application/json'} + response = requests.request( + "GET", url, headers=headers, auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.not_found) + res = response.json() + self.assertIn( + {"error-type":"application", "error-tag":"data-missing", "error-message":"Request could not be completed because the relevant data model content does not exist"}, + res['errors']['error']) + + def test_22_service_path_delete_rdm_check(self): + url = ("{}/config/network-topology:network-topology/topology/topology-netconf/" + "node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/" + "roadm-connections/SRG1-PP3-TXRX-DEG1-TTP-TXRX-7" + .format(self.restconf_baseurl)) + headers = {'content-type': 'application/json'} + response = requests.request( + "GET", url, headers=headers, auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.not_found) + res = response.json() + self.assertIn( + {"error-type":"application", "error-tag":"data-missing", "error-message":"Request could not be completed because the relevant data model content does not exist"}, + res['errors']['error']) + + def test_23_service_path_delete_xpdr_check(self): + url = ("{}/config/network-topology:network-topology/topology/topology-netconf/" + "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/" + "interface/XPDR1-NETWORK1-7" + .format(self.restconf_baseurl)) + headers = {'content-type': 'application/json'} + response = requests.request( + "GET", url, headers=headers, auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.not_found) + res = response.json() + self.assertIn( + {"error-type":"application", "error-tag":"data-missing", "error-message":"Request could not be completed because the relevant data model content does not exist"}, + res['errors']['error']) + + def test_24_service_path_delete_xpdr_check(self): + url = ("{}/config/network-topology:network-topology/topology/topology-netconf/" + "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/" + "interface/XPDR1-NETWORK1-OTU" + .format(self.restconf_baseurl)) + headers = {'content-type': 'application/json'} + response = requests.request( + "GET", url, headers=headers, auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.not_found) + res = response.json() + self.assertIn( + {"error-type":"application", "error-tag":"data-missing", "error-message":"Request could not be completed because the relevant data model content does not exist"}, + res['errors']['error']) + + def test_25_service_path_delete_xpdr_check(self): + url = ("{}/config/network-topology:network-topology/topology/topology-netconf/" + "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/" + "interface/XPDR1-NETWORK1-ODU" + .format(self.restconf_baseurl)) + headers = {'content-type': 'application/json'} + response = requests.request( + "GET", url, headers=headers, auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.not_found) + res = response.json() + self.assertIn( + {"error-type":"application", "error-tag":"data-missing", "error-message":"Request could not be completed because the relevant data model content does not exist"}, + res['errors']['error']) + + def test_26_service_path_delete_xpdr_check(self): + url = ("{}/config/network-topology:network-topology/topology/topology-netconf/" + "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/" + "interface/XPDR1-CLIENT1-ETHERNET" + .format(self.restconf_baseurl)) + headers = {'content-type': 'application/json'} + response = requests.request( + "GET", url, headers=headers, auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.not_found) + res = response.json() + self.assertIn( + {"error-type":"application", "error-tag":"data-missing", "error-message":"Request could not be completed because the relevant data model content does not exist"}, + res['errors']['error']) + + def test_27_service_path_delete_xpdr_check(self): + url = ("{}/config/network-topology:network-topology/topology/topology-netconf/" + "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/" + "circuit-packs/1%2F0%2F1-PLUG-NET" + .format(self.restconf_baseurl)) + headers = {'content-type': 'application/json'} + response = requests.request( + "GET", url, headers=headers, auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + self.assertEqual('not-reserved-available', res["circuit-packs"][0]['equipment-state']) + + def test_28_service_path_delete_xpdr_check(self): + url = ("{}/config/network-topology:network-topology/topology/topology-netconf/" + "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/" + "circuit-packs/1%2F0%2F1-PLUG-CLIENT" + .format(self.restconf_baseurl)) + headers = {'content-type': 'application/json'} + response = requests.request( + "GET", url, headers=headers, auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + self.assertEqual('not-reserved-available', res["circuit-packs"][0]['equipment-state']) + + def test_29_rdm_device_disconnected(self): + url = ("{}/config/network-topology:" + "network-topology/topology/topology-netconf/node/ROADM-A1" + .format(self.restconf_baseurl)) + headers = {'content-type': 'application/json'} + response = requests.request( + "DELETE", url, headers=headers, + auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.ok) + time.sleep(20) + + def test_30_xpdr_device_disconnected(self): + url = ("{}/config/network-topology:" + "network-topology/topology/topology-netconf/node/XPDR-A1" + .format(self.restconf_baseurl)) + headers = {'content-type': 'application/json'} + response = requests.request( + "DELETE", url, headers=headers, + auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.ok) + time.sleep(20) + + +if __name__ == "__main__": + unittest.main(verbosity=2, failfast=True) diff --git a/tests/transportpce_tests/2.2.1/test_topology.py b/tests/transportpce_tests/2.2.1/test_topology.py new file mode 100644 index 000000000..06a7e37be --- /dev/null +++ b/tests/transportpce_tests/2.2.1/test_topology.py @@ -0,0 +1,1213 @@ +#!/usr/bin/env python + +############################################################################## +#Copyright (c) 2017 Orange, Inc. and others. All rights reserved. +# +# All rights reserved. This program and the accompanying materials +# are made available under the terms of the Apache License, Version 2.0 +# which accompanies this distribution, and is available at +# http://www.apache.org/licenses/LICENSE-2.0 +############################################################################## + +import json +import os +import psutil +import requests +import signal +import shutil +import subprocess +import time +import unittest +import logging + +class TransportPCEtesting(unittest.TestCase): + + honeynode_process1 = None + honeynode_process2 = None + honeynode_process3 = None + honeynode_process4 = None + odl_process = None + restconf_baseurl = "http://localhost:8181/restconf" + +#START_IGNORE_XTESTING + + @classmethod + def __start_honeynode1(cls): + executable = ("./honeynode/2.2.1/honeynode-distribution/target/honeynode-distribution-1.18.01-hc" + "/honeynode-distribution-1.18.01/honeycomb-tpce") + if os.path.isfile(executable): + with open('honeynode1.log', 'w') as outfile: + cls.honeynode_process1 = subprocess.Popen( + [executable, "17840", "sample_configs/openroadm/2.2.1/oper-XPDRA.xml"], + stdout=outfile) + + @classmethod + def __start_honeynode2(cls): + executable = ("./honeynode/2.2.1/honeynode-distribution/target/honeynode-distribution-1.18.01-hc" + "/honeynode-distribution-1.18.01/honeycomb-tpce") + if os.path.isfile(executable): + with open('honeynode2.log', 'w') as outfile: + cls.honeynode_process2 = subprocess.Popen( + [executable, "17841", "sample_configs/openroadm/2.2.1/oper-ROADMA.xml"], + stdout=outfile) + + @classmethod + def __start_honeynode3(cls): + executable = ("./honeynode/2.2.1/honeynode-distribution/target/honeynode-distribution-1.18.01-hc" + "/honeynode-distribution-1.18.01/honeycomb-tpce") + if os.path.isfile(executable): + with open('honeynode1.log', 'w') as outfile: + cls.honeynode_process3 = subprocess.Popen( + [executable, "17842", "sample_configs/openroadm/2.2.1/oper-ROADMB.xml"], + stdout=outfile) + + @classmethod + def __start_honeynode4(cls): + executable = ("./honeynode/2.2.1/honeynode-distribution/target/honeynode-distribution-1.18.01-hc" + "/honeynode-distribution-1.18.01/honeycomb-tpce") + if os.path.isfile(executable): + with open('honeynode2.log', 'w') as outfile: + cls.honeynode_process4 = subprocess.Popen( + [executable, "17843", "sample_configs/openroadm/2.2.1/oper-ROADMC.xml"], + stdout=outfile) + + @classmethod + def __start_odl(cls): + executable = "../karaf/target/assembly/bin/karaf" + with open('odl.log', 'w') as outfile: + cls.odl_process = subprocess.Popen( + ["bash", executable, "server"], stdout=outfile, + stdin=open(os.devnull)) + + @classmethod + def setUpClass(cls): + cls.__start_honeynode1() + time.sleep(20) + cls.__start_honeynode2() + time.sleep(20) + cls.__start_honeynode3() + time.sleep(20) + cls.__start_honeynode4() + time.sleep(20) + cls.__start_odl() + time.sleep(60) + + @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.honeynode_process1.pid).children(): + child.send_signal(signal.SIGINT) + child.wait() + cls.honeynode_process1.send_signal(signal.SIGINT) + cls.honeynode_process1.wait() + for child in psutil.Process(cls.honeynode_process2.pid).children(): + child.send_signal(signal.SIGINT) + child.wait() + cls.honeynode_process2.send_signal(signal.SIGINT) + cls.honeynode_process2.wait() + for child in psutil.Process(cls.honeynode_process3.pid).children(): + child.send_signal(signal.SIGINT) + child.wait() + cls.honeynode_process3.send_signal(signal.SIGINT) + cls.honeynode_process3.wait() + for child in psutil.Process(cls.honeynode_process4.pid).children(): + child.send_signal(signal.SIGINT) + child.wait() + cls.honeynode_process4.send_signal(signal.SIGINT) + cls.honeynode_process4.wait() + + + def setUp(self): + time.sleep(30) + +#END_IGNORE_XTESTING + + def test_01_connect_ROADM_A1(self): + #Config ROADMA + url = ("{}/config/network-topology:" + "network-topology/topology/topology-netconf/node/ROADM-A1" + .format(self.restconf_baseurl)) + data = {"node": [{ + "node-id": "ROADM-A1", + "netconf-node-topology:username": "admin", + "netconf-node-topology:password": "admin", + "netconf-node-topology:host": "127.0.0.1", + "netconf-node-topology:port": "17841", + "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) + time.sleep(20) + + 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')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + logging.info(res) + self.assertEqual(res['network'][0]['node'][0]['node-id'],'NodeA') + self.assertEqual(res['network'][0]['node'][0]['org-openroadm-clli-network:clli'],'NodeA') + + 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')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + self.assertEqual(res['network'][0]['node'][0]['node-id'],'ROADM-A1') + self.assertEqual(res['network'][0]['node'][0]['supporting-node'][0]['network-ref'],'clli-network') + self.assertEqual(res['network'][0]['node'][0]['supporting-node'][0]['node-ref'],'NodeA') + self.assertEqual(res['network'][0]['node'][0]['org-openroadm-network:node-type'],'ROADM') + self.assertEqual(res['network'][0]['node'][0]['org-openroadm-network:model'],'') + + 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')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + #Tests related to links + nbLink=len(res['network'][0]['ietf-network-topology:link']) + self.assertEqual(nbLink,6) + expressLink=['ROADM-A1-DEG2-DEG2-CTP-TXRXtoROADM-A1-DEG1-DEG1-CTP-TXRX', + 'ROADM-A1-DEG1-DEG1-CTP-TXRXtoROADM-A1-DEG2-DEG2-CTP-TXRX'] + addLink=['ROADM-A1-SRG1-SRG1-CP-TXRXtoROADM-A1-DEG2-DEG2-CTP-TXRX', + 'ROADM-A1-SRG1-SRG1-CP-TXRXtoROADM-A1-DEG1-DEG1-CTP-TXRX'] + dropLink=['ROADM-A1-DEG1-DEG1-CTP-TXRXtoROADM-A1-SRG1-SRG1-CP-TXRX', + 'ROADM-A1-DEG2-DEG2-CTP-TXRXtoROADM-A1-SRG1-SRG1-CP-TXRX'] + for i in range(0,nbLink): + linkId = res['network'][0]['ietf-network-topology:link'][i]['link-id'] + if (res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-network-topology:link-type']=='EXPRESS-LINK'): + find= linkId in expressLink + self.assertEqual(find, True) + expressLink.remove(linkId) + elif (res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-network-topology:link-type']=='ADD-LINK'): + find= linkId in addLink + self.assertEqual(find, True) + addLink.remove(linkId) + elif (res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-network-topology:link-type']=='DROP-LINK'): + find= linkId in dropLink + self.assertEqual(find, True) + dropLink.remove(linkId) + else: + self.assertFalse(True) + self.assertEqual(len(expressLink),0) + self.assertEqual(len(addLink),0) + self.assertEqual(len(dropLink),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')) + res = response.json() + #Tests related to nodes + self.assertEqual(response.status_code, requests.codes.ok) + with open('./transportpce_tests/log/response.log', 'a') as outfile1: + outfile1.write(str(len(res['network'][0]['node']))) + nbNode=len(res['network'][0]['node']) + self.assertEqual(nbNode,3) + listNode=['ROADM-A1-SRG1','ROADM-A1-DEG1','ROADM-A1-DEG2'] + for i in range(0,nbNode): + self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'ROADM-A1'}, + res['network'][0]['node'][i]['supporting-node']) + nodeType=res['network'][0]['node'][i]['org-openroadm-network-topology:node-type'] + nodeId=res['network'][0]['node'][i]['node-id'] + if(nodeId=='ROADM-A1-SRG1'): + #Test related to SRG1 + self.assertEqual(nodeType,'SRG') + self.assertEqual(len(res['network'][0]['node'][i]['ietf-network-topology:termination-point']),5) + self.assertIn({'tp-id': 'SRG1-CP-TXRX', 'org-openroadm-network-topology:tp-type': 'SRG-TXRX-CP'}, + res['network'][0]['node'][i]['ietf-network-topology:termination-point']) + self.assertIn({'tp-id': 'SRG1-PP1-TXRX', 'org-openroadm-network-topology:tp-type': 'SRG-TXRX-PP'}, + res['network'][0]['node'][i]['ietf-network-topology:termination-point']) + listNode.remove(nodeId) + elif(nodeId=='ROADM-A1-DEG1'): + #Test related to DEG1 + self.assertEqual(nodeType,'DEGREE') + self.assertIn({'tp-id': 'DEG1-TTP-TXRX', 'org-openroadm-network-topology:tp-type': 'DEGREE-TXRX-TTP'}, + res['network'][0]['node'][i]['ietf-network-topology:termination-point']) + self.assertIn({'tp-id': 'DEG1-CTP-TXRX', 'org-openroadm-network-topology:tp-type': 'DEGREE-TXRX-CTP'}, + res['network'][0]['node'][i]['ietf-network-topology:termination-point']) + listNode.remove(nodeId) + elif(nodeId=='ROADM-A1-DEG2'): + #Test related to DEG2 + self.assertEqual(nodeType,'DEGREE') + self.assertIn({'tp-id': 'DEG2-TTP-TXRX', 'org-openroadm-network-topology:tp-type': 'DEGREE-TXRX-TTP'}, + res['network'][0]['node'][i]['ietf-network-topology:termination-point']) + self.assertIn({'tp-id': 'DEG2-CTP-TXRX', 'org-openroadm-network-topology:tp-type': 'DEGREE-TXRX-CTP'}, + res['network'][0]['node'][i]['ietf-network-topology:termination-point']) + listNode.remove(nodeId) + else: + self.assertFalse(True) + self.assertEqual(len(listNode),0) + + def test_06_connect_XPDRA(self): + url = ("{}/config/network-topology:" + "network-topology/topology/topology-netconf/node/XPDR-A1" + .format(self.restconf_baseurl)) + data = {"node": [{ + "node-id": "XPDR-A1", + "netconf-node-topology:username": "admin", + "netconf-node-topology:password": "admin", + "netconf-node-topology:host": "127.0.0.1", + "netconf-node-topology:port": "17840", + "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) + time.sleep(30) + + def test_07_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')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + self.assertEqual(res['network'][0]['node'][0]['node-id'],'NodeA') + self.assertEqual(res['network'][0]['node'][0]['org-openroadm-clli-network:clli'],'NodeA') + + def test_08_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')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + nbNode=len(res['network'][0]['node']) + self.assertEqual(nbNode,2) + for i in range(0,nbNode): + self.assertEqual(res['network'][0]['node'][i]['supporting-node'][0]['network-ref'],'clli-network') + self.assertEqual(res['network'][0]['node'][i]['supporting-node'][0]['node-ref'],'NodeA') + nodeId=res['network'][0]['node'][i]['node-id'] + if(nodeId=='XPDR-A1'): + self.assertEqual(res['network'][0]['node'][i]['org-openroadm-network:node-type'],'XPONDER') + self.assertEqual(res['network'][0]['node'][i]['org-openroadm-network:model'],'') + elif(nodeId=='ROADM-A1'): + self.assertEqual(res['network'][0]['node'][i]['org-openroadm-network:node-type'],'ROADM') + self.assertEqual(res['network'][0]['node'][i]['org-openroadm-network:model'],'') + else: + self.assertFalse(True) + + def test_09_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')) + res = response.json() + #Tests related to nodes + self.assertEqual(response.status_code, requests.codes.ok) + with open('./transportpce_tests/log/response.log', 'a') as outfile1: + outfile1.write(str(len(res['network'][0]['node']))) + nbNode=len(res['network'][0]['node']) + self.assertEqual(nbNode,4) + listNode=['XPDR-A1-XPDR1','ROADM-A1-SRG1','ROADM-A1-DEG1','ROADM-A1-DEG2'] + for i in range(0,nbNode): + nodeType=res['network'][0]['node'][i]['org-openroadm-network-topology:node-type'] + nodeId=res['network'][0]['node'][i]['node-id'] + #Tests related to XPDRA nodes + if(nodeId=='XPDR-A1-XPDR1'): + self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'XPDR-A1'}, + res['network'][0]['node'][i]['supporting-node']) + self.assertEqual(nodeType,'XPONDER') + nbTps=len(res['network'][0]['node'][i]['ietf-network-topology:termination-point']) + self.assertTrue(nbTps >= 4) + client = 0 + network = 0 + for j in range(0,nbTps): + tpType=res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['org-openroadm-network-topology:tp-type'] + if (tpType=='XPONDER-CLIENT'): + client += 1 + elif (tpType=='XPONDER-NETWORK'): + network += 1 + self.assertTrue(client >= 2) + self.assertTrue(network >= 2) + listNode.remove(nodeId) + elif(nodeId=='ROADM-A1-SRG1'): + #Test related to SRG1 + self.assertEqual(nodeType,'SRG') + self.assertEqual(len(res['network'][0]['node'][i]['ietf-network-topology:termination-point']),5) + self.assertIn({'tp-id': 'SRG1-CP-TXRX', 'org-openroadm-network-topology:tp-type': 'SRG-TXRX-CP'}, + res['network'][0]['node'][i]['ietf-network-topology:termination-point']) + self.assertIn({'tp-id': 'SRG1-PP1-TXRX', 'org-openroadm-network-topology:tp-type': 'SRG-TXRX-PP'}, + res['network'][0]['node'][i]['ietf-network-topology:termination-point']) + self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'ROADM-A1'}, + res['network'][0]['node'][i]['supporting-node']) + listNode.remove(nodeId) + elif(nodeId=='ROADM-A1-DEG1'): + #Test related to DEG1 + self.assertEqual(nodeType,'DEGREE') + self.assertIn({'tp-id': 'DEG1-TTP-TXRX', 'org-openroadm-network-topology:tp-type': 'DEGREE-TXRX-TTP'}, + res['network'][0]['node'][i]['ietf-network-topology:termination-point']) + self.assertIn({'tp-id': 'DEG1-CTP-TXRX', 'org-openroadm-network-topology:tp-type': 'DEGREE-TXRX-CTP'}, + res['network'][0]['node'][i]['ietf-network-topology:termination-point']) + self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'ROADM-A1'}, + res['network'][0]['node'][i]['supporting-node']) + listNode.remove(nodeId) + elif(nodeId=='ROADM-A1-DEG2'): + #Test related to DEG2 + self.assertEqual(nodeType,'DEGREE') + self.assertIn({'tp-id': 'DEG2-TTP-TXRX', 'org-openroadm-network-topology:tp-type': 'DEGREE-TXRX-TTP'}, + res['network'][0]['node'][i]['ietf-network-topology:termination-point']) + self.assertIn({'tp-id': 'DEG2-CTP-TXRX', 'org-openroadm-network-topology:tp-type': 'DEGREE-TXRX-CTP'}, + res['network'][0]['node'][i]['ietf-network-topology:termination-point']) + self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'ROADM-A1'}, + res['network'][0]['node'][i]['supporting-node']) + listNode.remove(nodeId) + else: + self.assertFalse(True) + self.assertEqual(len(listNode),0) + + #Connect the tail XPDRA to ROADMA and vice versa + def test_10_connect_tail_xpdr_rdm(self): + #Connect the tail: XPDRA to ROADMA + url = ("{}/operations/transportpce-networkutils:init-xpdr-rdm-links" + .format(self.restconf_baseurl)) + data = {"networkutils:input": { + "networkutils:links-input": { + "networkutils:xpdr-node": "XPDR-A1", + "networkutils:xpdr-num": "1", + "networkutils:network-num": "1", + "networkutils:rdm-node": "ROADM-A1", + "networkutils:srg-num": "1", + "networkutils:termination-point-num": "SRG1-PP1-TXRX" + } + } + } + headers = {'content-type': 'application/json'} + response = requests.request( + "POST", url, data=json.dumps(data), headers=headers, + auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.ok) + time.sleep(10) + + def test_11_connect_tail_rdm_xpdr(self): + #Connect the tail: ROADMA to XPDRA + url = ("{}/operations/transportpce-networkutils:init-rdm-xpdr-links" + .format(self.restconf_baseurl)) + data = {"networkutils:input": { + "networkutils:links-input": { + "networkutils:xpdr-node": "XPDR-A1", + "networkutils:xpdr-num": "1", + "networkutils:network-num": "1", + "networkutils:rdm-node": "ROADM-A1", + "networkutils:srg-num": "1", + "networkutils:termination-point-num": "SRG1-PP1-TXRX" + } + } + } + headers = {'content-type': 'application/json'} + response = requests.request( + "POST", url, data=json.dumps(data), headers=headers, + auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.ok) + time.sleep(10) + + def test_12_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')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + #Tests related to links + nbLink=len(res['network'][0]['ietf-network-topology:link']) + self.assertEqual(nbLink,8) + expressLink=['ROADM-A1-DEG2-DEG2-CTP-TXRXtoROADM-A1-DEG1-DEG1-CTP-TXRX','ROADM-A1-DEG1-DEG1-CTP-TXRXtoROADM-A1-DEG2-DEG2-CTP-TXRX'] + addLink=['ROADM-A1-SRG1-SRG1-CP-TXRXtoROADM-A1-DEG2-DEG2-CTP-TXRX','ROADM-A1-SRG1-SRG1-CP-TXRXtoROADM-A1-DEG1-DEG1-CTP-TXRX',] + dropLink=['ROADM-A1-DEG1-DEG1-CTP-TXRXtoROADM-A1-SRG1-SRG1-CP-TXRX','ROADM-A1-DEG2-DEG2-CTP-TXRXtoROADM-A1-SRG1-SRG1-CP-TXRX'] + XPDR_IN=['ROADM-A1-SRG1-SRG1-PP1-TXRXtoXPDR-A1-XPDR1-XPDR1-NETWORK1'] + XPDR_OUT=['XPDR-A1-XPDR1-XPDR1-NETWORK1toROADM-A1-SRG1-SRG1-PP1-TXRX'] + for i in range(0,nbLink): + nodeType=res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-network-topology:link-type'] + linkId=res['network'][0]['ietf-network-topology:link'][i]['link-id'] + if(nodeType=='EXPRESS-LINK'): + find= linkId in expressLink + self.assertEqual(find, True) + expressLink.remove(linkId) + elif(nodeType=='ADD-LINK'): + find= linkId in addLink + self.assertEqual(find, True) + addLink.remove(linkId) + elif(nodeType=='DROP-LINK'): + find= linkId in dropLink + self.assertEqual(find, True) + dropLink.remove(linkId) + elif(nodeType=='XPONDER-INPUT'): + find= linkId in XPDR_IN + self.assertEqual(find, True) + XPDR_IN.remove(linkId) + elif(nodeType=='XPONDER-OUTPUT'): + find= linkId in XPDR_OUT + self.assertEqual(find, True) + XPDR_OUT.remove(linkId) + else: + self.assertFalse(True) + self.assertEqual(len(expressLink),0) + self.assertEqual(len(addLink),0) + self.assertEqual(len(dropLink),0) + self.assertEqual(len(XPDR_IN),0) + self.assertEqual(len(XPDR_OUT),0) + + def test_13_connect_ROADMC(self): + #Config ROADMC + url = ("{}/config/network-topology:" + "network-topology/topology/topology-netconf/node/ROADM-C1" + .format(self.restconf_baseurl)) + data = {"node": [{ + "node-id": "ROADM-C1", + "netconf-node-topology:username": "admin", + "netconf-node-topology:password": "admin", + "netconf-node-topology:host": "127.0.0.1", + "netconf-node-topology:port": "17843", + "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) + time.sleep(30) + + def test_14_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')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + nbNode=len(res['network'][0]['node']) + listNode=['NodeA','NodeC'] + for i in range(0,nbNode): + nodeId = res['network'][0]['node'][i]['node-id'] + find= nodeId in listNode + self.assertEqual(find, True) + if(nodeId=='NodeA'): + self.assertEqual(res['network'][0]['node'][i]['org-openroadm-clli-network:clli'],'NodeA') + else: + self.assertEqual(res['network'][0]['node'][i]['org-openroadm-clli-network:clli'],'NodeC') + listNode.remove(nodeId) + + self.assertEqual(len(listNode),0) + + def test_15_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')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + nbNode=len(res['network'][0]['node']) + self.assertEqual(nbNode,3) + listNode=['XPDR-A1','ROADM-A1','ROADM-C1'] + for i in range(0,nbNode): + self.assertEqual(res['network'][0]['node'][i]['supporting-node'][0]['network-ref'],'clli-network') + nodeId=res['network'][0]['node'][i]['node-id'] + if(nodeId=='XPDR-A1'): + self.assertEqual(res['network'][0]['node'][i]['supporting-node'][0]['node-ref'],'NodeA') + self.assertEqual(res['network'][0]['node'][i]['org-openroadm-network:node-type'],'XPONDER') + self.assertEqual(res['network'][0]['node'][i]['org-openroadm-network:model'],'') + listNode.remove(nodeId) + elif(nodeId=='ROADM-A1'): + self.assertEqual(res['network'][0]['node'][i]['supporting-node'][0]['node-ref'],'NodeA') + self.assertEqual(res['network'][0]['node'][i]['org-openroadm-network:node-type'],'ROADM') + self.assertEqual(res['network'][0]['node'][i]['org-openroadm-network:model'],'') + listNode.remove(nodeId) + elif(nodeId=='ROADM-C1'): + self.assertEqual(res['network'][0]['node'][i]['supporting-node'][0]['node-ref'],'NodeC') + self.assertEqual(res['network'][0]['node'][i]['org-openroadm-network:node-type'],'ROADM') + self.assertEqual(res['network'][0]['node'][i]['org-openroadm-network:model'],'') + listNode.remove(nodeId) + else: + self.assertFalse(True) + self.assertEqual(len(listNode),0) + + def test_16_getROADMLinkOpenRoadmTopology(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')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + #Tests related to links + nbLink=len(res['network'][0]['ietf-network-topology:link']) + self.assertEqual(nbLink,16) + expressLink=['ROADM-A1-DEG2-DEG2-CTP-TXRXtoROADM-A1-DEG1-DEG1-CTP-TXRX','ROADM-A1-DEG1-DEG1-CTP-TXRXtoROADM-A1-DEG2-DEG2-CTP-TXRX', + 'ROADM-C1-DEG2-DEG2-CTP-TXRXtoROADM-C1-DEG1-DEG1-CTP-TXRX','ROADM-C1-DEG1-DEG1-CTP-TXRXtoROADM-C1-DEG2-DEG2-CTP-TXRX'] + addLink=['ROADM-A1-SRG1-SRG1-CP-TXRXtoROADM-A1-DEG2-DEG2-CTP-TXRX','ROADM-A1-SRG1-SRG1-CP-TXRXtoROADM-A1-DEG1-DEG1-CTP-TXRX', + 'ROADM-C1-SRG1-SRG1-CP-TXRXtoROADM-C1-DEG2-DEG2-CTP-TXRX','ROADM-C1-SRG1-SRG1-CP-TXRXtoROADM-C1-DEG1-DEG1-CTP-TXRX'] + dropLink=['ROADM-A1-DEG1-DEG1-CTP-TXRXtoROADM-A1-SRG1-SRG1-CP-TXRX','ROADM-A1-DEG2-DEG2-CTP-TXRXtoROADM-A1-SRG1-SRG1-CP-TXRX', + 'ROADM-C1-DEG1-DEG1-CTP-TXRXtoROADM-C1-SRG1-SRG1-CP-TXRX','ROADM-C1-DEG2-DEG2-CTP-TXRXtoROADM-C1-SRG1-SRG1-CP-TXRX'] + R2RLink=['ROADM-A1-DEG2-DEG2-TTP-TXRXtoROADM-C1-DEG1-DEG1-TTP-TXRX','ROADM-C1-DEG1-DEG1-TTP-TXRXtoROADM-A1-DEG2-DEG2-TTP-TXRX'] + XPDR_IN=['ROADM-A1-SRG1-SRG1-PP1-TXRXtoXPDR-A1-XPDR1-XPDR1-NETWORK1'] + XPDR_OUT=['XPDR-A1-XPDR1-XPDR1-NETWORK1toROADM-A1-SRG1-SRG1-PP1-TXRX'] + for i in range(0,nbLink): + nodeType=res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-network-topology:link-type'] + linkId=res['network'][0]['ietf-network-topology:link'][i]['link-id'] + if(nodeType=='EXPRESS-LINK'): + find= linkId in expressLink + self.assertEqual(find, True) + expressLink.remove(linkId) + elif(nodeType=='ADD-LINK'): + find= linkId in addLink + self.assertEqual(find, True) + addLink.remove(linkId) + elif(nodeType=='DROP-LINK'): + find= linkId in dropLink + self.assertEqual(find, True) + dropLink.remove(linkId) + elif(nodeType=='ROADM-TO-ROADM'): + find= linkId in R2RLink + self.assertEqual(find, True) + R2RLink.remove(linkId) + elif(nodeType=='XPONDER-INPUT'): + find= linkId in XPDR_IN + self.assertEqual(find, True) + XPDR_IN.remove(linkId) + elif(nodeType=='XPONDER-OUTPUT'): + find= linkId in XPDR_OUT + self.assertEqual(find, True) + XPDR_OUT.remove(linkId) + else: + self.assertFalse(True) + self.assertEqual(len(expressLink),0) + self.assertEqual(len(addLink),0) + self.assertEqual(len(dropLink),0) + self.assertEqual(len(R2RLink),0) + self.assertEqual(len(XPDR_IN),0) + self.assertEqual(len(XPDR_OUT),0) + + def test_17_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')) + res = response.json() + #Tests related to nodes + self.assertEqual(response.status_code, requests.codes.ok) + nbNode=len(res['network'][0]['node']) + self.assertEqual(nbNode,7) + listNode=['XPDR-A1-XPDR1', + 'ROADM-A1-SRG1','ROADM-A1-DEG1','ROADM-A1-DEG2', + 'ROADM-C1-SRG1','ROADM-C1-DEG1', 'ROADM-C1-DEG2'] + #************************Tests related to XPDRA nodes + for i in range(0,nbNode): + nodeType=res['network'][0]['node'][i]['org-openroadm-network-topology:node-type'] + nodeId=res['network'][0]['node'][i]['node-id'] + if(nodeId=='XPDR-A1-XPDR1'): + self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'XPDR-A1'}, + res['network'][0]['node'][i]['supporting-node']) + self.assertEqual(nodeType,'XPONDER') + nbTps=len(res['network'][0]['node'][i]['ietf-network-topology:termination-point']) + self.assertTrue(nbTps >= 4) + client = 0 + network = 0 + for j in range(0,nbTps): + tpType=res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['org-openroadm-network-topology:tp-type'] + if (tpType=='XPONDER-CLIENT'): + client += 1 + elif (tpType=='XPONDER-NETWORK'): + network += 1 + self.assertTrue(client > 2) + self.assertTrue(network > 2) + listNode.remove(nodeId) + elif(nodeId=='ROADM-A1-SRG1'): + #Test related to SRG1 + self.assertEqual(len(res['network'][0]['node'][i]['ietf-network-topology:termination-point']),5) + self.assertIn({'tp-id': 'SRG1-CP-TXRX', 'org-openroadm-network-topology:tp-type': 'SRG-TXRX-CP'}, + res['network'][0]['node'][i]['ietf-network-topology:termination-point']) + self.assertIn({'tp-id': 'SRG1-PP1-TXRX', 'org-openroadm-network-topology:tp-type': 'SRG-TXRX-PP'}, + res['network'][0]['node'][i]['ietf-network-topology:termination-point']) + self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'ROADM-A1'}, + res['network'][0]['node'][i]['supporting-node']) + self.assertEqual(res['network'][0]['node'][i]['org-openroadm-network-topology:node-type'],'SRG') + listNode.remove(nodeId) + elif(nodeId=='ROADM-A1-DEG1'): + #Test related to DEG1 + self.assertIn({'tp-id': 'DEG1-TTP-TXRX', 'org-openroadm-network-topology:tp-type': 'DEGREE-TXRX-TTP'}, + res['network'][0]['node'][i]['ietf-network-topology:termination-point']) + self.assertIn({'tp-id': 'DEG1-CTP-TXRX', 'org-openroadm-network-topology:tp-type': 'DEGREE-TXRX-CTP'}, + res['network'][0]['node'][i]['ietf-network-topology:termination-point']) + self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'ROADM-A1'}, + res['network'][0]['node'][i]['supporting-node']) + self.assertEqual(res['network'][0]['node'][i]['org-openroadm-network-topology:node-type'],'DEGREE') + listNode.remove(nodeId) + elif(nodeId=='ROADM-A1-DEG2'): + #Test related to DEG2 + self.assertIn({'tp-id': 'DEG2-TTP-TXRX', 'org-openroadm-network-topology:tp-type': 'DEGREE-TXRX-TTP'}, + res['network'][0]['node'][i]['ietf-network-topology:termination-point']) + self.assertIn({'tp-id': 'DEG2-CTP-TXRX', 'org-openroadm-network-topology:tp-type': 'DEGREE-TXRX-CTP'}, + res['network'][0]['node'][i]['ietf-network-topology:termination-point']) + self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'ROADM-A1'}, + res['network'][0]['node'][i]['supporting-node']) + self.assertEqual(res['network'][0]['node'][i]['org-openroadm-network-topology:node-type'],'DEGREE') + listNode.remove(nodeId) + elif(nodeId=='ROADM-C1-SRG1'): + #Test related to SRG1 + self.assertEqual(len(res['network'][0]['node'][i]['ietf-network-topology:termination-point']),5) + self.assertIn({'tp-id': 'SRG1-CP-TXRX', 'org-openroadm-network-topology:tp-type': 'SRG-TXRX-CP'}, + res['network'][0]['node'][i]['ietf-network-topology:termination-point']) + self.assertIn({'tp-id': 'SRG1-PP1-TXRX', 'org-openroadm-network-topology:tp-type': 'SRG-TXRX-PP'}, + res['network'][0]['node'][i]['ietf-network-topology:termination-point']) + self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'ROADM-C1'}, + res['network'][0]['node'][i]['supporting-node']) + self.assertEqual(res['network'][0]['node'][i]['org-openroadm-network-topology:node-type'],'SRG') + listNode.remove(nodeId) + elif(nodeId=='ROADM-C1-DEG1'): + #Test related to DEG1 + self.assertIn({'tp-id': 'DEG1-TTP-TXRX', 'org-openroadm-network-topology:tp-type': 'DEGREE-TXRX-TTP'}, + res['network'][0]['node'][i]['ietf-network-topology:termination-point']) + self.assertIn({'tp-id': 'DEG1-CTP-TXRX', 'org-openroadm-network-topology:tp-type': 'DEGREE-TXRX-CTP'}, + res['network'][0]['node'][i]['ietf-network-topology:termination-point']) + self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'ROADM-C1'}, + res['network'][0]['node'][i]['supporting-node']) + self.assertEqual(res['network'][0]['node'][i]['org-openroadm-network-topology:node-type'],'DEGREE') + listNode.remove(nodeId) + elif(nodeId=='ROADM-C1-DEG2'): + #Test related to DEG1 + self.assertIn({'tp-id': 'DEG2-TTP-TXRX', 'org-openroadm-network-topology:tp-type': 'DEGREE-TXRX-TTP'}, + res['network'][0]['node'][i]['ietf-network-topology:termination-point']) + self.assertIn({'tp-id': 'DEG2-CTP-TXRX', 'org-openroadm-network-topology:tp-type': 'DEGREE-TXRX-CTP'}, + res['network'][0]['node'][i]['ietf-network-topology:termination-point']) + self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'ROADM-C1'}, + res['network'][0]['node'][i]['supporting-node']) + self.assertEqual(res['network'][0]['node'][i]['org-openroadm-network-topology:node-type'],'DEGREE') + listNode.remove(nodeId) + else: + self.assertFalse(True) + self.assertEqual(len(listNode),0) + + def test_18_connect_ROADMB(self): + url = ("{}/config/network-topology:" + "network-topology/topology/topology-netconf/node/ROADM-B1" + .format(self.restconf_baseurl)) + data = {"node": [{ + "node-id": "ROADM-B1", + "netconf-node-topology:username": "admin", + "netconf-node-topology:password": "admin", + "netconf-node-topology:host": "127.0.0.1", + "netconf-node-topology:port": "17842", + "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) + time.sleep(30) + + def test_19_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')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + nbNode=len(res['network'][0]['node']) + listNode=['NodeA','NodeB','NodeC'] + for i in range(0,nbNode): + nodeId = res['network'][0]['node'][i]['node-id'] + find= nodeId in listNode + self.assertEqual(find, True) + if(nodeId=='NodeA'): + self.assertEqual(res['network'][0]['node'][i]['org-openroadm-clli-network:clli'],'NodeA') + elif(nodeId=='NodeB'): + self.assertEqual(res['network'][0]['node'][i]['org-openroadm-clli-network:clli'],'NodeB') + else: + self.assertEqual(res['network'][0]['node'][i]['org-openroadm-clli-network:clli'],'NodeC') + listNode.remove(nodeId) + + self.assertEqual(len(listNode),0) + + def test_20_verifyDegree(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')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + #Tests related to links + nbLink=len(res['network'][0]['ietf-network-topology:link']) + listR2RLink=['ROADM-A1-DEG2-DEG2-TTP-TXRXtoROADM-C1-DEG1-DEG1-TTP-TXRX','ROADM-C1-DEG1-DEG1-TTP-TXRXtoROADM-A1-DEG2-DEG2-TTP-TXRX', + 'ROADM-A1-DEG1-DEG1-TTP-TXRXtoROADM-B1-DEG1-DEG1-TTP-TXRX','ROADM-C1-DEG2-DEG2-TTP-TXRXtoROADM-B1-DEG2-DEG2-TTP-TXRX', + 'ROADM-B1-DEG1-DEG1-TTP-TXRXtoROADM-A1-DEG1-DEG1-TTP-TXRX','ROADM-B1-DEG2-DEG2-TTP-TXRXtoROADM-C1-DEG2-DEG2-TTP-TXRX'] + for i in range(0,nbLink): + if res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-network-topology:link-type'] == 'ROADM-TO-ROADM': + link_id = res['network'][0]['ietf-network-topology:link'][i]['link-id'] + find= link_id in listR2RLink + self.assertEqual(find, True) + listR2RLink.remove(link_id) + self.assertEqual(len(listR2RLink),0) + + def test_21_verifyOppositeLinkTopology(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')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + #Write the response in the log + with open('./transportpce_tests/log/response.log', 'a') as outfile1: + outfile1.write(str(res)) + #Tests related to links + nbLink=len(res['network'][0]['ietf-network-topology:link']) + self.assertEqual(nbLink,22) + for i in range(0,nbLink): + link_id=res['network'][0]['ietf-network-topology:link'][i]['link-id'] + link_type=res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-network-topology:link-type'] + link_src=res['network'][0]['ietf-network-topology:link'][i]['source']['source-node'] + link_dest=res['network'][0]['ietf-network-topology:link'][i]['destination']['dest-node'] + oppLink_id=res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-common-network:opposite-link'] + #Find the opposite link + url_oppLink="{}/config/ietf-network:networks/network/openroadm-topology/ietf-network-topology:link/"+oppLink_id + url = (url_oppLink.format(self.restconf_baseurl)) + headers = {'content-type': 'application/json'} + response_oppLink = requests.request("GET", url, headers=headers, auth=('admin', 'admin')) + self.assertEqual(response_oppLink.status_code, requests.codes.ok) + res_oppLink = response_oppLink.json() + self.assertEqual(res_oppLink['ietf-network-topology:link'][0]['org-openroadm-common-network:opposite-link'],link_id) + self.assertEqual(res_oppLink['ietf-network-topology:link'][0]['source']['source-node'],link_dest) + self.assertEqual(res_oppLink['ietf-network-topology:link'][0]['destination']['dest-node'],link_src) + oppLink_type=res_oppLink['ietf-network-topology:link'][0]['org-openroadm-network-topology:link-type'] + if link_type=='ADD-LINK': + self.assertEqual(oppLink_type, 'DROP-LINK') + elif link_type=='DROP-LINK': + self.assertEqual(oppLink_type, 'ADD-LINK') + elif link_type=='EXPRESS-LINK': + self.assertEqual(oppLink_type, 'EXPRESS-LINK') + elif link_type=='ROADM-TO-ROADM': + self.assertEqual(oppLink_type, 'ROADM-TO-ROADM') + elif link_type=='XPONDER-INPUT': + self.assertEqual(oppLink_type, 'XPONDER-OUTPUT') + elif link_type=='XPONDER-OUTPUT': + self.assertEqual(oppLink_type, 'XPONDER-INPUT') + time.sleep(5) + + def test_22_disconnect_ROADMB(self): + #Delete in the topology-netconf + url = ("{}/config/network-topology:" + "network-topology/topology/topology-netconf/node/ROADM-B1" + .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) + #Delete in the clli-network + url = ("{}/config/ietf-network:networks/network/clli-network/node/NodeB" + .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) + + def test_23_disconnect_ROADMC(self): + #Delete in the topology-netconf + url = ("{}/config/network-topology:" + "network-topology/topology/topology-netconf/node/ROADM-C1" + .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) + #Delete in the clli-network + url = ("{}/config/ietf-network:networks/network/clli-network/node/NodeC" + .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) + +# def test_24_check_roadm2roadm_links_deletion(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')) +# self.assertEqual(response.status_code, requests.codes.ok) +# res = response.json() +# #Write the response in the log +# with open('./transportpce_tests/log/response.log', 'a') as outfile1: +# outfile1.write(str(res)) +# #Tests related to links +# nbLink=len(res['network'][0]['ietf-network-topology:link']) +# self.assertEqual(nbLink,8) +# expressLink=['ROADMA-DEG2-DEG2-CTP-TXRXtoROADMA-DEG1-DEG1-CTP-TXRX','ROADMA-DEG1-DEG1-CTP-TXRXtoROADMA-DEG2-DEG2-CTP-TXRX'] +# addLink=['ROADMA-SRG1-SRG1-CP-TXRXtoROADMA-DEG2-DEG2-CTP-TXRX','ROADMA-SRG1-SRG1-CP-TXRXtoROADMA-DEG1-DEG1-CTP-TXRX',] +# dropLink=['ROADMA-DEG1-DEG1-CTP-TXRXtoROADMA-SRG1-SRG1-CP-TXRX','ROADMA-DEG2-DEG2-CTP-TXRXtoROADMA-SRG1-SRG1-CP-TXRX'] +# XPDR_IN=['ROADMA-SRG1-SRG1-PP1-TXRXtoXPDRA-XPDR1-XPDR1-NETWORK1'] +# XPDR_OUT=['XPDRA-XPDR1-XPDR1-NETWORK1toROADMA-SRG1-SRG1-PP1-TXRX'] +# for i in range(0,nbLink): +# nodeType=res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-network-topology:link-type'] +# linkId=res['network'][0]['ietf-network-topology:link'][i]['link-id'] +# if(nodeType=='EXPRESS-LINK'): +# find= linkId in expressLink +# self.assertEqual(find, True) +# expressLink.remove(linkId) +# elif(nodeType=='ADD-LINK'): +# find= linkId in addLink +# self.assertEqual(find, True) +# addLink.remove(linkId) +# elif(nodeType=='DROP-LINK'): +# find= linkId in dropLink +# self.assertEqual(find, True) +# dropLink.remove(linkId) +# elif(nodeType=='XPONDER-INPUT'): +# find= linkId in XPDR_IN +# self.assertEqual(find, True) +# XPDR_IN.remove(linkId) +# elif(nodeType=='XPONDER-OUTPUT'): +# find= linkId in XPDR_OUT +# self.assertEqual(find, True) +# XPDR_OUT.remove(linkId) +# else: +# self.assertFalse(True) +# self.assertEqual(len(expressLink),0) +# self.assertEqual(len(addLink),0) +# self.assertEqual(len(dropLink),0) +# self.assertEqual(len(XPDR_IN),0) +# self.assertEqual(len(XPDR_OUT),0) +# +# for i in range(0,nbLink): +# self.assertNotEqual(res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-network-topology:link-type'],'ROADM-TO-ROADM') +# self.assertNotEqual(res['network'][0]['ietf-network-topology:link'][i]['link-id'],'ROADMC-SRG1-SRG1-CP-TXRXtoROADMC-DEG1-DEG1-CTP-TXRX') +# self.assertNotEqual(res['network'][0]['ietf-network-topology:link'][i]['link-id'],'ROADMC-DEG1-DEG1-CTP-TXRXtoROADMC-SRG1-SRG1-CP-TXRX') +# self.assertNotEqual(res['network'][0]['ietf-network-topology:link'][i]['link-id'],'ROADMC-SRG1-SRG1-CP-TXRXtoROADMC-DEG2-DEG1-CTP-TXRX') +# self.assertNotEqual(res['network'][0]['ietf-network-topology:link'][i]['link-id'],'ROADMC-DEG1-DEG2-CTP-TXRXtoROADMC-SRG1-SRG1-CP-TXRX') +# self.assertNotEqual(res['network'][0]['ietf-network-topology:link'][i]['link-id'],'ROADMC-DEG1-DEG1-CTP-TXRXtoROADMC-DEG2-DEG2-CTP-TXRX') +# self.assertNotEqual(res['network'][0]['ietf-network-topology:link'][i]['link-id'],'ROADMC-DEG2-DEG2-CTP-TXRXtoROADMC-DEG1-DEG1-CTP-TXRX') + + def test_25_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')) + res = response.json() + #Tests related to nodes + self.assertEqual(response.status_code, requests.codes.ok) + with open('./transportpce_tests/log/response.log', 'a') as outfile1: + outfile1.write(str(len(res['network'][0]['node']))) + nbNode=len(res['network'][0]['node']) + self.assertEqual(nbNode,4) + listNode=['XPDR-A1-XPDR1','ROADM-A1-SRG1','ROADM-A1-DEG1','ROADM-A1-DEG2'] + for i in range(0,nbNode): + nodeType=res['network'][0]['node'][i]['org-openroadm-network-topology:node-type'] + nodeId=res['network'][0]['node'][i]['node-id'] + #Tests related to XPDRA nodes + if(nodeId=='XPDR-A1-XPDR1'): + nbTp = len(res['network'][0]['node'][i]['ietf-network-topology:termination-point']) + for j in range(0, nbTp): + tpid = res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['tp-id'] + if (tpid == 'XPDR1-CLIENT1'): + self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j] + ['org-openroadm-network-topology:tp-type'], 'XPONDER-CLIENT') + self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j] + ['org-openroadm-network-topology:xpdr-client-attributes']['tail-equipment-id'], + 'XPDR1-NETWORK1') + self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'XPDR-A1'}, + res['network'][0]['node'][i]['supporting-node']) + listNode.remove(nodeId) + elif(nodeId=='ROADM-A1-SRG1'): + #Test related to SRG1 + self.assertEqual(nodeType,'SRG') + self.assertEqual(len(res['network'][0]['node'][i]['ietf-network-topology:termination-point']),5) + self.assertIn({'tp-id': 'SRG1-CP-TXRX', 'org-openroadm-network-topology:tp-type': 'SRG-TXRX-CP'}, + res['network'][0]['node'][i]['ietf-network-topology:termination-point']) + self.assertIn({'tp-id': 'SRG1-PP1-TXRX', 'org-openroadm-network-topology:tp-type': 'SRG-TXRX-PP'}, + res['network'][0]['node'][i]['ietf-network-topology:termination-point']) + self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'ROADM-A1'}, + res['network'][0]['node'][i]['supporting-node']) + listNode.remove(nodeId) + elif(nodeId=='ROADM-A1-DEG1'): + #Test related to DEG1 + self.assertEqual(nodeType,'DEGREE') + self.assertIn({'tp-id': 'DEG1-TTP-TXRX', 'org-openroadm-network-topology:tp-type': 'DEGREE-TXRX-TTP'}, + res['network'][0]['node'][i]['ietf-network-topology:termination-point']) + self.assertIn({'tp-id': 'DEG1-CTP-TXRX', 'org-openroadm-network-topology:tp-type': 'DEGREE-TXRX-CTP'}, + res['network'][0]['node'][i]['ietf-network-topology:termination-point']) + self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'ROADM-A1'}, + res['network'][0]['node'][i]['supporting-node']) + listNode.remove(nodeId) + elif(nodeId=='ROADM-A1-DEG2'): + #Test related to DEG2 + self.assertEqual(nodeType,'DEGREE') + self.assertIn({'tp-id': 'DEG2-TTP-TXRX', 'org-openroadm-network-topology:tp-type': 'DEGREE-TXRX-TTP'}, + res['network'][0]['node'][i]['ietf-network-topology:termination-point']) + self.assertIn({'tp-id': 'DEG2-CTP-TXRX', 'org-openroadm-network-topology:tp-type': 'DEGREE-TXRX-CTP'}, + res['network'][0]['node'][i]['ietf-network-topology:termination-point']) + self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'ROADM-A1'}, + res['network'][0]['node'][i]['supporting-node']) + listNode.remove(nodeId) + else: + self.assertFalse(True) + self.assertEqual(len(listNode),0) + #Test related to SRG1 of ROADMC + for i in range(0,nbNode): + self.assertNotEqual(res['network'][0]['node'][i]['node-id'],'ROADM-C1-SRG1') + self.assertNotEqual(res['network'][0]['node'][i]['node-id'],'ROADM-C1-DEG1') + self.assertNotEqual(res['network'][0]['node'][i]['node-id'],'ROADM-C1-DEG2') + + def test_26_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')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + nbNode=len(res['network'][0]['node']) + self.assertEqual(nbNode,2) + for i in range(0,nbNode-1): + self.assertNotEqual(res['network'][0]['node'][i]['node-id'],'ROADM-C1') + + def test_27_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')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + nbNode=len(res['network'][0]['node']) + self.assertEqual(nbNode,1) + for i in range(0,nbNode-1): + self.assertNotEqual(res['network'][0]['node'][1]['org-openroadm-clli-network:clli'],'NodeC') + + def test_28_disconnect_XPDRA(self): + url = ("{}/config/network-topology:" + "network-topology/topology/topology-netconf/node/XPDR-A1" + .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) + + def test_29_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')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + nbNode=len(res['network'][0]['node']) + self.assertEqual(nbNode,1) + self.assertEqual(res['network'][0]['node'][0]['org-openroadm-clli-network:clli'],'NodeA') + + def test_30_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')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + nbNode=len(res['network'][0]['node']) + self.assertEqual(nbNode,1) + for i in range(0,nbNode): + self.assertNotEqual(res['network'][0]['node'][i]['node-id'],'XPDR-A1') + + def test_31_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')) + res = response.json() + #Tests related to nodes + self.assertEqual(response.status_code, requests.codes.ok) + nbNode=len(res['network'][0]['node']) + self.assertEqual(nbNode,3) + listNode=['ROADM-A1-SRG1','ROADM-A1-DEG1','ROADM-A1-DEG2'] + for i in range(0,nbNode): + self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'ROADM-A1'}, + res['network'][0]['node'][i]['supporting-node']) + nodeType=res['network'][0]['node'][i]['org-openroadm-network-topology:node-type'] + nodeId=res['network'][0]['node'][i]['node-id'] + if(nodeId=='ROADM-A1-SRG1'): + #Test related to SRG1 + self.assertEqual(nodeType,'SRG') + self.assertEqual(len(res['network'][0]['node'][i]['ietf-network-topology:termination-point']),5) + self.assertIn({'tp-id': 'SRG1-CP-TXRX', 'org-openroadm-network-topology:tp-type': 'SRG-TXRX-CP'}, + res['network'][0]['node'][i]['ietf-network-topology:termination-point']) + self.assertIn({'tp-id': 'SRG1-PP1-TXRX', 'org-openroadm-network-topology:tp-type': 'SRG-TXRX-PP'}, + res['network'][0]['node'][i]['ietf-network-topology:termination-point']) + listNode.remove(nodeId) + elif(nodeId=='ROADM-A1-DEG1'): + #Test related to DEG1 + self.assertEqual(nodeType,'DEGREE') + self.assertIn({'tp-id': 'DEG1-TTP-TXRX', 'org-openroadm-network-topology:tp-type': 'DEGREE-TXRX-TTP'}, + res['network'][0]['node'][i]['ietf-network-topology:termination-point']) + self.assertIn({'tp-id': 'DEG1-CTP-TXRX', 'org-openroadm-network-topology:tp-type': 'DEGREE-TXRX-CTP'}, + res['network'][0]['node'][i]['ietf-network-topology:termination-point']) + listNode.remove(nodeId) + elif(nodeId=='ROADM-A1-DEG2'): + #Test related to DEG2 + self.assertEqual(nodeType,'DEGREE') + self.assertIn({'tp-id': 'DEG2-TTP-TXRX', 'org-openroadm-network-topology:tp-type': 'DEGREE-TXRX-TTP'}, + res['network'][0]['node'][i]['ietf-network-topology:termination-point']) + self.assertIn({'tp-id': 'DEG2-CTP-TXRX', 'org-openroadm-network-topology:tp-type': 'DEGREE-TXRX-CTP'}, + res['network'][0]['node'][i]['ietf-network-topology:termination-point']) + listNode.remove(nodeId) + else: + self.assertFalse(True) + self.assertEqual(len(listNode),0) + + def test_32_disconnect_ROADM_XPDRA_link(self): + #Link-1 + url = ("{}/config/ietf-network:networks/network/openroadm-topology/ietf-network-topology:" + "link/XPDR-A1-XPDR1-XPDR1-NETWORK1toROADM-A1-SRG1-SRG1-PP1-TXRX" + .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) + #Link-2 + url = ("{}/config/ietf-network:networks/network/openroadm-topology/ietf-network-topology:" + "link/ROADM-A1-SRG1-SRG1-PP1-TXRXtoXPDR-A1-XPDR1-XPDR1-NETWORK1" + .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) + + def test_33_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')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + nbLink=len(res['network'][0]['ietf-network-topology:link']) + self.assertEqual(nbLink, 12) + expressLink=['ROADM-A1-DEG2-DEG2-CTP-TXRXtoROADM-A1-DEG1-DEG1-CTP-TXRX','ROADM-A1-DEG1-DEG1-CTP-TXRXtoROADM-A1-DEG2-DEG2-CTP-TXRX'] + addLink=['ROADM-A1-SRG1-SRG1-CP-TXRXtoROADM-A1-DEG2-DEG2-CTP-TXRX','ROADM-A1-SRG1-SRG1-CP-TXRXtoROADM-A1-DEG1-DEG1-CTP-TXRX'] + dropLink=['ROADM-A1-DEG1-DEG1-CTP-TXRXtoROADM-A1-SRG1-SRG1-CP-TXRX','ROADM-A1-DEG2-DEG2-CTP-TXRXtoROADM-A1-SRG1-SRG1-CP-TXRX'] + roadmtoroadmLink = 0 + for i in range(0,nbLink): + if (res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-network-topology:link-type']=='EXPRESS-LINK'): + link_id = res['network'][0]['ietf-network-topology:link'][i]['link-id'] + find= link_id in expressLink + self.assertEqual(find, True) + expressLink.remove(link_id) + elif (res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-network-topology:link-type']=='ADD-LINK'): + link_id = res['network'][0]['ietf-network-topology:link'][i]['link-id'] + find= link_id in addLink + self.assertEqual(find, True) + addLink.remove(link_id) + elif (res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-network-topology:link-type']=='DROP-LINK'): + link_id = res['network'][0]['ietf-network-topology:link'][i]['link-id'] + find= link_id in dropLink + self.assertEqual(find, True) + dropLink.remove(link_id) + else: + roadmtoroadmLink += 1 + self.assertEqual(len(expressLink),0) + self.assertEqual(len(addLink),0) + self.assertEqual(len(dropLink),0) + self.assertEqual(roadmtoroadmLink, 6) + for i in range(0,nbLink): + self.assertNotEqual(res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-network-topology:link-type'],'XPONDER-OUTPUT') + self.assertNotEqual(res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-network-topology:link-type'],'XPONDER-INPUT') + + def test_34_disconnect_ROADMA(self): + url = ("{}/config/network-topology:" + "network-topology/topology/topology-netconf/node/ROADM-A1" + .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) + #Delete in the clli-network + url = ("{}/config/ietf-network:networks/network/clli-network/node/NodeA" + .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) + + def test_35_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')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + self.assertNotIn('node', res['network'][0]) + + def test_36_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')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + self.assertNotIn('node', res['network'][0]) + + def test_37_check_roadm2roadm_link_persistence(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')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + nbLink=len(res['network'][0]['ietf-network-topology:link']) + self.assertNotIn('node', res['network'][0]) + self.assertEqual(nbLink, 6) + +if __name__ == "__main__": + #logging.basicConfig(filename='./transportpce_tests/log/response.log',filemode='w',level=logging.DEBUG) + #logging.debug('I am there') + unittest.main(verbosity=2) diff --git a/tox.ini b/tox.ini index e1c541739..4a4e53be2 100644 --- a/tox.ini +++ b/tox.ini @@ -18,14 +18,19 @@ commands = {py27,portmapping,topoPortMapping,rspn,topology,pce,olm,end2end}: - bash -c "if [ ! `which mvn` ]; then ./installMavenCentOS.sh ; fi" {py27,portmapping,topoPortMapping,rspn,topology,pce,olm,end2end}: - bash -c "(cd .. && mvn clean install -s tests/odl_settings.xml -DskipTests -Dmaven.javadoc.skip=true)" {py27,portmapping,topoPortMapping,rspn,topology,olm,end2end}: - ./buildHoneynode.sh - #py27: nosetests --with-xunit transportpce_tests/ - {py27,portmapping}: - nosetests --with-xunit transportpce_tests/test_portmapping.py - {py27,topoPortMapping}: - nosetests --with-xunit transportpce_tests/test_topoPortMapping.py - {py27,topology}: - nosetests --with-xunit transportpce_tests/test_topology.py - {py27,rspn}: - nosetests --with-xunit transportpce_tests/test_renderer_service_path_nominal.py - {py27,pce}: - nosetests --with-xunit transportpce_tests/test_pce.py - {py27,olm}: - nosetests --with-xunit transportpce_tests/test_olm.py - {py27,end2end}: - nosetests --with-xunit transportpce_tests/test_end2end.py + #py27: nosetests --with-xunit transportpce_tests/1.2.1/ + {py27,portmapping}: - nosetests --with-xunit transportpce_tests/1.2.1/test_portmapping.py + {py27,topoPortMapping}: - nosetests --with-xunit transportpce_tests/1.2.1/test_topoPortMapping.py + {py27,topology}: - nosetests --with-xunit transportpce_tests/1.2.1/test_topology.py + {py27,rspn}: - nosetests --with-xunit transportpce_tests/1.2.1/test_renderer_service_path_nominal.py + {py27,pce}: - nosetests --with-xunit transportpce_tests/1.2.1/test_pce.py + {py27,olm}: - nosetests --with-xunit transportpce_tests/1.2.1/test_olm.py + {py27,end2end}: - nosetests --with-xunit transportpce_tests/1.2.1/test_end2end.py + {ordm221,portmapping221,rspn221,topology221,olm221}: - ./buildHoneynode.sh 2.2.1 + {ordm221,portmapping221}: - nosetests --with-xunit transportpce_tests/2.2.1/test_portmapping.py + {ordm221,topology221}: - nosetests --with-xunit transportpce_tests/2.2.1/test_topology.py + {ordm221,rspn221}: - nosetests --with-xunit transportpce_tests/2.2.1/test_renderer_service_path_nominal.py + {ordm221,olm221}: - nosetests --with-xunit transportpce_tests/2.2.1/test_olm.py [testenv:docs] -- 2.36.6