osm_mon/test/OpenStack/test_common.py

   1 # Copyright 2017 Intel Research and Development Ireland Limited
   2 # *************************************************************
   3
   4 # This file is part of OSM Monitoring module
   5 # All Rights Reserved to Intel Corporation
   6
   7 # Licensed under the Apache License, Version 2.0 (the "License"); you may
   8 # not use this file except in compliance with the License. You may obtain
   9 # a copy of the License at
  10
  11 #         http://www.apache.org/licenses/LICENSE-2.0
  12
  13 # Unless required by applicable law or agreed to in writing, software
  14 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
  15 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  16 # License for the specific language governing permissions and limitations
  17 # under the License.
  18
  19 # For those usages not covered by the Apache License, Version 2.0 please
  20 # contact: helena.mcgough@intel.com or adrian.hoban@intel.com
  21 ##
  22 """Tests for all common OpenStack methods."""
  23
  24 import json
  25
  26 import logging
  27
  28 import unittest
  29
  30 from keystoneclient.v3 import client
  31
  32 import mock
  33
  34 from osm_mon.core.auth import AuthManager
  35 from osm_mon.core.database import VimCredentials
  36 from osm_mon.plugins.OpenStack.common import Common
  37 from osm_mon.plugins.OpenStack.settings import Config
  38
  39 import requests
  40
  41 __author__ = "Helena McGough"
  42
  43 log = logging.getLogger(__name__)
  44
  45
  46 class Message(object):
  47     """Mock a message for an access credentials request."""
  48
  49     def __init__(self):
  50         """Initialise the topic and value of access_cred message."""
  51         self.topic = "access_credentials"
  52         self.value = json.dumps({"mock_value": "mock_details",
  53                                  "vim_type": "OPENSTACK",
  54                                  "access_config":
  55                                      {"openstack_site": "my_site",
  56                                       "user": "my_user",
  57                                       "password": "my_password",
  58                                       "vim_tenant_name": "my_tenant"}})
  59
  60
  61 class TestCommon(unittest.TestCase):
  62     """Test the common class for OpenStack plugins."""
  63
  64     def setUp(self):
  65         """Test Setup."""
  66         super(TestCommon, self).setUp()
  67         self.common = Common()
  68         self.creds = VimCredentials()
  69         self.creds.id = 'test_id'
  70         self.creds.user = 'user'
  71         self.creds.url = 'url'
  72         self.creds.password = 'password'
  73         self.creds.tenant_name = 'tenant_name'
  74
  75     @mock.patch.object(AuthManager, "get_credentials")
  76     @mock.patch.object(Config, "instance")
  77     @mock.patch.object(client, "Client")
  78     def test_get_auth_token(self, key_client, cfg, get_creds):
  79         """Test generating a new authentication token."""
  80         get_creds.return_value = self.creds
  81         Common.get_auth_token('test_id')
  82         get_creds.assert_called_with('test_id')
  83         key_client.assert_called_with(auth_url='url', password='password', tenant_name='tenant_name', username='user')
  84
  85     @mock.patch.object(requests, 'post')
  86     def test_post_req(self, post):
  87         """Testing a post request."""
  88         Common.perform_request("url", "auth_token", req_type="post",
  89                                     payload="payload")
  90
  91         post.assert_called_with("url", data="payload", headers=mock.ANY,
  92                                 timeout=mock.ANY)
  93
  94     @mock.patch.object(requests, 'get')
  95     def test_get_req(self, get):
  96         """Testing a get request."""
  97         # Run the defualt get request without any parameters
  98         Common.perform_request("url", "auth_token", req_type="get")
  99
 100         get.assert_called_with("url", params=None, headers=mock.ANY,
 101                                timeout=mock.ANY)
 102
 103         # Test with some parameters specified
 104         get.reset_mock()
 105         Common.perform_request("url", "auth_token", req_type="get",
 106                                     params="some parameters")
 107
 108         get.assert_called_with("url", params="some parameters",
 109                                headers=mock.ANY, timeout=mock.ANY)
 110
 111     @mock.patch.object(requests, 'put')
 112     def test_put_req(self, put):
 113         """Testing a put request."""
 114         Common.perform_request("url", "auth_token", req_type="put",
 115                                     payload="payload")
 116         put.assert_called_with("url", data="payload", headers=mock.ANY,
 117                                timeout=mock.ANY)
 118
 119     @mock.patch.object(requests, 'delete')
 120     def test_delete_req(self, delete):
 121         """Testing a delete request."""
 122         Common.perform_request("url", "auth_token", req_type="delete")
 123
 124         delete.assert_called_with("url", headers=mock.ANY, timeout=mock.ANY)