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-SNAPSHOT-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/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/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 clli-network
167 url = ("{}/config/ietf-network:network/clli-network/node/NodeA"
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)
175 #Delete in the openroadm-network
176 url = ("{}/config/ietf-network:network/openroadm-network/node/ROADMA"
177 .format(self.restconf_baseurl))
179 headers = {'content-type': 'application/json'}
180 response = requests.request(
181 "DELETE", url, data=json.dumps(data), headers=headers,
182 auth=('admin', 'admin'))
183 self.assertEqual(response.status_code, requests.codes.ok)
187 def test_04_connect_xpdr(self):
189 url = ("{}/config/network-topology:"
190 "network-topology/topology/topology-netconf/node/XPDRA"
191 .format(self.restconf_baseurl))
194 "netconf-node-topology:username": "admin",
195 "netconf-node-topology:password": "admin",
196 "netconf-node-topology:host": "127.0.0.1",
197 "netconf-node-topology:port": "17830",
198 "netconf-node-topology:tcp-only": "false",
199 "netconf-node-topology:pass-through": {}}]}
200 headers = {'content-type': 'application/json'}
201 response = requests.request(
202 "PUT", url, data=json.dumps(data), headers=headers,
203 auth=('admin', 'admin'))
204 self.assertEqual(response.status_code, requests.codes.created)
205 #seems sometimes to return 200 instead of 201
206 #self.assertEqual(response.status_code, requests.codes.ok)
209 #Verify the termination points related to XPDR
210 def test_05_compareOpenroadmTopologyPortMapping(self):
212 for p in(1,nbXPDR+1):
213 url_topo = "{}/config/ietf-network:network/openroadm-topology/node/XPDRA-XPDR"+`p`
214 with open('./transportpce_tests/log/topoPortMap.log', 'a') as outfile1:
215 outfile1.write('Config: '+`p`+' : '+url_topo+'\n')
216 url = (url_topo.format(self.restconf_baseurl))
217 headers = {'content-type': 'application/json'}
218 response_topo = requests.request(
219 "GET", url, headers=headers, auth=('admin', 'admin'))
220 self.assertEqual(response_topo.status_code, requests.codes.ok)
221 res_topo = response_topo.json()
222 nbTP=len(res_topo['node'][0]['ietf-network-topology:termination-point'])
223 for i in range(0,nbTP):
224 tp_id=res_topo['node'][0]['ietf-network-topology:termination-point'][i]['tp-id']
225 url_map = "{}/config/portmapping:network/nodes/XPDRA/mapping/"+tp_id
226 with open('./transportpce_tests/log/topoPortMap.log', 'a') as outfile1:
227 outfile1.write('Config: '+`i`+'/'+ `nbTP`+' : '+url_map+'\n')
228 url = url_map.format(self.restconf_baseurl)
229 headers = {'content-type': 'application/json'}
230 response_portMap = requests.request(
231 "GET", url, headers=headers, auth=('admin', 'admin'))
232 self.assertEqual(response_portMap.status_code, requests.codes.ok)
233 #Verify the tail equipment id of the client
234 xpdr_client=res_topo['node'][0]['ietf-network-topology:termination-point'][i]["org-openroadm-network-topology:xpdr-client-attributes"]["tail-equipment-id"]
235 url_map = "{}/config/portmapping:network/nodes/XPDRA/mapping/"+xpdr_client
236 with open('./transportpce_tests/log/topoPortMap.log', 'a') as outfile1:
237 outfile1.write('Config: '+`i`+'/'+ `nbTP`+' : '+xpdr_client+'\n')
238 url = url_map.format(self.restconf_baseurl)
239 headers = {'content-type': 'application/json'}
240 response_xpdrClient = requests.request(
241 "GET", url, headers=headers, auth=('admin', 'admin'))
242 self.assertEqual(response_xpdrClient.status_code, requests.codes.ok)
243 #Verify the tail equipment id of the network
244 xpdr_network=res_topo['node'][0]['ietf-network-topology:termination-point'][i]["org-openroadm-network-topology:xpdr-network-attributes"]["tail-equipment-id"]
245 url_map = "{}/config/portmapping:network/nodes/XPDRA/mapping/"+xpdr_network
246 with open('./transportpce_tests/log/topoPortMap.log', 'a') as outfile1:
247 outfile1.write('Config: '+`i`+'/'+ `nbTP`+' : '+xpdr_network+'\n')
248 url = url_map.format(self.restconf_baseurl)
249 headers = {'content-type': 'application/json'}
250 response_xpdrNetwork = requests.request(
251 "GET", url, headers=headers, auth=('admin', 'admin'))
252 self.assertEqual(response_xpdrNetwork.status_code, requests.codes.ok)
254 #Disconnect the XPDRA
255 def test_06_disconnect_device(self):
256 url = ("{}/config/network-topology:"
257 "network-topology/topology/topology-netconf/node/XPDRA"
258 .format(self.restconf_baseurl))
260 headers = {'content-type': 'application/json'}
261 response = requests.request(
262 "DELETE", url, data=json.dumps(data), headers=headers,
263 auth=('admin', 'admin'))
264 self.assertEqual(response.status_code, requests.codes.ok)
265 #Delete in the openroadm-network
266 url = ("{}/config/ietf-network:network/openroadm-network/node/XPDRA"
267 .format(self.restconf_baseurl))
269 headers = {'content-type': 'application/json'}
270 response = requests.request(
271 "DELETE", url, data=json.dumps(data), headers=headers,
272 auth=('admin', 'admin'))
273 self.assertEqual(response.status_code, requests.codes.ok)
276 if __name__ == "__main__":
277 #logging.basicConfig(filename='./transportpce_tests/log/response.log',filemode='w',level=logging.DEBUG)
278 #logging.debug('I am there')
279 unittest.main(verbosity=2)