upgrade portmapping YANG to introduce OTN
[transportpce.git] / tests / transportpce_tests / 2.2.1 / test_renderer_service_path_nominal.py
1 #!/usr/bin/env python
2
3 #############################################################################
4 # Copyright (c) 2017 Orange, Inc. and others.  All rights reserved.
5 #
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 #############################################################################
11
12 import unittest
13 import requests
14 import time
15 import subprocess
16 import signal
17 import json
18 import os
19 import psutil
20 import shutil
21 from unittest.result import failfast
22 import test_utils
23
24
25 class TransportPCERendererTesting(unittest.TestCase):
26
27     honeynode_process1 = None
28     honeynode_process2 = None
29     odl_process = None
30     restconf_baseurl = "http://localhost:8181/restconf"
31
32 #START_IGNORE_XTESTING
33
34     @classmethod
35     def setUpClass(cls):
36         print ("starting honeynode1...")
37         cls.honeynode_process1 = test_utils.start_xpdra_honeynode()
38         time.sleep(20)
39
40         print ("starting honeynode2...")
41         cls.honeynode_process2 = test_utils.start_roadma_honeynode()
42         time.sleep(20)
43         print ("all honeynodes started")
44
45         print ("starting opendaylight...")
46         cls.odl_process = test_utils.start_tpce()
47         time.sleep(60)
48         print ("opendaylight started")
49
50     @classmethod
51     def tearDownClass(cls):
52         for child in psutil.Process(cls.odl_process.pid).children():
53             child.send_signal(signal.SIGINT)
54             child.wait()
55         cls.odl_process.send_signal(signal.SIGINT)
56         cls.odl_process.wait()
57         for child in psutil.Process(cls.honeynode_process1.pid).children():
58             child.send_signal(signal.SIGINT)
59             child.wait()
60         cls.honeynode_process1.send_signal(signal.SIGINT)
61         cls.honeynode_process1.wait()
62         for child in psutil.Process(cls.honeynode_process2.pid).children():
63             child.send_signal(signal.SIGINT)
64             child.wait()
65         cls.honeynode_process2.send_signal(signal.SIGINT)
66         cls.honeynode_process2.wait()
67
68     def setUp(self):
69         print ("execution of {}".format(self.id().split(".")[-1]))
70         time.sleep(10)
71
72 #END_IGNORE_XTESTING
73
74     def test_01_rdm_device_connected(self):
75         url = ("{}/config/network-topology:"
76                "network-topology/topology/topology-netconf/node/ROADM-A1"
77                .format(self.restconf_baseurl))
78         data = {"node": [{
79              "node-id": "ROADM-A1",
80              "netconf-node-topology:username": "admin",
81              "netconf-node-topology:password": "admin",
82              "netconf-node-topology:host": "127.0.0.1",
83              "netconf-node-topology:port": "17841",
84              "netconf-node-topology:tcp-only": "false",
85              "netconf-node-topology:pass-through": {}}]}
86         headers = {'content-type': 'application/json'}
87         response = requests.request(
88              "PUT", url, data=json.dumps(data), headers=headers,
89               auth=('admin', 'admin'))
90         self.assertEqual(response.status_code, requests.codes.created)
91         time.sleep(20)
92
93     def test_02_xpdr_device_connected(self):
94         url = ("{}/config/network-topology:"
95                "network-topology/topology/topology-netconf/node/XPDR-A1"
96               .format(self.restconf_baseurl))
97         data = {"node": [{
98             "node-id": "XPDR-A1",
99             "netconf-node-topology:username": "admin",
100             "netconf-node-topology:password": "admin",
101             "netconf-node-topology:host": "127.0.0.1",
102             "netconf-node-topology:port": "17840",
103             "netconf-node-topology:tcp-only": "false",
104             "netconf-node-topology:pass-through": {}}]}
105         headers = {'content-type': 'application/json'}
106         response = requests.request(
107             "PUT", url, data=json.dumps(data), headers=headers,
108             auth=('admin', 'admin'))
109         self.assertEqual(response.status_code, requests.codes.created)
110         time.sleep(20)
111
112     def test_03_rdm_portmapping(self):
113         url = ("{}/config/transportpce-portmapping:network/"
114                "nodes/ROADM-A1"
115                .format(self.restconf_baseurl))
116         headers = {'content-type': 'application/json'}
117         response = requests.request(
118              "GET", url, headers=headers, auth=('admin', 'admin'))
119         self.assertEqual(response.status_code, requests.codes.ok)
120         res = response.json()
121         self.assertIn(
122              {'supporting-port': 'L1', 'supporting-circuit-pack-name': '1/0',
123               'logical-connection-point': 'DEG1-TTP-TXRX', 'port-direction': 'bidirectional'},
124              res['nodes'][0]['mapping'])
125         self.assertIn(
126              {'supporting-port': 'C3', 'supporting-circuit-pack-name': '3/0',
127               'logical-connection-point': 'SRG1-PP3-TXRX', 'port-direction': 'bidirectional'},
128              res['nodes'][0]['mapping'])
129
130     def test_04_xpdr_portmapping(self):
131         url = ("{}/config/transportpce-portmapping:network/"
132                "nodes/XPDR-A1"
133                .format(self.restconf_baseurl))
134         headers = {'content-type': 'application/json'}
135         response = requests.request(
136              "GET", url, headers=headers, auth=('admin', 'admin'))
137         self.assertEqual(response.status_code, requests.codes.ok)
138         res = response.json()
139         self.assertIn(
140              {'supporting-port': '1', 'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
141               'logical-connection-point': 'XPDR1-NETWORK1', 'port-qual': 'xpdr-network',
142               'port-direction': 'bidirectional', 'connection-map-lcp': 'XPDR1-CLIENT1'},
143              res['nodes'][0]['mapping'])
144         self.assertIn(
145              {'supporting-port': 'C1',
146               'supporting-circuit-pack-name': '1/0/1-PLUG-CLIENT',
147               'logical-connection-point': 'XPDR1-CLIENT1', 'port-direction': 'bidirectional',
148               'connection-map-lcp': 'XPDR1-NETWORK1', 'port-qual': 'xpdr-client'},
149              res['nodes'][0]['mapping'])
150
151     def test_05_service_path_create(self):
152         url = "{}/operations/transportpce-device-renderer:service-path".format(self.restconf_baseurl)
153         data = {"renderer:input": {
154              "renderer:service-name": "service_test",
155              "renderer:wave-number": "7",
156              "renderer:modulation-format": "qpsk",
157              "renderer:operation": "create",
158              "renderer:nodes": [
159                  {"renderer:node-id": "ROADM-A1",
160                   "renderer:src-tp": "SRG1-PP3-TXRX",
161                   "renderer:dest-tp": "DEG1-TTP-TXRX"},
162                  {"renderer:node-id": "XPDR-A1",
163                   "renderer:src-tp": "XPDR1-CLIENT1",
164                   "renderer:dest-tp": "XPDR1-NETWORK1"}]}}
165         headers = {'content-type': 'application/json'}
166         response = requests.request(
167              "POST", url, data=json.dumps(data),
168              headers=headers, auth=('admin', 'admin'))
169         self.assertEqual(response.status_code, requests.codes.ok)
170         res = response.json()
171         self.assertIn('Roadm-connection successfully created for nodes: ROADM-A1', res["output"]["result"])
172
173     def test_06_service_path_create_rdm_check(self):
174         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
175                 "node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
176                 "interface/DEG1-TTP-TXRX-nmc-7"
177                 .format(self.restconf_baseurl))
178         headers = {'content-type': 'application/json'}
179         response = requests.request(
180              "GET", url, headers=headers, auth=('admin', 'admin'))
181         self.assertEqual(response.status_code, requests.codes.ok)
182         res = response.json()
183         self.assertDictContainsSubset({'name': 'DEG1-TTP-TXRX-nmc-7', 'administrative-state': 'inService',
184               'supporting-circuit-pack-name': '1/0',
185               'type': 'org-openroadm-interfaces:networkMediaChannelConnectionTerminationPoint',
186               'supporting-port': 'L1'}, res['interface'][0])
187         self.assertDictEqual(
188              {u'frequency': 195.8, u'width': 40},
189              res['interface'][0]['org-openroadm-network-media-channel-interfaces:nmc-ctp'])
190
191     def test_07_service_path_create_rdm_check(self):
192         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
193                 "node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
194                 "interface/DEG1-TTP-TXRX-mc-7"
195                 .format(self.restconf_baseurl))
196         headers = {'content-type': 'application/json'}
197         response = requests.request(
198              "GET", url, headers=headers, auth=('admin', 'admin'))
199         self.assertEqual(response.status_code, requests.codes.ok)
200         res = response.json()
201         self.assertDictContainsSubset({'name': 'DEG1-TTP-TXRX-mc-7', 'administrative-state': 'inService',
202               'supporting-circuit-pack-name': '1/0',
203               'type': 'org-openroadm-interfaces:mediaChannelTrailTerminationPoint',
204               'supporting-port': 'L1'}, res['interface'][0])
205         self.assertDictEqual(
206              {u'min-freq': 195.775, u'max-freq': 195.825},
207              res['interface'][0]['org-openroadm-media-channel-interfaces:mc-ttp'])
208
209
210     def test_08_service_path_create_rdm_check(self):
211         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
212                 "node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
213                 "interface/SRG1-PP3-TXRX-nmc-7"
214                 .format(self.restconf_baseurl))
215         headers = {'content-type': 'application/json'}
216         response = requests.request(
217              "GET", url, headers=headers, auth=('admin', 'admin'))
218         self.assertEqual(response.status_code, requests.codes.ok)
219         res = response.json()
220         self.assertDictContainsSubset({'name': 'SRG1-PP3-TXRX-nmc-7', 'administrative-state': 'inService',
221               'supporting-circuit-pack-name': '3/0',
222               'type': 'org-openroadm-interfaces:networkMediaChannelConnectionTerminationPoint',
223               'supporting-port': 'C3'}, res['interface'][0])
224         self.assertDictEqual(
225              {u'frequency': 195.8, u'width': 40},
226              res['interface'][0]['org-openroadm-network-media-channel-interfaces:nmc-ctp'])
227
228     # -mc supporting interfaces must not be created for SRG, only degrees
229     def test_09_service_path_create_rdm_check(self):
230         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
231                 "node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
232                 "interface/SRG1-PP3-TXRX-mc-7"
233                 .format(self.restconf_baseurl))
234         headers = {'content-type': 'application/json'}
235         response = requests.request(
236              "GET", url, headers=headers, auth=('admin', 'admin'))
237         self.assertEqual(response.status_code, requests.codes.not_found)
238         res = response.json()
239         self.assertIn(
240              {"error-type":"application", "error-tag":"data-missing",
241               "error-message":"Request could not be completed because the relevant data model content does not exist"},
242              res['errors']['error'])
243
244     def test_10_service_path_create_rdm_check(self):
245         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
246                "node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
247                "roadm-connections/SRG1-PP3-TXRX-DEG1-TTP-TXRX-7"
248                .format(self.restconf_baseurl))
249         headers = {'content-type': 'application/json'}
250         response = requests.request(
251              "GET", url, headers=headers, auth=('admin', 'admin'))
252         self.assertEqual(response.status_code, requests.codes.ok)
253         res = response.json()
254         self.assertDictContainsSubset(
255              {'connection-name': 'SRG1-PP3-TXRX-DEG1-TTP-TXRX-7',
256               'opticalControlMode': 'off'},
257              res['roadm-connections'][0])
258         self.assertDictEqual(
259              {'src-if': 'SRG1-PP3-TXRX-nmc-7'},
260              res['roadm-connections'][0]['source'])
261         self.assertDictEqual(
262              {'dst-if': 'DEG1-TTP-TXRX-nmc-7'},
263              res['roadm-connections'][0]['destination'])
264
265     def test_11_service_path_create_xpdr_check(self):
266         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
267                 "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
268                 "interface/XPDR1-NETWORK1-7"
269                 .format(self.restconf_baseurl))
270         headers = {'content-type': 'application/json'}
271         response = requests.request(
272              "GET", url, headers=headers, auth=('admin', 'admin'))
273         self.assertEqual(response.status_code, requests.codes.ok)
274         res = response.json()
275         self.assertDictContainsSubset(
276              {'name': 'XPDR1-NETWORK1-7', 'administrative-state': 'inService',
277               'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
278               'type': 'org-openroadm-interfaces:opticalChannel',
279               'supporting-port': '1'},
280              res['interface'][0])
281         self.assertDictEqual(
282              {u'rate': u'org-openroadm-common-types:R100G',
283               u'transmit-power':-5,
284               u'frequency': 195.8},
285              res['interface'][0]['org-openroadm-optical-channel-interfaces:och'])
286
287     def test_12_service_path_create_xpdr_check(self):
288         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
289                 "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
290                 "interface/XPDR1-NETWORK1-OTU"
291                 .format(self.restconf_baseurl))
292         headers = {'content-type': 'application/json'}
293         response = requests.request(
294              "GET", url, headers=headers, auth=('admin', 'admin'))
295         self.assertEqual(response.status_code, requests.codes.ok)
296         res = response.json()
297         self.assertDictContainsSubset(
298              {'name': 'XPDR1-NETWORK1-OTU', 'administrative-state': 'inService',
299               'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
300               'type': 'org-openroadm-interfaces:otnOtu',
301               'supporting-port': '1',
302               'supporting-interface': 'XPDR1-NETWORK1-7'},
303              res['interface'][0])
304         self.assertDictEqual(
305              {u'rate': u'org-openroadm-otn-common-types:OTU4',
306               u'fec': u'scfec'},
307              res['interface'][0]['org-openroadm-otn-otu-interfaces:otu'])
308
309     def test_13_service_path_create_xpdr_check(self):
310         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
311                 "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
312                 "interface/XPDR1-NETWORK1-ODU"
313                 .format(self.restconf_baseurl))
314         headers = {'content-type': 'application/json'}
315         response = requests.request(
316              "GET", url, headers=headers, auth=('admin', 'admin'))
317         self.assertEqual(response.status_code, requests.codes.ok)
318         res = response.json()
319         self.assertDictContainsSubset(
320              {'name': 'XPDR1-NETWORK1-ODU', 'administrative-state': 'inService',
321               'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
322               'type': 'org-openroadm-interfaces:otnOdu',
323               'supporting-port': '1',
324               'supporting-interface': 'XPDR1-NETWORK1-OTU'},
325              res['interface'][0])
326         self.assertDictContainsSubset(
327              {'rate': 'org-openroadm-otn-common-types:ODU4',
328               u'monitoring-mode': u'terminated'},
329              res['interface'][0]['org-openroadm-otn-odu-interfaces:odu'])
330         self.assertDictEqual({u'exp-payload-type': u'07', u'payload-type': u'07'}, 
331                              res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']['opu'])
332
333     def test_14_service_path_create_xpdr_check(self):
334         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
335                "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
336                "interface/XPDR1-CLIENT1-ETHERNET"
337                .format(self.restconf_baseurl))
338         headers = {'content-type': 'application/json'}
339         response = requests.request(
340              "GET", url, headers=headers, auth=('admin', 'admin'))
341         self.assertEqual(response.status_code, requests.codes.ok)
342         res = response.json()
343         self.assertDictContainsSubset(
344              {'name': 'XPDR1-CLIENT1-ETHERNET', 'administrative-state': 'inService',
345               'supporting-circuit-pack-name': '1/0/1-PLUG-CLIENT',
346               'type': 'org-openroadm-interfaces:ethernetCsmacd',
347               'supporting-port': 'C1'},
348              res['interface'][0])
349         self.assertDictEqual(
350              {u'fec': u'off', u'speed': 100000},
351              res['interface'][0]['org-openroadm-ethernet-interfaces:ethernet'])
352
353     def test_15_service_path_create_xpdr_check(self):
354         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
355                 "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
356                 "circuit-packs/1%2F0%2F1-PLUG-NET"
357                 .format(self.restconf_baseurl))
358         headers = {'content-type': 'application/json'}
359         response = requests.request(
360              "GET", url, headers=headers, auth=('admin', 'admin'))
361         self.assertEqual(response.status_code, requests.codes.ok)
362         res = response.json()
363         self.assertIn('not-reserved-inuse', res['circuit-packs'][0]["equipment-state"])
364
365     def test_16_service_path_create_xpdr_check(self):
366         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
367                 "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
368                 "circuit-packs/1%2F0%2F1-PLUG-CLIENT"
369                 .format(self.restconf_baseurl))
370         headers = {'content-type': 'application/json'}
371         response = requests.request(
372              "GET", url, headers=headers, auth=('admin', 'admin'))
373         self.assertEqual(response.status_code, requests.codes.ok)
374         res = response.json()
375         self.assertIn('not-reserved-inuse', res['circuit-packs'][0]["equipment-state"])
376
377     def test_17_service_path_delete(self):
378         url = "{}/operations/transportpce-device-renderer:service-path".format(self.restconf_baseurl)
379         data = {"renderer:input": {
380              "renderer:service-name": "service_test",
381              "renderer:wave-number": "7",
382              "renderer:operation": "delete",
383              "renderer:nodes": [
384                  {"renderer:node-id": "ROADM-A1",
385                   "renderer:src-tp": "SRG1-PP3-TXRX",
386                   "renderer:dest-tp": "DEG1-TTP-TXRX"},
387                  {"renderer:node-id": "XPDR-A1",
388                   "renderer:src-tp": "XPDR1-CLIENT1",
389                   "renderer:dest-tp": "XPDR1-NETWORK1"}]}}
390         headers = {'content-type': 'application/json'}
391         response = requests.request(
392              "POST", url, data=json.dumps(data),
393              headers=headers, auth=('admin', 'admin'))
394         self.assertEqual(response.status_code, requests.codes.ok)
395         self.assertEqual(response.json(), {
396              'output': {'result': 'Request processed', 'success': True}})
397
398     def test_18_service_path_delete_rdm_check(self):
399         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
400                 "node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
401                 "interface/DEG1-TTP-TXRX-mc-7"
402                 .format(self.restconf_baseurl))
403         headers = {'content-type': 'application/json'}
404         response = requests.request(
405              "GET", url, headers=headers, auth=('admin', 'admin'))
406         self.assertEqual(response.status_code, requests.codes.not_found)
407         res = response.json()
408         self.assertIn(
409              {"error-type":"application", "error-tag":"data-missing",
410               "error-message":"Request could not be completed because the relevant data model content does not exist"},
411              res['errors']['error'])
412
413     def test_19_service_path_delete_rdm_check(self):
414         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
415                 "node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
416                 "interface/DEG1-TTP-TXRX-nmc-7"
417                 .format(self.restconf_baseurl))
418         headers = {'content-type': 'application/json'}
419         response = requests.request(
420              "GET", url, headers=headers, auth=('admin', 'admin'))
421         self.assertEqual(response.status_code, requests.codes.not_found)
422         res = response.json()
423         self.assertIn(
424              {"error-type":"application", "error-tag":"data-missing",
425               "error-message":"Request could not be completed because the relevant data model content does not exist"},
426              res['errors']['error'])
427
428     def test_20_service_path_delete_rdm_check(self):
429         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
430                "node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
431                "interface/SRG1-PP3-TXRX-mc-7"
432                .format(self.restconf_baseurl))
433         headers = {'content-type': 'application/json'}
434         response = requests.request(
435              "GET", url, headers=headers, auth=('admin', 'admin'))
436         self.assertEqual(response.status_code, requests.codes.not_found)
437         res = response.json()
438         self.assertIn(
439              {"error-type":"application", "error-tag":"data-missing", "error-message":"Request could not be completed because the relevant data model content does not exist"},
440              res['errors']['error'])
441
442     def test_21_service_path_delete_rdm_check(self):
443         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
444                "node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
445                "interface/SRG1-PP3-TXRX-nmc-7"
446                .format(self.restconf_baseurl))
447         headers = {'content-type': 'application/json'}
448         response = requests.request(
449              "GET", url, headers=headers, auth=('admin', 'admin'))
450         self.assertEqual(response.status_code, requests.codes.not_found)
451         res = response.json()
452         self.assertIn(
453              {"error-type":"application", "error-tag":"data-missing", "error-message":"Request could not be completed because the relevant data model content does not exist"},
454              res['errors']['error'])
455
456     def test_22_service_path_delete_rdm_check(self):
457         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
458                "node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
459                "roadm-connections/SRG1-PP3-TXRX-DEG1-TTP-TXRX-7"
460                .format(self.restconf_baseurl))
461         headers = {'content-type': 'application/json'}
462         response = requests.request(
463              "GET", url, headers=headers, auth=('admin', 'admin'))
464         self.assertEqual(response.status_code, requests.codes.not_found)
465         res = response.json()
466         self.assertIn(
467              {"error-type":"application", "error-tag":"data-missing", "error-message":"Request could not be completed because the relevant data model content does not exist"},
468              res['errors']['error'])
469
470     def test_23_service_path_delete_xpdr_check(self):
471         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
472                "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
473                "interface/XPDR1-NETWORK1-7"
474                .format(self.restconf_baseurl))
475         headers = {'content-type': 'application/json'}
476         response = requests.request(
477              "GET", url, headers=headers, auth=('admin', 'admin'))
478         self.assertEqual(response.status_code, requests.codes.not_found)
479         res = response.json()
480         self.assertIn(
481              {"error-type":"application", "error-tag":"data-missing", "error-message":"Request could not be completed because the relevant data model content does not exist"},
482              res['errors']['error'])
483
484     def test_24_service_path_delete_xpdr_check(self):
485         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
486                "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
487                "interface/XPDR1-NETWORK1-OTU"
488                .format(self.restconf_baseurl))
489         headers = {'content-type': 'application/json'}
490         response = requests.request(
491              "GET", url, headers=headers, auth=('admin', 'admin'))
492         self.assertEqual(response.status_code, requests.codes.not_found)
493         res = response.json()
494         self.assertIn(
495              {"error-type":"application", "error-tag":"data-missing", "error-message":"Request could not be completed because the relevant data model content does not exist"},
496              res['errors']['error'])
497
498     def test_25_service_path_delete_xpdr_check(self):
499         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
500                "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
501                "interface/XPDR1-NETWORK1-ODU"
502                .format(self.restconf_baseurl))
503         headers = {'content-type': 'application/json'}
504         response = requests.request(
505              "GET", url, headers=headers, auth=('admin', 'admin'))
506         self.assertEqual(response.status_code, requests.codes.not_found)
507         res = response.json()
508         self.assertIn(
509              {"error-type":"application", "error-tag":"data-missing", "error-message":"Request could not be completed because the relevant data model content does not exist"},
510              res['errors']['error'])
511
512     def test_26_service_path_delete_xpdr_check(self):
513         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
514                "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
515                "interface/XPDR1-CLIENT1-ETHERNET"
516                .format(self.restconf_baseurl))
517         headers = {'content-type': 'application/json'}
518         response = requests.request(
519              "GET", url, headers=headers, auth=('admin', 'admin'))
520         self.assertEqual(response.status_code, requests.codes.not_found)
521         res = response.json()
522         self.assertIn(
523              {"error-type":"application", "error-tag":"data-missing", "error-message":"Request could not be completed because the relevant data model content does not exist"},
524              res['errors']['error'])
525
526     def test_27_service_path_delete_xpdr_check(self):
527         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
528                "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
529                "circuit-packs/1%2F0%2F1-PLUG-NET"
530                .format(self.restconf_baseurl))
531         headers = {'content-type': 'application/json'}
532         response = requests.request(
533             "GET", url, headers=headers, auth=('admin', 'admin'))
534         self.assertEqual(response.status_code, requests.codes.ok)
535         res = response.json()
536         self.assertEqual('not-reserved-available', res["circuit-packs"][0]['equipment-state'])
537
538     def test_28_service_path_delete_xpdr_check(self):
539         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
540                "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
541                "circuit-packs/1%2F0%2F1-PLUG-CLIENT"
542                .format(self.restconf_baseurl))
543         headers = {'content-type': 'application/json'}
544         response = requests.request(
545             "GET", url, headers=headers, auth=('admin', 'admin'))
546         self.assertEqual(response.status_code, requests.codes.ok)
547         res = response.json()
548         self.assertEqual('not-reserved-available', res["circuit-packs"][0]['equipment-state'])
549
550     def test_29_rdm_device_disconnected(self):
551         url = ("{}/config/network-topology:"
552                "network-topology/topology/topology-netconf/node/ROADM-A1"
553                .format(self.restconf_baseurl))
554         headers = {'content-type': 'application/json'}
555         response = requests.request(
556               "DELETE", url, headers=headers,
557               auth=('admin', 'admin'))
558         self.assertEqual(response.status_code, requests.codes.ok)
559         time.sleep(20)
560
561     def test_30_xpdr_device_disconnected(self):
562         url = ("{}/config/network-topology:"
563                 "network-topology/topology/topology-netconf/node/XPDR-A1"
564                .format(self.restconf_baseurl))
565         headers = {'content-type': 'application/json'}
566         response = requests.request(
567              "DELETE", url, headers=headers,
568              auth=('admin', 'admin'))
569         self.assertEqual(response.status_code, requests.codes.ok)
570         time.sleep(20)
571
572
573 if __name__ == "__main__":
574     unittest.main(verbosity=2, failfast=True)