3 ##############################################################################
4 #Copyright (c) 2017 Orange, Inc. and others. All rights reserved.
6 # All rights reserved. This program and the accompanying materials
7 # are made available under the terms of the Apache License, Version 2.0
8 # which accompanies this distribution, and is available at
9 # http://www.apache.org/licenses/LICENSE-2.0
10 ##############################################################################
23 class TransportPCEtesting(unittest.TestCase):
25 testtools_process1 = None
26 testtools_process2 = None
28 restconf_baseurl = "http://localhost:8181/restconf"
31 def __start_testtools(cls):
32 executable = ("./netconf/netconf/tools/netconf-testtool/target/"
33 "netconf-testtool-1.5.0-executable.jar")
34 if os.path.isfile(executable):
35 if not os.path.exists("transportpce_tests/log"):
36 os.makedirs("transportpce_tests/log")
37 if os.path.isfile("./transportpce_tests/log/topoPortMap.log"):
38 os.remove("transportpce_tests/log/topoPortMap.log")
39 with open('transportpce_tests/log/testtools_ROADMA.log', 'w') as outfile1:
40 cls.testtools_process1 = subprocess.Popen(
41 ["java", "-jar", executable, "--schemas-dir", "schemas",
42 "--initial-config-xml", "sample_configs/nodes_config/sample-config-ROADMA.xml","--starting-port","17831"],
44 with open('transportpce_tests/log/testtools_XPDRA.log', 'w') as outfile2:
45 cls.testtools_process2 = subprocess.Popen(
46 ["java", "-jar", executable, "--schemas-dir", "schemas",
47 "--initial-config-xml", "sample_configs/nodes_config/sample-config-XPDRA.xml","--starting-port","17830"],
52 executable = "../karaf/target/assembly/bin/karaf"
53 with open('transportpce_tests/log/odl.log', 'w') as outfile:
54 cls.odl_process = subprocess.Popen(
55 ["bash", executable, "server"], stdout=outfile,
56 stdin=open(os.devnull))
60 cls.__start_testtools()
65 def tearDownClass(cls):
66 cls.testtools_process1.send_signal(signal.SIGINT)
67 cls.testtools_process1.wait()
68 cls.testtools_process2.send_signal(signal.SIGINT)
69 cls.testtools_process2.wait()
70 for child in psutil.Process(cls.odl_process.pid).children():
71 child.send_signal(signal.SIGINT)
73 cls.odl_process.send_signal(signal.SIGINT)
74 cls.odl_process.wait()
80 def test_01_connect_roadma(self):
82 url = ("{}/config/network-topology:"
83 "network-topology/topology/topology-netconf/node/ROADMA"
84 .format(self.restconf_baseurl))
87 "netconf-node-topology:username": "admin",
88 "netconf-node-topology:password": "admin",
89 "netconf-node-topology:host": "127.0.0.1",
90 "netconf-node-topology:port": "17831",
91 "netconf-node-topology:tcp-only": "false",
92 "netconf-node-topology:pass-through": {}}]}
93 headers = {'content-type': 'application/json'}
94 response = requests.request(
95 "PUT", url, data=json.dumps(data), headers=headers,
96 auth=('admin', 'admin'))
97 self.assertEqual(response.status_code, requests.codes.created)
98 #seems sometimes to return 200 instead of 201
99 #self.assertEqual(response.status_code, requests.codes.ok)
102 #Verify the termination points of the ROADMA
103 def test_02_compareOpenroadmTopologyPortMapping(self):
104 #Verify the termination points related to the SRGs
106 for s in range(1,nbSrg+1):
107 url_topo="{}/config/ietf-network:network/openroadm-topology/node/ROADMA-SRG"+`s`
108 with open('./transportpce_tests/log/topoPortMap.log', 'a') as outfile1:
109 outfile1.write('Config: '+`s`+' : '+url_topo+'\n')
110 url = (url_topo.format(self.restconf_baseurl))
111 headers = {'content-type': 'application/json'}
112 response_topo = requests.request(
113 "GET", url, headers=headers, auth=('admin', 'admin'))
114 self.assertEqual(response_topo.status_code, requests.codes.ok)
115 res_topo = response_topo.json()
116 nbTP=len(res_topo['node'][0]['ietf-network-topology:termination-point'])
117 for i in range(0,nbTP):
118 tp_id=res_topo['node'][0]['ietf-network-topology:termination-point'][i]['tp-id']
119 if(not "CP" in tp_id):
120 url_map="{}/config/transportpce-portmapping:network/nodes/ROADMA/mapping/"+tp_id
121 with open('./transportpce_tests/log/topoPortMap.log', 'a') as outfile1:
122 outfile1.write('Config: '+`i`+'/'+ `nbTP`+' : '+url_map+'\n')
123 url = (url_map.format(self.restconf_baseurl))
124 headers = {'content-type': 'application/json'}
125 response_portMap = requests.request(
126 "GET", url, headers=headers, auth=('admin', 'admin'))
127 self.assertEqual(response_portMap.status_code, requests.codes.ok)
129 #Verify the termination points related to the degrees
131 for d in range(1,nbDeg+1):
132 url_topo="{}/config/ietf-network:network/openroadm-topology/node/ROADMA-DEG"+`d`
133 with open('./transportpce_tests/log/topoPortMap.log', 'a') as outfile1:
134 outfile1.write(url_topo+'\n')
135 url = (url_topo.format(self.restconf_baseurl))
136 headers = {'content-type': 'application/json'}
137 response_topo = requests.request(
138 "GET", url, headers=headers, auth=('admin', 'admin'))
139 self.assertEqual(response_topo.status_code, requests.codes.ok)
140 res_topo = response_topo.json()
141 nbTP=len(res_topo['node'][0]['ietf-network-topology:termination-point'])
142 for i in range(0,nbTP):
143 tp_id=res_topo['node'][0]['ietf-network-topology:termination-point'][i]['tp-id']
144 if(not "CTP" in tp_id):
145 url_map ="{}/config/transportpce-portmapping:network/nodes/ROADMA/mapping/"+tp_id
146 with open('./transportpce_tests/log/topoPortMap.log', 'a') as outfile1:
147 outfile1.write('Config: '+`i`+'/'+ `nbTP`+' : '+url_map+'\n')
148 url = (url_map.format(self.restconf_baseurl))
149 headers = {'content-type': 'application/json'}
150 response_portMap = requests.request(
151 "GET", url, headers=headers, auth=('admin', 'admin'))
152 self.assertEqual(response_portMap.status_code, requests.codes.ok)
155 #Disconnect the ROADMA
156 def test_03_disconnect_device(self):
157 url = ("{}/config/network-topology:"
158 "network-topology/topology/topology-netconf/node/ROADMA"
159 .format(self.restconf_baseurl))
161 headers = {'content-type': 'application/json'}
162 response = requests.request(
163 "DELETE", url, data=json.dumps(data), headers=headers,
164 auth=('admin', 'admin'))
165 self.assertEqual(response.status_code, requests.codes.ok)
166 #Delete in the openroadm-network
167 url = ("{}/config/ietf-network:network/openroadm-network/node/ROADMA"
168 .format(self.restconf_baseurl))
170 headers = {'content-type': 'application/json'}
171 response = requests.request(
172 "DELETE", url, data=json.dumps(data), headers=headers,
173 auth=('admin', 'admin'))
174 self.assertEqual(response.status_code, requests.codes.ok)
178 def test_04_connect_xpdr(self):
180 url = ("{}/config/network-topology:"
181 "network-topology/topology/topology-netconf/node/XPDRA"
182 .format(self.restconf_baseurl))
185 "netconf-node-topology:username": "admin",
186 "netconf-node-topology:password": "admin",
187 "netconf-node-topology:host": "127.0.0.1",
188 "netconf-node-topology:port": "17830",
189 "netconf-node-topology:tcp-only": "false",
190 "netconf-node-topology:pass-through": {}}]}
191 headers = {'content-type': 'application/json'}
192 response = requests.request(
193 "PUT", url, data=json.dumps(data), headers=headers,
194 auth=('admin', 'admin'))
195 self.assertEqual(response.status_code, requests.codes.created)
196 #seems sometimes to return 200 instead of 201
197 #self.assertEqual(response.status_code, requests.codes.ok)
200 #Verify the termination points related to XPDR
201 def test_05_compareOpenroadmTopologyPortMapping(self):
203 for p in(1,nbXPDR+1):
206 url_topo = "{}/config/ietf-network:network/openroadm-topology/node/XPDRA-XPDR"+`p`
207 with open('./transportpce_tests/log/topoPortMap.log', 'a') as outfile1:
208 outfile1.write('Config: '+`p`+' : '+url_topo+'\n')
209 url = (url_topo.format(self.restconf_baseurl))
210 headers = {'content-type': 'application/json'}
211 response_topo = requests.request(
212 "GET", url, headers=headers, auth=('admin', 'admin'))
213 self.assertEqual(response_topo.status_code, requests.codes.ok)
214 res_topo = response_topo.json()
215 nbTP=len(res_topo['node'][0]['ietf-network-topology:termination-point'])
216 for i in range(0,nbTP):
217 tp_id=res_topo['node'][0]['ietf-network-topology:termination-point'][i]['tp-id']
218 url_map = "{}/config/transportpce-portmapping:network/nodes/XPDRA/mapping/"+tp_id
219 with open('./transportpce_tests/log/topoPortMap.log', 'a') as outfile1:
220 outfile1.write('Config: '+`i`+'/'+ `nbTP`+' : '+url_map+'\n')
221 url = url_map.format(self.restconf_baseurl)
222 headers = {'content-type': 'application/json'}
223 response_portMap = requests.request(
224 "GET", url, headers=headers, auth=('admin', 'admin'))
225 self.assertEqual(response_portMap.status_code, requests.codes.ok)
226 if("CLIENT" in tp_id):
227 #Verify the tail equipment id of the client
228 xpdr_client=res_topo['node'][0]['ietf-network-topology:termination-point'][i]["org-openroadm-network-topology:xpdr-client-attributes"]["tail-equipment-id"]
229 url_map = "{}/config/transportpce-portmapping:network/nodes/XPDRA/mapping/"+xpdr_client
230 with open('./transportpce_tests/log/topoPortMap.log', 'a') as outfile1:
231 outfile1.write('Config: '+`i`+'/'+ `nbTP`+' : '+xpdr_client+'\n')
232 url = url_map.format(self.restconf_baseurl)
233 headers = {'content-type': 'application/json'}
234 response_xpdrClient = requests.request(
235 "GET", url, headers=headers, auth=('admin', 'admin'))
236 self.assertEqual(response_xpdrClient.status_code, requests.codes.ok)
237 if("NETWORK" in tp_id):
238 #Verify the tail equipment id of the network
239 xpdr_network=res_topo['node'][0]['ietf-network-topology:termination-point'][i]["org-openroadm-network-topology:xpdr-network-attributes"]["tail-equipment-id"]
240 url_map = "{}/config/transportpce-portmapping:network/nodes/XPDRA/mapping/"+xpdr_network
241 with open('./transportpce_tests/log/topoPortMap.log', 'a') as outfile1:
242 outfile1.write('Config: '+`i`+'/'+ `nbTP`+' : '+xpdr_network+'\n')
243 url = url_map.format(self.restconf_baseurl)
244 headers = {'content-type': 'application/json'}
245 response_xpdrNetwork = requests.request(
246 "GET", url, headers=headers, auth=('admin', 'admin'))
247 self.assertEqual(response_xpdrNetwork.status_code, requests.codes.ok)
249 #Disconnect the XPDRA
250 def test_06_disconnect_device(self):
251 url = ("{}/config/network-topology:"
252 "network-topology/topology/topology-netconf/node/XPDRA"
253 .format(self.restconf_baseurl))
255 headers = {'content-type': 'application/json'}
256 response = requests.request(
257 "DELETE", url, data=json.dumps(data), headers=headers,
258 auth=('admin', 'admin'))
259 self.assertEqual(response.status_code, requests.codes.ok)
260 #Delete in the openroadm-network
261 url = ("{}/config/ietf-network:network/openroadm-network/node/XPDRA"
262 .format(self.restconf_baseurl))
264 headers = {'content-type': 'application/json'}
265 response = requests.request(
266 "DELETE", url, data=json.dumps(data), headers=headers,
267 auth=('admin', 'admin'))
268 self.assertEqual(response.status_code, requests.codes.ok)
271 if __name__ == "__main__":
272 #logging.basicConfig(filename='./transportpce_tests/log/response.log',filemode='w',level=logging.DEBUG)
273 #logging.debug('I am there')
274 unittest.main(verbosity=2)