Update topology functional tests
[transportpce.git] / tests / transportpce_tests / 1.2.1 / test_topology.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 json
13 import os
14 import psutil
15 import requests
16 import signal
17 import shutil
18 import subprocess
19 import time
20 import unittest
21 import logging
22
23 class TransportPCETopologyTesting(unittest.TestCase):
24
25     honeynode_process1 = None
26     honeynode_process2 = None
27     honeynode_process3 = None
28     honeynode_process4 = None
29     odl_process = None
30     restconf_baseurl = "http://localhost:8181/restconf"
31
32 #START_IGNORE_XTESTING
33
34     @classmethod
35     def __start_honeynode1(cls):
36         executable = ("./honeynode/2.1/honeynode-distribution/target/honeynode-distribution-1.18.01-hc"
37                       "/honeynode-distribution-1.18.01/honeycomb-tpce")
38         if os.path.isfile(executable):
39             with open('honeynode1.log', 'w') as outfile:
40                 cls.honeynode_process1 = subprocess.Popen(
41                     [executable, "17830", "sample_configs/openroadm/2.1/oper-XPDRA.xml"],
42                     stdout=outfile)
43
44     @classmethod
45     def __start_honeynode2(cls):
46         executable = ("./honeynode/2.1/honeynode-distribution/target/honeynode-distribution-1.18.01-hc"
47                       "/honeynode-distribution-1.18.01/honeycomb-tpce")
48         if os.path.isfile(executable):
49             with open('honeynode2.log', 'w') as outfile:
50                 cls.honeynode_process2 = subprocess.Popen(
51                     [executable, "17831", "sample_configs/openroadm/2.1/oper-ROADMA.xml"],
52                     stdout=outfile)
53
54     @classmethod
55     def __start_honeynode3(cls):
56         executable = ("./honeynode/2.1/honeynode-distribution/target/honeynode-distribution-1.18.01-hc"
57                       "/honeynode-distribution-1.18.01/honeycomb-tpce")
58         if os.path.isfile(executable):
59             with open('honeynode3.log', 'w') as outfile:
60                 cls.honeynode_process3 = subprocess.Popen(
61                     [executable, "17832", "sample_configs/openroadm/2.1/oper-ROADMB.xml"],
62                     stdout=outfile)
63
64     @classmethod
65     def __start_honeynode4(cls):
66         executable = ("./honeynode/2.1/honeynode-distribution/target/honeynode-distribution-1.18.01-hc"
67                       "/honeynode-distribution-1.18.01/honeycomb-tpce")
68         if os.path.isfile(executable):
69             with open('honeynode4.log', 'w') as outfile:
70                 cls.honeynode_process4 = subprocess.Popen(
71                     [executable, "17833", "sample_configs/openroadm/2.1/oper-ROADMC.xml"],
72                     stdout=outfile)
73
74     @classmethod
75     def __start_odl(cls):
76         executable = "../karaf/target/assembly/bin/karaf"
77         with open('odl.log', 'w') as outfile:
78             cls.odl_process = subprocess.Popen(
79                 ["bash", executable, "server"], stdout=outfile,
80                 stdin=open(os.devnull))
81
82     @classmethod
83     def setUpClass(cls):
84         cls.__start_honeynode1()
85         time.sleep(20)
86         cls.__start_honeynode2()
87         time.sleep(20)
88         cls.__start_honeynode3()
89         time.sleep(20)
90         cls.__start_honeynode4()
91         time.sleep(20)
92         cls.__start_odl()
93         time.sleep(60)
94
95     @classmethod
96     def tearDownClass(cls):
97         for child in psutil.Process(cls.odl_process.pid).children():
98             child.send_signal(signal.SIGINT)
99             child.wait()
100         cls.odl_process.send_signal(signal.SIGINT)
101         cls.odl_process.wait()
102         for child in psutil.Process(cls.honeynode_process1.pid).children():
103             child.send_signal(signal.SIGINT)
104             child.wait()
105         cls.honeynode_process1.send_signal(signal.SIGINT)
106         cls.honeynode_process1.wait()
107         for child in psutil.Process(cls.honeynode_process2.pid).children():
108             child.send_signal(signal.SIGINT)
109             child.wait()
110         cls.honeynode_process2.send_signal(signal.SIGINT)
111         cls.honeynode_process2.wait()
112
113         for child in psutil.Process(cls.honeynode_process3.pid).children():
114             child.send_signal(signal.SIGINT)
115             child.wait()
116         cls.honeynode_process3.send_signal(signal.SIGINT)
117         cls.honeynode_process3.wait()
118
119         for child in psutil.Process(cls.honeynode_process4.pid).children():
120             child.send_signal(signal.SIGINT)
121             child.wait()
122         cls.honeynode_process4.send_signal(signal.SIGINT)
123         cls.honeynode_process4.wait()
124
125     def setUp(self):
126         time.sleep(30)
127
128 #END_IGNORE_XTESTING
129
130     def test_01_connect_ROADMA(self):
131         #Config ROADMA
132         url = ("{}/config/network-topology:"
133                 "network-topology/topology/topology-netconf/node/ROADMA"
134                .format(self.restconf_baseurl))
135         data = {"node": [{
136              "node-id": "ROADMA",
137              "netconf-node-topology:username": "admin",
138              "netconf-node-topology:password": "admin",
139              "netconf-node-topology:host": "127.0.0.1",
140              "netconf-node-topology:port": "17831",
141              "netconf-node-topology:tcp-only": "false",
142              "netconf-node-topology:pass-through": {}}]}
143         headers = {'content-type': 'application/json'}
144         response = requests.request(
145              "PUT", url, data=json.dumps(data), headers=headers,
146              auth=('admin', 'admin'))
147         self.assertEqual(response.status_code, requests.codes.created)
148         time.sleep(20)
149
150     def test_02_getClliNetwork(self):
151         url = ("{}/config/ietf-network:networks/network/clli-network"
152               .format(self.restconf_baseurl))
153         headers = {'content-type': 'application/json'}
154         response = requests.request(
155             "GET", url, headers=headers, auth=('admin', 'admin'))
156         self.assertEqual(response.status_code, requests.codes.ok)
157         res = response.json()
158         self.assertEqual(res['network'][0]['node'][0]['node-id'],'NodeA')
159         self.assertEqual(res['network'][0]['node'][0]['org-openroadm-clli-network:clli'],'NodeA')
160
161     def test_03_getOpenRoadmNetwork(self):
162         url = ("{}/config/ietf-network:networks/network/openroadm-network"
163               .format(self.restconf_baseurl))
164         headers = {'content-type': 'application/json'}
165         response = requests.request(
166             "GET", url, headers=headers, auth=('admin', 'admin'))
167         self.assertEqual(response.status_code, requests.codes.ok)
168         res = response.json()
169         self.assertEqual(res['network'][0]['node'][0]['node-id'],'ROADMA')
170         self.assertEqual(res['network'][0]['node'][0]['supporting-node'][0]['network-ref'],'clli-network')
171         self.assertEqual(res['network'][0]['node'][0]['supporting-node'][0]['node-ref'],'NodeA')
172         self.assertEqual(res['network'][0]['node'][0]['org-openroadm-network:node-type'],'ROADM')
173         self.assertEqual(res['network'][0]['node'][0]['org-openroadm-network:model'],'2')
174
175     def test_04_getLinks_OpenroadmTopology(self):
176         url = ("{}/config/ietf-network:networks/network/openroadm-topology"
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         #Tests related to links
184         nbLink=len(res['network'][0]['ietf-network-topology:link'])
185         self.assertEqual(nbLink,10)
186         expressLink=['ROADMA-DEG2-DEG2-CTP-TXRXtoROADMA-DEG1-DEG1-CTP-TXRX','ROADMA-DEG1-DEG1-CTP-TXRXtoROADMA-DEG2-DEG2-CTP-TXRX']
187         addLink=['ROADMA-SRG1-SRG1-CP-TXRXtoROADMA-DEG2-DEG2-CTP-TXRX','ROADMA-SRG1-SRG1-CP-TXRXtoROADMA-DEG1-DEG1-CTP-TXRX',
188                  'ROADMA-SRG3-SRG3-CP-TXRXtoROADMA-DEG2-DEG2-CTP-TXRX','ROADMA-SRG3-SRG3-CP-TXRXtoROADMA-DEG1-DEG1-CTP-TXRX']
189         dropLink=['ROADMA-DEG1-DEG1-CTP-TXRXtoROADMA-SRG1-SRG1-CP-TXRX','ROADMA-DEG2-DEG2-CTP-TXRXtoROADMA-SRG1-SRG1-CP-TXRX',
190                   'ROADMA-DEG1-DEG1-CTP-TXRXtoROADMA-SRG3-SRG3-CP-TXRX','ROADMA-DEG2-DEG2-CTP-TXRXtoROADMA-SRG3-SRG3-CP-TXRX']
191         for i in range(0,nbLink):
192             linkId = res['network'][0]['ietf-network-topology:link'][i]['link-id']
193             if (res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-network-topology:link-type']=='EXPRESS-LINK'):
194                 find= linkId in expressLink
195                 self.assertEqual(find, True)
196                 expressLink.remove(linkId)
197             elif (res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-network-topology:link-type']=='ADD-LINK'):
198                 find= linkId in addLink
199                 self.assertEqual(find, True)
200                 addLink.remove(linkId)
201             elif (res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-network-topology:link-type']=='DROP-LINK'):
202                 find= linkId in dropLink
203                 self.assertEqual(find, True)
204                 dropLink.remove(linkId)
205             else:
206                 self.assertFalse(True)
207         self.assertEqual(len(expressLink),0)
208         self.assertEqual(len(addLink),0)
209         self.assertEqual(len(dropLink),0)
210
211     def test_05_getNodes_OpenRoadmTopology(self):
212         url = ("{}/config/ietf-network:networks/network/openroadm-topology"
213               .format(self.restconf_baseurl))
214         headers = {'content-type': 'application/json'}
215         response = requests.request(
216             "GET", url, headers=headers, auth=('admin', 'admin'))
217         res = response.json()
218         #Tests related to nodes
219         self.assertEqual(response.status_code, requests.codes.ok)
220         nbNode=len(res['network'][0]['node'])
221         self.assertEqual(nbNode,4)
222         listNode=['ROADMA-SRG1','ROADMA-SRG3','ROADMA-DEG1','ROADMA-DEG2']
223         for i in range(0,nbNode):
224             self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'ROADMA'},
225                           res['network'][0]['node'][i]['supporting-node'])
226             nodeType=res['network'][0]['node'][i]['org-openroadm-network-topology:node-type']
227             nodeId=res['network'][0]['node'][i]['node-id']
228             if(nodeId=='ROADMA-SRG1'):
229                 #Test related to SRG1
230                 self.assertEqual(nodeType,'SRG')
231                 self.assertEqual(len(res['network'][0]['node'][i]['ietf-network-topology:termination-point']),17)
232                 self.assertIn({'tp-id': 'SRG1-CP-TXRX', 'org-openroadm-network-topology:tp-type': 'SRG-TXRX-CP'},
233                               res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
234                 self.assertIn({'tp-id': 'SRG1-PP1-TXRX', 'org-openroadm-network-topology:tp-type': 'SRG-TXRX-PP'},
235                               res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
236                 listNode.remove(nodeId)
237             elif(nodeId=='ROADMA-SRG3'):
238                 #Test related to SRG1
239                 self.assertEqual(nodeType,'SRG')
240                 self.assertEqual(len(res['network'][0]['node'][i]['ietf-network-topology:termination-point']),17)
241                 self.assertIn({'tp-id': 'SRG3-CP-TXRX', 'org-openroadm-network-topology:tp-type': 'SRG-TXRX-CP'},
242                               res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
243                 self.assertIn({'tp-id': 'SRG3-PP1-TXRX', 'org-openroadm-network-topology:tp-type': 'SRG-TXRX-PP'},
244                               res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
245                 listNode.remove(nodeId)
246             elif(nodeId=='ROADMA-DEG1'):
247                 #Test related to DEG1
248                 self.assertEqual(nodeType,'DEGREE')
249                 self.assertIn({'tp-id': 'DEG1-TTP-TXRX', 'org-openroadm-network-topology:tp-type': 'DEGREE-TXRX-TTP'},
250                               res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
251                 self.assertIn({'tp-id': 'DEG1-CTP-TXRX', 'org-openroadm-network-topology:tp-type': 'DEGREE-TXRX-CTP'},
252                               res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
253                 listNode.remove(nodeId)
254             elif(nodeId=='ROADMA-DEG2'):
255                 #Test related to DEG2
256                 self.assertEqual(nodeType,'DEGREE')
257                 self.assertIn({'tp-id': 'DEG2-TTP-TXRX', 'org-openroadm-network-topology:tp-type': 'DEGREE-TXRX-TTP'},
258                               res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
259                 self.assertIn({'tp-id': 'DEG2-CTP-TXRX', 'org-openroadm-network-topology:tp-type': 'DEGREE-TXRX-CTP'},
260                               res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
261                 listNode.remove(nodeId)
262             else:
263                 self.assertFalse(True)
264         self.assertEqual(len(listNode),0)
265
266     def test_06_connect_XPDRA(self):
267          url = ("{}/config/network-topology:"
268                  "network-topology/topology/topology-netconf/node/XPDRA"
269                 .format(self.restconf_baseurl))
270          data = {"node": [{
271               "node-id": "XPDRA",
272               "netconf-node-topology:username": "admin",
273               "netconf-node-topology:password": "admin",
274               "netconf-node-topology:host": "127.0.0.1",
275               "netconf-node-topology:port": "17830",
276               "netconf-node-topology:tcp-only": "false",
277               "netconf-node-topology:pass-through": {}}]}
278          headers = {'content-type': 'application/json'}
279          response = requests.request(
280               "PUT", url, data=json.dumps(data), headers=headers,
281               auth=('admin', 'admin'))
282          self.assertEqual(response.status_code, requests.codes.created)
283          time.sleep(30)
284
285     def test_07_getClliNetwork(self):
286          url = ("{}/config/ietf-network:networks/network/clli-network"
287                .format(self.restconf_baseurl))
288          headers = {'content-type': 'application/json'}
289          response = requests.request(
290              "GET", url, headers=headers, auth=('admin', 'admin'))
291          self.assertEqual(response.status_code, requests.codes.ok)
292          res = response.json()
293          self.assertEqual(res['network'][0]['node'][0]['node-id'],'NodeA')
294          self.assertEqual(res['network'][0]['node'][0]['org-openroadm-clli-network:clli'],'NodeA')
295
296     def test_08_getOpenRoadmNetwork(self):
297          url = ("{}/config/ietf-network:networks/network/openroadm-network"
298                .format(self.restconf_baseurl))
299          headers = {'content-type': 'application/json'}
300          response = requests.request(
301              "GET", url, headers=headers, auth=('admin', 'admin'))
302          self.assertEqual(response.status_code, requests.codes.ok)
303          res = response.json()
304          nbNode=len(res['network'][0]['node'])
305          self.assertEqual(nbNode,2)
306          for i in range(0,nbNode):
307              self.assertEqual(res['network'][0]['node'][i]['supporting-node'][0]['network-ref'],'clli-network')
308              self.assertEqual(res['network'][0]['node'][i]['supporting-node'][0]['node-ref'],'NodeA')
309              nodeId=res['network'][0]['node'][i]['node-id']
310              if(nodeId=='XPDRA'):
311                 self.assertEqual(res['network'][0]['node'][i]['org-openroadm-network:node-type'],'XPONDER')
312                 self.assertEqual(res['network'][0]['node'][i]['org-openroadm-network:model'],'1')
313              elif(nodeId=='ROADMA'):
314                 self.assertEqual(res['network'][0]['node'][i]['org-openroadm-network:node-type'],'ROADM')
315                 self.assertEqual(res['network'][0]['node'][i]['org-openroadm-network:model'],'2')
316              else:
317                 self.assertFalse(True)
318
319     def test_09_getNodes_OpenRoadmTopology(self):
320          url = ("{}/config/ietf-network:networks/network/openroadm-topology"
321                .format(self.restconf_baseurl))
322          headers = {'content-type': 'application/json'}
323          response = requests.request(
324              "GET", url, headers=headers, auth=('admin', 'admin'))
325          res = response.json()
326          #Tests related to nodes
327          self.assertEqual(response.status_code, requests.codes.ok)
328          nbNode=len(res['network'][0]['node'])
329          self.assertEqual(nbNode,5)
330          listNode=['XPDRA-XPDR1','ROADMA-SRG1','ROADMA-SRG3','ROADMA-DEG1','ROADMA-DEG2']
331          for i in range(0,nbNode):
332              nodeType=res['network'][0]['node'][i]['org-openroadm-network-topology:node-type']
333              nodeId=res['network'][0]['node'][i]['node-id']
334              #Tests related to XPDRA nodes
335              if(nodeId=='XPDRA-XPDR1'):
336                  self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'XPDRA'},
337                                res['network'][0]['node'][i]['supporting-node'])
338                  self.assertEqual(nodeType,'XPONDER')
339                  nbTps=len(res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
340                  client = 0
341                  network = 0
342                  for j in range(0,nbTps):
343                      tpType=res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['org-openroadm-network-topology:tp-type']
344                      tpId=res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['tp-id']
345                      if (tpType=='XPONDER-CLIENT'):
346                          client += 1
347                      elif (tpType=='XPONDER-NETWORK'):
348                          network += 1
349                      if (tpId == 'XPDR1-NETWORK2'):
350                          self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['transportpce-topology:associated-connection-map-port'],'XPDR1-CLIENT3')
351                      if (tpId == 'XPDR1-CLIENT3'):
352                          self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['transportpce-topology:associated-connection-map-port'],'XPDR1-NETWORK2')
353                  self.assertTrue(client == 4)
354                  self.assertTrue(network == 2)
355                  listNode.remove(nodeId)
356              elif(nodeId=='ROADMA-SRG1'):
357                  #Test related to SRG1
358                  self.assertEqual(nodeType,'SRG')
359                  self.assertEqual(len(res['network'][0]['node'][i]['ietf-network-topology:termination-point']),17)
360                  self.assertIn({'tp-id': 'SRG1-CP-TXRX', 'org-openroadm-network-topology:tp-type': 'SRG-TXRX-CP'},
361                                res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
362                  self.assertIn({'tp-id': 'SRG1-PP1-TXRX', 'org-openroadm-network-topology:tp-type': 'SRG-TXRX-PP'},
363                                res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
364                  self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'ROADMA'},
365                                res['network'][0]['node'][i]['supporting-node'])
366                  listNode.remove(nodeId)
367              elif(nodeId=='ROADMA-SRG3'):
368                  #Test related to SRG1
369                  self.assertEqual(nodeType,'SRG')
370                  self.assertEqual(len(res['network'][0]['node'][i]['ietf-network-topology:termination-point']),17)
371                  self.assertIn({'tp-id': 'SRG3-CP-TXRX', 'org-openroadm-network-topology:tp-type': 'SRG-TXRX-CP'},
372                                res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
373                  self.assertIn({'tp-id': 'SRG3-PP1-TXRX', 'org-openroadm-network-topology:tp-type': 'SRG-TXRX-PP'},
374                                res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
375                  self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'ROADMA'},
376                                res['network'][0]['node'][i]['supporting-node'])
377                  listNode.remove(nodeId)
378              elif(nodeId=='ROADMA-DEG1'):
379                 #Test related to DEG1
380                 self.assertEqual(nodeType,'DEGREE')
381                 self.assertIn({'tp-id': 'DEG1-TTP-TXRX', 'org-openroadm-network-topology:tp-type': 'DEGREE-TXRX-TTP'},
382                               res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
383                 self.assertIn({'tp-id': 'DEG1-CTP-TXRX', 'org-openroadm-network-topology:tp-type': 'DEGREE-TXRX-CTP'},
384                               res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
385                 self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'ROADMA'},
386                               res['network'][0]['node'][i]['supporting-node'])
387                 listNode.remove(nodeId)
388              elif(nodeId=='ROADMA-DEG2'):
389                 #Test related to DEG2
390                 self.assertEqual(nodeType,'DEGREE')
391                 self.assertIn({'tp-id': 'DEG2-TTP-TXRX', 'org-openroadm-network-topology:tp-type': 'DEGREE-TXRX-TTP'},
392                               res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
393                 self.assertIn({'tp-id': 'DEG2-CTP-TXRX', 'org-openroadm-network-topology:tp-type': 'DEGREE-TXRX-CTP'},
394                               res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
395                 self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'ROADMA'},
396                               res['network'][0]['node'][i]['supporting-node'])
397                 listNode.remove(nodeId)
398              else:
399                 self.assertFalse(True)
400          self.assertEqual(len(listNode),0)
401
402     #Connect the tail XPDRA to ROADMA and vice versa
403     def test_10_connect_tail_xpdr_rdm(self):
404          #Connect the tail: XPDRA to ROADMA
405         url = ("{}/operations/transportpce-networkutils:init-xpdr-rdm-links"
406                 .format(self.restconf_baseurl))
407         data = {"networkutils:input": {
408              "networkutils:links-input": {
409                "networkutils:xpdr-node": "XPDRA",
410                "networkutils:xpdr-num": "1",
411                "networkutils:network-num": "1",
412                "networkutils:rdm-node": "ROADMA",
413                "networkutils:srg-num": "1",
414                "networkutils:termination-point-num": "SRG1-PP1-TXRX"
415             }
416           }
417         }
418         headers = {'content-type': 'application/json'}
419         response = requests.request(
420               "POST", url, data=json.dumps(data), headers=headers,
421               auth=('admin', 'admin'))
422         self.assertEqual(response.status_code, requests.codes.ok)
423         time.sleep(10)
424
425     def test_11_connect_tail_rdm_xpdr(self):
426          #Connect the tail: ROADMA to XPDRA
427          url = ("{}/operations/transportpce-networkutils:init-rdm-xpdr-links"
428                  .format(self.restconf_baseurl))
429          data = {"networkutils:input": {
430               "networkutils:links-input": {
431                 "networkutils:xpdr-node": "XPDRA",
432                 "networkutils:xpdr-num": "1",
433                 "networkutils:network-num": "1",
434                 "networkutils:rdm-node": "ROADMA",
435                 "networkutils:srg-num": "1",
436                 "networkutils:termination-point-num": "SRG1-PP1-TXRX"
437              }
438             }
439          }
440          headers = {'content-type': 'application/json'}
441          response = requests.request(
442                "POST", url, data=json.dumps(data), headers=headers,
443                auth=('admin', 'admin'))
444          self.assertEqual(response.status_code, requests.codes.ok)
445          time.sleep(10)
446
447     def test_12_getLinks_OpenRoadmTopology(self):
448         url = ("{}/config/ietf-network:networks/network/openroadm-topology"
449                .format(self.restconf_baseurl))
450         headers = {'content-type': 'application/json'}
451         response = requests.request(
452              "GET", url, headers=headers, auth=('admin', 'admin'))
453         self.assertEqual(response.status_code, requests.codes.ok)
454         res = response.json()
455         #Tests related to links
456         nbLink=len(res['network'][0]['ietf-network-topology:link'])
457         self.assertEqual(nbLink,12)
458         expressLink=['ROADMA-DEG2-DEG2-CTP-TXRXtoROADMA-DEG1-DEG1-CTP-TXRX','ROADMA-DEG1-DEG1-CTP-TXRXtoROADMA-DEG2-DEG2-CTP-TXRX']
459         addLink=['ROADMA-SRG1-SRG1-CP-TXRXtoROADMA-DEG2-DEG2-CTP-TXRX','ROADMA-SRG1-SRG1-CP-TXRXtoROADMA-DEG1-DEG1-CTP-TXRX',
460                  'ROADMA-SRG3-SRG3-CP-TXRXtoROADMA-DEG2-DEG2-CTP-TXRX','ROADMA-SRG3-SRG3-CP-TXRXtoROADMA-DEG1-DEG1-CTP-TXRX']
461         dropLink=['ROADMA-DEG1-DEG1-CTP-TXRXtoROADMA-SRG1-SRG1-CP-TXRX','ROADMA-DEG2-DEG2-CTP-TXRXtoROADMA-SRG1-SRG1-CP-TXRX',
462                   'ROADMA-DEG1-DEG1-CTP-TXRXtoROADMA-SRG3-SRG3-CP-TXRX','ROADMA-DEG2-DEG2-CTP-TXRXtoROADMA-SRG3-SRG3-CP-TXRX']
463         XPDR_IN=['ROADMA-SRG1-SRG1-PP1-TXRXtoXPDRA-XPDR1-XPDR1-NETWORK1']
464         XPDR_OUT=['XPDRA-XPDR1-XPDR1-NETWORK1toROADMA-SRG1-SRG1-PP1-TXRX']
465         for i in range(0,nbLink):
466             nodeType=res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-network-topology:link-type']
467             linkId=res['network'][0]['ietf-network-topology:link'][i]['link-id']
468             if(nodeType=='EXPRESS-LINK'):
469                 find= linkId in expressLink
470                 self.assertEqual(find, True)
471                 expressLink.remove(linkId)
472             elif(nodeType=='ADD-LINK'):
473                 find= linkId in addLink
474                 self.assertEqual(find, True)
475                 addLink.remove(linkId)
476             elif(nodeType=='DROP-LINK'):
477                 find= linkId in dropLink
478                 self.assertEqual(find, True)
479                 dropLink.remove(linkId)
480             elif(nodeType=='XPONDER-INPUT'):
481                 find= linkId in XPDR_IN
482                 self.assertEqual(find, True)
483                 XPDR_IN.remove(linkId)
484             elif(nodeType=='XPONDER-OUTPUT'):
485                 find= linkId in XPDR_OUT
486                 self.assertEqual(find, True)
487                 XPDR_OUT.remove(linkId)
488             else:
489                 self.assertFalse(True)
490         self.assertEqual(len(expressLink),0)
491         self.assertEqual(len(addLink),0)
492         self.assertEqual(len(dropLink),0)
493         self.assertEqual(len(XPDR_IN),0)
494         self.assertEqual(len(XPDR_OUT),0)
495
496     def test_13_connect_ROADMC(self):
497         #Config ROADMC
498         url = ("{}/config/network-topology:"
499                 "network-topology/topology/topology-netconf/node/ROADMC"
500                 .format(self.restconf_baseurl))
501         data = {"node": [{
502              "node-id": "ROADMC",
503              "netconf-node-topology:username": "admin",
504              "netconf-node-topology:password": "admin",
505              "netconf-node-topology:host": "127.0.0.1",
506              "netconf-node-topology:port": "17833",
507              "netconf-node-topology:tcp-only": "false",
508              "netconf-node-topology:pass-through": {}}]}
509         headers = {'content-type': 'application/json'}
510         response = requests.request(
511              "PUT", url, data=json.dumps(data), headers=headers,
512              auth=('admin', 'admin'))
513         self.assertEqual(response.status_code, requests.codes.created)
514         time.sleep(30)
515
516     def test_14_omsAttributes_ROADMA_ROADMC(self):
517         # Config ROADMA-ROADMC oms-attributes
518         url = ("{}/config/ietf-network:networks/network/openroadm-topology/ietf-network-topology:"
519                "link/ROADMA-DEG1-DEG1-TTP-TXRXtoROADMC-DEG2-DEG2-TTP-TXRX/org-openroadm-network-topology:"
520                "OMS-attributes/span"
521                .format(self.restconf_baseurl))
522         data = {"span": {
523             "clfi": "fiber1",
524             "auto-spanloss": "true",
525             "engineered-spanloss": 12.2,
526             "link-concatenation": [{
527                 "SRLG-Id": 0,
528                 "fiber-type": "smf",
529                 "SRLG-length": 100000,
530                 "pmd": 0.5}]}}
531         headers = {'content-type': 'application/json'}
532         response = requests.request(
533             "PUT", url, data=json.dumps(data), headers=headers,
534             auth=('admin', 'admin'))
535         self.assertEqual(response.status_code, requests.codes.created)
536
537     def test_15_omsAttributes_ROADMC_ROADMA(self):
538         # Config ROADMC-ROADMA oms-attributes
539         url = ("{}/config/ietf-network:networks/network/openroadm-topology/ietf-network-topology:"
540                "link/ROADMC-DEG2-DEG2-TTP-TXRXtoROADMA-DEG1-DEG1-TTP-TXRX/org-openroadm-network-topology:"
541                "OMS-attributes/span"
542                .format(self.restconf_baseurl))
543         data = {"span": {
544             "clfi": "fiber1",
545             "auto-spanloss": "true",
546             "engineered-spanloss": 12.2,
547             "link-concatenation": [{
548                 "SRLG-Id": 0,
549                 "fiber-type": "smf",
550                 "SRLG-length": 100000,
551                 "pmd": 0.5}]}}
552         headers = {'content-type': 'application/json'}
553         response = requests.request(
554             "PUT", url, data=json.dumps(data), headers=headers,
555             auth=('admin', 'admin'))
556         self.assertEqual(response.status_code, requests.codes.created)
557
558     def test_16_getClliNetwork(self):
559         url = ("{}/config/ietf-network:networks/network/clli-network"
560               .format(self.restconf_baseurl))
561         headers = {'content-type': 'application/json'}
562         response = requests.request(
563             "GET", url, headers=headers, auth=('admin', 'admin'))
564         self.assertEqual(response.status_code, requests.codes.ok)
565         res = response.json()
566         nbNode=len(res['network'][0]['node'])
567         listNode=['NodeA','NodeC']
568         for i in range(0,nbNode):
569             nodeId = res['network'][0]['node'][i]['node-id']
570             find= nodeId in listNode
571             self.assertEqual(find, True)
572             if(nodeId=='NodeA'):
573                 self.assertEqual(res['network'][0]['node'][i]['org-openroadm-clli-network:clli'],'NodeA')
574             else:
575                 self.assertEqual(res['network'][0]['node'][i]['org-openroadm-clli-network:clli'],'NodeC')
576             listNode.remove(nodeId)
577
578         self.assertEqual(len(listNode),0)
579
580     def test_17_getOpenRoadmNetwork(self):
581         url = ("{}/config/ietf-network:networks/network/openroadm-network"
582               .format(self.restconf_baseurl))
583         headers = {'content-type': 'application/json'}
584         response = requests.request(
585             "GET", url, headers=headers, auth=('admin', 'admin'))
586         self.assertEqual(response.status_code, requests.codes.ok)
587         res = response.json()
588         nbNode=len(res['network'][0]['node'])
589         self.assertEqual(nbNode,3)
590         listNode=['XPDRA','ROADMA','ROADMC']
591         for i in range(0,nbNode):
592             self.assertEqual(res['network'][0]['node'][i]['supporting-node'][0]['network-ref'],'clli-network')
593             nodeId=res['network'][0]['node'][i]['node-id']
594             if(nodeId=='XPDRA'):
595                 self.assertEqual(res['network'][0]['node'][i]['supporting-node'][0]['node-ref'],'NodeA')
596                 self.assertEqual(res['network'][0]['node'][i]['org-openroadm-network:node-type'],'XPONDER')
597                 self.assertEqual(res['network'][0]['node'][i]['org-openroadm-network:model'],'1')
598                 listNode.remove(nodeId)
599             elif(nodeId=='ROADMA'):
600                 self.assertEqual(res['network'][0]['node'][i]['supporting-node'][0]['node-ref'],'NodeA')
601                 self.assertEqual(res['network'][0]['node'][i]['org-openroadm-network:node-type'],'ROADM')
602                 self.assertEqual(res['network'][0]['node'][i]['org-openroadm-network:model'],'2')
603                 listNode.remove(nodeId)
604             elif(nodeId=='ROADMC'):
605                 self.assertEqual(res['network'][0]['node'][i]['supporting-node'][0]['node-ref'],'NodeC')
606                 self.assertEqual(res['network'][0]['node'][i]['org-openroadm-network:node-type'],'ROADM')
607                 self.assertEqual(res['network'][0]['node'][i]['org-openroadm-network:model'],'2')
608                 listNode.remove(nodeId)
609             else:
610                 self.assertFalse(True)
611         self.assertEqual(len(listNode),0)
612
613     def test_18_getROADMLinkOpenRoadmTopology(self):
614         url = ("{}/config/ietf-network:networks/network/openroadm-topology"
615               .format(self.restconf_baseurl))
616         headers = {'content-type': 'application/json'}
617         response = requests.request(
618             "GET", url, headers=headers, auth=('admin', 'admin'))
619         self.assertEqual(response.status_code, requests.codes.ok)
620         res = response.json()
621         #Tests related to links
622         nbLink=len(res['network'][0]['ietf-network-topology:link'])
623         self.assertEqual(nbLink,20)
624         expressLink=['ROADMA-DEG2-DEG2-CTP-TXRXtoROADMA-DEG1-DEG1-CTP-TXRX','ROADMA-DEG1-DEG1-CTP-TXRXtoROADMA-DEG2-DEG2-CTP-TXRX',
625                      'ROADMC-DEG2-DEG2-CTP-TXRXtoROADMC-DEG1-DEG1-CTP-TXRX','ROADMC-DEG1-DEG1-CTP-TXRXtoROADMC-DEG2-DEG2-CTP-TXRX']
626         addLink=['ROADMA-SRG1-SRG1-CP-TXRXtoROADMA-DEG2-DEG2-CTP-TXRX','ROADMA-SRG1-SRG1-CP-TXRXtoROADMA-DEG1-DEG1-CTP-TXRX',
627                  'ROADMA-SRG3-SRG3-CP-TXRXtoROADMA-DEG2-DEG2-CTP-TXRX','ROADMA-SRG3-SRG3-CP-TXRXtoROADMA-DEG1-DEG1-CTP-TXRX',
628                  'ROADMC-SRG1-SRG1-CP-TXRXtoROADMC-DEG2-DEG2-CTP-TXRX','ROADMC-SRG1-SRG1-CP-TXRXtoROADMC-DEG1-DEG1-CTP-TXRX']
629         dropLink=['ROADMA-DEG1-DEG1-CTP-TXRXtoROADMA-SRG1-SRG1-CP-TXRX','ROADMA-DEG2-DEG2-CTP-TXRXtoROADMA-SRG1-SRG1-CP-TXRX',
630                   'ROADMA-DEG1-DEG1-CTP-TXRXtoROADMA-SRG3-SRG3-CP-TXRX','ROADMA-DEG2-DEG2-CTP-TXRXtoROADMA-SRG3-SRG3-CP-TXRX',
631                   'ROADMC-DEG1-DEG1-CTP-TXRXtoROADMC-SRG1-SRG1-CP-TXRX','ROADMC-DEG2-DEG2-CTP-TXRXtoROADMC-SRG1-SRG1-CP-TXRX']
632         R2RLink=['ROADMA-DEG1-DEG1-TTP-TXRXtoROADMC-DEG2-DEG2-TTP-TXRX','ROADMC-DEG2-DEG2-TTP-TXRXtoROADMA-DEG1-DEG1-TTP-TXRX']
633         XPDR_IN=['ROADMA-SRG1-SRG1-PP1-TXRXtoXPDRA-XPDR1-XPDR1-NETWORK1']
634         XPDR_OUT=['XPDRA-XPDR1-XPDR1-NETWORK1toROADMA-SRG1-SRG1-PP1-TXRX']
635         for i in range(0,nbLink):
636             nodeType=res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-network-topology:link-type']
637             linkId=res['network'][0]['ietf-network-topology:link'][i]['link-id']
638             if(nodeType=='EXPRESS-LINK'):
639                 find= linkId in expressLink
640                 self.assertEqual(find, True)
641                 expressLink.remove(linkId)
642             elif(nodeType=='ADD-LINK'):
643                 find= linkId in addLink
644                 self.assertEqual(find, True)
645                 addLink.remove(linkId)
646             elif(nodeType=='DROP-LINK'):
647                 find= linkId in dropLink
648                 self.assertEqual(find, True)
649                 dropLink.remove(linkId)
650             elif(nodeType=='ROADM-TO-ROADM'):
651                 find= linkId in R2RLink
652                 self.assertEqual(find, True)
653                 R2RLink.remove(linkId)
654             elif(nodeType=='XPONDER-INPUT'):
655                 find= linkId in XPDR_IN
656                 self.assertEqual(find, True)
657                 XPDR_IN.remove(linkId)
658             elif(nodeType=='XPONDER-OUTPUT'):
659                 find= linkId in XPDR_OUT
660                 self.assertEqual(find, True)
661                 XPDR_OUT.remove(linkId)
662             else:
663                 self.assertFalse(True)
664         self.assertEqual(len(expressLink),0)
665         self.assertEqual(len(addLink),0)
666         self.assertEqual(len(dropLink),0)
667         self.assertEqual(len(R2RLink),0)
668         self.assertEqual(len(XPDR_IN),0)
669         self.assertEqual(len(XPDR_OUT),0)
670
671     def test_19_getLinkOmsAttributesOpenRoadmTopology(self):
672         url = ("{}/config/ietf-network:networks/network/openroadm-topology"
673               .format(self.restconf_baseurl))
674         headers = {'content-type': 'application/json'}
675         response = requests.request(
676             "GET", url, headers=headers, auth=('admin', 'admin'))
677         self.assertEqual(response.status_code, requests.codes.ok)
678         res = response.json()
679         #Tests related to links
680         nbLink=len(res['network'][0]['ietf-network-topology:link'])
681         self.assertEqual(nbLink,20)
682         R2RLink=['ROADMA-DEG1-DEG1-TTP-TXRXtoROADMC-DEG2-DEG2-TTP-TXRX',
683                  'ROADMC-DEG2-DEG2-TTP-TXRXtoROADMA-DEG1-DEG1-TTP-TXRX']
684         for i in range(0,nbLink):
685             nodeType = res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-network-topology:link-type']
686             link_id=res['network'][0]['ietf-network-topology:link'][i]['link-id']
687             if(link_id in R2RLink):
688                 find = False
689                 spanLoss = res['network'][0]['ietf-network-topology:link'][i][
690                     'org-openroadm-network-topology:OMS-attributes']['span']["engineered-spanloss"]
691                 length = res['network'][0]['ietf-network-topology:link'][i][
692                     'org-openroadm-network-topology:OMS-attributes']['span']['link-concatenation'][0]['SRLG-length']
693                 if((spanLoss!=None)&(length!=None)):
694                     find = True
695                 self.assertTrue(find)
696                 R2RLink.remove(link_id)
697         self.assertEqual(len(R2RLink),0)
698
699     def test_20_getNodes_OpenRoadmTopology(self):
700          url = ("{}/config/ietf-network:networks/network/openroadm-topology"
701                .format(self.restconf_baseurl))
702          headers = {'content-type': 'application/json'}
703          response = requests.request(
704              "GET", url, headers=headers, auth=('admin', 'admin'))
705          res = response.json()
706          #Tests related to nodes
707          self.assertEqual(response.status_code, requests.codes.ok)
708          nbNode=len(res['network'][0]['node'])
709          self.assertEqual(nbNode,8)
710          listNode=['XPDRA-XPDR1',
711                    'ROADMA-SRG1','ROADMA-SRG3','ROADMA-DEG1','ROADMA-DEG2',
712                    'ROADMC-SRG1','ROADMC-DEG1','ROADMC-DEG2']
713          #************************Tests related to XPDRA nodes
714          for i in range(0,nbNode):
715              nodeType=res['network'][0]['node'][i]['org-openroadm-network-topology:node-type']
716              nodeId=res['network'][0]['node'][i]['node-id']
717              if(nodeId=='XPDRA-XPDR1'):
718                  self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'XPDRA'},
719                                res['network'][0]['node'][i]['supporting-node'])
720                  self.assertEqual(nodeType,'XPONDER')
721                  nbTps=len(res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
722                  self.assertTrue(nbTps == 6)
723                  client = 0
724                  network = 0
725                  for j in range(0,nbTps):
726                      tpType=res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['org-openroadm-network-topology:tp-type']
727                      if (tpType=='XPONDER-CLIENT'):
728                          client += 1
729                      elif (tpType=='XPONDER-NETWORK'):
730                          network += 1
731                  self.assertTrue(client == 4)
732                  self.assertTrue(network == 2)
733                  listNode.remove(nodeId)
734              elif(nodeId=='ROADMA-SRG1'):
735                  #Test related to SRG1
736                  self.assertEqual(len(res['network'][0]['node'][i]['ietf-network-topology:termination-point']),17)
737                  self.assertIn({'tp-id': 'SRG1-CP-TXRX', 'org-openroadm-network-topology:tp-type': 'SRG-TXRX-CP'},
738                                res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
739                  self.assertIn({'tp-id': 'SRG1-PP1-TXRX', 'org-openroadm-network-topology:tp-type': 'SRG-TXRX-PP'},
740                                res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
741                  self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'ROADMA'},
742                                res['network'][0]['node'][i]['supporting-node'])
743                  self.assertEqual(res['network'][0]['node'][i]['org-openroadm-network-topology:node-type'],'SRG')
744                  listNode.remove(nodeId)
745              elif(nodeId=='ROADMA-SRG3'):
746                  #Test related to SRG1
747                  self.assertEqual(len(res['network'][0]['node'][i]['ietf-network-topology:termination-point']),17)
748                  self.assertIn({'tp-id': 'SRG3-CP-TXRX', 'org-openroadm-network-topology:tp-type': 'SRG-TXRX-CP'},
749                                res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
750                  self.assertIn({'tp-id': 'SRG3-PP1-TXRX', 'org-openroadm-network-topology:tp-type': 'SRG-TXRX-PP'},
751                                res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
752                  self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'ROADMA'},
753                                res['network'][0]['node'][i]['supporting-node'])
754                  self.assertEqual(res['network'][0]['node'][i]['org-openroadm-network-topology:node-type'],'SRG')
755                  listNode.remove(nodeId)
756              elif(nodeId=='ROADMA-DEG1'):
757                  #Test related to DEG1
758                  self.assertIn({'tp-id': 'DEG1-TTP-TXRX', 'org-openroadm-network-topology:tp-type': 'DEGREE-TXRX-TTP'},
759                                res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
760                  self.assertIn({'tp-id': 'DEG1-CTP-TXRX', 'org-openroadm-network-topology:tp-type': 'DEGREE-TXRX-CTP'},
761                                res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
762                  self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'ROADMA'},
763                                res['network'][0]['node'][i]['supporting-node'])
764                  self.assertEqual(res['network'][0]['node'][i]['org-openroadm-network-topology:node-type'],'DEGREE')
765                  listNode.remove(nodeId)
766              elif(nodeId=='ROADMA-DEG2'):
767                  #Test related to DEG2
768                  self.assertIn({'tp-id': 'DEG2-TTP-TXRX', 'org-openroadm-network-topology:tp-type': 'DEGREE-TXRX-TTP'},
769                                res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
770                  self.assertIn({'tp-id': 'DEG2-CTP-TXRX', 'org-openroadm-network-topology:tp-type': 'DEGREE-TXRX-CTP'},
771                                res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
772                  self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'ROADMA'},
773                                res['network'][0]['node'][i]['supporting-node'])
774                  self.assertEqual(res['network'][0]['node'][i]['org-openroadm-network-topology:node-type'],'DEGREE')
775                  listNode.remove(nodeId)
776              elif(nodeId=='ROADMC-SRG1'):
777                  #Test related to SRG1
778                  self.assertEqual(len(res['network'][0]['node'][i]['ietf-network-topology:termination-point']),17)
779                  self.assertIn({'tp-id': 'SRG1-CP-TXRX', 'org-openroadm-network-topology:tp-type': 'SRG-TXRX-CP'},
780                                res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
781                  self.assertIn({'tp-id': 'SRG1-PP1-TXRX', 'org-openroadm-network-topology:tp-type': 'SRG-TXRX-PP'},
782                                res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
783                  self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'ROADMC'},
784                                res['network'][0]['node'][i]['supporting-node'])
785                  self.assertEqual(res['network'][0]['node'][i]['org-openroadm-network-topology:node-type'],'SRG')
786                  listNode.remove(nodeId)
787              elif(nodeId=='ROADMC-DEG1'):
788                  #Test related to DEG1
789                  self.assertIn({'tp-id': 'DEG1-TTP-TXRX', 'org-openroadm-network-topology:tp-type': 'DEGREE-TXRX-TTP'},
790                                res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
791                  self.assertIn({'tp-id': 'DEG1-CTP-TXRX', 'org-openroadm-network-topology:tp-type': 'DEGREE-TXRX-CTP'},
792                                res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
793                  self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'ROADMC'},
794                                res['network'][0]['node'][i]['supporting-node'])
795                  self.assertEqual(res['network'][0]['node'][i]['org-openroadm-network-topology:node-type'],'DEGREE')
796                  listNode.remove(nodeId)
797              elif(nodeId=='ROADMC-DEG2'):
798                  #Test related to DEG2
799                  self.assertIn({'tp-id': 'DEG2-TTP-TXRX', 'org-openroadm-network-topology:tp-type': 'DEGREE-TXRX-TTP'},
800                                res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
801                  self.assertIn({'tp-id': 'DEG2-CTP-TXRX', 'org-openroadm-network-topology:tp-type': 'DEGREE-TXRX-CTP'},
802                                res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
803                  self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'ROADMC'},
804                                res['network'][0]['node'][i]['supporting-node'])
805                  self.assertEqual(res['network'][0]['node'][i]['org-openroadm-network-topology:node-type'],'DEGREE')
806                  listNode.remove(nodeId)
807              else:
808                 self.assertFalse(True)
809          self.assertEqual(len(listNode),0)
810
811     def test_21_connect_ROADMB(self):
812         url = ("{}/config/network-topology:"
813                 "network-topology/topology/topology-netconf/node/ROADMB"
814                .format(self.restconf_baseurl))
815         data = {"node": [{
816              "node-id": "ROADMB",
817              "netconf-node-topology:username": "admin",
818              "netconf-node-topology:password": "admin",
819              "netconf-node-topology:host": "127.0.0.1",
820              "netconf-node-topology:port": "17832",
821              "netconf-node-topology:tcp-only": "false",
822              "netconf-node-topology:pass-through": {}}]}
823         headers = {'content-type': 'application/json'}
824         response = requests.request(
825              "PUT", url, data=json.dumps(data), headers=headers,
826              auth=('admin', 'admin'))
827         self.assertEqual(response.status_code, requests.codes.created)
828         time.sleep(30)
829
830     def test_22_omsAttributes_ROADMA_ROADMB(self):
831         # Config ROADMA-ROADMB oms-attributes
832         url = ("{}/config/ietf-network:networks/network/openroadm-topology/ietf-network-topology:"
833                "link/ROADMA-DEG2-DEG2-TTP-TXRXtoROADMB-DEG1-DEG1-TTP-TXRX/org-openroadm-network-topology:"
834                "OMS-attributes/span"
835                .format(self.restconf_baseurl))
836         data = {"span": {
837                 "clfi": "fiber1",
838                 "auto-spanloss": "true",
839                 "engineered-spanloss": 12.2,
840                 "link-concatenation": [{
841                     "SRLG-Id": 0,
842                     "fiber-type": "smf",
843                     "SRLG-length": 100000,
844                     "pmd": 0.5}]}}
845         headers = {'content-type': 'application/json'}
846         response = requests.request(
847                 "PUT", url, data=json.dumps(data), headers=headers,
848                 auth=('admin', 'admin'))
849         self.assertEqual(response.status_code, requests.codes.created)
850
851     def test_23_omsAttributes_ROADMB_ROADMA(self):
852         # Config ROADMB-ROADMA oms-attributes
853         url = ("{}/config/ietf-network:networks/network/openroadm-topology/ietf-network-topology:"
854                "link/ROADMB-DEG1-DEG1-TTP-TXRXtoROADMA-DEG2-DEG2-TTP-TXRX/org-openroadm-network-topology:"
855                "OMS-attributes/span"
856                .format(self.restconf_baseurl))
857         data = {"span": {
858                 "clfi": "fiber1",
859                 "auto-spanloss": "true",
860                 "engineered-spanloss": 12.2,
861                 "link-concatenation": [{
862                     "SRLG-Id": 0,
863                     "fiber-type": "smf",
864                     "SRLG-length": 100000,
865                     "pmd": 0.5}]}}
866         headers = {'content-type': 'application/json'}
867         response = requests.request(
868                 "PUT", url, data=json.dumps(data), headers=headers,
869                 auth=('admin', 'admin'))
870         self.assertEqual(response.status_code, requests.codes.created)
871
872     def test_24_omsAttributes_ROADMB_ROADMC(self):
873         # Config ROADMB-ROADMC oms-attributes
874         url = ("{}/config/ietf-network:networks/network/openroadm-topology/ietf-network-topology:"
875                "link/ROADMB-DEG2-DEG2-TTP-TXRXtoROADMC-DEG1-DEG1-TTP-TXRX/org-openroadm-network-topology:"
876                "OMS-attributes/span"
877                .format(self.restconf_baseurl))
878         data = {"span": {
879                 "clfi": "fiber1",
880                 "auto-spanloss": "true",
881                 "engineered-spanloss": 12.2,
882                 "link-concatenation": [{
883                     "SRLG-Id": 0,
884                     "fiber-type": "smf",
885                     "SRLG-length": 100000,
886                     "pmd": 0.5}]}}
887         headers = {'content-type': 'application/json'}
888         response = requests.request(
889                 "PUT", url, data=json.dumps(data), headers=headers,
890                 auth=('admin', 'admin'))
891         self.assertEqual(response.status_code, requests.codes.created)
892
893     def test_25_omsAttributes_ROADMC_ROADMB(self):
894         # Config ROADMC-ROADMB oms-attributes
895         url = ("{}/config/ietf-network:networks/network/openroadm-topology/ietf-network-topology:"
896                "link/ROADMC-DEG1-DEG1-TTP-TXRXtoROADMB-DEG2-DEG2-TTP-TXRX/org-openroadm-network-topology:"
897                "OMS-attributes/span"
898                .format(self.restconf_baseurl))
899         data = {"span": {
900                 "clfi": "fiber1",
901                 "auto-spanloss": "true",
902                 "engineered-spanloss": 12.2,
903                 "link-concatenation": [{
904                     "SRLG-Id": 0,
905                     "fiber-type": "smf",
906                     "SRLG-length": 100000,
907                     "pmd": 0.5}]}}
908         headers = {'content-type': 'application/json'}
909         response = requests.request(
910                 "PUT", url, data=json.dumps(data), headers=headers,
911                 auth=('admin', 'admin'))
912         self.assertEqual(response.status_code, requests.codes.created)
913
914     def test_26_getClliNetwork(self):
915         url = ("{}/config/ietf-network:networks/network/clli-network"
916               .format(self.restconf_baseurl))
917         headers = {'content-type': 'application/json'}
918         response = requests.request(
919             "GET", url, headers=headers, auth=('admin', 'admin'))
920         self.assertEqual(response.status_code, requests.codes.ok)
921         res = response.json()
922         nbNode=len(res['network'][0]['node'])
923         listNode=['NodeA','NodeB','NodeC']
924         for i in range(0,nbNode):
925             nodeId = res['network'][0]['node'][i]['node-id']
926             find= nodeId in listNode
927             self.assertEqual(find, True)
928             if(nodeId=='NodeA'):
929                 self.assertEqual(res['network'][0]['node'][i]['org-openroadm-clli-network:clli'],'NodeA')
930             elif(nodeId=='NodeB'):
931                 self.assertEqual(res['network'][0]['node'][i]['org-openroadm-clli-network:clli'],'NodeB')
932             else:
933                 self.assertEqual(res['network'][0]['node'][i]['org-openroadm-clli-network:clli'],'NodeC')
934             listNode.remove(nodeId)
935
936         self.assertEqual(len(listNode),0)
937
938     def test_27_verifyDegree(self):
939         url = ("{}/config/ietf-network:networks/network/openroadm-topology"
940               .format(self.restconf_baseurl))
941         headers = {'content-type': 'application/json'}
942         response = requests.request(
943             "GET", url, headers=headers, auth=('admin', 'admin'))
944         self.assertEqual(response.status_code, requests.codes.ok)
945         res = response.json()
946         #Tests related to links
947         nbLink=len(res['network'][0]['ietf-network-topology:link'])
948         listR2RLink=['ROADMA-DEG1-DEG1-TTP-TXRXtoROADMC-DEG2-DEG2-TTP-TXRX','ROADMC-DEG2-DEG2-TTP-TXRXtoROADMA-DEG1-DEG1-TTP-TXRX',
949            'ROADMA-DEG2-DEG2-TTP-TXRXtoROADMB-DEG1-DEG1-TTP-TXRX','ROADMC-DEG1-DEG1-TTP-TXRXtoROADMB-DEG2-DEG2-TTP-TXRX',
950            'ROADMB-DEG1-DEG1-TTP-TXRXtoROADMA-DEG2-DEG2-TTP-TXRX','ROADMB-DEG2-DEG2-TTP-TXRXtoROADMC-DEG1-DEG1-TTP-TXRX']
951         for i in range(0,nbLink):
952             if res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-network-topology:link-type'] == 'ROADM-TO-ROADM':
953                 link_id = res['network'][0]['ietf-network-topology:link'][i]['link-id']
954                 find= link_id in listR2RLink
955                 self.assertEqual(find, True)
956                 listR2RLink.remove(link_id)
957         self.assertEqual(len(listR2RLink),0)
958
959     def test_28_verifyOppositeLinkTopology(self):
960         url = ("{}/config/ietf-network:networks/network/openroadm-topology"
961                .format(self.restconf_baseurl))
962         headers = {'content-type': 'application/json'}
963         response = requests.request(
964              "GET", url, headers=headers, auth=('admin', 'admin'))
965         self.assertEqual(response.status_code, requests.codes.ok)
966         res = response.json()
967         #Tests related to links
968         nbLink=len(res['network'][0]['ietf-network-topology:link'])
969         self.assertEqual(nbLink,34)
970         for i in range(0,nbLink):
971             link_id=res['network'][0]['ietf-network-topology:link'][i]['link-id']
972             link_type=res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-network-topology:link-type']
973             link_src=res['network'][0]['ietf-network-topology:link'][i]['source']['source-node']
974             link_dest=res['network'][0]['ietf-network-topology:link'][i]['destination']['dest-node']
975             oppLink_id=res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-common-network:opposite-link']
976             #Find the opposite link
977             url_oppLink="{}/config/ietf-network:networks/network/openroadm-topology/ietf-network-topology:link/"+oppLink_id
978             url = (url_oppLink.format(self.restconf_baseurl))
979             headers = {'content-type': 'application/json'}
980             response_oppLink = requests.request("GET", url, headers=headers, auth=('admin', 'admin'))
981             self.assertEqual(response_oppLink.status_code, requests.codes.ok)
982             res_oppLink = response_oppLink.json()
983             self.assertEqual(res_oppLink['ietf-network-topology:link'][0]['org-openroadm-common-network:opposite-link'],link_id)
984             self.assertEqual(res_oppLink['ietf-network-topology:link'][0]['source']['source-node'],link_dest)
985             self.assertEqual(res_oppLink['ietf-network-topology:link'][0]['destination']['dest-node'],link_src)
986             oppLink_type=res_oppLink['ietf-network-topology:link'][0]['org-openroadm-network-topology:link-type']
987             if link_type=='ADD-LINK':
988                 self.assertEqual(oppLink_type, 'DROP-LINK')
989             elif link_type=='DROP-LINK':
990                 self.assertEqual(oppLink_type, 'ADD-LINK')
991             elif link_type=='EXPRESS-LINK':
992                 self.assertEqual(oppLink_type, 'EXPRESS-LINK')
993             elif link_type=='ROADM-TO-ROADM':
994                 self.assertEqual(oppLink_type, 'ROADM-TO-ROADM')
995             elif link_type=='XPONDER-INPUT':
996                 self.assertEqual(oppLink_type, 'XPONDER-OUTPUT')
997             elif link_type=='XPONDER-OUTPUT':
998                 self.assertEqual(oppLink_type, 'XPONDER-INPUT')
999         time.sleep(5)
1000
1001     def test_29_getLinkOmsAttributesOpenRoadmTopology(self):
1002         url = ("{}/config/ietf-network:networks/network/openroadm-topology"
1003               .format(self.restconf_baseurl))
1004         headers = {'content-type': 'application/json'}
1005         response = requests.request(
1006             "GET", url, headers=headers, auth=('admin', 'admin'))
1007         self.assertEqual(response.status_code, requests.codes.ok)
1008         res = response.json()
1009         nbLink=len(res['network'][0]['ietf-network-topology:link'])
1010         R2RLink = ['ROADMA-DEG1-DEG1-TTP-TXRXtoROADMC-DEG2-DEG2-TTP-TXRX',
1011                        'ROADMC-DEG2-DEG2-TTP-TXRXtoROADMA-DEG1-DEG1-TTP-TXRX',
1012                        'ROADMA-DEG2-DEG2-TTP-TXRXtoROADMB-DEG1-DEG1-TTP-TXRX',
1013                        'ROADMC-DEG1-DEG1-TTP-TXRXtoROADMB-DEG2-DEG2-TTP-TXRX',
1014                        'ROADMB-DEG1-DEG1-TTP-TXRXtoROADMA-DEG2-DEG2-TTP-TXRX',
1015                        'ROADMB-DEG2-DEG2-TTP-TXRXtoROADMC-DEG1-DEG1-TTP-TXRX']
1016         for i in range(0,nbLink):
1017             nodeType = res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-network-topology:link-type']
1018             link_id=res['network'][0]['ietf-network-topology:link'][i]['link-id']
1019             if(link_id in R2RLink):
1020                 find = False
1021                 spanLoss = res['network'][0]['ietf-network-topology:link'][i][
1022                     'org-openroadm-network-topology:OMS-attributes']['span']["engineered-spanloss"]
1023                 length = res['network'][0]['ietf-network-topology:link'][i][
1024                     'org-openroadm-network-topology:OMS-attributes']['span']['link-concatenation'][0]['SRLG-length']
1025                 if((spanLoss!=None)&(length!=None)):
1026                     find = True
1027                 self.assertTrue(find)
1028                 R2RLink.remove(link_id)
1029         self.assertEqual(len(R2RLink),0)
1030
1031     def test_30_disconnect_ROADMB(self):
1032         #Delete in the topology-netconf
1033         url = ("{}/config/network-topology:"
1034                 "network-topology/topology/topology-netconf/node/ROADMB"
1035                .format(self.restconf_baseurl))
1036         data = {}
1037         headers = {'content-type': 'application/json'}
1038         response = requests.request(
1039              "DELETE", url, data=json.dumps(data), headers=headers,
1040              auth=('admin', 'admin'))
1041         self.assertEqual(response.status_code, requests.codes.ok)
1042         #Delete in the clli-network
1043         url = ("{}/config/ietf-network:networks/network/clli-network/node/NodeB"
1044                .format(self.restconf_baseurl))
1045         data = {}
1046         headers = {'content-type': 'application/json'}
1047         response = requests.request(
1048              "DELETE", url, data=json.dumps(data), headers=headers,
1049              auth=('admin', 'admin'))
1050         self.assertEqual(response.status_code, requests.codes.ok)
1051
1052     def test_31_disconnect_ROADMC(self):
1053         #Delete in the topology-netconf
1054         url = ("{}/config/network-topology:"
1055                 "network-topology/topology/topology-netconf/node/ROADMC"
1056                .format(self.restconf_baseurl))
1057         data = {}
1058         headers = {'content-type': 'application/json'}
1059         response = requests.request(
1060              "DELETE", url, data=json.dumps(data), headers=headers,
1061              auth=('admin', 'admin'))
1062         self.assertEqual(response.status_code, requests.codes.ok)
1063         #Delete in the clli-network
1064         url = ("{}/config/ietf-network:networks/network/clli-network/node/NodeC"
1065                .format(self.restconf_baseurl))
1066         data = {}
1067         headers = {'content-type': 'application/json'}
1068         response = requests.request(
1069              "DELETE", url, data=json.dumps(data), headers=headers,
1070              auth=('admin', 'admin'))
1071         self.assertEqual(response.status_code, requests.codes.ok)
1072
1073     def test_32_getNodes_OpenRoadmTopology(self):
1074         url = ("{}/config/ietf-network:networks/network/openroadm-topology"
1075               .format(self.restconf_baseurl))
1076         headers = {'content-type': 'application/json'}
1077         response = requests.request(
1078             "GET", url, headers=headers, auth=('admin', 'admin'))
1079         res = response.json()
1080         #Tests related to nodes
1081         self.assertEqual(response.status_code, requests.codes.ok)
1082         nbNode=len(res['network'][0]['node'])
1083         self.assertEqual(nbNode,5)
1084         listNode=['XPDRA-XPDR1','ROADMA-SRG1', 'ROADMA-SRG3','ROADMA-DEG1','ROADMA-DEG2']
1085         for i in range(0,nbNode):
1086             nodeType=res['network'][0]['node'][i]['org-openroadm-network-topology:node-type']
1087             nodeId=res['network'][0]['node'][i]['node-id']
1088             #Tests related to XPDRA nodes
1089             if(nodeId=='XPDRA-XPDR1'):
1090                 nbTp = len(res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
1091                 for j in range(0, nbTp):
1092                     tpid = res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['tp-id']
1093                     if (tpid == 'XPDR1-CLIENT1'):
1094                         self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
1095                                          ['org-openroadm-network-topology:tp-type'], 'XPONDER-CLIENT')
1096                         self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
1097                                          ['org-openroadm-network-topology:xpdr-client-attributes']['tail-equipment-id'],
1098                                          'XPDR1-NETWORK1')
1099                 self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'XPDRA'},
1100                     res['network'][0]['node'][i]['supporting-node'])
1101                 listNode.remove(nodeId)
1102             elif(nodeId=='ROADMA-SRG1'):
1103                 #Test related to SRG1
1104                 self.assertEqual(nodeType,'SRG')
1105                 self.assertEqual(len(res['network'][0]['node'][i]['ietf-network-topology:termination-point']),17)
1106                 self.assertIn({'tp-id': 'SRG1-CP-TXRX', 'org-openroadm-network-topology:tp-type': 'SRG-TXRX-CP'},
1107                               res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
1108                 self.assertIn({'tp-id': 'SRG1-PP1-TXRX', 'org-openroadm-network-topology:tp-type': 'SRG-TXRX-PP'},
1109                               res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
1110                 self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'ROADMA'},
1111                               res['network'][0]['node'][i]['supporting-node'])
1112                 listNode.remove(nodeId)
1113             elif(nodeId=='ROADMA-SRG3'):
1114                 #Test related to SRG1
1115                 self.assertEqual(nodeType,'SRG')
1116                 self.assertEqual(len(res['network'][0]['node'][i]['ietf-network-topology:termination-point']),17)
1117                 self.assertIn({'tp-id': 'SRG3-CP-TXRX', 'org-openroadm-network-topology:tp-type': 'SRG-TXRX-CP'},
1118                               res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
1119                 self.assertIn({'tp-id': 'SRG3-PP1-TXRX', 'org-openroadm-network-topology:tp-type': 'SRG-TXRX-PP'},
1120                               res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
1121                 self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'ROADMA'},
1122                               res['network'][0]['node'][i]['supporting-node'])
1123                 listNode.remove(nodeId)
1124             elif(nodeId=='ROADMA-DEG1'):
1125                 #Test related to DEG1
1126                 self.assertEqual(nodeType,'DEGREE')
1127                 self.assertIn({'tp-id': 'DEG1-TTP-TXRX', 'org-openroadm-network-topology:tp-type': 'DEGREE-TXRX-TTP'},
1128                     res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
1129                 self.assertIn({'tp-id': 'DEG1-CTP-TXRX', 'org-openroadm-network-topology:tp-type': 'DEGREE-TXRX-CTP'},
1130                     res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
1131                 self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'ROADMA'},
1132                     res['network'][0]['node'][i]['supporting-node'])
1133                 listNode.remove(nodeId)
1134             elif(nodeId=='ROADMA-DEG2'):
1135                 #Test related to DEG2
1136                 self.assertEqual(nodeType,'DEGREE')
1137                 self.assertIn({'tp-id': 'DEG2-TTP-TXRX', 'org-openroadm-network-topology:tp-type': 'DEGREE-TXRX-TTP'},
1138                               res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
1139                 self.assertIn({'tp-id': 'DEG2-CTP-TXRX', 'org-openroadm-network-topology:tp-type': 'DEGREE-TXRX-CTP'},
1140                               res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
1141                 self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'ROADMA'},
1142                               res['network'][0]['node'][i]['supporting-node'])
1143                 listNode.remove(nodeId)
1144             else:
1145                 self.assertFalse(True)
1146         self.assertEqual(len(listNode),0)
1147         #Test related to SRG1 of ROADMC
1148         for i in range(0,nbNode):
1149             self.assertNotEqual(res['network'][0]['node'][i]['node-id'],'ROADMC-SRG1')
1150             self.assertNotEqual(res['network'][0]['node'][i]['node-id'],'ROADMC-DEG1')
1151             self.assertNotEqual(res['network'][0]['node'][i]['node-id'],'ROADMC-DEG2')
1152
1153     def test_33_getOpenRoadmNetwork(self):
1154         url = ("{}/config/ietf-network:networks/network/openroadm-network"
1155               .format(self.restconf_baseurl))
1156         headers = {'content-type': 'application/json'}
1157         response = requests.request(
1158             "GET", url, headers=headers, auth=('admin', 'admin'))
1159         self.assertEqual(response.status_code, requests.codes.ok)
1160         res = response.json()
1161         nbNode=len(res['network'][0]['node'])
1162         self.assertEqual(nbNode,2)
1163         for i in range(0,nbNode-1):
1164             self.assertNotEqual(res['network'][0]['node'][i]['node-id'],'ROADMC')
1165             self.assertNotEqual(res['network'][0]['node'][i]['node-id'],'ROADMB')
1166
1167     def test_34_getClliNetwork(self):
1168         url = ("{}/config/ietf-network:networks/network/clli-network"
1169               .format(self.restconf_baseurl))
1170         headers = {'content-type': 'application/json'}
1171         response = requests.request(
1172             "GET", url, headers=headers, auth=('admin', 'admin'))
1173         self.assertEqual(response.status_code, requests.codes.ok)
1174         res = response.json()
1175         nbNode=len(res['network'][0]['node'])
1176         self.assertEqual(nbNode,1)
1177         for i in range(0,nbNode-1):
1178             self.assertNotEqual(res['network'][0]['node'][1]['org-openroadm-clli-network:clli'],'NodeC')
1179
1180     def test_35_disconnect_XPDRA(self):
1181         url = ("{}/config/network-topology:"
1182                "network-topology/topology/topology-netconf/node/XPDRA"
1183               .format(self.restconf_baseurl))
1184         data = {}
1185         headers = {'content-type': 'application/json'}
1186         response = requests.request(
1187             "DELETE", url, data=json.dumps(data), headers=headers,
1188             auth=('admin', 'admin'))
1189         self.assertEqual(response.status_code, requests.codes.ok)
1190
1191     def test_36_getClliNetwork(self):
1192         url = ("{}/config/ietf-network:networks/network/clli-network"
1193               .format(self.restconf_baseurl))
1194         headers = {'content-type': 'application/json'}
1195         response = requests.request(
1196             "GET", url, headers=headers, auth=('admin', 'admin'))
1197         self.assertEqual(response.status_code, requests.codes.ok)
1198         res = response.json()
1199         nbNode=len(res['network'][0]['node'])
1200         self.assertEqual(nbNode,1)
1201         self.assertEqual(res['network'][0]['node'][0]['org-openroadm-clli-network:clli'],'NodeA')
1202
1203     def test_37_getOpenRoadmNetwork(self):
1204         url = ("{}/config/ietf-network:networks/network/openroadm-network"
1205               .format(self.restconf_baseurl))
1206         headers = {'content-type': 'application/json'}
1207         response = requests.request(
1208             "GET", url, headers=headers, auth=('admin', 'admin'))
1209         self.assertEqual(response.status_code, requests.codes.ok)
1210         res = response.json()
1211         nbNode=len(res['network'][0]['node'])
1212         self.assertEqual(nbNode,1)
1213         for i in range(0,nbNode):
1214             self.assertNotEqual(res['network'][0]['node'][i]['node-id'],'XPDRA')
1215
1216     def test_38_getNodes_OpenRoadmTopology(self):
1217         url = ("{}/config/ietf-network:networks/network/openroadm-topology"
1218               .format(self.restconf_baseurl))
1219         headers = {'content-type': 'application/json'}
1220         response = requests.request(
1221             "GET", url, headers=headers, auth=('admin', 'admin'))
1222         res = response.json()
1223         #Tests related to nodes
1224         self.assertEqual(response.status_code, requests.codes.ok)
1225         nbNode=len(res['network'][0]['node'])
1226         self.assertEqual(nbNode,4)
1227         listNode=['ROADMA-SRG1','ROADMA-SRG3','ROADMA-DEG1','ROADMA-DEG2']
1228         for i in range(0,nbNode):
1229             self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'ROADMA'},
1230                           res['network'][0]['node'][i]['supporting-node'])
1231             nodeType=res['network'][0]['node'][i]['org-openroadm-network-topology:node-type']
1232             nodeId=res['network'][0]['node'][i]['node-id']
1233             if(nodeId=='ROADMA-SRG1'):
1234                 #Test related to SRG1
1235                 self.assertEqual(nodeType,'SRG')
1236                 self.assertEqual(len(res['network'][0]['node'][i]['ietf-network-topology:termination-point']),17)
1237                 self.assertIn({'tp-id': 'SRG1-CP-TXRX', 'org-openroadm-network-topology:tp-type': 'SRG-TXRX-CP'},
1238                               res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
1239                 self.assertIn({'tp-id': 'SRG1-PP1-TXRX', 'org-openroadm-network-topology:tp-type': 'SRG-TXRX-PP'},
1240                               res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
1241                 listNode.remove(nodeId)
1242             elif(nodeId=='ROADMA-SRG3'):
1243                 #Test related to SRG1
1244                 self.assertEqual(nodeType,'SRG')
1245                 self.assertEqual(len(res['network'][0]['node'][i]['ietf-network-topology:termination-point']),17)
1246                 self.assertIn({'tp-id': 'SRG3-CP-TXRX', 'org-openroadm-network-topology:tp-type': 'SRG-TXRX-CP'},
1247                               res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
1248                 self.assertIn({'tp-id': 'SRG3-PP1-TXRX', 'org-openroadm-network-topology:tp-type': 'SRG-TXRX-PP'},
1249                               res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
1250                 listNode.remove(nodeId)
1251             elif(nodeId=='ROADMA-DEG1'):
1252                 #Test related to DEG1
1253                 self.assertEqual(nodeType,'DEGREE')
1254                 self.assertIn({'tp-id': 'DEG1-TTP-TXRX', 'org-openroadm-network-topology:tp-type': 'DEGREE-TXRX-TTP'},
1255                               res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
1256                 self.assertIn({'tp-id': 'DEG1-CTP-TXRX', 'org-openroadm-network-topology:tp-type': 'DEGREE-TXRX-CTP'},
1257                               res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
1258                 listNode.remove(nodeId)
1259             elif(nodeId=='ROADMA-DEG2'):
1260                 #Test related to DEG2
1261                 self.assertEqual(nodeType,'DEGREE')
1262                 self.assertIn({'tp-id': 'DEG2-TTP-TXRX', 'org-openroadm-network-topology:tp-type': 'DEGREE-TXRX-TTP'},
1263                               res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
1264                 self.assertIn({'tp-id': 'DEG2-CTP-TXRX', 'org-openroadm-network-topology:tp-type': 'DEGREE-TXRX-CTP'},
1265                               res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
1266                 listNode.remove(nodeId)
1267             else:
1268                 self.assertFalse(True)
1269         self.assertEqual(len(listNode),0)
1270
1271     def test_39_disconnect_ROADM_XPDRA_link(self):
1272         #Link-1
1273         url = ("{}/config/ietf-network:networks/network/openroadm-topology/ietf-network-topology:"
1274                "link/XPDRA-XPDR1-XPDR1-NETWORK1toROADMA-SRG1-SRG1-PP1-TXRX"
1275                 .format(self.restconf_baseurl))
1276         data = {}
1277         headers = {'content-type': 'application/json'}
1278         response = requests.request(
1279              "DELETE", url, data=json.dumps(data), headers=headers,
1280              auth=('admin', 'admin'))
1281         self.assertEqual(response.status_code, requests.codes.ok)
1282         #Link-2
1283         url = ("{}/config/ietf-network:networks/network/openroadm-topology/ietf-network-topology:"
1284                "link/ROADMA-SRG1-SRG1-PP1-TXRXtoXPDRA-XPDR1-XPDR1-NETWORK1"
1285                 .format(self.restconf_baseurl))
1286         data = {}
1287         headers = {'content-type': 'application/json'}
1288         response = requests.request(
1289              "DELETE", url, data=json.dumps(data), headers=headers,
1290              auth=('admin', 'admin'))
1291         self.assertEqual(response.status_code, requests.codes.ok)
1292
1293     def test_40_getLinks_OpenRoadmTopology(self):
1294         url = ("{}/config/ietf-network:networks/network/openroadm-topology"
1295               .format(self.restconf_baseurl))
1296         headers = {'content-type': 'application/json'}
1297         response = requests.request(
1298             "GET", url, headers=headers, auth=('admin', 'admin'))
1299         self.assertEqual(response.status_code, requests.codes.ok)
1300         res = response.json()
1301         nbLink=len(res['network'][0]['ietf-network-topology:link'])
1302         self.assertEqual(nbLink,16)
1303         expressLink=['ROADMA-DEG2-DEG2-CTP-TXRXtoROADMA-DEG1-DEG1-CTP-TXRX','ROADMA-DEG1-DEG1-CTP-TXRXtoROADMA-DEG2-DEG2-CTP-TXRX']
1304         addLink=['ROADMA-SRG1-SRG1-CP-TXRXtoROADMA-DEG2-DEG2-CTP-TXRX','ROADMA-SRG1-SRG1-CP-TXRXtoROADMA-DEG1-DEG1-CTP-TXRX',
1305                  'ROADMA-SRG3-SRG3-CP-TXRXtoROADMA-DEG2-DEG2-CTP-TXRX','ROADMA-SRG3-SRG3-CP-TXRXtoROADMA-DEG1-DEG1-CTP-TXRX']
1306         dropLink=['ROADMA-DEG1-DEG1-CTP-TXRXtoROADMA-SRG1-SRG1-CP-TXRX','ROADMA-DEG2-DEG2-CTP-TXRXtoROADMA-SRG1-SRG1-CP-TXRX',
1307                   'ROADMA-DEG1-DEG1-CTP-TXRXtoROADMA-SRG3-SRG3-CP-TXRX','ROADMA-DEG2-DEG2-CTP-TXRXtoROADMA-SRG3-SRG3-CP-TXRX']
1308         roadmtoroadmLink = 0
1309         for i in range(0,nbLink):
1310             if (res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-network-topology:link-type']=='EXPRESS-LINK'):
1311                 link_id = res['network'][0]['ietf-network-topology:link'][i]['link-id']
1312                 find= link_id in expressLink
1313                 self.assertEqual(find, True)
1314                 expressLink.remove(link_id)
1315             elif (res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-network-topology:link-type']=='ADD-LINK'):
1316                 link_id = res['network'][0]['ietf-network-topology:link'][i]['link-id']
1317                 find= link_id in addLink
1318                 self.assertEqual(find, True)
1319                 addLink.remove(link_id)
1320             elif (res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-network-topology:link-type']=='DROP-LINK'):
1321                 link_id = res['network'][0]['ietf-network-topology:link'][i]['link-id']
1322                 find= link_id in dropLink
1323                 self.assertEqual(find, True)
1324                 dropLink.remove(link_id)
1325             else:
1326                 roadmtoroadmLink += 1
1327         self.assertEqual(len(expressLink),0)
1328         self.assertEqual(len(addLink),0)
1329         self.assertEqual(len(dropLink),0)
1330         self.assertEqual(roadmtoroadmLink, 6)
1331         for i in range(0,nbLink):
1332             self.assertNotEqual(res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-network-topology:link-type'],'XPONDER-OUTPUT')
1333             self.assertNotEqual(res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-network-topology:link-type'],'XPONDER-INPUT')
1334
1335     def test_41_disconnect_ROADMA(self):
1336         url = ("{}/config/network-topology:"
1337                 "network-topology/topology/topology-netconf/node/ROADMA"
1338                .format(self.restconf_baseurl))
1339         data = {}
1340         headers = {'content-type': 'application/json'}
1341         response = requests.request(
1342              "DELETE", url, data=json.dumps(data), headers=headers,
1343              auth=('admin', 'admin'))
1344         self.assertEqual(response.status_code, requests.codes.ok)
1345         #Delete in the clli-network
1346         url = ("{}/config/ietf-network:networks/network/clli-network/node/NodeA"
1347                .format(self.restconf_baseurl))
1348         data = {}
1349         headers = {'content-type': 'application/json'}
1350         response = requests.request(
1351              "DELETE", url, data=json.dumps(data), headers=headers,
1352              auth=('admin', 'admin'))
1353         self.assertEqual(response.status_code, requests.codes.ok)
1354
1355     def test_42_getClliNetwork(self):
1356         url = ("{}/config/ietf-network:networks/network/clli-network"
1357               .format(self.restconf_baseurl))
1358         headers = {'content-type': 'application/json'}
1359         response = requests.request(
1360             "GET", url, headers=headers, auth=('admin', 'admin'))
1361         self.assertEqual(response.status_code, requests.codes.ok)
1362         res = response.json()
1363         self.assertNotIn('node', res['network'][0])
1364
1365     def test_43_getOpenRoadmNetwork(self):
1366         url = ("{}/config/ietf-network:networks/network/openroadm-network"
1367               .format(self.restconf_baseurl))
1368         headers = {'content-type': 'application/json'}
1369         response = requests.request(
1370             "GET", url, headers=headers, auth=('admin', 'admin'))
1371         self.assertEqual(response.status_code, requests.codes.ok)
1372         res = response.json()
1373         self.assertNotIn('node', res['network'][0])
1374
1375     def test_44_check_roadm2roadm_link_persistence(self):
1376         url = ("{}/config/ietf-network:networks/network/openroadm-topology"
1377               .format(self.restconf_baseurl))
1378         headers = {'content-type': 'application/json'}
1379         response = requests.request(
1380             "GET", url, headers=headers, auth=('admin', 'admin'))
1381         self.assertEqual(response.status_code, requests.codes.ok)
1382         res = response.json()
1383         nbLink=len(res['network'][0]['ietf-network-topology:link'])
1384         self.assertNotIn('node', res['network'][0])
1385         self.assertEqual(nbLink, 6)
1386
1387 if __name__ == "__main__":
1388     unittest.main(verbosity=2)