# 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
        )

    @patch.object(ceph_hooks, 'leader_set')
    @patch.object(ceph_hooks, 'is_leader')
    def test_install(self, is_leader, leader_set):
        _install_packages = self.patch('install_packages')
        is_leader.return_value = True
        ceph_hooks.install()
        self.assertTrue(self.execd_preinstall.called)
        self.assertTrue(_install_packages.called)
        is_leader.assert_called_once()
        leader_set.assert_called_once_with(namespace_tenants=False)

    @patch.object(ceph_hooks, 'leader_set')
    @patch.object(ceph_hooks, 'is_leader')
    def test_install_without_namespacing(self, is_leader, leader_set):
        _install_packages = self.patch('install_packages')
        is_leader.return_value = True
        self.test_config.set('namespace-tenants', True)
        ceph_hooks.install()
        self.assertTrue(self.execd_preinstall.called)
        self.assertTrue(_install_packages.called)
        is_leader.assert_called_once()
        leader_set.assert_called_once_with(namespace_tenants=True)

    @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.object(ceph_hooks, 'leader_get')
    @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, _leader_get):
        self.cmp_pkgrevno.return_value = -1
        _leader_get.return_value = 'False'
        ceph_hooks.identity_joined()
        self.sys.exit.assert_called_with(1)

    @patch.object(ceph_hooks, 'leader_get')
    @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, _leader_get):

        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')
            _leader_get.return_value = 'False'
            ceph_hooks.identity_joined(relid='rid')
            self.relation_set.assert_has_calls([
                call(swift_service='swift',
                     swift_region='region1',
                     swift_public_url='http://myserv:80/swift/v1',
                     swift_internal_url='http://myserv:80/swift/v1',
                     swift_admin_url='http://myserv:80/swift',
                     requested_roles=expected,
                     relation_id='rid'),
                call(s3_service='s3',
                     s3_region='region1',
                     s3_public_url='http://myserv:80/',
                     s3_internal_url='http://myserv:80/',
                     s3_admin_url='http://myserv:80/',
                     relation_id='rid')
            ])

        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.object(ceph_hooks, 'leader_get')
    @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_namespaced(self, _config,
                                        _resolve_address, _leader_get):
        _leader_get.return_value = True

        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')
            _leader_get.return_value = 'True'
            ceph_hooks.identity_joined(relid='rid')
            self.relation_set.assert_has_calls([
                call(swift_service='swift',
                     swift_region='region1',
                     swift_public_url=(
                         'http://myserv:80/swift/v1/AUTH_$(project_id)s'),
                     swift_internal_url=(
                         'http://myserv:80/swift/v1/AUTH_$(project_id)s'),
                     swift_admin_url='http://myserv:80/swift',
                     requested_roles=expected,
                     relation_id='rid'),
                call(s3_service='s3',
                     s3_region='region1',
                     s3_public_url='http://myserv:80/',
                     s3_internal_url='http://myserv:80/',
                     s3_admin_url='http://myserv:80/',
                     relation_id='rid')
            ])

        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.object(ceph_hooks, 'leader_get')
    @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, _leader_get):
        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
        _leader_get.return_value = 'False'
        ceph_hooks.identity_joined(relid='rid')
        self.relation_set.assert_has_calls([
            call(swift_service='swift',
                 swift_region='RegionOne',
                 swift_public_url='http://files.example.com:80/swift/v1',
                 swift_internal_url='http://myserv:80/swift/v1',
                 swift_admin_url='http://myserv:80/swift',
                 requested_roles='Member,Admin',
                 relation_id='rid'),
            call(s3_service='s3',
                 s3_region='RegionOne',
                 s3_public_url='http://files.example.com:80/',
                 s3_internal_url='http://myserv:80/',
                 s3_admin_url='http://myserv:80/',
                 relation_id='rid')
        ])

    @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')])

# 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
        )

    @patch.object(ceph_hooks, 'leader_set')
    @patch.object(ceph_hooks, 'is_leader')
    def test_install(self, is_leader, leader_set):
        _install_packages = self.patch('install_packages')
        is_leader.return_value = True
        ceph_hooks.install()
        self.assertTrue(self.execd_preinstall.called)
        self.assertTrue(_install_packages.called)
        is_leader.assert_called_once()
        leader_set.assert_called_once_with(namespace_tenants=False)

    @patch.object(ceph_hooks, 'leader_set')
    @patch.object(ceph_hooks, 'is_leader')
    def test_install_without_namespacing(self, is_leader, leader_set):
        _install_packages = self.patch('install_packages')
        is_leader.return_value = True
        self.test_config.set('namespace-tenants', True)
        ceph_hooks.install()
        self.assertTrue(self.execd_preinstall.called)
        self.assertTrue(_install_packages.called)
        is_leader.assert_called_once()
        leader_set.assert_called_once_with(namespace_tenants=True)

    @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.object(ceph_hooks, 'leader_get')
    @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, _leader_get):
        self.cmp_pkgrevno.return_value = -1
        _leader_get.return_value = 'False'
        ceph_hooks.identity_joined()
        self.sys.exit.assert_called_with(1)

    @patch.object(ceph_hooks, 'leader_get')
    @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, _leader_get):

        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')
            _leader_get.return_value = 'False'
            ceph_hooks.identity_joined(relid='rid')
            self.relation_set.assert_has_calls([
                call(swift_service='swift',
                     swift_region='region1',
                     swift_public_url='http://myserv:80/swift/v1',
                     swift_internal_url='http://myserv:80/swift/v1',
                     swift_admin_url='http://myserv:80/swift',
                     requested_roles=expected,
                     relation_id='rid'),
                call(s3_service='s3',
                     s3_region='region1',
                     s3_public_url='http://myserv:80/',
                     s3_internal_url='http://myserv:80/',
                     s3_admin_url='http://myserv:80/',
                     relation_id='rid')
            ])

        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.object(ceph_hooks, 'leader_get')
    @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_namespaced(self, _config,
                                        _resolve_address, _leader_get):
        _leader_get.return_value = True

        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')
            _leader_get.return_value = 'True'
            ceph_hooks.identity_joined(relid='rid')
            self.relation_set.assert_has_calls([
                call(swift_service='swift',
                     swift_region='region1',
                     swift_public_url=(
                         'http://myserv:80/swift/v1/AUTH_$(project_id)s'),
                     swift_internal_url=(
                         'http://myserv:80/swift/v1/AUTH_$(project_id)s'),
                     swift_admin_url='http://myserv:80/swift',
                     requested_roles=expected,
                     relation_id='rid'),
                call(s3_service='s3',
                     s3_region='region1',
                     s3_public_url='http://myserv:80/',
                     s3_internal_url='http://myserv:80/',
                     s3_admin_url='http://myserv:80/',
                     relation_id='rid')
            ])

        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.object(ceph_hooks, 'leader_get')
    @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, _leader_get):
        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
        _leader_get.return_value = 'False'
        ceph_hooks.identity_joined(relid='rid')
        self.relation_set.assert_has_calls([
            call(swift_service='swift',
                 swift_region='RegionOne',
                 swift_public_url='http://files.example.com:80/swift/v1',
                 swift_internal_url='http://myserv:80/swift/v1',
                 swift_admin_url='http://myserv:80/swift',
                 requested_roles='Member,Admin',
                 relation_id='rid'),
            call(s3_service='s3',
                 s3_region='RegionOne',
                 s3_public_url='http://files.example.com:80/',
                 s3_internal_url='http://myserv:80/',
                 s3_admin_url='http://myserv:80/',
                 relation_id='rid')
        ])

    @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')])

# 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. # Setup Mocks # Call the routine # Verify calls # Verify that remove_check is called appropriately if we pass # checks_to_remove