# Copyright 2013 Big Switch Networks, 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 contextlib

import mock
from webob import exc

from neutron import context
from neutron.extensions import firewall
from neutron.plugins.common import constants as const
from neutron.services.firewall import fwaas_plugin
from neutron.tests import base
from neutron.tests.unit.db.firewall import test_db_firewall


FW_PLUGIN_KLASS = (
    "neutron.services.firewall.fwaas_plugin.FirewallPlugin"
)


class TestFirewallCallbacks(test_db_firewall.FirewallPluginDbTestCase):

    def setUp(self):
        super(TestFirewallCallbacks,
              self).setUp(fw_plugin=FW_PLUGIN_KLASS)
        self.callbacks = self.plugin.endpoints[0]

    def test_set_firewall_status(self):
        ctx = context.get_admin_context()
        with self.firewall_policy() as fwp:
            fwp_id = fwp['firewall_policy']['id']
            with self.firewall(firewall_policy_id=fwp_id,
                               admin_state_up=
                               test_db_firewall.ADMIN_STATE_UP) as fw:
                fw_id = fw['firewall']['id']
                res = self.callbacks.set_firewall_status(ctx, fw_id,
                                                         const.ACTIVE,
                                                         host='dummy')
                fw_db = self.plugin.get_firewall(ctx, fw_id)
                self.assertEqual(fw_db['status'], const.ACTIVE)
                self.assertTrue(res)
                res = self.callbacks.set_firewall_status(ctx, fw_id,
                                                         const.ERROR)
                fw_db = self.plugin.get_firewall(ctx, fw_id)
                self.assertEqual(fw_db['status'], const.ERROR)
                self.assertFalse(res)

    def test_set_firewall_status_pending_delete(self):
        ctx = context.get_admin_context()
        with self.firewall_policy() as fwp:
            fwp_id = fwp['firewall_policy']['id']
            with self.firewall(firewall_policy_id=fwp_id,
                               admin_state_up=
                               test_db_firewall.ADMIN_STATE_UP) as fw:
                fw_id = fw['firewall']['id']
                fw_db = self.plugin._get_firewall(ctx, fw_id)
                fw_db['status'] = const.PENDING_DELETE
                ctx.session.flush()
                res = self.callbacks.set_firewall_status(ctx, fw_id,
                                                         const.ACTIVE,
                                                         host='dummy')
                fw_db = self.plugin.get_firewall(ctx, fw_id)
                self.assertEqual(fw_db['status'], const.PENDING_DELETE)
                self.assertFalse(res)

    def test_firewall_deleted(self):
        ctx = context.get_admin_context()
        with self.firewall_policy() as fwp:
            fwp_id = fwp['firewall_policy']['id']
            with self.firewall(firewall_policy_id=fwp_id,
                               admin_state_up=test_db_firewall.ADMIN_STATE_UP,
                               do_delete=False) as fw:
                fw_id = fw['firewall']['id']
                with ctx.session.begin(subtransactions=True):
                    fw_db = self.plugin._get_firewall(ctx, fw_id)
                    fw_db['status'] = const.PENDING_DELETE
                    ctx.session.flush()
                    res = self.callbacks.firewall_deleted(ctx, fw_id,
                                                          host='dummy')
                    self.assertTrue(res)
                    self.assertRaises(firewall.FirewallNotFound,
                                      self.plugin.get_firewall,
                                      ctx, fw_id)

    def test_firewall_deleted_error(self):
        ctx = context.get_admin_context()
        with self.firewall_policy() as fwp:
            fwp_id = fwp['firewall_policy']['id']
            with self.firewall(
                firewall_policy_id=fwp_id,
                admin_state_up=test_db_firewall.ADMIN_STATE_UP,
            ) as fw:
                fw_id = fw['firewall']['id']
                res = self.callbacks.firewall_deleted(ctx, fw_id,
                                                      host='dummy')
                self.assertFalse(res)
                fw_db = self.plugin._get_firewall(ctx, fw_id)
                self.assertEqual(fw_db['status'], const.ERROR)

    def test_get_firewall_for_tenant(self):
        tenant_id = 'test-tenant'
        ctx = context.Context('', tenant_id)
        with contextlib.nested(self.firewall_rule(name='fwr1',
                                                  tenant_id=tenant_id),
                               self.firewall_rule(name='fwr2',
                                                  tenant_id=tenant_id),
                               self.firewall_rule(name='fwr3',
                                                  tenant_id=tenant_id)
                               ) as fr:
            with self.firewall_policy(tenant_id=tenant_id) as fwp:
                fwp_id = fwp['firewall_policy']['id']
                fw_rule_ids = [r['firewall_rule']['id'] for r in fr]
                data = {'firewall_policy':
                        {'firewall_rules': fw_rule_ids}}
                req = self.new_update_request('firewall_policies', data,
                                              fwp_id)
                res = req.get_response(self.ext_api)
                attrs = self._get_test_firewall_attrs()
                attrs['firewall_policy_id'] = fwp_id
                with self.firewall(firewall_policy_id=fwp_id,
                                   tenant_id=tenant_id,
                                   admin_state_up=
                                   test_db_firewall.ADMIN_STATE_UP) as fw:
                    fw_id = fw['firewall']['id']
                    res = self.callbacks.get_firewalls_for_tenant(ctx,
                                                                  host='dummy')
                    fw_rules = (
                        self.plugin._make_firewall_dict_with_rules(ctx,
                                                                   fw_id)
                    )
                    self.assertEqual(res[0], fw_rules)
                    self._compare_firewall_rule_lists(
                        fwp_id, fr, res[0]['firewall_rule_list'])

    def test_get_firewall_for_tenant_without_rules(self):
        tenant_id = 'test-tenant'
        ctx = context.Context('', tenant_id)
        with self.firewall_policy(tenant_id=tenant_id) as fwp:
            fwp_id = fwp['firewall_policy']['id']
            attrs = self._get_test_firewall_attrs()
            attrs['firewall_policy_id'] = fwp_id
            with self.firewall(firewall_policy_id=fwp_id, tenant_id=tenant_id,
                               admin_state_up=test_db_firewall.ADMIN_STATE_UP
                               ) as fw:
                    fw_list = [fw['firewall']]
                    f = self.callbacks.get_firewalls_for_tenant_without_rules
                    res = f(ctx, host='dummy')
                    for fw in res: