Add function to update object metadata

Change-Id: Ibeff2aa7cafd62b932f81fa17449657d9d83b19e
This commit is contained in:
Caleb Boylan 2016-05-26 15:25:20 -07:00
parent c726bc16e3
commit 92a3c9ea17
3 changed files with 38 additions and 1 deletions

View File

@ -546,7 +546,7 @@ class ObjectCreate(task_manager.Task):
class ObjectUpdate(task_manager.Task):
def main(self, client):
client.swift_client.post_object(**self.args)
return client.swift_client.post_object(**self.args)
class ObjectList(task_manager.Task):

View File

@ -4684,6 +4684,37 @@ class OpenStackCloud(object):
raise OpenStackCloudException(
'Failed at action ({action}) [{error}]:'.format(**r))
def update_object(self, container, name, metadata=None, **headers):
"""Update the metadtata of an object
:param container: The name of the container the object is in
:param name: Name for the object within the container.
:param headers: These will be passed through to the object update
API as HTTP Headers.
:param metadata: This dict will get changed into headers that set
metadata of the object
:raises: ``OpenStackCloudException`` on operation error.
"""
if not metadata:
metadata = {}
metadata_headers = {}
for (k, v) in metadata.items():
metadata_headers['x-object-meta-' + k] = v
headers = dict(headers, **metadata_headers)
try:
return self.manager.submitTask(
_tasks.ObjectUpdate(container=container, obj=name,
headers=headers))
except swift_exceptions.ClientException as e:
raise OpenStackCloudException(
"Object update failed: %s (%s/%s)" % (
e.http_reason, e.http_host, e.http_path))
def list_objects(self, container, full_listing=True):
try:
return self.manager.submitTask(_tasks.ObjectList(

View File

@ -66,6 +66,12 @@ class TestObject(base.BaseFunctionalTestCase):
'bar', self.demo_cloud.get_object_metadata(
container_name, name)['x-object-meta-foo']
)
self.demo_cloud.update_object(container=container_name, name=name,
metadata={'testk': 'testv'})
self.assertEqual(
'testv', self.demo_cloud.get_object_metadata(
container_name, name)['x-object-meta-testk']
)
self.assertIsNotNone(
self.demo_cloud.get_object(container_name, name))
self.assertEqual(