import os
import re
+import enum
from oslo_config import cfg
from oslo_log import log as logging
from six.moves import urllib
XML_NS_V1 = 'http://docs.openstack.org/api/openstack-block-storage/1.0/content'
XML_NS_V2 = 'http://docs.openstack.org/api/openstack-block-storage/2.0/content'
+METADATA_TYPES = enum.Enum('METADATA_TYPES', 'user image')
+
# Regex that matches alphanumeric characters, periods, hyphens,
# colons and underscores:
# under the License.
"""The Volume Image Metadata API extension."""
-
import logging
import six
+import webob
+from cinder.api import common
from cinder.api import extensions
from cinder.api.openstack import wsgi
from cinder.api import xmlutil
+from cinder import exception
+from cinder.i18n import _
from cinder import volume
super(VolumeImageMetadataController, self).__init__(*args, **kwargs)
self.volume_api = volume.API()
+ def _get_image_metadata(self, context, volume_id):
+ try:
+ volume = self.volume_api.get(context, volume_id)
+ meta = self.volume_api.get_volume_image_metadata(context, volume)
+ except exception.VolumeNotFound:
+ msg = _('Volume with volume id %s does not exist.') % volume_id
+ raise webob.exc.HTTPNotFound(explanation=msg)
+ return meta
+
def _get_all_images_metadata(self, context):
"""Returns the image metadata for all volumes."""
try:
image_meta = all_meta.get(vol['id'], {})
self._add_image_metadata(context, vol, image_meta)
+ @wsgi.action("os-set_image_metadata")
+ @wsgi.serializers(xml=common.MetadataTemplate)
+ @wsgi.deserializers(xml=common.MetadataDeserializer)
+ def create(self, req, id, body):
+ context = req.environ['cinder.context']
+ if authorize(context):
+ try:
+ metadata = body['os-set_image_metadata']['metadata']
+ except (KeyError, TypeError):
+ msg = _("Malformed request body.")
+ raise webob.exc.HTTPBadRequest(explanation=msg)
+ new_metadata = self._update_volume_image_metadata(context,
+ id,
+ metadata,
+ delete=False)
+
+ return {'metadata': new_metadata}
+
+ def _update_volume_image_metadata(self, context,
+ volume_id,
+ metadata,
+ delete=False):
+ try:
+ volume = self.volume_api.get(context, volume_id)
+ return self.volume_api.update_volume_metadata(
+ context,
+ volume,
+ metadata,
+ delete=False,
+ meta_type=common.METADATA_TYPES.image)
+ except exception.VolumeNotFound:
+ msg = _('Volume with volume id %s does not exist.') % volume_id
+ raise webob.exc.HTTPNotFound(explanation=msg)
+ except (ValueError, AttributeError):
+ msg = _("Malformed request body.")
+ raise webob.exc.HTTPBadRequest(explanation=msg)
+ except exception.InvalidVolumeMetadata as error:
+ raise webob.exc.HTTPBadRequest(explanation=error.msg)
+ except exception.InvalidVolumeMetadataSize as error:
+ raise webob.exc.HTTPRequestEntityTooLarge(explanation=error.msg)
+
+ @wsgi.action("os-unset_image_metadata")
+ def delete(self, req, id, body):
+ """Deletes an existing image metadata."""
+ context = req.environ['cinder.context']
+ if authorize(context):
+ try:
+ key = body['os-unset_image_metadata']['key']
+ except (KeyError, TypeError):
+ msg = _("Malformed request body.")
+ raise webob.exc.HTTPBadRequest(explanation=msg)
+
+ if key:
+ metadata = self._get_image_metadata(context, id)
+ if key not in metadata:
+ msg = _("Metadata item was not found.")
+ raise webob.exc.HTTPNotFound(explanation=msg)
+
+ try:
+ volume = self.volume_api.get(context, id)
+ self.volume_api.delete_volume_metadata(
+ context,
+ volume,
+ key,
+ meta_type=common.METADATA_TYPES.image)
+ except exception.VolumeNotFound:
+ msg = _('Volume does not exist.')
+ raise webob.exc.HTTPNotFound(explanation=msg)
+ return webob.Response(status_int=200)
+
class Volume_image_metadata(extensions.ExtensionDescriptor):
"""Show image metadata associated with the volume."""
delete=False):
try:
volume = self.volume_api.get(context, volume_id)
- return self.volume_api.update_volume_metadata(context,
- volume,
- metadata,
- delete)
+ return self.volume_api.update_volume_metadata(
+ context,
+ volume,
+ metadata,
+ delete,
+ meta_type=common.METADATA_TYPES.user)
except exception.VolumeNotFound as error:
raise webob.exc.HTTPNotFound(explanation=error.msg)
try:
volume = self.volume_api.get(context, volume_id)
- self.volume_api.delete_volume_metadata(context, volume, id)
+ self.volume_api.delete_volume_metadata(
+ context,
+ volume,
+ id,
+ meta_type=common.METADATA_TYPES.user)
except exception.VolumeNotFound as error:
raise webob.exc.HTTPNotFound(explanation=error.msg)
return webob.Response(status_int=200)
from oslo_db import concurrency as db_concurrency
from oslo_db import options as db_options
+from cinder.api import common
db_opts = [
cfg.BoolOpt('enable_new_services',
return IMPL.volume_metadata_get(context, volume_id)
-def volume_metadata_delete(context, volume_id, key):
+def volume_metadata_delete(context, volume_id, key,
+ meta_type=common.METADATA_TYPES.user):
"""Delete the given metadata item."""
- return IMPL.volume_metadata_delete(context, volume_id, key)
+ return IMPL.volume_metadata_delete(context, volume_id,
+ key, meta_type)
-def volume_metadata_update(context, volume_id, metadata, delete):
+def volume_metadata_update(context, volume_id, metadata,
+ delete, meta_type=common.METADATA_TYPES.user):
"""Update metadata if it exists, otherwise create it."""
- return IMPL.volume_metadata_update(context, volume_id, metadata, delete)
+ return IMPL.volume_metadata_update(context, volume_id, metadata,
+ delete, meta_type)
##################
from sqlalchemy.sql.expression import true
from sqlalchemy.sql import func
+from cinder.api import common
from cinder.common import sqlalchemyutils
from cinder.db.sqlalchemy import models
from cinder import exception
first()
if not result:
- raise notfound_exec(metadata_key=key, volume_id=volume_id)
+ if model is models.VolumeGlanceMetadata:
+ raise notfound_exec(id=volume_id)
+ else:
+ raise notfound_exec(metadata_key=key, volume_id=volume_id)
return result
models.VolumeMetadata, session=session)
+def _volume_image_metadata_get_query(context, volume_id, session=None):
+ return _volume_x_metadata_get_query(context, volume_id,
+ models.VolumeGlanceMetadata,
+ session=session)
+
+
@require_context
@require_volume_exists
def _volume_user_metadata_get(context, volume_id, session=None):
session=session)
+@require_context
+@require_volume_exists
+def _volume_image_metadata_update(context, volume_id, metadata, delete,
+ session=None):
+ return _volume_x_metadata_update(context, volume_id, metadata, delete,
+ models.VolumeGlanceMetadata,
+ exception.GlanceMetadataNotFound,
+ session=session)
+
+
@require_context
@require_volume_exists
def volume_metadata_get_item(context, volume_id, key):
@require_context
@require_volume_exists
@_retry_on_deadlock
-def volume_metadata_delete(context, volume_id, key):
- _volume_user_metadata_get_query(context, volume_id).\
- filter_by(key=key).\
- update({'deleted': True,
- 'deleted_at': timeutils.utcnow(),
- 'updated_at': literal_column('updated_at')})
+def volume_metadata_delete(context, volume_id, key, meta_type):
+ if meta_type == common.METADATA_TYPES.user:
+ (_volume_user_metadata_get_query(context, volume_id).
+ filter_by(key=key).
+ update({'deleted': True,
+ 'deleted_at': timeutils.utcnow(),
+ 'updated_at': literal_column('updated_at')}))
+ elif meta_type == common.METADATA_TYPES.image:
+ (_volume_image_metadata_get_query(context, volume_id).
+ filter_by(key=key).
+ update({'deleted': True,
+ 'deleted_at': timeutils.utcnow(),
+ 'updated_at': literal_column('updated_at')}))
+ else:
+ raise exception.InvalidMetadataType(metadata_type=meta_type,
+ id=volume_id)
@require_context
@require_volume_exists
@_retry_on_deadlock
-def volume_metadata_update(context, volume_id, metadata, delete):
- return _volume_user_metadata_update(context, volume_id, metadata, delete)
+def volume_metadata_update(context, volume_id, metadata, delete, meta_type):
+ if meta_type == common.METADATA_TYPES.user:
+ return _volume_user_metadata_update(context,
+ volume_id,
+ metadata,
+ delete)
+ elif meta_type == common.METADATA_TYPES.image:
+ return _volume_image_metadata_update(context,
+ volume_id,
+ metadata,
+ delete)
+ else:
+ raise exception.InvalidMetadataType(metadata_type=meta_type,
+ id=volume_id)
###################
message = _("Failed to copy metadata to volume: %(reason)s")
+class InvalidMetadataType(Invalid):
+ message = _("The type of metadata: %(metadata_type)s for volume/snapshot "
+ "%(id)s is invalid.")
+
+
class ImageCopyFailure(Invalid):
message = _("Failed to copy image to volume: %(reason)s")
import uuid
from xml.dom import minidom
+from oslo_serialization import jsonutils
from oslo_utils import timeutils
import webob
from cinder.api import common
+from cinder.api.contrib import volume_image_metadata
from cinder.api.openstack import wsgi
from cinder import db
+from cinder import exception
from cinder import test
from cinder.tests.unit.api import fakes
from cinder import volume
return {'fake': fake_image_metadata}
+def return_empty_image_metadata(*args, **kwargs):
+ return {}
+
+
+def volume_metadata_delete(context, volume_id, key, meta_type):
+ pass
+
+
+def fake_create_volume_metadata(context, volume_id, metadata,
+ delete, meta_type):
+ return fake_get_volume_image_metadata()
+
+
+def return_volume_nonexistent(*args, **kwargs):
+ raise exception.VolumeNotFound('bogus test message')
+
+
class VolumeImageMetadataTest(test.TestCase):
content_type = 'application/json'
fake_get_volumes_image_metadata)
self.stubs.Set(db, 'volume_get', fake_volume_get)
self.UUID = uuid.uuid4()
+ self.controller = (volume_image_metadata.
+ VolumeImageMetadataController())
def _make_request(self, url):
req = webob.Request.blank(url)
def test_get_volume(self):
res = self._make_request('/v2/fake/volumes/%s' % self.UUID)
- self.assertEqual(res.status_int, 200)
+ self.assertEqual(200, res.status_int)
self.assertEqual(self._get_image_metadata(res.body),
fake_image_metadata)
def test_list_detail_volumes(self):
res = self._make_request('/v2/fake/volumes/detail')
- self.assertEqual(res.status_int, 200)
+ self.assertEqual(200, res.status_int)
self.assertEqual(self._get_image_metadata_list(res.body)[0],
fake_image_metadata)
+ def test_create_image_metadata(self):
+ self.stubs.Set(volume.API, 'get_volume_image_metadata',
+ return_empty_image_metadata)
+ self.stubs.Set(db, 'volume_metadata_update',
+ fake_create_volume_metadata)
+
+ body = {"os-set_image_metadata": {"metadata": fake_image_metadata}}
+ req = webob.Request.blank('/v2/fake/volumes/1/action')
+ req.method = "POST"
+ req.body = jsonutils.dumps(body)
+ req.headers["content-type"] = "application/json"
+
+ res = req.get_response(fakes.wsgi_app())
+ self.assertEqual(200, res.status_int)
+ self.assertEqual(fake_image_metadata,
+ json.loads(res.body)["metadata"])
+
+ def test_create_with_keys_case_insensitive(self):
+ # If the keys in uppercase_and_lowercase, should return the one
+ # which server added
+ self.stubs.Set(volume.API, 'get_volume_image_metadata',
+ return_empty_image_metadata)
+ self.stubs.Set(db, 'volume_metadata_update',
+ fake_create_volume_metadata)
+
+ body = {
+ "os-set_image_metadata": {
+ "metadata": {
+ "Image_Id": "someid",
+ "image_name": "fake",
+ "Kernel_id": "somekernel",
+ "ramdisk_id": "someramdisk"
+ },
+ },
+ }
+
+ req = webob.Request.blank('/v2/fake/volumes/1/action')
+ req.method = 'POST'
+ req.body = jsonutils.dumps(body)
+ req.headers["content-type"] = "application/json"
+
+ res = req.get_response(fakes.wsgi_app())
+ self.assertEqual(200, res.status_int)
+ self.assertEqual(fake_image_metadata,
+ json.loads(res.body)["metadata"])
+
+ def test_create_empty_body(self):
+ req = fakes.HTTPRequest.blank('/v2/fake/volumes/1/action')
+ req.method = 'POST'
+ req.headers["content-type"] = "application/json"
+
+ self.assertRaises(webob.exc.HTTPBadRequest,
+ self.controller.create, req, 1, None)
+
+ def test_create_nonexistent_volume(self):
+ self.stubs.Set(volume.API, 'get', return_volume_nonexistent)
+
+ req = fakes.HTTPRequest.blank('/v2/fake/volumes/1/action')
+ req.method = 'POST'
+ req.content_type = "application/json"
+ body = {"os-set_image_metadata": {
+ "metadata": {"image_name": "fake"}}
+ }
+ req.body = jsonutils.dumps(body)
+ self.assertRaises(webob.exc.HTTPNotFound,
+ self.controller.create, req, 1, body)
+
+ def test_invalid_metadata_items_on_create(self):
+ self.stubs.Set(db, 'volume_metadata_update',
+ fake_create_volume_metadata)
+ req = fakes.HTTPRequest.blank('/v2/fake/volumes/1/action')
+ req.method = 'POST'
+ req.headers["content-type"] = "application/json"
+
+ data = {"os-set_image_metadata": {
+ "metadata": {"a" * 260: "value1"}}
+ }
+
+ # Test for long key
+ req.body = jsonutils.dumps(data)
+ self.assertRaises(webob.exc.HTTPRequestEntityTooLarge,
+ self.controller.create, req, 1, data)
+
+ # Test for long value
+ data = {"os-set_image_metadata": {
+ "metadata": {"key": "v" * 260}}
+ }
+ req.body = jsonutils.dumps(data)
+ self.assertRaises(webob.exc.HTTPRequestEntityTooLarge,
+ self.controller.create, req, 1, data)
+
+ # Test for empty key.
+ data = {"os-set_image_metadata": {
+ "metadata": {"": "value1"}}
+ }
+ req.body = jsonutils.dumps(data)
+ self.assertRaises(webob.exc.HTTPBadRequest,
+ self.controller.create, req, 1, data)
+
+ def test_delete(self):
+ self.stubs.Set(db, 'volume_metadata_delete',
+ volume_metadata_delete)
+
+ body = {"os-unset_image_metadata": {
+ "key": "ramdisk_id"}
+ }
+ req = webob.Request.blank('/v2/fake/volumes/1/action')
+ req.method = 'POST'
+ req.body = jsonutils.dumps(body)
+ req.headers["content-type"] = "application/json"
+
+ res = req.get_response(fakes.wsgi_app())
+ self.assertEqual(200, res.status_int)
+
+ def test_delete_meta_not_found(self):
+ data = {"os-unset_image_metadata": {
+ "key": "invalid_id"}
+ }
+ req = fakes.HTTPRequest.blank('/v2/fake/volumes/1/action')
+ req.method = 'POST'
+ req.body = jsonutils.dumps(data)
+ req.headers["content-type"] = "application/json"
+
+ self.assertRaises(webob.exc.HTTPNotFound,
+ self.controller.delete, req, 1, data)
+
+ def test_delete_nonexistent_volume(self):
+ self.stubs.Set(db, 'volume_metadata_delete',
+ return_volume_nonexistent)
+
+ body = {"os-unset_image_metadata": {
+ "key": "fake"}
+ }
+ req = fakes.HTTPRequest.blank('/v2/fake/volumes/1/action')
+ req.method = 'POST'
+ req.body = jsonutils.dumps(body)
+ req.headers["content-type"] = "application/json"
+
+ self.assertRaises(webob.exc.HTTPNotFound,
+ self.controller.delete, req, 1, body)
+
class ImageMetadataXMLDeserializer(common.MetadataXMLDeserializer):
metadata_node_name = "volume_image_metadata"
return stub_max_volume_metadata()
-def return_create_volume_metadata(context, volume_id, metadata, delete):
+def return_create_volume_metadata(context, volume_id, metadata, delete,
+ meta_type):
return stub_volume_metadata()
-def return_new_volume_metadata(context, volume_id, metadata, delete):
+def return_new_volume_metadata(context, volume_id, metadata,
+ delete, meta_type):
return stub_new_volume_metadata()
def return_create_volume_metadata_insensitive(context, snapshot_id,
- metadata, delete):
+ metadata, delete,
+ meta_type):
return stub_volume_metadata_insensitive()
return {}
-def return_empty_container_metadata(context, volume_id, metadata, delete):
+def return_empty_container_metadata(context, volume_id, metadata,
+ delete, meta_type):
return {}
-def delete_volume_metadata(context, volume_id, key):
+def delete_volume_metadata(context, volume_id, key, meta_type):
pass
return stub_max_volume_metadata()
-def return_create_volume_metadata(context, volume_id, metadata, delete):
+def return_create_volume_metadata(context, volume_id, metadata,
+ delete, meta_type):
return stub_volume_metadata()
-def return_new_volume_metadata(context, volume_id, metadata, delete):
+def return_new_volume_metadata(context, volume_id, metadata,
+ delete, meta_type):
return stub_new_volume_metadata()
def return_create_volume_metadata_insensitive(context, snapshot_id,
- metadata, delete):
+ metadata, delete,
+ meta_type):
return stub_volume_metadata_insensitive()
return {}
-def return_empty_container_metadata(context, volume_id, metadata, delete):
+def return_empty_container_metadata(context, volume_id, metadata,
+ delete, meta_type):
return {}
-def delete_volume_metadata(context, volume_id, key):
+def delete_volume_metadata(context, volume_id, key, meta_type):
pass
"volume:get": "rule:admin_or_owner",
"volume:get_all": "",
"volume:get_volume_metadata": "",
+ "volume:get_volume_image_metadata": "",
"volume:delete_volume_metadata": "",
"volume:update_volume_metadata": "",
"volume:get_volume_admin_metadata": "rule:admin_api",
import datetime
+import enum
from oslo_config import cfg
from oslo_utils import uuidutils
+from cinder.api import common
from cinder import context
from cinder import db
from cinder.db.sqlalchemy import api as sqlalchemy_api
self.assertEqual(should_be, db_meta)
+ def test_volume_metadata_update_with_metatype(self):
+ user_metadata1 = {'a': '1', 'c': '2'}
+ user_metadata2 = {'a': '3', 'd': '5'}
+ expected1 = {'a': '3', 'c': '2', 'd': '5'}
+ image_metadata1 = {'e': '1', 'f': '2'}
+ image_metadata2 = {'e': '3', 'g': '5'}
+ expected2 = {'e': '3', 'f': '2', 'g': '5'}
+ FAKE_METADATA_TYPE = enum.Enum('METADATA_TYPES', 'fake_type')
+
+ db.volume_create(self.ctxt, {'id': 1, 'metadata': user_metadata1})
+
+ # update user metatdata associated with volume.
+ db_meta = db.volume_metadata_update(
+ self.ctxt,
+ 1,
+ user_metadata2,
+ False,
+ meta_type=common.METADATA_TYPES.user)
+ self.assertEqual(expected1, db_meta)
+
+ # create image metatdata associated with volume.
+ db_meta = db.volume_metadata_update(
+ self.ctxt,
+ 1,
+ image_metadata1,
+ False,
+ meta_type=common.METADATA_TYPES.image)
+ self.assertEqual(image_metadata1, db_meta)
+
+ # update image metatdata associated with volume.
+ db_meta = db.volume_metadata_update(
+ self.ctxt,
+ 1,
+ image_metadata2,
+ False,
+ meta_type=common.METADATA_TYPES.image)
+ self.assertEqual(expected2, db_meta)
+
+ # update volume with invalid metadata type.
+ self.assertRaises(exception.InvalidMetadataType,
+ db.volume_metadata_update,
+ self.ctxt,
+ 1,
+ image_metadata1,
+ False,
+ FAKE_METADATA_TYPE.fake_type)
+
def test_volume_metadata_update_delete(self):
metadata1 = {'a': '1', 'c': '2'}
metadata2 = {'a': '3', 'd': '4'}
metadata.pop('c')
self.assertEqual(metadata, db.volume_metadata_get(self.ctxt, 1))
+ def test_volume_metadata_delete_with_metatype(self):
+ user_metadata = {'a': '1', 'c': '2'}
+ image_metadata = {'e': '1', 'f': '2'}
+ FAKE_METADATA_TYPE = enum.Enum('METADATA_TYPES', 'fake_type')
+
+ # test that user metadata deleted with meta_type specified.
+ db.volume_create(self.ctxt, {'id': 1, 'metadata': user_metadata})
+ db.volume_metadata_delete(self.ctxt, 1, 'c',
+ meta_type=common.METADATA_TYPES.user)
+ user_metadata.pop('c')
+ self.assertEqual(user_metadata, db.volume_metadata_get(self.ctxt, 1))
+
+ # update the image metadata associated with the volume.
+ db.volume_metadata_update(
+ self.ctxt,
+ 1,
+ image_metadata,
+ False,
+ meta_type=common.METADATA_TYPES.image)
+
+ # test that image metadata deleted with meta_type specified.
+ db.volume_metadata_delete(self.ctxt, 1, 'e',
+ meta_type=common.METADATA_TYPES.image)
+ image_metadata.pop('e')
+
+ # parse the result to build the dict.
+ rows = db.volume_glance_metadata_get(self.ctxt, 1)
+ result = {}
+ for row in rows:
+ result[row['key']] = row['value']
+ self.assertEqual(image_metadata, result)
+
+ # delete volume with invalid metadata type.
+ self.assertRaises(exception.InvalidMetadataType,
+ db.volume_metadata_delete,
+ self.ctxt,
+ 1,
+ 'f',
+ FAKE_METADATA_TYPE.fake_type)
+
def test_volume_glance_metadata_create(self):
volume = db.volume_create(self.ctxt, {'host': 'h1'})
db.volume_glance_metadata_create(self.ctxt, volume['id'],
import tempfile
import time
+import enum
import eventlet
import mock
from mox3 import mox
from stevedore import extension
from taskflow.engines.action_engine import engine
+from cinder.api import common
from cinder.backup import driver as backup_driver
from cinder.brick.local_dev import lvm as brick_lvm
from cinder.compute import nova
None,
test_meta)
+ def test_update_volume_metadata_with_metatype(self):
+ """Test update volume metadata with different metadata type."""
+ test_meta1 = {'fake_key1': 'fake_value1'}
+ test_meta2 = {'fake_key1': 'fake_value2'}
+ FAKE_METADATA_TYPE = enum.Enum('METADATA_TYPES', 'fake_type')
+ volume = tests_utils.create_volume(self.context, metadata=test_meta1,
+ **self.volume_params)
+ volume_id = volume['id']
+ self.volume.create_volume(self.context, volume_id)
+
+ volume_api = cinder.volume.api.API()
+
+ # update user metadata associated with the volume.
+ result_meta = volume_api.update_volume_metadata(
+ self.context,
+ volume,
+ test_meta2,
+ False,
+ common.METADATA_TYPES.user)
+ self.assertEqual(test_meta2, result_meta)
+
+ # create image metadata associated with the volume.
+ result_meta = volume_api.update_volume_metadata(
+ self.context,
+ volume,
+ test_meta1,
+ False,
+ common.METADATA_TYPES.image)
+ self.assertEqual(test_meta1, result_meta)
+
+ # update image metadata associated with the volume.
+ result_meta = volume_api.update_volume_metadata(
+ self.context,
+ volume,
+ test_meta2,
+ False,
+ common.METADATA_TYPES.image)
+ self.assertEqual(test_meta2, result_meta)
+
+ # update volume metadata with invalid metadta type.
+ self.assertRaises(exception.InvalidMetadataType,
+ volume_api.update_volume_metadata,
+ self.context,
+ volume,
+ test_meta1,
+ False,
+ FAKE_METADATA_TYPE.fake_type)
+
+ def test_delete_volume_metadata_with_metatype(self):
+ """Test delete volume metadata with different metadata type."""
+ test_meta1 = {'fake_key1': 'fake_value1', 'fake_key2': 'fake_value2'}
+ test_meta2 = {'fake_key1': 'fake_value1'}
+ FAKE_METADATA_TYPE = enum.Enum('METADATA_TYPES', 'fake_type')
+ volume = tests_utils.create_volume(self.context, metadata=test_meta1,
+ **self.volume_params)
+ volume_id = volume['id']
+ self.volume.create_volume(self.context, volume_id)
+
+ volume_api = cinder.volume.api.API()
+
+ # delete user metadata associated with the volume.
+ volume_api.delete_volume_metadata(
+ self.context,
+ volume,
+ 'fake_key2',
+ common.METADATA_TYPES.user)
+
+ self.assertEqual(test_meta2,
+ db.volume_metadata_get(self.context, volume_id))
+
+ # create image metadata associated with the volume.
+ result_meta = volume_api.update_volume_metadata(
+ self.context,
+ volume,
+ test_meta1,
+ False,
+ common.METADATA_TYPES.image)
+
+ self.assertEqual(test_meta1, result_meta)
+
+ # delete image metadata associated with the volume.
+ volume_api.delete_volume_metadata(
+ self.context,
+ volume,
+ 'fake_key2',
+ common.METADATA_TYPES.image)
+
+ # parse the result to build the dict.
+ rows = db.volume_glance_metadata_get(self.context, volume_id)
+ result = {}
+ for row in rows:
+ result[row['key']] = row['value']
+ self.assertEqual(test_meta2, result)
+
+ # delete volume metadata with invalid metadta type.
+ self.assertRaises(exception.InvalidMetadataType,
+ volume_api.delete_volume_metadata,
+ self.context,
+ volume,
+ 'fake_key1',
+ FAKE_METADATA_TYPE.fake_type)
+
def test_create_volume_uses_default_availability_zone(self):
"""Test setting availability_zone correctly during volume create."""
volume_api = cinder.volume.api.API()
from oslo_utils import uuidutils
import six
+from cinder.api import common
from cinder import context
from cinder.db import base
from cinder import exception
return dict(rv)
@wrap_check_policy
- def delete_volume_metadata(self, context, volume, key):
+ def delete_volume_metadata(self, context, volume,
+ key, meta_type=common.METADATA_TYPES.user):
"""Delete the given metadata item from a volume."""
- self.db.volume_metadata_delete(context, volume['id'], key)
+ self.db.volume_metadata_delete(context, volume['id'], key, meta_type)
LOG.info(_LI("Delete volume metadata completed successfully."),
resource=volume)
raise exception.InvalidVolumeMetadataSize(reason=msg)
@wrap_check_policy
- def update_volume_metadata(self, context, volume, metadata, delete=False):
+ def update_volume_metadata(self, context, volume,
+ metadata, delete=False,
+ meta_type=common.METADATA_TYPES.user):
"""Updates or creates volume metadata.
If delete is True, metadata items that are not specified in the
if delete:
_metadata = metadata
else:
- orig_meta = self.get_volume_metadata(context, volume)
+ if meta_type == common.METADATA_TYPES.user:
+ orig_meta = self.get_volume_metadata(context, volume)
+ elif meta_type == common.METADATA_TYPES.image:
+ try:
+ orig_meta = self.get_volume_image_metadata(context,
+ volume)
+ except exception.GlanceMetadataNotFound:
+ orig_meta = {}
+ else:
+ raise exception.InvalidMetadataType(metadata_type=meta_type,
+ id=volume['id'])
_metadata = orig_meta.copy()
_metadata.update(metadata)
self._check_metadata_properties(_metadata)
-
db_meta = self.db.volume_metadata_update(context, volume['id'],
- _metadata, delete)
+ _metadata,
+ delete,
+ meta_type)
# TODO(jdg): Implement an RPC call for drivers that may use this info
pbr<2.0,>=0.11
anyjson>=0.3.3
Babel>=1.3
+enum34;python_version=='2.7' or python_version=='2.6'
eventlet>=0.17.4
greenlet>=0.3.2
iso8601>=0.1.9
Code Review / openstack-build / cinder-build.git / commitdiff