from mock import MagicMock, patch, call from collections import OrderedDict from copy import deepcopy import charmhelpers.contrib.openstack.templating as templating import charmhelpers.contrib.openstack.utils import neutron_api_context as ncontext templating.OSConfigRenderer = MagicMock() with patch('charmhelpers.core.hookenv.config') as config: config.return_value = 'neutron' import neutron_api_utils as nutils from test_utils import ( CharmTestCase, patch_open, ) import charmhelpers.core.hookenv as hookenv TO_PATCH = [ 'apt_install', 'apt_update', 'apt_upgrade', 'b64encode', 'config', 'configure_installation_source', 'get_os_codename_install_source', 'log', 'neutron_plugin_attribute', 'os_release', 'pip_install', 'subprocess', 'is_elected_leader', ] openstack_origin_git = \ """repositories: - {name: requirements, repository: 'git://git.openstack.org/openstack/requirements', branch: stable/juno} - {name: neutron, repository: 'git://git.openstack.org/openstack/neutron', branch: stable/juno}""" def _mock_npa(plugin, attr, net_manager=None): plugins = { 'ovs': { 'config': '/etc/neutron/plugins/ml2/ml2_conf.ini', 'driver': 'neutron.plugins.ml2.plugin.Ml2Plugin', 'contexts': [], 'services': ['neutron-plugin-openvswitch-agent'], 'packages': [['neutron-plugin-openvswitch-agent']], 'server_packages': ['neutron-server', 'neutron-plugin-ml2'], 'server_services': ['neutron-server'] }, } return plugins[plugin][attr] class DummyIdentityServiceContext(): def __init__(self, return_value): self.return_value = return_value def __call__(self): return self.return_value class TestNeutronAPIUtils(CharmTestCase): def setUp(self): super(TestNeutronAPIUtils, self).setUp(nutils, TO_PATCH) self.config.side_effect = self.test_config.get self.test_config.set('region', 'region101') self.neutron_plugin_attribute.side_effect = _mock_npa def tearDown(self): # Reset cached cache hookenv.cache = {} def test_api_port(self): port = nutils.api_port('neutron-server') self.assertEqual(port, nutils.API_PORTS['neutron-server']) @patch.object(nutils, 'git_install_requested') def test_determine_packages(self, git_requested): git_requested.return_value = False pkg_list = nutils.determine_packages() expect = deepcopy(nutils.BASE_PACKAGES) expect.extend(['neutron-server', 'neutron-plugin-ml2']) self.assertItemsEqual(pkg_list, expect) @patch.object(nutils, 'git_install_requested') def test_determine_packages_kilo(self, git_requested): git_requested.return_value = False self.get_os_codename_install_source.return_value = 'kilo' pkg_list = nutils.determine_packages() expect = deepcopy(nutils.BASE_PACKAGES) expect.extend(['neutron-server', 'neutron-plugin-ml2']) expect.extend(nutils.KILO_PACKAGES) self.assertItemsEqual(pkg_list, expect) @patch.object(nutils, 'git_install_requested') def test_determine_packages_noplugin(self, git_requested): git_requested.return_value = False self.test_config.set('manage-neutron-plugin-legacy-mode', False) pkg_list = nutils.determine_packages() expect = deepcopy(nutils.BASE_PACKAGES) expect.extend(['neutron-server']) self.assertItemsEqual(pkg_list, expect) def test_determine_ports(self): port_list = nutils.determine_ports() self.assertItemsEqual(port_list, [9696]) @patch.object(nutils, 'manage_plugin') @patch('os.path.exists') def test_resource_map(self, _path_exists, _manage_plugin): _path_exists.return_value = False _manage_plugin.return_value = True _map = nutils.resource_map() confs = [nutils.NEUTRON_CONF, nutils.NEUTRON_DEFAULT, nutils.APACHE_CONF] [self.assertIn(q_conf, _map.keys()) for q_conf in confs] self.assertTrue(nutils.APACHE_24_CONF not in _map.keys()) @patch.object(nutils, 'manage_plugin') @patch('os.path.exists') def test_resource_map_apache24(self, _path_exists, _manage_plugin): _path_exists.return_value = True _manage_plugin.return_value = True _map = nutils.resource_map() confs = [nutils.NEUTRON_CONF, nutils.NEUTRON_DEFAULT, nutils.APACHE_24_CONF] [self.assertIn(q_conf, _map.keys()) for q_conf in confs] self.assertTrue(nutils.APACHE_CONF not in _map.keys()) @patch.object(nutils, 'manage_plugin') @patch('os.path.exists') def test_resource_map_noplugin(self, _path_exists, _manage_plugin): _path_exists.return_value = True _manage_plugin.return_value = False _map = nutils.resource_map() found_sdn_ctxt = False found_sdnconfig_ctxt = False for ctxt in _map[nutils.NEUTRON_CONF]['contexts']: if isinstance(ctxt, ncontext.NeutronApiSDNContext): found_sdn_ctxt = True for ctxt in _map[nutils.NEUTRON_DEFAULT]['contexts']: if isinstance(ctxt, ncontext.NeutronApiSDNConfigFileContext): found_sdnconfig_ctxt = True self.assertTrue(found_sdn_ctxt and found_sdnconfig_ctxt) @patch('os.path.exists') def test_restart_map(self, mock_path_exists): mock_path_exists.return_value = False _restart_map = nutils.restart_map() ML2CONF = "/etc/neutron/plugins/ml2/ml2_conf.ini" expect = OrderedDict([ (nutils.NEUTRON_CONF, { 'services': ['neutron-server'], }), (nutils.NEUTRON_DEFAULT, { 'services': ['neutron-server'], }), (ML2CONF, { 'services': ['neutron-server'], }), (nutils.APACHE_CONF, { 'services': ['apache2'], }), (nutils.HAPROXY_CONF, { 'services': ['haproxy'], }), ]) self.assertItemsEqual(_restart_map, expect) @patch('os.path.exists') def test_register_configs(self, mock_path_exists): mock_path_exists.return_value = False class _mock_OSConfigRenderer(): def __init__(self, templates_dir=None, openstack_release=None): self.configs = [] self.ctxts = [] def register(self, config, ctxt): self.configs.append(config) self.ctxts.append(ctxt) templating.OSConfigRenderer.side_effect = _mock_OSConfigRenderer _regconfs = nutils.register_configs() confs = ['/etc/neutron/neutron.conf', '/etc/default/neutron-server', '/etc/neutron/plugins/ml2/ml2_conf.ini', '/etc/apache2/sites-available/openstack_https_frontend', '/etc/haproxy/haproxy.cfg'] self.assertItemsEqual(_regconfs.configs, confs) @patch('os.path.isfile') def test_keystone_ca_cert_b64_no_cert_file(self, _isfile): _isfile.return_value = False cert = nutils.keystone_ca_cert_b64() self.assertEquals(cert, None) @patch('os.path.isfile') def test_keystone_ca_cert_b64(self, _isfile): _isfile.return_value = True with patch_open() as (_open, _file): nutils.keystone_ca_cert_b64() self.assertTrue(self.b64encode.called) @patch.object(nutils, 'migrate_neutron_database') @patch.object(nutils, 'stamp_neutron_database') @patch.object(nutils, 'git_install_requested') def test_do_openstack_upgrade_juno(self, git_requested, stamp_neutron_db, migrate_neutron_db): git_requested.return_value = False self.is_elected_leader.return_value = True self.config.side_effect = self.test_config.get self.test_config.set('openstack-origin', 'cloud:trusty-juno') self.os_release.return_value = 'icehouse' self.get_os_codename_install_source.return_value = 'juno' configs = MagicMock() nutils.do_openstack_upgrade(configs) self.os_release.assert_called_with('neutron-server') self.log.assert_called() self.configure_installation_source.assert_called_with( 'cloud:trusty-juno' ) self.apt_update.assert_called_with(fatal=True) dpkg_opts = [ '--option', 'Dpkg::Options::=--force-confnew', '--option', 'Dpkg::Options::=--force-confdef', ] self.apt_upgrade.assert_called_with(options=dpkg_opts, fatal=True, dist=True) pkgs = nutils.determine_packages() pkgs.sort() self.apt_install.assert_called_with(packages=pkgs, options=dpkg_opts, fatal=True) configs.set_release.assert_called_with(openstack_release='juno') self.assertItemsEqual(stamp_neutron_db.call_args_list, []) self.assertItemsEqual(migrate_neutron_db.call_args_list, []) @patch.object(charmhelpers.contrib.openstack.utils, 'get_os_codename_install_source') @patch.object(nutils, 'migrate_neutron_database') @patch.object(nutils, 'stamp_neutron_database') @patch.object(nutils, 'git_install_requested') def test_do_openstack_upgrade_kilo(self, git_requested, stamp_neutron_db, migrate_neutron_db, gsrc): git_requested.return_value = False self.is_elected_leader.return_value = True self.os_release.return_value = 'juno' self.config.side_effect = self.test_config.get self.test_config.set('openstack-origin', 'cloud:trusty-kilo') gsrc.return_value = 'kilo' self.get_os_codename_install_source.return_value = 'kilo' configs = MagicMock() nutils.do_openstack_upgrade(configs) self.os_release.assert_called_with('neutron-server') self.log.assert_called() self.configure_installation_source.assert_called_with( 'cloud:trusty-kilo' ) self.apt_update.assert_called_with(fatal=True) dpkg_opts = [ '--option', 'Dpkg::Options::=--force-confnew', '--option', 'Dpkg::Options::=--force-confdef', ] self.apt_upgrade.assert_called_with(options=dpkg_opts, fatal=True, dist=True) pkgs = nutils.determine_packages() pkgs.sort() self.apt_install.assert_called_with(packages=pkgs, options=dpkg_opts, fatal=True) configs.set_release.assert_called_with(openstack_release='kilo') stamp_neutron_db.assert_called_with('juno') migrate_neutron_db.assert_called_with() @patch.object(charmhelpers.contrib.openstack.utils, 'get_os_codename_install_source') @patch.object(nutils, 'migrate_neutron_database') @patch.object(nutils, 'stamp_neutron_database') @patch.object(nutils, 'git_install_requested') def test_do_openstack_upgrade_kilo_notleader(self, git_requested, stamp_neutron_db, migrate_neutron_db, gsrc): git_requested.return_value = False self.is_elected_leader.return_value = False self.os_release.return_value = 'juno' self.config.side_effect = self.test_config.get self.test_config.set('openstack-origin', 'cloud:trusty-kilo') gsrc.return_value = 'kilo' self.get_os_codename_install_source.return_value = 'kilo' configs = MagicMock() nutils.do_openstack_upgrade(configs) self.os_release.assert_called_with('neutron-server') self.log.assert_called() self.configure_installation_source.assert_called_with( 'cloud:trusty-kilo' ) self.apt_update.assert_called_with(fatal=True) dpkg_opts = [ '--option', 'Dpkg::Options::=--force-confnew', '--option', 'Dpkg::Options::=--force-confdef', ] self.apt_upgrade.assert_called_with(options=dpkg_opts, fatal=True, dist=True) pkgs = nutils.determine_packages() pkgs.sort() self.apt_install.assert_called_with(packages=pkgs, options=dpkg_opts, fatal=True) configs.set_release.assert_called_with(openstack_release='kilo') self.assertFalse(stamp_neutron_db.called) self.assertFalse(migrate_neutron_db.called) @patch.object(ncontext, 'IdentityServiceContext') @patch('neutronclient.v2_0.client.Client') def test_get_neutron_client(self, nclient, IdentityServiceContext): creds = { 'auth_protocol': 'http', 'auth_host': 'myhost', 'auth_port': '2222', 'admin_user': 'bob', 'admin_password': 'pa55w0rd', 'admin_tenant_name': 'tenant1', 'region': 'region2', } IdentityServiceContext.return_value = \ DummyIdentityServiceContext(return_value=creds) nutils.get_neutron_client() nclient.assert_called_with( username='bob', tenant_name='tenant1', password='pa55w0rd', auth_url='http://myhost:2222/v2.0', region_name='region2', ) @patch.object(ncontext, 'IdentityServiceContext') def test_get_neutron_client_noidservice(self, IdentityServiceContext): creds = {} IdentityServiceContext.return_value = \ DummyIdentityServiceContext(return_value=creds) self.assertEquals(nutils.get_neutron_client(), None) @patch.object(nutils, 'get_neutron_client') def test_router_feature_present_keymissing(self, get_neutron_client): routers = { 'routers': [ { u'status': u'ACTIVE', u'external_gateway_info': { u'network_id': u'eedffb9b-b93e-49c6-9545-47c656c9678e', u'enable_snat': True }, u'name': u'provider-router', u'admin_state_up': True, u'tenant_id': u'b240d06e38394780a3ea296138cdd174', u'routes': [], u'id': u'84182bc8-eede-4564-9c87-1a56bdb26a90', } ] } get_neutron_client.list_routers.return_value = routers self.assertEquals(nutils.router_feature_present('ha'), False) @patch.object(nutils, 'get_neutron_client') def test_router_feature_present_keyfalse(self, get_neutron_client): routers = { 'routers': [ { u'status': u'ACTIVE', u'external_gateway_info': { u'network_id': u'eedffb9b-b93e-49c6-9545-47c656c9678e', u'enable_snat': True }, u'name': u'provider-router', u'admin_state_up': True, u'tenant_id': u'b240d06e38394780a3ea296138cdd174', u'routes': [], u'id': u'84182bc8-eede-4564-9c87-1a56bdb26a90', u'ha': False, } ] } dummy_client = MagicMock() dummy_client.list_routers.return_value = routers get_neutron_client.return_value = dummy_client self.assertEquals(nutils.router_feature_present('ha'), False) @patch.object(nutils, 'get_neutron_client') def test_router_feature_present_keytrue(self, get_neutron_client): routers = { 'routers': [ { u'status': u'ACTIVE', u'external_gateway_info': { u'network_id': u'eedffb9b-b93e-49c6-9545-47c656c9678e', u'enable_snat': True }, u'name': u'provider-router', u'admin_state_up': True, u'tenant_id': u'b240d06e38394780a3ea296138cdd174', u'routes': [], u'id': u'84182bc8-eede-4564-9c87-1a56bdb26a90', u'ha': True, } ] } dummy_client = MagicMock() dummy_client.list_routers.return_value = routers get_neutron_client.return_value = dummy_client self.assertEquals(nutils.router_feature_present('ha'), True) @patch.object(nutils, 'get_neutron_client') def test_neutron_ready(self, get_neutron_client): dummy_client = MagicMock() dummy_client.list_routers.return_value = [] get_neutron_client.return_value = dummy_client self.assertEquals(nutils.neutron_ready(), True) @patch.object(nutils, 'get_neutron_client') def test_neutron_ready_noclient(self, get_neutron_client): get_neutron_client.return_value = None self.assertEquals(nutils.neutron_ready(), False) @patch.object(nutils, 'get_neutron_client') def test_neutron_ready_clientexception(self, get_neutron_client): dummy_client = MagicMock() dummy_client.list_routers.side_effect = Exception('Boom!') get_neutron_client.return_value = dummy_client self.assertEquals(nutils.neutron_ready(), False) @patch.object(nutils, 'git_install_requested') @patch.object(nutils, 'git_clone_and_install') @patch.object(nutils, 'git_post_install') @patch.object(nutils, 'git_pre_install') def test_git_install(self, git_pre, git_post, git_clone_and_install, git_requested): projects_yaml = openstack_origin_git git_requested.return_value = True nutils.git_install(projects_yaml) self.assertTrue(git_pre.called) git_clone_and_install.assert_called_with(openstack_origin_git, core_project='neutron') self.assertTrue(git_post.called) @patch.object(nutils, 'mkdir') @patch.object(nutils, 'write_file') @patch.object(nutils, 'add_user_to_group') @patch.object(nutils, 'add_group') @patch.object(nutils, 'adduser') def test_git_pre_install(self, adduser, add_group, add_user_to_group, write_file, mkdir): nutils.git_pre_install() adduser.assert_called_with('neutron', shell='/bin/bash', system_user=True) add_group.assert_called_with('neutron', system_group=True) add_user_to_group.assert_called_with('neutron', 'neutron') expected = [ call('/var/lib/neutron', owner='neutron', group='neutron', perms=0755, force=False), call('/var/lib/neutron/lock', owner='neutron', group='neutron', perms=0755, force=False), call('/var/log/neutron', owner='neutron', group='neutron', perms=0755, force=False), ] self.assertEquals(mkdir.call_args_list, expected) expected = [ call('/var/log/neutron/server.log', '', owner='neutron', group='neutron', perms=0600), ] self.assertEquals(write_file.call_args_list, expected) @patch.object(nutils, 'git_src_dir') @patch.object(nutils, 'service_restart') @patch.object(nutils, 'render') @patch.object(nutils, 'git_pip_venv_dir') @patch('os.path.join') @patch('os.path.exists') @patch('os.symlink') @patch('shutil.copytree') @patch('shutil.rmtree') @patch('subprocess.check_call') def test_git_post_install(self, check_call, rmtree, copytree, symlink, exists, join, venv, render, service_restart, git_src_dir): projects_yaml = openstack_origin_git join.return_value = 'joined-string' venv.return_value = '/mnt/openstack-git/venv' nutils.git_post_install(projects_yaml) expected = [ call('joined-string', '/etc/neutron'), call('joined-string', '/etc/neutron/plugins'), call('joined-string', '/etc/neutron/rootwrap.d'), ] copytree.assert_has_calls(expected) expected = [ call('joined-string', '/usr/local/bin/neutron-rootwrap'), call('joined-string', '/usr/local/bin/neutron-db-manage'), ] symlink.assert_has_calls(expected, any_order=True) neutron_api_context = { 'service_description': 'Neutron API server', 'charm_name': 'neutron-api', 'process_name': 'neutron-server', 'executable_name': 'joined-string', } expected = [ call('git/neutron_sudoers', '/etc/sudoers.d/neutron_sudoers', {}, perms=0o440), call('git/upstart/neutron-server.upstart', '/etc/init/neutron-server.conf', neutron_api_context, perms=0o644), ] self.assertEquals(render.call_args_list, expected) expected = [ call('neutron-server'), ] self.assertEquals(service_restart.call_args_list, expected) def test_stamp_neutron_database(self): nutils.stamp_neutron_database('icehouse') cmd = ['neutron-db-manage', '--config-file', '/etc/neutron/neutron.conf', '--config-file', '/etc/neutron/plugins/ml2/ml2_conf.ini', 'stamp', 'icehouse'] self.subprocess.check_output.assert_called_with(cmd) def test_migrate_neutron_database(self): nutils.migrate_neutron_database() cmd = ['neutron-db-manage', '--config-file', '/etc/neutron/neutron.conf', '--config-file', '/etc/neutron/plugins/ml2/ml2_conf.ini', 'upgrade', 'head'] self.subprocess.check_output.assert_called_with(cmd) def test_manage_plugin_true(self): self.test_config.set('manage-neutron-plugin-legacy-mode', True) manage = nutils.manage_plugin() self.assertTrue(manage) def test_manage_plugin_false(self): self.test_config.set('manage-neutron-plugin-legacy-mode', False) manage = nutils.manage_plugin() self.assertFalse(manage)