x/collectd-openstack-plugins
Code Issues Proposed changes
collectd-openstack-plugins/collectd_ceilometer/tests/test_keystone_light.py
Michal Ptacek 68c6f2cc3e Adding Keystone V3 API support 
in master branch keystone v2.0 api is no longer supported, this patch
is introducing v3 api support.

Change-Id: I5ed5f65f34033b6a4c550704bb186dfa8d0fc82c
Closes-Bug: #1614892
2016-11-10 21:22:59 +00:00

550 lines
19 KiB
Python
Raw Blame History

# -*- coding: utf-8 -*-
# Copyright 2010-2011 OpenStack Foundation
# Copyright (c) 2015 Intel Corporation
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
"""Plugin tests"""
from __future__ import unicode_literals
from collectd_ceilometer import keystone_light
from collectd_ceilometer.keystone_light import ClientV2
from collectd_ceilometer.keystone_light import ClientV3
from collectd_ceilometer.keystone_light import MissingServices
import mock
import unittest
class KeystoneLightTestV3(unittest.TestCase):
"""Test the keystone light client with 3.0 keystone api"""
def setUp(self):
super(KeystoneLightTestV3, self).setUp()
self.test_authtoken = "c5bbb1c9a27e470fb482de2a718e08c2"
self.test_public_endpoint = "http://public_endpoint"
self.test_internal_endpoint = "http://iternal_endpoint"
self.test_region = "RegionOne"
response = {"token": {
"is_domain": 'false',
"methods": [
"password"
],
"roles": [
{
"id": "eacf519eb1264cba9ad645355ce1f6ec",
"name": "ResellerAdmin"
},
{
"id": "63e481b5d5f545ecb8947072ff34f10d",
"name": "admin"
}
],
"is_admin_project": 'false',
"project": {
"domain": {
"id": "default",
"name": "Default"
},
"id": "97467f21efb2493c92481429a04df7bd",
"name": "service"
},
"catalog": [
{
"endpoints": [
{
"url": self.test_public_endpoint + '/',
"interface": "public",
"region": "RegionOne",
"region_id": self.test_region,
"id": "5e1d9a45d7d442ca8971a5112b2e89b5"
},
{
"url": "http://127.0.0.1:8777",
"interface": "admin",
"region": "RegionOne",
"region_id": self.test_region,
"id": "5e8b536fde6049d381ee540c018905d1"
},
{
"url": self.test_internal_endpoint + '/',
"interface": "internal",
"region": "RegionOne",
"region_id": self.test_region,
"id": "db90c733ddd9466696bc5aaec43b18d0"
}
],
"type": "metering",
"id": "f6c15a041d574bc190c70815a14ab851",
"name": "ceilometer"
}
]
}
}
self.mock_response = mock.Mock()
self.mock_response.json.return_value = response
self.mock_response.headers = {
'X-Subject-Token': "c5bbb1c9a27e470fb482de2a718e08c2"
}
@mock.patch('collectd_ceilometer.keystone_light.requests.post')
def test_refresh(self, mock_post):
"""Test refresh"""
mock_post.return_value = self.mock_response
client = ClientV3("test_auth_url", "test_username",
"test_password", "test_tenant")
self.assertEqual(client.auth_token, self.test_authtoken)
expected_args = {
'headers': {'Accept': 'application/json'},
'json': {
'auth': {
"identity": {
"methods": ["password"],
"password": {
"user": {
"name": u'test_username',
"domain": {"id": "default"},
"password": u'test_password'
}
}
},
"scope": {
"project": {
"name": u'test_tenant',
"domain": {"id": "default"}
}
}
}
}
}
mock_post.assert_called_once_with(
'test_auth_url/auth/tokens',
json=expected_args['json'],
headers=expected_args['headers'],
)
@mock.patch('collectd_ceilometer.keystone_light.requests.post')
def test_getservice_endpoint(self, mock_post):
"""Test getservice endpoint"""
mock_post.return_value = self.mock_response
client = ClientV3("test_auth_url", "test_username",
"test_password", "test_tenant")
client.refresh()
endpoint = client.get_service_endpoint('ceilometer')
self.assertEqual(endpoint, self.test_internal_endpoint)
endpoint = client.get_service_endpoint('ceilometer', 'publicURL')
self.assertEqual(endpoint, self.test_public_endpoint)
endpoint = client.get_service_endpoint('ceilometer', 'publicURL',
self.test_region)
self.assertEqual(endpoint, self.test_public_endpoint)
with self.assertRaises(MissingServices):
client.get_service_endpoint('badname')
@mock.patch('collectd_ceilometer.keystone_light.requests.post')
def test_getservice_endpoint_error(self, mock_post):
"""Test getservice endpoint error"""
response = {"token": {
"is_domain": 'false',
"methods": [
"password"
],
"roles": [
{
"id": "eacf519eb1264cba9ad645355ce1f6ec",
"name": "ResellerAdmin"
},
{
"id": "63e481b5d5f545ecb8947072ff34f10d",
"name": "admin"
}
],
"is_admin_project": 'false',
"project": {
"domain": {
"id": "default",
"name": "Default"
},
"id": "97467f21efb2493c92481429a04df7bd",
"name": "service"
},
"catalog": [
{
"endpoints": [],
"type": "metering",
"id": "f6c15a041d574bc190c70815a14ab851",
"name": "badname"
}
]
}
}
self.mock_response = mock.Mock()
self.mock_response.json.return_value = response
self.mock_response.headers = {
'X-Subject-Token': "c5bbb1c9a27e470fb482de2a718e08c2"
}
mock_post.return_value = self.mock_response
client = ClientV3("test_auth_url", "test_username",
"test_password", "test_tenant")
client.refresh()
with self.assertRaises(MissingServices):
client.get_service_endpoint('ceilometer')
@mock.patch('collectd_ceilometer.keystone_light.requests.post')
def test_invalidresponse_missing_token(self, mock_post):
"""Test invalid response: missing access"""
response = {'badresponse': None}
mock_response = mock.Mock()
mock_response.json.return_value = response
mock_response.headers = {
'X-Subject-Token': "c5bbb1c9a27e470fb482de2a718e08c2"
}
mock_post.return_value = mock_response
client = keystone_light.ClientV3("test_auth_url", "test_username",
"test_password", "test_tenant")
with self.assertRaises(keystone_light.InvalidResponse):
client.refresh()
@mock.patch('collectd_ceilometer.keystone_light.requests.post')
def test_invalidresponse_missing_catalog(self, mock_post):
"""Test invalid response: missing catalog"""
response = {"token": {
"is_domain": 'false',
"methods": [
"password"
],
"roles": [
{
"id": "eacf519eb1264cba9ad645355ce1f6ec",
"name": "ResellerAdmin"
},
{
"id": "63e481b5d5f545ecb8947072ff34f10d",
"name": "admin"
}
],
"is_admin_project": 'false',
"project": {
"domain": {
"id": "default",
"name": "Default"
},
"id": "97467f21efb2493c92481429a04df7bd",
"name": "service"
},
}
}
mock_response = mock.Mock()
mock_response.json.return_value = response
mock_response.headers = {
'X-Subject-Token': "c5bbb1c9a27e470fb482de2a718e08c2"
}
mock_post.return_value = mock_response
client = keystone_light.ClientV3("test_auth_url", "test_username",
"test_password", "test_tenant")
with self.assertRaises(keystone_light.InvalidResponse):
client.refresh()
@mock.patch('collectd_ceilometer.keystone_light.requests.post')
def test_invalidresponse_missing_token_http_header(self, mock_post):
"""Test invalid response: missing token in header"""
response = {"token": {
"is_domain": 'false',
"methods": [
"password"
],
"roles": [
{
"id": "eacf519eb1264cba9ad645355ce1f6ec",
"name": "ResellerAdmin"
},
{
"id": "63e481b5d5f545ecb8947072ff34f10d",
"name": "admin"
}
],
"is_admin_project": 'false',
"project": {
"domain": {
"id": "default",
"name": "Default"
},
"id": "97467f21efb2493c92481429a04df7bd",
"name": "service"
},
"catalog": [
{
"endpoints": [
{
"url": self.test_public_endpoint + '/',
"interface": "public",
"region": "RegionOne",
"region_id": self.test_region,
"id": "5e1d9a45d7d442ca8971a5112b2e89b5"
},
{
"url": "http://127.0.0.1:8777",
"interface": "admin",
"region": "RegionOne",
"region_id": self.test_region,
"id": "5e8b536fde6049d381ee540c018905d1"
},
{
"url": self.test_internal_endpoint + '/',
"interface": "internal",
"region": "RegionOne",
"region_id": self.test_region,
"id": "db90c733ddd9466696bc5aaec43b18d0"
}
],
"type": "metering",
"id": "f6c15a041d574bc190c70815a14ab851",
"name": "ceilometer"
}
]
}
}
mock_response = mock.Mock()
mock_response.json.return_value = response
mock_post.return_value = mock_response
client = keystone_light.ClientV3("test_auth_url", "test_username",
"test_password", "test_tenant")
with self.assertRaises(keystone_light.InvalidResponse):
client.refresh()
class KeystoneLightTestV2(unittest.TestCase):
"""Test the keystone light client with 2.0 keystone api"""
def setUp(self):
super(KeystoneLightTestV2, self).setUp()
self.test_authtoken = "c5bbb1c9a27e470fb482de2a718e08c2"
self.test_public_endpoint = "http://public_endpoint"
self.test_internal_endpoint = "http://iternal_endpoint"
self.test_region = "RegionOne"
response = {'access': {
"token": {
"issued_at": "2015-09-04T08:59:09.991646",
"expires": "2015-09-04T09:59:09Z",
"id": self.test_authtoken,
"tenant": {
"enabled": True,
"description": None,
"name": "service",
"id": "fdeec62f6c794c8dbfda448a83de9ce2"
},
"audit_ids": [
"Pig7hVfGQjSuUnt1Hc5mCg"
]
},
"serviceCatalog": [{
"endpoints_links": [],
"endpoints": [{
"adminURL": "http://10.237.214.74:8777/",
"region": self.test_region,
"publicURL": self.test_public_endpoint + '/',
"internalURL": self.test_internal_endpoint,
"id": "ac95b1a24a854ec7a4b63b08ed4cbd83"
}],
"type": "metering",
"name": "ceilometer"
}, ],
}}
self.mock_response = mock.Mock()
self.mock_response.json.return_value = response
@mock.patch('collectd_ceilometer.keystone_light.requests.post')
def test_refresh(self, mock_post):
"""Test refresh"""
mock_post.return_value = self.mock_response
client = ClientV2("test_auth_url", "test_username",
"test_password", "test_tenant")
client.refresh()
self.assertEqual(mock_post.call_count, 1)
self.assertEqual(client.auth_token, self.test_authtoken)
expected_args = {
'headers': {'Accept': 'application/json'},
'json': {
'auth': {
'tenantName': u'test_tenant',
'passwordCredentials': {
'username': u'test_username',
'password': u'test_password'
}
}
}
}
self.assertEqual(mock_post.call_args[0], (u'test_auth_url/tokens',))
self.assertEqual(mock_post.call_args[1], expected_args)
@mock.patch('collectd_ceilometer.keystone_light.requests.post')
def test_getservice_endpoint(self, mock_post):
"""Test getservice endpoint"""
mock_post.return_value = self.mock_response
client = ClientV2("test_auth_url", "test_username",
"test_password", "test_tenant")
client.refresh()
endpoint = client.get_service_endpoint('ceilometer')
self.assertEqual(endpoint, self.test_internal_endpoint)
endpoint = client.get_service_endpoint('ceilometer', 'publicURL')
self.assertEqual(endpoint, self.test_public_endpoint)
endpoint = client.get_service_endpoint('ceilometer', 'publicURL',
self.test_region)
self.assertEqual(endpoint, self.test_public_endpoint)
with self.assertRaises(MissingServices):
client.get_service_endpoint('badname')
@mock.patch('collectd_ceilometer.keystone_light.requests.post')
def test_getservice_endpoint_error(self, mock_post):
"""Test getservice endpoint error"""
response = {'access': {
"token": {
"id": "authtoken",
},
"serviceCatalog": [{
"endpoints_links": [],
"endpoints": [],
"type": "metering",
"missingname": "ceilometer"
},
],
}}
self.mock_response = mock.Mock()
self.mock_response.json.return_value = response
mock_post.return_value = self.mock_response
client = ClientV2("test_auth_url", "test_username",
"test_password", "test_tenant")
client.refresh()
with self.assertRaises(MissingServices):
client.get_service_endpoint('ceilometer')
@mock.patch('collectd_ceilometer.keystone_light.requests.post')
def test_invalidresponse_missing_access(self, mock_post):
"""Test invalid response: missing access"""
response = {'badresponse': None}
mock_response = mock.Mock()
mock_response.json.return_value = response
mock_post.return_value = mock_response
client = keystone_light.ClientV2("test_auth_url", "test_username",
"test_password", "test_tenant")
with self.assertRaises(keystone_light.InvalidResponse):
client.refresh()
@mock.patch('collectd_ceilometer.keystone_light.requests.post')
def test_invalidresponse_missing_servicecatalog(self, mock_post):
"""Test invalid response: missing servicecatalog"""
response = {'access': {
'token': None
}
}
mock_response = mock.Mock()
mock_response.json.return_value = response
mock_post.return_value = mock_response
client = keystone_light.ClientV2("test_auth_url", "test_username",
"test_password", "test_tenant")
with self.assertRaises(keystone_light.InvalidResponse):
client.refresh()
@mock.patch('collectd_ceilometer.keystone_light.requests.post')
def test_invalidresponse_missing_token(self, mock_post):
"""Test invalid response: missing token"""
response = {'access': {
"serviceCatalog": []
}}
mock_response = mock.Mock()
mock_response.json.return_value = response
mock_post.return_value = mock_response
client = keystone_light.ClientV2("test_auth_url", "test_username",
"test_password", "test_tenant")
with self.assertRaises(keystone_light.InvalidResponse):
client.refresh()
@mock.patch('collectd_ceilometer.keystone_light.requests.post')
def test_invalidresponse_missing_id(self, mock_post):
"""Test invalid response: missing id"""
response = {'access': {
"serviceCatalog": [],
"token": None
}, }
mock_response = mock.Mock()
mock_response.json.return_value = response
mock_post.return_value = mock_response
client = keystone_light.ClientV2("test_auth_url", "test_username",
"test_password", "test_tenant")
with self.assertRaises(keystone_light.InvalidResponse):
client.refresh()
View Git Blame Copy Permalink