Synnefo » Pithos

        raise RangeNotSatisfiable('Supplied offset is beyond object limits')

    if src_object:

        src_container, src_name = split_container_object_string(src_object)

        src_version = request.META.get('HTTP_X_SOURCE_VERSION')

        try:

            src_size, src_hashmap = backend.get_object_hashmap(request.user, v_account, src_container, src_name, src_version)

                      content_type=None, content_range=None,

                      content_encoding=None, content_disposition=None,

                      x_object_bytes=None, x_object_manifest=None,

        """updates an object"""

        account = account or self.account

        args = locals()

        for elem in ['self', 'container', 'object']:

            args.pop(elem)

        return OOS_Client.update_object(self, container, object, **args)

    def update_object_using_chunks(self, container, object, f=stdin,

        return self._chunked_transfer(path, 'POST', f, headers=headers,

                                      blocksize=blocksize)

    def delete_object(self, container, object, until=None, account=None):

        """deletes an object or the object history until the date provided"""

        account = account or self.account

from optparse import OptionParser

from os import environ

from sys import argv, exit, stdin, stdout

from datetime import datetime

import json

import logging

import time as _time

import os

_cli_commands = {}

def cli_command(*args):

    def __init__(self, name, argv):

        parser = OptionParser('%%prog %s [options] %s' % (name, self.syntax))

        parser.add_option('--host', dest='host', metavar='HOST',

        parser.add_option('--user', dest='user', metavar='USERNAME',

                          help='use account USERNAME')

        parser.add_option('--token', dest='token', metavar='AUTH',

                          help='use account AUTH')

        parser.add_option('--api', dest='api', metavar='API',

        parser.add_option('-v', action='store_true', dest='verbose',

                          default=False, help='use verbose output')

        parser.add_option('-d', action='store_true', dest='debug',

    for id, t in data['versions']:

        f.write('%s @ %s\n' % (str(id).rjust(30), datetime.fromtimestamp(t)))

def main():

    try:

        name = argv[1]

# or implied, of GRNET S.A.

from lib.client import Pithos_Client, Fault

from xml.dom import minidom

from StringIO import StringIO

import json

import time as _time

import types

import hashlib

import mimetypes

import random

import datetime

                "%A, %d-%b-%y %H:%M:%S GMT",

                "%a, %d %b %Y %H:%M:%S GMT"]

OTHER_ACCOUNTS = {

    '0001': 'verigak',

    '0002': 'chazapis',

class BaseTestCase(unittest.TestCase):

    #TODO unauthorized request

    def setUp(self):

        #self.headers = {

        #    'account': ('x-account-container-count',

        #                'x-account-bytes-used',

        self.assertEqual(10000, len(self.client.list_objects('pithos')))

    def test_list_empty_params(self):

        if objects:

            objects = objects.strip().split('\n')

        self.assertEqual(objects,

                                              self.obj['name'])

        self.assertEqual(content[0:dl], data)

        self.assertEqual(content[dl:fl], self.obj['data'][dl:fl])

class ObjectDelete(BaseTestCase):

    def setUp(self):

        #self.client.share_object('c', 'o2', self.o2_sharing)

    def test_listing(self):

                              self.o1_sharing_with[0],

                              self.o1_sharing_with[1],

        self.assertTrue('test' in self.other.list_shared_by_others())

class TestGreek(BaseTestCase):

        self.client.create_container('φάκελος')

        o = self.upload_random_data('φάκελος', 'ο1')

        self.client.share_object('φάκελος', 'ο1', ['test:σεφς'])

                            '0009',

                            'διογένης',

        self.assert_not_raises_fault(401, chef.retrieve_object_metadata,

        #check write access

        self.client.share_object('φάκελος', 'ο1', ['διογένης'], read=False)

        new_data = get_random_data()

        self.assert_not_raises_fault(401, chef.update_object,

                                     'φάκελος', 'ο1', StringIO(new_data),

        server_data = self.client.retrieve_object('φάκελος', 'ο1')

        self.assertEqual(server_data[:len(o['data'])], o['data'])

        self.assertEqual('', self.client.retrieve_object('φάκελος', 'άπαντα'))

    def test_update_from_another_object(self):

class TestPermissions(BaseTestCase):

    def setUp(self):

        o = self.upload_random_data('c', 'o')

        self.client.share_object('c', 'o', ['test:pithosdev'])

        for token, account in OTHER_ACCOUNTS.items():

            if account in self.authorized:

                self.assert_not_raises_fault(401, cl.retrieve_object_metadata,

            else:

                self.assert_raises_fault(401, cl.retrieve_object_metadata,

    #def test_write_access(self):

    #    chef.update_object('φάκελος', 'ο1', StringIO(new_data))

    #    self.assert_not_raises_fault(401, chef.update_object,

    #                                 'φάκελος', 'ο1', StringIO(new_data),

    #    

    #    server_data = self.client.retrieve_object('φάκελος', 'ο1')

    #    self.assertEqual(server_data[:len(o['data'])], o['data'])