rwlaunchpad/plugins/rwvns/test/test_top_datastore.py

   1
   2 #
   3 #   Copyright 2016 RIFT.IO Inc
   4 #
   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
   8 #
   9 #       http://www.apache.org/licenses/LICENSE-2.0
  10 #
  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.
  16 #
  17
  18 import datetime
  19 import logging
  20 import unittest
  21
  22 import rwlogger
  23
  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
  28
  29 from create_stackedl2topology import MyL2Topology
  30
  31 from rift.topmgr import (
  32     NwtopDataStore,
  33 )
  34 logger = logging.getLogger('sdntop')
  35
  36 NUM_NWS = 1
  37 NUM_NODES_L2_NW = 6
  38 NUM_TPS_L2_NW = 20
  39 NUM_LINKS = 20
  40
  41 class SdnTopStoreNetworkTest(unittest.TestCase):
  42     def setUp(self):
  43         """
  44           Initialize Top data store
  45         """
  46         self._nwtopdata_store = NwtopDataStore(logger)
  47         self.test_nwtop = RwTl.YangData_IetfNetwork()
  48
  49         self.l2top = MyL2Topology(self.test_nwtop, logger)
  50         self.l2top.setup_all()
  51
  52         # Get initial test data
  53         self.l2net1 = self.l2top.find_nw("L2HostNetwork-1")
  54         # Create initial nw
  55         self._nwtopdata_store.create_network("L2HostNetwork-1", self.l2net1)
  56
  57         # Add test data
  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")
  63
  64     def tearDown(self):
  65         self.l2net1 = None
  66         self.new_l2net = None
  67         logger.info("SdnTopStoreNetworkTest: Done with tests")
  68
  69     def test_create_network(self):
  70         """
  71            Test: Create first l2 network
  72         """
  73         num_nodes = 0
  74         num_tps = 0
  75         logger.debug("SdnTopStoreNetworkTest: Create network ")
  76         # Get test data
  77         # Created durign setup phase
  78         assert self.l2net1 is not None
  79         # Use data store APIs
  80         # Network already stored
  81         # Verify data created
  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)
  86         for node in nw.node:
  87             logger.debug("...Node id %s", node.node_id)
  88             num_nodes += 1
  89             for tp in node.termination_point:
  90                 logger.debug("...Tp id %s", tp.tp_id)
  91                 num_tps += 1
  92         self.assertEqual(num_nodes, NUM_NODES_L2_NW)
  93         self.assertEqual(num_tps, NUM_TPS_L2_NW)
  94
  95
  96     def test_add_network(self):
  97         """
  98            Test: Add another network, Check network id
  99         """
 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)
 108
 109     def test_add_networktype(self):
 110         """
 111            Test: Add another network, Check network type
 112         """
 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)
 121         # Add new test data
 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)
 129
 130     def test_add_networkl2name(self):
 131         """
 132            Test: Add another network, Check L2 network name
 133         """
 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)
 143         # Add new test data
 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")
 151
 152
 153 class SdnTopStoreNetworkNodeTest(unittest.TestCase):
 154     def setUp(self):
 155         """
 156           Initialize Top data store
 157         """
 158         self._nwtopdata_store = NwtopDataStore(logger)
 159         self.test_nwtop = RwTl.YangData_IetfNetwork()
 160
 161         self.l2top = MyL2Topology(self.test_nwtop, logger)
 162         self.l2top.setup_all()
 163
 164         # Get initial test data
 165         self.l2net1 = self.l2top.find_nw("L2HostNetwork-1")
 166         # Create initial nw
 167         self._nwtopdata_store.create_network("L2HostNetwork-1", self.l2net1)
 168         # Get test data
 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")
 176
 177     def tearDown(self):
 178         logger.info("SdnTopStoreTest: Done with  NetworkNodetest")
 179
 180
 181     def test_add_network_node(self):
 182         """
 183            Test: Add a node to existing network
 184                  Test all parameters
 185         """
 186         num_nodes = 0
 187         num_tps = 0
 188         logger.debug("SdnTopStoreTest: Add network node")
 189         # Add test data
 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")
 200
 201     #@unittest.skip("Skipping")
 202     def test_update_network_node(self):
 203         """
 204            Test: Updat a node to existing network
 205         """
 206         num_nodes = 0
 207         num_tps = 0
 208         logger.debug("SdnTopStoreTest: Update network node")
 209         # Add test data
 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")
 224
 225     #@unittest.skip("Skipping")
 226     def test_update_network_node_l2attr1(self):
 227         """
 228            Test: Update a node to existing network
 229         """
 230         num_nodes = 0
 231         num_tps = 0
 232         logger.debug("SdnTopStoreTest: Update network node")
 233         # Add test data
 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")
 248
 249         # Add test data
 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")
 258
 259     def test_update_network_node_l2attr2(self):
 260         """
 261            Test: Updat a node to existing network
 262         """
 263         num_nodes = 0
 264         num_tps = 0
 265         logger.debug("SdnTopStoreTest: Update network node")
 266         # Add test data
 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")
 281
 282         # Add test data
 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)
 291
 292         # Add test data
 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)
 301
 302
 303 class SdnTopStoreNetworkNodeTpTest(unittest.TestCase):
 304     def setUp(self):
 305         """
 306           Initialize Top data store
 307         """
 308         self._nwtopdata_store = NwtopDataStore(logger)
 309         self.test_nwtop = RwTl.YangData_IetfNetwork()
 310
 311         self.l2top = MyL2Topology(self.test_nwtop, logger)
 312         self.l2top.setup_all()
 313
 314         # Get initial test data
 315         self.l2net1 = self.l2top.find_nw("L2HostNetwork-1")
 316         # Create initial nw
 317         self._nwtopdata_store.create_network("L2HostNetwork-1", self.l2net1)
 318         # Get test data
 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")
 328
 329     def tearDown(self):
 330         logger.info("SdnTopStoreTest: Done with  NetworkNodeTptest")
 331
 332         self.new_l2net = None
 333         self.node2 = None
 334         self.tp1 = None
 335
 336     def test_add_network_node_tp(self):
 337         """
 338            Test: Add a node to existing network
 339         """
 340         num_nodes = 0
 341         num_tps = 0
 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")
 354
 355     def test_update_network_node_tp(self):
 356         """
 357            Test: Update a tp to existing network, add all tp elements
 358         """
 359         num_nodes = 0
 360         num_tps = 0
 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")
 381
 382     def test_update_network_node_tp2(self):
 383         """
 384            Test: Update a tp to existing network, change tp elements
 385         """
 386         num_nodes = 0
 387         num_tps = 0
 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")
 407
 408         # Change frame size
 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")
 424
 425         # Change MAC address
 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")
 441
 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")
 459
 460     def test_update_extra_network_node_tp2(self):
 461         """
 462            Test: Update a tp to existing network, change tp elements
 463         """
 464         num_nodes = 0
 465         num_tps = 0
 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")
 480
 481
 482
 483 class SdnTopStoreNetworkLinkTest(unittest.TestCase):
 484     def setUp(self):
 485         """
 486           Initialize Top data store
 487         """
 488         self._nwtopdata_store = NwtopDataStore(logger)
 489         self.test_nwtop = RwTl.YangData_IetfNetwork()
 490
 491         self.l2top = MyL2Topology(self.test_nwtop, logger)
 492         self.l2top.setup_all()
 493
 494         # Get initial test data
 495         self.l2net1 = self.l2top.find_nw("L2HostNetwork-1")
 496         # Create initial nw
 497         self._nwtopdata_store.create_network("L2HostNetwork-1", self.l2net1)
 498         # Get test data
 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"
 503
 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"
 508
 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")
 514
 515     def tearDown(self):
 516         logger.info("SdnTopStoreTest: Done with  NetworkLinkTest")
 517
 518         self.new_l2net = None
 519         self.src_node = None
 520         self.tp1 = None
 521         self.dest_node = None
 522         self.tp2 = None
 523
 524     def test_add_network_link(self):
 525         """
 526            Test: Add a link to existing network
 527         """
 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)
 559
 560     def test_add_extra_network_link(self):
 561         """
 562            Test: Add a link to existing network
 563         """
 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)
 586
 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)
 605
 606     def test_add_network_link_l2attr(self):
 607         """
 608            Test: Check L2 link attributes
 609         """
 610         logger.info("SdnTopStoreTest: Add network link L2 attributes")
 611         # Create test state
 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)
 621         # Verify data state
 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)
 636
 637     def test_change_network_link_l2attr(self):
 638         """
 639            Test: Change L2 link attributes
 640         """
 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)
 667
 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)
 675         # Verify test state
 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")
 679
 680     def test_change_network_link_dest_tp(self):
 681         """
 682            Test: Change L2 link attributes
 683         """
 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)
 710
 711         # Create test state
 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)
 720         # Verify test state
 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)
 724
 725
 726
 727
 728 if __name__ == "__main__":
 729     logging.basicConfig(level=logging.INFO)
 730     unittest.main()
 731
 732
 733
 734