# 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 collections
import copy
import sys
import uuid

import eventlet
import mock
from neutron_lib import constants as const
from neutron_lib import exceptions
from oslo_config import cfg
import oslo_messaging
import testtools

from neutron.agent.common import config
from neutron.agent.dhcp import agent as dhcp_agent
from neutron.agent import dhcp_agent as entry
from neutron.agent.linux import dhcp
from neutron.agent.linux import interface
from neutron.common import config as common_config
from neutron.common import constants as n_const
from neutron.common import utils
from neutron.conf.agent import dhcp as dhcp_config
from neutron.tests import base


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


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,
                              ipv6_ra_mode=None, ipv6_address_mode=None,
                              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_ipv6_subnet = dhcp.DictModel(dict(id='bbbbbbbb-1111-2222-bbbbbbbbbbbb',
                              network_id='12345678-1234-5678-1234567890ab',
                              cidr='2001:0db8::0/64', enable_dhcp=True,
                              tenant_id=fake_tenant_id,
                              gateway_ip='2001:0db8::1', ip_version=6,
                              ipv6_ra_mode='slaac', ipv6_address_mode=None))

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_fixed_ip_subnet2 = dhcp.DictModel(dict(id='', subnet_id=fake_subnet2.id,
                                ip_address='172.9.8.9'))
fake_fixed_ip2 = dhcp.DictModel(dict(id='', subnet_id=fake_subnet1.id,
                                ip_address='172.9.9.10'))
fake_fixed_ipv6 = dhcp.DictModel(dict(id='', subnet_id=fake_ipv6_subnet.id,
                                 ip_address='2001:db8::a8bb:ccff:fedd:ee99'))
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',
                            device_owner='',
                            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_dhcp_port = dhcp.DictModel(dict(id='12345678-1234-aaaa-123456789022',
                            device_id='dhcp-12345678-1234-aaaa-123456789022',
                            device_owner=const.DEVICE_OWNER_DHCP,
                            allocation_pools=fake_subnet1_allocation_pools,
                            mac_address='aa:bb:cc:dd:ee:22',
                            network_id='12345678-1234-5678-1234567890ab',
                            fixed_ips=[fake_fixed_ip2]))

fake_port2 = dhcp.DictModel(dict(id='12345678-1234-aaaa-123456789000',
                            device_id='dhcp-12345678-1234-aaaa-123456789000',
                            device_owner='',
                            mac_address='aa:bb:cc:dd:ee:99',
                            network_id='12345678-1234-5678-1234567890ab',
                            revision_number=77,
                            fixed_ips=[fake_fixed_ip2]))

fake_ipv6_port = dhcp.DictModel(dict(id='12345678-1234-aaaa-123456789000',
                                device_owner='',
                                mac_address='aa:bb:cc:dd:ee:99',
                                network_id='12345678-1234-5678-1234567890ab',
                                fixed_ips=[fake_fixed_ipv6]))

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_meta_dvr_port = dhcp.DictModel(fake_meta_port.copy())
fake_meta_dvr_port.device_owner = const.DEVICE_OWNER_DVR_INTERFACE

fake_dist_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_DVR_INTERFACE,
                                device_id='forzanapoli',
                                fixed_ips=[fake_meta_fixed_ip]))

FAKE_NETWORK_UUID = '12345678-1234-5678-1234567890ab'
FAKE_NETWORK_DHCP_NS = "qdhcp-%s" % FAKE_NETWORK_UUID

fake_network = dhcp.NetModel(dict(id=FAKE_NETWORK_UUID,
                             tenant_id='aaaaaaaa-aaaa-aaaa-aaaaaaaaaaaa',
                             admin_state_up=True,
                             subnets=[fake_subnet1, fake_subnet2],
                             ports=[fake_port1]))

fake_network_ipv6 = dhcp.NetModel(dict(
                             id='12345678-1234-5678-1234567890ab',
                             tenant_id='aaaaaaaa-aaaa-aaaa-aaaaaaaaaaaa',
                             admin_state_up=True,
                             subnets=[fake_ipv6_subnet],
                             ports=[fake_ipv6_port]))

fake_network_ipv6_ipv4 = dhcp.NetModel(dict(
                             id='12345678-1234-5678-1234567890ab',
                             tenant_id='aaaaaaaa-aaaa-aaaa-aaaaaaaaaaaa',
                             admin_state_up=True,
                             subnets=[fake_ipv6_subnet, fake_subnet1],
                             ports=[fake_port1]))

isolated_network = dhcp.NetModel(
    dict(
        id='12345678-1234-5678-1234567890ab',
        tenant_id='aaaaaaaa-aaaa-aaaa-aaaaaaaaaaaa',
        admin_state_up=True,
        subnets=[fake_subnet1],
        ports=[fake_port1]))

nonisolated_dist_network = dhcp.NetModel(
    dict(
        id='12345678-1234-5678-1234567890ab',
        tenant_id='aaaaaaaa-aaaa-aaaa-aaaaaaaaaaaa',
        admin_state_up=True,
        subnets=[fake_subnet1],
        ports=[fake_port1, fake_port2]))

empty_network = dhcp.NetModel(
    dict(
        id='12345678-1234-5678-1234567890ab',
        tenant_id='aaaaaaaa-aaaa-aaaa-aaaaaaaaaaaa',
        admin_state_up=True,
        subnets=[fake_subnet1],
        ports=[]))

fake_meta_network = dhcp.NetModel(
    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_meta_dvr_network = dhcp.NetModel(fake_meta_network.copy())
fake_meta_dvr_network.ports = [fake_meta_dvr_port]

fake_dist_network = dhcp.NetModel(
    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_dist_port]))

fake_down_network = dhcp.NetModel(
    dict(id='12345678-dddd-dddd-1234567890ab',
         tenant_id='aaaaaaaa-aaaa-aaaa-aaaaaaaaaaaa',
         admin_state_up=False,
         subnets=[],
         ports=[]))


class TestDhcpAgent(base.BaseTestCase):
    def setUp(self):
        super(TestDhcpAgent, self).setUp()
        entry.register_options(cfg.CONF)
        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()

        self.mock_ip_wrapper_p = mock.patch("neutron.agent.linux.ip_lib."
                                            "IPWrapper")
        self.mock_ip_wrapper = self.mock_ip_wrapper_p.start()

    def test_init_host(self):
        dhcp = dhcp_agent.DhcpAgent(HOSTNAME)
        with mock.patch.object(dhcp, 'sync_state') as sync_state:
            dhcp.init_host()
            sync_state.assert_called_once_with()

    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')
        mock_start_ready = mock.patch.object(
            dhcp_agent.DhcpAgentWithStateReport, 'start_ready_ports_loop',
            autospec=True).start()
        with mock.patch.object(dhcp_agent.DhcpAgentWithStateReport,
                               'sync_state',
                               autospec=True) as mock_sync_state:
            with mock.patch.object(dhcp_agent.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',
                            base.etcdir('neutron.conf')]
                        cfg.CONF.register_opts(dhcp_config.DHCP_AGENT_OPTS)
                        config.register_interface_driver_opts_helper(cfg.CONF)
                        config.register_agent_state_opts_helper(cfg.CONF)
                        cfg.CONF.register_opts(interface.OPTS)
                        common_config.init(sys.argv[1:])
                        agent_mgr = dhcp_agent.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)
                        mock_start_ready.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)])

    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',
                  'start_ready_ports_loop']])
            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()
                mocks['start_ready_ports_loop'].assert_called_once_with()

    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,
                                            mock.ANY,
                                            mock.ANY,
                                            mock.ANY)

    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
        dhcp = dhcp_agent.DhcpAgent(HOSTNAME)
        with mock.patch.object(dhcp,
                               'schedule_resync') as schedule_resync:
            self.assertIsNone(dhcp.call_driver('foo', network))
            self.driver.assert_called_once_with(cfg.CONF,
                                                mock.ANY,
                                                mock.ANY,
                                                mock.ANY,
                                                mock.ANY)
            self.assertEqual(expected_sync, schedule_resync.called)

    def test_call_driver_ip_address_generation_failure(self):
        error = oslo_messaging.RemoteError(
            exc_type='IpAddressGenerationFailure')
        self._test_call_driver_failure(exc=error, expected_sync=False)

    def test_call_driver_failure(self):
        self._test_call_driver_failure()

    def test_call_driver_remote_error_net_not_found(self):
        self._test_call_driver_failure(
            exc=oslo_messaging.RemoteError(exc_type='NetworkNotFound'),
            trace_level='warning')

    def test_call_driver_network_not_found(self):
        self._test_call_driver_failure(
            exc=exceptions.NetworkNotFound(net_id='1'),
            trace_level='warning')

    def test_call_driver_conflict(self):
        self._test_call_driver_failure(
            exc=exceptions.Conflict(),
            trace_level='warning',
            expected_sync=False)

    def _test_sync_state_helper(self, known_net_ids, active_net_ids):
        active_networks = set(mock.Mock(id=netid) for netid in active_net_ids)

        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 ['disable_dhcp_helper', 'cache',
                                           'safe_configure_dhcp_for_network']])

            with mock.patch.multiple(dhcp, **attrs_to_mock) as mocks:
                mocks['cache'].get_network_ids.return_value = known_net_ids
                mocks['cache'].get_port_ids.return_value = range(4)
                dhcp.sync_state()

                diff = set(known_net_ids) - set(active_net_ids)
                exp_disable = [mock.call(net_id) for net_id in diff]
                mocks['cache'].assert_has_calls([mock.call.get_network_ids()])
                mocks['disable_dhcp_helper'].assert_has_calls(exp_disable)
                self.assertEqual(set(range(4)), dhcp.dhcp_ready_ports)

    def test_sync_state_initial(self):
        self._test_sync_state_helper([], ['a'])

    def test_sync_state_same(self):
        self._test_sync_state_helper(['a'], ['a'])

    def test_sync_state_disabled_net(self):
        self._test_sync_state_helper(['b'], ['a'])

    def test_sync_state_waitall(self):
        with mock.patch.object(dhcp_agent.eventlet.GreenPool, 'waitall') as w:
            active_net_ids = ['1', '2', '3', '4', '5']
            known_net_ids = ['1', '2', '3', '4', '5']
            self._test_sync_state_helper(known_net_ids, active_net_ids)
            w.assert_called_once_with()

    def test_sync_state_for_all_networks_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)
                with mock.patch.object(dhcp,
                                       'schedule_resync') as schedule_resync:
                    dhcp.sync_state()

                    self.assertTrue(log.called)
                    self.assertTrue(schedule_resync.called)

    def test_sync_state_for_one_network_plugin_error(self):
        with mock.patch(DHCP_PLUGIN) as plug:
            mock_plugin = mock.Mock()
            exc = Exception()
            mock_plugin.get_active_networks_info.side_effect = exc
            plug.return_value = mock_plugin

            with mock.patch.object(dhcp_agent.LOG, 'exception') as log:
                dhcp = dhcp_agent.DhcpAgent(HOSTNAME)
                with mock.patch.object(dhcp,
                                       'schedule_resync') as schedule_resync:
                    dhcp.sync_state(['foo_network'])

                    self.assertTrue(log.called)
                    schedule_resync.assert_called_with(exc, 'foo_network')

    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)

    def test_start_ready_ports_loop(self):
        dhcp = dhcp_agent.DhcpAgent(HOSTNAME)
        with mock.patch.object(dhcp_agent.eventlet, 'spawn') as spawn:
            dhcp.start_ready_ports_loop()
            spawn.assert_called_once_with(dhcp._dhcp_ready_ports_loop)

    def test__dhcp_ready_ports_doesnt_log_exception_on_timeout(self):
        dhcp = dhcp_agent.DhcpAgent(HOSTNAME)
        dhcp.dhcp_ready_ports = set(range(4))

        with mock.patch.object(dhcp.plugin_rpc, 'dhcp_ready_on_ports',
                               side_effect=oslo_messaging.MessagingTimeout):
            # exit after 2 iterations
            with mock.patch.object(dhcp_agent.eventlet, 'sleep',
                                   side_effect=[0, 0, RuntimeError]):
                with mock.patch.object(dhcp_agent.LOG, 'exception') as lex:
                    with testtools.ExpectedException(RuntimeError):
                        dhcp._dhcp_ready_ports_loop()
        self.assertFalse(lex.called)

    def test__dhcp_ready_ports_disables_on_incompatible_server(self):
        dhcp = dhcp_agent.DhcpAgent(HOSTNAME)
        dhcp.agent_state = dict(configurations=dict(notifies_port_ready=True))
        dhcp.dhcp_ready_ports = set(range(4))

        side_effect = oslo_messaging.RemoteError(exc_type='NoSuchMethod')
        with mock.patch.object(dhcp.plugin_rpc, 'dhcp_ready_on_ports',
                               side_effect=side_effect):
            with mock.patch.object(dhcp_agent.eventlet, 'sleep',
                                   side_effect=[None, RuntimeError]) as sleep:
                with testtools.ExpectedException(RuntimeError):
                    dhcp._dhcp_ready_ports_loop()
                # should have slept for 5 minutes
                sleep.assert_called_with(300)

    def test__dhcp_ready_ports_loop(self):
        dhcp = dhcp_agent.DhcpAgent(HOSTNAME)
        dhcp.dhcp_ready_ports = set(range(4))

        with mock.patch.object(dhcp.plugin_rpc, 'dhcp_ready_on_ports',
                               side_effect=[RuntimeError, 0]) as ready:
            # exit after 2 iterations
            with mock.patch.object(dhcp_agent.eventlet, 'sleep',
                                   side_effect=[0, 0, RuntimeError]):
                with testtools.ExpectedException(RuntimeError):
                    dhcp._dhcp_ready_ports_loop()
        # should have been called with all ports again after the failure
        ready.assert_has_calls([mock.call(set(range(4)))] * 2)

    def test_dhcp_ready_ports_updates_after_enable_dhcp(self):
        dhcp = dhcp_agent.DhcpAgent(HOSTNAME)
        self.assertEqual(set(), dhcp.dhcp_ready_ports)
        dhcp.configure_dhcp_for_network(fake_network)
        self.assertEqual({fake_port1.id}, dhcp.dhcp_ready_ports)

    def test_report_state_revival_logic(self):
        dhcp = dhcp_agent.DhcpAgentWithStateReport(HOSTNAME)
        with mock.patch.object(dhcp.state_rpc,
                               'report_state') as report_state,\
            mock.patch.object(dhcp, "run"):
            report_state.return_value = n_const.AGENT_ALIVE
            dhcp._report_state()
            self.assertEqual({}, dhcp.needs_resync_reasons)

            report_state.return_value = n_const.AGENT_REVIVED
            dhcp._report_state()
            self.assertEqual(dhcp.needs_resync_reasons[None],
                             ['Agent has just been revived'])

    def test_periodic_resync_helper(self):
        with mock.patch.object(dhcp_agent.eventlet, 'sleep') as sleep:
            dhcp = dhcp_agent.DhcpAgent(HOSTNAME)
            resync_reasons = collections.OrderedDict(
                (('a', 'reason1'), ('b', 'reason2')))
            dhcp.needs_resync_reasons = resync_reasons
            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(resync_reasons.keys())
                sleep.assert_called_once_with(dhcp.conf.resync_interval)
                self.assertEqual(0, len(dhcp.needs_resync_reasons))

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

            self.assertFalse(dhcp.cache.get_network_ids())
            self.assertTrue(log.called)

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

        self.assertEqual(set(networks), set(dhcp.cache.get_network_ids()))

    def test_none_interface_driver(self):
        cfg.CONF.set_override('interface_driver', None)
        self.assertRaises(SystemExit, dhcp.DeviceManager,
                          cfg.CONF, None)

    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.bar')
        self.assertRaises(SystemExit, dhcp.DeviceManager,
                          cfg.CONF, None)


class TestLogArgs(base.BaseTestCase):

    def test_log_args_without_log_dir_and_file(self):
        conf_dict = {'debug': True,
                     '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)

    def test_log_args_without_log_file(self):
        conf_dict = {'debug': True,
                     'log_dir': '/etc/tests',
                     'log_file': None,
                     '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']
        args = config.get_log_args(conf, 'log_file_name')
        self.assertEqual(expected_args, args)

    def test_log_args_with_log_dir_and_file(self):
        conf_dict = {'debug': True,
                     '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)

    def test_log_args_without_log_dir(self):
        conf_dict = {'debug': True,
                     '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)

    def test_log_args_with_filelog_and_syslog(self):
        conf_dict = {'debug': 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',
                         '--log-file=log_file_name',
                         '--log-dir=/etc/tests/tests']
        args = config.get_log_args(conf, 'log_file_name')
        self.assertEqual(expected_args, args)


class TestDhcpAgentEventHandler(base.BaseTestCase):
    def setUp(self):
        super(TestDhcpAgentEventHandler, self).setUp()
        config.register_interface_driver_opts_helper(cfg.CONF)
        cfg.CONF.set_override('interface_driver',
                              'neutron.agent.linux.interface.NullDriver')
        entry.register_options(cfg.CONF)  # register all dhcp cfg options

        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()
        self.cache.is_port_message_stale.return_value = False
        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()
        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.schedule_resync_p = mock.patch.object(self.dhcp,
                                                   'schedule_resync')
        self.schedule_resync = self.schedule_resync_p.start()
        self.external_process_p = mock.patch(
            'neutron.agent.linux.external_process.ProcessManager'
        )
        self.external_process = self.external_process_p.start()

    def _process_manager_constructor_call(self, ns=FAKE_NETWORK_DHCP_NS):
        return mock.call(conf=cfg.CONF,
                         uuid=FAKE_NETWORK_UUID,
                         namespace=ns,
                         default_cmd_callback=mock.ANY)

    def _enable_dhcp_helper(self, network, enable_isolated_metadata=False,
                            is_isolated_network=False):
        self.dhcp._process_monitor = mock.Mock()
        if enable_isolated_metadata:
            cfg.CONF.set_override('enable_isolated_metadata', True)
        self.plugin.get_network_info.return_value = network
        self.dhcp.enable_dhcp_helper(network.id)
        self.plugin.assert_has_calls([
            mock.call.get_network_info(network.id)])
        self.call_driver.assert_called_once_with('enable', network)
        self.cache.assert_has_calls([mock.call.put(network)])
        if is_isolated_network and enable_isolated_metadata:
            self.external_process.assert_has_calls([
                self._process_manager_constructor_call(),
                mock.call().enable()])
        elif not enable_isolated_metadata:
            self.external_process.assert_has_calls([
                self._process_manager_constructor_call(ns=None),
                mock.call().disable()])
        else:
            self.assertFalse(self.external_process.call_count)

    def test_enable_dhcp_helper_enable_metadata_isolated_network(self):
        self._enable_dhcp_helper(isolated_network,
                                 enable_isolated_metadata=True,
                                 is_isolated_network=True)

    def test_enable_dhcp_helper_enable_metadata_no_gateway(self):
        isolated_network_no_gateway = copy.deepcopy(isolated_network)
        isolated_network_no_gateway.subnets[0].gateway_ip = None

        self._enable_dhcp_helper(isolated_network_no_gateway,
                                 enable_isolated_metadata=True,
                                 is_isolated_network=True)

    def test_enable_dhcp_helper_enable_metadata_nonisolated_network(self):
        nonisolated_network = copy.deepcopy(isolated_network)
        nonisolated_network.ports[0].device_owner = (
            const.DEVICE_OWNER_ROUTER_INTF)
        nonisolated_network.ports[0].fixed_ips[0].ip_address = '172.9.9.1'

        self._enable_dhcp_helper(nonisolated_network,
                                 enable_isolated_metadata=True,
                                 is_isolated_network=False)

    def test_enable_dhcp_helper_enable_metadata_nonisolated_dist_network(self):
        nonisolated_dist_network.ports[0].device_owner = (
            const.DEVICE_OWNER_ROUTER_INTF)
        nonisolated_dist_network.ports[0].fixed_ips[0].ip_address = '172.9.9.1'
        nonisolated_dist_network.ports[1].device_owner = (
            const.DEVICE_OWNER_DVR_INTERFACE)
        nonisolated_dist_network.ports[1].fixed_ips[0].ip_address = '172.9.9.1'

        self._enable_dhcp_helper(nonisolated_dist_network,
                                 enable_isolated_metadata=True,
                                 is_isolated_network=False)

    def test_enable_dhcp_helper_enable_metadata_empty_network(self):
        self._enable_dhcp_helper(empty_network,
                                 enable_isolated_metadata=True,
                                 is_isolated_network=True)

    def test_enable_dhcp_helper_enable_metadata_ipv6_ipv4_network(self):
        self._enable_dhcp_helper(fake_network_ipv6_ipv4,
                                 enable_isolated_metadata=True,
                                 is_isolated_network=True)

    def test_enable_dhcp_helper_driver_failure_ipv6_ipv4_network(self):
        self.plugin.get_network_info.return_value = fake_network_ipv6_ipv4
        self.call_driver.return_value = False
        cfg.CONF.set_override('enable_isolated_metadata', True)
        with mock.patch.object(
            self.dhcp, 'enable_isolated_metadata_proxy') as enable_metadata:
            self.dhcp.enable_dhcp_helper(fake_network_ipv6_ipv4.id)
            self.plugin.assert_has_calls(
                [mock.call.get_network_info(fake_network_ipv6_ipv4.id)])
            self.call_driver.assert_called_once_with('enable',
                                                     fake_network_ipv6_ipv4)
            self.assertFalse(self.cache.called)
            self.assertFalse(enable_metadata.called)
            self.assertFalse(self.external_process.called)

    def test_enable_dhcp_helper(self):
        self._enable_dhcp_helper(fake_network)

    def test_enable_dhcp_helper_ipv6_network(self):
        self._enable_dhcp_helper(fake_network_ipv6)

    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)

    def test_enable_dhcp_helper_network_none(self):
        self.plugin.get_network_info.return_value = None
        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.assertFalse(self.dhcp.schedule_resync.called)

    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.schedule_resync.called)
            self.assertFalse(self.cache.called)
            self.assertFalse(self.external_process.called)

    def test_enable_dhcp_helper_driver_failure(self):
        self.plugin.get_network_info.return_value = fake_network
        self.call_driver.return_value = False
        cfg.CONF.set_override('enable_isolated_metadata', True)
        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)

    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([
                self._process_manager_constructor_call(ns=None),
                mock.call().disable()])
        else:
            self.assertFalse(self.external_process.call_count)

    def test_disable_dhcp_helper_known_network_isolated_metadata(self):
        self._disable_dhcp_helper_known_network(isolated_metadata=True)

    def test_disable_dhcp_helper_known_network(self):
        self._disable_dhcp_helper_known_network()

    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)

    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([
                self._process_manager_constructor_call(ns=None),
                mock.call().disable()
            ])
        else:
            self.assertFalse(self.external_process.call_count)

    def test_disable_dhcp_helper_driver_failure_isolated_metadata(self):
        self._disable_dhcp_helper_driver_failure(isolated_metadata=True)