def test_DELETE(self):
        """ Test swift.object_server.ObjectController.DELETE """
        req = Request.blank('/sda1/p/a/c',
                            environ={'REQUEST_METHOD': 'DELETE'})
        resp = self.object_controller.DELETE(req)
        self.assertEquals(resp.status_int, 400)

        req = Request.blank('/sda1/p/a/c/o',
                            environ={'REQUEST_METHOD': 'DELETE'})
        resp = self.object_controller.DELETE(req)
        self.assertEquals(resp.status_int, 400)
        # self.assertRaises(KeyError, self.object_controller.DELETE, req)

        timestamp = normalize_timestamp(time())
        req = Request.blank('/sda1/p/a/c/o',
                            environ={'REQUEST_METHOD': 'DELETE'},
                            headers={'X-Timestamp': timestamp})
        resp = self.object_controller.DELETE(req)
        self.assertEquals(resp.status_int, 404)

        sleep(.00001)
        timestamp = normalize_timestamp(time())
        req = Request.blank('/sda1/p/a/c/o', environ={'REQUEST_METHOD': 'PUT'},
                            headers={
                                'X-Timestamp': timestamp,
                                'Content-Type': 'application/octet-stream',
                                'Content-Length': '4',
                                })
        req.body = 'test'
        resp = self.object_controller.PUT(req)
        self.assertEquals(resp.status_int, 201)

        timestamp = normalize_timestamp(float(timestamp) - 1)
        req = Request.blank('/sda1/p/a/c/o',
                            environ={'REQUEST_METHOD': 'DELETE'},
                            headers={'X-Timestamp': timestamp})
        resp = self.object_controller.DELETE(req)
        self.assertEquals(resp.status_int, 204)
        objfile = os.path.join(self.testdir, 'sda1',
            storage_directory(object_server.DATADIR, 'p',
                              hash_path('a', 'c', 'o')),
            timestamp + '.ts')
        self.assert_(os.path.isfile(objfile))

        sleep(.00001)
        timestamp = normalize_timestamp(time())
        req = Request.blank('/sda1/p/a/c/o',
                            environ={'REQUEST_METHOD': 'DELETE'},
                            headers={'X-Timestamp': timestamp})
        resp = self.object_controller.DELETE(req)
        self.assertEquals(resp.status_int, 204)
        objfile = os.path.join(self.testdir, 'sda1',
            storage_directory(object_server.DATADIR, 'p',
                              hash_path('a', 'c', 'o')),
            timestamp + '.ts')
        self.assert_(os.path.isfile(objfile))

 def test_hash_path(self):
     # Yes, these tests are deliberately very fragile. We want to make sure
     # that if someones changes the results hash_path produces, they know it
     self.assertEquals(utils.hash_path("a"), "1c84525acb02107ea475dcd3d09c2c58")
     self.assertEquals(utils.hash_path("a", "c"), "33379ecb053aa5c9e356c68997cbb59e")
     self.assertEquals(utils.hash_path("a", "c", "o"), "06fbf0b514e5199dfc4e00f42eb5ea83")
     self.assertEquals(utils.hash_path("a", "c", "o", raw_digest=False), "06fbf0b514e5199dfc4e00f42eb5ea83")
     self.assertEquals(
         utils.hash_path("a", "c", "o", raw_digest=True),
         "\x06\xfb\xf0\xb5\x14\xe5\x19\x9d\xfcN\x00\xf4.\xb5\xea\x83",
     )
     self.assertRaises(ValueError, utils.hash_path, "a", object="o")

 def test_updates_storage_policy_index(self):
     device = 'dev'
     part = '9'
     object_parts = ('a', 'c', 'o')
     df = self._make_open_diskfile(device, part, *object_parts,
                                   policy_idx=1)
     object_hash = utils.hash_path(*object_parts)
     expected = df.get_metadata()
     self.sender.connection = FakeConnection()
     self.sender.job = {'device': device, 'partition': part,
                        'policy_idx': 1}
     self.sender.node = {}
     self.sender.send_list = [object_hash]
     self.sender.send_delete = mock.MagicMock()
     self.sender.send_put = mock.MagicMock()
     self.sender.response = FakeResponse(
         chunk_body=(
             ':UPDATES: START\r\n'
             ':UPDATES: END\r\n'))
     self.sender.updates()
     args, _kwargs = self.sender.send_put.call_args
     path, df = args
     self.assertEqual(path, '/a/c/o')
     self.assert_(isinstance(df, diskfile.DiskFile))
     self.assertEqual(expected, df.get_metadata())
     self.assertEqual(os.path.join(self.testdir, 'dev/objects-1/9/',
                                   object_hash[-3:], object_hash),
                      df._datadir)

 def test_updates_put(self):
     device = 'dev'
     part = '9'
     object_parts = ('a', 'c', 'o')
     df = self._make_open_diskfile(device, part, *object_parts)
     object_hash = utils.hash_path(*object_parts)
     expected = df.get_metadata()
     self.sender.connection = FakeConnection()
     self.sender.job = {'device': device, 'partition': part}
     self.sender.node = {}
     self.sender.send_list = [object_hash]
     self.sender.send_delete = mock.MagicMock()
     self.sender.send_put = mock.MagicMock()
     self.sender.response = FakeResponse(
         chunk_body=(
             ':UPDATES: START\r\n'
             ':UPDATES: END\r\n'))
     self.sender.updates()
     self.assertEqual(self.sender.send_delete.mock_calls, [])
     self.assertEqual(1, len(self.sender.send_put.mock_calls))
     args, _kwargs = self.sender.send_put.call_args
     path, df = args
     self.assertEqual(path, '/a/c/o')
     self.assert_(isinstance(df, diskfile.DiskFile))
     self.assertEqual(expected, df.get_metadata())
     # note that the put line isn't actually sent since we mock send_put;
     # send_put is tested separately.
     self.assertEqual(
         ''.join(self.sender.connection.sent),
         '11\r\n:UPDATES: START\r\n\r\n'
         'f\r\n:UPDATES: END\r\n\r\n')

 def test_updates_is_deleted(self):
     device = 'dev'
     part = '9'
     object_parts = ('a', 'c', 'o')
     df = self._make_open_diskfile(device, part, *object_parts)
     object_hash = utils.hash_path(*object_parts)
     delete_timestamp = utils.normalize_timestamp(time.time())
     df.delete(delete_timestamp)
     self.sender.connection = FakeConnection()
     self.sender.job = {'device': device, 'partition': part}
     self.sender.node = {}
     self.sender.send_list = [object_hash]
     self.sender.send_delete = mock.MagicMock()
     self.sender.send_put = mock.MagicMock()
     self.sender.response = FakeResponse(
         chunk_body=(
             ':UPDATES: START\r\n'
             ':UPDATES: END\r\n'))
     self.sender.updates()
     self.sender.send_delete.assert_called_once_with(
         '/a/c/o', delete_timestamp)
     self.assertEqual(self.sender.send_put.mock_calls, [])
     # note that the delete line isn't actually sent since we mock
     # send_delete; send_delete is tested separately.
     self.assertEqual(
         ''.join(self.sender.connection.sent),
         '11\r\n:UPDATES: START\r\n\r\n'
         'f\r\n:UPDATES: END\r\n\r\n')

    def __init__(self, path, device, partition, account, container, obj,
                 logger, disk_chunk_size=65536,
                 bytes_per_sync=(512 * 1024 * 1024),
                 iter_hook=None, threadpool=None, obj_dir='objects',
                 mount_check=False):
        if mount_check and not check_mount(path, device):
            raise DiskFileDeviceUnavailable()
        self.disk_chunk_size = disk_chunk_size
        self.bytes_per_sync = bytes_per_sync
        self.iter_hook = iter_hook
        self.name = '/' + '/'.join((account, container, obj))
        name_hash = hash_path(account, container, obj)
        self.datadir = join(
            path, device, storage_directory(obj_dir, partition, name_hash))
        self.device_path = join(path, device)
        self.tmpdir = join(path, device, 'tmp')
        self.logger = logger
        self._metadata = None
        self.data_file = None
        self._data_file_size = None
        self.fp = None
        self.iter_etag = None
        self.started_at_0 = False
        self.read_to_eof = False
        self.quarantined_dir = None
        self.suppress_file_closing = False
        self._verify_close = False
        self.threadpool = threadpool or ThreadPool(nthreads=0)

        # FIXME(clayg): this attribute is set after open and affects the
        # behavior of the class (i.e. public interface)
        self.keep_cache = False

 def test_update_send_delete(self):
     device = 'dev'
     part = '9'
     object_parts = ('a', 'c', 'o')
     df = self._make_open_diskfile(device, part, *object_parts)
     object_hash = utils.hash_path(*object_parts)
     delete_timestamp = utils.normalize_timestamp(time.time())
     df.delete(delete_timestamp)
     self.sender.connection = FakeConnection()
     self.sender.job = {'device': device, 'partition': part}
     self.sender.node = {}
     self.sender.send_list = [object_hash]
     self.sender.response = FakeResponse(
         chunk_body=(
             ':UPDATES: START\r\n'
             ':UPDATES: END\r\n'))
     self.sender.updates()
     self.assertEqual(
         ''.join(self.sender.connection.sent),
         '11\r\n:UPDATES: START\r\n\r\n'
         '30\r\n'
         'DELETE /a/c/o\r\n'
         'X-Timestamp: %s\r\n\r\n\r\n'
         'f\r\n:UPDATES: END\r\n\r\n'
         % delete_timestamp
     )

 def test_run_once_1(self):
     conf = dict(swift_dir=self.testdir, devices=self.devices,
                 mount_check='false', timeout='300', stats_interval='1')
     replicator = object_replicator.ObjectReplicator(conf)
     was_connector = object_replicator.http_connect
     object_replicator.http_connect = mock_http_connect(200)
     cur_part = '0'
     df = self.df_mgr.get_diskfile('sda', cur_part, 'a', 'c', 'o',
                                   policy_idx=1)
     mkdirs(df._datadir)
     f = open(os.path.join(df._datadir,
                           normalize_timestamp(time.time()) + '.data'),
              'wb')
     f.write('1234567890')
     f.close()
     ohash = hash_path('a', 'c', 'o')
     data_dir = ohash[-3:]
     whole_path_from = os.path.join(self.objects_1, cur_part, data_dir)
     process_arg_checker = []
     ring = replicator.get_object_ring(1)
     nodes = [node for node in
              ring.get_part_nodes(int(cur_part))
              if node['ip'] not in _ips()]
     rsync_mods = tuple(['%s::object/sda/objects-1/%s' %
                         (node['ip'], cur_part) for node in nodes])
     for node in nodes:
         process_arg_checker.append(
             (0, '', ['rsync', whole_path_from, rsync_mods]))
     with _mock_process(process_arg_checker):
         replicator.run_once()
     self.assertFalse(process_errors)
     object_replicator.http_connect = was_connector

    def setup_bad_zero_byte(self, with_ts=False):
        self.auditor = auditor.ObjectAuditor(self.conf)
        self.auditor.log_time = 0
        ts_file_path = ""
        if with_ts:

            name_hash = hash_path("a", "c", "o")
            dir_path = os.path.join(self.devices, "sda", storage_directory(DATADIR, "0", name_hash))
            ts_file_path = os.path.join(dir_path, "99999.ts")
            if not os.path.exists(dir_path):
                mkdirs(dir_path)
            fp = open(ts_file_path, "w")
            fp.close()

        etag = md5()
        with self.disk_file.mkstemp() as (fd, tmppath):
            etag = etag.hexdigest()
            metadata = {"ETag": etag, "X-Timestamp": str(normalize_timestamp(time.time())), "Content-Length": 10}
            self.disk_file.put(fd, tmppath, metadata)
            etag = md5()
            etag = etag.hexdigest()
            metadata["ETag"] = etag
            write_metadata(fd, metadata)
        if self.disk_file.data_file:
            return self.disk_file.data_file
        return ts_file_path

def print_ring_locations(ring, datadir, account, container=None):
    """
    print out ring locations of specified type

    :param ring: ring instance
    :param datadir: high level directory to store account/container/objects
    :param acount: account name
    :param container: container name
    """
    if ring is None or datadir is None or account is None:
        raise ValueError("None type")
    storage_type = "account"
    if container:
        storage_type = "container"
    try:
        part, nodes = ring.get_nodes(account, container, None)
    except (ValueError, AttributeError):
        raise ValueError("Ring error")
    else:
        path_hash = hash_path(account, container, None)
        print "\nRing locations:"
        for node in nodes:
            print (
                "  %s:%s - /srv/node/%s/%s/%s.db"
                % (node["ip"], node["port"], node["device"], storage_directory(datadir, part, path_hash), path_hash)
            )
        print "\nnote: /srv/node is used as default value of `devices`, the " "real value is set in the %s config file on each storage node." % storage_type

def print_ring_locations(ring, datadir, account, container=None):
    """
    print out ring locations of specified type

    :param ring: ring instance
    :param datadir: high level directory to store account/container/objects
    :param account: account name
    :param container: container name
    """
    if ring is None or datadir is None or account is None:
        raise ValueError('None type')
    storage_type = 'account'
    if container:
        storage_type = 'container'
    try:
        part, nodes = ring.get_nodes(account, container, None)
    except (ValueError, AttributeError):
        raise ValueError('Ring error')
    else:
        path_hash = hash_path(account, container, None)
        print '\nRing locations:'
        for node in nodes:
            print ('  %s:%s - /srv/node/%s/%s/%s.db' %
                   (node['ip'], node['port'], node['device'],
                    storage_directory(datadir, part, path_hash),
                    path_hash))
        print '\nnote: /srv/node is used as default value of `devices`, the ' \
            'real value is set in the %s config file on each storage node.' % \
            storage_type

 def test_PUT_user_metadata(self):
     timestamp = normalize_timestamp(time())
     req = Request.blank('/sda1/p/a/c/o', environ={'REQUEST_METHOD': 'PUT'},
             headers={'X-Timestamp': timestamp,
                      'Content-Type': 'text/plain',
                      'ETag': 'b114ab7b90d9ccac4bd5d99cc7ebb568',
                      'X-Object-Meta-1': 'One',
                      'X-Object-Meta-Two': 'Two'})
     req.body = 'VERIFY THREE'
     resp = self.object_controller.PUT(req)
     self.assertEquals(resp.status_int, 201)
     objfile = os.path.join(self.testdir, 'sda1',
         storage_directory(object_server.DATADIR, 'p',
                           hash_path('a', 'c', 'o')),
         timestamp + '.data')
     self.assert_(os.path.isfile(objfile))
     self.assertEquals(open(objfile).read(), 'VERIFY THREE')
     self.assertEquals(pickle.loads(getxattr(objfile,
     object_server.METADATA_KEY)),
                       {'X-Timestamp': timestamp,
                        'Content-Length': '12',
                        'ETag': 'b114ab7b90d9ccac4bd5d99cc7ebb568',
                        'Content-Type': 'text/plain',
                        'name': '/a/c/o',
                        'X-Object-Meta-1': 'One',
                        'X-Object-Meta-Two': 'Two'})

 def test_PUT_overwrite(self):
     req = Request.blank('/sda1/p/a/c/o', environ={'REQUEST_METHOD': 'PUT'},
             headers={'X-Timestamp': normalize_timestamp(time()),
                      'Content-Length': '6',
                      'Content-Type': 'application/octet-stream'})
     req.body = 'VERIFY'
     resp = self.object_controller.PUT(req)
     self.assertEquals(resp.status_int, 201)
     sleep(.00001)
     timestamp = normalize_timestamp(time())
     req = Request.blank('/sda1/p/a/c/o', environ={'REQUEST_METHOD': 'PUT'},
                         headers={'X-Timestamp': timestamp,
                                  'Content-Type': 'text/plain',
                                  'Content-Encoding': 'gzip'})
     req.body = 'VERIFY TWO'
     resp = self.object_controller.PUT(req)
     self.assertEquals(resp.status_int, 201)
     objfile = os.path.join(self.testdir, 'sda1',
         storage_directory(object_server.DATADIR, 'p',
                           hash_path('a', 'c', 'o')),
         timestamp + '.data')
     self.assert_(os.path.isfile(objfile))
     self.assertEquals(open(objfile).read(), 'VERIFY TWO')
     self.assertEquals(pickle.loads(getxattr(objfile,
                         object_server.METADATA_KEY)),
                       {'X-Timestamp': timestamp,
                        'Content-Length': '10',
                        'ETag': 'b381a4c5dab1eaa1eb9711fa647cd039',
                        'Content-Type': 'text/plain',
                        'name': '/a/c/o',
                        'Content-Encoding': 'gzip'})

 def test_run_once_recover_from_failure(self):
     replicator = object_replicator.ObjectReplicator(
         dict(swift_dir=self.testdir, devices=self.devices, mount_check="false", timeout="300", stats_interval="1")
     )
     was_connector = object_replicator.http_connect
     try:
         object_replicator.http_connect = mock_http_connect(200)
         # Write some files into '1' and run replicate- they should be moved
         # to the other partitoins and then node should get deleted.
         cur_part = "1"
         df = DiskFile(self.devices, "sda", cur_part, "a", "c", "o", FakeLogger())
         mkdirs(df.datadir)
         f = open(os.path.join(df.datadir, normalize_timestamp(time.time()) + ".data"), "wb")
         f.write("1234567890")
         f.close()
         ohash = hash_path("a", "c", "o")
         data_dir = ohash[-3:]
         whole_path_from = os.path.join(self.objects, cur_part, data_dir)
         process_arg_checker = []
         nodes = [node for node in self.ring.get_part_nodes(int(cur_part)) if node["ip"] not in _ips()]
         for node in nodes:
             rsync_mod = "%s::object/sda/objects/%s" % (node["ip"], cur_part)
             process_arg_checker.append((0, "", ["rsync", whole_path_from, rsync_mod]))
         self.assertTrue(os.access(os.path.join(self.objects, "1", data_dir, ohash), os.F_OK))
         with _mock_process(process_arg_checker):
             replicator.run_once()
         self.assertFalse(process_errors)
         for i, result in [("0", True), ("1", False), ("2", True), ("3", True)]:
             self.assertEquals(
                 os.access(os.path.join(self.objects, i, object_replicator.HASH_FILE), os.F_OK), result
             )
     finally:
         object_replicator.http_connect = was_connector

    def test_hash_suffix_multi_file_two(self):
        df = diskfile.DiskFile(self.devices, 'sda', '0', 'a', 'c', 'o',
                               FakeLogger())
        mkdirs(df.datadir)
        for tdiff in [1, 50, 100, 500]:
            suffs = ['.meta', '.data']
            if tdiff > 50:
                suffs.append('.ts')
            for suff in suffs:
                f = open(
                    os.path.join(
                        df.datadir,
                        normalize_timestamp(int(time()) - tdiff) + suff),
                    'wb')
                f.write('1234567890')
                f.close()

        ohash = hash_path('a', 'c', 'o')
        data_dir = ohash[-3:]
        whole_path_from = os.path.join(self.objects, '0', data_dir)
        hsh_path = os.listdir(whole_path_from)[0]
        whole_hsh_path = os.path.join(whole_path_from, hsh_path)

        diskfile.hash_suffix(whole_path_from, 99)
        # only the meta and data should be left
        self.assertEquals(len(os.listdir(whole_hsh_path)), 2)

    def get_nodes(self, account, container=None, obj=None):
        """
        Get the partition and nodes for an account/container/object.
        If a node is responsible for more than one replica, it will
        only appear in the output once.

        :param account: account name
        :param container: container name
        :param obj: object name
        :returns: a tuple of (partition, list of node dicts)

        Each node dict will have at least the following keys:

        ======  ===============================================================
        id      unique integer identifier amongst devices
        weight  a float of the relative weight of this device as compared to
                others; this indicates how many partitions the builder will try
                to assign to this device
        zone    integer indicating which zone the device is in; a given
                partition will not be assigned to multiple devices within the
                same zone
        ip      the ip address of the device
        port    the tcp port of the device
        device  the device's name on disk (sdb1, for example)
        meta    general use 'extra' field; for example: the online date, the
                hardware description
        ======  ===============================================================
        """
        key = hash_path(account, container, obj, raw_digest=True)
        if time() > self._rtime:
            self._reload()
        part = unpack_from('>I', key)[0] >> self._part_shift
        seen_ids = set()
        return part, [self.devs[r[part]] for r in self._replica2part2dev_id
                if not (r[part] in seen_ids or seen_ids.add(r[part]))]

    def test_run_once(self):
        replicator = object_replicator.ObjectReplicator(
            dict(swift_dir=self.testdir, devices=self.devices, mount_check="false", timeout="300", stats_interval="1")
        )
        was_connector = object_replicator.http_connect
        object_replicator.http_connect = mock_http_connect(200)
        cur_part = "0"
        df = DiskFile(self.devices, "sda", cur_part, "a", "c", "o", FakeLogger())
        mkdirs(df.datadir)
        f = open(os.path.join(df.datadir, normalize_timestamp(time.time()) + ".data"), "wb")
        f.write("1234567890")
        f.close()
        ohash = hash_path("a", "c", "o")
        data_dir = ohash[-3:]
        whole_path_from = os.path.join(self.objects, cur_part, data_dir)
        process_arg_checker = []
        nodes = [node for node in self.ring.get_part_nodes(int(cur_part)) if node["ip"] not in _ips()]
        for node in nodes:
            rsync_mod = "%s::object/sda/objects/%s" % (node["ip"], cur_part)
            process_arg_checker.append((0, "", ["rsync", whole_path_from, rsync_mod]))
        with _mock_process(process_arg_checker):
            replicator.run_once()
        self.assertFalse(process_errors)

        object_replicator.http_connect = was_connector

 def _get_account_broker(self, drive, part, account, **kwargs):
     hsh = hash_path(account)
     db_dir = storage_directory(DATADIR, part, hsh)
     db_path = os.path.join(self.root, drive, db_dir, hsh + ".db")
     kwargs.setdefault("account", account)
     kwargs.setdefault("logger", self.logger)
     return AccountBroker(db_path, **kwargs)

    def setup_bad_zero_byte(self, with_ts=False):
        self.auditor = auditor.ObjectAuditor(self.conf)
        self.auditor.log_time = 0
        ts_file_path = ''
        if with_ts:
            name_hash = hash_path('a', 'c', 'o')
            dir_path = os.path.join(
                self.devices, 'sda',
                storage_directory(get_data_dir(0), '0', name_hash))
            ts_file_path = os.path.join(dir_path, '99999.ts')
            if not os.path.exists(dir_path):
                mkdirs(dir_path)
            fp = open(ts_file_path, 'w')
            write_metadata(fp, {'X-Timestamp': '99999', 'name': '/a/c/o'})
            fp.close()

        etag = md5()
        with self.disk_file.create() as writer:
            etag = etag.hexdigest()
            metadata = {
                'ETag': etag,
                'X-Timestamp': str(normalize_timestamp(time.time())),
                'Content-Length': 10,
            }
            writer.put(metadata)
            etag = md5()
            etag = etag.hexdigest()
            metadata['ETag'] = etag
            write_metadata(writer._fd, metadata)
        return ts_file_path

    def test_run_once(self):
        replicator = object_replicator.ObjectReplicator(
            dict(swift_dir=self.testdir, devices=self.devices,
                mount_check='false', timeout='300', stats_interval='1'))
        was_connector = object_replicator.http_connect
        object_replicator.http_connect = mock_http_connect(200)
        cur_part = '0'
        df = DiskFile(self.devices, 'sda', cur_part, 'a', 'c', 'o',
                      FakeLogger())
        mkdirs(df.datadir)
        f = open(os.path.join(df.datadir,
                              normalize_timestamp(time.time()) + '.data'),
                 'wb')
        f.write('1234567890')
        f.close()
        ohash = hash_path('a', 'c', 'o')
        data_dir = ohash[-3:]
        whole_path_from = os.path.join(self.objects, cur_part, data_dir)
        process_arg_checker = []
        nodes = [node for node in
                 self.ring.get_part_nodes(int(cur_part))
                     if node['ip'] not in _ips()]
        for node in nodes:
            rsync_mod = '%s::object/sda/objects/%s' % (node['ip'], cur_part)
            process_arg_checker.append(
                (0, '', ['rsync', whole_path_from, rsync_mod]))
        with _mock_process(process_arg_checker):
            replicator.run_once()
        self.assertFalse(process_errors)

        object_replicator.http_connect = was_connector