openstack
/
charm-barbican
Code Issues Proposed changes
Browse Source

Refactor charm to work with declarative-changes 

This is a refactor of the charm (no new functionality) to work with the
declarative helpers in charms.openstack.  This means that much of the
boiler plate has disappeared from the charm and into charms.openstack
including the tests.

Change-Id: I4305eb34f314f8f5f1a7d1807508bf1e4dbfb84c
Depends-On: I3c74f60bb4ed7901828902118697f310622c4061
Depends-On: Ic81e65f5a072f67cbd2322e4cfd0eec9a5895823
changes/11/367511/2
Alex Kavanagh 6 years ago
parent
21aa29f8a9
commit
aa47fd5b0e
6 changed files with 226 additions and 508 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 3
      requirements.txt
  2. 11
      src/actions/actions.py
  3. 249
      src/lib/charm/openstack/barbican.py
  4. 80
      src/reactive/barbican_handlers.py
  5. 189
      unit_tests/test_barbican_handlers.py
  6. 202
      unit_tests/test_lib_charm_openstack_barbican.py

3
requirements.txt
Unescape View File

@ -1,2 +1,3 @@
charm-tools
#charm-tools
git+https://github.com/juju/charm-tools#egg=charm-tools
simplejson

11
src/actions/actions.py
Unescape View File

@ -24,10 +24,11 @@ basic.bootstrap_charm_deps()
basic.init_config_states()
import charms.reactive as reactive
import charmhelpers.core.hookenv as hookenv
import charms_openstack.charm
import charm.openstack.barbican as barbican
# import the barbican module to get the charm definitions created.
import charm.openstack.barbican # noqa
def generate_mkek_action(*args):
@ -41,7 +42,8 @@ def generate_mkek_action(*args):
"Can't generate an MKEK in associated HSM because HSM is not "
"available.")
return
barbican.generate_mkek(hsm)
with charms_openstack.charm.provide_charm_instance as barbican_charm:
barbican_charm.generate_mkek(hsm)
def generate_hmac_action(*args):
@ -54,7 +56,8 @@ def generate_hmac_action(*args):
hookenv.action_fail(
"Can't generate an HMAC in associated HSM because HSM is not "
"available.")
barbican.generate_hmac(hsm)
with charms_openstack.charm.provide_charm_instance as barbican_charm:
barbican_charm.generate_hmac(hsm)
# Actions to function mapping, to allow for illegal python action names that

249
src/lib/charm/openstack/barbican.py
Unescape View File

@ -19,10 +19,7 @@ from __future__ import absolute_import
import subprocess
import charmhelpers.contrib.openstack.utils as ch_utils
import charmhelpers.core.hookenv as hookenv
import charmhelpers.core.unitdata as unitdata
import charmhelpers.fetch
import charms_openstack.charm
import charms_openstack.adapters
@ -38,158 +35,80 @@ BARBICAN_WSGI_CONF = '/etc/apache2/conf-available/barbican-api.conf'
OPENSTACK_RELEASE_KEY = 'barbican-charm.openstack-release-version'
###
# Handler functions for events that are interesting to the Barbican charms
# select the default release function
charms_openstack.charm.use_defaults('charm.default-select-release')
def install():
"""Use the singleton from the BarbicanCharm to install the packages on the
unit
"""
unitdata.kv().unset(OPENSTACK_RELEASE_KEY)
BarbicanCharm.singleton.install()
###
# Implementation of the Barbican Charm classes
def setup_endpoint(keystone):
"""When the keystone interface connects, register this unit in the keystone
catalogue.
# Add some properties to the configuration for templates/code to use with the
# charm instance. The config_validator is called when the configuration is
# loaded, and the properties are to add those names to the config object.
:param keystone: instance of KeystoneRequires() class from i/f
"""
charm = BarbicanCharm.singleton
keystone.register_endpoints(charm.service_type,
charm.region,
charm.public_url,
charm.internal_url,
charm.admin_url)
@charms_openstack.adapters.config_property
def validate_keystone_api_version(config):
if config.keystone_api_version not in ['2', '3', 'none']:
raise ValueError(
"Unsupported keystone-api-version ({}). It should be 2 or 3"
.format(config.keystone_api_version))
def render_configs(interfaces_list):
"""Using a list of interfaces, render the configs and, if they have
changes, restart the services on the unit.
@charms_openstack.adapters.config_property
def barbican_api_keystone_pipeline(config):
if config.keystone_api_version == "2":
return 'cors keystone_authtoken context apiapp'
else:
return 'cors keystone_v3_authtoken context apiapp'
:param interfaces_list: [RelationBase] interfaces from reactive
"""
BarbicanCharm.singleton.render_with_interfaces(interfaces_list)
@charms_openstack.adapters.config_property
def barbican_api_pipeline(config):
return {
"2": "cors keystone_authtoken context apiapp",
"3": "cors keystone_v3_authtoken context apiapp",
"none": "cors unauthenticated-context apiapp"
}[config.keystone_api_version]
def generate_mkek(hsm):
"""Ask barbican to generate an MKEK in the backend store using the HSM.
This assumes that an HSM is available, and configured. Uses the charm.
"""
BarbicanCharm.singleton.action_generate_mkek(hsm)
@charms_openstack.adapters.config_property
def barbican_api_keystone_audit_pipeline(config):
if config.keystone_api_version == "2":
return 'keystone_authtoken context audit apiapp'
else:
return 'keystone_v3_authtoken context audit apiapp'
def generate_hmac(hsm):
"""Ask barbican to generate an HMAC in the backend store using the HSM.
This assumes that an HSM is available, and configured. Uses the charm.
"""
BarbicanCharm.singleton.action_generate_hmac(hsm)
# Adapt the barbican-hsm-plugin relation for use in rendering the config
# for Barbican. Note that the HSM relation is optional, so we have a class
# variable 'exists' that we can test in the template to see if we should
# render HSM parameters into the template.
def assess_status():
"""Just call the BarbicanCharm.singleton.assess_status() command to update
status on the unit.
"""
BarbicanCharm.singleton.assess_status()
@charms_openstack.adapters.adapter_property('hsm')
def library_path(hsm):
"""Provide a library_path property to the template if it exists"""
try:
return hsm.relation.plugin_data['library_path']
except:
return ''
def configure_ssl(keystone=None):
"""Use the singleton from the BarbicanCharm to configure ssl
@charms_openstack.adapters.adapter_property('hsm')
def login(hsm):
"""Provide a login property to the template if it exists"""
try:
return hsm.relation.plugin_data['login']
except:
return ''
:param keystone: KeystoneRequires() interface class
"""
BarbicanCharm.singleton.configure_ssl(keystone)
###
# Implementation of the Barbican Charm classes
class BarbicanConfigurationAdapter(
charms_openstack.adapters.APIConfigurationAdapter):
def __init__(self, port_map=None):
super(BarbicanConfigurationAdapter, self).__init__(
service_name='barbican',
port_map=port_map)
if self.keystone_api_version not in ['2', '3', 'none']:
raise ValueError(
"Unsupported keystone-api-version ({}). It should be 2 or 3"
.format(self.keystone_api_version))
@property
def barbican_api_keystone_pipeline(self):
if self.keystone_api_version == "2":
return 'cors keystone_authtoken context apiapp'
else:
return 'cors keystone_v3_authtoken context apiapp'
@property
def barbican_api_pipeline(self):
return {
"2": "cors keystone_authtoken context apiapp",
"3": "cors keystone_v3_authtoken context apiapp",
"none": "cors unauthenticated-context apiapp"
}[self.keystone_api_version]
@property
def barbican_api_keystone_audit_pipeline(self):
if self.keystone_api_version == "2":
return 'keystone_authtoken context audit apiapp'
else:
return 'keystone_v3_authtoken context audit apiapp'
class HSMAdapter(charms_openstack.adapters.OpenStackRelationAdapter):
"""Adapt the barbican-hsm-plugin relation for use in rendering the config
for Barbican. Note that the HSM relation is optional, so we have a class
variable 'exists' that we can test in the template to see if we should
render HSM parameters into the template.
"""
interface_type = 'hsm'
@property
def library_path(self):
"""Provide a library_path property to the template if it exists"""
try:
return self.relation.plugin_data['library_path']
except:
return ''
@property
def login(self):
"""Provide a login property to the template if it exists"""
try:
return self.relation.plugin_data['login']
except:
return ''
@property
def slot_id(self):
"""Provide a slot_id property to the template if it exists"""
try:
return self.relation.plugin_data['slot_id']
except:
return ''
class BarbicanAdapters(charms_openstack.adapters.OpenStackAPIRelationAdapters):
"""
Adapters class for the Barbican charm.
This plumbs in the BarbicanConfigurationAdapter as the ConfigurationAdapter
to provide additional properties.
"""
relation_adapters = {
'hsm': HSMAdapter,
}
def __init__(self, relations):
super(BarbicanAdapters, self).__init__(
relations,
options_instance=BarbicanConfigurationAdapter(
port_map=BarbicanCharm.api_ports))
@charms_openstack.adapters.adapter_property('hsm')
def slot_id(hsm):
"""Provide a slot_id property to the template if it exists"""
try:
return hsm.relation.plugin_data['slot_id']
except:
return ''
class BarbicanCharm(charms_openstack.charm.HAOpenStackCharm):
@ -220,22 +139,34 @@ class BarbicanCharm(charms_openstack.charm.HAOpenStackCharm):
BARBICAN_WSGI_CONF: services,
}
adapters_class = BarbicanAdapters
ha_resources = ['vips', 'haproxy']
def install(self):
"""Customise the installation, configure the source and then call the
parent install() method to install the packages
def get_amqp_credentials(self):
"""Provide the default amqp username and vhost as a tuple.
:returns (username, host): two strings to send to the amqp provider.
"""
# DEBUG - until seed random change lands into xenial cloud archive
# BUG #1599550 - barbican + softhsm2 + libssl1.0.0:
# pkcs11:_generate_random() fails
# WARNING: This charm can't be released into stable until the bug is
# fixed.
charmhelpers.fetch.add_source("ppa:ajkavanagh/barbican")
self.configure_source()
# and do the actual install
super(BarbicanCharm, self).install()
return (self.config['rabbit-user'], self.config['rabbit-vhost'])
def get_database_setup(self):
"""Provide the default database credentials as a list of 3-tuples
returns a structure of:
[
{'database': <database>,
'username': <username>,
'hostname': <hostname of this unit>
'prefix': <the optional prefix for the database>, },
]
:returns [{'database': ...}, ...]: credentials for multiple databases
"""
return [
dict(
database=self.config['database'],
username=self.config['database-user'],
hostname=hookenv.unit_private_ip(), )
]
def action_generate_mkek(self, hsm):
"""Generate an MKEK on a connected HSM. Requires that an HSM is
@ -308,19 +239,3 @@ class BarbicanCharm(charms_openstack.charm.HAOpenStackCharm):
required_relations.append('hsm')
return super(BarbicanCharm, self).states_to_check(
required_relations=required_relations)
# Determine the charm class by the supported release
@charms_openstack.charm.register_os_release_selector
def select_release():
"""Determine the release based on the python-keystonemiddleware that is
installed.
Note that this function caches the release after the first install so that
it doesn't need to keep going and getting it from the package information.
"""
release_version = unitdata.kv().get(OPENSTACK_RELEASE_KEY, None)
if release_version is None:
release_version = ch_utils.os_release('python-keystonemiddleware')
unitdata.kv().set(OPENSTACK_RELEASE_KEY, release_version)
return release_version

80
src/reactive/barbican_handlers.py
Unescape View File

@ -18,46 +18,28 @@ from __future__ import absolute_import
import charms.reactive as reactive
import charmhelpers.core.hookenv as hookenv
# This charm's library contains all of the handler code associated with
# barbican
import charm.openstack.barbican as barbican
# use a synthetic state to ensure that it get it to be installed independent of
# the install hook.
@reactive.when_not('charm.installed')
def install_packages():
barbican.install()
reactive.set_state('charm.installed')
import charms_openstack.charm as charm
@reactive.when('amqp.connected')
def setup_amqp_req(amqp):
"""Use the amqp interface to request access to the amqp broker using our
local configuration.
"""
amqp.request_access(username=hookenv.config('rabbit-user'),
vhost=hookenv.config('rabbit-vhost'))
barbican.assess_status()
@reactive.when('shared-db.connected')
def setup_database(database):
"""On receiving database credentials, configure the database on the
interface.
"""
database.configure(hookenv.config('database'),
hookenv.config('database-user'),
hookenv.unit_private_ip())
barbican.assess_status()
# This charm's library contains all of the handler code associated with
# barbican -- we need to import it to get the definitions for the charm.
import charm.openstack.barbican as barbican # noqa
@reactive.when('identity-service.connected')
def setup_endpoint(keystone):
barbican.setup_endpoint(keystone)
barbican.assess_status()
# Use the charms.openstack defaults for common states and hooks
charm.use_defaults(
'charm.installed',
'amqp.connected',
'shared-db.connected',
'identity-service.connected',
'identity-service.available', # enables SSL support
'config.changed',
'update-status')
# Note that because of the way reactive.when works, (which is to 'find' the
# __code__ segment of the decorated function, it's very, very difficult to add
# other kinds of decorators here. This rules out adding other things into the
# charm args list. It is also CPython dependent.
@reactive.when('shared-db.available')
@reactive.when('identity-service.available')
@reactive.when('amqp.available')
@ -65,25 +47,11 @@ def render_stuff(*args):
"""Render the configuration for Barbican when all the interfaces are
available.
Note that the HSM interface is optional (hence the @when_any) and thus is
only used if it is available.
Note that the HSM interface is optional and thus is only used if it is
available.
"""
# Get the optional hsm relation, if it is available for rendering.
hsm = reactive.RelationBase.from_state('hsm.available')
if hsm is not None:
args = args + (hsm, )
barbican.render_configs(args)
barbican.assess_status()
@reactive.when('config.changed')
def config_changed():
"""When the configuration changes, assess the unit's status to update any
juju state required"""
barbican.assess_status()
@reactive.when('identity-service.available')
def configure_ssl(keystone):
"""Configure SSL access to Barbican if requested"""
barbican.configure_ssl(keystone)
hookenv.log("about to call the render_configs with {}".format(args))
with charm.provide_charm_instance() as barbican_charm:
barbican_charm.render_with_interfaces(
charm.optional_interfaces(args, 'hsm.available'))
barbican_charm.assess_status()

189
unit_tests/test_barbican_handlers.py
Unescape View File

@ -15,164 +15,53 @@
from __future__ import absolute_import
from __future__ import print_function
import unittest
import mock
import reactive.barbican_handlers as handlers
_when_args = {}
_when_not_args = {}
def mock_hook_factory(d):
def mock_hook(*args, **kwargs):
def inner(f):
# remember what we were passed. Note that we can't actually
# determine the class we're attached to, as the decorator only gets
# the function.
try:
d[f.__name__].append(dict(args=args, kwargs=kwargs))
except KeyError:
d[f.__name__] = [dict(args=args, kwargs=kwargs)]
return f
return inner
return mock_hook
class TestBarbicanHandlers(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls._patched_when = mock.patch('charms.reactive.when',
mock_hook_factory(_when_args))
cls._patched_when_started = cls._patched_when.start()
cls._patched_when_not = mock.patch('charms.reactive.when_not',
mock_hook_factory(_when_not_args))
cls._patched_when_not_started = cls._patched_when_not.start()
# force requires to rerun the mock_hook decorator:
# try except is Python2/Python3 compatibility as Python3 has moved
# reload to importlib.
try:
reload(handlers)
except NameError:
import importlib
importlib.reload(handlers)
@classmethod
def tearDownClass(cls):
cls._patched_when.stop()
cls._patched_when_started = None
cls._patched_when = None
cls._patched_when_not.stop()
cls._patched_when_not_started = None
cls._patched_when_not = None
# and fix any breakage we did to the module
try:
reload(handlers)
except NameError:
import importlib
importlib.reload(handlers)
def setUp(self):
self._patches = {}
self._patches_start = {}
def tearDown(self):
for k, v in self._patches.items():
v.stop()
setattr(self, k, None)
self._patches = None
self._patches_start = None
def patch(self, obj, attr, return_value=None):
mocked = mock.patch.object(obj, attr)
self._patches[attr] = mocked
started = mocked.start()
started.return_value = return_value
self._patches_start[attr] = started
setattr(self, attr, started)
def test_registered_hooks(self):
# test that the hooks actually registered the relation expressions that
# are meaningful for this interface: this is to handle regressions.
# The keys are the function names that the hook attaches to.
when_patterns = {
'setup_amqp_req': ('amqp.connected', ),
'setup_database': ('shared-db.connected', ),
'setup_endpoint': ('identity-service.connected', ),
'render_stuff': ('shared-db.available',
'identity-service.available',
'amqp.available',),
'config_changed': ('config.changed', ),
'configure_ssl': ('identity-service.available', ),
}
when_not_patterns = {
'install_packages': ('charm.installed', ),
import charms_openstack.test_utils as test_utils
class TestRegisteredHooks(test_utils.TestRegisteredHooks):
def test_hooks(self):
defaults = [
'charm.installed',
'amqp.connected',
'shared-db.connected',
'identity-service.connected',
'identity-service.available', # enables SSL support
'config.changed',
'update-status']
hook_set = {
'when': {
'render_stuff': ('shared-db.available',
'identity-service.available',
'amqp.available',),
}
}
# check the when hooks are attached to the expected functions
for t, p in [(_when_args, when_patterns),
(_when_not_args, when_not_patterns)]:
for f, args in t.items():
# check that function is in patterns
# print("f: {}, args: {}".format(f, args))
self.assertTrue(f in p.keys())
# check that the lists are equal
l = [a['args'][0] for a in args]
self.assertEqual(l, sorted(p[f]))
# test that the hooks were registered via the
# reactive.barbican_handlers
self.registered_hooks_test_helper(handlers, hook_set, defaults)
def test_install_packages(self):
self.patch(handlers.barbican, 'install')
self.patch(handlers.reactive, 'set_state')
handlers.install_packages()
self.install.assert_called_once_with()
self.set_state.assert_called_once_with('charm.installed')
def test_setup_amqp_req(self):
amqp = mock.MagicMock()
self.patch(handlers.hookenv, 'config')
reply = {
'rabbit-user': 'user1',
'rabbit-vhost': 'vhost1',
}
self.config.side_effect = lambda x: reply[x]
self.patch(handlers.barbican, 'assess_status')
handlers.setup_amqp_req(amqp)
amqp.request_access.assert_called_once_with(
username='user1', vhost='vhost1')
self.assess_status.assert_called_once_with()
class TestRenderStuff(test_utils.PatchHelper):
def test_database(self):
database = mock.MagicMock()
self.patch(handlers.hookenv, 'config')
reply = {
'database': 'db1',
'database-user': 'dbuser1',
}
self.config.side_effect = lambda x: reply[x]
self.patch(handlers.hookenv, 'unit_private_ip', 'private_ip')
self.patch(handlers.barbican, 'assess_status')
handlers.setup_database(database)
database.configure.assert_called_once_with(
'db1', 'dbuser1', 'private_ip')
self.assess_status.assert_called_once_with()
def test_render_stuff(self):
barbican_charm = mock.MagicMock()
self.patch_object(handlers.charm, 'provide_charm_instance',
new=mock.MagicMock())
self.provide_charm_instance().__enter__.return_value = barbican_charm
self.provide_charm_instance().__exit__.return_value = None
self.patch_object(handlers.charm, 'optional_interfaces')
def test_setup_endpoint(self):
self.patch(handlers.barbican, 'setup_endpoint')
self.patch(handlers.barbican, 'assess_status')
handlers.setup_endpoint('endpoint_object')
self.setup_endpoint.assert_called_once_with('endpoint_object')
self.assess_status.assert_called_once_with()
def _optional_interfaces(args, *interfaces):
self.assertEqual(interfaces, ('hsm.available', ))
return args + ('hsm', )
self.optional_interfaces.side_effect = _optional_interfaces
def test_render_stuff(self):
self.patch(handlers.barbican, 'render_configs')
self.patch(handlers.barbican, 'assess_status')
self.patch(handlers.reactive.RelationBase, 'from_state',
return_value='hsm')
handlers.render_stuff('arg1', 'arg2')
self.render_configs.assert_called_once_with(('arg1', 'arg2', 'hsm'))
self.assess_status.assert_called_once_with()
self.from_state.assert_called_once_with('hsm.available')
barbican_charm.render_with_interfaces.assert_called_once_with(
('arg1', 'arg2', 'hsm'))
barbican_charm.assess_status.assert_called_once_with()

202
unit_tests/test_lib_charm_openstack_barbican.py
Unescape View File

@ -15,151 +15,93 @@
from __future__ import absolute_import
from __future__ import print_function
import unittest
import mock
import charms_openstack.test_utils as test_utils
import charm.openstack.barbican as barbican
class Helper(unittest.TestCase):
class Helper(test_utils.PatchHelper):
def setUp(self):
self._patches = {}
self._patches_start = {}
# patch out the select_release to always return 'mitaka'
self.patch(barbican.unitdata, 'kv')
_getter = mock.MagicMock()
_getter.get.return_value = barbican.BarbicanCharm.release
self.kv.return_value = _getter
def tearDown(self):
for k, v in self._patches.items():
v.stop()
setattr(self, k, None)
self._patches = None
self._patches_start = None
def patch(self, obj, attr, return_value=None, **kwargs):
mocked = mock.patch.object(obj, attr, **kwargs)
self._patches[attr] = mocked
started = mocked.start()
started.return_value = return_value
self._patches_start[attr] = started
setattr(self, attr, started)
class TestOpenStackBarbican(Helper):
def test_install(self):
self.patch(barbican.BarbicanCharm, 'set_config_defined_certs_and_keys')
self.patch(barbican.BarbicanCharm.singleton, 'install')
barbican.install()
self.install.assert_called_once_with()
super().setUp()
self.patch_release(barbican.BarbicanCharm.release)
def test_setup_endpoint(self):
self.patch(barbican.BarbicanCharm, 'set_config_defined_certs_and_keys')
self.patch(barbican.BarbicanCharm, 'service_type',
new_callable=mock.PropertyMock)
self.patch(barbican.BarbicanCharm, 'region',
new_callable=mock.PropertyMock)
self.patch(barbican.BarbicanCharm, 'public_url',
new_callable=mock.PropertyMock)
self.patch(barbican.BarbicanCharm, 'internal_url',
new_callable=mock.PropertyMock)
self.patch(barbican.BarbicanCharm, 'admin_url',
new_callable=mock.PropertyMock)
self.service_type.return_value = 'type1'
self.region.return_value = 'region1'
self.public_url.return_value = 'public_url'
self.internal_url.return_value = 'internal_url'
self.admin_url.return_value = 'admin_url'
keystone = mock.MagicMock()
barbican.setup_endpoint(keystone)
keystone.register_endpoints.assert_called_once_with(
'type1', 'region1', 'public_url', 'internal_url', 'admin_url')
def test_render_configs(self):
self.patch(barbican.BarbicanCharm, 'set_config_defined_certs_and_keys')
self.patch(barbican.BarbicanCharm.singleton, 'render_with_interfaces')
barbican.render_configs('interfaces-list')
self.render_with_interfaces.assert_called_once_with(
'interfaces-list')
class TestCustomProperties(Helper):
def test_validate_keystone_api_version(self):
config = mock.MagicMock()
for v in ['2', '3', 'none']:
config.keystone_api_version = v
barbican.validate_keystone_api_version(config)
# ensure that it fails
with self.assertRaises(ValueError):
config.keystone_api_version = 'fail-me'
barbican.validate_keystone_api_version(config)
class TestBarbicanConfigurationAdapter(Helper):
@mock.patch('charmhelpers.core.hookenv.config')
def test_barbican_configuration_adapter(self, config):
self.patch(
barbican.charms_openstack.adapters.APIConfigurationAdapter,
'get_network_addresses')
reply = {
'keystone-api-version': '2',
}
config.side_effect = lambda: reply
# Make one with no errors, api version 2
a = barbican.BarbicanConfigurationAdapter()
self.assertEqual(a.barbican_api_keystone_pipeline,
def test_barbican_api_keystone_pipeline(self):
config = mock.MagicMock()
config.keystone_api_version = '2'
self.assertEqual(barbican.barbican_api_keystone_pipeline(config),
'cors keystone_authtoken context apiapp')
self.assertEqual(a.barbican_api_pipeline,
'cors keystone_authtoken context apiapp')
# Now test it with api version 3
reply['keystone-api-version'] = '3'
a = barbican.BarbicanConfigurationAdapter()
self.assertEqual(a.barbican_api_keystone_pipeline,
'cors keystone_v3_authtoken context apiapp')
self.assertEqual(a.barbican_api_pipeline,
config.keystone_api_version = ''
self.assertEqual(barbican.barbican_api_keystone_pipeline(config),
'cors keystone_v3_authtoken context apiapp')
# and a 'none' version
reply['keystone-api-version'] = 'none'
a = barbican.BarbicanConfigurationAdapter()
self.assertEqual(a.barbican_api_keystone_pipeline,
def test_barbican_api_pipeline(self):
config = mock.MagicMock()
config.keystone_api_version = '2'
self.assertEqual(barbican.barbican_api_pipeline(config),
'cors keystone_authtoken context apiapp')
config.keystone_api_version = '3'
self.assertEqual(barbican.barbican_api_pipeline(config),
'cors keystone_v3_authtoken context apiapp')
self.assertEqual(a.barbican_api_pipeline,
config.keystone_api_version = 'none'
self.assertEqual(barbican.barbican_api_pipeline(config),
'cors unauthenticated-context apiapp')
# finally, try to create an invalid one.
reply['keystone-api-version'] = None
with self.assertRaises(ValueError):
a = barbican.BarbicanConfigurationAdapter()
def test_barbican_api_keystone_audit_pipeline(self):
config = mock.MagicMock()
config.keystone_api_version = '2'
self.assertEqual(barbican.barbican_api_keystone_audit_pipeline(config),
'keystone_authtoken context audit apiapp')
config.keystone_api_version = ''
self.assertEqual(barbican.barbican_api_keystone_audit_pipeline(config),
'keystone_v3_authtoken context audit apiapp')
class TestBarbicanAdapters(Helper):
@mock.patch('charmhelpers.core.hookenv.config')
def test_barbican_adapters(self, config):
reply = {
'keystone-api-version': '2',
# for the charms.openstack code, which breaks if we don't have:
'os-public-hostname': 'host',
'os-internal-hostname': 'internal',
'os-admin-hostname': 'admin',
class TestHSMProperties(Helper):
def setUp(self):
super().setUp()
self.data_none = {}
self.data_set = {
'library_path': 'a-path',
'login': 'a-login',
'slot_id': 'a-slot_id',
}
def cf(key=None):
if key is not None:
return reply[key]
return reply
def test_library_path(self):
hsm = mock.MagicMock()
hsm.relation.plugin_data = self.data_none
self.assertEqual(barbican.library_path(hsm), '')
hsm.relation.plugin_data = self.data_set
self.assertEqual(barbican.library_path(hsm), 'a-path')
config.side_effect = cf
amqp_relation = mock.MagicMock()
amqp_relation.relation_name = 'amqp'
shared_db_relation = mock.MagicMock()
shared_db_relation.relation_name = 'shared_db'
other_relation = mock.MagicMock()
other_relation.relation_name = 'other'
other_relation.thingy = 'help'
# verify that the class is created with a BarbicanConfigurationAdapter
b = barbican.BarbicanAdapters([amqp_relation,
shared_db_relation,
other_relation])
# ensure that the relevant things got put on.
self.assertTrue(
isinstance(
b.other,
barbican.charms_openstack.adapters.OpenStackRelationAdapter))
self.assertTrue(isinstance(b.options,
barbican.BarbicanConfigurationAdapter))
def test_login(self):
hsm = mock.MagicMock()
hsm.relation.plugin_data = self.data_none
self.assertEqual(barbican.login(hsm), '')
hsm.relation.plugin_data = self.data_set
self.assertEqual(barbican.login(hsm), 'a-login')
def test_slot_id(self):
hsm = mock.MagicMock()
hsm.relation.plugin_data = self.data_none
self.assertEqual(barbican.slot_id(hsm), '')
hsm.relation.plugin_data = self.data_set
self.assertEqual(barbican.slot_id(hsm), 'a-slot_id')
class TestBarbicanCharm(Helper):
@ -171,7 +113,7 @@ class TestBarbicanCharm(Helper):
'login': '1234',
'slot_id': 'slot1'
}
self.patch(barbican.hookenv, 'config')
self.patch_object(barbican.hookenv, 'config')
config = {
'mkek-key-length': 5,
'label-mkek': 'the-label'
@ -183,8 +125,8 @@ class TestBarbicanCharm(Helper):
return config
self.config.side_effect = cf
self.patch(barbican.subprocess, 'check_call')
self.patch(barbican.hookenv, 'log')
self.patch_object(barbican.subprocess, 'check_call')
self.patch_object(barbican.hookenv, 'log')
# try generating a an mkek with no failure
c = barbican.BarbicanCharm()
c.action_generate_mkek(hsm)
@ -218,7 +160,7 @@ class TestBarbicanCharm(Helper):
'login': '1234',
'slot_id': 'slot1'
}
self.patch(barbican.hookenv, 'config')
self.patch_object(barbican.hookenv, 'config')
config = {
'hmac-key-length': 5,
'label-hmac': 'the-label'
@ -230,8 +172,8 @@ class TestBarbicanCharm(Helper):
return config
self.config.side_effect = cf
self.patch(barbican.subprocess, 'check_call')
self.patch(barbican.hookenv, 'log')
self.patch_object(barbican.subprocess, 'check_call')
self.patch_object(barbican.hookenv, 'log')
# try generating a an hmac with no failure
c = barbican.BarbicanCharm()
c.action_generate_hmac(hsm)

Write Preview
Loading…
Cancel
Save