manila/manila/tests/api/v1/test_share_servers.py

# Copyright 2014 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 webob import exc

from manila.api.v1 import share_servers
from manila.common import constants
from manila import context
from manila.db import api as db_api
from manila import exception
from manila import policy
from manila import test


fake_share_server_list = {
    'share_servers': [
        {
            'status': constants.STATUS_ACTIVE,
            'updated_at': None,
            'host': 'fake_host',
            'share_network_id': 'fake_sn_id',
            'share_network_name': 'fake_sn_name',
            'project_id': 'fake_project_id',
            'id': 'fake_server_id',
        },
        {
            'status': constants.STATUS_ERROR,
            'updated_at': None,
            'host': 'fake_host_2',
            'share_network_id': 'fake_sn_id_2',
            'share_network_name': 'fake_sn_id_2',
            'project_id': 'fake_project_id_2',
            'id': 'fake_server_id_2',
        },
    ]
}


fake_share_server_get_result = {
    'share_server': {
        'status': constants.STATUS_ACTIVE,
        'created_at': None,
        'updated_at': None,
        'host': 'fake_host',
        'share_network_name': 'fake_sn_name',
        'share_network_id': 'fake_sn_id',
        'project_id': 'fake_project_id',
        'id': 'fake_server_id',
        'backend_details': {
            'fake_key_1': 'fake_value_1',
            'fake_key_2': 'fake_value_2',
        }
    }
}

share_server_backend_details = {
    'fake_key_1': 'fake_value_1',
    'fake_key_2': 'fake_value_2',
}

fake_share_server_backend_details_get_result = {
    'details': share_server_backend_details
}


CONTEXT = context.get_admin_context()


class FakeShareServer(object):

    def __init__(self, *args, **kwargs):
        super(FakeShareServer, self).__init__()
        self.id = kwargs.get('id', 'fake_server_id')
        if 'created_at' in kwargs:
            self.created_at = kwargs.get('created_at', None)
        self.updated_at = kwargs.get('updated_at', None)
        self.host = kwargs.get('host', 'fake_host')
        self.share_network = kwargs.get('share_network', {
            'name': 'fake_sn_name', 'id': 'fake_sn_id',
            'project_id': 'fake_project_id'})
        self.share_network_id = kwargs.get('share_network_id',
                                           self.share_network['id'])
        self.status = kwargs.get('status', constants.STATUS_ACTIVE)
        self.project_id = self.share_network['project_id']
        self.backend_details = share_server_backend_details

    def __getitem__(self, item):
        return getattr(self, item)


def fake_share_server_get_all():
    fake_share_servers = [
        FakeShareServer(),
        FakeShareServer(id='fake_server_id_2',
                        host='fake_host_2',
                        share_network={
                            'name': None,
                            'id': 'fake_sn_id_2',
                            'project_id': 'fake_project_id_2'},
                        status=constants.STATUS_ERROR)
    ]
    return fake_share_servers


def fake_share_server_get():
    return FakeShareServer(created_at=None)


class FakeRequestAdmin(object):
    environ = {"manila.context": CONTEXT}
    GET = {}


class FakeRequestWithHost(FakeRequestAdmin):
    GET = {'host': fake_share_server_list['share_servers'][0]['host']}


class FakeRequestWithStatus(FakeRequestAdmin):
    GET = {'status': constants.STATUS_ERROR}


class FakeRequestWithProjectId(FakeRequestAdmin):
    GET = {'project_id': fake_share_server_get_all()[0].project_id}


class FakeRequestWithShareNetworkName(FakeRequestAdmin):
    GET = {
        'share_network': fake_share_server_get_all()[0].share_network['name'],
    }


class FakeRequestWithShareNetworkId(FakeRequestAdmin):
    GET = {
        'share_network': fake_share_server_get_all()[0].share_network['id'],
    }


class FakeRequestWithFakeFilter(FakeRequestAdmin):
    GET = {'fake_key': 'fake_value'}


class ShareServerAPITest(test.TestCase):

    def setUp(self):
        super(ShareServerAPITest, self).setUp()
        self.controller = share_servers.ShareServerController()
        self.resource_name = self.controller.resource_name
        self.mock_object(policy, 'check_policy',
                         mock.Mock(return_value=True))
        self.mock_object(db_api, 'share_server_get_all',
                         mock.Mock(return_value=fake_share_server_get_all()))

    def test_index_no_filters(self):
        result = self.controller.index(FakeRequestAdmin)
        policy.check_policy.assert_called_once_with(
            CONTEXT, self.resource_name, 'index')
        db_api.share_server_get_all.assert_called_once_with(CONTEXT)
        self.assertEqual(fake_share_server_list, result)

    def test_index_host_filter(self):
        result = self.controller.index(FakeRequestWithHost)
        policy.check_policy.assert_called_once_with(
            CONTEXT, self.resource_name, 'index')
        db_api.share_server_get_all.assert_called_once_with(CONTEXT)
        self.assertEqual([fake_share_server_list['share_servers'][0]],
                         result['share_servers'])

    def test_index_status_filter(self):
        result = self.controller.index(FakeRequestWithStatus)
        policy.check_policy.assert_called_once_with(
            CONTEXT, self.resource_name, 'index')
        db_api.share_server_get_all.assert_called_once_with(CONTEXT)
        self.assertEqual([fake_share_server_list['share_servers'][1]],
                         result['share_servers'])

    def test_index_project_id_filter(self):
        result = self.controller.index(FakeRequestWithProjectId)
        policy.check_policy.assert_called_once_with(
            CONTEXT, self.resource_name, 'index')
        db_api.share_server_get_all.assert_called_once_with(CONTEXT)
        self.assertEqual([fake_share_server_list['share_servers'][0]],
                         result['share_servers'])

    def test_index_share_network_filter_by_name(self):
        result = self.controller.index(FakeRequestWithShareNetworkName)
        policy.check_policy.assert_called_once_with(
            CONTEXT, self.resource_name, 'index')
        db_api.share_server_get_all.assert_called_once_with(CONTEXT)
        self.assertEqual([fake_share_server_list['share_servers'][0]],
                         result['share_servers'])

    def test_index_share_network_filter_by_id(self):
        result = self.controller.index(FakeRequestWithShareNetworkId)
        policy.check_policy.assert_called_once_with(
            CONTEXT, self.resource_name, 'index')
        db_api.share_server_get_all.assert_called_once_with(CONTEXT)
        self.assertEqual([fake_share_server_list['share_servers'][0]],
                         result['share_servers'])

    def test_index_fake_filter(self):
        result = self.controller.index(FakeRequestWithFakeFilter)
        policy.check_policy.assert_called_once_with(
            CONTEXT, self.resource_name, 'index')
        db_api.share_server_get_all.assert_called_once_with(CONTEXT)
        self.assertEqual(0, len(result['share_servers']))

    def test_show(self):
        self.mock_object(db_api, 'share_server_get',
                         mock.Mock(return_value=fake_share_server_get()))
        result = self.controller.show(
            FakeRequestAdmin,
            fake_share_server_get_result['share_server']['id'])
        policy.check_policy.assert_called_once_with(
            CONTEXT, self.resource_name, 'show')
        db_api.share_server_get.assert_called_once_with(
            CONTEXT, fake_share_server_get_result['share_server']['id'])
        self.assertEqual(fake_share_server_get_result['share_server'],
                         result['share_server'])

    def test_details(self):
        self.mock_object(db_api, 'share_server_get',
                         mock.Mock(return_value=fake_share_server_get()))
        result = self.controller.details(
            FakeRequestAdmin,
            fake_share_server_get_result['share_server']['id'])
        policy.check_policy.assert_called_once_with(
            CONTEXT, self.resource_name, 'details')
        db_api.share_server_get.assert_called_once_with(
            CONTEXT, fake_share_server_get_result['share_server']['id'])
        self.assertEqual(fake_share_server_backend_details_get_result,
                         result)

    def test_delete_active_server(self):
        share_server = FakeShareServer(status=constants.STATUS_ACTIVE)
        self.mock_object(db_api, 'share_server_get',
                         mock.Mock(return_value=share_server))
        self.mock_object(self.controller.share_api, 'delete_share_server')
        self.controller.delete(
            FakeRequestAdmin,
            fake_share_server_get_result['share_server']['id'])
        policy.check_policy.assert_called_once_with(
            CONTEXT, self.resource_name, 'delete')
        db_api.share_server_get.assert_called_once_with(
            CONTEXT, fake_share_server_get_result['share_server']['id'])
        self.controller.share_api.delete_share_server.assert_called_once_with(
            CONTEXT, share_server)

    def test_delete_error_server(self):
        share_server = FakeShareServer(status=constants.STATUS_ERROR)
        self.mock_object(db_api, 'share_server_get',
                         mock.Mock(return_value=share_server))
        self.mock_object(self.controller.share_api, 'delete_share_server')
        self.controller.delete(
            FakeRequestAdmin,
            fake_share_server_get_result['share_server']['id'])
        policy.check_policy.assert_called_once_with(
            CONTEXT, self.resource_name, 'delete')
        db_api.share_server_get.assert_called_once_with(
            CONTEXT, fake_share_server_get_result['share_server']['id'])
        self.controller.share_api.delete_share_server.assert_called_once_with(
            CONTEXT, share_server)

    def test_delete_used_server(self):
        share_server_id = fake_share_server_get_result['share_server']['id']

        def raise_not_share_server_in_use(*args, **kwargs):
            raise exception.ShareServerInUse(share_server_id=share_server_id)

        share_server = fake_share_server_get()
        self.mock_object(db_api, 'share_server_get',
                         mock.Mock(return_value=share_server))
        self.mock_object(self.controller.share_api, 'delete_share_server',
                         mock.Mock(side_effect=raise_not_share_server_in_use))
        self.assertRaises(exc.HTTPConflict,
                          self.controller.delete,
                          FakeRequestAdmin,
                          share_server_id)
        db_api.share_server_get.assert_called_once_with(CONTEXT,
                                                        share_server_id)
        self.controller.share_api.delete_share_server.assert_called_once_with(
            CONTEXT, share_server)

    def test_delete_not_found(self):
        share_server_id = fake_share_server_get_result['share_server']['id']

        def raise_not_found(*args, **kwargs):
            raise exception.ShareServerNotFound(
                share_server_id=share_server_id)

        self.mock_object(db_api, 'share_server_get',
                         mock.Mock(side_effect=raise_not_found))
        self.assertRaises(exc.HTTPNotFound,
                          self.controller.delete,
                          FakeRequestAdmin,
                          share_server_id)
        db_api.share_server_get.assert_called_once_with(
            CONTEXT, share_server_id)
        policy.check_policy.assert_called_once_with(
            CONTEXT, self.resource_name, 'delete')

    def test_delete_creating_server(self):
        share_server = FakeShareServer(status=constants.STATUS_CREATING)
        self.mock_object(db_api, 'share_server_get',
                         mock.Mock(return_value=share_server))
        self.assertRaises(exc.HTTPForbidden,
                          self.controller.delete,
                          FakeRequestAdmin,
                          share_server['id'])
        policy.check_policy.assert_called_once_with(
            CONTEXT,
            self.resource_name, 'delete')

    def test_delete_deleting_server(self):
        share_server = FakeShareServer(status=constants.STATUS_DELETING)
        self.mock_object(db_api, 'share_server_get',
                         mock.Mock(return_value=share_server))
        self.assertRaises(exc.HTTPForbidden,
                          self.controller.delete,
                          FakeRequestAdmin,
                          share_server['id'])
        policy.check_policy.assert_called_once_with(
            CONTEXT, self.resource_name, 'delete')