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 TransportPCEPortMappingTesting(unittest.TestCase):
25 testtools_process1 = None
26 testtools_process2 = None
28 restconf_baseurl = "http://127.0.0.1:8181/restconf"
31 def __start_testtools(cls):
32 executable = ("./netconf/netconf/tools/netconf-testtool/target/"
33 "netconf-testtool-1.3.1-executable.jar")
34 if os.path.isfile(executable):
35 with open('testtools1.log', 'w') as outfile:
36 cls.testtools_process1 = subprocess.Popen(
37 ["java", "-jar", executable, "--schemas-dir", "schemas",
38 "--initial-config-xml", "sample_configs/ord_1.2.1/sample-config-ROADM.xml",
39 "--starting-port","17830"],
41 with open('testtools2.log', 'w') as outfile:
42 cls.testtools_process2 = subprocess.Popen(
43 ["java", "-jar", executable, "--schemas-dir", "schemas",
44 "--initial-config-xml", "sample_configs/ord_1.2.1/sample-config-XPDR.xml",
45 "--starting-port","17831"],
50 executable = "../karaf/target/assembly/bin/karaf"
51 with open('odl.log', 'w') as outfile:
52 cls.odl_process = subprocess.Popen(
53 ["bash", executable], stdout=outfile,
54 stdin=open(os.devnull))
58 cls.__start_testtools()
63 def tearDownClass(cls):
64 cls.testtools_process1.send_signal(signal.SIGINT)
65 cls.testtools_process1.wait()
66 cls.testtools_process2.send_signal(signal.SIGINT)
67 cls.testtools_process2.wait()
68 for child in psutil.Process(cls.odl_process.pid).children():
69 child.send_signal(signal.SIGINT)
71 cls.odl_process.send_signal(signal.SIGINT)
72 cls.odl_process.wait()
75 print ("execution of {}".format(self.id().split(".")[-1]))
78 def test_01_restconfAPI(self):
79 url = ("{}/operational/network-topology:network-topology/topology/"
80 "topology-netconf/node/controller-config".format(self.restconf_baseurl))
81 headers = {'content-type': 'application/json'}
82 response = requests.request("GET", url, headers=headers, auth=('admin', 'admin'))
83 self.assertEqual(response.status_code, requests.codes.ok)
85 self.assertEqual(res['node'] [0] ['netconf-node-topology:connection-status'],
88 def test_02_restconfAPI(self):
89 url = ("{}/config/portmapping:network/nodes/controller-config"
90 .format(self.restconf_baseurl))
91 headers = {'content-type': 'application/json'}
92 response = requests.request(
93 "GET", url, headers=headers, auth=('admin', 'admin'))
94 self.assertEqual(response.status_code, requests.codes.not_found)
97 {"error-type":"application","error-tag":"data-missing",
98 "error-message":"Request could not be completed because the relevant data model content does not exist "},
99 res['errors']['error'])
101 def test_03_rdm_device_connected(self):
102 url = ("{}/config/network-topology:"
103 "network-topology/topology/topology-netconf/node/ROADMA"
104 .format(self.restconf_baseurl))
107 "netconf-node-topology:username": "admin",
108 "netconf-node-topology:password": "admin",
109 "netconf-node-topology:host": "127.0.0.1",
110 "netconf-node-topology:port": "17830",
111 "netconf-node-topology:tcp-only": "false",
112 "netconf-node-topology:pass-through": {}}]}
113 headers = {'content-type': 'application/json'}
114 response = requests.request(
115 "PUT", url, data=json.dumps(data), headers=headers,
116 auth=('admin', 'admin'))
117 self.assertEqual(response.status_code, requests.codes.created)
120 def test_04_rdm_device_connected(self):
121 url = ("{}/operational/network-topology:"
122 "network-topology/topology/topology-netconf/node/ROADMA"
123 .format(self.restconf_baseurl))
124 headers = {'content-type': 'application/json'}
125 response = requests.request(
126 "GET", url, headers=headers, auth=('admin', 'admin'))
127 self.assertEqual(response.status_code, requests.codes.ok)
128 res = response.json()
130 res['node'][0]['netconf-node-topology:connection-status'],
134 def test_05_rdm_portmapping_DEG1_TTP_TXRX(self):
135 url = ("{}/config/portmapping:network/"
136 "nodes/ROADMA/mapping/DEG1-TTP-TXRX"
137 .format(self.restconf_baseurl))
138 headers = {'content-type': 'application/json'}
139 response = requests.request(
140 "GET", url, headers=headers, auth=('admin', 'admin'))
141 self.assertEqual(response.status_code, requests.codes.ok)
142 res = response.json()
144 {'supporting-port': 'L1', 'supporting-circuit-pack-name': '2/0',
145 'logical-connection-point': 'DEG1-TTP-TXRX'},
148 def test_06_rdm_portmapping_SRG1_PP7_TXRX(self):
149 url = ("{}/config/portmapping:network/"
150 "nodes/ROADMA/mapping/SRG1-PP7-TXRX"
151 .format(self.restconf_baseurl))
152 headers = {'content-type': 'application/json'}
153 response = requests.request(
154 "GET", url, headers=headers, auth=('admin', 'admin'))
155 self.assertEqual(response.status_code, requests.codes.ok)
156 res = response.json()
158 {'supporting-port': 'C7', 'supporting-circuit-pack-name': '4/0',
159 'logical-connection-point': 'SRG1-PP7-TXRX'},
162 def test_07_rdm_portmapping_SRG3_PP1_TXRX(self):
163 url = ("{}/config/portmapping:network/"
164 "nodes/ROADMA/mapping/SRG3-PP1-TXRX"
165 .format(self.restconf_baseurl))
166 headers = {'content-type': 'application/json'}
167 response = requests.request(
168 "GET", url, headers=headers, auth=('admin', 'admin'))
169 self.assertEqual(response.status_code, requests.codes.ok)
170 res = response.json()
172 {'supporting-port': 'C1', 'supporting-circuit-pack-name': '5/0',
173 'logical-connection-point': 'SRG3-PP1-TXRX'},
176 def test_08_xpdr_device_connected(self):
177 url = ("{}/config/network-topology:"
178 "network-topology/topology/topology-netconf/node/XPDRA"
179 .format(self.restconf_baseurl))
182 "netconf-node-topology:username": "admin",
183 "netconf-node-topology:password": "admin",
184 "netconf-node-topology:host": "127.0.0.1",
185 "netconf-node-topology:port": "17831",
186 "netconf-node-topology:tcp-only": "false",
187 "netconf-node-topology:pass-through": {}}]}
188 headers = {'content-type': 'application/json'}
189 response = requests.request(
190 "PUT", url, data=json.dumps(data), headers=headers,
191 auth=('admin', 'admin'))
192 self.assertEqual(response.status_code, requests.codes.created)
195 def test_09_xpdr_device_connected(self):
196 url = ("{}/operational/network-topology:"
197 "network-topology/topology/topology-netconf/node/XPDRA"
198 .format(self.restconf_baseurl))
199 headers = {'content-type': 'application/json'}
200 response = requests.request(
201 "GET", url, headers=headers, auth=('admin', 'admin'))
202 self.assertEqual(response.status_code, requests.codes.ok)
203 res = response.json()
205 res['node'][0]['netconf-node-topology:connection-status'],
209 def test_10_xpdr_portmapping_NETWORK1(self):
210 url = ("{}/config/portmapping:network/"
211 "nodes/XPDRA/mapping/XPDR-LINE1"
212 .format(self.restconf_baseurl))
213 headers = {'content-type': 'application/json'}
214 response = requests.request(
215 "GET", url, headers=headers, auth=('admin', 'admin'))
216 self.assertEqual(response.status_code, requests.codes.ok)
217 res = response.json()
219 {'supporting-port': '1', 'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
220 'logical-connection-point': 'XPDR-LINE1'},
223 def test_11_xpdr_portmapping_CLIENT1(self):
224 url = ("{}/config/portmapping:network/"
225 "nodes/XPDRA/mapping/XPDR-CLNT1"
226 .format(self.restconf_baseurl))
227 headers = {'content-type': 'application/json'}
228 response = requests.request(
229 "GET", url, headers=headers, auth=('admin', 'admin'))
230 self.assertEqual(response.status_code, requests.codes.ok)
231 res = response.json()
233 {'supporting-port': '1',
234 'supporting-circuit-pack-name': '1/0/2-PLUG-CLIENT',
235 'logical-connection-point': 'XPDR-CLNT1'},
238 def test_12_xpdr_device_disconnected(self):
239 url = ("{}/config/network-topology:"
240 "network-topology/topology/topology-netconf/node/XPDRA"
241 .format(self.restconf_baseurl))
242 headers = {'content-type': 'application/json'}
243 response = requests.request(
244 "DELETE", url, headers=headers,
245 auth=('admin', 'admin'))
246 self.assertEqual(response.status_code, requests.codes.ok)
249 def test_13_xpdr_device_disconnected(self):
250 url = ("{}/operational/network-topology:network-topology/topology/"
251 "topology-netconf/node/XPDRA".format(self.restconf_baseurl))
252 headers = {'content-type': 'application/json'}
253 response = requests.request(
254 "GET", url, headers=headers, auth=('admin', 'admin'))
255 self.assertEqual(response.status_code, requests.codes.not_found)
256 res = response.json()
258 {"error-type":"application","error-tag":"data-missing",
259 "error-message":"Request could not be completed because the relevant data model content does not exist "},
260 res['errors']['error'])
262 def test_14_xpdr_device_disconnected(self):
263 url = ("{}/config/portmapping:network/"
265 .format(self.restconf_baseurl))
266 headers = {'content-type': 'application/json'}
267 response = requests.request(
268 "GET", url, headers=headers, auth=('admin', 'admin'))
269 self.assertEqual(response.status_code, requests.codes.not_found)
270 res = response.json()
272 {"error-type":"application","error-tag":"data-missing",
273 "error-message":"Request could not be completed because the relevant data model content does not exist "},
274 res['errors']['error'])
276 def test_15_rdm_device_disconnected(self):
277 url = ("{}/config/network-topology:"
278 "network-topology/topology/topology-netconf/node/ROADMA"
279 .format(self.restconf_baseurl))
280 headers = {'content-type': 'application/json'}
281 response = requests.request(
282 "DELETE", url, headers=headers,
283 auth=('admin', 'admin'))
284 self.assertEqual(response.status_code, requests.codes.ok)
287 def test_16_rdm_device_disconnected(self):
288 url = ("{}/operational/network-topology:network-topology/topology/"
289 "topology-netconf/node/ROADMA".format(self.restconf_baseurl))
290 headers = {'content-type': 'application/json'}
291 response = requests.request(
292 "GET", url, headers=headers, auth=('admin', 'admin'))
293 self.assertEqual(response.status_code, requests.codes.not_found)
294 res = response.json()
296 {"error-type":"application","error-tag":"data-missing",
297 "error-message":"Request could not be completed because the relevant data model content does not exist "},
298 res['errors']['error'])
300 def test_17_rdm_device_disconnected(self):
301 url = ("{}/config/portmapping:network/nodes/ROADMA"
302 .format(self.restconf_baseurl))
303 headers = {'content-type': 'application/json'}
304 response = requests.request(
305 "GET", url, headers=headers, auth=('admin', 'admin'))
306 self.assertEqual(response.status_code, requests.codes.not_found)
307 res = response.json()
309 {"error-type":"application","error-tag":"data-missing",
310 "error-message":"Request could not be completed because the relevant data model content does not exist "},
311 res['errors']['error'])
314 if __name__ == "__main__":
315 unittest.main(verbosity=2)