Test alignment between topology and portmapping

- connect one ROADM and one XPDR
- check the alignment of the TPs naming between both the topology
and what appears in the port mapping
- test 5 fails because naming of TP  are not aligned for the XPDR

JIIRA: TRNSPRTPCE-5
Change-Id: I514f2cae964c8cb12e2bb8700a19d946f4059b72
Signed-off-by: Ahmed Triki <ahmed.triki@orange.com>

diff --git a/tests/tox.ini b/tests/tox.ini
index e8ceff59fffe60299f64063977bb4577a2d295af..a20fbdb77c7b8a7deda7ef47c71829ad17437ba1 100644 (file)
--- a/tests/tox.ini
+++ b/tests/tox.ini
@@ -11,10 +11,11 @@ whitelist_externals = bash
 commands =
   {p27,portmapping,topology,pce}: bash -c "(cd .. && mvn clean install -DskipTests -Dmaven.javadoc.skip=true)"
   {py27,portmapping}: ./build.sh
-  {topology}: ./buildTesttool.sh
+  {topology,topoPortMapping}: ./buildTesttool.sh
   {p27,servicehandler}: ./SH_func_tests.sh
   py27: nosetests --with-xunit transportpce_tests/
   portmapping: nosetests --with-xunit transportpce_tests/test_portmapping.py
+  topoPortMapping: nosetests --with-xunit transportpce_tests/test_topoPortMapping.py
   topology: nosetests --with-xunit transportpce_tests/test_topology.py
   pce: nosetests --with-xunit transportpce_tests/test_pce.py
   servicehandler: nosetests --with-xunit transportpce_tests/test_servicehandler.py

diff --git a/tests/transportpce_tests/test_topoPortMapping.py b/tests/transportpce_tests/test_topoPortMapping.py
new file mode 100644 (file)
index 0000000..02e5a0a
--- /dev/null
+++ b/tests/transportpce_tests/test_topoPortMapping.py
@@ -0,0 +1,279 @@
+#!/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):
+
+    testtools_process1 = None
+    testtools_process2 = None
+    odl_process = None
+    restconf_baseurl = "http://localhost:8181/restconf"
+
+    @classmethod
+    def __start_testtools(cls):
+        executable = ("./netconf/netconf/tools/netconf-testtool/target/"
+                      "netconf-testtool-1.3.1-executable.jar")
+        if os.path.isfile(executable):
+            if not os.path.exists("transportpce_tests/log"):
+                os.makedirs("transportpce_tests/log")
+            if os.path.isfile("./transportpce_tests/log/topoPortMap.log"):
+                os.remove("transportpce_tests/log/topoPortMap.log")
+            with open('transportpce_tests/log/testtools_ROADMA.log', 'w') as outfile1:
+                cls.testtools_process1 = subprocess.Popen(
+                    ["java", "-jar", executable, "--schemas-dir", "schemas",
+                     "--initial-config-xml", "sample_configs/nodes_config/sample-config-ROADMA.xml","--starting-port","17831"],
+                    stdout=outfile1)
+            with open('transportpce_tests/log/testtools_XPDRA.log', 'w') as outfile2:
+                cls.testtools_process2 = subprocess.Popen(
+                    ["java", "-jar", executable, "--schemas-dir", "schemas",
+                     "--initial-config-xml", "sample_configs/nodes_config/sample-config-XPDRA.xml","--starting-port","17830"],
+                    stdout=outfile2)
+
+    @classmethod
+    def __start_odl(cls):
+        executable = "../karaf/target/assembly/bin/karaf"
+        with open('transportpce_tests/log/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_testtools()
+        cls.__start_odl()
+        time.sleep(40)
+
+    @classmethod
+    def tearDownClass(cls):
+        cls.testtools_process1.send_signal(signal.SIGINT)
+        cls.testtools_process1.wait()
+        cls.testtools_process2.send_signal(signal.SIGINT)
+        cls.testtools_process2.wait()
+        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()
+
+    def setUp(self):
+        time.sleep(10)
+
+    #Connect the ROADMA
+    def test_01_connect_roadma(self):
+        #Config ROADMA
+        url = ("{}/config/network-topology:"
+                "network-topology/topology/topology-netconf/node/ROADMA"
+               .format(self.restconf_baseurl))
+        data = {"node": [{
+             "node-id": "ROADMA",
+             "netconf-node-topology:username": "admin",
+             "netconf-node-topology:password": "admin",
+             "netconf-node-topology:host": "127.0.0.1",
+             "netconf-node-topology:port": "17831",
+             "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)
+        #seems sometimes to return 200 instead of 201
+        #self.assertEqual(response.status_code, requests.codes.ok)
+        time.sleep(10)
+
+    #Verify the termination points of the ROADMA
+    def test_02_compareOpenroadmTopologyPortMapping(self):
+        #Verify the termination points related to the SRGs
+        nbSrg=1
+        for s in range(1,nbSrg+1):
+            url_topo="{}/config/ietf-network:network/openroadm-topology/node/ROADMA-SRG"+`s`
+            with open('./transportpce_tests/log/topoPortMap.log', 'a') as outfile1:
+                outfile1.write('Config: '+`s`+' : '+url_topo+'\n')
+            url = (url_topo.format(self.restconf_baseurl))
+            headers = {'content-type': 'application/json'}
+            response_topo = requests.request(
+                "GET", url, headers=headers, auth=('admin', 'admin'))
+            self.assertEqual(response_topo.status_code, requests.codes.ok)
+            res_topo = response_topo.json()
+            nbTP=len(res_topo['node'][0]['ietf-network-topology:termination-point'])
+            for i in range(0,nbTP):
+                tp_id=res_topo['node'][0]['ietf-network-topology:termination-point'][i]['tp-id']
+                if(not "CP" in tp_id):
+                    url_map="{}/config/portmapping:network/nodes/ROADMA/mapping/"+tp_id
+                    with open('./transportpce_tests/log/topoPortMap.log', 'a') as outfile1:
+                        outfile1.write('Config: '+`i`+'/'+ `nbTP`+' : '+url_map+'\n')
+                    url = (url_map.format(self.restconf_baseurl))
+                    headers = {'content-type': 'application/json'}
+                    response_portMap = requests.request(
+                        "GET", url, headers=headers, auth=('admin', 'admin'))
+                    self.assertEqual(response_portMap.status_code, requests.codes.ok)
+
+        #Verify the termination points related to the degrees
+        nbDeg=2
+        for d in range(1,nbDeg+1):
+            url_topo="{}/config/ietf-network:network/openroadm-topology/node/ROADMA-DEG"+`d`
+            with open('./transportpce_tests/log/topoPortMap.log', 'a') as outfile1:
+                outfile1.write(url_topo+'\n')
+            url = (url_topo.format(self.restconf_baseurl))
+            headers = {'content-type': 'application/json'}
+            response_topo = requests.request(
+                "GET", url, headers=headers, auth=('admin', 'admin'))
+            self.assertEqual(response_topo.status_code, requests.codes.ok)
+            res_topo = response_topo.json()
+            nbTP=len(res_topo['node'][0]['ietf-network-topology:termination-point'])
+            for i in range(0,nbTP):
+                tp_id=res_topo['node'][0]['ietf-network-topology:termination-point'][i]['tp-id']
+                if(not "CTP" in tp_id):
+                    url_map ="{}/config/portmapping:network/nodes/ROADMA/mapping/"+tp_id
+                    with open('./transportpce_tests/log/topoPortMap.log', 'a') as outfile1:
+                        outfile1.write('Config: '+`i`+'/'+ `nbTP`+' : '+url_map+'\n')
+                    url = (url_map.format(self.restconf_baseurl))
+                    headers = {'content-type': 'application/json'}
+                    response_portMap = requests.request(
+                        "GET", url, headers=headers, auth=('admin', 'admin'))
+                    self.assertEqual(response_portMap.status_code, requests.codes.ok)
+        time.sleep(1)
+
+    #Disconnect the ROADMA
+    def test_03_disconnect_device(self):
+        url = ("{}/config/network-topology:"
+                "network-topology/topology/topology-netconf/node/ROADMA"
+               .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: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)
+        #Delete in the openroadm-network
+        url = ("{}/config/ietf-network:network/openroadm-network/node/ROADMA"
+               .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)
+        time.sleep(5)
+
+    #Connect the XPDRA
+    def test_04_connect_xpdr(self):
+         #Config XPDRA
+         url = ("{}/config/network-topology:"
+                 "network-topology/topology/topology-netconf/node/XPDRA"
+                .format(self.restconf_baseurl))
+         data = {"node": [{
+              "node-id": "XPDRA",
+              "netconf-node-topology:username": "admin",
+              "netconf-node-topology:password": "admin",
+              "netconf-node-topology:host": "127.0.0.1",
+              "netconf-node-topology:port": "17830",
+              "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)
+         #seems sometimes to return 200 instead of 201
+         #self.assertEqual(response.status_code, requests.codes.ok)
+         time.sleep(15)
+
+    #Verify the termination points related to XPDR
+    def test_05_compareOpenroadmTopologyPortMapping(self):
+        nbXPDR=1
+        for p in(1,nbXPDR+1):
+            url_topo = "{}/config/ietf-network:network/openroadm-topology/node/XPDRA-XPDR"+`p`
+            with open('./transportpce_tests/log/topoPortMap.log', 'a') as outfile1:
+                outfile1.write('Config: '+`p`+' : '+url_topo+'\n')
+            url = (url_topo.format(self.restconf_baseurl))
+            headers = {'content-type': 'application/json'}
+            response_topo = requests.request(
+                "GET", url, headers=headers, auth=('admin', 'admin'))
+            self.assertEqual(response_topo.status_code, requests.codes.ok)
+            res_topo = response_topo.json()
+            nbTP=len(res_topo['node'][0]['ietf-network-topology:termination-point'])
+            for i in range(0,nbTP):
+                tp_id=res_topo['node'][0]['ietf-network-topology:termination-point'][i]['tp-id']
+                url_map = "{}/config/portmapping:network/nodes/XPDRA/mapping/"+tp_id
+                with open('./transportpce_tests/log/topoPortMap.log', 'a') as outfile1:
+                    outfile1.write('Config: '+`i`+'/'+ `nbTP`+' : '+url_map+'\n')
+                url = url_map.format(self.restconf_baseurl)
+                headers = {'content-type': 'application/json'}
+                response_portMap = requests.request(
+                    "GET", url, headers=headers, auth=('admin', 'admin'))
+                self.assertEqual(response_portMap.status_code, requests.codes.ok)
+                #Verify the tail equipment id of the client
+                xpdr_client=res_topo['node'][0]['ietf-network-topology:termination-point'][i]["org-openroadm-network-topology:xpdr-client-attributes"]["tail-equipment-id"]
+                url_map = "{}/config/portmapping:network/nodes/XPDRA/mapping/"+xpdr_client
+                with open('./transportpce_tests/log/topoPortMap.log', 'a') as outfile1:
+                    outfile1.write('Config: '+`i`+'/'+ `nbTP`+' : '+xpdr_client+'\n')
+                url = url_map.format(self.restconf_baseurl)
+                headers = {'content-type': 'application/json'}
+                response_xpdrClient = requests.request(
+                    "GET", url, headers=headers, auth=('admin', 'admin'))
+                self.assertEqual(response_xpdrClient.status_code, requests.codes.ok)
+                #Verify the tail equipment id of the network
+                xpdr_network=res_topo['node'][0]['ietf-network-topology:termination-point'][i]["org-openroadm-network-topology:xpdr-network-attributes"]["tail-equipment-id"]
+                url_map = "{}/config/portmapping:network/nodes/XPDRA/mapping/"+xpdr_network
+                with open('./transportpce_tests/log/topoPortMap.log', 'a') as outfile1:
+                    outfile1.write('Config: '+`i`+'/'+ `nbTP`+' : '+xpdr_network+'\n')
+                url = url_map.format(self.restconf_baseurl)
+                headers = {'content-type': 'application/json'}
+                response_xpdrNetwork = requests.request(
+                    "GET", url, headers=headers, auth=('admin', 'admin'))
+                self.assertEqual(response_xpdrNetwork.status_code, requests.codes.ok)
+
+    #Disconnect the XPDRA
+    def test_06_disconnect_device(self):
+        url = ("{}/config/network-topology:"
+               "network-topology/topology/topology-netconf/node/XPDRA"
+              .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 openroadm-network
+        url = ("{}/config/ietf-network:network/openroadm-network/node/XPDRA"
+               .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)
+
+
+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)