osm_mon/test/core/test_common_consumer.py

   1 import unittest
   2
   3 import mock
   4 from kafka import KafkaProducer
   5 from kafka.errors import KafkaError
   6 from osm_common import dbmongo
   7
   8 from osm_mon.core.database import VimCredentials, DatabaseManager
   9 from osm_mon.core.message_bus.common_consumer import CommonConsumer
  10
  11
  12 @mock.patch.object(dbmongo.DbMongo, "db_connect", mock.Mock())
  13 class CommonConsumerTest(unittest.TestCase):
  14
  15     def setUp(self):
  16         try:
  17             KafkaProducer(bootstrap_servers='localhost:9092',
  18                           key_serializer=str.encode,
  19                           value_serializer=str.encode
  20                           )
  21         except KafkaError:
  22             self.skipTest('Kafka server not present.')
  23
  24     @mock.patch.object(DatabaseManager, "get_credentials")
  25     def test_get_vim_type(self, get_creds):
  26         mock_creds = VimCredentials()
  27         mock_creds.id = 'test_id'
  28         mock_creds.user = 'user'
  29         mock_creds.url = 'url'
  30         mock_creds.password = 'password'
  31         mock_creds.tenant_name = 'tenant_name'
  32         mock_creds.type = 'openstack'
  33
  34         get_creds.return_value = mock_creds
  35
  36         common_consumer = CommonConsumer()
  37         vim_type = common_consumer.get_vim_type('test_id')
  38
  39         self.assertEqual(vim_type, 'openstack')
  40
  41     @mock.patch.object(dbmongo.DbMongo, "get_one")
  42     def test_get_vdur(self, get_one):
  43         get_one.return_value = {'_id': 'a314c865-aee7-4d9b-9c9d-079d7f857f01',
  44                                 '_admin': {
  45                                     'projects_read': ['admin'], 'created': 1526044312.102287,
  46                                     'modified': 1526044312.102287, 'projects_write': ['admin']
  47                                 },
  48                                 'vim-account-id': 'c1740601-7287-48c8-a2c9-bce8fee459eb',
  49                                 'nsr-id-ref': '5ec3f571-d540-4cb0-9992-971d1b08312e',
  50                                 'vdur': [
  51                                     {
  52                                         'internal-connection-point': [],
  53                                         'vdu-id-ref': 'ubuntuvnf_vnfd-VM',
  54                                         'id': 'ffd73f33-c8bb-4541-a977-44dcc3cbe28d',
  55                                         'vim-id': '27042672-5190-4209-b844-95bbaeea7ea7',
  56                                         'name': 'ubuntuvnf_vnfd-VM'
  57                                     }
  58                                 ],
  59                                 'vnfd-ref': 'ubuntuvnf_vnfd',
  60                                 'member-vnf-index-ref': '1',
  61                                 'created-time': 1526044312.0999322,
  62                                 'vnfd-id': 'a314c865-aee7-4d9b-9c9d-079d7f857f01',
  63                                 'id': 'a314c865-aee7-4d9b-9c9d-079d7f857f01'}
  64         common_consumer = CommonConsumer()
  65         vdur = common_consumer.common_db.get_vdur('5ec3f571-d540-4cb0-9992-971d1b08312e', '1', 'ubuntuvnf_vnfd-VM')
  66         expected_vdur = {
  67             'internal-connection-point': [],
  68             'vdu-id-ref': 'ubuntuvnf_vnfd-VM',
  69             'id': 'ffd73f33-c8bb-4541-a977-44dcc3cbe28d',
  70             'vim-id': '27042672-5190-4209-b844-95bbaeea7ea7',
  71             'name': 'ubuntuvnf_vnfd-VM'
  72         }
  73
  74         self.assertDictEqual(vdur, expected_vdur)
  75
  76
  77 if __name__ == '__main__':
  78     unittest.main()