3 # Copyright 2016 RIFT.IO Inc
5 # Licensed under the Apache License, Version 2.0 (the "License");
6 # you may not use this file except in compliance with the License.
7 # You may obtain a copy of the License at
9 # http://www.apache.org/licenses/LICENSE-2.0
11 # Unless required by applicable law or agreed to in writing, software
12 # distributed under the License is distributed on an "AS IS" BASIS,
13 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 # See the License for the specific language governing permissions and
15 # limitations under the License.
24 # from gi.repository import IetfNetworkYang
25 from gi
.repository
import IetfL2TopologyYang
as l2Tl
26 from gi
.repository
import RwTopologyYang
as RwTl
27 # from gi.repository.RwTypes import RwStatus
29 from create_stackedl2topology
import MyL2Topology
31 from rift
.topmgr
import (
34 logger
= logging
.getLogger('sdntop')
41 class SdnTopStoreNetworkTest(unittest
.TestCase
):
44 Initialize Top data store
46 self
._nwtopdata
_store
= NwtopDataStore(logger
)
47 self
.test_nwtop
= RwTl
.YangData_IetfNetwork()
49 self
.l2top
= MyL2Topology(self
.test_nwtop
, logger
)
50 self
.l2top
.setup_all()
52 # Get initial test data
53 self
.l2net1
= self
.l2top
.find_nw("L2HostNetwork-1")
55 self
._nwtopdata
_store
.create_network("L2HostNetwork-1", self
.l2net1
)
58 self
.l2net1
= self
.l2top
.find_nw("L2HostNetwork-1")
59 assert self
.l2net1
is not None
60 self
.new_l2net
= RwTl
.YangData_IetfNetwork_Network()
61 self
.new_l2net
.network_id
= "L2HostNetwork-2"
62 logger
.info("SdnTopStoreNetworkTest: setUp")
67 logger
.info("SdnTopStoreNetworkTest: Done with tests")
69 def test_create_network(self
):
71 Test: Create first l2 network
75 logger
.debug("SdnTopStoreNetworkTest: Create network ")
77 # Created durign setup phase
78 assert self
.l2net1
is not None
80 # Network already stored
82 nw
= self
._nwtopdata
_store
.get_network("L2HostNetwork-1")
83 logger
.debug("...Network id %s", nw
.network_id
)
84 self
.assertEqual(nw
.network_id
, "L2HostNetwork-1")
85 logger
.debug("...Network name %s", nw
.l2_network_attributes
.name
)
87 logger
.debug("...Node id %s", node
.node_id
)
89 for tp
in node
.termination_point
:
90 logger
.debug("...Tp id %s", tp
.tp_id
)
92 self
.assertEqual(num_nodes
, NUM_NODES_L2_NW
)
93 self
.assertEqual(num_tps
, NUM_TPS_L2_NW
)
96 def test_add_network(self
):
98 Test: Add another network, Check network id
100 logger
.debug("SdnTopStoreNetworkTest: Add network ")
101 # Use data store APIs
102 self
._nwtopdata
_store
.create_network("L2HostNetwork-2", self
.new_l2net
)
103 # Verify data created
104 nw
= self
._nwtopdata
_store
.get_network("L2HostNetwork-2")
105 self
.assertIsNotNone(nw
)
106 self
.assertEqual(nw
.network_id
, "L2HostNetwork-2")
107 self
.assertEqual(len(self
._nwtopdata
_store
._networks
), 2)
109 def test_add_networktype(self
):
111 Test: Add another network, Check network type
113 logger
.debug("SdnTopStoreTest: Add network type ")
114 # Use data store APIs
115 self
._nwtopdata
_store
.create_network("L2HostNetwork-2", self
.new_l2net
)
116 # Verify data created
117 nw
= self
._nwtopdata
_store
.get_network("L2HostNetwork-2")
118 self
.assertIsNotNone(nw
)
119 self
.assertEqual(nw
.network_id
, "L2HostNetwork-2")
120 self
.assertEqual(len(self
._nwtopdata
_store
._networks
), 2)
122 self
.new_l2net
.network_types
.l2_network
= self
.new_l2net
.network_types
.l2_network
.new()
123 logger
.debug("Adding update l2net..%s", self
.new_l2net
)
124 # Use data store APIs
125 self
._nwtopdata
_store
.update_network("L2HostNetwork-2", self
.new_l2net
)
126 # Verify data created
127 nw
= self
._nwtopdata
_store
.get_network("L2HostNetwork-2")
128 self
.assertIsNotNone(nw
.network_types
.l2_network
)
130 def test_add_networkl2name(self
):
132 Test: Add another network, Check L2 network name
134 logger
.debug("SdnTopStoreTest: Add L2 network name ")
135 # Use data store APIs
136 self
.new_l2net
.network_types
.l2_network
= self
.new_l2net
.network_types
.l2_network
.new()
137 self
._nwtopdata
_store
.create_network("L2HostNetwork-2", self
.new_l2net
)
138 # Verify data created
139 nw
= self
._nwtopdata
_store
.get_network("L2HostNetwork-2")
140 self
.assertIsNotNone(nw
)
141 self
.assertEqual(nw
.network_id
, "L2HostNetwork-2")
142 self
.assertEqual(len(self
._nwtopdata
_store
._networks
), 2)
144 self
.new_l2net
.l2_network_attributes
.name
= "L2networkName"
145 logger
.debug("Adding update l2net..%s", self
.new_l2net
)
146 # Use data store APIs
147 self
._nwtopdata
_store
.update_network("L2HostNetwork-2", self
.new_l2net
)
148 # Verify data created
149 nw
= self
._nwtopdata
_store
.get_network("L2HostNetwork-2")
150 self
.assertEqual(nw
.l2_network_attributes
.name
, "L2networkName")
153 class SdnTopStoreNetworkNodeTest(unittest
.TestCase
):
156 Initialize Top data store
158 self
._nwtopdata
_store
= NwtopDataStore(logger
)
159 self
.test_nwtop
= RwTl
.YangData_IetfNetwork()
161 self
.l2top
= MyL2Topology(self
.test_nwtop
, logger
)
162 self
.l2top
.setup_all()
164 # Get initial test data
165 self
.l2net1
= self
.l2top
.find_nw("L2HostNetwork-1")
167 self
._nwtopdata
_store
.create_network("L2HostNetwork-1", self
.l2net1
)
169 self
.l2net1
= self
.l2top
.find_nw("L2HostNetwork-1")
170 assert self
.l2net1
is not None
171 self
.new_l2net
= RwTl
.YangData_IetfNetwork_Network()
172 self
.new_l2net
.network_id
= "L2HostNetwork-1"
173 self
.node2
= self
.new_l2net
.node
.add()
174 self
.node2
.node_id
= "TempNode2"
175 logger
.info("SdnTopStoreTest: setUp NetworkNodetest")
178 logger
.info("SdnTopStoreTest: Done with NetworkNodetest")
181 def test_add_network_node(self
):
183 Test: Add a node to existing network
188 logger
.debug("SdnTopStoreTest: Add network node")
190 self
.node2
.node_id
= "TempNode2"
191 # Use data store APIs
192 self
._nwtopdata
_store
.update_network("L2HostNetwork-1", self
.new_l2net
)
193 # Verify data created
194 nw
= self
._nwtopdata
_store
.get_network("L2HostNetwork-1")
195 self
.assertIsNotNone(nw
)
196 self
.assertEqual(nw
.network_id
, "L2HostNetwork-1")
197 self
.assertEqual(len(self
._nwtopdata
_store
._networks
), 1)
198 self
.assertEqual(len(nw
.node
), NUM_NODES_L2_NW
+ 1)
199 self
.assertEqual(nw
.node
[NUM_NODES_L2_NW
].node_id
, "TempNode2")
201 #@unittest.skip("Skipping")
202 def test_update_network_node(self
):
204 Test: Updat a node to existing network
208 logger
.debug("SdnTopStoreTest: Update network node")
210 self
.node2
.node_id
= "TempNode2"
211 self
.node2
.l2_node_attributes
.description
= "TempNode2 desc"
212 self
.node2
.l2_node_attributes
.name
= "Nice Name2"
213 # Use data store APIs
214 self
._nwtopdata
_store
.update_network("L2HostNetwork-1", self
.new_l2net
)
215 # Verify data created
216 nw
= self
._nwtopdata
_store
.get_network("L2HostNetwork-1")
217 self
.assertIsNotNone(nw
)
218 self
.assertEqual(nw
.network_id
, "L2HostNetwork-1")
219 self
.assertEqual(len(self
._nwtopdata
_store
._networks
), 1)
220 self
.assertEqual(len(nw
.node
), NUM_NODES_L2_NW
+ 1)
221 self
.assertEqual(nw
.node
[NUM_NODES_L2_NW
].node_id
, "TempNode2")
222 self
.assertEqual(nw
.node
[NUM_NODES_L2_NW
].l2_node_attributes
.description
, "TempNode2 desc")
223 self
.assertEqual(nw
.node
[NUM_NODES_L2_NW
].l2_node_attributes
.name
, "Nice Name2")
225 #@unittest.skip("Skipping")
226 def test_update_network_node_l2attr1(self
):
228 Test: Update a node to existing network
232 logger
.debug("SdnTopStoreTest: Update network node")
234 self
.node2
.node_id
= "TempNode2"
235 self
.node2
.l2_node_attributes
.description
= "TempNode2 desc"
236 self
.node2
.l2_node_attributes
.name
= "Nice Name3"
237 # Use data store APIs
238 self
._nwtopdata
_store
.update_network("L2HostNetwork-1", self
.new_l2net
)
239 # Verify data created
240 nw
= self
._nwtopdata
_store
.get_network("L2HostNetwork-1")
241 self
.assertIsNotNone(nw
)
242 self
.assertEqual(nw
.network_id
, "L2HostNetwork-1")
243 self
.assertEqual(len(self
._nwtopdata
_store
._networks
), 1)
244 self
.assertEqual(len(nw
.node
), NUM_NODES_L2_NW
+ 1)
245 self
.assertEqual(nw
.node
[NUM_NODES_L2_NW
].node_id
, "TempNode2")
246 self
.assertEqual(nw
.node
[NUM_NODES_L2_NW
].l2_node_attributes
.description
, "TempNode2 desc")
247 self
.assertEqual(nw
.node
[NUM_NODES_L2_NW
].l2_node_attributes
.name
, "Nice Name3")
250 self
.node2
.l2_node_attributes
.name
= "Nice Name4"
251 logger
.debug("Network %s", self
.new_l2net
)
252 # Use data store APIs
253 self
._nwtopdata
_store
.update_network("L2HostNetwork-1", self
.new_l2net
)
254 # Verify data created
255 nw
= self
._nwtopdata
_store
.get_network("L2HostNetwork-1")
256 logger
.debug("Node %s", nw
.node
[NUM_NODES_L2_NW
])
257 self
.assertEqual(nw
.node
[NUM_NODES_L2_NW
].l2_node_attributes
.name
, "Nice Name4")
259 def test_update_network_node_l2attr2(self
):
261 Test: Updat a node to existing network
265 logger
.debug("SdnTopStoreTest: Update network node")
267 self
.node2
.node_id
= "TempNode2"
268 self
.node2
.l2_node_attributes
.description
= "TempNode2 desc"
269 self
.node2
.l2_node_attributes
.name
= "Nice Name3"
270 # Use data store APIs
271 self
._nwtopdata
_store
.update_network("L2HostNetwork-1", self
.new_l2net
)
272 # Verify data created
273 nw
= self
._nwtopdata
_store
.get_network("L2HostNetwork-1")
274 self
.assertIsNotNone(nw
)
275 self
.assertEqual(nw
.network_id
, "L2HostNetwork-1")
276 self
.assertEqual(len(self
._nwtopdata
_store
._networks
), 1)
277 self
.assertEqual(len(nw
.node
), NUM_NODES_L2_NW
+ 1)
278 self
.assertEqual(nw
.node
[NUM_NODES_L2_NW
].node_id
, "TempNode2")
279 self
.assertEqual(nw
.node
[NUM_NODES_L2_NW
].l2_node_attributes
.description
, "TempNode2 desc")
280 self
.assertEqual(nw
.node
[NUM_NODES_L2_NW
].l2_node_attributes
.name
, "Nice Name3")
283 self
.node2
.l2_node_attributes
.management_address
.append("10.0.0.1")
284 logger
.debug("Network %s", self
.new_l2net
)
285 # Use data store APIs
286 self
._nwtopdata
_store
.update_network("L2HostNetwork-1", self
.new_l2net
)
287 # Verify data created
288 nw
= self
._nwtopdata
_store
.get_network("L2HostNetwork-1")
289 self
.assertEqual(nw
.node
[NUM_NODES_L2_NW
].l2_node_attributes
.name
, "Nice Name3")
290 self
.assertEqual(len(nw
.node
[NUM_NODES_L2_NW
].l2_node_attributes
.management_address
), 1)
293 self
.node2
.l2_node_attributes
.management_address
.append("10.0.0.2")
294 logger
.debug("Network %s", self
.new_l2net
)
295 # Use data store APIs
296 self
._nwtopdata
_store
.update_network("L2HostNetwork-1", self
.new_l2net
)
297 # Verify data created
298 nw
= self
._nwtopdata
_store
.get_network("L2HostNetwork-1")
299 self
.assertEqual(nw
.node
[NUM_NODES_L2_NW
].l2_node_attributes
.name
, "Nice Name3")
300 self
.assertEqual(len(nw
.node
[NUM_NODES_L2_NW
].l2_node_attributes
.management_address
), 2)
303 class SdnTopStoreNetworkNodeTpTest(unittest
.TestCase
):
306 Initialize Top data store
308 self
._nwtopdata
_store
= NwtopDataStore(logger
)
309 self
.test_nwtop
= RwTl
.YangData_IetfNetwork()
311 self
.l2top
= MyL2Topology(self
.test_nwtop
, logger
)
312 self
.l2top
.setup_all()
314 # Get initial test data
315 self
.l2net1
= self
.l2top
.find_nw("L2HostNetwork-1")
317 self
._nwtopdata
_store
.create_network("L2HostNetwork-1", self
.l2net1
)
319 self
.l2net1
= self
.l2top
.find_nw("L2HostNetwork-1")
320 assert self
.l2net1
is not None
321 self
.new_l2net
= RwTl
.YangData_IetfNetwork_Network()
322 self
.new_l2net
.network_id
= "L2HostNetwork-1"
323 self
.node2
= self
.new_l2net
.node
.add()
324 self
.node2
.node_id
= "TempNode2"
325 self
.tp1
= self
.node2
.termination_point
.add()
326 self
.tp1
.tp_id
= "TempTp1"
327 logger
.info("SdnTopStoreTest: setUp NetworkNodeTptest")
330 logger
.info("SdnTopStoreTest: Done with NetworkNodeTptest")
332 self
.new_l2net
= None
336 def test_add_network_node_tp(self
):
338 Test: Add a node to existing network
342 logger
.debug("SdnTopStoreTest: Update network ")
343 # Use data store APIs
344 self
._nwtopdata
_store
.update_network("L2HostNetwork-1", self
.new_l2net
)
345 # Verify data created
346 nw
= self
._nwtopdata
_store
.get_network("L2HostNetwork-1")
347 self
.assertIsNotNone(nw
)
348 self
.assertEqual(nw
.network_id
, "L2HostNetwork-1")
349 self
.assertEqual(len(self
._nwtopdata
_store
._networks
), 1)
350 self
.assertEqual(len(nw
.node
), NUM_NODES_L2_NW
+ 1)
351 self
.assertEqual(nw
.node
[NUM_NODES_L2_NW
].node_id
, "TempNode2")
352 self
.assertEqual(len(nw
.node
[NUM_NODES_L2_NW
].termination_point
), 1)
353 self
.assertEqual(nw
.node
[NUM_NODES_L2_NW
].termination_point
[0].tp_id
, "TempTp1")
355 def test_update_network_node_tp(self
):
357 Test: Update a tp to existing network, add all tp elements
361 logger
.debug("SdnTopStoreTest: Update network ")
362 self
.tp1
.tp_id
= "TempTp1"
363 self
.tp1
.l2_termination_point_attributes
.description
= "TempTp1 Desc"
364 self
.tp1
.l2_termination_point_attributes
.maximum_frame_size
= 1296
365 self
.tp1
.l2_termination_point_attributes
.mac_address
= "00:1e:67:98:28:01"
366 self
.tp1
.l2_termination_point_attributes
.eth_encapsulation
= "l2t:ethernet"
367 # Use data store APIs
368 self
._nwtopdata
_store
.update_network("L2HostNetwork-1", self
.new_l2net
)
369 # Verify data created
370 nw
= self
._nwtopdata
_store
.get_network("L2HostNetwork-1")
371 self
.assertIsNotNone(nw
)
372 self
.assertEqual(nw
.network_id
, "L2HostNetwork-1")
373 self
.assertEqual(len(self
._nwtopdata
_store
._networks
), 1)
374 self
.assertEqual(len(nw
.node
), NUM_NODES_L2_NW
+ 1)
375 self
.assertEqual(nw
.node
[NUM_NODES_L2_NW
].node_id
, "TempNode2")
376 self
.assertEqual(len(nw
.node
[NUM_NODES_L2_NW
].termination_point
), 1)
377 self
.assertEqual(nw
.node
[NUM_NODES_L2_NW
].termination_point
[0].tp_id
, "TempTp1")
378 self
.assertEqual(nw
.node
[NUM_NODES_L2_NW
].termination_point
[0].l2_termination_point_attributes
.description
, "TempTp1 Desc")
379 self
.assertEqual(nw
.node
[NUM_NODES_L2_NW
].termination_point
[0].l2_termination_point_attributes
.maximum_frame_size
, 1296)
380 self
.assertEqual(nw
.node
[NUM_NODES_L2_NW
].termination_point
[0].l2_termination_point_attributes
.mac_address
, "00:1e:67:98:28:01")
382 def test_update_network_node_tp2(self
):
384 Test: Update a tp to existing network, change tp elements
388 logger
.debug("SdnTopStoreTest: Update network ")
389 self
.tp1
.tp_id
= "TempTp1"
390 self
.tp1
.l2_termination_point_attributes
.description
= "TempTp1 Desc"
391 self
.tp1
.l2_termination_point_attributes
.maximum_frame_size
= 1296
392 self
.tp1
.l2_termination_point_attributes
.mac_address
= "00:1e:67:98:28:01"
393 # Use data store APIs
394 self
._nwtopdata
_store
.update_network("L2HostNetwork-1", self
.new_l2net
)
395 # Verify data created
396 nw
= self
._nwtopdata
_store
.get_network("L2HostNetwork-1")
397 self
.assertIsNotNone(nw
)
398 self
.assertEqual(nw
.network_id
, "L2HostNetwork-1")
399 self
.assertEqual(len(self
._nwtopdata
_store
._networks
), 1)
400 self
.assertEqual(len(nw
.node
), NUM_NODES_L2_NW
+ 1)
401 self
.assertEqual(nw
.node
[NUM_NODES_L2_NW
].node_id
, "TempNode2")
402 self
.assertEqual(len(nw
.node
[NUM_NODES_L2_NW
].termination_point
), 1)
403 self
.assertEqual(nw
.node
[NUM_NODES_L2_NW
].termination_point
[0].tp_id
, "TempTp1")
404 self
.assertEqual(nw
.node
[NUM_NODES_L2_NW
].termination_point
[0].l2_termination_point_attributes
.description
, "TempTp1 Desc")
405 self
.assertEqual(nw
.node
[NUM_NODES_L2_NW
].termination_point
[0].l2_termination_point_attributes
.maximum_frame_size
, 1296)
406 self
.assertEqual(nw
.node
[NUM_NODES_L2_NW
].termination_point
[0].l2_termination_point_attributes
.mac_address
, "00:1e:67:98:28:01")
409 self
.tp1
.l2_termination_point_attributes
.maximum_frame_size
= 1396
410 # Use data store APIs
411 self
._nwtopdata
_store
.update_network("L2HostNetwork-1", self
.new_l2net
)
412 # Verify data created
413 nw
= self
._nwtopdata
_store
.get_network("L2HostNetwork-1")
414 self
.assertIsNotNone(nw
)
415 self
.assertEqual(nw
.network_id
, "L2HostNetwork-1")
416 self
.assertEqual(len(self
._nwtopdata
_store
._networks
), 1)
417 self
.assertEqual(len(nw
.node
), NUM_NODES_L2_NW
+ 1)
418 self
.assertEqual(nw
.node
[NUM_NODES_L2_NW
].node_id
, "TempNode2")
419 self
.assertEqual(len(nw
.node
[NUM_NODES_L2_NW
].termination_point
), 1)
420 self
.assertEqual(nw
.node
[NUM_NODES_L2_NW
].termination_point
[0].tp_id
, "TempTp1")
421 self
.assertEqual(nw
.node
[NUM_NODES_L2_NW
].termination_point
[0].l2_termination_point_attributes
.description
, "TempTp1 Desc")
422 self
.assertEqual(nw
.node
[NUM_NODES_L2_NW
].termination_point
[0].l2_termination_point_attributes
.maximum_frame_size
, 1396)
423 self
.assertEqual(nw
.node
[NUM_NODES_L2_NW
].termination_point
[0].l2_termination_point_attributes
.mac_address
, "00:1e:67:98:28:01")
426 self
.tp1
.l2_termination_point_attributes
.mac_address
= "00:1e:67:98:28:02"
427 # Use data store APIs
428 self
._nwtopdata
_store
.update_network("L2HostNetwork-1", self
.new_l2net
)
429 # Verify data created
430 nw
= self
._nwtopdata
_store
.get_network("L2HostNetwork-1")
431 self
.assertIsNotNone(nw
)
432 self
.assertEqual(nw
.network_id
, "L2HostNetwork-1")
433 self
.assertEqual(len(self
._nwtopdata
_store
._networks
), 1)
434 self
.assertEqual(len(nw
.node
), NUM_NODES_L2_NW
+ 1)
435 self
.assertEqual(nw
.node
[NUM_NODES_L2_NW
].node_id
, "TempNode2")
436 self
.assertEqual(len(nw
.node
[NUM_NODES_L2_NW
].termination_point
), 1)
437 self
.assertEqual(nw
.node
[NUM_NODES_L2_NW
].termination_point
[0].tp_id
, "TempTp1")
438 self
.assertEqual(nw
.node
[NUM_NODES_L2_NW
].termination_point
[0].l2_termination_point_attributes
.description
, "TempTp1 Desc")
439 self
.assertEqual(nw
.node
[NUM_NODES_L2_NW
].termination_point
[0].l2_termination_point_attributes
.maximum_frame_size
, 1396)
440 self
.assertEqual(nw
.node
[NUM_NODES_L2_NW
].termination_point
[0].l2_termination_point_attributes
.mac_address
, "00:1e:67:98:28:02")
442 # Add encapsulation type
443 self
.tp1
.l2_termination_point_attributes
.eth_encapsulation
= "l2t:ethernet"
444 # Use data store APIs
445 self
._nwtopdata
_store
.update_network("L2HostNetwork-1", self
.new_l2net
)
446 # Verify data created
447 nw
= self
._nwtopdata
_store
.get_network("L2HostNetwork-1")
448 self
.assertIsNotNone(nw
)
449 self
.assertEqual(nw
.network_id
, "L2HostNetwork-1")
450 self
.assertEqual(len(self
._nwtopdata
_store
._networks
), 1)
451 self
.assertEqual(len(nw
.node
), NUM_NODES_L2_NW
+ 1)
452 self
.assertEqual(nw
.node
[NUM_NODES_L2_NW
].node_id
, "TempNode2")
453 self
.assertEqual(len(nw
.node
[NUM_NODES_L2_NW
].termination_point
), 1)
454 self
.assertEqual(nw
.node
[NUM_NODES_L2_NW
].termination_point
[0].tp_id
, "TempTp1")
455 self
.assertEqual(nw
.node
[NUM_NODES_L2_NW
].termination_point
[0].l2_termination_point_attributes
.description
, "TempTp1 Desc")
456 self
.assertEqual(nw
.node
[NUM_NODES_L2_NW
].termination_point
[0].l2_termination_point_attributes
.maximum_frame_size
, 1396)
457 self
.assertEqual(nw
.node
[NUM_NODES_L2_NW
].termination_point
[0].l2_termination_point_attributes
.mac_address
, "00:1e:67:98:28:02")
458 self
.assertEqual(nw
.node
[NUM_NODES_L2_NW
].termination_point
[0].l2_termination_point_attributes
.eth_encapsulation
, "l2t:ethernet")
460 def test_update_extra_network_node_tp2(self
):
462 Test: Update a tp to existing network, change tp elements
466 logger
.debug("SdnTopStoreTest: Update network ")
467 self
.tp2
= self
.node2
.termination_point
.add()
468 self
.tp2
.tp_id
= "TempTp2"
469 # Use data store APIs
470 self
._nwtopdata
_store
.update_network("L2HostNetwork-1", self
.new_l2net
)
471 # Verify data created
472 nw
= self
._nwtopdata
_store
.get_network("L2HostNetwork-1")
473 self
.assertIsNotNone(nw
)
474 self
.assertEqual(nw
.network_id
, "L2HostNetwork-1")
475 self
.assertEqual(len(self
._nwtopdata
_store
._networks
), 1)
476 self
.assertEqual(len(nw
.node
), NUM_NODES_L2_NW
+ 1)
477 self
.assertEqual(nw
.node
[NUM_NODES_L2_NW
].node_id
, "TempNode2")
478 self
.assertEqual(len(nw
.node
[NUM_NODES_L2_NW
].termination_point
), 2)
479 self
.assertEqual(nw
.node
[NUM_NODES_L2_NW
].termination_point
[1].tp_id
, "TempTp2")
483 class SdnTopStoreNetworkLinkTest(unittest
.TestCase
):
486 Initialize Top data store
488 self
._nwtopdata
_store
= NwtopDataStore(logger
)
489 self
.test_nwtop
= RwTl
.YangData_IetfNetwork()
491 self
.l2top
= MyL2Topology(self
.test_nwtop
, logger
)
492 self
.l2top
.setup_all()
494 # Get initial test data
495 self
.l2net1
= self
.l2top
.find_nw("L2HostNetwork-1")
497 self
._nwtopdata
_store
.create_network("L2HostNetwork-1", self
.l2net1
)
499 self
.l2net1
= self
.l2top
.find_nw("L2HostNetwork-1")
500 assert self
.l2net1
is not None
501 self
.new_l2net
= RwTl
.YangData_IetfNetwork_Network()
502 self
.new_l2net
.network_id
= "L2HostNetwork-1"
504 self
.src_node
= self
.new_l2net
.node
.add()
505 self
.src_node
.node_id
= "TempNode1"
506 self
.tp1
= self
.src_node
.termination_point
.add()
507 self
.tp1
.tp_id
= "TempTp1"
509 self
.dest_node
= self
.new_l2net
.node
.add()
510 self
.dest_node
.node_id
= "TempNode2"
511 self
.tp2
= self
.dest_node
.termination_point
.add()
512 self
.tp2
.tp_id
= "TempTp2"
513 logger
.info("SdnTopStoreTest: setUp NetworkLinkTest")
516 logger
.info("SdnTopStoreTest: Done with NetworkLinkTest")
518 self
.new_l2net
= None
521 self
.dest_node
= None
524 def test_add_network_link(self
):
526 Test: Add a link to existing network
528 logger
.info("SdnTopStoreTest: Update network link")
529 self
._nwtopdata
_store
.update_network("L2HostNetwork-1", self
.new_l2net
)
530 # Verify data created
531 nw
= self
._nwtopdata
_store
.get_network("L2HostNetwork-1")
532 self
.assertIsNotNone(nw
)
533 self
.assertEqual(nw
.network_id
, "L2HostNetwork-1")
534 self
.assertEqual(len(self
._nwtopdata
_store
._networks
), 1)
535 self
.assertEqual(len(nw
.node
), NUM_NODES_L2_NW
+ 2)
536 self
.assertEqual(nw
.node
[NUM_NODES_L2_NW
].node_id
, "TempNode1")
537 self
.assertEqual(len(nw
.node
[NUM_NODES_L2_NW
].termination_point
), 1)
538 self
.assertEqual(nw
.node
[NUM_NODES_L2_NW
].termination_point
[0].tp_id
, "TempTp1")
539 self
.assertEqual(nw
.node
[NUM_NODES_L2_NW
+1].node_id
, "TempNode2")
540 self
.assertEqual(nw
.node
[NUM_NODES_L2_NW
+1].termination_point
[0].tp_id
, "TempTp2")
541 self
.assertEqual(len(nw
.link
), NUM_LINKS
)
542 self
.link1
= self
.new_l2net
.link
.add()
543 self
.link1
.link_id
= "Link1"
544 self
.link1
.source
.source_node
= self
.src_node
.node_id
545 self
.link1
.source
.source_tp
= self
.tp1
.tp_id
546 self
.link1
.destination
.dest_node
= self
.dest_node
.node_id
547 self
.link1
.destination
.dest_tp
= self
.tp2
.tp_id
548 # Use data store APIs
549 logger
.info("SdnTopStoreTest: Update network link - Part 2")
550 self
._nwtopdata
_store
.update_network("L2HostNetwork-1", self
.new_l2net
)
551 nw
= self
._nwtopdata
_store
.get_network("L2HostNetwork-1")
552 # Verify data created
553 self
.assertEqual(nw
.link
[NUM_LINKS
].link_id
, "Link1")
554 self
.assertEqual(nw
.link
[NUM_LINKS
].source
.source_node
, self
.src_node
.node_id
)
555 self
.assertEqual(nw
.link
[NUM_LINKS
].source
.source_tp
, self
.tp1
.tp_id
)
556 self
.assertEqual(nw
.link
[NUM_LINKS
].destination
.dest_node
, self
.dest_node
.node_id
)
557 self
.assertEqual(nw
.link
[NUM_LINKS
].destination
.dest_tp
, self
.tp2
.tp_id
)
558 self
.assertEqual(len(nw
.link
), NUM_LINKS
+ 1)
560 def test_add_extra_network_link(self
):
562 Test: Add a link to existing network
564 logger
.info("SdnTopStoreTest: Update extra network link")
565 # Create initial state
566 self
.link1
= self
.new_l2net
.link
.add()
567 self
.link1
.link_id
= "Link1"
568 self
.link1
.source
.source_node
= self
.src_node
.node_id
569 self
.link1
.source
.source_tp
= self
.tp1
.tp_id
570 self
.link1
.destination
.dest_node
= self
.dest_node
.node_id
571 self
.link1
.destination
.dest_tp
= self
.tp2
.tp_id
572 self
._nwtopdata
_store
.update_network("L2HostNetwork-1", self
.new_l2net
)
573 # Verify initial state
574 nw
= self
._nwtopdata
_store
.get_network("L2HostNetwork-1")
575 self
.assertIsNotNone(nw
)
576 self
.assertEqual(nw
.network_id
, "L2HostNetwork-1")
577 self
.assertEqual(len(self
._nwtopdata
_store
._networks
), 1)
578 self
.assertEqual(len(nw
.node
), NUM_NODES_L2_NW
+ 2)
579 self
.assertEqual(nw
.node
[NUM_NODES_L2_NW
].node_id
, "TempNode1")
580 self
.assertEqual(len(nw
.node
[NUM_NODES_L2_NW
].termination_point
), 1)
581 self
.assertEqual(nw
.node
[NUM_NODES_L2_NW
].termination_point
[0].tp_id
, "TempTp1")
582 self
.assertEqual(nw
.node
[NUM_NODES_L2_NW
+1].node_id
, "TempNode2")
583 self
.assertEqual(nw
.node
[NUM_NODES_L2_NW
+1].termination_point
[0].tp_id
, "TempTp2")
584 self
.assertEqual(nw
.link
[NUM_LINKS
].link_id
, "Link1")
585 self
.assertEqual(len(nw
.link
), NUM_LINKS
+ 1)
587 # Add extra link (reverse)
588 self
.link2
= self
.new_l2net
.link
.add()
589 self
.link2
.link_id
= "Link2"
590 self
.link2
.source
.source_node
= self
.dest_node
.node_id
591 self
.link2
.source
.source_tp
= self
.tp2
.tp_id
592 self
.link2
.destination
.dest_node
= self
.src_node
.node_id
593 self
.link2
.destination
.dest_tp
= self
.tp1
.tp_id
594 # Use data store APIs
595 logger
.info("SdnTopStoreTest: Update extra network link - Part 2")
596 self
._nwtopdata
_store
.update_network("L2HostNetwork-1", self
.new_l2net
)
597 nw
= self
._nwtopdata
_store
.get_network("L2HostNetwork-1")
598 # Verify data created
599 self
.assertEqual(nw
.link
[NUM_LINKS
+1].link_id
, "Link2")
600 self
.assertEqual(len(nw
.link
), NUM_LINKS
+ 2)
601 self
.assertEqual(nw
.link
[NUM_LINKS
+1].source
.source_node
, self
.dest_node
.node_id
)
602 self
.assertEqual(nw
.link
[NUM_LINKS
+1].source
.source_tp
, self
.tp2
.tp_id
)
603 self
.assertEqual(nw
.link
[NUM_LINKS
+1].destination
.dest_node
, self
.src_node
.node_id
)
604 self
.assertEqual(nw
.link
[NUM_LINKS
+1].destination
.dest_tp
, self
.tp1
.tp_id
)
606 def test_add_network_link_l2attr(self
):
608 Test: Check L2 link attributes
610 logger
.info("SdnTopStoreTest: Add network link L2 attributes")
612 self
.link1
= self
.new_l2net
.link
.add()
613 self
.link1
.link_id
= "Link1"
614 self
.link1
.source
.source_node
= self
.src_node
.node_id
615 self
.link1
.source
.source_tp
= self
.tp1
.tp_id
616 self
.link1
.destination
.dest_node
= self
.dest_node
.node_id
617 self
.link1
.destination
.dest_tp
= self
.tp2
.tp_id
618 self
.link1
.l2_link_attributes
.name
= "Link L2 name"
619 self
.link1
.l2_link_attributes
.rate
= 10000
620 self
._nwtopdata
_store
.update_network("L2HostNetwork-1", self
.new_l2net
)
622 nw
= self
._nwtopdata
_store
.get_network("L2HostNetwork-1")
623 self
.assertIsNotNone(nw
)
624 self
.assertEqual(nw
.network_id
, "L2HostNetwork-1")
625 self
.assertEqual(len(self
._nwtopdata
_store
._networks
), 1)
626 self
.assertEqual(len(nw
.node
), NUM_NODES_L2_NW
+ 2)
627 self
.assertEqual(nw
.node
[NUM_NODES_L2_NW
].node_id
, "TempNode1")
628 self
.assertEqual(len(nw
.node
[NUM_NODES_L2_NW
].termination_point
), 1)
629 self
.assertEqual(nw
.node
[NUM_NODES_L2_NW
].termination_point
[0].tp_id
, "TempTp1")
630 self
.assertEqual(nw
.node
[NUM_NODES_L2_NW
+1].node_id
, "TempNode2")
631 self
.assertEqual(nw
.node
[NUM_NODES_L2_NW
+1].termination_point
[0].tp_id
, "TempTp2")
632 self
.assertEqual(nw
.link
[NUM_LINKS
].link_id
, "Link1")
633 self
.assertEqual(len(nw
.link
), NUM_LINKS
+ 1)
634 self
.assertEqual(nw
.link
[NUM_LINKS
].l2_link_attributes
.name
, "Link L2 name")
635 self
.assertEqual(nw
.link
[NUM_LINKS
].l2_link_attributes
.rate
, 10000)
637 def test_change_network_link_l2attr(self
):
639 Test: Change L2 link attributes
641 logger
.info("SdnTopStoreTest: Change network link L2 attributes")
642 # Create initial state
643 self
.link1
= self
.new_l2net
.link
.add()
644 self
.link1
.link_id
= "Link1"
645 self
.link1
.source
.source_node
= self
.src_node
.node_id
646 self
.link1
.source
.source_tp
= self
.tp1
.tp_id
647 self
.link1
.destination
.dest_node
= self
.dest_node
.node_id
648 self
.link1
.destination
.dest_tp
= self
.tp2
.tp_id
649 self
.link1
.l2_link_attributes
.name
= "Link L2 name"
650 self
.link1
.l2_link_attributes
.rate
= 10000
651 self
._nwtopdata
_store
.update_network("L2HostNetwork-1", self
.new_l2net
)
652 # Verify initial state
653 nw
= self
._nwtopdata
_store
.get_network("L2HostNetwork-1")
654 self
.assertIsNotNone(nw
)
655 self
.assertEqual(nw
.network_id
, "L2HostNetwork-1")
656 self
.assertEqual(len(self
._nwtopdata
_store
._networks
), 1)
657 self
.assertEqual(len(nw
.node
), NUM_NODES_L2_NW
+ 2)
658 self
.assertEqual(nw
.node
[NUM_NODES_L2_NW
].node_id
, "TempNode1")
659 self
.assertEqual(len(nw
.node
[NUM_NODES_L2_NW
].termination_point
), 1)
660 self
.assertEqual(nw
.node
[NUM_NODES_L2_NW
].termination_point
[0].tp_id
, "TempTp1")
661 self
.assertEqual(nw
.node
[NUM_NODES_L2_NW
+1].node_id
, "TempNode2")
662 self
.assertEqual(nw
.node
[NUM_NODES_L2_NW
+1].termination_point
[0].tp_id
, "TempTp2")
663 self
.assertEqual(nw
.link
[NUM_LINKS
].link_id
, "Link1")
664 self
.assertEqual(len(nw
.link
), NUM_LINKS
+ 1)
665 self
.assertEqual(nw
.link
[NUM_LINKS
].l2_link_attributes
.name
, "Link L2 name")
666 self
.assertEqual(nw
.link
[NUM_LINKS
].l2_link_attributes
.rate
, 10000)
668 # Create initial state
669 self
.test_l2net
= RwTl
.YangData_IetfNetwork_Network()
670 self
.test_l2net
.network_id
= "L2HostNetwork-1"
671 self
.link1
= self
.test_l2net
.link
.add()
672 self
.link1
.link_id
= "Link1"
673 self
.link1
.l2_link_attributes
.name
= "Link L2 updated name"
674 self
._nwtopdata
_store
.update_network("L2HostNetwork-1", self
.test_l2net
)
676 nw
= self
._nwtopdata
_store
.get_network("L2HostNetwork-1")
677 self
.assertEqual(nw
.network_id
, "L2HostNetwork-1")
678 self
.assertEqual(nw
.link
[NUM_LINKS
].l2_link_attributes
.name
, "Link L2 updated name")
680 def test_change_network_link_dest_tp(self
):
682 Test: Change L2 link attributes
684 logger
.info("SdnTopStoreTest: Change network link dest-tp")
685 # Create initial state
686 self
.link1
= self
.new_l2net
.link
.add()
687 self
.link1
.link_id
= "Link1"
688 self
.link1
.source
.source_node
= self
.src_node
.node_id
689 self
.link1
.source
.source_tp
= self
.tp1
.tp_id
690 self
.link1
.destination
.dest_node
= self
.dest_node
.node_id
691 self
.link1
.destination
.dest_tp
= self
.tp2
.tp_id
692 self
.link1
.l2_link_attributes
.name
= "Link L2 name"
693 self
.link1
.l2_link_attributes
.rate
= 10000
694 self
._nwtopdata
_store
.update_network("L2HostNetwork-1", self
.new_l2net
)
695 # Verify initial state
696 nw
= self
._nwtopdata
_store
.get_network("L2HostNetwork-1")
697 self
.assertIsNotNone(nw
)
698 self
.assertEqual(nw
.network_id
, "L2HostNetwork-1")
699 self
.assertEqual(len(self
._nwtopdata
_store
._networks
), 1)
700 self
.assertEqual(len(nw
.node
), NUM_NODES_L2_NW
+ 2)
701 self
.assertEqual(nw
.node
[NUM_NODES_L2_NW
].node_id
, "TempNode1")
702 self
.assertEqual(len(nw
.node
[NUM_NODES_L2_NW
].termination_point
), 1)
703 self
.assertEqual(nw
.node
[NUM_NODES_L2_NW
].termination_point
[0].tp_id
, "TempTp1")
704 self
.assertEqual(nw
.node
[NUM_NODES_L2_NW
+1].node_id
, "TempNode2")
705 self
.assertEqual(nw
.node
[NUM_NODES_L2_NW
+1].termination_point
[0].tp_id
, "TempTp2")
706 self
.assertEqual(nw
.link
[NUM_LINKS
].link_id
, "Link1")
707 self
.assertEqual(len(nw
.link
), NUM_LINKS
+ 1)
708 self
.assertEqual(nw
.link
[NUM_LINKS
].l2_link_attributes
.name
, "Link L2 name")
709 self
.assertEqual(nw
.link
[NUM_LINKS
].l2_link_attributes
.rate
, 10000)
712 self
.test_l2net
= RwTl
.YangData_IetfNetwork_Network()
713 self
.test_l2net
.network_id
= "L2HostNetwork-1"
714 self
.link1
= self
.test_l2net
.link
.add()
715 self
.link1
.link_id
= "Link1"
716 # Changing dest node params
717 self
.link1
.destination
.dest_node
= self
.src_node
.node_id
718 self
.link1
.destination
.dest_tp
= self
.tp1
.tp_id
719 self
._nwtopdata
_store
.update_network("L2HostNetwork-1", self
.test_l2net
)
721 nw
= self
._nwtopdata
_store
.get_network("L2HostNetwork-1")
722 self
.assertEqual(nw
.network_id
, "L2HostNetwork-1")
723 self
.assertEqual(nw
.link
[NUM_LINKS
].destination
.dest_node
, self
.src_node
.node_id
)
728 if __name__
== "__main__":
729 logging
.basicConfig(level
=logging
.INFO
)