# vim: tabstop=4 shiftwidth=4 softtabstop=4

# Copyright 2012 OpenStack LLC
# 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 unittest

import mock
from sqlalchemy.ext import sqlsoup

from quantum.agent import dhcp_agent
from quantum.agent.common import config
from quantum.agent.linux import interface


class FakeModel:
    def __init__(self, id_, **kwargs):
        self.id = id_
        self.__dict__.update(kwargs)

    def __str__(self):
        return str(self.__dict__)


class FakePortModel(FakeModel):
    fixed_ips = []


class FakeFixedIPModel(object):

    def __init__(self, ip_address, cidr):
        self.subnet = FakeSubnetModel(cidr)
        self.ip_address = ip_address


class FakeSubnetModel(object):

    def __init__(self, cidr):
        self.cidr = cidr


class TestDhcpAgent(unittest.TestCase):
    def setUp(self):
        self.conf = config.setup_conf()
        self.conf.register_opts(dhcp_agent.DhcpAgent.OPTS)
        self.driver_cls_p = mock.patch(
            'quantum.agent.dhcp_agent.importutils.import_class')
        self.driver = mock.Mock(name='driver')
        self.driver_cls = self.driver_cls_p.start()
        self.driver_cls.return_value = self.driver
        self.dhcp = dhcp_agent.DhcpAgent(self.conf)
        self.dhcp.polling_interval = 0

    def tearDown(self):
        self.driver_cls_p.stop()

    def test_dhcp_agent_main(self):
        with mock.patch('quantum.agent.dhcp_agent.DeviceManager') as dev_mgr:
            with mock.patch('quantum.agent.dhcp_agent.DhcpAgent') as dhcp:
                with mock.patch('quantum.agent.dhcp_agent.sys') as mock_sys:
                    mock_sys.argv = []
                    dhcp_agent.main()
                    dev_mgr.assert_called_once(mock.ANY, 'sudo')
                    dhcp.assert_has_calls([
                        mock.call(mock.ANY),
                        mock.call().daemon_loop()])

    def test_daemon_loop_survives_get_network_state_delta_failure(self):
        def stop_loop(*args):
            self.dhcp._run = False
            return None

        with mock.patch.object(self.dhcp, 'get_network_state_delta') as state:
            state.side_effect = stop_loop
            self.dhcp.daemon_loop()

    def test_daemon_loop_completes_single_pass(self):
        self.dhcp._network_dhcp_enable = mock.Mock(return_value=True)
        with mock.patch.object(self.dhcp, 'get_network_state_delta') as state:
            with mock.patch.object(self.dhcp, 'call_driver') as call_driver:
                with mock.patch('quantum.agent.dhcp_agent.time') as time:
                    time.sleep = mock.Mock(side_effect=RuntimeError('stop'))
                    state.return_value = dict(new=['new_net'],
                                              updated=['updated_net'],
                                              deleted=['deleted_net'])

                    self.assertRaises(RuntimeError, self.dhcp.daemon_loop)
                    call_driver.assert_has_calls(
                        [mock.call('enable', 'new_net'),
                         mock.call('reload_allocations', 'updated_net'),
                         mock.call('disable', 'deleted_net')])

    def test_state_builder_network_admin_down(self):
        fake_network1 = FakeModel(1, admin_state_up=True)
        fake_network2 = FakeModel(2, admin_state_up=False)

        fake_subnet1 = FakeModel(1, network_id=1, enable_dhcp=True)
        fake_subnet2 = FakeModel(2, network_id=2, enable_dhcp=True)
        fake_subnet3 = FakeModel(3, network_id=2, enable_dhcp=True)

        fake_network1.subnets = [fake_subnet1]
        fake_network2.subnets = [fake_subnet2, fake_subnet3]

        fake_subnet1.network = fake_network1
        fake_subnet2.network = fake_network2
        fake_subnet3.network = fake_network2

        fake_allocation = [
            FakeModel(2, subnet_id=1),
            FakeModel(3, subnet_id=2)
        ]

        fake_subnets = [fake_subnet1, fake_subnet2, fake_subnet3]
        fake_networks = [fake_network1, fake_network2]

        db = mock.Mock()
        db.subnets.all = mock.Mock(return_value=fake_subnets)
        db.networks.all = mock.Mock(return_value=fake_networks)
        db.ipallocations.all = mock.Mock(return_value=fake_allocation)
        self.dhcp.db = db
        state = self.dhcp._state_builder()

        self.assertEquals(state.networks, set([1]))

        expected_subnets = set([
            (hash(str(fake_subnets[0])), 1),
        ])
        self.assertEquals(state.subnet_hashes, expected_subnets)

        expected_ipalloc = set([
            (hash(str(fake_allocation[0])), 1),
        ])
        self.assertEquals(state.ipalloc_hashes, expected_ipalloc)

    def test_state_builder_network_dhcp_partial_disable(self):
        fake_network1 = FakeModel(1, admin_state_up=True)
        fake_network2 = FakeModel(2, admin_state_up=True)

        fake_subnet1 = FakeModel(1, network_id=1, enable_dhcp=True)
        fake_subnet2 = FakeModel(2, network_id=2, enable_dhcp=False)
        fake_subnet3 = FakeModel(3, network_id=2, enable_dhcp=True)

        fake_network1.subnets = [fake_subnet1]
        fake_network2.subnets = [fake_subnet2, fake_subnet3]

        fake_subnet1.network = fake_network1
        fake_subnet2.network = fake_network2
        fake_subnet3.network = fake_network2

        fake_allocation = [
            FakeModel(2, subnet_id=1),
            FakeModel(3, subnet_id=2),
            FakeModel(4, subnet_id=3),
        ]

        fake_subnets = [fake_subnet1, fake_subnet2, fake_subnet3]
        fake_networks = [fake_network1, fake_network2]

        db = mock.Mock()
        db.subnets.all = mock.Mock(return_value=fake_subnets)
        db.networks.all = mock.Mock(return_value=fake_networks)
        db.ipallocations.all = mock.Mock(return_value=fake_allocation)
        self.dhcp.db = db
        state = self.dhcp._state_builder()

        self.assertEquals(state.networks, set([1, 2]))

        expected_subnets = set([
            (hash(str(fake_subnets[0])), 1),
            (hash(str(fake_subnets[2])), 2),
        ])
        self.assertEquals(state.subnet_hashes, expected_subnets)

        expected_ipalloc = set([
            (hash(str(fake_allocation[0])), 1),
            (hash(str(fake_allocation[2])), 2),
        ])
        self.assertEquals(state.ipalloc_hashes, expected_ipalloc)

    def test_state_builder_network_dhcp_all_disable(self):
        fake_network1 = FakeModel(1, admin_state_up=True)
        fake_network2 = FakeModel(2, admin_state_up=True)

        fake_subnet1 = FakeModel(1, network_id=1, enable_dhcp=True)
        fake_subnet2 = FakeModel(2, network_id=2, enable_dhcp=False)
        fake_subnet3 = FakeModel(3, network_id=2, enable_dhcp=False)

        fake_network1.subnets = [fake_subnet1]
        fake_network2.subnets = [fake_subnet2, fake_subnet3]

        fake_subnet1.network = fake_network1
        fake_subnet2.network = fake_network2
        fake_subnet3.network = fake_network2

        fake_allocation = [
            FakeModel(2, subnet_id=1),
            FakeModel(3, subnet_id=2),
            FakeModel(4, subnet_id=3),
        ]

        fake_subnets = [fake_subnet1, fake_subnet2, fake_subnet3]
        fake_networks = [fake_network1, fake_network2]

        db = mock.Mock()
        db.subnets.all = mock.Mock(return_value=fake_subnets)
        db.networks.all = mock.Mock(return_value=fake_networks)
        db.ipallocations.all = mock.Mock(return_value=fake_allocation)
        self.dhcp.db = db
        state = self.dhcp._state_builder()

        self.assertEquals(state.networks, set([1]))

        expected_subnets = set([
            (hash(str(fake_subnets[0])), 1)
        ])
        self.assertEquals(state.subnet_hashes, expected_subnets)

        expected_ipalloc = set([
            (hash(str(fake_allocation[0])), 1)
        ])
        self.assertEquals(state.ipalloc_hashes, expected_ipalloc)

    def test_state_builder_mixed(self):
        fake_network1 = FakeModel(1, admin_state_up=True)
        fake_network2 = FakeModel(2, admin_state_up=True)
        fake_network3 = FakeModel(3, admin_state_up=False)

        fake_subnet1 = FakeModel(1, network_id=1, enable_dhcp=True)
        fake_subnet2 = FakeModel(2, network_id=2, enable_dhcp=False)
        fake_subnet3 = FakeModel(3, network_id=3, enable_dhcp=True)

        fake_network1.subnets = [fake_subnet1]
        fake_network2.subnets = [fake_subnet2]
        fake_network3.subnets = [fake_subnet3]

        fake_subnet1.network = fake_network1
        fake_subnet2.network = fake_network2
        fake_subnet3.network = fake_network3

        fake_allocation = [
            FakeModel(2, subnet_id=1)
        ]

        fake_subnets = [fake_subnet1, fake_subnet2, fake_subnet3]
        fake_networks = [fake_network1, fake_network2, fake_network3]

        db = mock.Mock()
        db.subnets.all = mock.Mock(return_value=fake_subnets)
        db.networks.all = mock.Mock(return_value=fake_networks)
        db.ipallocations.all = mock.Mock(return_value=fake_allocation)
        self.dhcp.db = db
        state = self.dhcp._state_builder()

        self.assertEquals(state.networks, set([1]))

        expected_subnets = set([
            (hash(str(fake_subnets[0])), 1),
        ])
        self.assertEquals(state.subnet_hashes, expected_subnets)

        expected_ipalloc = set([
            (hash(str(fake_allocation[0])), 1),
        ])
        self.assertEquals(state.ipalloc_hashes, expected_ipalloc)

    def _network_state_helper(self, before, after):
        with mock.patch.object(self.dhcp, '_state_builder') as state_builder:
            state_builder.return_value = after
            self.dhcp.prev_state = before
            return self.dhcp.get_network_state_delta()

    def test_get_network_state_fresh(self):
        new_state = dhcp_agent.State(set([1]), set([(3, 1)]),
                                     set([(11, 1)]))

        delta = self._network_state_helper(self.dhcp.prev_state, new_state)
        self.assertEqual(delta,
                         dict(new=set([1]), deleted=set(), updated=set()))

    def test_get_network_state_new_subnet_on_known_network(self):
        prev_state = dhcp_agent.State(set([1]), set([(3, 1)]), set([(11, 1)]))
        new_state = dhcp_agent.State(set([1]),
                                     set([(3, 1), (4, 1)]),
                                     set([(11, 1)]))

        delta = self._network_state_helper(prev_state, new_state)
        self.assertEqual(delta,
                         dict(new=set(), deleted=set(), updated=set([1])))

    def test_get_network_state_new_ipallocation(self):
        prev_state = dhcp_agent.State(set([1]),
                                      set([(3, 1)]),
                                      set([(11, 1)]))
        new_state = dhcp_agent.State(set([1]),
                                     set([(3, 1)]),
                                     set([(11, 1), (12, 1)]))

        delta = self._network_state_helper(prev_state, new_state)
        self.assertEqual(delta,
                         dict(new=set(), deleted=set(), updated=set([1])))

    def test_get_network_state_delete_subnet_on_known_network(self):
        prev_state = dhcp_agent.State(set([1]),
                                      set([(3, 1), (4, 1)]),
                                      set([(11, 1)]))
        new_state = dhcp_agent.State(set([1]),
                                     set([(3, 1)]),
                                     set([(11, 1)]))

        delta = self._network_state_helper(prev_state, new_state)
        self.assertEqual(delta,
                         dict(new=set(), deleted=set(), updated=set([1])))

    def test_get_network_state_deleted_ipallocation(self):
        prev_state = dhcp_agent.State(set([1]),
                                      set([(3, 1)]),
                                      set([(11, 1), (12, 1)]))
        new_state = dhcp_agent.State(set([1]),
                                     set([(3, 1)]),
                                     set([(11, 1)]))

        delta = self._network_state_helper(prev_state, new_state)
        self.assertEqual(delta,
                         dict(new=set(), deleted=set(), updated=set([1])))

    def test_get_network_state_deleted_network(self):
        prev_state = dhcp_agent.State(set([1]),
                                      set([(3, 1)]),
                                      set([(11, 1), (12, 1)]))
        new_state = dhcp_agent.State(set(), set(), set())

        delta = self._network_state_helper(prev_state, new_state)
        self.assertEqual(delta,
                         dict(new=set(), deleted=set([1]), updated=set()))

    def test_get_network_state_changed_subnet_and_deleted_network(self):
        prev_state = dhcp_agent.State(set([1, 2]),
                                      set([(3, 1), (2, 2)]),
                                      set([(11, 1), (12, 1)]))
        new_state = dhcp_agent.State(set([1]),
                                     set([(4, 1)]),
                                     set([(11, 1), (12, 1)]))

        delta = self._network_state_helper(prev_state, new_state)
        self.assertEqual(delta,
                         dict(new=set(), deleted=set([2]), updated=set([1])))

    def test_call_driver(self):
        with mock.patch.object(self.dhcp, 'db') as db:
            db.networks = mock.Mock()
            db.networks.filter_by = mock.Mock(
                return_value=mock.Mock(return_value=FakeModel('1')))
            with mock.patch.object(dhcp_agent, 'DeviceManager') as dev_mgr:
                self.dhcp.call_driver('foo', '1')
                dev_mgr.assert_called()
                self.driver.assert_called_once_with(self.conf,
                                                    mock.ANY,
                                                    'sudo',
                                                    mock.ANY)


class TestDeviceManager(unittest.TestCase):
    def setUp(self):
        self.conf = config.setup_conf()
        self.conf.register_opts(dhcp_agent.DeviceManager.OPTS)
        self.conf.set_override('interface_driver',
                               'quantum.agent.linux.interface.NullDriver')
        self.conf.root_helper = 'sudo'
        self.conf.use_namespaces = True

        self.client_cls_p = mock.patch('quantumclient.v2_0.client.Client')
        client_cls = self.client_cls_p.start()
        self.client_inst = mock.Mock()
        client_cls.return_value = self.client_inst

        self.device_exists_p = mock.patch(
            'quantum.agent.linux.ip_lib.device_exists')
        self.device_exists = self.device_exists_p.start()

        self.dvr_cls_p = mock.patch('quantum.agent.linux.interface.NullDriver')
        driver_cls = self.dvr_cls_p.start()
        self.mock_driver = mock.MagicMock()
        self.mock_driver.DEV_NAME_LEN = (
            interface.LinuxInterfaceDriver.DEV_NAME_LEN)
        driver_cls.return_value = self.mock_driver

    def tearDown(self):
        self.dvr_cls_p.stop()
        self.device_exists_p.stop()
        self.client_cls_p.stop()

    def test_setup(self):
        port_id = '12345678-1234-aaaa-1234567890ab'
        network_id = '12345678-1234-5678-1234567890ab'
        fake_subnets = [FakeModel('12345678-aaaa-aaaa-1234567890ab'),
                        FakeModel('12345678-bbbb-bbbb-1234567890ab')]

        fake_network = FakeModel(network_id,
                                 tenant_id='aaaaaaaa-aaaa-aaaa-aaaaaaaaaaaa',
                                 subnets=fake_subnets)

        fake_port = FakePortModel(port_id, mac_address='aa:bb:cc:dd:ee:ff',
                                  network_id=network_id,
                                  allocations=[])
        fake_port.fixed_ips.append(FakeFixedIPModel('172.9.9.9',
                                                    '172.9.9.0/24'))
        port_dict = dict(mac_address='aa:bb:cc:dd:ee:ff',
                         allocations=[], id=1)

        self.client_inst.create_port.return_value = dict(port=port_dict)
        self.device_exists.return_value = False

        # fake the db
        filter_by_result = mock.Mock()
        filter_by_result.one = mock.Mock(return_value=fake_port)

        self.filter_called = False

        def get_filter_results(*args, **kwargs):
            if self.filter_called:
                return filter_by_result
            else:
                self.filter_called = True
                raise sqlsoup.SQLAlchemyError()

        mock_db = mock.Mock()
        mock_db.ports = mock.Mock(name='ports2')
        mock_db.ports.filter_by = mock.Mock(
            name='filter_by',
            side_effect=get_filter_results)

        self.mock_driver.get_device_name.return_value = 'tap12345678-12'

        dh = dhcp_agent.DeviceManager(self.conf, mock_db)
        dh.setup(fake_network)

        self.client_inst.assert_has_calls([
            mock.call.create_port(mock.ANY)])

        self.mock_driver.assert_has_calls([
            mock.call.get_device_name(mock.ANY),
            mock.call.plug(network_id,
                           port_id,
                           'tap12345678-12',
                           'aa:bb:cc:dd:ee:ff',
                           namespace=network_id),
            mock.call.init_l3('tap12345678-12', ['172.9.9.9/24'],
                              namespace=network_id)]
        )

    def test_destroy(self):
        fake_subnets = [FakeModel('12345678-aaaa-aaaa-1234567890ab'),
                        FakeModel('12345678-bbbb-bbbb-1234567890ab')]

        fake_network = FakeModel('12345678-1234-5678-1234567890ab',
                                 tenant_id='aaaaaaaa-aaaa-aaaa-aaaaaaaaaaaa',
                                 subnets=fake_subnets)

        fake_port = FakeModel('12345678-1234-aaaa-1234567890ab',
                              mac_address='aa:bb:cc:dd:ee:ff')

        port_dict = dict(mac_address='aa:bb:cc:dd:ee:ff', allocations=[], id=1)

        self.client_inst.create_port.return_value = dict(port=port_dict)
        self.device_exists.return_value = False

        # fake the db
        filter_by_result = mock.Mock()
        filter_by_result.one = mock.Mock(return_value=fake_port)

        self.filter_called = False

        def get_filter_results(*args, **kwargs):
            if self.filter_called:
                return filter_by_result
            else:
                self.filter_called = True
                raise sqlsoup.SQLAlchemyError()

        mock_db = mock.Mock()
        mock_db.ports = mock.Mock(name='ports2')
        mock_db.ports.filter_by = mock.Mock(
            name='filter_by',
            side_effect=get_filter_results)

        with mock.patch('quantum.agent.linux.interface.NullDriver') as dvr_cls:
            mock_driver = mock.MagicMock()
            mock_driver.DEV_NAME_LEN = (
                interface.LinuxInterfaceDriver.DEV_NAME_LEN)
            mock_driver.port = fake_port
            mock_driver.get_device_name.return_value = 'tap12345678-12'
            dvr_cls.return_value = mock_driver

            dh = dhcp_agent.DeviceManager(self.conf, mock_db)
            dh.destroy(fake_network)

            dvr_cls.assert_called_once_with(self.conf)
            mock_driver.assert_has_calls(
                [mock.call.get_device_name(mock.ANY),
                 mock.call.unplug('tap12345678-12')])


class TestAugmentingWrapper(unittest.TestCase):
    def test_simple_wrap(self):
        net = mock.Mock()
        db = mock.Mock()
        net.name = 'foo'
        wrapped = dhcp_agent.AugmentingWrapper(net, db)
        self.assertEqual(wrapped.name, 'foo')
        self.assertEqual(repr(net), repr(wrapped))
