openstack/trove
Code Issues Proposed changes

Merge "Refactor scenario tests to facilitate multi-group"

Browse Source
This commit is contained in:
Jenkins 2016-05-12 19:55:22 +00:00 committed by Gerrit Code Review
commit 5c364df2c1
21 changed files with 494 additions and 255 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
trove/tests/int_tests.py
View File

@ -156,9 +156,6 @@ guest_log_groups.extend([guest_log_group.GROUP])
instance_actions_groups = list(instance_create_groups)
instance_actions_groups.extend([instance_actions_group.GROUP])
instance_module_groups = list(instance_create_groups)
instance_module_groups.extend([module_group.GROUP_INSTANCE_MODULE])
module_groups = list(instance_create_groups)
module_groups.extend([module_group.GROUP])
@ -187,7 +184,6 @@ register(["database"], database_actions_groups)
register(["guest_log"], guest_log_groups)
register(["instance", "instance_actions"], instance_actions_groups)
register(["instance_create"], instance_create_groups)
register(["instance_module"], instance_module_groups)
register(["module"], module_groups)
register(["module_create"], module_create_groups)
register(["replication"], replication_groups)

9
trove/tests/scenario/groups/backup_group.py
View File

@ -17,6 +17,7 @@ from proboscis import test
from trove.tests.scenario.groups import instance_create_group
from trove.tests.scenario.groups.test_group import TestGroup
from trove.tests.scenario.runners import test_runners
GROUP = "scenario.backup_restore_group"
@ -25,13 +26,19 @@ GROUP_BACKUP_LIST = "scenario.backup_list_group"
GROUP_RESTORE = "scenario.restore_group"
class BackupRunnerFactory(test_runners.RunnerFactory):
_runner_ns = 'backup_runners'
_runner_cls = 'BackupRunner'
@test(depends_on_groups=[instance_create_group.GROUP], groups=[GROUP])
class BackupGroup(TestGroup):
"""Test Backup and Restore functionality."""
def __init__(self):
super(BackupGroup, self).__init__(
'backup_runners', 'BackupRunner')
BackupRunnerFactory.instance())
@test(groups=[GROUP_BACKUP])
def backup_create_instance_invalid(self):

9
trove/tests/scenario/groups/cluster_actions_group.py
View File

@ -16,17 +16,24 @@
from proboscis import test
from trove.tests.scenario.groups.test_group import TestGroup
from trove.tests.scenario.runners import test_runners
GROUP = "scenario.cluster_actions_group"
class ClusterActionsRunnerFactory(test_runners.RunnerFactory):
_runner_ns = 'cluster_actions_runners'
_runner_cls = 'ClusterActionsRunner'
@test(groups=[GROUP])
class ClusterActionsGroup(TestGroup):
def __init__(self):
super(ClusterActionsGroup, self).__init__(
'cluster_actions_runners', 'ClusterActionsRunner')
ClusterActionsRunnerFactory.instance())
@test
def cluster_create(self):

9
trove/tests/scenario/groups/configuration_group.py
View File

@ -17,17 +17,24 @@ from proboscis import test
from trove.tests.scenario.groups import instance_create_group
from trove.tests.scenario.groups.test_group import TestGroup
from trove.tests.scenario.runners import test_runners
GROUP = "scenario.configuration_group"
class ConfigurationRunnerFactory(test_runners.RunnerFactory):
_runner_ns = 'configuration_runners'
_runner_cls = 'ConfigurationRunner'
@test(depends_on_groups=[instance_create_group.GROUP], groups=[GROUP])
class ConfigurationGroup(TestGroup):
def __init__(self):
super(ConfigurationGroup, self).__init__(
'configuration_runners', 'ConfigurationRunner')
ConfigurationRunnerFactory.instance())
@test
def create_bad_group(self):

13
trove/tests/scenario/groups/database_actions_group.py
View File

@ -17,19 +17,26 @@ from proboscis import test
from trove.tests.scenario.groups import instance_create_group
from trove.tests.scenario.groups.test_group import TestGroup
from trove.tests.scenario.runners import test_runners
GROUP = "scenario.database_actions_group"
class DatabaseActionsRunnerFactory(test_runners.RunnerFactory):
_runner_ns = 'database_actions_runners'
_runner_cls = 'DatabaseActionsRunner'
@test(depends_on_groups=[instance_create_group.GROUP], groups=[GROUP])
class DatabaseActionsGroup(TestGroup):
def __init__(self):
super(DatabaseActionsGroup, self).__init__(
'database_actions_runners', 'DatabaseActionsRunner')
self.instance_create_runner = self.get_runner(
'instance_create_runners', 'InstanceCreateRunner')
DatabaseActionsRunnerFactory.instance())
self.instance_create_runner = (
instance_create_group.InstanceCreateRunnerFactory.create())
@test
def create_initialized_instance(self):

9
trove/tests/scenario/groups/guest_log_group.py
View File

@ -16,18 +16,25 @@ from proboscis import test
from trove.tests.scenario.groups import instance_create_group
from trove.tests.scenario.groups.test_group import TestGroup
from trove.tests.scenario.runners import test_runners
GROUP = "scenario.guest_log_group"
class GuestLogRunnerFactory(test_runners.RunnerFactory):
_runner_ns = 'guest_log_runners'
_runner_cls = 'GuestLogRunner'
@test(depends_on_groups=[instance_create_group.GROUP], groups=[GROUP])
class GuestLogGroup(TestGroup):
"""Test Guest Log functionality."""
def __init__(self):
super(GuestLogGroup, self).__init__(
'guest_log_runners', 'GuestLogRunner')
GuestLogRunnerFactory.instance())
@test
def test_log_list(self):

9
trove/tests/scenario/groups/instance_actions_group.py
View File

@ -17,17 +17,24 @@ from proboscis import test
from trove.tests.scenario.groups import instance_create_group
from trove.tests.scenario.groups.test_group import TestGroup
from trove.tests.scenario.runners import test_runners
GROUP = "scenario.instance_actions_group"
class InstanceActionsRunnerFactory(test_runners.RunnerFactory):
_runner_ns = 'instance_actions_runners'
_runner_cls = 'InstanceActionsRunner'
@test(depends_on_groups=[instance_create_group.GROUP], groups=[GROUP])
class InstanceActionsGroup(TestGroup):
def __init__(self):
super(InstanceActionsGroup, self).__init__(
'instance_actions_runners', 'InstanceActionsRunner')
InstanceActionsRunnerFactory.instance())
@test
def instance_restart(self):

15
trove/tests/scenario/groups/instance_create_group.py
View File

@ -15,21 +15,28 @@
from proboscis import test
from trove.tests.api.instances import InstanceSetup
from trove.tests import PRE_INSTANCES
from trove.tests.scenario.groups.test_group import TestGroup
from trove.tests.scenario.runners import test_runners
GROUP = "scenario.instance_create_group"
@test(depends_on_classes=[InstanceSetup], runs_after_groups=[PRE_INSTANCES],
groups=[GROUP])
class InstanceCreateRunnerFactory(test_runners.RunnerFactory):
_runner_ns = 'instance_create_runners'
_runner_cls = 'InstanceCreateRunner'
@test(groups=[GROUP],
depends_on_groups=["services.initialize"],
runs_after_groups=[PRE_INSTANCES])
class InstanceCreateGroup(TestGroup):
def __init__(self):
super(InstanceCreateGroup, self).__init__(
'instance_create_runners', 'InstanceCreateRunner')
InstanceCreateRunnerFactory.instance())
@test
def create_empty_instance(self):

9
trove/tests/scenario/groups/instance_delete_group.py
View File

@ -25,11 +25,18 @@ from trove.tests.scenario.groups import replication_group
from trove.tests.scenario.groups import root_actions_group
from trove.tests.scenario.groups.test_group import TestGroup
from trove.tests.scenario.groups import user_actions_group
from trove.tests.scenario.runners import test_runners
GROUP = "scenario.instance_delete_group"
class InstanceDeleteRunnerFactory(test_runners.RunnerFactory):
_runner_ns = 'instance_delete_runners'
_runner_cls = 'InstanceDeleteRunner'
@test(depends_on_groups=[instance_create_group.GROUP],
groups=[GROUP],
runs_after_groups=[backup_group.GROUP_BACKUP,
@ -44,7 +51,7 @@ class InstanceDeleteGroup(TestGroup):
def __init__(self):
super(InstanceDeleteGroup, self).__init__(
'instance_delete_runners', 'InstanceDeleteRunner')
InstanceDeleteRunnerFactory.instance())
@test
def instance_delete(self):

51
trove/tests/scenario/groups/module_group.py
View File

@ -18,21 +18,28 @@ from proboscis import test
from trove.tests.scenario.groups import instance_create_group
from trove.tests.scenario.groups.test_group import TestGroup
from trove.tests.scenario.runners import test_runners
GROUP = "scenario.module_group"
GROUP_MODULE_CREATE = "scenario.module_create_group"
GROUP_INSTANCE_MODULE = "scenario.instance_module_group"
GROUP_MODULE_INSTANCE = "scenario.module_instance_group"
GROUP_MODULE_DELETE = "scenario.module_delete_group"
class ModuleRunnerFactory(test_runners.RunnerFactory):
_runner_ns = 'module_runners'
_runner_cls = 'ModuleRunner'
@test(groups=[GROUP, GROUP_MODULE_CREATE])
class ModuleGroup(TestGroup):
"""Test Module functionality."""
def __init__(self):
super(ModuleGroup, self).__init__(
'module_runners', 'ModuleRunner')
ModuleRunnerFactory.instance())
@test(groups=[GROUP, GROUP_MODULE_CREATE])
def module_delete_existing(self):
@ -319,63 +326,63 @@ class ModuleGroup(TestGroup):
@test(depends_on_groups=[instance_create_group.GROUP,
GROUP_MODULE_CREATE],
groups=[GROUP, GROUP_INSTANCE_MODULE])
groups=[GROUP, GROUP_MODULE_INSTANCE])
class ModuleInstanceGroup(TestGroup):
"""Test Instance Module functionality."""
def __init__(self):
super(ModuleInstanceGroup, self).__init__(
'module_runners', 'ModuleRunner')
ModuleRunnerFactory.instance())
@test(groups=[GROUP, GROUP_INSTANCE_MODULE])
@test(groups=[GROUP, GROUP_MODULE_INSTANCE])
def module_list_instance_empty(self):
"""Check that the instance has no modules associated."""
self.test_runner.run_module_list_instance_empty()
@test(groups=[GROUP, GROUP_INSTANCE_MODULE],
@test(groups=[GROUP, GROUP_MODULE_INSTANCE],
runs_after=[module_list_instance_empty])
def module_instances_empty(self):
"""Check that the module hasn't been applied to any instances."""
self.test_runner.run_module_instances_empty()
@test(groups=[GROUP, GROUP_INSTANCE_MODULE],
@test(groups=[GROUP, GROUP_MODULE_INSTANCE],
runs_after=[module_instances_empty])
def module_query_empty(self):
"""Check that the instance has no modules applied."""
self.test_runner.run_module_query_empty()
@test(groups=[GROUP, GROUP_INSTANCE_MODULE],
@test(groups=[GROUP, GROUP_MODULE_INSTANCE],
runs_after=[module_query_empty])
def module_apply(self):
"""Check that module-apply works."""
self.test_runner.run_module_apply()
@test(groups=[GROUP, GROUP_INSTANCE_MODULE],
@test(groups=[GROUP, GROUP_MODULE_INSTANCE],
depends_on=[module_apply])
def module_list_instance_after_apply(self):
"""Check that the instance has one module associated."""
self.test_runner.run_module_list_instance_after_apply()
@test(groups=[GROUP, GROUP_INSTANCE_MODULE],
@test(groups=[GROUP, GROUP_MODULE_INSTANCE],
depends_on=[module_apply])
def module_query_after_apply(self):
"""Check that module-query works."""
self.test_runner.run_module_query_after_apply()
@test(groups=[GROUP, GROUP_INSTANCE_MODULE],
@test(groups=[GROUP, GROUP_MODULE_INSTANCE],
depends_on=[module_apply],
runs_after=[module_query_after_apply])
def create_inst_with_mods(self):
"""Check that creating an instance with modules works."""
self.test_runner.run_create_inst_with_mods()
@test(groups=[GROUP, GROUP_INSTANCE_MODULE],
@test(groups=[GROUP, GROUP_MODULE_INSTANCE],
depends_on=[module_apply])
def module_delete_applied(self):
"""Ensure that deleting an applied module fails."""
self.test_runner.run_module_delete_applied()
@test(groups=[GROUP, GROUP_INSTANCE_MODULE],
@test(groups=[GROUP, GROUP_MODULE_INSTANCE],
depends_on=[module_apply],
runs_after=[module_list_instance_after_apply,
module_query_after_apply])
@ -383,54 +390,54 @@ class ModuleInstanceGroup(TestGroup):
"""Check that module-remove works."""
self.test_runner.run_module_remove()
@test(groups=[GROUP, GROUP_INSTANCE_MODULE],
@test(groups=[GROUP, GROUP_MODULE_INSTANCE],
depends_on=[module_remove])
def module_query_empty_after(self):
"""Check that the instance has no modules applied after remove."""
self.test_runner.run_module_query_empty()
@test(groups=[GROUP, GROUP_INSTANCE_MODULE],
@test(groups=[GROUP, GROUP_MODULE_INSTANCE],
depends_on=[create_inst_with_mods],
runs_after=[module_query_empty_after])
def wait_for_inst_with_mods(self):
"""Wait for create instance with modules to finish."""
self.test_runner.run_wait_for_inst_with_mods()
@test(groups=[GROUP, GROUP_INSTANCE_MODULE],
@test(groups=[GROUP, GROUP_MODULE_INSTANCE],
depends_on=[wait_for_inst_with_mods])
def module_query_after_inst_create(self):
"""Check that module-query works on new instance."""
self.test_runner.run_module_query_after_inst_create()
@test(groups=[GROUP, GROUP_INSTANCE_MODULE],
@test(groups=[GROUP, GROUP_MODULE_INSTANCE],
depends_on=[wait_for_inst_with_mods],
runs_after=[module_query_after_inst_create])
def module_retrieve_after_inst_create(self):
"""Check that module-retrieve works on new instance."""
self.test_runner.run_module_retrieve_after_inst_create()
@test(groups=[GROUP, GROUP_INSTANCE_MODULE],
@test(groups=[GROUP, GROUP_MODULE_INSTANCE],
depends_on=[wait_for_inst_with_mods],
runs_after=[module_retrieve_after_inst_create])
def module_query_after_inst_create_admin(self):
"""Check that module-query works for admin."""
self.test_runner.run_module_query_after_inst_create_admin()
@test(groups=[GROUP, GROUP_INSTANCE_MODULE],
@test(groups=[GROUP, GROUP_MODULE_INSTANCE],
depends_on=[wait_for_inst_with_mods],
runs_after=[module_query_after_inst_create_admin])
def module_retrieve_after_inst_create_admin(self):
"""Check that module-retrieve works for admin."""
self.test_runner.run_module_retrieve_after_inst_create_admin()
@test(groups=[GROUP, GROUP_INSTANCE_MODULE],
@test(groups=[GROUP, GROUP_MODULE_INSTANCE],
depends_on=[wait_for_inst_with_mods],
runs_after=[module_retrieve_after_inst_create_admin])
def module_delete_auto_applied(self):
"""Ensure that module-delete on auto-applied module fails."""
self.test_runner.run_module_delete_auto_applied()
@test(groups=[GROUP, GROUP_INSTANCE_MODULE],
@test(groups=[GROUP, GROUP_MODULE_INSTANCE],
depends_on=[wait_for_inst_with_mods],
runs_after=[module_delete_auto_applied])
def delete_inst_with_mods(self):
@ -445,7 +452,7 @@ class ModuleDeleteGroup(TestGroup):
def __init__(self):
super(ModuleDeleteGroup, self).__init__(
'module_runners', 'ModuleRunner')
ModuleRunnerFactory.instance())
@test(groups=[GROUP, GROUP_MODULE_DELETE])
def module_delete_non_existent(self):

9
trove/tests/scenario/groups/negative_cluster_actions_group.py
View File

@ -16,17 +16,24 @@
from proboscis import test
from trove.tests.scenario.groups.test_group import TestGroup
from trove.tests.scenario.runners import test_runners
GROUP = "scenario.negative_cluster_actions_group"
class NegativeClusterActionsRunnerFactory(test_runners.RunnerFactory):
_runner_ns = 'negative_cluster_actions_runners'
_runner_cls = 'NegativeClusterActionsRunner'
@test(groups=[GROUP])
class NegativeClusterActionsGroup(TestGroup):
def __init__(self):
super(NegativeClusterActionsGroup, self).__init__(
'negative_cluster_actions_runners', 'NegativeClusterActionsRunner')
NegativeClusterActionsRunnerFactory.instance())
@test
def create_constrained_size_cluster(self):

9
trove/tests/scenario/groups/replication_group.py
View File

@ -17,18 +17,25 @@ from proboscis import test
from trove.tests.scenario.groups import instance_create_group
from trove.tests.scenario.groups.test_group import TestGroup
from trove.tests.scenario.runners import test_runners
GROUP = "scenario.replication_group"
class ReplicationRunnerFactory(test_runners.RunnerFactory):
_runner_ns = 'replication_runners'
_runner_cls = 'ReplicationRunner'
@test(depends_on_groups=[instance_create_group.GROUP], groups=[GROUP])
class ReplicationGroup(TestGroup):
"""Test Replication functionality."""
def __init__(self):
super(ReplicationGroup, self).__init__(
'replication_runners', 'ReplicationRunner')
ReplicationRunnerFactory.instance())
@test
def add_data_for_replication(self):

16
trove/tests/scenario/groups/root_actions_group.py
View File

@ -15,23 +15,29 @@
from proboscis import test
from trove.tests.scenario.groups import backup_group
from trove.tests.scenario.groups import instance_create_group
from trove.tests.scenario.groups.test_group import TestGroup
from trove.tests.scenario.runners import test_runners
GROUP = "scenario.root_actions_group"
class RootActionsRunnerFactory(test_runners.RunnerFactory):
_runner_ns = 'root_actions_runners'
_runner_cls = 'RootActionsRunner'
@test(depends_on_groups=[instance_create_group.GROUP], groups=[GROUP])
class RootActionsGroup(TestGroup):
def __init__(self):
super(RootActionsGroup, self).__init__(
'root_actions_runners', 'RootActionsRunner')
self.backup_runner = self.get_runner(
'backup_runners', 'BackupRunner')
self.backup_runner2 = self.get_runner(
'backup_runners', 'BackupRunner')
RootActionsRunnerFactory.instance())
self.backup_runner = backup_group.BackupRunnerFactory.create()
self.backup_runner2 = backup_group.BackupRunnerFactory.create()
@test
def check_root_never_enabled(self):

84
trove/tests/scenario/groups/test_group.py
View File

@ -16,92 +16,12 @@
import abc
import six
from trove.common.strategies.strategy import Strategy
from trove.tests.config import CONFIG
@six.add_metaclass(abc.ABCMeta)
class TestGroup(object):
TEST_RUNNERS_NS = 'trove.tests.scenario.runners'
TEST_HELPERS_NS = 'trove.tests.scenario.helpers'
TEST_HELPER_MODULE_NAME = 'test_helper'
TEST_HELPER_BASE_NAME = 'TestHelper'
def __init__(self, runner_module_name, runner_base_name, *args, **kwargs):
self._test_runner = self.get_runner(
runner_module_name, runner_base_name, *args, **kwargs)
def get_runner(self, runner_module_name, runner_base_name,
*args, **kwargs):
class_prefix = self._get_test_datastore()
runner_cls = self._load_dynamic_class(
runner_module_name, class_prefix, runner_base_name,
self.TEST_RUNNERS_NS)
runner = runner_cls(*args, **kwargs)
runner._test_helper = self.get_helper()
return runner
def get_helper(self):
class_prefix = self._get_test_datastore()
helper_cls = self._load_dynamic_class(
self.TEST_HELPER_MODULE_NAME, class_prefix,
self.TEST_HELPER_BASE_NAME, self.TEST_HELPERS_NS)
return helper_cls(self._build_class_name(
class_prefix, self.TEST_HELPER_BASE_NAME, strip_test=True))
def _get_test_datastore(self):
return CONFIG.dbaas_datastore
def _load_dynamic_class(self, module_name, class_prefix, base_name,
namespace):
"""Try to load a datastore specific class if it exists; use the
default otherwise.
"""
# This is for overridden Runner classes
impl = self._build_class_path(module_name,
class_prefix, base_name)
cls = self._load_class('runner', impl, namespace)
if not cls:
# This is for overridden Helper classes
module = module_name.replace('test', class_prefix.lower())
impl = self._build_class_path(module, class_prefix, base_name,
strip_test=True)
cls = self._load_class('helper', impl, namespace)
if not cls:
# Just import the base class
impl = self._build_class_path(module_name, '', base_name)
cls = self._load_class(None, impl, namespace)
return cls
def _load_class(self, load_type, impl, namespace):
cls = None
if not load_type or load_type in impl.lower():
try:
cls = Strategy.get_strategy(impl, namespace)
except ImportError as ie:
# Only fail silently if it's something we expect,
# such as a missing override class. Anything else
# shouldn't be suppressed.
l_msg = ie.message.lower()
if load_type not in l_msg or (
'no module named' not in l_msg and
'cannot be found' not in l_msg):
raise
return cls
def _build_class_path(self, module_name, class_prefix, class_base,
strip_test=False):
class_name = self._build_class_name(class_prefix, class_base,
strip_test)
return '%s.%s' % (module_name, class_name)
def _build_class_name(self, class_prefix, base_name, strip_test=False):
base = (base_name.replace('Test', '') if strip_test else base_name)
return '%s%s' % (class_prefix.capitalize(), base)
def __init__(self, test_runner):
self._test_runner = test_runner
@property
def test_runner(self):

18
trove/tests/scenario/groups/user_actions_group.py
View File

@ -15,23 +15,31 @@
from proboscis import test
from trove.tests.scenario.groups import database_actions_group
from trove.tests.scenario.groups import instance_create_group
from trove.tests.scenario.groups.test_group import TestGroup
from trove.tests.scenario.runners import test_runners
GROUP = "scenario.user_actions_group"
class UserActionsRunnerFactory(test_runners.RunnerFactory):
_runner_ns = 'user_actions_runners'
_runner_cls = 'UserActionsRunner'
@test(depends_on_groups=[instance_create_group.GROUP], groups=[GROUP])
class UserActionsGroup(TestGroup):
def __init__(self):
super(UserActionsGroup, self).__init__(
'user_actions_runners', 'UserActionsRunner')
self.instance_create_runner = self.get_runner(
'instance_create_runners', 'InstanceCreateRunner')
self.database_actions_runner = self.get_runner(
'database_actions_runners', 'DatabaseActionsRunner')
UserActionsRunnerFactory.instance())
self.instance_create_runner = (
instance_create_group.InstanceCreateRunnerFactory.create())
self.database_actions_runner = (
database_actions_group.DatabaseActionsRunnerFactory.create())
@test
def create_initialized_instance(self):

6
trove/tests/scenario/helpers/mysql_helper.py
View File

@ -33,11 +33,11 @@ class MysqlHelper(SqlHelper):
{'name': 'db2'}, {"name": 'db3'}]
def get_valid_user_definitions(self):
return [{'name': 'user1', 'password': 'password1', 'databases': [],
return [{'name': 'a_user1', 'password': 'password1', 'databases': [],
'host': '127.0.0.1'},
{'name': 'user2', 'password': 'password1',
{'name': 'a_user2', 'password': 'password1',
'databases': [{'name': 'db1'}], 'host': '0.0.0.0'},
{'name': 'user3', 'password': 'password1',
{'name': 'a_user3', 'password': 'password1',
'databases': [{'name': 'db1'}, {'name': 'db2'}]}]
def get_dynamic_group(self):

16
trove/tests/scenario/runners/configuration_runners.py
View File

@ -18,12 +18,9 @@ import json
from proboscis import SkipTest
from trove.common.utils import generate_uuid
from trove.tests.config import CONFIG
from trove.tests.scenario.runners.test_runners import TestRunner
from trove.tests.util.check import CollectionCheck
from trove.tests.util.check import TypeCheck
from trove.tests.util import create_dbaas_client
from trove.tests.util.users import Requirements
from troveclient.compat import exceptions
@ -39,7 +36,6 @@ class ConfigurationRunner(TestRunner):
self.non_dynamic_inst_count = 0
self.initial_group_count = 0
self.additional_group_count = 0
self.other_client = None
self.config_id_for_inst = None
self.config_inst_id = None
@ -273,20 +269,12 @@ class ConfigurationRunner(TestRunner):
def assert_conf_get_unauthorized_user(
self, config_id, expected_exception=exceptions.NotFound,
expected_http_code=404):
self._create_other_client()
self.assert_raises(
expected_exception, None,
self.other_client.configurations.get, config_id)
self.unauth_client.configurations.get, config_id)
# we're using a different client, so we'll check the return code
# on it explicitly, instead of depending on 'assert_raises'
self.assert_client_code(expected_http_code, client=self.other_client)
def _create_other_client(self):
if not self.other_client:
requirements = Requirements(is_admin=False)
other_user = CONFIG.users.find_user(
requirements, black_list=[self.instance_info.user.auth_user])
self.other_client = create_dbaas_client(other_user)
self.assert_client_code(expected_http_code, client=self.unauth_client)
def run_non_dynamic_conf_get_unauthorized_user(
self, expected_exception=exceptions.NotFound,

50
trove/tests/scenario/runners/instance_create_runners.py
View File

@ -17,9 +17,10 @@ import json
from proboscis import SkipTest
from trove.tests.api.instances import CheckInstance, InstanceTestInfo
from trove.tests.config import CONFIG
from trove.tests.scenario.helpers.test_helper import DataType
from trove.tests.scenario.runners.test_runners import CheckInstance
from trove.tests.scenario.runners.test_runners import InstanceTestInfo
from trove.tests.scenario.runners.test_runners import TestRunner
@ -32,7 +33,8 @@ class InstanceCreateRunner(TestRunner):
self.init_inst_users = None
self.init_inst_host = None
self.init_inst_data = None
self.init_config_group_id = None
self.init_inst_config_group_id = None
self.config_group_id = None
def run_empty_instance_create(
self, expected_states=['BUILD', 'ACTIVE'], expected_http_code=200):
@ -40,20 +42,21 @@ class InstanceCreateRunner(TestRunner):
flavor = self._get_instance_flavor()
trove_volume_size = CONFIG.get('trove_volume_size', 1)
info = self.assert_instance_create(
instance_info = self.assert_instance_create(
name, flavor, trove_volume_size, [], [], None, None,
CONFIG.dbaas_datastore, CONFIG.dbaas_datastore_version,
expected_states, expected_http_code, create_helper_user=True)
# Update the shared instance info.
self.instance_info.databases = info.databases
self.instance_info.users = info.users
self.instance_info.dbaas_datastore = info.dbaas_datastore
self.instance_info.dbaas_datastore_version = (info.
dbaas_datastore_version)
self.instance_info.dbaas_flavor_href = info.dbaas_flavor_href
self.instance_info.volume = info.volume
self.instance_info.id = info.id
self.instance_info.id = instance_info.id
self.instance_info.name = instance_info.name
self.instance_info.databases = instance_info.databases
self.instance_info.users = instance_info.users
self.instance_info.dbaas_datastore = instance_info.dbaas_datastore
self.instance_info.dbaas_datastore_version = (
instance_info.dbaas_datastore_version)
self.instance_info.dbaas_flavor_href = instance_info.dbaas_flavor_href
self.instance_info.volume = instance_info.volume
def run_initial_configuration_create(self, expected_http_code=200):
dynamic_config = self.test_helper.get_dynamic_group()
@ -69,7 +72,7 @@ class InstanceCreateRunner(TestRunner):
datastore_version=self.instance_info.dbaas_datastore_version)
self.assert_client_code(expected_http_code)
self.init_config_group_id = result.id
self.config_group_id = result.id
else:
raise SkipTest("No groups defined.")
@ -83,6 +86,7 @@ class InstanceCreateRunner(TestRunner):
# test instances.
raise SkipTest("Using an existing instance.")
configuration_id = configuration_id or self.config_group_id
name = self.instance_info.name + '_init'
flavor = self._get_instance_flavor()
trove_volume_size = CONFIG.get('trove_volume_size', 1)
@ -90,15 +94,12 @@ class InstanceCreateRunner(TestRunner):
if with_dbs else [])
self.init_inst_users = (self.test_helper.get_valid_user_definitions()
if with_users else [])
if configuration_id:
self.init_config_group_id = configuration_id
if (self.init_inst_dbs or self.init_inst_users or
self.init_config_group_id):
self.init_inst_config_group_id = configuration_id
if (self.init_inst_dbs or self.init_inst_users or configuration_id):
info = self.assert_instance_create(
name, flavor, trove_volume_size,
self.init_inst_dbs, self.init_inst_users,
self.init_config_group_id, None,
configuration_id, None,
CONFIG.dbaas_datastore, CONFIG.dbaas_datastore_version,
expected_states, expected_http_code,
create_helper_user=create_helper_user)
@ -244,7 +245,7 @@ class InstanceCreateRunner(TestRunner):
if self.init_inst_id:
self.assert_instance_properties(
self.init_inst_id, self.init_inst_dbs, self.init_inst_users,
self.init_config_group_id, self.init_inst_data)
self.init_inst_config_group_id, self.init_inst_data)
def assert_instance_properties(
self, instance_id, expected_dbs_definitions,
@ -316,10 +317,17 @@ class InstanceCreateRunner(TestRunner):
self.assert_all_gone(self.init_inst_id, expected_states[-1])
else:
raise SkipTest("Cleanup is not required.")
self.init_inst_id = None
self.init_inst_dbs = None
self.init_inst_users = None
self.init_inst_host = None
self.init_inst_data = None
self.init_inst_config_group_id = None
def run_initial_configuration_delete(self, expected_http_code=202):
if self.init_config_group_id:
self.auth_client.configurations.delete(self.init_config_group_id)
if self.config_group_id:
self.auth_client.configurations.delete(self.config_group_id)
self.assert_client_code(expected_http_code)
else:
raise SkipTest("Cleanup is not required.")
self.config_group_id = None

132
trove/tests/scenario/runners/module_runners.py
View File

@ -26,27 +26,6 @@ from trove.module import models
from trove.tests.scenario.runners.test_runners import TestRunner
# Variables here are set up to be used across multiple groups,
# since each group will instantiate a new runner
random_data = Crypto.Random.new().read(20)
test_modules = []
module_count_prior_to_create = 0
module_ds_count_prior_to_create = 0
module_ds_all_count_prior_to_create = 0
module_all_tenant_count_prior_to_create = 0
module_auto_apply_count_prior_to_create = 0
module_admin_count_prior_to_create = 0
module_other_count_prior_to_create = 0
module_create_count = 0
module_ds_create_count = 0
module_ds_all_create_count = 0
module_all_tenant_create_count = 0
module_auto_apply_create_count = 0
module_admin_create_count = 0
module_other_create_count = 0
class ModuleRunner(TestRunner):
def __init__(self):
@ -62,13 +41,30 @@ class ModuleRunner(TestRunner):
self.MODULE_NEG_CONTENTS = 'contents for negative tests'
self.MODULE_BINARY_SUFFIX = '_bin_auto'
self.MODULE_BINARY_SUFFIX2 = self.MODULE_BINARY_SUFFIX + '_2'
self.MODULE_BINARY_CONTENTS = random_data
self.MODULE_BINARY_CONTENTS = Crypto.Random.new().read(20)
self.MODULE_BINARY_CONTENTS2 = '\x00\xFF\xea\x9c\x11\xfeok\xb1\x8ax'
self.mod_inst_id = None
self.temp_module = None
self._module_type = None
self.test_modules = []
self.module_count_prior_to_create = 0
self.module_ds_count_prior_to_create = 0
self.module_ds_all_count_prior_to_create = 0
self.module_all_tenant_count_prior_to_create = 0
self.module_auto_apply_count_prior_to_create = 0
self.module_admin_count_prior_to_create = 0
self.module_other_count_prior_to_create = 0
self.module_create_count = 0
self.module_ds_create_count = 0
self.module_ds_all_create_count = 0
self.module_all_tenant_create_count = 0
self.module_auto_apply_create_count = 0
self.module_admin_create_count = 0
self.module_other_create_count = 0
@property
def module_type(self):
if not self._module_type:
@ -77,9 +73,9 @@ class ModuleRunner(TestRunner):
@property
def main_test_module(self):
if not test_modules or not test_modules[0]:
if not self.test_modules or not self.test_modules[0]:
SkipTest("No main module created")
return test_modules[0]
return self.test_modules[0]
def build_module_args(self, extra=None):
extra = extra or ''
@ -103,7 +99,7 @@ class ModuleRunner(TestRunner):
def _find_module(self, match_fn, not_found_message, find_all=False):
found = [] if find_all else None
for test_module in test_modules:
for test_module in self.test_modules:
if match_fn(test_module):
if find_all:
found.append(test_module)
@ -203,29 +199,22 @@ class ModuleRunner(TestRunner):
def run_module_create(self):
# Necessary to test that the count increases.
global module_count_prior_to_create
global module_ds_count_prior_to_create
global module_ds_all_count_prior_to_create
global module_all_tenant_count_prior_to_create
global module_auto_apply_count_prior_to_create
global module_admin_count_prior_to_create
global module_other_count_prior_to_create
module_count_prior_to_create = len(
self.module_count_prior_to_create = len(
self.auth_client.modules.list())
module_ds_count_prior_to_create = len(
self.module_ds_count_prior_to_create = len(
self.auth_client.modules.list(
datastore=self.instance_info.dbaas_datastore))
module_ds_all_count_prior_to_create = len(
self.module_ds_all_count_prior_to_create = len(
self.auth_client.modules.list(
datastore=models.Modules.MATCH_ALL_NAME))
module_all_tenant_count_prior_to_create = len(
self.module_all_tenant_count_prior_to_create = len(
self.unauth_client.modules.list())
module_auto_apply_count_prior_to_create = len(
self.module_auto_apply_count_prior_to_create = len(
[module for module in self.admin_client.modules.list()
if module.auto_apply])
module_admin_count_prior_to_create = len(
self.module_admin_count_prior_to_create = len(
self.admin_client.modules.list())
module_other_count_prior_to_create = len(
self.module_other_count_prior_to_create = len(
self.unauth_client.modules.list())
name, description, contents = self.build_module_args()
self.assert_module_create(
@ -248,30 +237,22 @@ class ModuleRunner(TestRunner):
datastore=datastore, datastore_version=datastore_version,
auto_apply=auto_apply,
live_update=live_update, visible=visible)
global module_create_count
global module_ds_create_count
global module_ds_all_create_count
global module_auto_apply_create_count
global module_all_tenant_create_count
global module_admin_create_count
global module_other_create_count
if (client == self.auth_client or
(client == self.admin_client and visible)):
module_create_count += 1
self.module_create_count += 1
if datastore:
module_ds_create_count += 1
self.module_ds_create_count += 1
else:
module_ds_all_create_count += 1
self.module_ds_all_create_count += 1
elif not visible:
module_admin_create_count += 1
self.module_admin_create_count += 1
else:
module_other_create_count += 1
self.module_other_create_count += 1
if all_tenants and visible:
module_all_tenant_create_count += 1
self.module_all_tenant_create_count += 1
if auto_apply and visible:
module_auto_apply_create_count += 1
global test_modules
test_modules.append(result)
self.module_auto_apply_create_count += 1
self.test_modules.append(result)
tenant_id = None
tenant = models.Modules.MATCH_ALL_NAME
@ -413,7 +394,7 @@ class ModuleRunner(TestRunner):
def run_module_list(self):
self.assert_module_list(
self.auth_client,
module_count_prior_to_create + module_create_count)
self.module_count_prior_to_create + self.module_create_count)
def assert_module_list(self, client, expected_count, datastore=None,
skip_validation=False):
@ -441,8 +422,9 @@ class ModuleRunner(TestRunner):
def run_module_list_unauth_user(self):
self.assert_module_list(
self.unauth_client,
module_all_tenant_count_prior_to_create +
module_all_tenant_create_count + module_other_create_count)
(self.module_all_tenant_count_prior_to_create +
self.module_all_tenant_create_count +
self.module_other_create_count))
def run_module_create_admin_all(self):
name, description, contents = self.build_module_args(
@ -517,20 +499,21 @@ class ModuleRunner(TestRunner):
def run_module_list_again(self):
self.assert_module_list(
self.auth_client,
module_count_prior_to_create + module_create_count,
self.module_count_prior_to_create + self.module_create_count,
skip_validation=True)
def run_module_list_ds(self):
self.assert_module_list(
self.auth_client,
module_ds_count_prior_to_create + module_ds_create_count,
self.module_ds_count_prior_to_create + self.module_ds_create_count,
datastore=self.instance_info.dbaas_datastore,
skip_validation=True)
def run_module_list_ds_all(self):
self.assert_module_list(
self.auth_client,
module_ds_all_count_prior_to_create + module_ds_all_create_count,
(self.module_ds_all_count_prior_to_create +
self.module_ds_all_create_count),
datastore=models.Modules.MATCH_ALL_NAME,
skip_validation=True)
@ -545,10 +528,10 @@ class ModuleRunner(TestRunner):
def run_module_list_admin(self):
self.assert_module_list(
self.admin_client,
(module_admin_count_prior_to_create +
module_create_count +
module_admin_create_count +
module_other_create_count),
(self.module_admin_count_prior_to_create +
self.module_create_count +
self.module_admin_create_count +
self.module_other_create_count),
skip_validation=True)
def run_module_update(self):
@ -599,17 +582,16 @@ class ModuleRunner(TestRunner):
def assert_module_update(self, client, module_id, **kwargs):
result = client.modules.update(module_id, **kwargs)
global test_modules
found = False
index = -1
for test_module in test_modules:
for test_module in self.test_modules:
index += 1
if test_module.id == module_id:
found = True
break
if not found:
self.fail("Could not find updated module in module list")
test_modules[index] = result
self.test_modules[index] = result
expected_args = {}
for key, value in kwargs.items():
@ -701,7 +683,7 @@ class ModuleRunner(TestRunner):
def run_module_list_instance_empty(self):
self.assert_module_list_instance(
self.auth_client, self.instance_info.id,
module_auto_apply_count_prior_to_create)
self.module_auto_apply_count_prior_to_create)
def assert_module_list_instance(self, client, instance_id, expected_count,
expected_http_code=200):
@ -728,7 +710,7 @@ class ModuleRunner(TestRunner):
def run_module_query_empty(self):
self.assert_module_query(self.auth_client, self.instance_info.id,
module_auto_apply_count_prior_to_create)
self.module_auto_apply_count_prior_to_create)
def assert_module_query(self, client, instance_id, expected_count,
expected_http_code=200, expected_results=None):
@ -826,7 +808,7 @@ class ModuleRunner(TestRunner):
self.auth_client, self.instance_info.id, 1)
def run_module_query_after_apply(self):
expected_count = module_auto_apply_count_prior_to_create + 1
expected_count = self.module_auto_apply_count_prior_to_create + 1
expected_results = self.create_default_query_expected_results(
[self.main_test_module])
self.assert_module_query(self.auth_client, self.instance_info.id,
@ -1029,23 +1011,23 @@ class ModuleRunner(TestRunner):
def run_module_delete(self):
expected_count = len(self.auth_client.modules.list()) - 1
test_module = test_modules.pop(0)
test_module = self.test_modules.pop(0)
self.assert_module_delete(self.auth_client, test_module.id,
expected_count)
def run_module_delete_admin(self):
start_count = count = len(self.admin_client.modules.list())
for test_module in test_modules:
for test_module in self.test_modules:
count -= 1
self.report.log("Deleting module '%s' (tenant: %s)" % (
test_module.name, test_module.tenant_id))
self.assert_module_delete(self.admin_client, test_module.id, count)
self.assert_not_equal(start_count, count, "Nothing was deleted")
count = len(self.admin_client.modules.list())
self.assert_equal(module_admin_count_prior_to_create, count,
self.assert_equal(self.module_admin_count_prior_to_create, count,
"Wrong number of admin modules after deleting all")
count = len(self.auth_client.modules.list())
self.assert_equal(module_count_prior_to_create, count,
self.assert_equal(self.module_count_prior_to_create, count,
"Wrong number of modules after deleting all")
def assert_module_delete(self, client, module_id, expected_count):

2
trove/tests/scenario/runners/replication_runners.py
View File

@ -14,8 +14,8 @@
# under the License.
from trove.common import utils
from trove.tests.api.instances import CheckInstance
from trove.tests.scenario.helpers.test_helper import DataType
from trove.tests.scenario.runners.test_runners import CheckInstance
from trove.tests.scenario.runners.test_runners import TestRunner
from troveclient.compat import exceptions

270
trove/tests/scenario/runners/test_runners.py
View File

@ -13,6 +13,7 @@
# License for the specific language governing permissions and limitations
# under the License.
import datetime
import os
import time as timer
@ -23,15 +24,137 @@ from troveclient.compat import exceptions
from trove.common import cfg
from trove.common import exception
from trove.common.strategies.strategy import Strategy
from trove.common import utils
from trove.common.utils import poll_until, build_polling_task
from trove.tests.api.instances import instance_info
from trove.tests.config import CONFIG
from trove.tests.util.check import AttrCheck
from trove.tests.util import create_dbaas_client
from trove.tests.util.users import Requirements
CONF = cfg.CONF
TEST_RUNNERS_NS = 'trove.tests.scenario.runners'
TEST_HELPERS_NS = 'trove.tests.scenario.helpers'
TEST_HELPER_MODULE_NAME = 'test_helper'
TEST_HELPER_BASE_NAME = 'TestHelper'
class RunnerFactory(object):
_test_runner = None
_runner_ns = None
_runner_cls = None
@classmethod
def instance(cls):
"""Returns the current instance of the runner, or creates a new
one if none exists. This is useful to have multiple 'group' classes
use the same runner so that state is maintained.
"""
if not cls._test_runner:
cls._test_runner = cls.create()
return cls._test_runner
@classmethod
def create(cls):
"""Returns a new instance of the runner. Tests that require a 'fresh'
runner (typically from a different 'group') can call this.
"""
return cls._get_runner(cls._runner_ns, cls._runner_cls)
@classmethod
def _get_runner(cls, runner_module_name, runner_base_name,
*args, **kwargs):
class_prefix = cls._get_test_datastore()
runner_cls = cls._load_dynamic_class(
runner_module_name, class_prefix, runner_base_name,
TEST_RUNNERS_NS)
runner = runner_cls(*args, **kwargs)
runner._test_helper = cls._get_helper()
return runner
@classmethod
def _get_helper(cls):
class_prefix = cls._get_test_datastore()
helper_cls = cls._load_dynamic_class(
TEST_HELPER_MODULE_NAME, class_prefix,
TEST_HELPER_BASE_NAME, TEST_HELPERS_NS)
return helper_cls(cls._build_class_name(
class_prefix, TEST_HELPER_BASE_NAME, strip_test=True))
@classmethod
def _get_test_datastore(cls):
return CONFIG.dbaas_datastore
@classmethod
def _load_dynamic_class(cls, module_name, class_prefix, base_name,
namespace):
"""Try to load a datastore specific class if it exists; use the
default otherwise.
"""
# This is for overridden Runner classes
impl = cls._build_class_path(module_name, class_prefix, base_name)
clazz = cls._load_class('runner', impl, namespace)
if not clazz:
# This is for overridden Helper classes
module = module_name.replace('test', class_prefix.lower())
impl = cls._build_class_path(
module, class_prefix, base_name, strip_test=True)
clazz = cls._load_class('helper', impl, namespace)
if not clazz:
# Just import the base class
impl = cls._build_class_path(module_name, '', base_name)
clazz = cls._load_class(None, impl, namespace)
return clazz
@classmethod
def _load_class(cls, load_type, impl, namespace):
clazz = None
if not load_type or load_type in impl.lower():
try:
clazz = Strategy.get_strategy(impl, namespace)
except ImportError as ie:
# Only fail silently if it's something we expect,
# such as a missing override class. Anything else
# shouldn't be suppressed.
l_msg = ie.message.lower()
if load_type not in l_msg or (
'no module named' not in l_msg and
'cannot be found' not in l_msg):
raise
return clazz
@classmethod
def _build_class_path(cls, module_name, class_prefix, class_base,
strip_test=False):
class_name = cls._build_class_name(
class_prefix, class_base, strip_test)
return '%s.%s' % (module_name, class_name)
@classmethod
def _build_class_name(cls, class_prefix, base_name, strip_test=False):
base = (base_name.replace('Test', '') if strip_test else base_name)
return '%s%s' % (class_prefix.capitalize(), base)
class InstanceTestInfo(object):
"""Stores new instance information used by dependent tests."""
def __init__(self):
self.id = None # The ID of the instance in the database.
self.name = None # Test name, generated each test run.
self.dbaas_flavor_href = None # The flavor of the instance.
self.dbaas_datastore = None # The datastore id
self.dbaas_datastore_version = None # The datastore version id
self.volume = None # The volume the instance will have.
self.nics = None # The dict of type/id for nics used on the intance.
self.user = None # The user instance who owns the instance.
self.users = None # The users created on the instance.
class TestRunner(object):
@ -64,21 +187,27 @@ class TestRunner(object):
EPHEMERAL_SUPPORT = not VOLUME_SUPPORT and CONFIG.get('device_path', None)
ROOT_PARTITION = not (VOLUME_SUPPORT or CONFIG.get('device_path', None))
# Here's where the info for the 'main' test instance goes
instance_info = InstanceTestInfo()
report = CONFIG.get_report()
def __init__(self, sleep_time=10, timeout=1200):
self.def_sleep_time = sleep_time
self.def_timeout = timeout
self.instance_info = instance_info
instance_info.dbaas_datastore = CONFIG.dbaas_datastore
instance_info.dbaas_datastore_version = CONFIG.dbaas_datastore_version
self.instance_info.name = "TEST_" + datetime.datetime.strftime(
datetime.datetime.now(), '%Y-%m-%d_%H:%M:%S')
self.instance_info.dbaas_datastore = CONFIG.dbaas_datastore
self.instance_info.dbaas_datastore_version = (
CONFIG.dbaas_datastore_version)
self.instance_info.user = CONFIG.users.find_user_by_name('alt_demo')
if self.VOLUME_SUPPORT:
instance_info.volume = {'size': CONFIG.get('trove_volume_size', 1)}
self.instance_info.volume = {
'size': CONFIG.get('trove_volume_size', 1)}
else:
instance_info.volume = None
self.instance_info.volume = None
self.auth_client = create_dbaas_client(self.instance_info.user)
self._auth_client = None
self._unauth_client = None
self._admin_client = None
self._swift_client = None
@ -151,6 +280,16 @@ class TestRunner(object):
def test_helper(self, test_helper):
self._test_helper = test_helper
@property
def auth_client(self):
if not self._auth_client:
self._auth_client = self._create_authorized_client()
return self._auth_client
def _create_authorized_client(self):
"""Create a client from the normal 'authorized' user."""
return create_dbaas_client(self.instance_info.user)
@property
def unauth_client(self):
if not self._unauth_client:
@ -223,7 +362,11 @@ class TestRunner(object):
@property
def is_using_existing_instance(self):
return self.has_env_flag(self.USE_INSTANCE_ID_FLAG)
return TestRunner.using_existing_instance()
@staticmethod
def using_existing_instance():
return TestRunner.has_env_flag(TestRunner.USE_INSTANCE_ID_FLAG)
@staticmethod
def has_env_flag(flag_name):
@ -449,3 +592,114 @@ class TestRunner(object):
return (database_def,
_get_credentials(credentials),
_get_credentials(credentials_root))
class CheckInstance(AttrCheck):
"""Class to check various attributes of Instance details."""
def __init__(self, instance):
super(CheckInstance, self).__init__()
self.instance = instance
self.volume_support = TestRunner.VOLUME_SUPPORT
self.existing_instance = TestRunner.is_using_existing_instance
def flavor(self):
if 'flavor' not in self.instance:
self.fail("'flavor' not found in instance.")
else:
allowed_attrs = ['id', 'links']
self.contains_allowed_attrs(
self.instance['flavor'], allowed_attrs,
msg="Flavor")
self.links(self.instance['flavor']['links'])
def datastore(self):
if 'datastore' not in self.instance:
self.fail("'datastore' not found in instance.")
else:
allowed_attrs = ['type', 'version']
self.contains_allowed_attrs(
self.instance['datastore'], allowed_attrs,
msg="datastore")
def volume_key_exists(self):
if 'volume' not in self.instance:
self.fail("'volume' not found in instance.")
return False
return True
def volume(self):
if not self.volume_support:
return
if self.volume_key_exists():
allowed_attrs = ['size']
if self.existing_instance:
allowed_attrs.append('used')
self.contains_allowed_attrs(
self.instance['volume'], allowed_attrs,
msg="Volumes")
def used_volume(self):
if not self.volume_support:
return
if self.volume_key_exists():
allowed_attrs = ['size', 'used']
print(self.instance)
self.contains_allowed_attrs(
self.instance['volume'], allowed_attrs,
msg="Volumes")
def volume_mgmt(self):
if not self.volume_support:
return
if self.volume_key_exists():
allowed_attrs = ['description', 'id', 'name', 'size']
self.contains_allowed_attrs(
self.instance['volume'], allowed_attrs,
msg="Volumes")
def addresses(self):
allowed_attrs = ['addr', 'version']
print(self.instance)
networks = ['usernet']
for network in networks:
for address in self.instance['addresses'][network]:
self.contains_allowed_attrs(
address, allowed_attrs,
msg="Address")
def guest_status(self):
allowed_attrs = ['created_at', 'deleted', 'deleted_at', 'instance_id',
'state', 'state_description', 'updated_at']
self.contains_allowed_attrs(
self.instance['guest_status'], allowed_attrs,
msg="Guest status")
def mgmt_volume(self):
if not self.volume_support:
return
allowed_attrs = ['description', 'id', 'name', 'size']
self.contains_allowed_attrs(
self.instance['volume'], allowed_attrs,
msg="Volume")
def replica_of(self):
if 'replica_of' not in self.instance:
self.fail("'replica_of' not found in instance.")
else:
allowed_attrs = ['id', 'links']
self.contains_allowed_attrs(
self.instance['replica_of'], allowed_attrs,
msg="Replica-of links not found")
self.links(self.instance['replica_of']['links'])
def slaves(self):
if 'replicas' not in self.instance:
self.fail("'replicas' not found in instance.")
else:
allowed_attrs = ['id', 'links']
for slave in self.instance['replicas']:
self.contains_allowed_attrs(
slave, allowed_attrs,
msg="Replica links not found")
self.links(slave['links'])