¡@

Home 

OpenStack Study: test_volume_host_attribute.py

OpenStack Index

**** CubicPower OpenStack Study ****

# Copyright 2012 OpenStack Foundation

#

# 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 datetime

import json

import uuid

from lxml import etree

import webob

from cinder import context

from cinder import db

from cinder import test

from cinder.tests.api import fakes

from cinder import volume

**** CubicPower OpenStack Study ****

def fake_volume_get(*args, **kwargs):

    return {

        'id': 'fake',

        'host': 'host001',

        'status': 'available',

        'size': 5,

        'availability_zone': 'somewhere',

        'created_at': datetime.datetime.now(),

        'attach_status': None,

        'display_name': 'anothervolume',

        'display_description': 'Just another volume!',

        'volume_type_id': None,

        'snapshot_id': None,

        'project_id': 'fake',

        'migration_status': None,

        '_name_id': 'fake2',

    }

**** CubicPower OpenStack Study ****

def fake_volume_get_all(*args, **kwargs):

    return [fake_volume_get()]

**** CubicPower OpenStack Study ****

def app():

    # no auth, just let environ['cinder.context'] pass through

    api = fakes.router.APIRouter()

    mapper = fakes.urlmap.URLMap()

    mapper['/v2'] = api

    return mapper

**** CubicPower OpenStack Study ****

class VolumeHostAttributeTest(test.TestCase):

**** CubicPower OpenStack Study ****

    def setUp(self):

        super(VolumeHostAttributeTest, self).setUp()

        self.stubs.Set(volume.API, 'get', fake_volume_get)

        self.stubs.Set(volume.API, 'get_all', fake_volume_get_all)

        self.stubs.Set(db, 'volume_get', fake_volume_get)

        self.UUID = uuid.uuid4()

**** CubicPower OpenStack Study ****

    def test_get_volume_allowed(self):

        ctx = context.RequestContext('admin', 'fake', True)

        req = webob.Request.blank('/v2/fake/volumes/%s' % self.UUID)

        req.method = 'GET'

        req.environ['cinder.context'] = ctx

        res = req.get_response(app())

        vol = json.loads(res.body)['volume']

        self.assertEqual(vol['os-vol-host-attr:host'], 'host001')

**** CubicPower OpenStack Study ****

    def test_get_volume_unallowed(self):

        ctx = context.RequestContext('non-admin', 'fake', False)

        req = webob.Request.blank('/v2/fake/volumes/%s' % self.UUID)

        req.method = 'GET'

        req.environ['cinder.context'] = ctx

        res = req.get_response(app())

        vol = json.loads(res.body)['volume']

        self.assertNotIn('os-vol-host-attr:host', vol)

**** CubicPower OpenStack Study ****

    def test_list_detail_volumes_allowed(self):

        ctx = context.RequestContext('admin', 'fake', True)

        req = webob.Request.blank('/v2/fake/volumes/detail')

        req.method = 'GET'

        req.environ['cinder.context'] = ctx

        res = req.get_response(app())

        vol = json.loads(res.body)['volumes']

        self.assertEqual(vol[0]['os-vol-host-attr:host'], 'host001')

**** CubicPower OpenStack Study ****

    def test_list_detail_volumes_unallowed(self):

        ctx = context.RequestContext('non-admin', 'fake', False)

        req = webob.Request.blank('/v2/fake/volumes/detail')

        req.method = 'GET'

        req.environ['cinder.context'] = ctx

        res = req.get_response(app())

        vol = json.loads(res.body)['volumes']

        self.assertNotIn('os-vol-host-attr:host', vol[0])

**** CubicPower OpenStack Study ****

    def test_list_simple_volumes_no_host(self):

        ctx = context.RequestContext('admin', 'fake', True)

        req = webob.Request.blank('/v2/fake/volumes')

        req.method = 'GET'

        req.environ['cinder.context'] = ctx

        res = req.get_response(app())

        vol = json.loads(res.body)['volumes']

        self.assertNotIn('os-vol-host-attr:host', vol[0])

**** CubicPower OpenStack Study ****

    def test_get_volume_xml(self):

        ctx = context.RequestContext('admin', 'fake', True)

        req = webob.Request.blank('/v2/fake/volumes/%s' % self.UUID)

        req.method = 'GET'

        req.accept = 'application/xml'

        req.environ['cinder.context'] = ctx

        res = req.get_response(app())

        vol = etree.XML(res.body)

        host_key = ('{http://docs.openstack.org/volume/ext/'

                    'volume_host_attribute/api/v1}host')

        self.assertEqual(vol.get(host_key), 'host001')

**** CubicPower OpenStack Study ****

    def test_list_volumes_detail_xml(self):

        ctx = context.RequestContext('admin', 'fake', True)

        req = webob.Request.blank('/v2/fake/volumes/detail')

        req.method = 'GET'

        req.accept = 'application/xml'

        req.environ['cinder.context'] = ctx

        res = req.get_response(app())

        vol = list(etree.XML(res.body))[0]

        host_key = ('{http://docs.openstack.org/volume/ext/'

                    'volume_host_attribute/api/v1}host')

        self.assertEqual(vol.get(host_key), 'host001')