# 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): 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') self.test_config.set('operator-roles', 'admin') 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='admin', relation_id='rid', admin_url='http://myserv:80/swift') @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()