# 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 __future__ import absolute_import from __future__ import print_function import contextlib import unittest import mock import charm.openstack.designate as designate def FakeConfig(init_dict): def _config(key=None): return init_dict[key] if key else init_dict return _config class Helper(unittest.TestCase): def setUp(self): self._patches = {} self._patches_start = {} self.ch_config_patch = mock.patch('charmhelpers.core.hookenv.config') self.ch_config = self.ch_config_patch.start() self.ch_config.side_effect = lambda: {'ssl_param': None} def tearDown(self): for k, v in self._patches.items(): v.stop() setattr(self, k, None) self._patches = None self._patches_start = None self.ch_config_patch.stop() def patch(self, obj, attr, return_value=None, **kwargs): mocked = mock.patch.object(obj, attr, **kwargs) self._patches[attr] = mocked started = mocked.start() started.return_value = return_value self._patches_start[attr] = started setattr(self, attr, started) def patch_object(self, obj, attr, return_value=None, name=None, new=None): if name is None: name = attr if new is not None: mocked = mock.patch.object(obj, attr, new=new) else: mocked = mock.patch.object(obj, attr) self._patches[name] = mocked started = mocked.start() if new is None: started.return_value = return_value self._patches_start[name] = started setattr(self, name, started) class TestDesignateDBAdapter(Helper): def fake_get_uri(self, prefix): return 'mysql://uri/{}-database'.format(prefix) def test_designate_uri(self): relation = mock.MagicMock() a = designate.DesignateDBAdapter(relation) self.patch(designate.DesignateDBAdapter, 'get_uri') self.get_uri.side_effect = self.fake_get_uri self.assertEqual(a.designate_uri, 'mysql://uri/designate-database') self.assertEqual(a.designate_pool_uri, 'mysql://uri/dpm-database') class TestBindRNDCRelationAdapter(Helper): def test_slave_ips(self): relation = mock.MagicMock() relation.slave_ips.return_value = 'slave_ips_info' a = designate.BindRNDCRelationAdapter(relation) self.assertEqual(a.slave_ips, 'slave_ips_info') def test_pool_configs(self): relation = mock.MagicMock() _slave_ips = [ {'unit': 'unit/1', 'address': 'addr1'}, {'unit': 'unit/2', 'address': 'addr2'}] with mock.patch.object(designate.BindRNDCRelationAdapter, 'slave_ips', new=_slave_ips): a = designate.BindRNDCRelationAdapter(relation) expect = [{'address': 'addr1', 'nameserver': 'nameserver_unit_1', 'pool_target': 'nameserver_unit_1'}, {'address': 'addr2', 'nameserver': 'nameserver_unit_2', 'pool_target': 'nameserver_unit_2'}] self.assertEqual(a.pool_config, expect) self.assertEqual( a.pool_targets, 'nameserver_unit_1, nameserver_unit_2') self.assertEqual(a.slave_addresses, 'addr1:53, addr2:53') def test_rndc_info(self): relation = mock.MagicMock() relation.rndc_info = 'rndcstuff' a = designate.BindRNDCRelationAdapter(relation) self.assertEqual(a.rndc_info, 'rndcstuff') class TestDesignateConfigurationAdapter(Helper): def test_designate_configuration_adapter_pool_info(self): relation = mock.MagicMock() self.patch( designate.openstack_adapters.APIConfigurationAdapter, 'get_network_addresses') test_config = { 'dns_slaves': 'ip1:port1:key1 ip2:port2:key2', } with mock.patch.object(designate.openstack_adapters.hookenv, 'config', new=lambda: test_config): a = designate.DesignateConfigurationAdapter(relation) expect = [{'address': 'ip1', 'nameserver': 'nameserver_ip1', 'pool_target': 'nameserver_ip1', 'rndc_key_file': '/etc/designate/rndc_ip1.key'}, {'address': 'ip2', 'nameserver': 'nameserver_ip2', 'pool_target': 'nameserver_ip2', 'rndc_key_file': '/etc/designate/rndc_ip2.key'}] self.assertEqual(a.pool_config, expect) self.assertEqual(a.pool_targets, 'nameserver_ip1, nameserver_ip2') self.assertEqual(a.slave_addresses, 'ip1:53, ip2:53') def test_designate_configuration_domains(self): relation = mock.MagicMock() self.patch( designate.openstack_adapters.APIConfigurationAdapter, 'get_network_addresses') test_config = { 'nova-domain': 'bob.com', 'neutron-domain': 'bill.com', } domain_map = { 'bob.com': 12, 'bill.com': 13, } with mock.patch.object(designate.hookenv, 'config', side_effect=FakeConfig(test_config)): self.patch(designate.DesignateCharm, 'get_domain_id') self.get_domain_id.side_effect = lambda x: domain_map.get(x) a = designate.DesignateConfigurationAdapter(relation) self.assertEqual(a.nova_domain_id, 12) self.assertEqual(a.neutron_domain_id, 13) def test_designate_configuration_daemon_args(self): relation = mock.MagicMock() self.patch( designate.openstack_adapters.APIConfigurationAdapter, 'get_network_addresses') self.patch(designate.os.path, 'exists', return_value=True) a = designate.DesignateConfigurationAdapter(relation) self.assertEqual( a.nova_conf_args, '--config-file=/etc/designate/conf.d/nova_sink.cfg') self.assertEqual( a.neutron_conf_args, '--config-file=/etc/designate/conf.d/neutron_sink.cfg') self.patch(designate.os.path, 'exists', return_value=False) self.assertEqual(a.nova_conf_args, '') self.assertEqual(a.neutron_conf_args, '') def test_rndc_master_ip(self): relation = mock.MagicMock() self.patch( designate.openstack_adapters.APIConfigurationAdapter, 'get_network_addresses') self.patch(designate.os_ip, 'resolve_address', return_value='intip') a = designate.DesignateConfigurationAdapter(relation) self.assertEqual(a.rndc_master_ip, 'intip') def test_also_notifies_hosts(self): relation = mock.MagicMock test_config = { 'also-notifies': '10.0.0.1:53 10.0.0.2:10053', } with mock.patch.object(designate.hookenv, 'config', side_effect=FakeConfig(test_config)): expect = [{'address': '10.0.0.1', 'port': '53'}, {'address': '10.0.0.2', 'port': '10053'}] a = designate.DesignateConfigurationAdapter(relation) self.assertEqual(a.also_notifies_hosts, expect) class TestDesignateCharm(Helper): def test_install(self): self.patch(designate.DesignateCharm, 'configure_source') self.patch(designate.DesignateCharm, 'update_api_ports') self.ch_config.side_effect = lambda: {'openstack-origin': 'distro'} a = designate.DesignateCharm(release='mitaka') a.install() self.configure_source.assert_called_with() def test_render_base_config(self): self.patch(designate.DesignateCharm, 'haproxy_enabled') self.patch( designate.DesignateCharm, 'render_with_interfaces') self.patch_object(designate.DesignateCharm, 'haproxy_enabled', new=lambda x: True) a = designate.DesignateCharm(release='mitaka') a.render_base_config('interface_list') expect_configs = [ '/root/novarc', '/etc/designate/designate.conf', '/etc/designate/rndc.key', '/etc/default/openstack', '/etc/haproxy/haproxy.cfg'] self.render_with_interfaces.assert_called_with( 'interface_list', configs=expect_configs) def test_render_full_config(self): self.patch( designate.DesignateCharm, 'render_with_interfaces') a = designate.DesignateCharm(release='mitaka') a.render_full_config('interface_list') self.render_with_interfaces.assert_called_with('interface_list') def test_write_key_file(self): self.patch(designate.host, 'write_file') a = designate.DesignateCharm(release='mitaka') a.write_key_file('unit1', 'keydigest') self.write_file.assert_called_with( '/etc/designate/rndc_unit1.key', mock.ANY, owner='root', group='designate', perms=0o440) def test_render_rndc_keys(self): test_config = { 'dns-slaves': '10.0.0.10:port1:key1 192.168.23.4:port2:key2', } self.patch(designate.DesignateCharm, 'write_key_file') with mock.patch.object(designate.hookenv, 'config', side_effect=FakeConfig(test_config)): a = designate.DesignateCharm(release='mitaka') a.render_rndc_keys() calls = [ mock.call('10_0_0_10', 'key1'), mock.call('192_168_23_4', 'key2'), ] self.write_key_file.assert_has_calls(calls) def test_get_domain_id(self): self.patch(designate.DesignateCharm, 'ensure_api_responding') self.patch(designate.subprocess, 'check_output') self.check_output.return_value = b'hi\n' self.assertEqual(designate.DesignateCharm.get_domain_id('domain'), 'hi') self.check_output.assert_called_with( ['reactive/designate_utils.py', 'domain-get', '--domain-name', 'domain']) def test_create_domain(self): self.patch(designate.DesignateCharm, 'ensure_api_responding') self.patch(designate.subprocess, 'check_call') designate.DesignateCharm.create_domain('domain', 'email') self.check_call.assert_called_with( ['reactive/designate_utils.py', 'domain-create', '--domain-name', 'domain', '--email', 'email']) def test_create_server(self): self.patch(designate.subprocess, 'check_call') self.patch(designate.DesignateCharm, 'ensure_api_responding') designate.DesignateCharm.create_server('nameservername') self.check_call.assert_called_with( ['reactive/designate_utils.py', 'server-create', '--server-name', 'nameservername']) def test_domain_init_done(self): self.patch(designate.hookenv, 'leader_get') self.leader_get.return_value = True a = designate.DesignateCharm(release='mitaka') self.assertTrue(a.domain_init_done()) self.leader_get.return_value = False a = designate.DesignateCharm(release='mitaka') self.assertFalse(a.domain_init_done()) def test_create_initial_servers_and_domains(self): test_config = { 'nameservers': 'dnsserverrec1. dnsserverrec2', 'nova-domain': 'novadomain', 'nova-domain-email': 'novaemail', 'neutron-domain': 'neutrondomain', 'neutron-domain-email': 'neutronemail', } self.patch(designate.DesignateCharm, 'ensure_api_responding') self.ensure_api_responding.return_value = True self.patch(designate.hookenv, 'is_leader', return_value=True) self.patch(designate.hookenv, 'leader_set') self.patch(designate.hookenv, 'leader_get', return_value=False) self.patch(designate.DesignateCharm, 'create_server') self.patch(designate.DesignateCharm, 'create_domain') @contextlib.contextmanager def fake_check_zone_ids(a, b): yield self.patch(designate.DesignateCharm, 'check_zone_ids', new=fake_check_zone_ids) with mock.patch.object(designate.hookenv, 'config', side_effect=FakeConfig(test_config)): designate.DesignateCharm.create_initial_servers_and_domains() self.create_server.assert_has_calls([mock.call('dnsserverrec1.'), mock.call('dnsserverrec2.')]) calls = [ mock.call('novadomain', 'novaemail'), mock.call('neutrondomain', 'neutronemail')] self.create_domain.assert_has_calls(calls) def test_check_zone_ids_change(self): self.patch(designate.hookenv, 'leader_set') DOMAIN_LOOKSUPS = ['novaid1', 'neutronid1', 'novaid1', 'neutronid2'] def fake_get_domain_id(a): return DOMAIN_LOOKSUPS.pop() self.patch(designate.DesignateCharm, 'get_domain_id', side_effect=fake_get_domain_id) with designate.DesignateCharm.check_zone_ids('novadom', 'neutrondom'): pass self.leader_set.assert_called_once_with({'domain-init-done': mock.ANY}) def test_check_zone_ids_nochange(self): self.patch(designate.hookenv, 'leader_set') DOMAIN_LOOKSUPS = ['novaid1', 'neutronid1', 'novaid1', 'neutronid1'] def fake_get_domain_id(a): return DOMAIN_LOOKSUPS.pop() self.patch(designate.DesignateCharm, 'get_domain_id', side_effect=fake_get_domain_id) with designate.DesignateCharm.check_zone_ids('novadom', 'neutrondom'): pass self.assertFalse(self.leader_set.called) def test_render_nrpe(self): self.patch_object(designate.nrpe, 'add_init_service_checks') charm_instance = designate.DesignateCharm(release='queens') charm_instance.render_nrpe() self.add_init_service_checks.assert_has_calls([ mock.call().add_init_service_checks( mock.ANY, charm_instance.services, mock.ANY ), ]) class TestDesignateQueensCharm(Helper): def test_upgrade(self): self.patch(designate.DesignateCharm, 'run_upgrade') self.patch(designate.relations, 'endpoint_from_flag') endpoint = mock.MagicMock() self.endpoint_from_flag.return_value = endpoint a = designate.DesignateCharmQueens(release='queens') a.run_upgrade() self.run_upgrade.assert_called_once_with(interfaces_list=None) endpoint.request_restart.assert_called_once_with()