openstack/charm-ceph-radosgw
Code Issues Proposed changes
0a9daf723e
charm-ceph-radosgw/unit_tests/test_hooks.py
Rodrigo Barbieri fb2f757494 Add config option for keystone admin roles 
RADOS Gateway supports setting keystone operator and admin
roles. RADOS Gateway requires admin roles for keystone users
to change their user quota. Regular operator/member roles
are not allowed to do so.

The lack of this config option prevents swift users with admin
roles from being able to set their quotas. Therefore, a config
option 'admin-roles' is now added to the charm to map to
'rgw keystone accepted admin roles' RADOS Gateway config.

Please note that this is only effective from Luminous
Ceph Release.

Change-Id: Ic0b9aa39eef9fbc6c43eb4e66ab72d90787c2017
Closes-Bug: #1831577
2019-07-01 17:37:19 -03:00

715 lines
27 KiB
Python
Raw Blame History

# Copyright 2016 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.
from mock import (
patch, call, MagicMock, ANY
)
from test_utils import (
CharmTestCase,
)
from charmhelpers.contrib.openstack.ip import PUBLIC
with patch('charmhelpers.contrib.hardening.harden.harden') as mock_dec:
mock_dec.side_effect = (lambda *dargs, **dkwargs: lambda f:
lambda *args, **kwargs: f(*args, **kwargs))
with patch('charmhelpers.fetch.apt_install'):
with patch('utils.register_configs'):
import hooks as ceph_hooks
TO_PATCH = [
'CONFIGS',
'add_source',
'apt_update',
'apt_install',
'apt_purge',
'config',
'cmp_pkgrevno',
'execd_preinstall',
'log',
'open_port',
'os',
'relation_ids',
'relation_set',
'relation_get',
'related_units',
'status_set',
'subprocess',
'sys',
'generate_ha_relation_data',
'get_relation_ip',
'disable_unused_apache_sites',
'service_reload',
'service_stop',
'service_restart',
'service',
'service_name',
'socket',
'restart_map',
'systemd_based_radosgw',
'request_per_unit_key',
'get_certificate_request',
'process_certificates',
'filter_installed_packages',
'filter_missing_packages',
'ceph_utils',
'multisite_deployment',
]
class CephRadosGWTests(CharmTestCase):
def setUp(self):
super(CephRadosGWTests, self).setUp(ceph_hooks, TO_PATCH)
self.config.side_effect = self.test_config.get
self.test_config.set('source', 'distro')
self.test_config.set('key', 'secretkey')
self.test_config.set('use-syslog', False)
self.cmp_pkgrevno.return_value = 0
self.service_name.return_value = 'radosgw'
self.request_per_unit_key.return_value = False
self.systemd_based_radosgw.return_value = False
self.filter_installed_packages.side_effect = lambda pkgs: pkgs
self.filter_missing_packages.side_effect = lambda pkgs: pkgs
self.multisite_deployment.return_value = False
def test_upgrade_available(self):
_vers = {
'distro': 'luminous',
'cloud:bionic-rocky': 'mimic',
}
mock_config = MagicMock()
self.test_config.set('source', 'cloud:bionic-rocky')
mock_config.get.side_effect = self.test_config.get
mock_config.previous.return_value = 'distro'
self.config.side_effect = None
self.config.return_value = mock_config
self.ceph_utils.UPGRADE_PATHS = {
'luminous': 'mimic',
}
self.ceph_utils.resolve_ceph_version.side_effect = (
lambda v: _vers.get(v)
)
self.assertTrue(ceph_hooks.upgrade_available())
@patch.object(ceph_hooks, 'upgrade_available')
def test_install_packages(self, upgrade_available):
mock_config = MagicMock()
mock_config.get.side_effect = self.test_config.get
mock_config.changed.return_value = True
self.config.side_effect = None
self.config.return_value = mock_config
upgrade_available.return_value = False
ceph_hooks.install_packages()
self.add_source.assert_called_with('distro', 'secretkey')
self.apt_update.assert_called_with(fatal=True)
self.apt_purge.assert_called_with(ceph_hooks.APACHE_PACKAGES)
self.apt_install.assert_called_with(ceph_hooks.PACKAGES,
fatal=True)
mock_config.changed.assert_called_with('source')
self.filter_installed_packages.assert_called_with(
ceph_hooks.PACKAGES
)
self.filter_missing_packages.assert_called_with(
ceph_hooks.APACHE_PACKAGES
)
@patch.object(ceph_hooks, 'upgrade_available')
def test_install_packages_upgrades(self, upgrade_available):
mock_config = MagicMock()
mock_config.get.side_effect = self.test_config.get
mock_config.changed.return_value = True
self.config.side_effect = None
self.config.return_value = mock_config
upgrade_available.return_value = True
ceph_hooks.install_packages()
self.add_source.assert_called_with('distro', 'secretkey')
self.apt_update.assert_called_with(fatal=True)
self.apt_purge.assert_called_with(ceph_hooks.APACHE_PACKAGES)
self.apt_install.assert_called_with(ceph_hooks.PACKAGES,
fatal=True)
mock_config.changed.assert_called_with('source')
self.filter_installed_packages.assert_not_called()
self.filter_missing_packages.assert_called_with(
ceph_hooks.APACHE_PACKAGES
)
def test_install(self):
_install_packages = self.patch('install_packages')
ceph_hooks.install()
self.assertTrue(self.execd_preinstall.called)
self.assertTrue(_install_packages.called)
@patch.object(ceph_hooks, 'certs_joined')
@patch.object(ceph_hooks, 'update_nrpe_config')
def test_config_changed(self, update_nrpe_config, mock_certs_joined):
_install_packages = self.patch('install_packages')
_relations = {
'certificates': ['certificates:1']
}
self.relation_ids.side_effect = lambda name: _relations.get(name, [])
ceph_hooks.config_changed()
self.assertTrue(_install_packages.called)
self.CONFIGS.write_all.assert_called_with()
update_nrpe_config.assert_called_with()
mock_certs_joined.assert_called_once_with('certificates:1')
@patch.object(ceph_hooks, 'is_request_complete',
lambda *args, **kwargs: True)
def test_mon_relation(self):
_ceph = self.patch('ceph')
_ceph.import_radosgw_key.return_value = True
self.relation_get.return_value = 'seckey'
self.socket.gethostname.return_value = 'testinghostname'
ceph_hooks.mon_relation()
self.relation_set.assert_not_called()
self.service_restart.assert_called_once_with('radosgw')
self.service.assert_called_once_with('enable', 'radosgw')
_ceph.import_radosgw_key.assert_called_with('seckey',
name='rgw.testinghostname')
self.CONFIGS.write_all.assert_called_with()
@patch.object(ceph_hooks, 'is_request_complete',
lambda *args, **kwargs: True)
def test_mon_relation_request_key(self):
_ceph = self.patch('ceph')
_ceph.import_radosgw_key.return_value = True
self.relation_get.return_value = 'seckey'
self.socket.gethostname.return_value = 'testinghostname'
self.request_per_unit_key.return_value = True
ceph_hooks.mon_relation()
self.relation_set.assert_called_with(
relation_id=None,
key_name='rgw.testinghostname'
)
self.service_restart.assert_called_once_with('radosgw')
self.service.assert_called_once_with('enable', 'radosgw')
_ceph.import_radosgw_key.assert_called_with('seckey',
name='rgw.testinghostname')
self.CONFIGS.write_all.assert_called_with()
@patch.object(ceph_hooks, 'is_request_complete',
lambda *args, **kwargs: True)
def test_mon_relation_nokey(self):
_ceph = self.patch('ceph')
_ceph.import_radosgw_key.return_value = False
self.relation_get.return_value = None
ceph_hooks.mon_relation()
self.assertFalse(_ceph.import_radosgw_key.called)
self.service_restart.assert_not_called()
self.service.assert_not_called()
self.CONFIGS.write_all.assert_called_with()
@patch.object(ceph_hooks, 'send_request_if_needed')
@patch.object(ceph_hooks, 'is_request_complete',
lambda *args, **kwargs: False)
def test_mon_relation_send_broker_request(self,
mock_send_request_if_needed):
_ceph = self.patch('ceph')
_ceph.import_radosgw_key.return_value = False
self.relation_get.return_value = 'seckey'
ceph_hooks.mon_relation()
self.service_restart.assert_not_called()
self.service.assert_not_called()
self.assertFalse(_ceph.import_radosgw_key.called)
self.assertFalse(self.CONFIGS.called)
self.assertTrue(mock_send_request_if_needed.called)
def test_gateway_relation(self):
self.get_relation_ip.return_value = '10.0.0.1'
ceph_hooks.gateway_relation()
self.relation_set.assert_called_with(hostname='10.0.0.1', port=80)
@patch('charmhelpers.contrib.openstack.ip.service_name',
lambda *args: 'ceph-radosgw')
@patch('charmhelpers.contrib.openstack.ip.config')
def test_identity_joined_early_version(self, _config):
self.cmp_pkgrevno.return_value = -1
ceph_hooks.identity_joined()
self.sys.exit.assert_called_with(1)
@patch('charmhelpers.contrib.openstack.ip.service_name',
lambda *args: 'ceph-radosgw')
@patch('charmhelpers.contrib.openstack.ip.resolve_address')
@patch('charmhelpers.contrib.openstack.ip.config')
def test_identity_joined(self, _config, _resolve_address):
def _test_identify_joined(expected):
self.related_units = ['unit/0']
self.cmp_pkgrevno.return_value = 1
_resolve_address.return_value = 'myserv'
_config.side_effect = self.test_config.get
self.test_config.set('region', 'region1')
ceph_hooks.identity_joined(relid='rid')
self.relation_set.assert_called_with(
service='swift',
region='region1',
public_url='http://myserv:80/swift/v1',
internal_url='http://myserv:80/swift/v1',
requested_roles=expected,
relation_id='rid',
admin_url='http://myserv:80/swift')
inputs = [{'operator': 'foo', 'admin': 'bar', 'expected': 'foo,bar'},
{'operator': 'foo', 'expected': 'foo'},
{'admin': 'bar', 'expected': 'bar'},
{'expected': ''}]
for input in inputs:
self.test_config.set('operator-roles', input.get('operator', ''))
self.test_config.set('admin-roles', input.get('admin', ''))
_test_identify_joined(input['expected'])
@patch('charmhelpers.contrib.openstack.ip.service_name',
lambda *args: 'ceph-radosgw')
@patch('charmhelpers.contrib.openstack.ip.is_clustered')
@patch('charmhelpers.contrib.openstack.ip.unit_get')
@patch('charmhelpers.contrib.openstack.ip.config')
def test_identity_joined_public_name(self, _config, _unit_get,
_is_clustered):
self.related_units = ['unit/0']
_config.side_effect = self.test_config.get
self.test_config.set('os-public-hostname', 'files.example.com')
_unit_get.return_value = 'myserv'
_is_clustered.return_value = False
ceph_hooks.identity_joined(relid='rid')
self.relation_set.assert_called_with(
service='swift',
region='RegionOne',
public_url='http://files.example.com:80/swift/v1',
internal_url='http://myserv:80/swift/v1',
requested_roles='Member,Admin',
relation_id='rid',
admin_url='http://myserv:80/swift')
@patch.object(ceph_hooks, 'identity_joined')
def test_identity_changed(self, mock_identity_joined):
ceph_hooks.identity_changed()
self.CONFIGS.write_all.assert_called_with()
self.assertTrue(mock_identity_joined.called)
@patch('charmhelpers.contrib.openstack.ip.is_clustered')
@patch('charmhelpers.contrib.openstack.ip.unit_get')
@patch('charmhelpers.contrib.openstack.ip.config')
def test_canonical_url_ipv6(self, _config, _unit_get, _is_clustered):
ipv6_addr = '2001:db8:85a3:8d3:1319:8a2e:370:7348'
_config.side_effect = self.test_config.get
_unit_get.return_value = ipv6_addr
_is_clustered.return_value = False
self.assertEqual(ceph_hooks.canonical_url({}, PUBLIC),
'http://[%s]' % ipv6_addr)
def test_cluster_joined(self):
self.get_relation_ip.side_effect = ['10.0.0.1',
'10.0.1.1',
'10.0.2.1',
'10.0.3.1']
self.test_config.set('os-public-network', '10.0.0.0/24')
self.test_config.set('os-admin-network', '10.0.1.0/24')
self.test_config.set('os-internal-network', '10.0.2.0/24')
ceph_hooks.cluster_joined()
self.relation_set.assert_has_calls(
[call(relation_id=None,
relation_settings={
'admin-address': '10.0.0.1',
'public-address': '10.0.2.1',
'internal-address': '10.0.1.1',
'private-address': '10.0.3.1'})])
@patch.object(ceph_hooks, 'certs_changed')
def test_cluster_changed(self, mock_certs_changed):
_id_joined = self.patch('identity_joined')
_relations = {
'identity-service': ['rid'],
'certificates': ['certificates:1'],
}
self.relation_ids.side_effect = lambda name: _relations.get(name)
self.related_units.return_value = ['vault/0', 'vault/1']
ceph_hooks.cluster_changed()
self.CONFIGS.write_all.assert_called_with()
_id_joined.assert_called_with(relid='rid')
mock_certs_changed.assert_has_calls([
call('certificates:1', 'vault/0'),
call('certificates:1', 'vault/1')
])
def test_ha_relation_joined(self):
self.generate_ha_relation_data.return_value = {
'test': 'data'
}
ceph_hooks.ha_relation_joined(relation_id='ha:1')
self.relation_set.assert_called_with(
relation_id='ha:1',
test='data'
)
def test_ha_relation_changed(self):
_id_joined = self.patch('identity_joined')
self.relation_get.return_value = True
self.relation_ids.return_value = ['rid']
ceph_hooks.ha_relation_changed()
_id_joined.assert_called_with(relid='rid')
def test_certs_joined(self):
self.get_certificate_request.return_value = {'foo': 'baa'}
ceph_hooks.certs_joined('certificates:1')
self.relation_set.assert_called_once_with(
relation_id='certificates:1',
relation_settings={'foo': 'baa'}
)
self.get_certificate_request.assert_called_once_with()
@patch.object(ceph_hooks, 'configure_https')
def test_certs_changed(self, mock_configure_https):
ceph_hooks.certs_changed('certificates:1', 'vault/0')
self.process_certificates.assert_called_once_with(
'ceph-radosgw',
'certificates:1',
'vault/0'
)
mock_configure_https.assert_called_once_with()
class MiscMultisiteTests(CharmTestCase):
TO_PATCH = [
'restart_nonce_changed',
'relation_ids',
'related_units',
'leader_get',
'is_leader',
'master_relation_joined',
'slave_relation_changed',
'service_restart',
'service_name',
]
_relation_ids = {
'master': ['master:1'],
'slave': ['slave:1'],
}
_related_units = {
'master:1': ['rgw/0', 'rgw/1'],
'slave:1': ['rgw-s/0', 'rgw-s/1'],
}
def setUp(self):
super(MiscMultisiteTests, self).setUp(ceph_hooks,
self.TO_PATCH)
self.relation_ids.side_effect = (
lambda endpoint: self._relation_ids.get(endpoint) or []
)
self.related_units.side_effect = (
lambda rid: self._related_units.get(rid) or []
)
self.service_name.return_value = 'rgw@hostname'
def test_leader_settings_changed(self):
self.restart_nonce_changed.return_value = True
self.is_leader.return_value = False
ceph_hooks.leader_settings_changed()
self.service_restart.assert_called_once_with('rgw@hostname')
self.master_relation_joined.assert_called_once_with('master:1')
def test_process_multisite_relations(self):
ceph_hooks.process_multisite_relations()
self.master_relation_joined.assert_called_once_with('master:1')
self.slave_relation_changed.assert_has_calls([
call('slave:1', 'rgw-s/0'),
call('slave:1', 'rgw-s/1'),
])
class CephRadosMultisiteTests(CharmTestCase):
TO_PATCH = [
'ready_for_service',
'canonical_url',
'relation_set',
'relation_get',
'leader_get',
'config',
'is_leader',
'multisite',
'leader_set',
'service_restart',
'service_name',
'log',
'multisite_deployment',
'systemd_based_radosgw',
]
def setUp(self):
super(CephRadosMultisiteTests, self).setUp(ceph_hooks,
self.TO_PATCH)
self.config.side_effect = self.test_config.get
self.ready_for_service.return_value = True
self.canonical_url.return_value = 'http://rgw'
self.service_name.return_value = 'rgw@hostname'
self.multisite_deployment.return_value = True
self.systemd_based_radosgw.return_value = True
class MasterMultisiteTests(CephRadosMultisiteTests):
_complete_config = {
'realm': 'testrealm',
'zonegroup': 'testzonegroup',
'zone': 'testzone',
}
_leader_data = {
'access_key': 'mykey',
'secret': 'mysecret',
}
_leader_data_done = {
'access_key': 'mykey',
'secret': 'mysecret',
'restart_nonce': 'foobar',
}
def test_master_relation_joined_missing_config(self):
ceph_hooks.master_relation_joined('master:1')
self.config.assert_has_calls([
call('realm'),
call('zonegroup'),
call('zone'),
])
self.relation_set.assert_not_called()
def test_master_relation_joined_create_everything(self):
for k, v in self._complete_config.items():
self.test_config.set(k, v)
self.is_leader.return_value = True
self.leader_get.side_effect = lambda attr: self._leader_data.get(attr)
self.multisite.list_realms.return_value = []
self.multisite.list_zonegroups.return_value = []
self.multisite.list_zones.return_value = []
self.multisite.list_users.return_value = []
self.multisite.create_system_user.return_value = (
'mykey', 'mysecret',
)
ceph_hooks.master_relation_joined('master:1')
self.config.assert_has_calls([
call('realm'),
call('zonegroup'),
call('zone'),
])
self.multisite.create_realm.assert_called_once_with(
'testrealm',
default=True,
)
self.multisite.create_zonegroup.assert_called_once_with(
'testzonegroup',
endpoints=['http://rgw:80'],
default=True,
master=True,
realm='testrealm',
)
self.multisite.create_zone.assert_called_once_with(
'testzone',
endpoints=['http://rgw:80'],
default=True,
master=True,
zonegroup='testzonegroup',
)
self.multisite.create_system_user.assert_called_once_with(
ceph_hooks.MULTISITE_SYSTEM_USER
)
self.multisite.modify_zone.assert_called_once_with(
'testzone',
access_key='mykey',
secret='mysecret',
)
self.multisite.update_period.assert_has_calls([
call(fatal=False),
call(),
])
self.service_restart.assert_called_once_with('rgw@hostname')
self.leader_set.assert_has_calls([
call(access_key='mykey',
secret='mysecret'),
call(restart_nonce=ANY),
])
self.relation_set.assert_called_with(
relation_id='master:1',
access_key='mykey',
secret='mysecret',
)
def test_master_relation_joined_create_nothing(self):
for k, v in self._complete_config.items():
self.test_config.set(k, v)
self.is_leader.return_value = True
self.leader_get.side_effect = (
lambda attr: self._leader_data_done.get(attr)
)
self.multisite.list_realms.return_value = ['testrealm']
self.multisite.list_zonegroups.return_value = ['testzonegroup']
self.multisite.list_zones.return_value = ['testzone']
self.multisite.list_users.return_value = [
ceph_hooks.MULTISITE_SYSTEM_USER
]
ceph_hooks.master_relation_joined('master:1')
self.multisite.create_realm.assert_not_called()
self.multisite.create_zonegroup.assert_not_called()
self.multisite.create_zone.assert_not_called()
self.multisite.create_system_user.assert_not_called()
self.multisite.update_period.assert_not_called()
self.service_restart.assert_not_called()
self.leader_set.assert_not_called()
def test_master_relation_joined_not_leader(self):
for k, v in self._complete_config.items():
self.test_config.set(k, v)
self.is_leader.return_value = False
self.leader_get.side_effect = lambda attr: self._leader_data.get(attr)
ceph_hooks.master_relation_joined('master:1')
self.relation_set.assert_called_once_with(
relation_id='master:1',
realm='testrealm',
zonegroup='testzonegroup',
url='http://rgw:80',
access_key='mykey',
secret='mysecret',
)
self.multisite.list_realms.assert_not_called()
class SlaveMultisiteTests(CephRadosMultisiteTests):
_complete_config = {
'realm': 'testrealm',
'zonegroup': 'testzonegroup',
'zone': 'testzone2',
}
_test_relation = {
'realm': 'testrealm',
'zonegroup': 'testzonegroup',
'access_key': 'anotherkey',
'secret': 'anothersecret',
'url': 'http://master:80'
}
_test_bad_relation = {
'realm': 'anotherrealm',
'zonegroup': 'anotherzg',
'access_key': 'anotherkey',
'secret': 'anothersecret',
'url': 'http://master:80'
}
def test_slave_relation_changed(self):
for k, v in self._complete_config.items():
self.test_config.set(k, v)
self.is_leader.return_value = True
self.leader_get.return_value = None
self.relation_get.return_value = self._test_relation
self.multisite.list_realms.return_value = []
self.multisite.list_zones.return_value = []
ceph_hooks.slave_relation_changed('slave:1', 'rgw/0')
self.config.assert_has_calls([
call('realm'),
call('zonegroup'),
call('zone'),
])
self.multisite.pull_realm.assert_called_once_with(
url=self._test_relation['url'],
access_key=self._test_relation['access_key'],
secret=self._test_relation['secret'],
)
self.multisite.pull_period.assert_called_once_with(
url=self._test_relation['url'],
access_key=self._test_relation['access_key'],
secret=self._test_relation['secret'],
)
self.multisite.set_default_realm.assert_called_once_with(
'testrealm'
)
self.multisite.create_zone.assert_called_once_with(
'testzone2',
endpoints=['http://rgw:80'],
default=False,
master=False,
zonegroup='testzonegroup',
access_key=self._test_relation['access_key'],
secret=self._test_relation['secret'],
)
self.multisite.update_period.assert_has_calls([
call(fatal=False),
call(),
])
self.service_restart.assert_called_once()
self.leader_set.assert_called_once_with(restart_nonce=ANY)
def test_slave_relation_changed_incomplete_relation(self):
for k, v in self._complete_config.items():
self.test_config.set(k, v)
self.is_leader.return_value = True
self.relation_get.return_value = {}
ceph_hooks.slave_relation_changed('slave:1', 'rgw/0')
self.config.assert_not_called()
def test_slave_relation_changed_mismatching_config(self):
for k, v in self._complete_config.items():
self.test_config.set(k, v)
self.is_leader.return_value = True
self.relation_get.return_value = self._test_bad_relation
ceph_hooks.slave_relation_changed('slave:1', 'rgw/0')
self.config.assert_has_calls([
call('realm'),
call('zonegroup'),
call('zone'),
])
self.multisite.list_realms.assert_not_called()
def test_slave_relation_changed_not_leader(self):
self.is_leader.return_value = False
ceph_hooks.slave_relation_changed('slave:1', 'rgw/0')
self.relation_get.assert_not_called()
@patch.object(ceph_hooks, 'apt_install')
@patch.object(ceph_hooks, 'services')
@patch.object(ceph_hooks, 'nrpe')
def test_update_nrpe_config(self, nrpe, services, apt_install):
# Setup Mocks
nrpe.get_nagios_hostname.return_value = 'foo'
nrpe.get_nagios_unit_name.return_value = 'bar'
nrpe_setup = MagicMock()
nrpe.NRPE.return_value = nrpe_setup
services.return_value = ['baz', 'qux']
# Call the routine
ceph_hooks.update_nrpe_config()
# Verify calls
apt_install.assert_called()
nrpe.get_nagios_hostname.assert_called()
nrpe.get_nagios_unit_name.assert_called()
nrpe.copy_nrpe_checks.assert_called()
nrpe.remove_check.assert_not_called()
nrpe.add_init_service_checks.assert_called_with(nrpe_setup,
['baz', 'qux'], 'bar')
nrpe.add_haproxy_checks.assert_called_with(nrpe_setup, 'bar')
nrpe_setup.write.assert_called()
# Verify that remove_check is called appropriately if we pass
# checks_to_remove
ceph_hooks.update_nrpe_config(checks_to_remove=['quux', 'quuux'])
nrpe_setup.remove_check.assert_has_calls([call(shortname='quux'),
call(shortname='quuux')])
View Git Blame Copy Permalink