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 honeynode_process1 = None
26 honeynode_process2 = None
28 restconf_baseurl = "http://localhost:8181/restconf"
30 #START_IGNORE_XTESTING
33 def __init_logfile(cls):
34 if not os.path.exists("transportpce_tests/log"):
35 os.makedirs("transportpce_tests/log")
36 if os.path.isfile("./transportpce_tests/log/topoPortMap.log"):
37 os.remove("transportpce_tests/log/topoPortMap.log")
40 def __start_honeynode1(cls):
41 executable = ("./honeynode/2.1/honeynode-distribution/target/honeynode-distribution-1.18.01-hc"
42 "/honeynode-distribution-1.18.01/honeycomb-tpce")
43 if os.path.isfile(executable):
44 with open('honeynode1.log', 'w') as outfile:
45 cls.honeynode_process1 = subprocess.Popen(
46 [executable, "17830", "sample_configs/openroadm/2.1/oper-XPDRA.xml"],
51 def __start_honeynode2(cls):
52 executable = ("./honeynode/2.1/honeynode-distribution/target/honeynode-distribution-1.18.01-hc"
53 "/honeynode-distribution-1.18.01/honeycomb-tpce")
54 if os.path.isfile(executable):
55 with open('honeynode2.log', 'w') as outfile:
56 cls.honeynode_process2 = subprocess.Popen(
57 [executable, "17831", "sample_configs/openroadm/2.1/oper-ROADMA.xml"],
63 executable = "../karaf/target/assembly/bin/karaf"
64 with open('transportpce_tests/log/odl.log', 'w') as outfile:
65 cls.odl_process = subprocess.Popen(
66 ["bash", executable, "server"], stdout=outfile,
67 stdin=open(os.devnull))
73 cls.__start_honeynode1()
75 cls.__start_honeynode2()
81 def tearDownClass(cls):
82 for child in psutil.Process(cls.odl_process.pid).children():
83 child.send_signal(signal.SIGINT)
85 cls.odl_process.send_signal(signal.SIGINT)
86 cls.odl_process.wait()
87 for child in psutil.Process(cls.honeynode_process1.pid).children():
88 child.send_signal(signal.SIGINT)
90 cls.honeynode_process1.send_signal(signal.SIGINT)
91 cls.honeynode_process1.wait()
92 for child in psutil.Process(cls.honeynode_process2.pid).children():
93 child.send_signal(signal.SIGINT)
95 cls.honeynode_process2.send_signal(signal.SIGINT)
96 cls.honeynode_process2.wait()
104 def test_01_connect_roadma(self):
106 url = ("{}/config/network-topology:"
107 "network-topology/topology/topology-netconf/node/ROADMA"
108 .format(self.restconf_baseurl))
111 "netconf-node-topology:username": "admin",
112 "netconf-node-topology:password": "admin",
113 "netconf-node-topology:host": "127.0.0.1",
114 "netconf-node-topology:port": "17831",
115 "netconf-node-topology:tcp-only": "false",
116 "netconf-node-topology:pass-through": {}}]}
117 headers = {'content-type': 'application/json'}
118 response = requests.request(
119 "PUT", url, data=json.dumps(data), headers=headers,
120 auth=('admin', 'admin'))
121 self.assertEqual(response.status_code, requests.codes.created)
122 #seems sometimes to return 200 instead of 201
123 #self.assertEqual(response.status_code, requests.codes.ok)
126 #Verify the termination points of the ROADMA
127 def test_02_compareOpenroadmTopologyPortMapping(self):
128 #Verify the termination points related to the SRGs
130 for s in range(1,nbSrg+1):
131 url_topo="{}/config/ietf-network:networks/network/openroadm-topology/node/ROADMA-SRG"+`s`
132 with open('./transportpce_tests/log/topoPortMap.log', 'a') as outfile1:
133 outfile1.write('Config: '+`s`+' : '+url_topo+'\n')
134 url = (url_topo.format(self.restconf_baseurl))
135 headers = {'content-type': 'application/json'}
136 response_topo = requests.request(
137 "GET", url, headers=headers, auth=('admin', 'admin'))
138 self.assertEqual(response_topo.status_code, requests.codes.ok)
139 res_topo = response_topo.json()
140 nbTP=len(res_topo['node'][0]['ietf-network-topology:termination-point'])
141 for i in range(0,nbTP):
142 tp_id=res_topo['node'][0]['ietf-network-topology:termination-point'][i]['tp-id']
143 if(not "CP" in tp_id):
144 url_map="{}/config/transportpce-portmapping:network/nodes/ROADMA/mapping/"+tp_id
145 with open('./transportpce_tests/log/topoPortMap.log', 'a') as outfile1:
146 outfile1.write('Config: '+`i`+'/'+ `nbTP`+' : '+url_map+'\n')
147 url = (url_map.format(self.restconf_baseurl))
148 headers = {'content-type': 'application/json'}
149 response_portMap = requests.request(
150 "GET", url, headers=headers, auth=('admin', 'admin'))
151 self.assertEqual(response_portMap.status_code, requests.codes.ok)
153 #Verify the termination points related to the degrees
155 for d in range(1,nbDeg+1):
156 url_topo="{}/config/ietf-network:networks/network/openroadm-topology/node/ROADMA-DEG"+`d`
157 with open('./transportpce_tests/log/topoPortMap.log', 'a') as outfile1:
158 outfile1.write(url_topo+'\n')
159 url = (url_topo.format(self.restconf_baseurl))
160 headers = {'content-type': 'application/json'}
161 response_topo = requests.request(
162 "GET", url, headers=headers, auth=('admin', 'admin'))
163 self.assertEqual(response_topo.status_code, requests.codes.ok)
164 res_topo = response_topo.json()
165 nbTP=len(res_topo['node'][0]['ietf-network-topology:termination-point'])
166 for i in range(0,nbTP):
167 tp_id=res_topo['node'][0]['ietf-network-topology:termination-point'][i]['tp-id']
168 if(not "CTP" in tp_id):
169 url_map ="{}/config/transportpce-portmapping:network/nodes/ROADMA/mapping/"+tp_id
170 with open('./transportpce_tests/log/topoPortMap.log', 'a') as outfile1:
171 outfile1.write('Config: '+`i`+'/'+ `nbTP`+' : '+url_map+'\n')
172 url = (url_map.format(self.restconf_baseurl))
173 headers = {'content-type': 'application/json'}
174 response_portMap = requests.request(
175 "GET", url, headers=headers, auth=('admin', 'admin'))
176 self.assertEqual(response_portMap.status_code, requests.codes.ok)
179 #Disconnect the ROADMA
180 def test_03_disconnect_device(self):
181 url = ("{}/config/network-topology:"
182 "network-topology/topology/topology-netconf/node/ROADMA"
183 .format(self.restconf_baseurl))
185 headers = {'content-type': 'application/json'}
186 response = requests.request(
187 "DELETE", url, data=json.dumps(data), headers=headers,
188 auth=('admin', 'admin'))
189 self.assertEqual(response.status_code, requests.codes.ok)
190 #Delete in the openroadm-network
191 # url = ("{}/config/ietf-network:networks/network/openroadm-network/node/ROADMA"
192 # .format(self.restconf_baseurl))
194 # headers = {'content-type': 'application/json'}
195 # response = requests.request(
196 # "DELETE", url, data=json.dumps(data), headers=headers,
197 # auth=('admin', 'admin'))
198 # self.assertEqual(response.status_code, requests.codes.ok)
202 def test_04_connect_xpdr(self):
204 url = ("{}/config/network-topology:"
205 "network-topology/topology/topology-netconf/node/XPDRA"
206 .format(self.restconf_baseurl))
209 "netconf-node-topology:username": "admin",
210 "netconf-node-topology:password": "admin",
211 "netconf-node-topology:host": "127.0.0.1",
212 "netconf-node-topology:port": "17830",
213 "netconf-node-topology:tcp-only": "false",
214 "netconf-node-topology:pass-through": {}}]}
215 headers = {'content-type': 'application/json'}
216 response = requests.request(
217 "PUT", url, data=json.dumps(data), headers=headers,
218 auth=('admin', 'admin'))
219 self.assertEqual(response.status_code, requests.codes.created)
220 #seems sometimes to return 200 instead of 201
221 #self.assertEqual(response.status_code, requests.codes.ok)
224 #Verify the termination points related to XPDR
225 @unittest.expectedFailure
226 def test_05_compareOpenroadmTopologyPortMapping(self):
228 for p in(1,nbXPDR+1):
231 url_topo = "{}/config/ietf-network:networks/network/openroadm-topology/node/XPDRA-XPDR"+`p`
232 with open('./transportpce_tests/log/topoPortMap.log', 'a') as outfile1:
233 outfile1.write('Config: '+`p`+' : '+url_topo+'\n')
234 url = (url_topo.format(self.restconf_baseurl))
235 headers = {'content-type': 'application/json'}
236 response_topo = requests.request(
237 "GET", url, headers=headers, auth=('admin', 'admin'))
238 self.assertEqual(response_topo.status_code, requests.codes.ok)
239 res_topo = response_topo.json()
240 nbTP=len(res_topo['node'][0]['ietf-network-topology:termination-point'])
241 for i in range(0,nbTP):
242 tp_id=res_topo['node'][0]['ietf-network-topology:termination-point'][i]['tp-id']
243 url_map = "{}/config/transportpce-portmapping:network/nodes/XPDRA/mapping/"+tp_id
244 with open('./transportpce_tests/log/topoPortMap.log', 'a') as outfile1:
245 outfile1.write('Config: '+`i`+'/'+ `nbTP`+' : '+url_map+'\n')
246 url = url_map.format(self.restconf_baseurl)
247 headers = {'content-type': 'application/json'}
248 response_portMap = requests.request(
249 "GET", url, headers=headers, auth=('admin', 'admin'))
250 self.assertEqual(response_portMap.status_code, requests.codes.ok)
251 if("CLIENT" in tp_id):
252 #Verify the tail equipment id of the client
253 xpdr_client=res_topo['node'][0]['ietf-network-topology:termination-point'][i]["org-openroadm-network-topology:xpdr-client-attributes"]["tail-equipment-id"]
254 url_map = "{}/config/transportpce-portmapping:network/nodes/XPDRA/mapping/"+xpdr_client
255 with open('./transportpce_tests/log/topoPortMap.log', 'a') as outfile1:
256 outfile1.write('Config: '+`i`+'/'+ `nbTP`+' : '+xpdr_client+'\n')
257 url = url_map.format(self.restconf_baseurl)
258 headers = {'content-type': 'application/json'}
259 response_xpdrClient = requests.request(
260 "GET", url, headers=headers, auth=('admin', 'admin'))
261 self.assertEqual(response_xpdrClient.status_code, requests.codes.ok)
262 if("NETWORK" in tp_id):
263 #Verify the tail equipment id of the network
264 xpdr_network=res_topo['node'][0]['ietf-network-topology:termination-point'][i]["org-openroadm-network-topology:xpdr-network-attributes"]["tail-equipment-id"]
265 url_map = "{}/config/transportpce-portmapping:network/nodes/XPDRA/mapping/"+xpdr_network
266 with open('./transportpce_tests/log/topoPortMap.log', 'a') as outfile1:
267 outfile1.write('Config: '+`i`+'/'+ `nbTP`+' : '+xpdr_network+'\n')
268 url = url_map.format(self.restconf_baseurl)
269 headers = {'content-type': 'application/json'}
270 response_xpdrNetwork = requests.request(
271 "GET", url, headers=headers, auth=('admin', 'admin'))
272 self.assertEqual(response_xpdrNetwork.status_code, requests.codes.ok)
274 #Disconnect the XPDRA
275 def test_06_disconnect_device(self):
276 url = ("{}/config/network-topology:"
277 "network-topology/topology/topology-netconf/node/XPDRA"
278 .format(self.restconf_baseurl))
280 headers = {'content-type': 'application/json'}
281 response = requests.request(
282 "DELETE", url, data=json.dumps(data), headers=headers,
283 auth=('admin', 'admin'))
284 self.assertEqual(response.status_code, requests.codes.ok)
285 #Delete in the openroadm-network
286 # url = ("{}/config/ietf-network:networks/network/openroadm-network/node/XPDRA"
287 # .format(self.restconf_baseurl))
289 # headers = {'content-type': 'application/json'}
290 # response = requests.request(
291 # "DELETE", url, data=json.dumps(data), headers=headers,
292 # auth=('admin', 'admin'))
293 # self.assertEqual(response.status_code, requests.codes.ok)
296 if __name__ == "__main__":
297 #logging.basicConfig(filename='./transportpce_tests/log/response.log',filemode='w',level=logging.DEBUG)
298 #logging.debug('I am there')
299 unittest.main(verbosity=2)