be8889e581
The new methods assertActionRanOK & assertRemoteRunOK are useful in setup as well as tests so it makes sense to move them out of the test class.
396 lines
16 KiB
Python
396 lines
16 KiB
Python
# Copyright 2018 Canonical Ltd.
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
|
# you may not use this file except in compliance with the License.
|
|
# You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
# See the License for the specific language governing permissions and
|
|
# limitations under the License.
|
|
"""Module containing base class for implementing charm tests."""
|
|
import contextlib
|
|
import logging
|
|
import subprocess
|
|
import unittest
|
|
|
|
import zaza.model as model
|
|
import zaza.charm_lifecycle.utils as lifecycle_utils
|
|
import zaza.openstack.utilities.openstack as openstack_utils
|
|
import zaza.openstack.utilities.generic as generic_utils
|
|
|
|
|
|
def skipIfNotHA(service_name):
|
|
"""Run decorator to skip tests if application not in HA configuration."""
|
|
def _skipIfNotHA_inner_1(f):
|
|
def _skipIfNotHA_inner_2(*args, **kwargs):
|
|
ips = model.get_app_ips(
|
|
service_name)
|
|
if len(ips) > 1:
|
|
return f(*args, **kwargs)
|
|
else:
|
|
logging.warn("Skipping HA test for non-ha service {}".format(
|
|
service_name))
|
|
return _skipIfNotHA_inner_2
|
|
|
|
return _skipIfNotHA_inner_1
|
|
|
|
|
|
def skipUntilVersion(service, package, release):
|
|
"""Run decorator to skip this test if application version is too low."""
|
|
def _skipUntilVersion_inner_1(f):
|
|
def _skipUntilVersion_inner_2(*args, **kwargs):
|
|
package_version = generic_utils.get_pkg_version(service, package)
|
|
try:
|
|
subprocess.check_call(['dpkg', '--compare-versions',
|
|
package_version, 'ge', release],
|
|
stderr=subprocess.STDOUT,
|
|
universal_newlines=True)
|
|
return f(*args, **kwargs)
|
|
except subprocess.CalledProcessError as cp:
|
|
logging.warn("Skipping test for older ({})"
|
|
"service {}, requested {}".format(
|
|
package_version, service, release))
|
|
return _skipUntilVersion_inner_2
|
|
return _skipUntilVersion_inner_1
|
|
|
|
|
|
def audit_assertions(action,
|
|
expected_passes,
|
|
expected_failures=None,
|
|
expected_to_pass=True):
|
|
"""Check expected assertion failures in security-checklist actions.
|
|
|
|
:param action: Action object from running the security-checklist action
|
|
:type action: juju.action.Action
|
|
:param expected_passes: List of test names that are expected to pass
|
|
:type expected_passes: List[str]
|
|
:param expected_failures: List of test names that are expected to fail
|
|
:type expected_failures: List[str]
|
|
:raises: AssertionError if the assertion fails.
|
|
"""
|
|
if expected_failures is None:
|
|
expected_failures = []
|
|
if expected_to_pass:
|
|
assert action.data["status"] == "completed", \
|
|
"Security check is expected to pass by default"
|
|
else:
|
|
assert action.data["status"] == "failed", \
|
|
"Security check is not expected to pass by default"
|
|
|
|
results = action.data['results']
|
|
for key, value in results.items():
|
|
if key in expected_failures:
|
|
assert "FAIL" in value, "Unexpected test pass: {}".format(key)
|
|
if key in expected_passes:
|
|
assert value == "PASS", "Unexpected failure: {}".format(key)
|
|
|
|
|
|
class BaseCharmTest(unittest.TestCase):
|
|
"""Generic helpers for testing charms."""
|
|
|
|
run_resource_cleanup = False
|
|
|
|
@classmethod
|
|
def resource_cleanup(cls):
|
|
"""Cleanup any resources created during the test run.
|
|
|
|
Override this method with a method which removes any resources
|
|
which were created during the test run. If the test sets
|
|
"self.run_resource_cleanup = True" then cleanup will be
|
|
performed.
|
|
"""
|
|
pass
|
|
|
|
@classmethod
|
|
def tearDown(cls):
|
|
"""Run teardown for test class."""
|
|
if cls.run_resource_cleanup:
|
|
logging.info('Running resource cleanup')
|
|
cls.resource_cleanup()
|
|
|
|
@classmethod
|
|
def setUpClass(cls, application_name=None, model_alias=None):
|
|
"""Run setup for test class to create common resources."""
|
|
cls.model_aliases = model.get_juju_model_aliases()
|
|
if model_alias:
|
|
cls.model_name = cls.model_aliases[model_alias]
|
|
else:
|
|
cls.model_name = model.get_juju_model()
|
|
cls.test_config = lifecycle_utils.get_charm_config(fatal=False)
|
|
if application_name:
|
|
cls.application_name = application_name
|
|
else:
|
|
cls.application_name = cls.test_config['charm_name']
|
|
cls.lead_unit = model.get_lead_unit_name(
|
|
cls.application_name,
|
|
model_name=cls.model_name)
|
|
logging.debug('Leader unit is {}'.format(cls.lead_unit))
|
|
|
|
def config_current(self, application_name=None, keys=None):
|
|
"""Get Current Config of an application normalized into key-values.
|
|
|
|
:param application_name: String application name for use when called
|
|
by a charm under test other than the object's
|
|
application.
|
|
:type application_name: Optional[str]
|
|
:param keys: iterable of strs to index into the current config. If
|
|
None, return all keys from the config
|
|
:type keys: Optional[Iterable[str]]
|
|
:return: Dictionary of requested config from application
|
|
:rtype: Dict[str, Any]
|
|
"""
|
|
if not application_name:
|
|
application_name = self.application_name
|
|
|
|
_app_config = model.get_application_config(application_name)
|
|
|
|
keys = keys or _app_config.keys()
|
|
return {
|
|
k: _app_config.get(k, {}).get('value')
|
|
for k in keys
|
|
}
|
|
|
|
@staticmethod
|
|
def _stringed_value_config(config):
|
|
"""Stringify values in a dict.
|
|
|
|
Workaround:
|
|
libjuju refuses to accept data with types other than strings
|
|
through the zuzu.model.set_application_config
|
|
|
|
:param config: Config dictionary with any typed values
|
|
:type config: Dict[str,Any]
|
|
:return: Config Dictionary with string-ly typed values
|
|
:rtype: Dict[str,str]
|
|
"""
|
|
# if v is None, stringify to ''
|
|
# otherwise use a strict cast with str(...)
|
|
return {
|
|
k: '' if v is None else str(v)
|
|
for k, v in config.items()
|
|
}
|
|
|
|
@contextlib.contextmanager
|
|
def config_change(self, default_config, alternate_config,
|
|
application_name=None):
|
|
"""Run change config tests.
|
|
|
|
Change config to `alternate_config`, wait for idle workload status,
|
|
yield, return config to `default_config` and wait for idle workload
|
|
status before return from function.
|
|
|
|
Example usage:
|
|
with self.config_change({'preferred-api-version': '2'},
|
|
{'preferred-api-version': '3'}):
|
|
do_something()
|
|
|
|
:param default_config: Dict of charm settings to set on completion
|
|
:type default_config: dict
|
|
:param alternate_config: Dict of charm settings to change to
|
|
:type alternate_config: dict
|
|
:param application_name: String application name for use when called
|
|
by a charm under test other than the object's
|
|
application.
|
|
:type application_name: str
|
|
"""
|
|
if not application_name:
|
|
application_name = self.application_name
|
|
|
|
# we need to compare config values to what is already applied before
|
|
# attempting to set them. otherwise the model will behave differently
|
|
# than we would expect while waiting for completion of the change
|
|
app_config = self.config_current(
|
|
application_name, keys=alternate_config.keys()
|
|
)
|
|
|
|
if all(item in app_config.items()
|
|
for item in alternate_config.items()):
|
|
logging.debug('alternate_config equals what is already applied '
|
|
'config')
|
|
yield
|
|
if default_config == alternate_config:
|
|
logging.debug('default_config also equals what is already '
|
|
'applied config')
|
|
return
|
|
logging.debug('alternate_config already set, and default_config '
|
|
'needs to be applied before return')
|
|
else:
|
|
logging.debug('Changing charm setting to {}'
|
|
.format(alternate_config))
|
|
model.set_application_config(
|
|
application_name,
|
|
self._stringed_value_config(alternate_config),
|
|
model_name=self.model_name)
|
|
|
|
logging.debug(
|
|
'Waiting for units to execute config-changed hook')
|
|
model.wait_for_agent_status(model_name=self.model_name)
|
|
|
|
logging.debug(
|
|
'Waiting for units to reach target states')
|
|
model.wait_for_application_states(
|
|
model_name=self.model_name,
|
|
states=self.test_config.get('target_deploy_status', {}))
|
|
# TODO: Optimize with a block on a specific application until idle.
|
|
model.block_until_all_units_idle()
|
|
|
|
yield
|
|
|
|
logging.debug('Restoring charm setting to {}'.format(default_config))
|
|
model.set_application_config(
|
|
application_name,
|
|
self._stringed_value_config(default_config),
|
|
model_name=self.model_name)
|
|
|
|
logging.debug(
|
|
'Waiting for units to reach target states')
|
|
model.wait_for_application_states(
|
|
model_name=self.model_name,
|
|
states=self.test_config.get('target_deploy_status', {}))
|
|
# TODO: Optimize with a block on a specific application until idle.
|
|
model.block_until_all_units_idle()
|
|
|
|
def restart_on_changed(self, config_file, default_config, alternate_config,
|
|
default_entry, alternate_entry, services,
|
|
pgrep_full=False):
|
|
"""Run restart on change tests.
|
|
|
|
Test that changing config results in config file being updates and
|
|
services restarted. Return config to default_config afterwards
|
|
|
|
:param config_file: Config file to check for settings
|
|
:type config_file: str
|
|
:param default_config: Dict of charm settings to set on completion
|
|
:type default_config: dict
|
|
:param alternate_config: Dict of charm settings to change to
|
|
:type alternate_config: dict
|
|
:param default_entry: Config file entries that correspond to
|
|
default_config
|
|
:type default_entry: dict
|
|
:param alternate_entry: Config file entries that correspond to
|
|
alternate_config
|
|
:type alternate_entry: dict
|
|
:param services: Services expected to be restarted when config_file is
|
|
changed.
|
|
:type services: list
|
|
:param pgrep_full: Should pgrep be used rather than pidof to identify
|
|
a service.
|
|
:type pgrep_full: bool
|
|
"""
|
|
# lead_unit is only useed to grab a timestamp, the assumption being
|
|
# that all the units times are in sync.
|
|
|
|
mtime = model.get_unit_time(
|
|
self.lead_unit,
|
|
model_name=self.model_name)
|
|
logging.debug('Remote unit timestamp {}'.format(mtime))
|
|
|
|
with self.config_change(default_config, alternate_config):
|
|
# If this is not an OSLO config file set default_config={}
|
|
if alternate_entry:
|
|
logging.debug(
|
|
'Waiting for updates to propagate to {}'
|
|
.format(config_file))
|
|
model.block_until_oslo_config_entries_match(
|
|
self.application_name,
|
|
config_file,
|
|
alternate_entry,
|
|
model_name=self.model_name)
|
|
else:
|
|
model.block_until_all_units_idle(model_name=self.model_name)
|
|
|
|
# Config update has occured and hooks are idle. Any services should
|
|
# have been restarted by now:
|
|
logging.debug(
|
|
'Waiting for services ({}) to be restarted'.format(services))
|
|
model.block_until_services_restarted(
|
|
self.application_name,
|
|
mtime,
|
|
services,
|
|
model_name=self.model_name,
|
|
pgrep_full=pgrep_full)
|
|
|
|
# If this is not an OSLO config file set default_config={}
|
|
if default_entry:
|
|
logging.debug(
|
|
'Waiting for updates to propagate to '.format(config_file))
|
|
model.block_until_oslo_config_entries_match(
|
|
self.application_name,
|
|
config_file,
|
|
default_entry,
|
|
model_name=self.model_name)
|
|
else:
|
|
model.block_until_all_units_idle(model_name=self.model_name)
|
|
|
|
@contextlib.contextmanager
|
|
def pause_resume(self, services, pgrep_full=False):
|
|
"""Run Pause and resume tests.
|
|
|
|
Pause and then resume a unit checking that services are in the
|
|
required state after each action
|
|
|
|
:param services: Services expected to be restarted when the unit is
|
|
paused/resumed.
|
|
:type services: list
|
|
:param pgrep_full: Should pgrep be used rather than pidof to identify
|
|
a service.
|
|
:type pgrep_full: bool
|
|
"""
|
|
model.block_until_service_status(
|
|
self.lead_unit,
|
|
services,
|
|
'running',
|
|
model_name=self.model_name,
|
|
pgrep_full=pgrep_full)
|
|
model.block_until_unit_wl_status(
|
|
self.lead_unit,
|
|
'active',
|
|
model_name=self.model_name)
|
|
generic_utils.assertActionRanOK(model.run_action(
|
|
self.lead_unit,
|
|
'pause',
|
|
model_name=self.model_name))
|
|
model.block_until_unit_wl_status(
|
|
self.lead_unit,
|
|
'maintenance',
|
|
model_name=self.model_name)
|
|
model.block_until_all_units_idle(model_name=self.model_name)
|
|
model.block_until_service_status(
|
|
self.lead_unit,
|
|
services,
|
|
'stopped',
|
|
model_name=self.model_name,
|
|
pgrep_full=pgrep_full)
|
|
yield
|
|
generic_utils.assertActionRanOK(model.run_action(
|
|
self.lead_unit,
|
|
'resume',
|
|
model_name=self.model_name))
|
|
model.block_until_unit_wl_status(
|
|
self.lead_unit,
|
|
'active',
|
|
model_name=self.model_name)
|
|
model.block_until_all_units_idle(model_name=self.model_name)
|
|
model.block_until_service_status(
|
|
self.lead_unit,
|
|
services,
|
|
'running',
|
|
model_name=self.model_name,
|
|
pgrep_full=pgrep_full)
|
|
|
|
|
|
class OpenStackBaseTest(BaseCharmTest):
|
|
"""Generic helpers for testing OpenStack API charms."""
|
|
|
|
@classmethod
|
|
def setUpClass(cls, application_name=None, model_alias=None):
|
|
"""Run setup for test class to create common resources."""
|
|
super(OpenStackBaseTest, cls).setUpClass()
|
|
cls.keystone_session = openstack_utils.get_overcloud_keystone_session(
|
|
model_name=cls.model_name)
|
|
cls.cacert = openstack_utils.get_cacert()
|