import mock
-from plugins.OpenStack.common import Common
-from plugins.OpenStack.settings import Config
+from osm_mon.core.auth import AuthManager
+from osm_mon.core.database import VimCredentials
+from osm_mon.plugins.OpenStack.common import Common
+from osm_mon.plugins.OpenStack.settings import Config
import requests
self.value = json.dumps({"mock_value": "mock_details",
"vim_type": "OPENSTACK",
"access_config":
- {"openstack_site": "my_site",
- "user": "my_user",
- "password": "my_password",
- "vim_tenant_name": "my_tenant"}})
+ {"openstack_site": "my_site",
+ "user": "my_user",
+ "password": "my_password",
+ "vim_tenant_name": "my_tenant"}})
class TestCommon(unittest.TestCase):
"""Test Setup."""
super(TestCommon, self).setUp()
self.common = Common()
-
- @mock.patch.object(client, "Client")
- def test_authenticate_exists(self, key_client):
- """Testing if an authentication token already exists."""
- # If the auth_token is already generated a new one will not be creates
- self.common._auth_token = "my_auth_token"
- token = self.common._authenticate()
-
- self.assertEqual(token, "my_auth_token")
-
+ self.creds = VimCredentials()
+ self.creds.id = 'test_id'
+ self.creds.user = 'user'
+ self.creds.url = 'url'
+ self.creds.password = 'password'
+ self.creds.tenant_name = 'tenant_name'
+
+ @mock.patch.object(AuthManager, "get_credentials")
@mock.patch.object(Config, "instance")
@mock.patch.object(client, "Client")
- def test_authenticate_none(self, key_client, cfg):
+ def test_get_auth_token(self, key_client, cfg, get_creds):
"""Test generating a new authentication token."""
- # If auth_token doesn't exist one will try to be created with keystone
- # With the configuration values from the environment
- self.common._auth_token = None
- config = cfg.return_value
- url = config.OS_AUTH_URL
- user = config.OS_USERNAME
- pword = config.OS_PASSWORD
- tenant = config.OS_TENANT_NAME
-
- self.common._authenticate()
-
- key_client.assert_called_with(auth_url=url,
- username=user,
- password=pword,
- tenant_name=tenant)
- key_client.reset_mock()
-
- @mock.patch.object(client, "Client")
- def test_authenticate_access_cred(self, key_client):
- """Test generating an auth_token using access_credentials from SO."""
- # Mock valid message from SO
- self.common._auth_token = None
- message = Message()
-
- self.common._authenticate(message=message)
-
- # The class variables are set for each consifugration
- self.assertEqual(self.common.openstack_url, "my_site")
- self.assertEqual(self.common.user, "my_user")
- self.assertEqual(self.common.password, "my_password")
- self.assertEqual(self.common.tenant, "my_tenant")
- key_client.assert_called
+ get_creds.return_value = self.creds
+ Common.get_auth_token('test_id')
+ get_creds.assert_called_with('test_id')
+ key_client.assert_called_with(auth_url='url', password='password', tenant_name='tenant_name', username='user')
@mock.patch.object(requests, 'post')
def test_post_req(self, post):
"""Testing a post request."""
- self.common._perform_request("url", "auth_token", req_type="post",
- payload="payload")
+ Common.perform_request("url", "auth_token", req_type="post",
+ payload="payload")
post.assert_called_with("url", data="payload", headers=mock.ANY,
timeout=mock.ANY)
def test_get_req(self, get):
"""Testing a get request."""
# Run the defualt get request without any parameters
- self.common._perform_request("url", "auth_token", req_type="get")
+ Common.perform_request("url", "auth_token", req_type="get")
get.assert_called_with("url", params=None, headers=mock.ANY,
timeout=mock.ANY)
# Test with some parameters specified
get.reset_mock()
- self.common._perform_request("url", "auth_token", req_type="get",
- params="some parameters")
+ Common.perform_request("url", "auth_token", req_type="get",
+ params="some parameters")
get.assert_called_with("url", params="some parameters",
headers=mock.ANY, timeout=mock.ANY)
@mock.patch.object(requests, 'put')
def test_put_req(self, put):
"""Testing a put request."""
- self.common._perform_request("url", "auth_token", req_type="put",
- payload="payload")
+ Common.perform_request("url", "auth_token", req_type="put",
+ payload="payload")
put.assert_called_with("url", data="payload", headers=mock.ANY,
timeout=mock.ANY)
@mock.patch.object(requests, 'delete')
def test_delete_req(self, delete):
"""Testing a delete request."""
- self.common._perform_request("url", "auth_token", req_type="delete")
+ Common.perform_request("url", "auth_token", req_type="delete")
delete.assert_called_with("url", headers=mock.ANY, timeout=mock.ANY)