¡@
OpenStack Study: test_dhcp_agent.py**** CubicPower OpenStack Study ****
# vim: tabstop=4 shiftwidth=4 softtabstop=4 # Copyright 2012 OpenStack Foundation # All Rights Reserved. # # 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. import copy import os import sys import uuid import eventlet import mock from oslo.config import cfg import testtools from neutron.agent.common import config from neutron.agent import dhcp_agent from neutron.agent.dhcp_agent import DhcpAgentWithStateReport from neutron.agent.linux import dhcp from neutron.agent.linux import interface from neutron.common import constants as const from neutron.common import exceptions from neutron.openstack.common.rpc import common from neutron.tests import base ROOTDIR = os.path.dirname(os.path.dirname(__file__)) ETCDIR = os.path.join(ROOTDIR, 'etc') HOSTNAME = 'hostname' dev_man = dhcp.DeviceManager rpc_api = dhcp_agent.DhcpPluginApi DEVICE_MANAGER = '%s.%s' % (dev_man.__module__, dev_man.__name__) DHCP_PLUGIN = '%s.%s' % (rpc_api.__module__, rpc_api.__name__)
**** CubicPower OpenStack Study **** def etcdir(*p): return os.path.join(ETCDIR, *p) fake_tenant_id = 'aaaaaaaa-aaaa-aaaa-aaaaaaaaaaaa' fake_subnet1_allocation_pools = dhcp.DictModel(dict(id='', start='172.9.9.2', end='172.9.9.254')) fake_subnet1 = dhcp.DictModel(dict(id='bbbbbbbb-bbbb-bbbb-bbbbbbbbbbbb', network_id='12345678-1234-5678-1234567890ab', cidr='172.9.9.0/24', enable_dhcp=True, name='', tenant_id=fake_tenant_id, gateway_ip='172.9.9.1', host_routes=[], dns_nameservers=[], ip_version=4, allocation_pools=fake_subnet1_allocation_pools)) fake_subnet2_allocation_pools = dhcp.DictModel(dict(id='', start='172.9.8.2', end='172.9.8.254')) fake_subnet2 = dhcp.DictModel(dict(id='dddddddd-dddd-dddd-dddddddddddd', network_id='12345678-1234-5678-1234567890ab', cidr='172.9.8.0/24', enable_dhcp=False, name='', tenant_id=fake_tenant_id, gateway_ip='172.9.8.1', host_routes=[], dns_nameservers=[], ip_version=4, allocation_pools=fake_subnet2_allocation_pools)) fake_subnet3 = dhcp.DictModel(dict(id='bbbbbbbb-1111-2222-bbbbbbbbbbbb', network_id='12345678-1234-5678-1234567890ab', cidr='192.168.1.1/24', enable_dhcp=True)) fake_meta_subnet = dhcp.DictModel(dict(id='bbbbbbbb-1111-2222-bbbbbbbbbbbb', network_id='12345678-1234-5678-1234567890ab', cidr='169.254.169.252/30', gateway_ip='169.254.169.253', enable_dhcp=True)) fake_fixed_ip1 = dhcp.DictModel(dict(id='', subnet_id=fake_subnet1.id, ip_address='172.9.9.9')) fake_meta_fixed_ip = dhcp.DictModel(dict(id='', subnet=fake_meta_subnet, ip_address='169.254.169.254')) fake_allocation_pool_subnet1 = dhcp.DictModel(dict(id='', start='172.9.9.2', end='172.9.9.254')) fake_port1 = dhcp.DictModel(dict(id='12345678-1234-aaaa-1234567890ab', device_id='dhcp-12345678-1234-aaaa-1234567890ab', allocation_pools=fake_subnet1_allocation_pools, mac_address='aa:bb:cc:dd:ee:ff', network_id='12345678-1234-5678-1234567890ab', fixed_ips=[fake_fixed_ip1])) fake_port2 = dhcp.DictModel(dict(id='12345678-1234-aaaa-123456789000', mac_address='aa:bb:cc:dd:ee:99', network_id='12345678-1234-5678-1234567890ab', fixed_ips=[])) fake_meta_port = dhcp.DictModel(dict(id='12345678-1234-aaaa-1234567890ab', mac_address='aa:bb:cc:dd:ee:ff', network_id='12345678-1234-5678-1234567890ab', device_owner=const.DEVICE_OWNER_ROUTER_INTF, device_id='forzanapoli', fixed_ips=[fake_meta_fixed_ip])) fake_network = dhcp.NetModel(True, dict(id='12345678-1234-5678-1234567890ab', tenant_id='aaaaaaaa-aaaa-aaaa-aaaaaaaaaaaa', admin_state_up=True, subnets=[fake_subnet1, fake_subnet2], ports=[fake_port1])) fake_meta_network = dhcp.NetModel( True, dict(id='12345678-1234-5678-1234567890ab', tenant_id='aaaaaaaa-aaaa-aaaa-aaaaaaaaaaaa', admin_state_up=True, subnets=[fake_meta_subnet], ports=[fake_meta_port])) fake_down_network = dhcp.NetModel( True, dict(id='12345678-dddd-dddd-1234567890ab', tenant_id='aaaaaaaa-aaaa-aaaa-aaaaaaaaaaaa', admin_state_up=False, subnets=[], ports=[]))
**** CubicPower OpenStack Study **** class TestDhcpAgent(base.BaseTestCase):
**** CubicPower OpenStack Study **** def setUp(self): super(TestDhcpAgent, self).setUp() dhcp_agent.register_options() cfg.CONF.set_override('interface_driver', 'neutron.agent.linux.interface.NullDriver') # disable setting up periodic state reporting cfg.CONF.set_override('report_interval', 0, 'AGENT') self.driver_cls_p = mock.patch( 'neutron.agent.dhcp_agent.importutils.import_class') self.driver = mock.Mock(name='driver') self.driver.existing_dhcp_networks.return_value = [] self.driver_cls = self.driver_cls_p.start() self.driver_cls.return_value = self.driver self.mock_makedirs_p = mock.patch("os.makedirs") self.mock_makedirs = self.mock_makedirs_p.start()
**** CubicPower OpenStack Study **** def test_dhcp_agent_manager(self): state_rpc_str = 'neutron.agent.rpc.PluginReportStateAPI' # sync_state is needed for this test cfg.CONF.set_override('report_interval', 1, 'AGENT') with mock.patch.object(DhcpAgentWithStateReport, 'sync_state', autospec=True) as mock_sync_state: with mock.patch.object(DhcpAgentWithStateReport, 'periodic_resync', autospec=True) as mock_periodic_resync: with mock.patch(state_rpc_str) as state_rpc: with mock.patch.object(sys, 'argv') as sys_argv: sys_argv.return_value = [ 'dhcp', '--config-file', etcdir('neutron.conf.test')] cfg.CONF.register_opts(dhcp_agent.DhcpAgent.OPTS) config.register_interface_driver_opts_helper(cfg.CONF) config.register_agent_state_opts_helper(cfg.CONF) config.register_root_helper(cfg.CONF) cfg.CONF.register_opts(dhcp.OPTS) cfg.CONF.register_opts(interface.OPTS) cfg.CONF(project='neutron') agent_mgr = DhcpAgentWithStateReport('testhost') eventlet.greenthread.sleep(1) agent_mgr.after_start() mock_sync_state.assert_called_once_with(agent_mgr) mock_periodic_resync.assert_called_once_with(agent_mgr) state_rpc.assert_has_calls( [mock.call(mock.ANY), mock.call().report_state(mock.ANY, mock.ANY, mock.ANY)])
**** CubicPower OpenStack Study **** def test_dhcp_agent_main_agent_manager(self): logging_str = 'neutron.agent.common.config.setup_logging' launcher_str = 'neutron.openstack.common.service.ServiceLauncher' with mock.patch(logging_str): with mock.patch.object(sys, 'argv') as sys_argv: with mock.patch(launcher_str) as launcher: sys_argv.return_value = ['dhcp', '--config-file', etcdir('neutron.conf.test')] dhcp_agent.main() launcher.assert_has_calls( [mock.call(), mock.call().launch_service(mock.ANY), mock.call().wait()])
**** CubicPower OpenStack Study **** def test_run_completes_single_pass(self): with mock.patch(DEVICE_MANAGER): dhcp = dhcp_agent.DhcpAgent(HOSTNAME) attrs_to_mock = dict( [(a, mock.DEFAULT) for a in ['sync_state', 'periodic_resync']]) with mock.patch.multiple(dhcp, **attrs_to_mock) as mocks: dhcp.run() mocks['sync_state'].assert_called_once_with() mocks['periodic_resync'].assert_called_once_with()
**** CubicPower OpenStack Study **** def test_call_driver(self): network = mock.Mock() network.id = '1' dhcp = dhcp_agent.DhcpAgent(cfg.CONF) self.assertTrue(dhcp.call_driver('foo', network)) self.driver.assert_called_once_with(cfg.CONF, mock.ANY, 'sudo', mock.ANY, mock.ANY)
**** CubicPower OpenStack Study **** def _test_call_driver_failure(self, exc=None, trace_level='exception', expected_sync=True): network = mock.Mock() network.id = '1' self.driver.return_value.foo.side_effect = exc or Exception with mock.patch.object(dhcp_agent.LOG, trace_level) as log: dhcp = dhcp_agent.DhcpAgent(HOSTNAME) self.assertIsNone(dhcp.call_driver('foo', network)) self.driver.assert_called_once_with(cfg.CONF, mock.ANY, 'sudo', mock.ANY, mock.ANY) self.assertEqual(log.call_count, 1) self.assertEqual(expected_sync, dhcp.needs_resync)
**** CubicPower OpenStack Study **** def test_call_driver_failure(self): self._test_call_driver_failure()
**** CubicPower OpenStack Study **** def test_call_driver_remote_error_net_not_found(self): self._test_call_driver_failure( exc=common.RemoteError(exc_type='NetworkNotFound'), trace_level='warning')
**** CubicPower OpenStack Study **** def test_call_driver_network_not_found(self): self._test_call_driver_failure( exc=exceptions.NetworkNotFound(net_id='1'), trace_level='warning')
**** CubicPower OpenStack Study **** def test_call_driver_conflict(self): self._test_call_driver_failure( exc=exceptions.Conflict(), trace_level='warning', expected_sync=False)
**** CubicPower OpenStack Study **** def _test_sync_state_helper(self, known_networks, active_networks): with mock.patch(DHCP_PLUGIN) as plug: mock_plugin = mock.Mock() mock_plugin.get_active_networks_info.return_value = active_networks plug.return_value = mock_plugin dhcp = dhcp_agent.DhcpAgent(HOSTNAME) attrs_to_mock = dict( [(a, mock.DEFAULT) for a in ['refresh_dhcp_helper', 'disable_dhcp_helper', 'cache']]) with mock.patch.multiple(dhcp, **attrs_to_mock) as mocks: mocks['cache'].get_network_ids.return_value = known_networks dhcp.sync_state() exp_refresh = [ mock.call(net_id) for net_id in active_networks] diff = set(known_networks) - set(active_networks) exp_disable = [mock.call(net_id) for net_id in diff] mocks['cache'].assert_has_calls([mock.call.get_network_ids()]) mocks['refresh_dhcp_helper'].assert_has_called(exp_refresh) mocks['disable_dhcp_helper'].assert_has_called(exp_disable)
**** CubicPower OpenStack Study **** def test_sync_state_initial(self): self._test_sync_state_helper([], ['a'])
**** CubicPower OpenStack Study **** def test_sync_state_same(self): self._test_sync_state_helper(['a'], ['a'])
**** CubicPower OpenStack Study **** def test_sync_state_disabled_net(self): self._test_sync_state_helper(['b'], ['a'])
**** CubicPower OpenStack Study **** def test_sync_state_waitall(self): class mockNetwork(): id = '0' admin_state_up = True subnets = [] def __init__(self, id): self.id = id with mock.patch.object(dhcp_agent.eventlet.GreenPool, 'waitall') as w: active_networks = [mockNetwork('1'), mockNetwork('2'), mockNetwork('3'), mockNetwork('4'), mockNetwork('5')] known_networks = ['1', '2', '3', '4', '5'] self._test_sync_state_helper(known_networks, active_networks) w.assert_called_once_with()
**** CubicPower OpenStack Study **** def test_sync_state_plugin_error(self): with mock.patch(DHCP_PLUGIN) as plug: mock_plugin = mock.Mock() mock_plugin.get_active_networks_info.side_effect = Exception plug.return_value = mock_plugin with mock.patch.object(dhcp_agent.LOG, 'exception') as log: dhcp = dhcp_agent.DhcpAgent(HOSTNAME) dhcp.sync_state() self.assertTrue(log.called) self.assertTrue(dhcp.needs_resync)
**** CubicPower OpenStack Study **** def test_periodic_resync(self): dhcp = dhcp_agent.DhcpAgent(HOSTNAME) with mock.patch.object(dhcp_agent.eventlet, 'spawn') as spawn: dhcp.periodic_resync() spawn.assert_called_once_with(dhcp._periodic_resync_helper)
**** CubicPower OpenStack Study **** def test_periodoc_resync_helper(self): with mock.patch.object(dhcp_agent.eventlet, 'sleep') as sleep: dhcp = dhcp_agent.DhcpAgent(HOSTNAME) dhcp.needs_resync = True with mock.patch.object(dhcp, 'sync_state') as sync_state: sync_state.side_effect = RuntimeError with testtools.ExpectedException(RuntimeError): dhcp._periodic_resync_helper() sync_state.assert_called_once_with() sleep.assert_called_once_with(dhcp.conf.resync_interval) self.assertFalse(dhcp.needs_resync)
**** CubicPower OpenStack Study **** def test_populate_cache_on_start_without_active_networks_support(self): # emul dhcp driver that doesn't support retrieving of active networks self.driver.existing_dhcp_networks.side_effect = NotImplementedError with mock.patch.object(dhcp_agent.LOG, 'debug') as log: dhcp = dhcp_agent.DhcpAgent(HOSTNAME) self.driver.existing_dhcp_networks.assert_called_once_with( dhcp.conf, cfg.CONF.root_helper ) self.assertFalse(dhcp.cache.get_network_ids()) self.assertTrue(log.called)
**** CubicPower OpenStack Study **** def test_populate_cache_on_start(self): networks = ['aaa', 'bbb'] self.driver.existing_dhcp_networks.return_value = networks dhcp = dhcp_agent.DhcpAgent(HOSTNAME) self.driver.existing_dhcp_networks.assert_called_once_with( dhcp.conf, cfg.CONF.root_helper ) self.assertEqual(set(networks), set(dhcp.cache.get_network_ids()))
**** CubicPower OpenStack Study **** def test_none_interface_driver(self): cfg.CONF.set_override('interface_driver', None) with mock.patch.object(dhcp, 'LOG') as log: self.assertRaises(SystemExit, dhcp.DeviceManager, cfg.CONF, 'sudo', None) msg = 'An interface driver must be specified' log.error.assert_called_once_with(msg)
**** CubicPower OpenStack Study **** def test_nonexistent_interface_driver(self): # Temporarily turn off mock, so could use the real import_class # to import interface_driver. self.driver_cls_p.stop() self.addCleanup(self.driver_cls_p.start) cfg.CONF.set_override('interface_driver', 'foo') with mock.patch.object(dhcp, 'LOG') as log: self.assertRaises(SystemExit, dhcp.DeviceManager, cfg.CONF, 'sudo', None) log.error.assert_called_once()
**** CubicPower OpenStack Study **** class TestLogArgs(base.BaseTestCase):
**** CubicPower OpenStack Study **** def test_log_args_without_log_dir_and_file(self): conf_dict = {'debug': True, 'verbose': False, 'log_dir': None, 'log_file': None, 'use_syslog': True, 'syslog_log_facility': 'LOG_USER'} conf = dhcp.DictModel(conf_dict) expected_args = ['--debug', '--use-syslog', '--syslog-log-facility=LOG_USER'] args = config.get_log_args(conf, 'log_file_name') self.assertEqual(expected_args, args)
**** CubicPower OpenStack Study **** def test_log_args_without_log_file(self): conf_dict = {'debug': True, 'verbose': True, 'log_dir': '/etc/tests', 'log_file': None, 'use_syslog': False, 'syslog_log_facility': 'LOG_USER'} conf = dhcp.DictModel(conf_dict) expected_args = ['--debug', '--verbose', '--log-file=log_file_name', '--log-dir=/etc/tests'] args = config.get_log_args(conf, 'log_file_name') self.assertEqual(expected_args, args)
**** CubicPower OpenStack Study **** def test_log_args_with_log_dir_and_file(self): conf_dict = {'debug': True, 'verbose': False, 'log_dir': '/etc/tests', 'log_file': 'tests/filelog', 'use_syslog': False, 'syslog_log_facility': 'LOG_USER'} conf = dhcp.DictModel(conf_dict) expected_args = ['--debug', '--log-file=log_file_name', '--log-dir=/etc/tests/tests'] args = config.get_log_args(conf, 'log_file_name') self.assertEqual(expected_args, args)
**** CubicPower OpenStack Study **** def test_log_args_without_log_dir(self): conf_dict = {'debug': True, 'verbose': False, 'log_file': 'tests/filelog', 'log_dir': None, 'use_syslog': False, 'syslog_log_facility': 'LOG_USER'} conf = dhcp.DictModel(conf_dict) expected_args = ['--debug', '--log-file=log_file_name', '--log-dir=tests'] args = config.get_log_args(conf, 'log_file_name') self.assertEqual(expected_args, args)
**** CubicPower OpenStack Study **** def test_log_args_with_filelog_and_syslog(self): conf_dict = {'debug': True, 'verbose': True, 'log_file': 'tests/filelog', 'log_dir': '/etc/tests', 'use_syslog': True, 'syslog_log_facility': 'LOG_USER'} conf = dhcp.DictModel(conf_dict) expected_args = ['--debug', '--verbose', '--log-file=log_file_name', '--log-dir=/etc/tests/tests'] args = config.get_log_args(conf, 'log_file_name') self.assertEqual(expected_args, args)
**** CubicPower OpenStack Study **** class TestDhcpAgentEventHandler(base.BaseTestCase):
**** CubicPower OpenStack Study **** def setUp(self): super(TestDhcpAgentEventHandler, self).setUp() config.register_interface_driver_opts_helper(cfg.CONF) cfg.CONF.register_opts(dhcp.OPTS) cfg.CONF.set_override('interface_driver', 'neutron.agent.linux.interface.NullDriver') config.register_root_helper(cfg.CONF) cfg.CONF.register_opts(dhcp_agent.DhcpAgent.OPTS) self.plugin_p = mock.patch(DHCP_PLUGIN) plugin_cls = self.plugin_p.start() self.plugin = mock.Mock() plugin_cls.return_value = self.plugin self.cache_p = mock.patch('neutron.agent.dhcp_agent.NetworkCache') cache_cls = self.cache_p.start() self.cache = mock.Mock() cache_cls.return_value = self.cache self.mock_makedirs_p = mock.patch("os.makedirs") self.mock_makedirs = self.mock_makedirs_p.start() self.mock_init_p = mock.patch('neutron.agent.dhcp_agent.' 'DhcpAgent._populate_networks_cache') self.mock_init = self.mock_init_p.start() with mock.patch.object(dhcp.Dnsmasq, 'check_version') as check_v: check_v.return_value = dhcp.Dnsmasq.MINIMUM_VERSION self.dhcp = dhcp_agent.DhcpAgent(HOSTNAME) self.call_driver_p = mock.patch.object(self.dhcp, 'call_driver') self.call_driver = self.call_driver_p.start() self.external_process_p = mock.patch( 'neutron.agent.linux.external_process.ProcessManager' ) self.external_process = self.external_process_p.start()
**** CubicPower OpenStack Study **** def tearDown(self): self.external_process_p.stop() self.call_driver_p.stop() self.cache_p.stop() self.plugin_p.stop() self.mock_makedirs_p.stop() self.mock_init_p.stop() super(TestDhcpAgentEventHandler, self).tearDown()
**** CubicPower OpenStack Study **** def _enable_dhcp_helper(self, isolated_metadata=False): if isolated_metadata: cfg.CONF.set_override('enable_isolated_metadata', True) self.plugin.get_network_info.return_value = fake_network self.dhcp.enable_dhcp_helper(fake_network.id) self.plugin.assert_has_calls( [mock.call.get_network_info(fake_network.id)]) self.call_driver.assert_called_once_with('enable', fake_network) self.cache.assert_has_calls([mock.call.put(fake_network)]) if isolated_metadata: self.external_process.assert_has_calls([ mock.call( cfg.CONF, '12345678-1234-5678-1234567890ab', 'sudo', 'qdhcp-12345678-1234-5678-1234567890ab'), mock.call().enable(mock.ANY) ]) else: self.assertFalse(self.external_process.call_count)
**** CubicPower OpenStack Study **** def test_enable_dhcp_helper_enable_isolated_metadata(self): self._enable_dhcp_helper(isolated_metadata=True)
**** CubicPower OpenStack Study **** def test_enable_dhcp_helper(self): self._enable_dhcp_helper()
**** CubicPower OpenStack Study **** def test_enable_dhcp_helper_down_network(self): self.plugin.get_network_info.return_value = fake_down_network self.dhcp.enable_dhcp_helper(fake_down_network.id) self.plugin.assert_has_calls( [mock.call.get_network_info(fake_down_network.id)]) self.assertFalse(self.call_driver.called) self.assertFalse(self.cache.called) self.assertFalse(self.external_process.called)
**** CubicPower OpenStack Study **** def test_enable_dhcp_helper_network_none(self): self.plugin.get_network_info.return_value = None with mock.patch.object(dhcp_agent.LOG, 'warn') as log: self.dhcp.enable_dhcp_helper('fake_id') self.plugin.assert_has_calls( [mock.call.get_network_info('fake_id')]) self.assertFalse(self.call_driver.called) self.assertTrue(log.called) self.assertFalse(self.dhcp.needs_resync)
**** CubicPower OpenStack Study **** def test_enable_dhcp_helper_exception_during_rpc(self): self.plugin.get_network_info.side_effect = Exception with mock.patch.object(dhcp_agent.LOG, 'exception') as log: self.dhcp.enable_dhcp_helper(fake_network.id) self.plugin.assert_has_calls( [mock.call.get_network_info(fake_network.id)]) self.assertFalse(self.call_driver.called) self.assertTrue(log.called) self.assertTrue(self.dhcp.needs_resync) self.assertFalse(self.cache.called) self.assertFalse(self.external_process.called)
**** CubicPower OpenStack Study **** def test_enable_dhcp_helper_driver_failure(self): self.plugin.get_network_info.return_value = fake_network self.call_driver.return_value = False self.dhcp.enable_dhcp_helper(fake_network.id) self.plugin.assert_has_calls( [mock.call.get_network_info(fake_network.id)]) self.call_driver.assert_called_once_with('enable', fake_network) self.assertFalse(self.cache.called) self.assertFalse(self.external_process.called)
**** CubicPower OpenStack Study **** def _disable_dhcp_helper_known_network(self, isolated_metadata=False): if isolated_metadata: cfg.CONF.set_override('enable_isolated_metadata', True) self.cache.get_network_by_id.return_value = fake_network self.dhcp.disable_dhcp_helper(fake_network.id) self.cache.assert_has_calls( [mock.call.get_network_by_id(fake_network.id)]) self.call_driver.assert_called_once_with('disable', fake_network) if isolated_metadata: self.external_process.assert_has_calls([ mock.call( cfg.CONF, '12345678-1234-5678-1234567890ab', 'sudo', 'qdhcp-12345678-1234-5678-1234567890ab'), mock.call().disable() ]) else: self.assertFalse(self.external_process.call_count)
**** CubicPower OpenStack Study **** def test_disable_dhcp_helper_known_network_isolated_metadata(self): self._disable_dhcp_helper_known_network(isolated_metadata=True)
**** CubicPower OpenStack Study **** def test_disable_dhcp_helper_known_network(self): self._disable_dhcp_helper_known_network()
**** CubicPower OpenStack Study **** def test_disable_dhcp_helper_unknown_network(self): self.cache.get_network_by_id.return_value = None self.dhcp.disable_dhcp_helper('abcdef') self.cache.assert_has_calls( [mock.call.get_network_by_id('abcdef')]) self.assertEqual(0, self.call_driver.call_count) self.assertFalse(self.external_process.called)
**** CubicPower OpenStack Study **** def _disable_dhcp_helper_driver_failure(self, isolated_metadata=False): if isolated_metadata: cfg.CONF.set_override('enable_isolated_metadata', True) self.cache.get_network_by_id.return_value = fake_network self.call_driver.return_value = False self.dhcp.disable_dhcp_helper(fake_network.id) self.cache.assert_has_calls( [mock.call.get_network_by_id(fake_network.id)]) self.call_driver.assert_called_once_with('disable', fake_network) self.cache.assert_has_calls( [mock.call.get_network_by_id(fake_network.id)]) if isolated_metadata: self.external_process.assert_has_calls([ mock.call( cfg.CONF, '12345678-1234-5678-1234567890ab', 'sudo', 'qdhcp-12345678-1234-5678-1234567890ab'), mock.call().disable() ]) else: self.assertFalse(self.external_process.call_count)
**** CubicPower OpenStack Study **** def test_disable_dhcp_helper_driver_failure_isolated_metadata(self): self._disable_dhcp_helper_driver_failure(isolated_metadata=True)
**** CubicPower OpenStack Study **** def test_disable_dhcp_helper_driver_failure(self): self._disable_dhcp_helper_driver_failure()
**** CubicPower OpenStack Study **** def test_enable_isolated_metadata_proxy(self): class_path = 'neutron.agent.linux.external_process.ProcessManager' with mock.patch(class_path) as ext_process: self.dhcp.enable_isolated_metadata_proxy(fake_network) ext_process.assert_has_calls([ mock.call( cfg.CONF, '12345678-1234-5678-1234567890ab', 'sudo', 'qdhcp-12345678-1234-5678-1234567890ab'), mock.call().enable(mock.ANY) ])
**** CubicPower OpenStack Study **** def test_disable_isolated_metadata_proxy(self): class_path = 'neutron.agent.linux.external_process.ProcessManager' with mock.patch(class_path) as ext_process: self.dhcp.disable_isolated_metadata_proxy(fake_network) ext_process.assert_has_calls([ mock.call( cfg.CONF, '12345678-1234-5678-1234567890ab', 'sudo', 'qdhcp-12345678-1234-5678-1234567890ab'), mock.call().disable() ])
**** CubicPower OpenStack Study **** def test_enable_isolated_metadata_proxy_with_metadata_network(self): cfg.CONF.set_override('enable_metadata_network', True) cfg.CONF.set_override('debug', True) cfg.CONF.set_override('log_file', 'test.log') class_path = 'neutron.agent.linux.ip_lib.IPWrapper' self.external_process_p.stop() # Ensure the mock is restored if this test fail try: with mock.patch(class_path) as ip_wrapper: self.dhcp.enable_isolated_metadata_proxy(fake_meta_network) ip_wrapper.assert_has_calls([mock.call( 'sudo', 'qdhcp-12345678-1234-5678-1234567890ab'), mock.call().netns.execute([ 'neutron-ns-metadata-proxy', mock.ANY, mock.ANY, '--router_id=forzanapoli', mock.ANY, mock.ANY, '--debug', ('--log-file=neutron-ns-metadata-proxy-%s.log' % fake_meta_network.id)]) ]) finally: self.external_process_p.start()
**** CubicPower OpenStack Study **** def test_network_create_end(self): payload = dict(network=dict(id=fake_network.id)) with mock.patch.object(self.dhcp, 'enable_dhcp_helper') as enable: self.dhcp.network_create_end(None, payload) enable.assertCalledOnceWith(fake_network.id)
**** CubicPower OpenStack Study **** def test_network_update_end_admin_state_up(self): payload = dict(network=dict(id=fake_network.id, admin_state_up=True)) with mock.patch.object(self.dhcp, 'enable_dhcp_helper') as enable: self.dhcp.network_update_end(None, payload) enable.assertCalledOnceWith(fake_network.id)
**** CubicPower OpenStack Study **** def test_network_update_end_admin_state_down(self): payload = dict(network=dict(id=fake_network.id, admin_state_up=False)) with mock.patch.object(self.dhcp, 'disable_dhcp_helper') as disable: self.dhcp.network_update_end(None, payload) disable.assertCalledOnceWith(fake_network.id)
**** CubicPower OpenStack Study **** def test_network_delete_end(self): payload = dict(network_id=fake_network.id) with mock.patch.object(self.dhcp, 'disable_dhcp_helper') as disable: self.dhcp.network_delete_end(None, payload) disable.assertCalledOnceWith(fake_network.id)
**** CubicPower OpenStack Study **** def test_refresh_dhcp_helper_no_dhcp_enabled_networks(self): network = dhcp.NetModel(True, dict(id='net-id', tenant_id='aaaaaaaa-aaaa-aaaa-aaaaaaaaaaaa', admin_state_up=True, subnets=[], ports=[])) self.cache.get_network_by_id.return_value = network self.plugin.get_network_info.return_value = network with mock.patch.object(self.dhcp, 'disable_dhcp_helper') as disable: self.dhcp.refresh_dhcp_helper(network.id) disable.assert_called_once_with(network.id) self.assertFalse(self.cache.called) self.assertFalse(self.call_driver.called) self.cache.assert_has_calls( [mock.call.get_network_by_id('net-id')])
**** CubicPower OpenStack Study **** def test_refresh_dhcp_helper_exception_during_rpc(self): network = dhcp.NetModel(True, dict(id='net-id', tenant_id='aaaaaaaa-aaaa-aaaa-aaaaaaaaaaaa', admin_state_up=True, subnets=[], ports=[])) self.cache.get_network_by_id.return_value = network self.plugin.get_network_info.side_effect = Exception with mock.patch.object(dhcp_agent.LOG, 'exception') as log: self.dhcp.refresh_dhcp_helper(network.id) self.assertFalse(self.call_driver.called) self.cache.assert_has_calls( [mock.call.get_network_by_id('net-id')]) self.assertTrue(log.called) self.assertTrue(self.dhcp.needs_resync)
**** CubicPower OpenStack Study **** def test_subnet_update_end(self): payload = dict(subnet=dict(network_id=fake_network.id)) self.cache.get_network_by_id.return_value = fake_network self.plugin.get_network_info.return_value = fake_network self.dhcp.subnet_update_end(None, payload) self.cache.assert_has_calls([mock.call.put(fake_network)]) self.call_driver.assert_called_once_with('reload_allocations', fake_network)
**** CubicPower OpenStack Study **** def test_subnet_update_end_restart(self): new_state = dhcp.NetModel(True, dict(id=fake_network.id, tenant_id=fake_network.tenant_id, admin_state_up=True, subnets=[fake_subnet1, fake_subnet3], ports=[fake_port1])) payload = dict(subnet=dict(network_id=fake_network.id)) self.cache.get_network_by_id.return_value = fake_network self.plugin.get_network_info.return_value = new_state self.dhcp.subnet_update_end(None, payload) self.cache.assert_has_calls([mock.call.put(new_state)]) self.call_driver.assert_called_once_with('restart', new_state)
**** CubicPower OpenStack Study **** def test_subnet_update_end_delete_payload(self): prev_state = dhcp.NetModel(True, dict(id=fake_network.id, tenant_id=fake_network.tenant_id, admin_state_up=True, subnets=[fake_subnet1, fake_subnet3], ports=[fake_port1])) payload = dict(subnet_id=fake_subnet1.id) self.cache.get_network_by_subnet_id.return_value = prev_state self.cache.get_network_by_id.return_value = prev_state self.plugin.get_network_info.return_value = fake_network self.dhcp.subnet_delete_end(None, payload) self.cache.assert_has_calls([ mock.call.get_network_by_subnet_id( 'bbbbbbbb-bbbb-bbbb-bbbbbbbbbbbb'), mock.call.get_network_by_id('12345678-1234-5678-1234567890ab'), mock.call.put(fake_network)]) self.call_driver.assert_called_once_with('restart', fake_network)
**** CubicPower OpenStack Study **** def test_port_update_end(self): payload = dict(port=vars(fake_port2)) self.cache.get_network_by_id.return_value = fake_network self.cache.get_port_by_id.return_value = fake_port2 self.dhcp.port_update_end(None, payload) self.cache.assert_has_calls( [mock.call.get_network_by_id(fake_port2.network_id), mock.call.put_port(mock.ANY)]) self.call_driver.assert_called_once_with('reload_allocations', fake_network)
**** CubicPower OpenStack Study **** def test_port_update_change_ip_on_port(self): payload = dict(port=vars(fake_port1)) self.cache.get_network_by_id.return_value = fake_network updated_fake_port1 = copy.deepcopy(fake_port1) updated_fake_port1.fixed_ips[0].ip_address = '172.9.9.99' self.cache.get_port_by_id.return_value = updated_fake_port1 self.dhcp.port_update_end(None, payload) self.cache.assert_has_calls( [mock.call.get_network_by_id(fake_port1.network_id), mock.call.put_port(mock.ANY)]) self.call_driver.assert_has_calls( [mock.call.call_driver('reload_allocations', fake_network)])
**** CubicPower OpenStack Study **** def test_port_delete_end(self): payload = dict(port_id=fake_port2.id) self.cache.get_network_by_id.return_value = fake_network self.cache.get_port_by_id.return_value = fake_port2 self.dhcp.port_delete_end(None, payload) self.cache.assert_has_calls( [mock.call.get_port_by_id(fake_port2.id), mock.call.get_network_by_id(fake_network.id), mock.call.remove_port(fake_port2)]) self.call_driver.assert_has_calls( [mock.call.call_driver('reload_allocations', fake_network)])
**** CubicPower OpenStack Study **** def test_port_delete_end_unknown_port(self): payload = dict(port_id='unknown') self.cache.get_port_by_id.return_value = None self.dhcp.port_delete_end(None, payload) self.cache.assert_has_calls([mock.call.get_port_by_id('unknown')]) self.assertEqual(self.call_driver.call_count, 0)
**** CubicPower OpenStack Study **** class TestDhcpPluginApiProxy(base.BaseTestCase):
**** CubicPower OpenStack Study **** def setUp(self): super(TestDhcpPluginApiProxy, self).setUp() self.proxy = dhcp_agent.DhcpPluginApi('foo', {}, None) self.proxy.host = 'foo' self.call_p = mock.patch.object(self.proxy, 'call') self.call = self.call_p.start() self.make_msg_p = mock.patch.object(self.proxy, 'make_msg') self.make_msg = self.make_msg_p.start()
**** CubicPower OpenStack Study **** def tearDown(self): self.make_msg_p.stop() self.call_p.stop() super(TestDhcpPluginApiProxy, self).tearDown()
**** CubicPower OpenStack Study **** def test_get_network_info(self): self.call.return_value = dict(a=1) retval = self.proxy.get_network_info('netid') self.assertEqual(retval.a, 1) self.assertTrue(self.call.called) self.make_msg.assert_called_once_with('get_network_info', network_id='netid', host='foo')
**** CubicPower OpenStack Study **** def test_get_dhcp_port(self): self.call.return_value = dict(a=1) retval = self.proxy.get_dhcp_port('netid', 'devid') self.assertEqual(retval.a, 1) self.assertTrue(self.call.called) self.make_msg.assert_called_once_with('get_dhcp_port', network_id='netid', device_id='devid', host='foo')
**** CubicPower OpenStack Study **** def test_get_dhcp_port_none(self): self.call.return_value = None self.assertIsNone(self.proxy.get_dhcp_port('netid', 'devid'))
**** CubicPower OpenStack Study **** def test_get_active_networks_info(self): self.proxy.get_active_networks_info() self.make_msg.assert_called_once_with('get_active_networks_info', host='foo')
**** CubicPower OpenStack Study **** def test_create_dhcp_port(self): port_body = ( {'port': {'name': '', 'admin_state_up': True, 'network_id': fake_network.id, 'tenant_id': fake_network.tenant_id, 'fixed_ips': [{'subnet_id': fake_fixed_ip1.subnet_id}], 'device_id': mock.ANY}}) self.proxy.create_dhcp_port(port_body) self.make_msg.assert_called_once_with('create_dhcp_port', port=port_body, host='foo')
**** CubicPower OpenStack Study **** def test_create_dhcp_port_none(self): self.call.return_value = None port_body = ( {'port': {'name': '', 'admin_state_up': True, 'network_id': fake_network.id, 'tenant_id': fake_network.tenant_id, 'fixed_ips': [{'subnet_id': fake_fixed_ip1.subnet_id}], 'device_id': mock.ANY}}) self.assertIsNone(self.proxy.create_dhcp_port(port_body))
**** CubicPower OpenStack Study **** def test_update_dhcp_port_none(self): self.call.return_value = None port_body = {'port': {'fixed_ips': [{'subnet_id': fake_fixed_ip1.subnet_id}]}} self.assertIsNone(self.proxy.update_dhcp_port(fake_port1.id, port_body))
**** CubicPower OpenStack Study **** def test_update_dhcp_port(self): port_body = {'port': {'fixed_ips': [{'subnet_id': fake_fixed_ip1.subnet_id}]}} self.proxy.update_dhcp_port(fake_port1.id, port_body) self.make_msg.assert_called_once_with('update_dhcp_port', port_id=fake_port1.id, port=port_body, host='foo')
**** CubicPower OpenStack Study **** def test_release_dhcp_port(self): self.proxy.release_dhcp_port('netid', 'devid') self.assertTrue(self.call.called) self.make_msg.assert_called_once_with('release_dhcp_port', network_id='netid', device_id='devid', host='foo')
**** CubicPower OpenStack Study **** def test_release_port_fixed_ip(self): self.proxy.release_port_fixed_ip('netid', 'devid', 'subid') self.assertTrue(self.call.called) self.make_msg.assert_called_once_with('release_port_fixed_ip', network_id='netid', subnet_id='subid', device_id='devid', host='foo')
**** CubicPower OpenStack Study **** class TestNetworkCache(base.BaseTestCase):
**** CubicPower OpenStack Study **** def test_put_network(self): nc = dhcp_agent.NetworkCache() nc.put(fake_network) self.assertEqual(nc.cache, {fake_network.id: fake_network}) self.assertEqual(nc.subnet_lookup, {fake_subnet1.id: fake_network.id, fake_subnet2.id: fake_network.id}) self.assertEqual(nc.port_lookup, {fake_port1.id: fake_network.id})
**** CubicPower OpenStack Study **** def test_put_network_existing(self): prev_network_info = mock.Mock() nc = dhcp_agent.NetworkCache() with mock.patch.object(nc, 'remove') as remove: nc.cache[fake_network.id] = prev_network_info nc.put(fake_network) remove.assert_called_once_with(prev_network_info) self.assertEqual(nc.cache, {fake_network.id: fake_network}) self.assertEqual(nc.subnet_lookup, {fake_subnet1.id: fake_network.id, fake_subnet2.id: fake_network.id}) self.assertEqual(nc.port_lookup, {fake_port1.id: fake_network.id})
**** CubicPower OpenStack Study **** def test_remove_network(self): nc = dhcp_agent.NetworkCache() nc.cache = {fake_network.id: fake_network} nc.subnet_lookup = {fake_subnet1.id: fake_network.id, fake_subnet2.id: fake_network.id} nc.port_lookup = {fake_port1.id: fake_network.id} nc.remove(fake_network) self.assertEqual(len(nc.cache), 0) self.assertEqual(len(nc.subnet_lookup), 0) self.assertEqual(len(nc.port_lookup), 0)
**** CubicPower OpenStack Study **** def test_get_network_by_id(self): nc = dhcp_agent.NetworkCache() nc.put(fake_network) self.assertEqual(nc.get_network_by_id(fake_network.id), fake_network)
**** CubicPower OpenStack Study **** def test_get_network_ids(self): nc = dhcp_agent.NetworkCache() nc.put(fake_network) self.assertEqual(nc.get_network_ids(), [fake_network.id])
**** CubicPower OpenStack Study **** def test_get_network_by_subnet_id(self): nc = dhcp_agent.NetworkCache() nc.put(fake_network) self.assertEqual(nc.get_network_by_subnet_id(fake_subnet1.id), fake_network)
**** CubicPower OpenStack Study **** def test_get_network_by_port_id(self): nc = dhcp_agent.NetworkCache() nc.put(fake_network) self.assertEqual(nc.get_network_by_port_id(fake_port1.id), fake_network)
**** CubicPower OpenStack Study **** def test_put_port(self): fake_net = dhcp.NetModel( True, dict(id='12345678-1234-5678-1234567890ab', tenant_id='aaaaaaaa-aaaa-aaaa-aaaaaaaaaaaa', subnets=[fake_subnet1], ports=[fake_port1])) nc = dhcp_agent.NetworkCache() nc.put(fake_net) nc.put_port(fake_port2) self.assertEqual(len(nc.port_lookup), 2) self.assertIn(fake_port2, fake_net.ports)
**** CubicPower OpenStack Study **** def test_put_port_existing(self): fake_net = dhcp.NetModel( True, dict(id='12345678-1234-5678-1234567890ab', tenant_id='aaaaaaaa-aaaa-aaaa-aaaaaaaaaaaa', subnets=[fake_subnet1], ports=[fake_port1, fake_port2])) nc = dhcp_agent.NetworkCache() nc.put(fake_net) nc.put_port(fake_port2) self.assertEqual(len(nc.port_lookup), 2) self.assertIn(fake_port2, fake_net.ports)
**** CubicPower OpenStack Study **** def test_remove_port_existing(self): fake_net = dhcp.NetModel( True, dict(id='12345678-1234-5678-1234567890ab', tenant_id='aaaaaaaa-aaaa-aaaa-aaaaaaaaaaaa', subnets=[fake_subnet1], ports=[fake_port1, fake_port2])) nc = dhcp_agent.NetworkCache() nc.put(fake_net) nc.remove_port(fake_port2) self.assertEqual(len(nc.port_lookup), 1) self.assertNotIn(fake_port2, fake_net.ports)
**** CubicPower OpenStack Study **** def test_get_port_by_id(self): nc = dhcp_agent.NetworkCache() nc.put(fake_network) self.assertEqual(nc.get_port_by_id(fake_port1.id), fake_port1)
**** CubicPower OpenStack Study **** class FakeV4Network: id = 'aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa' subnets = [FakeV4Subnet()] ports = [FakePort1()]
**** CubicPower OpenStack Study **** class FakeV4NetworkNoGateway: id = 'cccccccc-cccc-cccc-cccc-cccccccccccc' subnets = [FakeV4SubnetNoGateway()] ports = [FakePort1()]
**** CubicPower OpenStack Study **** class TestDeviceManager(base.BaseTestCase):
**** CubicPower OpenStack Study **** def setUp(self): super(TestDeviceManager, self).setUp() config.register_interface_driver_opts_helper(cfg.CONF) config.register_use_namespaces_opts_helper(cfg.CONF) cfg.CONF.register_opts(dhcp_agent.DhcpAgent.OPTS) cfg.CONF.register_opts(dhcp.OPTS) cfg.CONF.set_override('interface_driver', 'neutron.agent.linux.interface.NullDriver') config.register_root_helper(cfg.CONF) cfg.CONF.set_override('use_namespaces', True) cfg.CONF.set_override('enable_isolated_metadata', True) self.device_exists_p = mock.patch( 'neutron.agent.linux.ip_lib.device_exists') self.device_exists = self.device_exists_p.start() self.dvr_cls_p = mock.patch('neutron.agent.linux.interface.NullDriver') self.iproute_cls_p = mock.patch('neutron.agent.linux.' 'ip_lib.IpRouteCommand') driver_cls = self.dvr_cls_p.start() iproute_cls = self.iproute_cls_p.start() self.mock_driver = mock.MagicMock() self.mock_driver.DEV_NAME_LEN = ( interface.LinuxInterfaceDriver.DEV_NAME_LEN) self.mock_iproute = mock.MagicMock() driver_cls.return_value = self.mock_driver iproute_cls.return_value = self.mock_iproute
**** CubicPower OpenStack Study **** def tearDown(self): self.dvr_cls_p.stop() self.device_exists_p.stop() self.iproute_cls_p.stop() super(TestDeviceManager, self).tearDown()
**** CubicPower OpenStack Study **** def _test_setup_helper(self, device_exists, reuse_existing=False, net=None, port=None): net = net or fake_network port = port or fake_port1 plugin = mock.Mock() plugin.create_dhcp_port.return_value = port or fake_port1 plugin.get_dhcp_port.return_value = port or fake_port1 self.device_exists.return_value = device_exists self.mock_driver.get_device_name.return_value = 'tap12345678-12' dh = dhcp.DeviceManager(cfg.CONF, cfg.CONF.root_helper, plugin) dh._set_default_route = mock.Mock() interface_name = dh.setup(net, reuse_existing) self.assertEqual(interface_name, 'tap12345678-12') plugin.assert_has_calls([ mock.call.create_dhcp_port( {'port': {'name': '', 'admin_state_up': True, 'network_id': net.id, 'tenant_id': net.tenant_id, 'fixed_ips': [{'subnet_id': fake_fixed_ip1.subnet_id}], 'device_id': mock.ANY}})]) expected_ips = ['172.9.9.9/24', '169.254.169.254/16'] expected = [ mock.call.get_device_name(port), mock.call.init_l3( 'tap12345678-12', expected_ips, namespace=net.namespace)] if not reuse_existing: expected.insert(1, mock.call.plug(net.id, port.id, 'tap12345678-12', 'aa:bb:cc:dd:ee:ff', namespace=net.namespace)) self.mock_driver.assert_has_calls(expected) dh._set_default_route.assert_called_once_with(net, port)
**** CubicPower OpenStack Study **** def test_setup(self): cfg.CONF.set_override('enable_metadata_network', False) self._test_setup_helper(False) cfg.CONF.set_override('enable_metadata_network', True) self._test_setup_helper(False)
**** CubicPower OpenStack Study **** def test_setup_device_exists(self): with testtools.ExpectedException(exceptions.PreexistingDeviceFailure): self._test_setup_helper(True)
**** CubicPower OpenStack Study **** def test_setup_device_exists_reuse(self): self._test_setup_helper(True, True)
**** CubicPower OpenStack Study **** def test_create_dhcp_port_raise_conflict(self): plugin = mock.Mock() dh = dhcp.DeviceManager(cfg.CONF, cfg.CONF.root_helper, plugin) plugin.create_dhcp_port.return_value = None self.assertRaises(exceptions.Conflict, dh.setup_dhcp_port, fake_network)
**** CubicPower OpenStack Study **** def test_create_dhcp_port_create_new(self): plugin = mock.Mock() dh = dhcp.DeviceManager(cfg.CONF, cfg.CONF.root_helper, plugin) plugin.create_dhcp_port.return_value = fake_network.ports[0] dh.setup_dhcp_port(fake_network) plugin.assert_has_calls([ mock.call.create_dhcp_port( {'port': {'name': '', 'admin_state_up': True, 'network_id': fake_network.id, 'tenant_id': fake_network.tenant_id, 'fixed_ips': [{'subnet_id': fake_fixed_ip1.subnet_id}], 'device_id': mock.ANY}})])
**** CubicPower OpenStack Study **** def test_create_dhcp_port_update_add_subnet(self): plugin = mock.Mock() dh = dhcp.DeviceManager(cfg.CONF, cfg.CONF.root_helper, plugin) fake_network_copy = copy.deepcopy(fake_network) fake_network_copy.ports[0].device_id = dh.get_device_id(fake_network) fake_network_copy.subnets[1].enable_dhcp = True plugin.update_dhcp_port.return_value = fake_network.ports[0] dh.setup_dhcp_port(fake_network_copy) port_body = {'port': { 'fixed_ips': [{'subnet_id': fake_fixed_ip1.subnet_id, 'ip_address': fake_fixed_ip1.ip_address}, {'subnet_id': fake_subnet2.id}]}} plugin.assert_has_calls([ mock.call.update_dhcp_port(fake_network_copy.ports[0].id, port_body)])
**** CubicPower OpenStack Study **** def test_update_dhcp_port_raises_conflict(self): plugin = mock.Mock() dh = dhcp.DeviceManager(cfg.CONF, cfg.CONF.root_helper, plugin) fake_network_copy = copy.deepcopy(fake_network) fake_network_copy.ports[0].device_id = dh.get_device_id(fake_network) fake_network_copy.subnets[1].enable_dhcp = True plugin.update_dhcp_port.return_value = None self.assertRaises(exceptions.Conflict, dh.setup_dhcp_port, fake_network_copy)
**** CubicPower OpenStack Study **** def test_create_dhcp_port_no_update_or_create(self): plugin = mock.Mock() dh = dhcp.DeviceManager(cfg.CONF, cfg.CONF.root_helper, plugin) fake_network_copy = copy.deepcopy(fake_network) fake_network_copy.ports[0].device_id = dh.get_device_id(fake_network) dh.setup_dhcp_port(fake_network_copy) self.assertFalse(plugin.setup_dhcp_port.called) self.assertFalse(plugin.update_dhcp_port.called)
**** CubicPower OpenStack Study **** def test_destroy(self): fake_net = dhcp.NetModel( True, dict(id='12345678-1234-5678-1234567890ab', tenant_id='aaaaaaaa-aaaa-aaaa-aaaaaaaaaaaa')) fake_port = dhcp.DictModel( dict(id='12345678-1234-aaaa-1234567890ab', mac_address='aa:bb:cc:dd:ee:ff')) with mock.patch('neutron.agent.linux.interface.NullDriver') as dvr_cls: mock_driver = mock.MagicMock() mock_driver.get_device_name.return_value = 'tap12345678-12' dvr_cls.return_value = mock_driver plugin = mock.Mock() plugin.get_dhcp_port.return_value = fake_port dh = dhcp.DeviceManager(cfg.CONF, cfg.CONF.root_helper, plugin) dh.destroy(fake_net, 'tap12345678-12') dvr_cls.assert_called_once_with(cfg.CONF) mock_driver.assert_has_calls( [mock.call.unplug('tap12345678-12', namespace='qdhcp-' + fake_net.id)]) plugin.assert_has_calls( [mock.call.release_dhcp_port(fake_net.id, mock.ANY)])
**** CubicPower OpenStack Study **** def test_get_interface_name(self): fake_net = dhcp.NetModel( True, dict(id='12345678-1234-5678-1234567890ab', tenant_id='aaaaaaaa-aaaa-aaaa-aaaaaaaaaaaa')) fake_port = dhcp.DictModel( dict(id='12345678-1234-aaaa-1234567890ab', mac_address='aa:bb:cc:dd:ee:ff')) with mock.patch('neutron.agent.linux.interface.NullDriver') as dvr_cls: mock_driver = mock.MagicMock() mock_driver.get_device_name.return_value = 'tap12345678-12' dvr_cls.return_value = mock_driver plugin = mock.Mock() plugin.get_dhcp_port.return_value = fake_port dh = dhcp.DeviceManager(cfg.CONF, cfg.CONF.root_helper, plugin) dh.get_interface_name(fake_net, fake_port) dvr_cls.assert_called_once_with(cfg.CONF) mock_driver.assert_has_calls( [mock.call.get_device_name(fake_port)]) self.assertEqual(len(plugin.mock_calls), 0)
**** CubicPower OpenStack Study **** def test_get_device_id(self): fake_net = dhcp.NetModel( True, dict(id='12345678-1234-5678-1234567890ab', tenant_id='aaaaaaaa-aaaa-aaaa-aaaaaaaaaaaa')) expected = ('dhcp1ae5f96c-c527-5079-82ea-371a01645457-12345678-1234-' '5678-1234567890ab') with mock.patch('socket.gethostname') as get_host: with mock.patch('uuid.uuid5') as uuid5: uuid5.return_value = '1ae5f96c-c527-5079-82ea-371a01645457' get_host.return_value = 'localhost' dh = dhcp.DeviceManager(cfg.CONF, cfg.CONF.root_helper, None) self.assertEqual(dh.get_device_id(fake_net), expected) uuid5.assert_called_once_with(uuid.NAMESPACE_DNS, 'localhost')
**** CubicPower OpenStack Study **** def _get_device_manager_with_mock_device(self, conf, device): dh = dhcp.DeviceManager(conf, cfg.CONF.root_helper, None) dh._get_device = mock.Mock(return_value=device) return dh
**** CubicPower OpenStack Study **** def test_update(self): # Try with namespaces and no metadata network cfg.CONF.set_override('use_namespaces', True) cfg.CONF.set_override('enable_metadata_network', False) dh = dhcp.DeviceManager(cfg.CONF, cfg.CONF.root_helper, None) dh._set_default_route = mock.Mock() network = mock.Mock() port = mock.Mock() dh.plugin = mock.Mock() dh.plugin.get_dhcp_port.return_value = port dh.update(network) dh._set_default_route.assert_called_once_with(network, port) # No namespaces, shouldn't set default route. cfg.CONF.set_override('use_namespaces', False) cfg.CONF.set_override('enable_metadata_network', False) dh = dhcp.DeviceManager(cfg.CONF, cfg.CONF.root_helper, None) dh._set_default_route = mock.Mock() dh.plugin = mock.Mock() dh.plugin.get_dhcp_port.return_value = port dh.update(FakeV4Network()) self.assertFalse(dh._set_default_route.called) # Meta data network enabled, don't interfere with its gateway. cfg.CONF.set_override('use_namespaces', True) cfg.CONF.set_override('enable_metadata_network', True) dh = dhcp.DeviceManager(cfg.CONF, cfg.CONF.root_helper, None) dh._set_default_route = mock.Mock() dh.plugin = mock.Mock() dh.plugin.get_dhcp_port.return_value = port dh.update(FakeV4Network()) self.assertTrue(dh._set_default_route.called) # For completeness cfg.CONF.set_override('use_namespaces', False) cfg.CONF.set_override('enable_metadata_network', True) dh = dhcp.DeviceManager(cfg.CONF, cfg.CONF.root_helper, None) dh._set_default_route = mock.Mock() dh.update(FakeV4Network()) self.assertFalse(dh._set_default_route.called)
**** CubicPower OpenStack Study **** def test_set_default_route(self): device = mock.Mock() device.route.get_gateway.return_value = None # Basic one subnet with gateway. dh = self._get_device_manager_with_mock_device(cfg.CONF, device) network = FakeV4Network() dh._set_default_route(network, mock.Mock()) device.route.get_gateway.assert_called_once() self.assertFalse(device.route.delete_gateway.called) device.route.add_gateway.assert_called_once_with('192.168.0.1')
**** CubicPower OpenStack Study **** def test_set_default_route_no_subnet(self): device = mock.Mock() device.route.get_gateway.return_value = None # Try a namespace but no subnet. dh = self._get_device_manager_with_mock_device(cfg.CONF, device) network = FakeV4NetworkNoSubnet() dh._set_default_route(network, mock.Mock()) device.route.get_gateway.assert_called_once() self.assertFalse(device.route.delete_gateway.called) self.assertFalse(device.route.add_gateway.called)
**** CubicPower OpenStack Study **** def test_set_default_route_no_subnet_delete_gateway(self): device = mock.Mock() device.route.get_gateway.return_value = dict(gateway='192.168.0.1') # Try a namespace but no subnet where a gateway needs to be deleted. dh = self._get_device_manager_with_mock_device(cfg.CONF, device) network = FakeV4NetworkNoSubnet() dh._set_default_route(network, mock.Mock()) device.route.get_gateway.assert_called_once() device.route.delete_gateway.assert_called_once_with('192.168.0.1') self.assertFalse(device.route.add_gateway.called)
**** CubicPower OpenStack Study **** def test_set_default_route_no_gateway(self): device = mock.Mock() device.route.get_gateway.return_value = dict(gateway='192.168.0.1') # Try a subnet with no gateway dh = self._get_device_manager_with_mock_device(cfg.CONF, device) network = FakeV4NetworkNoGateway() dh._set_default_route(network, mock.Mock()) device.route.get_gateway.assert_called_once() device.route.delete_gateway.assert_called_once_with('192.168.0.1') self.assertFalse(device.route.add_gateway.called)
**** CubicPower OpenStack Study **** def test_set_default_route_do_nothing(self): device = mock.Mock() device.route.get_gateway.return_value = dict(gateway='192.168.0.1') # Try a subnet where the gateway doesn't change. Should do nothing. dh = self._get_device_manager_with_mock_device(cfg.CONF, device) network = FakeV4Network() dh._set_default_route(network, mock.Mock()) device.route.get_gateway.assert_called_once() self.assertFalse(device.route.delete_gateway.called) self.assertFalse(device.route.add_gateway.called)
**** CubicPower OpenStack Study **** def test_set_default_route_change_gateway(self): device = mock.Mock() device.route.get_gateway.return_value = dict(gateway='192.168.0.2') # Try a subnet with a gateway this is different than the current. dh = self._get_device_manager_with_mock_device(cfg.CONF, device) network = FakeV4Network() dh._set_default_route(network, mock.Mock()) device.route.get_gateway.assert_called_once() self.assertFalse(device.route.delete_gateway.called) device.route.add_gateway.assert_called_once_with('192.168.0.1')
**** CubicPower OpenStack Study **** def test_set_default_route_two_subnets(self): device = mock.Mock() device.route.get_gateway.return_value = None # Try two subnets. Should set gateway from the first. dh = self._get_device_manager_with_mock_device(cfg.CONF, device) network = FakeV4Network() subnet2 = FakeV4Subnet() subnet2.gateway_ip = '192.168.1.1' network.subnets = [subnet2, FakeV4Subnet()] dh._set_default_route(network, mock.Mock()) device.route.get_gateway.assert_called_once() self.assertFalse(device.route.delete_gateway.called) device.route.add_gateway.assert_called_once_with('192.168.1.1')
**** CubicPower OpenStack Study **** def test_get_device_dont_call_get_dhcp_port(self): dh = dhcp.DeviceManager(cfg.CONF, cfg.CONF.root_helper, None) dh.get_interface_name = mock.Mock() dh.plugin = mock.Mock() network = mock.Mock() port = mock.Mock() dh._get_device(network, port) self.assertFalse(dh.plugin.get_dhcp_port.called) dh.get_interface_name.assert_called_once_with(network, port)
**** CubicPower OpenStack Study **** class TestDictModel(base.BaseTestCase):
**** CubicPower OpenStack Study **** def test_basic_dict(self): d = dict(a=1, b=2) m = dhcp.DictModel(d) self.assertEqual(m.a, 1) self.assertEqual(m.b, 2)
**** CubicPower OpenStack Study **** def test_dict_has_sub_dict(self): d = dict(a=dict(b=2)) m = dhcp.DictModel(d) self.assertEqual(m.a.b, 2)
**** CubicPower OpenStack Study **** def test_dict_contains_list(self): d = dict(a=[1, 2]) m = dhcp.DictModel(d) self.assertEqual(m.a, [1, 2])
**** CubicPower OpenStack Study **** def test_dict_contains_list_of_dicts(self): d = dict(a=[dict(b=2), dict(c=3)]) m = dhcp.DictModel(d) self.assertEqual(m.a[0].b, 2) self.assertEqual(m.a[1].c, 3)
**** CubicPower OpenStack Study **** class TestNetModel(base.BaseTestCase):
**** CubicPower OpenStack Study **** def test_ns_name(self): network = dhcp.NetModel(True, {'id': 'foo'}) self.assertEqual(network.namespace, 'qdhcp-foo')
**** CubicPower OpenStack Study **** def test_ns_name_false_namespace(self): network = dhcp.NetModel(False, {'id': 'foo'}) self.assertIsNone(network.namespace)
**** CubicPower OpenStack Study **** def test_ns_name_none_namespace(self): network = dhcp.NetModel(None, {'id': 'foo'}) self.assertIsNone(network.namespace)
|