resources/tools/odltools/odltools/mdsal/tests/test_itm_state.py

   1 # Copyright 2018 Red Hat, Inc. and others. All Rights Reserved.
   2 #
   3 # Licensed under the Apache License, Version 2.0 (the "License");
   4 # you may not use this file except in compliance with the License.
   5 # You may obtain a copy of the License at
   6 #
   7 #     http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 # Unless required by applicable law or agreed to in writing, software
  10 # distributed under the License is distributed on an "AS IS" BASIS,
  11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12 # See the License for the specific language governing permissions and
  13 # limitations under the License.
  14
  15 import logging
  16 import unittest
  17 from odltools import logg
  18 from odltools.mdsal.models.itm_state import dpn_endpoints
  19 from odltools.mdsal.models.itm_state import DpnEndpoints
  20 from odltools.mdsal.models.itm_state import tunnels_state
  21 from odltools.mdsal.models.model import Model
  22 from odltools.mdsal import tests
  23
  24 logger = logging.getLogger("test.itmstate")
  25
  26
  27 class TestItmState(unittest.TestCase):
  28     def setUp(self):
  29         logg.Logger(logging.INFO, logging.INFO)
  30         args = tests.Args(path=tests.get_resources_path())
  31         self.itm_state_dpn_endpoints = dpn_endpoints(Model.CONFIG, args)
  32         self.itm_state_tunnels_state = tunnels_state(Model.OPERATIONAL, args)
  33
  34     def test_read_file(self):
  35         logger.debug("dpn-endpoints: %s", self.itm_state_dpn_endpoints.data)
  36         logger.debug("dpn-endpoints: \n%s",
  37                      self.itm_state_dpn_endpoints.pretty_format(self.itm_state_dpn_endpoints.data))
  38
  39     def test_get_ip_address(self):
  40         dpn_ids = self.itm_state_dpn_endpoints.get_dpn_ids()
  41         dpn_id = dpn_ids[0]
  42         ip_address = self.itm_state_dpn_endpoints.get_ip_address(dpn_id)
  43         logger.debug("dpn_id: %s, ip_address: %s", dpn_id, ip_address)
  44         self.assertEqual(dpn_id, 132319289050514)
  45         self.assertEqual(ip_address, "10.30.170.17")
  46
  47     def test_get_all(self):
  48         logger.debug("dpn-endpoints: %s", self.itm_state_dpn_endpoints.data)
  49         logger.debug("dpn-endpoints: \n%s",
  50                      self.itm_state_dpn_endpoints.pretty_format(self.itm_state_dpn_endpoints.data))
  51
  52         dpn_ids = self.itm_state_dpn_endpoints.get_dpn_ids()
  53         dpn_id = dpn_ids[0]
  54         dpn_teps_info = self.itm_state_dpn_endpoints.get_dpn_teps_info(dpn_id)
  55         logger.debug("dpn_teps_info for %s: %s", dpn_id, dpn_teps_info)
  56
  57         ip_address = self.itm_state_dpn_endpoints.get_ip_address(dpn_id)
  58         logger.debug("ip_address: %s", ip_address)
  59         self.assertEqual(ip_address, "10.30.170.17")
  60
  61         self.get_info(DpnEndpoints.CONTAINER)
  62         self.get_info(DpnEndpoints.CLIST)
  63         self.get_info(DpnEndpoints.TUNNEL_END_POINTS)
  64         self.get_info(DpnEndpoints.DPN_ID)
  65
  66     def get_info(self, key):
  67         info = self.itm_state_dpn_endpoints.get_kv(key, self.itm_state_dpn_endpoints.data, values=[])
  68         logger.debug("dpn info for %s: %s", key, info)
  69         return info
  70
  71     def test_get_ip_address_from_dpn_info(self):
  72         nodes = self.itm_state_dpn_endpoints.get_clist_by_key()
  73         node = nodes.get(132319289050514)
  74         self.assertIsNotNone(node)
  75         ip = self.itm_state_dpn_endpoints.get_ip_address_from_dpn_info(node)
  76         self.assertEqual("10.30.170.17", ip)
  77
  78     def test_get_tunnels_state(self):
  79         d = self.itm_state_tunnels_state.get_clist_by_key()
  80         self.assertIsNotNone(d and d['tun428ee8c4fe7'])
  81
  82
  83 if __name__ == '__main__':
  84     unittest.main()