osm_mon/test/core/test_common_consumer.py

   1 # -*- coding: utf-8 -*-
   2
   3 # Copyright 2018 Whitestack, LLC
   4 # *************************************************************
   5
   6 # This file is part of OSM Monitoring module
   7 # All Rights Reserved to Whitestack, LLC
   8
   9 # Licensed under the Apache License, Version 2.0 (the "License"); you may
  10 # not use this file except in compliance with the License. You may obtain
  11 # a copy of the License at
  12
  13 #         http://www.apache.org/licenses/LICENSE-2.0
  14
  15 # Unless required by applicable law or agreed to in writing, software
  16 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
  17 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  18 # License for the specific language governing permissions and limitations
  19 # under the License.
  20 # For those usages not covered by the Apache License, Version 2.0 please
  21 # contact: bdiaz@whitestack.com or glavado@whitestack.com
  22 ##
  23 import unittest
  24
  25 import mock
  26 from kafka import KafkaProducer
  27 from kafka.errors import KafkaError
  28 from osm_common import dbmongo
  29
  30 from osm_mon.core.database import VimCredentials, DatabaseManager
  31 from osm_mon.core.message_bus.common_consumer import CommonConsumer
  32
  33
  34 @mock.patch.object(dbmongo.DbMongo, "db_connect", mock.Mock())
  35 class CommonConsumerTest(unittest.TestCase):
  36
  37     def setUp(self):
  38         try:
  39             KafkaProducer(bootstrap_servers='localhost:9092',
  40                           key_serializer=str.encode,
  41                           value_serializer=str.encode
  42                           )
  43         except KafkaError:
  44             self.skipTest('Kafka server not present.')
  45
  46     @mock.patch.object(DatabaseManager, "get_credentials")
  47     def test_get_vim_type(self, get_creds):
  48         mock_creds = VimCredentials()
  49         mock_creds.id = 'test_id'
  50         mock_creds.user = 'user'
  51         mock_creds.url = 'url'
  52         mock_creds.password = 'password'
  53         mock_creds.tenant_name = 'tenant_name'
  54         mock_creds.type = 'openstack'
  55
  56         get_creds.return_value = mock_creds
  57
  58         common_consumer = CommonConsumer()
  59         vim_type = common_consumer.get_vim_type('test_id')
  60
  61         self.assertEqual(vim_type, 'openstack')
  62
  63     @mock.patch.object(dbmongo.DbMongo, "get_one")
  64     def test_get_vdur(self, get_one):
  65         get_one.return_value = {'_id': 'a314c865-aee7-4d9b-9c9d-079d7f857f01',
  66                                 '_admin': {
  67                                     'projects_read': ['admin'], 'created': 1526044312.102287,
  68                                     'modified': 1526044312.102287, 'projects_write': ['admin']
  69                                 },
  70                                 'vim-account-id': 'c1740601-7287-48c8-a2c9-bce8fee459eb',
  71                                 'nsr-id-ref': '5ec3f571-d540-4cb0-9992-971d1b08312e',
  72                                 'vdur': [
  73                                     {
  74                                         'internal-connection-point': [],
  75                                         'vdu-id-ref': 'ubuntuvnf_vnfd-VM',
  76                                         'id': 'ffd73f33-c8bb-4541-a977-44dcc3cbe28d',
  77                                         'vim-id': '27042672-5190-4209-b844-95bbaeea7ea7',
  78                                         'name': 'ubuntuvnf_vnfd-VM'
  79                                     }
  80                                 ],
  81                                 'vnfd-ref': 'ubuntuvnf_vnfd',
  82                                 'member-vnf-index-ref': '1',
  83                                 'created-time': 1526044312.0999322,
  84                                 'vnfd-id': 'a314c865-aee7-4d9b-9c9d-079d7f857f01',
  85                                 'id': 'a314c865-aee7-4d9b-9c9d-079d7f857f01'}
  86         common_consumer = CommonConsumer()
  87         vdur = common_consumer.common_db.get_vdur('5ec3f571-d540-4cb0-9992-971d1b08312e', '1', 'ubuntuvnf_vnfd-VM')
  88         expected_vdur = {
  89             'internal-connection-point': [],
  90             'vdu-id-ref': 'ubuntuvnf_vnfd-VM',
  91             'id': 'ffd73f33-c8bb-4541-a977-44dcc3cbe28d',
  92             'vim-id': '27042672-5190-4209-b844-95bbaeea7ea7',
  93             'name': 'ubuntuvnf_vnfd-VM'
  94         }
  95
  96         self.assertDictEqual(vdur, expected_vdur)
  97
  98
  99 if __name__ == '__main__':
 100     unittest.main()