Commit 5648bc36 authored by Sofia Papagiannaki's avatar Sofia Papagiannaki
Browse files

pithos: Extend backend

Provide an additional backend method 'get_object_by_uuid'
which gets the object's UUID as an input
and return metadata and permission information
about the object.
parent 96c0df47
......@@ -1867,6 +1867,48 @@ class ModularBackend(BaseBackend):
account, container, name = path.split('/', 2)
self._delete_object(user, account, container, name)
@debug_method
@backend_method
def get_object_by_uuid(self, uuid, version=None, domain='pithos',
user=None, check_permissions=True):
"""Return information for the object identified by the specific UUID
Raises:
NameError: UUID or version was not found
NotAllowedError: if check_permissions is True and user has not
access to the object
AssertionError: if check_permissions is True but user
is provided
"""
if user is not None and not check_permissions:
raise AssertionError('Inconsistent argument combination:'
'if user is provided '
'permission check should be enforced.')
uuid_ = self._validate_uuid(uuid)
if version is None:
props = self.node.latest_uuid(uuid_, CLUSTER_NORMAL)
if props is None:
raise NameError('No object found for this UUID.')
path, _ = props
else:
props = self.node.version_get_properties(version,
keys=('uuid', 'node'))
if not props:
raise NameError('No such version was found.')
uuid_, node = props
assert uuid_ == uuid
_, path = self.node.node_get_properties(node)
account, container, name = path.split('/', 2)
if check_permissions:
self._can_read_object(user, account, container, name)
user_ = user if user is not None else account
meta = self.get_object_meta(user_, account, container, name,
domain=domain, version=version,
include_user_defined=True)
perms = self.permissions.access_get(path)
return meta, perms, path
@debug_method
@backend_method
def get_public(self, user, public):
......
......@@ -15,7 +15,7 @@
from .common import CommonMixin
from .quota import TestQuotaMixin
from .delete_by_uuid import TestDeleteByUUIDMixin
from .uuid_methods import TestUUIDMixin
from .snapshots import TestSnapshotsMixin
from sqlalchemy import create_engine
......@@ -23,7 +23,7 @@ from sqlalchemy import create_engine
import os
import time
class TestSQLAlchemyBackend(CommonMixin, TestDeleteByUUIDMixin,
class TestSQLAlchemyBackend(CommonMixin, TestUUIDMixin,
TestQuotaMixin, TestSnapshotsMixin):
db_module = 'pithos.backends.lib.sqlalchemy'
db_connection_str = '%(scheme)s://%(user)s:%(pwd)s@%(host)s:%(port)s/%(name)s'
......@@ -58,7 +58,7 @@ class TestSQLAlchemyBackend(CommonMixin, TestDeleteByUUIDMixin,
c.execute('drop database %s' % cls.name)
c.connection.connection.set_isolation_level(1)
class TestSQLiteBackend(CommonMixin, TestDeleteByUUIDMixin, TestQuotaMixin,
class TestSQLiteBackend(CommonMixin, TestUUIDMixin, TestQuotaMixin,
TestSnapshotsMixin):
db_module = 'pithos.backends.lib.sqlite'
db_connection = location = '/tmp/test_pithos_backend.db'
......
......@@ -27,7 +27,66 @@ get_random_data = lambda length: get_random_word(length)[:length]
get_random_name = partial(get_random_word, length=8)
class TestDeleteByUUIDMixin(object):
class TestUUIDMixin(object):
def test_get_object_by_uuid(self):
container = get_random_name()
obj = get_random_name()
t = self.account, self.account, container, obj
self.b.put_container(*t[:-1])
permissions = {'read': ['somebody_else']}
self.upload_object(*t, permissions=permissions)
self.b.update_object_meta(*t, domain='test1', meta={'domain': 'test1'})
meta = self.b.get_object_meta(*t, include_user_defined=False)
v1 = meta['version']
self.b.update_object_meta(*t, domain='test2', meta={'domain': 'test2'})
meta = self.b.get_object_meta(*t, include_user_defined=False)
uuid = meta['uuid']
v2 = meta['version']
meta, permissions_, path = self.b.get_object_by_uuid(
uuid, domain='test1', check_permissions=False)
self.assertTrue('domain' in meta)
self.assertEqual(meta['domain'], 'test1')
self.assertEqual(permissions_, permissions)
self.assertEqual(path, '/'.join(t[1:]))
meta, permissions_, path = self.b.get_object_by_uuid(
uuid, domain='test1', version=v2, check_permissions=False)
self.assertTrue('domain' in meta)
self.assertEqual(meta['domain'], 'test1')
self.assertEqual(permissions_, permissions)
self.assertEqual(path, '/'.join(t[1:]))
meta, permissions_, path = self.b.get_object_by_uuid(
uuid, domain='test2', version=v2, check_permissions=False)
self.assertTrue('domain' in meta)
self.assertEqual(meta['domain'], 'test2')
self.assertEqual(permissions_, permissions)
self.assertEqual(path, '/'.join(t[1:]))
meta, permissions_, path = self.b.get_object_by_uuid(
uuid, domain='test2', version=v1, check_permissions=False)
self.assertTrue('domain' not in meta)
self.assertEqual(permissions_, permissions)
self.assertEqual(path, '/'.join(t[1:]))
meta, permissions_, path = self.b.get_object_by_uuid(
uuid, domain='test1', user='somebody', check_permissions=True)
self.assertTrue('domain' in meta)
self.assertEqual(meta['domain'], 'test1')
self.assertEqual(permissions_, permissions)
self.assertEqual(path, '/'.join(t[1:]))
self.assertRaises(NotAllowedError,
self.b.get_object_by_uuid, uuid, user='not_allowed',
check_permissions=True)
self.assertRaises(AssertionError, self.b.get_object_by_uuid, uuid,
domain='test1', user='somebody',
check_permissions=False)
def test_delete_by_uuid(self):
self.assertRaises(ValueError, self.b.delete_by_uuid, self.account,
uuid=None)
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment