417 lines
16 KiB
Python
417 lines
16 KiB
Python
# All Rights Reserved.
|
|
#
|
|
# 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.
|
|
|
|
import json
|
|
from unittest import mock
|
|
|
|
import jsonschema
|
|
|
|
from rally.env import env_mgr
|
|
from rally.env import platform
|
|
from rally import exceptions
|
|
|
|
from rally_openstack.environment.platforms import existing
|
|
from tests.unit import test
|
|
|
|
|
|
class PlatformBaseTestCase(test.TestCase):
|
|
|
|
def _check_schema(self, schema, obj):
|
|
jsonschema.validate(obj, schema)
|
|
|
|
def _check_health_schema(self, obj):
|
|
self._check_schema(env_mgr.EnvManager._HEALTH_FORMAT, obj)
|
|
|
|
def _check_cleanup_schema(self, obj):
|
|
self._check_schema(env_mgr.EnvManager._CLEANUP_FORMAT, obj)
|
|
|
|
def _check_info_schema(self, obj):
|
|
self._check_schema(env_mgr.EnvManager._INFO_FORMAT, obj)
|
|
|
|
|
|
class ExistingPlatformTestCase(PlatformBaseTestCase):
|
|
|
|
def test_validate_spec_schema(self):
|
|
spec = {
|
|
"existing@openstack": {
|
|
"auth_url": "url",
|
|
"admin": {
|
|
"username": "admin",
|
|
"password": "password123",
|
|
"tenant_name": "admin"
|
|
},
|
|
"users": [{
|
|
"username": "admin",
|
|
"password": "password123",
|
|
"tenant_name": "admin"
|
|
}]
|
|
}
|
|
}
|
|
result = platform.Platform.validate("existing@openstack", {},
|
|
spec, spec["existing@openstack"])
|
|
self.assertEqual([], result)
|
|
|
|
def test_validate_invalid_spec(self):
|
|
spec = {
|
|
"existing@openstack": {
|
|
"something_wrong": {
|
|
"username": "not_an_admin",
|
|
"password": "password123",
|
|
"project_name": "not_an_admin"
|
|
}
|
|
}
|
|
}
|
|
result = platform.Platform.validate("existing@openstack", {},
|
|
spec, spec["existing@openstack"])
|
|
self.assertNotEqual([], result)
|
|
|
|
def test_validate_spec_schema_with_api_info(self):
|
|
spec = {
|
|
"existing@openstack": {
|
|
"auth_url": "url",
|
|
"admin": {
|
|
"username": "admin",
|
|
"password": "password123",
|
|
"tenant_name": "admin"
|
|
},
|
|
"api_info": {
|
|
"nova": {"version": 1},
|
|
"cinder": {"version": 2, "service_type": "volumev2"}
|
|
}
|
|
}
|
|
}
|
|
result = platform.Platform.validate("existing@openstack", {},
|
|
spec, spec["existing@openstack"])
|
|
self.assertEqual([], result)
|
|
|
|
def test_create_users_only(self):
|
|
|
|
spec = {
|
|
"auth_url": "https://best",
|
|
"endpoint": "check_that_its_poped",
|
|
"users": [
|
|
{"project_name": "a", "username": "a", "password": "a"},
|
|
{"project_name": "b", "username": "b", "password": "b"}
|
|
]
|
|
}
|
|
|
|
self.assertEqual(
|
|
({
|
|
"admin": None,
|
|
"users": [
|
|
{
|
|
"auth_url": "https://best", "endpoint_type": None,
|
|
"region_name": None,
|
|
"domain_name": None,
|
|
"user_domain_name": "default",
|
|
"project_domain_name": "default",
|
|
"https_insecure": False, "https_cacert": None,
|
|
"tenant_name": "a", "username": "a", "password": "a"
|
|
},
|
|
{
|
|
"auth_url": "https://best", "endpoint_type": None,
|
|
"region_name": None,
|
|
"domain_name": None,
|
|
"user_domain_name": "default",
|
|
"project_domain_name": "default",
|
|
"https_insecure": False, "https_cacert": None,
|
|
"tenant_name": "b", "username": "b", "password": "b"
|
|
}
|
|
]
|
|
}, {}),
|
|
existing.OpenStack(spec).create())
|
|
|
|
def test_create_admin_only(self):
|
|
spec = {
|
|
"auth_url": "https://best",
|
|
"endpoint_type": "public",
|
|
"https_insecure": True,
|
|
"https_cacert": "/my.ca",
|
|
"profiler_hmac_key": "key",
|
|
"profiler_conn_str": "http://prof",
|
|
"admin": {
|
|
"domain_name": "d", "user_domain_name": "d",
|
|
"project_domain_name": "d", "project_name": "d",
|
|
"username": "d", "password": "d"
|
|
}
|
|
}
|
|
self.assertEqual(
|
|
(
|
|
{
|
|
"admin": {
|
|
"auth_url": "https://best",
|
|
"endpoint_type": "public",
|
|
"https_insecure": True, "https_cacert": "/my.ca",
|
|
"profiler_hmac_key": "key",
|
|
"profiler_conn_str": "http://prof",
|
|
"region_name": None, "domain_name": "d",
|
|
"user_domain_name": "d", "project_domain_name": "d",
|
|
"tenant_name": "d", "username": "d", "password": "d"
|
|
},
|
|
"users": []
|
|
},
|
|
{}
|
|
),
|
|
existing.OpenStack(spec).create())
|
|
|
|
def test_create_spec_from_sys_environ(self):
|
|
# keystone v2
|
|
sys_env = {
|
|
"OS_AUTH_URL": "https://example.com",
|
|
"OS_USERNAME": "user",
|
|
"OS_PASSWORD": "pass",
|
|
"OS_TENANT_NAME": "projectX",
|
|
"OS_INTERFACE": "publicURL",
|
|
"OS_REGION_NAME": "Region1",
|
|
"OS_CACERT": "Cacert",
|
|
"OS_CERT": "cert",
|
|
"OS_KEY": "key",
|
|
"OS_INSECURE": True,
|
|
"OSPROFILER_HMAC_KEY": "hmackey",
|
|
"OSPROFILER_CONN_STR": "https://example2.com",
|
|
}
|
|
|
|
result = existing.OpenStack.create_spec_from_sys_environ(sys_env)
|
|
self.assertTrue(result["available"])
|
|
self.assertEqual(
|
|
{
|
|
"admin": {
|
|
"username": "user",
|
|
"tenant_name": "projectX",
|
|
"password": "pass"
|
|
},
|
|
"auth_url": "https://example.com",
|
|
"endpoint_type": "public",
|
|
"region_name": "Region1",
|
|
"https_cacert": "Cacert",
|
|
"https_cert": "cert",
|
|
"https_key": "key",
|
|
"https_insecure": True,
|
|
"profiler_hmac_key": "hmackey",
|
|
"profiler_conn_str": "https://example2.com",
|
|
"api_info": {
|
|
"keystone": {
|
|
"version": 2,
|
|
"service_type": "identity"
|
|
}
|
|
}
|
|
}, result["spec"])
|
|
|
|
# keystone v3
|
|
sys_env["OS_IDENTITY_API_VERSION"] = "3"
|
|
|
|
result = existing.OpenStack.create_spec_from_sys_environ(sys_env)
|
|
|
|
self.assertEqual(
|
|
{
|
|
"admin": {
|
|
"username": "user",
|
|
"project_name": "projectX",
|
|
"user_domain_name": "Default",
|
|
"password": "pass",
|
|
"project_domain_name": "Default"
|
|
},
|
|
"endpoint_type": "public",
|
|
"auth_url": "https://example.com",
|
|
"region_name": "Region1",
|
|
"https_cacert": "Cacert",
|
|
"https_cert": "cert",
|
|
"https_key": "key",
|
|
"https_insecure": True,
|
|
"profiler_hmac_key": "hmackey",
|
|
"profiler_conn_str": "https://example2.com",
|
|
"api_info": {
|
|
"keystone": {
|
|
"version": 3,
|
|
"service_type": "identityv3"
|
|
}
|
|
}
|
|
}, result["spec"])
|
|
|
|
def test_create_spec_from_sys_environ_fails_with_missing_vars(self):
|
|
sys_env = {"OS_AUTH_URL": "https://example.com"}
|
|
result = existing.OpenStack.create_spec_from_sys_environ(sys_env)
|
|
self.assertFalse(result["available"])
|
|
self.assertIn("OS_USERNAME", result["message"])
|
|
self.assertIn("OS_PASSWORD", result["message"])
|
|
self.assertNotIn("OS_AUTH_URL", result["message"])
|
|
|
|
sys_env = {"OS_AUTH_URL": "https://example.com",
|
|
"OS_USERNAME": "user",
|
|
"OS_PASSWORD": "pass"}
|
|
result = existing.OpenStack.create_spec_from_sys_environ(sys_env)
|
|
self.assertFalse(result["available"])
|
|
self.assertIn("OS_PROJECT_NAME or OS_TENANT_NAME", result["message"])
|
|
|
|
def test_destroy(self):
|
|
self.assertIsNone(existing.OpenStack({}).destroy())
|
|
|
|
def test_cleanup(self):
|
|
result1 = existing.OpenStack({}).cleanup()
|
|
result2 = existing.OpenStack({}).cleanup(task_uuid="any")
|
|
self.assertEqual(result1, result2)
|
|
self.assertEqual(
|
|
{
|
|
"message": "Coming soon!",
|
|
"discovered": 0,
|
|
"deleted": 0,
|
|
"failed": 0,
|
|
"resources": {},
|
|
"errors": []
|
|
},
|
|
result1
|
|
)
|
|
self._check_cleanup_schema(result1)
|
|
|
|
@mock.patch("rally_openstack.common.osclients.Clients")
|
|
def test_check_health(self, mock_clients):
|
|
pdata = {
|
|
"admin": mock.MagicMock(),
|
|
"users": [mock.MagicMock(), mock.MagicMock()]
|
|
}
|
|
result = existing.OpenStack({}, platform_data=pdata).check_health()
|
|
self._check_health_schema(result)
|
|
self.assertEqual({"available": True}, result)
|
|
mock_clients.assert_has_calls(
|
|
[mock.call(pdata["users"][0]), mock.call().keystone(),
|
|
mock.call(pdata["users"][1]), mock.call().keystone(),
|
|
mock.call(pdata["admin"]), mock.call().verified_keystone()])
|
|
|
|
@mock.patch("rally_openstack.common.osclients.Clients")
|
|
def test_check_failed_with_native_rally_exc(self, mock_clients):
|
|
e = exceptions.RallyException("foo")
|
|
mock_clients.return_value.keystone.side_effect = e
|
|
pdata = {"admin": None,
|
|
"users": [{"username": "balbab", "password": "12345"}]}
|
|
result = existing.OpenStack({}, platform_data=pdata).check_health()
|
|
self._check_health_schema(result)
|
|
self.assertEqual(
|
|
{
|
|
"available": False,
|
|
"message": e.format_message(),
|
|
"traceback": mock.ANY
|
|
},
|
|
result)
|
|
|
|
@mock.patch("rally_openstack.common.osclients.Clients")
|
|
def test_check_failed_admin(self, mock_clients):
|
|
mock_clients.return_value.verified_keystone.side_effect = Exception
|
|
pdata = {"admin": {"username": "balbab", "password": "12345"},
|
|
"users": []}
|
|
result = existing.OpenStack({}, platform_data=pdata).check_health()
|
|
self._check_health_schema(result)
|
|
self.assertEqual(
|
|
{"available": False,
|
|
"message":
|
|
"Bad admin creds: \n%s"
|
|
% json.dumps({"username": "balbab", "password": "***",
|
|
"api_info": {}},
|
|
indent=2, sort_keys=True),
|
|
"traceback": mock.ANY},
|
|
result)
|
|
self.assertIn("Traceback (most recent call last)", result["traceback"])
|
|
|
|
@mock.patch("rally_openstack.common.osclients.Clients")
|
|
def test_check_failed_users(self, mock_clients):
|
|
mock_clients.return_value.keystone.side_effect = Exception
|
|
pdata = {"admin": None,
|
|
"users": [{"username": "balbab", "password": "12345"}]}
|
|
result = existing.OpenStack({}, platform_data=pdata).check_health()
|
|
self._check_health_schema(result)
|
|
self.assertEqual(
|
|
{"available": False,
|
|
"message":
|
|
"Bad user creds: \n%s"
|
|
% json.dumps({"username": "balbab", "password": "***",
|
|
"api_info": {}},
|
|
indent=2, sort_keys=True),
|
|
"traceback": mock.ANY},
|
|
result)
|
|
self.assertIn("Traceback (most recent call last)", result["traceback"])
|
|
|
|
@mock.patch("rally_openstack.common.osclients.Clients")
|
|
def test_check_health_with_api_info(self, mock_clients):
|
|
pdata = {"admin": mock.MagicMock(),
|
|
"users": [],
|
|
"api_info": {"fakeclient": "version"}}
|
|
result = existing.OpenStack({}, platform_data=pdata).check_health()
|
|
self._check_health_schema(result)
|
|
self.assertEqual({"available": True}, result)
|
|
mock_clients.assert_has_calls(
|
|
[mock.call(pdata["admin"]), mock.call().verified_keystone(),
|
|
mock.call().fakeclient.choose_version(),
|
|
mock.call().fakeclient.validate_version(
|
|
mock_clients.return_value.fakeclient.choose_version
|
|
.return_value),
|
|
mock.call().fakeclient.create_client()])
|
|
|
|
@mock.patch("rally_openstack.common.osclients.Clients")
|
|
def test_check_version_failed_with_api_info(self, mock_clients):
|
|
pdata = {"admin": mock.MagicMock(),
|
|
"users": [],
|
|
"api_info": {"fakeclient": "version"}}
|
|
|
|
def validate_version(version):
|
|
raise exceptions.RallyException("Version is not supported.")
|
|
(mock_clients.return_value.fakeclient
|
|
.validate_version) = validate_version
|
|
result = existing.OpenStack({}, platform_data=pdata).check_health()
|
|
self._check_health_schema(result)
|
|
self.assertEqual({"available": False,
|
|
"message": ("Invalid setting for 'fakeclient':"
|
|
" Version is not supported.")},
|
|
result)
|
|
|
|
@mock.patch("rally_openstack.common.osclients.Clients")
|
|
def test_check_unexpected_failed_with_api_info(self, mock_clients):
|
|
pdata = {"admin": mock.MagicMock(),
|
|
"users": [],
|
|
"api_info": {"fakeclient": "version"}}
|
|
|
|
def create_client():
|
|
raise Exception("Invalid client.")
|
|
|
|
(mock_clients.return_value.fakeclient
|
|
.choose_version.return_value) = "1.0"
|
|
mock_clients.return_value.fakeclient.create_client = create_client
|
|
result = existing.OpenStack({}, platform_data=pdata).check_health()
|
|
self._check_health_schema(result)
|
|
self.assertEqual({"available": False,
|
|
"message": ("Can not create 'fakeclient' with"
|
|
" 1.0 version."),
|
|
"traceback": mock.ANY},
|
|
result)
|
|
|
|
@mock.patch("rally_openstack.common.osclients.Clients")
|
|
def test_info(self, mock_clients):
|
|
mock_clients.return_value.services.return_value = {
|
|
"foo": "bar",
|
|
"volumev4": "__unknown__"}
|
|
platform_data = {
|
|
"admin": None,
|
|
"users": [{"username": "u1", "password": "123"}]
|
|
}
|
|
p = existing.OpenStack({}, platform_data=platform_data)
|
|
|
|
result = p.info()
|
|
mock_clients.assert_called_once_with(platform_data["users"][0])
|
|
mock_clients.return_value.services.assert_called_once_with()
|
|
self.assertEqual(
|
|
{
|
|
"info": {
|
|
"services": [{"type": "foo", "name": "bar"},
|
|
{"type": "volumev4"}]}},
|
|
result)
|
|
self._check_info_schema(result)
|