openstack/trove
Code Issues Proposed changes

Refactor scenario tests to facilitate multi-group

Browse Source 
The scenario tests need to have the group classes broken
up to allow interleaving of the different parts of the tests
so that the gate can finish in a reasonable amount of time.

The runners, however, need to be the same for each main group,
thus allowing state to be preserved between groups of the same
feature.  The creation of runners was refactored to allow for this.

The module tests were also modified to use the new paradym, which
allowed the removal of the global varaibles in the runner.

Dependencies on tests.api were also removed.

Change-Id: I66fcb2bce8c6cbc99c9ff071c242e4741500056d
Partial-Bug: #1571092
This commit is contained in:
Peter Stachowski 2016-04-19 22:16:17 -04:00
parent a3b3fcbd7d
commit cc7559f26a
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'])