nova/nova/tests/unit/api/openstack/compute/test_cloudpipe.py

# Copyright 2011 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 uuid as uuid_lib

from oslo_utils import timeutils
from webob import exc

from nova.api.openstack.compute import cloudpipe as cloudpipe_v21
from nova.compute import utils as compute_utils
import nova.conf
from nova import exception
from nova import objects
from nova import test
from nova.tests.unit.api.openstack import fakes
from nova.tests.unit import fake_network
from nova.tests.unit import matchers
from nova.tests import uuidsentinel as uuids
from nova import utils

CONF = nova.conf.CONF


project_id = str(uuid_lib.uuid4().hex)
uuid = uuids.fake


def fake_vpn_instance():
    return objects.Instance(
        id=7, image_ref=CONF.cloudpipe.vpn_image_id, vm_state='active',
        created_at=timeutils.parse_strtime('1981-10-20T00:00:00.000000'),
        uuid=uuid, project_id=project_id)


def compute_api_get_all_empty(context, search_opts=None):
    return []


def compute_api_get_all(context, search_opts=None):
        return [fake_vpn_instance()]


def utils_vpn_ping(addr, port, timeout=0.05, session_id=None):
    return True


class CloudpipeTestV21(test.NoDBTestCase):
    cloudpipe = cloudpipe_v21
    url = '/v2/fake/os-cloudpipe'

    def setUp(self):
        super(CloudpipeTestV21, self).setUp()
        self.controller = self.cloudpipe.CloudpipeController()
        self.stubs.Set(self.controller.compute_api, "get_all",
                       compute_api_get_all_empty)
        self.stubs.Set(utils, 'vpn_ping', utils_vpn_ping)
        self.req = fakes.HTTPRequest.blank('')

    def test_cloudpipe_list_no_network(self):

        def fake_get_nw_info_for_instance(instance):
            return {}

        self.stubs.Set(compute_utils, "get_nw_info_for_instance",
                       fake_get_nw_info_for_instance)
        self.stubs.Set(self.controller.compute_api, "get_all",
                       compute_api_get_all)
        res_dict = self.controller.index(self.req)
        response = {'cloudpipes': [{'project_id': project_id,
                                    'instance_id': uuid,
                                    'created_at': '1981-10-20T00:00:00Z'}]}
        self.assertEqual(res_dict, response)

    def test_cloudpipe_list(self):

        def network_api_get(context, network_id):
            self.assertEqual(context.project_id, project_id)
            return {'vpn_public_address': '127.0.0.1',
                    'vpn_public_port': 22}

        def fake_get_nw_info_for_instance(instance):
            return fake_network.fake_get_instance_nw_info(self)

        self.stubs.Set(compute_utils, "get_nw_info_for_instance",
                       fake_get_nw_info_for_instance)
        self.stubs.Set(self.controller.network_api, "get",
                       network_api_get)
        self.stubs.Set(self.controller.compute_api, "get_all",
                       compute_api_get_all)
        res_dict = self.controller.index(self.req)
        response = {'cloudpipes': [{'project_id': project_id,
                                    'internal_ip': '192.168.1.100',
                                    'public_ip': '127.0.0.1',
                                    'public_port': 22,
                                    'state': 'running',
                                    'instance_id': uuid,
                                    'created_at': '1981-10-20T00:00:00Z'}]}
        self.assertThat(res_dict, matchers.DictMatches(response))

    def test_cloudpipe_create(self):
        def launch_vpn_instance(context):
            return ([fake_vpn_instance()], 'fake-reservation')

        self.stubs.Set(self.controller.cloudpipe, 'launch_vpn_instance',
                       launch_vpn_instance)
        body = {'cloudpipe': {'project_id': project_id}}
        res_dict = self.controller.create(self.req, body=body)

        response = {'instance_id': uuid}
        self.assertEqual(res_dict, response)

    def test_cloudpipe_create_no_networks(self):
        def launch_vpn_instance(context):
            raise exception.NoMoreNetworks

        self.stubs.Set(self.controller.cloudpipe, 'launch_vpn_instance',
                       launch_vpn_instance)
        body = {'cloudpipe': {'project_id': project_id}}
        req = fakes.HTTPRequest.blank(self.url)
        self.assertRaises(exc.HTTPBadRequest,
                          self.controller.create, req, body=body)

    def test_cloudpipe_create_already_running(self):
        def launch_vpn_instance(*args, **kwargs):
            self.fail("Method should not have been called")

        self.stubs.Set(self.controller.cloudpipe, 'launch_vpn_instance',
                       launch_vpn_instance)
        self.stubs.Set(self.controller.compute_api, "get_all",
                       compute_api_get_all)
        body = {'cloudpipe': {'project_id': project_id}}
        req = fakes.HTTPRequest.blank(self.url)
        res_dict = self.controller.create(req, body=body)
        response = {'instance_id': uuid}
        self.assertEqual(res_dict, response)

    def test_cloudpipe_create_with_bad_project_id_failed(self):
        body = {'cloudpipe': {'project_id': 'bad.project.id'}}
        req = fakes.HTTPRequest.blank(self.url)
        self.assertRaises(exception.ValidationError,
                          self.controller.create, req, body=body)


class CloudpipePolicyEnforcementV21(test.NoDBTestCase):

    def setUp(self):
        super(CloudpipePolicyEnforcementV21, self).setUp()
        self.controller = cloudpipe_v21.CloudpipeController()
        self.req = fakes.HTTPRequest.blank('')

    def _common_policy_check(self, func, *arg, **kwarg):
        rule_name = "os_compute_api:os-cloudpipe"
        rule = {rule_name: "project:non_fake"}
        self.policy.set_rules(rule)
        exc = self.assertRaises(
            exception.PolicyNotAuthorized, func, *arg, **kwarg)
        self.assertEqual(
            "Policy doesn't allow %s to be performed." % rule_name,
            exc.format_message())

    def test_list_policy_failed(self):
        self._common_policy_check(self.controller.index, self.req)

    def test_create_policy_failed(self):
        body = {'cloudpipe': {'project_id': uuid}}
        self._common_policy_check(self.controller.create, self.req, body=body)

    def test_update_policy_failed(self):
        body = {"configure_project": {'vpn_ip': '192.168.1.1',
                                      'vpn_port': 2000}}
        self._common_policy_check(
            self.controller.update, self.req, uuid, body=body)