# Copyright 2013 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 mock
from mox3 import mox
from neutronclient.common import exceptions as n_exc
from neutronclient.v2_0 import client

from six.moves import range

from nova import context
from nova import exception
from nova.network.neutronv2 import api as neutronapi
from nova.network.security_group import neutron_driver
from nova import test


class TestNeutronDriver(test.NoDBTestCase):
    def setUp(self):
        super(TestNeutronDriver, self).setUp()
        self.mox.StubOutWithMock(neutronapi, 'get_client')
        self.moxed_client = self.mox.CreateMock(client.Client)
        neutronapi.get_client(mox.IgnoreArg()).MultipleTimes().AndReturn(
            self.moxed_client)
        self.context = context.RequestContext('userid', 'my_tenantid')
        setattr(self.context,
                'auth_token',
                'bff4a5a6b9eb4ea2a6efec6eefb77936')

    def test_list_with_project(self):
        project_id = '0af70a4d22cf4652824ddc1f2435dd85'
        security_groups_list = {'security_groups': []}
        self.moxed_client.list_security_groups(tenant_id=project_id).AndReturn(
            security_groups_list)
        self.mox.ReplayAll()

        sg_api = neutron_driver.SecurityGroupAPI()
        sg_api.list(self.context, project=project_id)

    def test_list_with_all_tenants_and_admin_context(self):
        project_id = '0af70a4d22cf4652824ddc1f2435dd85'
        search_opts = {'all_tenants': 1}
        security_groups_list = {'security_groups': []}
        admin_context = context.RequestContext('user1', project_id, True)
        self.mox.ReplayAll()

        with mock.patch.object(
                self.moxed_client,
                'list_security_groups',
                return_value=security_groups_list) as mock_list_secgroup:
            sg_api = neutron_driver.SecurityGroupAPI()
            sg_api.list(admin_context,
                        project=project_id,
                        search_opts=search_opts)

            mock_list_secgroup.assert_called_once_with()

    def test_list_without_all_tenants_and_admin_context(self):
        project_id = '0af70a4d22cf4652824ddc1f2435dd85'
        security_groups_list = {'security_groups': []}
        admin_context = context.RequestContext('user1', project_id, True)
        self.mox.ReplayAll()

        with mock.patch.object(
                self.moxed_client,
                'list_security_groups',
                return_value=security_groups_list) as mock_list_secgroup:
            sg_api = neutron_driver.SecurityGroupAPI()
            sg_api.list(admin_context, project=project_id)

            mock_list_secgroup.assert_called_once_with(tenant_id=project_id)

    def test_list_with_all_tenants_sec_name_and_admin_context(self):
        project_id = '0af70a4d22cf4652824ddc1f2435dd85'
        search_opts = {'all_tenants': 1}
        security_group_names = ['secgroup_ssh']
        security_groups_list = {'security_groups': []}
        admin_context = context.RequestContext('user1', project_id, True)
        self.mox.ReplayAll()

        with mock.patch.object(
                self.moxed_client,
                'list_security_groups',
                return_value=security_groups_list) as mock_list_secgroup:
            sg_api = neutron_driver.SecurityGroupAPI()
            sg_api.list(admin_context, project=project_id,
                        names=security_group_names,
                        search_opts=search_opts)

            mock_list_secgroup.assert_called_once_with(
                name=security_group_names,
                tenant_id=project_id)

    def test_list_with_all_tenants_sec_name_ids_and_admin_context(self):
        project_id = '0af70a4d22cf4652824ddc1f2435dd85'
        search_opts = {'all_tenants': 1}
        security_group_names = ['secgroup_ssh']
        security_group_ids = ['id1']
        security_groups_list = {'security_groups': []}
        admin_context = context.RequestContext('user1', project_id, True)
        self.mox.ReplayAll()

        with mock.patch.object(
                self.moxed_client,
                'list_security_groups',
                return_value=security_groups_list) as mock_list_secgroup:
            sg_api = neutron_driver.SecurityGroupAPI()
            sg_api.list(admin_context, project=project_id,
                        names=security_group_names,
                        ids=security_group_ids,
                        search_opts=search_opts)

            mock_list_secgroup.assert_called_once_with(
                name=security_group_names,
                id=security_group_ids,
                tenant_id=project_id)

    def test_list_with_all_tenants_not_admin(self):
        search_opts = {'all_tenants': 1}
        security_groups_list = {'security_groups': []}
        self.mox.ReplayAll()

        with mock.patch.object(
                self.moxed_client,
                'list_security_groups',
                return_value=security_groups_list) as mock_list_secgroup:
            sg_api = neutron_driver.SecurityGroupAPI()
            sg_api.list(self.context, project=self.context.tenant,
                        search_opts=search_opts)

            mock_list_secgroup.assert_called_once_with(
                tenant_id=self.context.tenant)

    def test_get_with_name_duplicated(self):
        sg_name = 'web_server'
        expected_sg_id = '85cc3048-abc3-43cc-89b3-377341426ac5'
        list_security_groups = {'security_groups':
                                [{'name': sg_name,
                                  'id': expected_sg_id,
                                  'tenant_id': self.context.tenant,
                                  'description': 'server',
                                  'rules': []}
                                ]}
        self.moxed_client.list_security_groups(name=sg_name, fields='id',
            tenant_id=self.context.tenant).AndReturn(list_security_groups)

        expected_sg = {'security_group': {'name': sg_name,
                                 'id': expected_sg_id,
                                 'tenant_id': self.context.tenant,
                                 'description': 'server', 'rules': []}}
        self.moxed_client.show_security_group(expected_sg_id).AndReturn(
            expected_sg)
        self.mox.ReplayAll()

        sg_api = neutron_driver.SecurityGroupAPI()
        observed_sg = sg_api.get(self.context, name=sg_name)
        expected_sg['security_group']['project_id'] = self.context.tenant
        del expected_sg['security_group']['tenant_id']
        self.assertEqual(expected_sg['security_group'], observed_sg)

    def test_get_with_invalid_name(self):
        sg_name = 'invalid_name'
        expected_sg_id = '85cc3048-abc3-43cc-89b3-377341426ac5'
        list_security_groups = {'security_groups':
                                [{'name': sg_name,
                                  'id': expected_sg_id,
                                  'tenant_id': self.context.tenant,
                                  'description': 'server',
                                  'rules': []}
                                ]}
        self.moxed_client.list_security_groups(name=sg_name, fields='id',
            tenant_id=self.context.tenant).AndReturn(list_security_groups)

        self.moxed_client.show_security_group(expected_sg_id).AndRaise(
            TypeError)
        self.mox.ReplayAll()

        sg_api = neutron_driver.SecurityGroupAPI()
        self.assertRaises(exception.SecurityGroupNotFound,
                          sg_api.get, self.context, name=sg_name)

    def test_create_security_group_with_bad_request(self):
        name = 'test-security-group'
        description = None
        body = {'security_group': {'name': name,
                                   'description': description}}
        message = "Invalid input. Reason: 'None' is not a valid string."
        self.moxed_client.create_security_group(
            body).AndRaise(n_exc.BadRequest(message=message))
        self.mox.ReplayAll()
        sg_api = neutron_driver.SecurityGroupAPI()
        self.assertRaises(exception.Invalid,
                          sg_api.create_security_group, self.context, name,
                          description)

    def test_create_security_group_exceed_quota(self):
        name = 'test-security-group'
        description = 'test-security-group'
        body = {'security_group': {'name': name,
                                   'description': description}}
        message = "Quota exceeded for resources: ['security_group']"
        self.moxed_client.create_security_group(
            body).AndRaise(n_exc.NeutronClientException(status_code=409,
                                                        message=message))
        self.mox.ReplayAll()
        sg_api = neutron_driver.SecurityGroupAPI()
        self.assertRaises(exception.SecurityGroupLimitExceeded,
                          sg_api.create_security_group, self.context, name,
                          description)

    def test_create_security_group_rules_exceed_quota(self):
        vals = {'protocol': 'tcp', 'cidr': '0.0.0.0/0',
                'parent_group_id': '7ae75663-277e-4a0e-8f87-56ea4e70cb47',
                'group_id': None, 'from_port': 1025, 'to_port': 1025}
        body = {'security_group_rules': [{'remote_group_id': None,
                'direction': 'ingress', 'protocol': 'tcp', 'ethertype': 'IPv4',
                'port_range_max': 1025, 'port_range_min': 1025,
                'security_group_id': '7ae75663-277e-4a0e-8f87-56ea4e70cb47',
                'remote_ip_prefix': '0.0.0.0/0'}]}
        name = 'test-security-group'
        message = "Quota exceeded for resources: ['security_group_rule']"
        self.moxed_client.create_security_group_rule(
            body).AndRaise(n_exc.NeutronClientException(status_code=409,
                                                        message=message))
        self.mox.ReplayAll()
        sg_api = neutron_driver.SecurityGroupAPI()
        self.assertRaises(exception.SecurityGroupLimitExceeded,
                          sg_api.add_rules, self.context, None, name, [vals])

    def test_create_security_group_rules_bad_request(self):
        vals = {'protocol': 'icmp', 'cidr': '0.0.0.0/0',
                'parent_group_id': '7ae75663-277e-4a0e-8f87-56ea4e70cb47',
                'group_id': None, 'to_port': 255}
        body = {'security_group_rules': [{'remote_group_id': None,
                'direction': 'ingress', 'protocol': 'icmp',
                'ethertype': 'IPv4', 'port_range_max': 255,
                'security_group_id': '7ae75663-277e-4a0e-8f87-56ea4e70cb47',
                'remote_ip_prefix': '0.0.0.0/0'}]}
        name = 'test-security-group'
        message = "ICMP code (port-range-max) 255 is provided but ICMP type" \
                  " (port-range-min) is missing"
        self.moxed_client.create_security_group_rule(
            body).AndRaise(n_exc.NeutronClientException(status_code=400,
                                                        message=message))
        self.mox.ReplayAll()
        sg_api = neutron_driver.SecurityGroupAPI()
        self.assertRaises(exception.Invalid, sg_api.add_rules,
                          self.context, None, name, [vals])

    def test_list_security_group_with_no_port_range_and_not_tcp_udp_icmp(self):
        sg1 = {'description': 'default',
               'id': '07f1362f-34f6-4136-819a-2dcde112269e',
               'name': 'default',
               'tenant_id': 'c166d9316f814891bcb66b96c4c891d6',
               'security_group_rules':
                   [{'direction': 'ingress',
                     'ethertype': 'IPv4',
                     'id': '0a4647f1-e1aa-488d-90e1-97a7d0293beb',
                      'port_range_max': None,
                      'port_range_min': None,
                      'protocol': '51',
                      'remote_group_id': None,
                      'remote_ip_prefix': None,
                      'security_group_id':
                           '07f1362f-34f6-4136-819a-2dcde112269e',
                      'tenant_id': 'c166d9316f814891bcb66b96c4c891d6'}]}

        self.moxed_client.list_security_groups().AndReturn(
            {'security_groups': [sg1]})
        self.mox.ReplayAll()
        sg_api = neutron_driver.SecurityGroupAPI()
        result = sg_api.list(self.context)
        expected = [{'rules':
            [{'from_port': -1, 'protocol': '51', 'to_port': -1,
              'parent_group_id': '07f1362f-34f6-4136-819a-2dcde112269e',
              'cidr': '0.0.0.0/0', 'group_id': None,
              'id': '0a4647f1-e1aa-488d-90e1-97a7d0293beb'}],
            'project_id': 'c166d9316f814891bcb66b96c4c891d6',
            'id': '07f1362f-34f6-4136-819a-2dcde112269e',
            'name': 'default', 'description': 'default'}]
        self.assertEqual(expected, result)

    def test_instances_security_group_bindings(self):
        server_id = 'c5a20e8d-c4b0-47cf-9dca-ebe4f758acb1'
        port1_id = '4c505aec-09aa-47bc-bcc0-940477e84dc0'
        port2_id = 'b3b31a53-6e29-479f-ae5c-00b7b71a6d44'
        sg1_id = '2f7ce969-1a73-4ef9-bbd6-c9a91780ecd4'
        sg2_id = '20c89ce5-9388-4046-896e-64ffbd3eb584'
        servers = [{'id': server_id}]
        ports = [{'id': port1_id, 'device_id': server_id,
                  'security_groups': [sg1_id]},
                 {'id': port2_id, 'device_id': server_id,
                  'security_groups': [sg2_id]}]
        port_list = {'ports': ports}
        sg1 = {'id': sg1_id, 'name': 'wol'}
        sg2 = {'id': sg2_id, 'name': 'eor'}
        security_groups_list = {'security_groups': [sg1, sg2]}

        sg_bindings = {server_id: [{'name': 'wol'}, {'name': 'eor'}]}

        self.moxed_client.list_ports(device_id=[server_id]).AndReturn(
            port_list)
        self.moxed_client.list_security_groups(
            id=mox.SameElementsAs([sg2_id, sg1_id])).AndReturn(
            security_groups_list)
        self.mox.ReplayAll()

        sg_api = neutron_driver.SecurityGroupAPI()
        result = sg_api.get_instances_security_groups_bindings(
                                  self.context, servers)
        self.assertEqual(result, sg_bindings)

    def _test_instances_security_group_bindings_scale(self, num_servers):
        max_query = 150
        sg1_id = '2f7ce969-1a73-4ef9-bbd6-c9a91780ecd4'
        sg2_id = '20c89ce5-9388-4046-896e-64ffbd3eb584'
        sg1 = {'id': sg1_id, 'name': 'wol'}
        sg2 = {'id': sg2_id, 'name': 'eor'}
        security_groups_list = {'security_groups': [sg1, sg2]}
        servers = []
        device_ids = []
        ports = []
        sg_bindings = {}
        for i in range(0, num_servers):
            server_id = "server-%d" % i
            port_id = "port-%d" % i
            servers.append({'id': server_id})
            device_ids.append(server_id)
            ports.append({'id': port_id,
                          'device_id': server_id,
                          'security_groups': [sg1_id, sg2_id]})
            sg_bindings[server_id] = [{'name': 'wol'}, {'name': 'eor'}]

        for x in range(0, num_servers, max_query):
            self.moxed_client.list_ports(
                       device_id=device_ids[x:x + max_query]).\
                       AndReturn({'ports': ports[x:x + max_query]})

        self.moxed_client.list_security_groups(
            id=mox.SameElementsAs([sg2_id, sg1_id])).AndReturn(
            security_groups_list)
        self.mox.ReplayAll()

        sg_api = neutron_driver.SecurityGroupAPI()
        result = sg_api.get_instances_security_groups_bindings(
                                  self.context, servers)
        self.assertEqual(result, sg_bindings)

    def test_instances_security_group_bindings_less_than_max(self):
        self._test_instances_security_group_bindings_scale(100)

    def test_instances_security_group_bindings_max(self):
        self._test_instances_security_group_bindings_scale(150)

    def test_instances_security_group_bindings_more_then_max(self):
        self._test_instances_security_group_bindings_scale(300)

    def test_instances_security_group_bindings_with_hidden_sg(self):
        servers = [{'id': 'server_1'}]
        ports = [{'id': '1', 'device_id': 'dev_1', 'security_groups': ['1']},
                 {'id': '2', 'device_id': 'dev_1', 'security_groups': ['2']}]
        port_list = {'ports': ports}
        sg1 = {'id': '1', 'name': 'wol'}
        # User doesn't have access to sg2
        security_groups_list = {'security_groups': [sg1]}

        sg_bindings = {'dev_1': [{'name': 'wol'}]}

        self.moxed_client.list_ports(device_id=['server_1']).AndReturn(
            port_list)
        self.moxed_client.\
            list_security_groups(id=mox.SameElementsAs(['1', '2'])).AndReturn(
                security_groups_list)
        self.mox.ReplayAll()

        sg_api = neutron_driver.SecurityGroupAPI()
        result = sg_api.get_instances_security_groups_bindings(
                                  self.context, servers)
        self.assertEqual(result, sg_bindings)

    def test_instance_empty_security_groups(self):

        port_list = {'ports': [{'id': 1, 'device_id': '1',
                     'security_groups': []}]}
        self.moxed_client.list_ports(device_id=['1']).AndReturn(port_list)
        self.mox.ReplayAll()
        sg_api = neutron_driver.SecurityGroupAPI()
        result = sg_api.get_instance_security_groups(self.context, '1')
        self.assertEqual([], result)


class TestNeutronDriverWithoutMock(test.NoDBTestCase):

    def test_validate_property(self):
        sg_api = neutron_driver.SecurityGroupAPI()

        sg_api.validate_property('foo', 'name', None)
        sg_api.validate_property('', 'name', None)
        self.assertRaises(exception.Invalid, sg_api.validate_property,
                          'a' * 256, 'name', None)
        self.assertRaises(exception.Invalid, sg_api.validate_property,
                          None, 'name', None)
