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

# Copyright 2011 OpenStack Foundation
# Copyright 2012 Justin Santa Barbara
#
#    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.

from lxml import etree
import mock
import mox
from oslo.config import cfg
import webob

from nova.api.openstack.compute.contrib import security_groups
from nova.api.openstack import wsgi
from nova.api.openstack import xmlutil
from nova import compute
from nova.compute import power_state
import nova.db
from nova import exception
from nova.objects import instance as instance_obj
from nova.openstack.common import jsonutils
from nova import quota
from nova import test
from nova.tests.api.openstack import fakes
from nova.tests import fake_instance
from nova.tests import utils

CONF = cfg.CONF
FAKE_UUID1 = 'a47ae74e-ab08-447f-8eee-ffd43fc46c16'
FAKE_UUID2 = 'c6e6430a-6563-4efa-9542-5e93c9e97d18'


class AttrDict(dict):
    def __getattr__(self, k):
        return self[k]


def security_group_template(**kwargs):
    sg = kwargs.copy()
    sg.setdefault('tenant_id', '123')
    sg.setdefault('name', 'test')
    sg.setdefault('description', 'test-description')
    return sg


def security_group_db(security_group, id=None):
    attrs = security_group.copy()
    if 'tenant_id' in attrs:
        attrs['project_id'] = attrs.pop('tenant_id')
    if id is not None:
        attrs['id'] = id
    attrs.setdefault('rules', [])
    attrs.setdefault('instances', [])
    return AttrDict(attrs)


def security_group_rule_template(**kwargs):
    rule = kwargs.copy()
    rule.setdefault('ip_protocol', 'tcp')
    rule.setdefault('from_port', 22)
    rule.setdefault('to_port', 22)
    rule.setdefault('parent_group_id', 2)
    return rule


def security_group_rule_db(rule, id=None):
    attrs = rule.copy()
    if 'ip_protocol' in attrs:
        attrs['protocol'] = attrs.pop('ip_protocol')
    return AttrDict(attrs)


def return_server(context, server_id, columns_to_join=None):
    return fake_instance.fake_db_instance(
        **{'id': int(server_id),
           'power_state': 0x01,
           'host': "localhost",
           'uuid': FAKE_UUID1,
           'name': 'asdf'})


def return_server_by_uuid(context, server_uuid, columns_to_join=None):
    return fake_instance.fake_db_instance(
        **{'id': 1,
           'power_state': 0x01,
           'host': "localhost",
           'uuid': server_uuid,
           'name': 'asdf'})


def return_non_running_server(context, server_id, columns_to_join=None):
    return fake_instance.fake_db_instance(
        **{'id': server_id, 'power_state': power_state.SHUTDOWN,
           'uuid': FAKE_UUID1, 'host': "localhost", 'name': 'asdf'})


def return_security_group_by_name(context, project_id, group_name):
    return {'id': 1, 'name': group_name,
            "instances": [{'id': 1, 'uuid': FAKE_UUID1}]}


def return_security_group_without_instances(context, project_id, group_name):
    return {'id': 1, 'name': group_name}


def return_server_nonexistent(context, server_id, columns_to_join=None):
    raise exception.InstanceNotFound(instance_id=server_id)


class TestSecurityGroups(test.TestCase):
    def setUp(self):
        super(TestSecurityGroups, self).setUp()

        self.controller = security_groups.SecurityGroupController()
        self.server_controller = (
            security_groups.ServerSecurityGroupController())
        self.manager = security_groups.SecurityGroupActionController()

        # This needs to be done here to set fake_id because the derived
        # class needs to be called first if it wants to set
        # 'security_group_api' and this setUp method needs to be called.
        if self.controller.security_group_api.id_is_uuid:
            self.fake_id = '11111111-1111-1111-1111-111111111111'
        else:
            self.fake_id = '11111111'

    def _assert_no_security_groups_reserved(self, context):
        """Check that no reservations are leaked during tests."""
        result = quota.QUOTAS.get_project_quotas(context, context.project_id)
        self.assertEqual(result['security_groups']['reserved'], 0)

    def test_create_security_group(self):
        sg = security_group_template()

        req = fakes.HTTPRequest.blank('/v2/fake/os-security-groups')
        res_dict = self.controller.create(req, {'security_group': sg})
        self.assertEqual(res_dict['security_group']['name'], 'test')
        self.assertEqual(res_dict['security_group']['description'],
                         'test-description')

    def test_create_security_group_with_no_name(self):
        sg = security_group_template()
        del sg['name']

        req = fakes.HTTPRequest.blank('/v2/fake/os-security-groups')
        self.assertRaises(webob.exc.HTTPUnprocessableEntity,
                          self.controller.create, req, sg)

        self._assert_no_security_groups_reserved(req.environ['nova.context'])

    def test_create_security_group_with_no_description(self):
        sg = security_group_template()
        del sg['description']

        req = fakes.HTTPRequest.blank('/v2/fake/os-security-groups')
        self.assertRaises(webob.exc.HTTPBadRequest, self.controller.create,
                          req, {'security_group': sg})

        self._assert_no_security_groups_reserved(req.environ['nova.context'])

    def test_create_security_group_with_blank_name(self):
        sg = security_group_template(name='')

        req = fakes.HTTPRequest.blank('/v2/fake/os-security-groups')
        self.assertRaises(webob.exc.HTTPBadRequest, self.controller.create,
                          req, {'security_group': sg})

        self._assert_no_security_groups_reserved(req.environ['nova.context'])

    def test_create_security_group_with_whitespace_name(self):
        sg = security_group_template(name=' ')

        req = fakes.HTTPRequest.blank('/v2/fake/os-security-groups')
        self.assertRaises(webob.exc.HTTPBadRequest, self.controller.create,
                          req, {'security_group': sg})

        self._assert_no_security_groups_reserved(req.environ['nova.context'])

    def test_create_security_group_with_blank_description(self):
        sg = security_group_template(description='')

        req = fakes.HTTPRequest.blank('/v2/fake/os-security-groups')
        self.assertRaises(webob.exc.HTTPBadRequest, self.controller.create,
                          req, {'security_group': sg})

        self._assert_no_security_groups_reserved(req.environ['nova.context'])

    def test_create_security_group_with_whitespace_description(self):
        sg = security_group_template(description=' ')

        req = fakes.HTTPRequest.blank('/v2/fake/os-security-groups')
        self.assertRaises(webob.exc.HTTPBadRequest, self.controller.create,
                          req, {'security_group': sg})

        self._assert_no_security_groups_reserved(req.environ['nova.context'])

    def test_create_security_group_with_duplicate_name(self):
        sg = security_group_template()

        # FIXME: Stub out _get instead of creating twice
        req = fakes.HTTPRequest.blank('/v2/fake/os-security-groups')
        self.controller.create(req, {'security_group': sg})

        req = fakes.HTTPRequest.blank('/v2/fake/os-security-groups')
        self.assertRaises(webob.exc.HTTPBadRequest, self.controller.create,
                          req, {'security_group': sg})

        self._assert_no_security_groups_reserved(req.environ['nova.context'])

    def test_create_security_group_with_no_body(self):
        req = fakes.HTTPRequest.blank('/v2/fake/os-security-groups')
        self.assertRaises(webob.exc.HTTPUnprocessableEntity,
                          self.controller.create, req, None)

        self._assert_no_security_groups_reserved(req.environ['nova.context'])

    def test_create_security_group_with_no_security_group(self):
        body = {'no-securityGroup': None}

        req = fakes.HTTPRequest.blank('/v2/fake/os-security-groups')
        self.assertRaises(webob.exc.HTTPUnprocessableEntity,
                          self.controller.create, req, body)

        self._assert_no_security_groups_reserved(req.environ['nova.context'])

    def test_create_security_group_above_255_characters_name(self):
        sg = security_group_template(name='1234567890' * 26)

        req = fakes.HTTPRequest.blank('/v2/fake/os-security-groups')
        self.assertRaises(webob.exc.HTTPBadRequest, self.controller.create,
                          req, {'security_group': sg})

        self._assert_no_security_groups_reserved(req.environ['nova.context'])

    def test_create_security_group_above_255_characters_description(self):
        sg = security_group_template(description='1234567890' * 26)

        req = fakes.HTTPRequest.blank('/v2/fake/os-security-groups')
        self.assertRaises(webob.exc.HTTPBadRequest, self.controller.create,
                          req, {'security_group': sg})

        self._assert_no_security_groups_reserved(req.environ['nova.context'])

    def test_create_security_group_non_string_name(self):
        sg = security_group_template(name=12)

        req = fakes.HTTPRequest.blank('/v2/fake/os-security-groups')
        self.assertRaises(webob.exc.HTTPBadRequest, self.controller.create,
                          req, {'security_group': sg})

        self._assert_no_security_groups_reserved(req.environ['nova.context'])

    def test_create_security_group_non_string_description(self):
        sg = security_group_template(description=12)

        req = fakes.HTTPRequest.blank('/v2/fake/os-security-groups')
        self.assertRaises(webob.exc.HTTPBadRequest, self.controller.create,
                          req, {'security_group': sg})

        self._assert_no_security_groups_reserved(req.environ['nova.context'])

    def test_create_security_group_quota_limit(self):
        req = fakes.HTTPRequest.blank('/v2/fake/os-security-groups')
        for num in range(1, CONF.quota_security_groups + 1):
            name = 'test%s' % num
            sg = security_group_template(name=name)
            res_dict = self.controller.create(req, {'security_group': sg})
            self.assertEqual(res_dict['security_group']['name'], name)

        sg = security_group_template()
        self.assertRaises(webob.exc.HTTPRequestEntityTooLarge,
                          self.controller.create,
                          req, {'security_group': sg})

    def test_get_security_group_list(self):
        groups = []
        for i, name in enumerate(['default', 'test']):
            sg = security_group_template(id=i + 1,
                                         name=name,
                                         description=name + '-desc',
                                         rules=[])
            groups.append(sg)
        expected = {'security_groups': groups}

        def return_security_groups(context, project_id):
            return [security_group_db(sg) for sg in groups]

        self.stubs.Set(nova.db, 'security_group_get_by_project',
                       return_security_groups)

        req = fakes.HTTPRequest.blank('/v2/fake/os-security-groups')
        res_dict = self.controller.index(req)

        self.assertEquals(res_dict, expected)

    def test_get_security_group_list_all_tenants(self):
        all_groups = []
        tenant_groups = []

        for i, name in enumerate(['default', 'test']):
            sg = security_group_template(id=i + 1,
                                         name=name,
                                         description=name + '-desc',
                                         rules=[])
            all_groups.append(sg)
            if name == 'default':
                tenant_groups.append(sg)

        all = {'security_groups': all_groups}
        tenant_specific = {'security_groups': tenant_groups}

        def return_all_security_groups(context):
            return [security_group_db(sg) for sg in all_groups]

        self.stubs.Set(nova.db, 'security_group_get_all',
                       return_all_security_groups)

        def return_tenant_security_groups(context, project_id):
            return [security_group_db(sg) for sg in tenant_groups]

        self.stubs.Set(nova.db, 'security_group_get_by_project',
                       return_tenant_security_groups)

        path = '/v2/fake/os-security-groups'

        req = fakes.HTTPRequest.blank(path, use_admin_context=True)
        res_dict = self.controller.index(req)
        self.assertEquals(res_dict, tenant_specific)

        req = fakes.HTTPRequest.blank('%s?all_tenants=1' % path,
                                      use_admin_context=True)
        res_dict = self.controller.index(req)
        self.assertEquals(res_dict, all)

    def test_get_security_group_by_instance(self):
        groups = []
        for i, name in enumerate(['default', 'test']):
            sg = security_group_template(id=i + 1,
                                         name=name,
                                         description=name + '-desc',
                                         rules=[])
            groups.append(sg)
        expected = {'security_groups': groups}

        def return_instance(context, server_id, columns_to_join=None):
            self.assertEquals(server_id, FAKE_UUID1)
            return return_server_by_uuid(context, server_id)

        self.stubs.Set(nova.db, 'instance_get_by_uuid',
                       return_instance)

        def return_security_groups(context, instance_uuid):
            self.assertEquals(instance_uuid, FAKE_UUID1)
            return [security_group_db(sg) for sg in groups]

        self.stubs.Set(nova.db, 'security_group_get_by_instance',
                       return_security_groups)

        req = fakes.HTTPRequest.blank('/v2/%s/servers/%s/os-security-groups' %
                                      ('fake', FAKE_UUID1))
        res_dict = self.server_controller.index(req, FAKE_UUID1)

        self.assertEquals(res_dict, expected)

    @mock.patch('nova.db.instance_get_by_uuid')
    @mock.patch('nova.db.security_group_get_by_instance', return_value=[])
    def test_get_security_group_empty_for_instance(self, mock_sec_group,
                                                   mock_db_get_ins):
        expected = {'security_groups': []}

        def return_instance(context, server_id,
                            columns_to_join=None, use_slave=False):
            self.assertEqual(server_id, FAKE_UUID1)
            return return_server_by_uuid(context, server_id)
        mock_db_get_ins.side_effect = return_instance
        req = fakes.HTTPRequest.blank('/v2/%s/servers/%s/os-security-groups' %
                                      ('fake', FAKE_UUID1))
        res_dict = self.server_controller.index(req, FAKE_UUID1)
        self.assertEqual(expected, res_dict)
        mock_sec_group.assert_called_once_with(req.environ['nova.context'],
                                               FAKE_UUID1)

    def test_get_security_group_by_instance_non_existing(self):
        self.stubs.Set(nova.db, 'instance_get', return_server_nonexistent)
        self.stubs.Set(nova.db, 'instance_get_by_uuid',
                       return_server_nonexistent)
        req = fakes.HTTPRequest.blank('/v2/fake/servers/1/os-security-groups')
        self.assertRaises(webob.exc.HTTPNotFound,
                          self.server_controller.index, req, '1')

    def test_get_security_group_by_instance_invalid_id(self):
        req = fakes.HTTPRequest.blank(
            '/v2/fake/servers/invalid/os-security-groups')
        self.assertRaises(webob.exc.HTTPNotFound,
                          self.server_controller.index, req, 'invalid')

    def test_get_security_group_by_id(self):
        sg = security_group_template(id=2, rules=[])

        def return_security_group(context, group_id):
            self.assertEquals(sg['id'], group_id)
            return security_group_db(sg)

        self.stubs.Set(nova.db, 'security_group_get',
                       return_security_group)

        req = fakes.HTTPRequest.blank('/v2/fake/os-security-groups/2')
        res_dict = self.controller.show(req, '2')

        expected = {'security_group': sg}
        self.assertEquals(res_dict, expected)

    def test_get_security_group_by_invalid_id(self):
        req = fakes.HTTPRequest.blank('/v2/fake/os-security-groups/invalid')
        self.assertRaises(webob.exc.HTTPBadRequest, self.controller.delete,
                          req, 'invalid')

    def test_get_security_group_by_non_existing_id(self):
        req = fakes.HTTPRequest.blank('/v2/fake/os-security-groups/%s' %
                                      self.fake_id)
        self.assertRaises(webob.exc.HTTPNotFound, self.controller.delete,
                          req, self.fake_id)

    def test_update_security_group(self):
        sg = security_group_template(id=2, rules=[])
        sg_update = security_group_template(id=2, rules=[],
                        name='update_name', description='update_desc')

        def return_security_group(context, group_id):
            self.assertEquals(sg['id'], group_id)
            return security_group_db(sg)

        def return_update_security_group(context, group_id, values,
                                         columns_to_join=None):
            self.assertEqual(sg_update['id'], group_id)
            self.assertEqual(sg_update['name'], values['name'])
            self.assertEqual(sg_update['description'], values['description'])
            return security_group_db(sg_update)

        self.stubs.Set(nova.db, 'security_group_update',
                       return_update_security_group)
        self.stubs.Set(nova.db, 'security_group_get',
                       return_security_group)

        req = fakes.HTTPRequest.blank('/v2/fake/os-security-groups/2')
        res_dict = self.controller.update(req, '2',
                                          {'security_group': sg_update})

        expected = {'security_group': sg_update}
        self.assertEquals(res_dict, expected)

    def test_update_security_group_name_to_default(self):
        sg = security_group_template(id=2, rules=[], name='default')

        def return_security_group(context, group_id):
            self.assertEquals(sg['id'], group_id)
            return security_group_db(sg)

        self.stubs.Set(nova.db, 'security_group_get',
                       return_security_group)

        req = fakes.HTTPRequest.blank('/v2/fake/os-security-groups/2')
        self.assertRaises(webob.exc.HTTPBadRequest, self.controller.update,
                          req, '2', {'security_group': sg})

    def test_update_default_security_group_fail(self):
        sg = security_group_template()

        req = fakes.HTTPRequest.blank('/v2/fake/os-security-groups/1')
        self.assertRaises(webob.exc.HTTPBadRequest, self.controller.update,
                          req, '1', {'security_group': sg})

    def test_delete_security_group_by_id(self):
        sg = security_group_template(id=1, rules=[])

        self.called = False

        def security_group_destroy(context, id):
            self.called = True

        def return_security_group(context, group_id):
            self.assertEquals(sg['id'], group_id)
            return security_group_db(sg)

        self.stubs.Set(nova.db, 'security_group_destroy',
                       security_group_destroy)
        self.stubs.Set(nova.db, 'security_group_get',
                       return_security_group)

        req = fakes.HTTPRequest.blank('/v2/fake/os-security-groups/1')
        self.controller.delete(req, '1')

        self.assertTrue(self.called)

    def test_delete_security_group_by_invalid_id(self):
        req = fakes.HTTPRequest.blank('/v2/fake/os-security-groups/invalid')
        self.assertRaises(webob.exc.HTTPBadRequest, self.controller.delete,
                          req, 'invalid')

    def test_delete_security_group_by_non_existing_id(self):
        req = fakes.HTTPRequest.blank('/v2/fake/os-security-groups/%s'
                                      % self.fake_id)
        self.assertRaises(webob.exc.HTTPNotFound, self.controller.delete,
                          req, self.fake_id)

    def test_delete_security_group_in_use(self):
        sg = security_group_template(id=1, rules=[])

        def security_group_in_use(context, id):
            return True

        def return_security_group(context, group_id):
            self.assertEquals(sg['id'], group_id)
            return security_group_db(sg)

        self.stubs.Set(nova.db, 'security_group_in_use',
                       security_group_in_use)
        self.stubs.Set(nova.db, 'security_group_get',
                       return_security_group)

        req = fakes.HTTPRequest.blank('/v2/fake/os-security-groups/1')
        self.assertRaises(webob.exc.HTTPBadRequest, self.controller.delete,
                          req, '1')

    def test_associate_by_non_existing_security_group_name(self):
        self.stubs.Set(nova.db, 'instance_get', return_server)
        self.assertEquals(return_server(None, '1'),
                          nova.db.instance_get(None, '1'))
        body = dict(addSecurityGroup=dict(name='non-existing'))

        req = fakes.HTTPRequest.blank('/v2/fake/servers/1/action')
        self.assertRaises(webob.exc.HTTPNotFound,
                          self.manager._addSecurityGroup, req, '1', body)

    def test_associate_by_invalid_server_id(self):
        body = dict(addSecurityGroup=dict(name='test'))

        req = fakes.HTTPRequest.blank('/v2/fake/servers/invalid/action')
        self.assertRaises(webob.exc.HTTPNotFound,
                          self.manager._addSecurityGroup, req, 'invalid', body)

    def test_associate_without_body(self):
        self.stubs.Set(nova.db, 'instance_get', return_server)
        body = dict(addSecurityGroup=None)

        req = fakes.HTTPRequest.blank('/v2/fake/servers/1/action')
        self.assertRaises(webob.exc.HTTPBadRequest,
                          self.manager._addSecurityGroup, req, '1', body)

    def test_associate_no_security_group_name(self):
        self.stubs.Set(nova.db, 'instance_get', return_server)
        body = dict(addSecurityGroup=dict())

        req = fakes.HTTPRequest.blank('/v2/fake/servers/1/action')
        self.assertRaises(webob.exc.HTTPBadRequest,
                          self.manager._addSecurityGroup, req, '1', body)

    def test_associate_security_group_name_with_whitespaces(self):
        self.stubs.Set(nova.db, 'instance_get', return_server)
        body = dict(addSecurityGroup=dict(name="   "))

        req = fakes.HTTPRequest.blank('/v2/fake/servers/1/action')
        self.assertRaises(webob.exc.HTTPBadRequest,
                          self.manager._addSecurityGroup, req, '1', body)

    def test_associate_non_existing_instance(self):
        self.stubs.Set(nova.db, 'instance_get', return_server_nonexistent)
        self.stubs.Set(nova.db, 'instance_get_by_uuid',
                       return_server_nonexistent)
        body = dict(addSecurityGroup=dict(name="test"))

        req = fakes.HTTPRequest.blank('/v2/fake/servers/1/action')
        self.assertRaises(webob.exc.HTTPNotFound,
                          self.manager._addSecurityGroup, req, '1', body)

    def test_associate_non_running_instance(self):
        self.stubs.Set(nova.db, 'instance_get', return_non_running_server)
        self.stubs.Set(nova.db, 'instance_get_by_uuid',
                       return_non_running_server)
        self.stubs.Set(nova.db, 'security_group_get_by_name',
                       return_security_group_without_instances)
        body = dict(addSecurityGroup=dict(name="test"))

        req = fakes.HTTPRequest.blank('/v2/fake/servers/1/action')
        self.manager._addSecurityGroup(req, '1', body)

    def test_associate_already_associated_security_group_to_instance(self):
        self.stubs.Set(nova.db, 'instance_get', return_server)
        self.stubs.Set(nova.db, 'instance_get_by_uuid',