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 plugins.OpenStack.common import Common
  35 from plugins.OpenStack.settings import Config
  36
  37 import requests
  38
  39 __author__ = "Helena McGough"
  40
  41 log = logging.getLogger(__name__)
  42
  43
  44 class Message(object):
  45     """Mock a message for an access credentials request."""
  46
  47     def __init__(self):
  48         """Initialise the topic and value of access_cred message."""
  49         self.topic = "access_credentials"
  50         self.value = json.dumps({"mock_value": "mock_details",
  51                                  "vim_type": "OPENSTACK",
  52                                  "access_config":
  53                                  {"openstack_site": "my_site",
  54                                   "user": "my_user",
  55                                   "password": "my_password",
  56                                   "vim_tenant_name": "my_tenant"}})
  57
  58
  59 class TestCommon(unittest.TestCase):
  60     """Test the common class for OpenStack plugins."""
  61
  62     def setUp(self):
  63         """Test Setup."""
  64         super(TestCommon, self).setUp()
  65         self.common = Common()
  66
  67     @mock.patch.object(client, "Client")
  68     def test_authenticate_exists(self, key_client):
  69         """Testing if an authentication token already exists."""
  70         # If the auth_token is already generated a new one will not be creates
  71         self.common._auth_token = "my_auth_token"
  72         token = self.common._authenticate()
  73
  74         self.assertEqual(token, "my_auth_token")
  75
  76     @mock.patch.object(Config, "instance")
  77     @mock.patch.object(client, "Client")
  78     def test_authenticate_none(self, key_client, cfg):
  79         """Test generating a new authentication token."""
  80         # If auth_token doesn't exist one will try to be created with keystone
  81         # With the configuration values from the environment
  82         self.common._auth_token = None
  83         config = cfg.return_value
  84         url = config.OS_AUTH_URL
  85         user = config.OS_USERNAME
  86         pword = config.OS_PASSWORD
  87         tenant = config.OS_TENANT_NAME
  88
  89         self.common._authenticate()
  90
  91         key_client.assert_called_with(auth_url=url,
  92                                       username=user,
  93                                       password=pword,
  94                                       tenant_name=tenant)
  95         key_client.reset_mock()
  96
  97     @mock.patch.object(client, "Client")
  98     def test_authenticate_access_cred(self, key_client):
  99         """Test generating an auth_token using access_credentials from SO."""
 100         # Mock valid message from SO
 101         self.common._auth_token = None
 102         message = Message()
 103
 104         self.common._authenticate(message=message)
 105
 106         # The class variables are set for each consifugration
 107         self.assertEqual(self.common.openstack_url, "my_site")
 108         self.assertEqual(self.common.user, "my_user")
 109         self.assertEqual(self.common.password, "my_password")
 110         self.assertEqual(self.common.tenant, "my_tenant")
 111         key_client.assert_called
 112
 113     @mock.patch.object(requests, 'post')
 114     def test_post_req(self, post):
 115         """Testing a post request."""
 116         self.common._perform_request("url", "auth_token", req_type="post",
 117                                      payload="payload")
 118
 119         post.assert_called_with("url", data="payload", headers=mock.ANY,
 120                                 timeout=mock.ANY)
 121
 122     @mock.patch.object(requests, 'get')
 123     def test_get_req(self, get):
 124         """Testing a get request."""
 125         # Run the defualt get request without any parameters
 126         self.common._perform_request("url", "auth_token", req_type="get")
 127
 128         get.assert_called_with("url", params=None, headers=mock.ANY,
 129                                timeout=mock.ANY)
 130
 131         # Test with some parameters specified
 132         get.reset_mock()
 133         self.common._perform_request("url", "auth_token", req_type="get",
 134                                      params="some parameters")
 135
 136         get.assert_called_with("url", params="some parameters",
 137                                headers=mock.ANY, timeout=mock.ANY)
 138
 139     @mock.patch.object(requests, 'put')
 140     def test_put_req(self, put):
 141         """Testing a put request."""
 142         self.common._perform_request("url", "auth_token", req_type="put",
 143                                      payload="payload")
 144         put.assert_called_with("url", data="payload", headers=mock.ANY,
 145                                timeout=mock.ANY)
 146
 147     @mock.patch.object(requests, 'delete')
 148     def test_delete_req(self, delete):
 149         """Testing a delete request."""
 150         self.common._perform_request("url", "auth_token", req_type="delete")
 151
 152         delete.assert_called_with("url", headers=mock.ANY, timeout=mock.ANY)