# Copyright (c) 2014 Red Hat, Inc.
# 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 functools

import mock
import netaddr
from neutron_lib import constants as l3_constants
from oslo_config import cfg
from oslo_log import log as logging
from oslo_utils import uuidutils
import testtools

from neutron.agent.common import config as agent_config
from neutron.agent.common import ovs_lib
from neutron.agent.l3 import agent as neutron_l3_agent
from neutron.agent import l3_agent as l3_agent_main
from neutron.agent.linux import external_process
from neutron.agent.linux import ip_lib
from neutron.agent.linux import utils
from neutron.common import config as common_config
from neutron.common import constants as n_const
from neutron.common import utils as common_utils
from neutron.tests.common import l3_test_common
from neutron.tests.common import net_helpers
from neutron.tests.functional import base


_uuid = uuidutils.generate_uuid


def get_ovs_bridge(br_name):
    return ovs_lib.OVSBridge(br_name)


class L3AgentTestFramework(base.BaseSudoTestCase):
    def setUp(self):
        super(L3AgentTestFramework, self).setUp()
        self.mock_plugin_api = mock.patch(
            'neutron.agent.l3.agent.L3PluginApi').start().return_value
        mock.patch('neutron.agent.rpc.PluginReportStateAPI').start()
        self.conf = self._configure_agent('agent1')
        self.agent = neutron_l3_agent.L3NATAgentWithStateReport('agent1',
                                                                self.conf)

    def _get_config_opts(self):
        config = cfg.ConfigOpts()
        config.register_opts(common_config.core_opts)
        config.register_opts(common_config.core_cli_opts)
        logging.register_options(config)
        agent_config.register_process_monitor_opts(config)
        return config

    def _configure_agent(self, host, agent_mode='dvr_snat'):
        conf = self._get_config_opts()
        l3_agent_main.register_opts(conf)
        conf.set_override(
            'interface_driver',
            'neutron.agent.linux.interface.OVSInterfaceDriver')

        br_int = self.useFixture(net_helpers.OVSBridgeFixture()).bridge
        br_ex = self.useFixture(net_helpers.OVSBridgeFixture()).bridge
        conf.set_override('ovs_integration_bridge', br_int.br_name)
        conf.set_override('external_network_bridge', br_ex.br_name)

        temp_dir = self.get_new_temp_dir()
        get_temp_file_path = functools.partial(self.get_temp_file_path,
                                               root=temp_dir)
        conf.set_override('state_path', temp_dir.path)
        # NOTE(cbrandily): log_file or log_dir must be set otherwise
        # metadata_proxy_watch_log has no effect
        conf.set_override('log_file',
                          get_temp_file_path('log_file'))
        conf.set_override('metadata_proxy_socket',
                          get_temp_file_path('metadata_proxy'))
        conf.set_override('ha_confs_path',
                          get_temp_file_path('ha_confs'))
        conf.set_override('external_pids',
                          get_temp_file_path('external/pids'))
        conf.set_override('host', host)
        conf.set_override('agent_mode', agent_mode)

        return conf

    def _get_agent_ovs_integration_bridge(self, agent):
        return get_ovs_bridge(agent.conf.ovs_integration_bridge)

    def generate_router_info(self, enable_ha, ip_version=4, extra_routes=True,
                             enable_fip=True, enable_snat=True,
                             num_internal_ports=1,
                             dual_stack=False, v6_ext_gw_with_sub=True):
        if ip_version == 6 and not dual_stack:
            enable_snat = False
            enable_fip = False
            extra_routes = False

        return l3_test_common.prepare_router_data(ip_version=ip_version,
                                                 enable_snat=enable_snat,
                                                 num_internal_ports=(
                                                     num_internal_ports),
                                                 enable_floating_ip=enable_fip,
                                                 enable_ha=enable_ha,
                                                 extra_routes=extra_routes,
                                                 dual_stack=dual_stack,
                                                 v6_ext_gw_with_sub=(
                                                     v6_ext_gw_with_sub))

    def _test_conntrack_disassociate_fip(self, ha):
        '''Test that conntrack immediately drops stateful connection
           that uses floating IP once it's disassociated.
        '''
        router_info = self.generate_router_info(enable_ha=ha)
        router = self.manage_router(self.agent, router_info)

        port = net_helpers.get_free_namespace_port(l3_constants.PROTO_NAME_TCP,
                                                   router.ns_name)
        client_address = '19.4.4.3'
        server_address = '35.4.0.4'

        def clean_fips(router):
            router.router[l3_constants.FLOATINGIP_KEY] = []

        clean_fips(router)
        self._add_fip(router, client_address, fixed_address=server_address)
        router.process(self.agent)

        router_ns = ip_lib.IPWrapper(namespace=router.ns_name)
        netcat = net_helpers.NetcatTester(
            router.ns_name, router.ns_name, client_address, port,
            protocol=net_helpers.NetcatTester.TCP)
        self.addCleanup(netcat.stop_processes)

        def assert_num_of_conntrack_rules(n):
            out = router_ns.netns.execute(["conntrack", "-L",
                                           "--orig-src", client_address])
            self.assertEqual(
                n, len([line for line in out.strip().split('\n') if line]))

        if ha:
            utils.wait_until_true(lambda: router.ha_state == 'master')

        with self.assert_max_execution_time(100):
            assert_num_of_conntrack_rules(0)

            self.assertTrue(netcat.test_connectivity())
            assert_num_of_conntrack_rules(1)

            clean_fips(router)
            router.process(self.agent)
            assert_num_of_conntrack_rules(0)

            with testtools.ExpectedException(RuntimeError):
                netcat.test_connectivity()

    def _test_update_floatingip_statuses(self, router_info):
        router = self.manage_router(self.agent, router_info)
        rpc = self.agent.plugin_rpc.update_floatingip_statuses
        self.assertTrue(rpc.called)

        # Assert that every defined FIP is updated via RPC
        expected_fips = set([
            (fip['id'], l3_constants.FLOATINGIP_STATUS_ACTIVE) for fip in
            router.router[l3_constants.FLOATINGIP_KEY]])
        call = [args[0] for args in rpc.call_args_list][0]
        actual_fips = set(
            [(fip_id, status) for fip_id, status in call[2].items()])
        self.assertEqual(expected_fips, actual_fips)

    def _gateway_check(self, gateway_ip, external_device):
        expected_gateway = gateway_ip
        ip_vers = netaddr.IPAddress(expected_gateway).version
        existing_gateway = (external_device.route.get_gateway(
            ip_version=ip_vers).get('gateway'))
        self.assertEqual(expected_gateway, existing_gateway)

    def _assert_ha_device(self, router):
        def ha_router_dev_name_getter(not_used):
            return router.get_ha_device_name()
        self.assertTrue(self.device_exists_with_ips_and_mac(
            router.router[l3_constants.HA_INTERFACE_KEY],
            ha_router_dev_name_getter, router.ns_name))

    def _assert_gateway(self, router, v6_ext_gw_with_sub=True):
        external_port = router.get_ex_gw_port()
        external_device_name = router.get_external_device_name(
            external_port['id'])
        external_device = ip_lib.IPDevice(external_device_name,