def setUp(self):
     utils.HASH_PATH_SUFFIX = 'endcap'
     utils.HASH_PATH_PREFIX = ''
     # Setup a test ring (stolen from common/test_ring.py)
     self.testdir = tempfile.mkdtemp()
     self.devices = os.path.join(self.testdir, 'node')
     rmtree(self.testdir, ignore_errors=1)
     os.mkdir(self.testdir)
     os.mkdir(self.devices)
     os.mkdir(os.path.join(self.devices, 'sda'))
     self.objects = os.path.join(self.devices, 'sda',
                                 diskfile.get_data_dir(0))
     self.objects_1 = os.path.join(self.devices, 'sda',
                                   diskfile.get_data_dir(1))
     os.mkdir(self.objects)
     os.mkdir(self.objects_1)
     self.parts = {}
     self.parts_1 = {}
     for part in ['0', '1', '2', '3']:
         self.parts[part] = os.path.join(self.objects, part)
         os.mkdir(os.path.join(self.objects, part))
         self.parts_1[part] = os.path.join(self.objects_1, part)
         os.mkdir(os.path.join(self.objects_1, part))
     _create_test_rings(self.testdir)
     self.conf = dict(
         swift_dir=self.testdir, devices=self.devices, mount_check='false',
         timeout='300', stats_interval='1')
     self.replicator = object_replicator.ObjectReplicator(self.conf)
     self.replicator.logger = FakeLogger()
     self.df_mgr = diskfile.DiskFileManager(self.conf,
                                            self.replicator.logger)

    def setUp(self):
        self.testdir = os.path.join(mkdtemp(), 'tmp_test_object_auditor')
        self.devices = os.path.join(self.testdir, 'node')
        self.rcache = os.path.join(self.testdir, 'object.recon')
        self.logger = FakeLogger()
        rmtree(self.testdir, ignore_errors=1)
        mkdirs(os.path.join(self.devices, 'sda'))
        os.mkdir(os.path.join(self.devices, 'sdb'))

        # policy 0
        self.objects = os.path.join(self.devices, 'sda', get_data_dir(0))
        self.objects_2 = os.path.join(self.devices, 'sdb', get_data_dir(0))
        os.mkdir(self.objects)
        # policy 1
        self.objects_p1 = os.path.join(self.devices, 'sda', get_data_dir(1))
        self.objects_2_p1 = os.path.join(self.devices, 'sdb', get_data_dir(1))
        os.mkdir(self.objects_p1)

        self.parts = self.parts_p1 = {}
        for part in ['0', '1', '2', '3']:
            self.parts[part] = os.path.join(self.objects, part)
            self.parts_p1[part] = os.path.join(self.objects_p1, part)
            os.mkdir(os.path.join(self.objects, part))
            os.mkdir(os.path.join(self.objects_p1, part))

        self.conf = dict(
            devices=self.devices,
            mount_check='false',
            object_size_stats='10,100,1024,10240')
        self.df_mgr = DiskFileManager(self.conf, self.logger)

        # diskfiles for policy 0, 1
        self.disk_file = self.df_mgr.get_diskfile('sda', '0', 'a', 'c', 'o', 0)
        self.disk_file_p1 = self.df_mgr.get_diskfile('sda', '0', 'a', 'c',
                                                     'o', 1)

    def setUp(self):
        self.testdir = os.path.join(mkdtemp(), "tmp_test_object_auditor")
        self.devices = os.path.join(self.testdir, "node")
        self.rcache = os.path.join(self.testdir, "object.recon")
        self.logger = FakeLogger()
        rmtree(self.testdir, ignore_errors=1)
        mkdirs(os.path.join(self.devices, "sda"))
        os.mkdir(os.path.join(self.devices, "sdb"))

        # policy 0
        self.objects = os.path.join(self.devices, "sda", get_data_dir(POLICIES[0]))
        self.objects_2 = os.path.join(self.devices, "sdb", get_data_dir(POLICIES[0]))
        os.mkdir(self.objects)
        # policy 1
        self.objects_p1 = os.path.join(self.devices, "sda", get_data_dir(POLICIES[1]))
        self.objects_2_p1 = os.path.join(self.devices, "sdb", get_data_dir(POLICIES[1]))
        os.mkdir(self.objects_p1)

        self.parts = self.parts_p1 = {}
        for part in ["0", "1", "2", "3"]:
            self.parts[part] = os.path.join(self.objects, part)
            self.parts_p1[part] = os.path.join(self.objects_p1, part)
            os.mkdir(os.path.join(self.objects, part))
            os.mkdir(os.path.join(self.objects_p1, part))

        self.conf = dict(devices=self.devices, mount_check="false", object_size_stats="10,100,1024,10240")
        self.df_mgr = DiskFileManager(self.conf, self.logger)

        # diskfiles for policy 0, 1
        self.disk_file = self.df_mgr.get_diskfile("sda", "0", "a", "c", "o", policy=POLICIES[0])
        self.disk_file_p1 = self.df_mgr.get_diskfile("sda", "0", "a", "c", "o", policy=POLICIES[1])

 def rsync(self, node, job, suffixes):
     """
     Uses rsync to implement the sync method. This was the first
     sync method in Swift.
     """
     if not os.path.exists(job["path"]):
         return False
     args = [
         "rsync",
         "--recursive",
         "--whole-file",
         "--human-readable",
         "--xattrs",
         "--itemize-changes",
         "--ignore-existing",
         "--timeout=%s" % self.rsync_io_timeout,
         "--contimeout=%s" % self.rsync_io_timeout,
         "--bwlimit=%s" % self.rsync_bwlimit,
     ]
     node_ip = rsync_ip(node["replication_ip"])
     if self.vm_test_mode:
         rsync_module = "%s::object%s" % (node_ip, node["replication_port"])
     else:
         rsync_module = "%s::object" % node_ip
     had_any = False
     for suffix in suffixes:
         spath = join(job["path"], suffix)
         if os.path.exists(spath):
             args.append(spath)
             had_any = True
     if not had_any:
         return False
     data_dir = get_data_dir(job["policy_idx"])
     args.append(join(rsync_module, node["device"], data_dir, job["partition"]))
     return self._rsync(args) == 0

 def storage_dir(self, server, node, part=None, policy=None):
     policy = policy or self.policy
     device_path = self.device_dir(server, node)
     path_parts = [device_path, get_data_dir(policy)]
     if part is not None:
         path_parts.append(str(part))
     return os.path.join(*path_parts)

 def rsync(self, node, job, suffixes):
     """
     Uses rsync to implement the sync method. This was the first
     sync method in Swift.
     """
     if not os.path.exists(job["path"]):
         return False, {}
     args = [
         "rsync",
         "--recursive",
         "--whole-file",
         "--human-readable",
         "--xattrs",
         "--itemize-changes",
         "--ignore-existing",
         "--timeout=%s" % self.rsync_io_timeout,
         "--contimeout=%s" % self.rsync_io_timeout,
         "--bwlimit=%s" % self.rsync_bwlimit,
     ]
     if self.rsync_compress and job["region"] != node["region"]:
         # Allow for compression, but only if the remote node is in
         # a different region than the local one.
         args.append("--compress")
     rsync_module = rsync_module_interpolation(self.rsync_module, node)
     had_any = False
     for suffix in suffixes:
         spath = join(job["path"], suffix)
         if os.path.exists(spath):
             args.append(spath)
             had_any = True
     if not had_any:
         return False, {}
     data_dir = get_data_dir(job["policy"])
     args.append(join(rsync_module, node["device"], data_dir, job["partition"]))
     return self._rsync(args) == 0, {}

    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 object_key(policy_index, hashpath, timestamp='',
               extension='.data', nounce=''):
    storage_policy = diskfile.get_data_dir(policy_index)
    if timestamp:
        return '%s.%s.%s%s.%s' % (storage_policy, hashpath, timestamp,
                                  extension, nounce)
    else:
        # for use with getPrevious
        return '%s.%s/' % (storage_policy, hashpath)

    def collect_parts(self, override_devices=None, override_partitions=None):
        """
        Helper for yielding partitions in the top level reconstructor
        """
        override_devices = override_devices or []
        override_partitions = override_partitions or []
        ips = whataremyips(self.bind_ip)
        for policy in POLICIES:
            if policy.policy_type != EC_POLICY:
                continue
            self._diskfile_mgr = self._df_router[policy]
            self.load_object_ring(policy)
            data_dir = get_data_dir(policy)
            local_devices = itertools.ifilter(
                lambda dev: dev and is_local_device(ips, self.port, dev["replication_ip"], dev["replication_port"]),
                policy.object_ring.devs,
            )

            for local_dev in local_devices:
                if override_devices and (local_dev["device"] not in override_devices):
                    continue
                dev_path = self._df_router[policy].get_dev_path(local_dev["device"])
                if not dev_path:
                    self.logger.warn(_("%s is not mounted"), local_dev["device"])
                    continue
                obj_path = join(dev_path, data_dir)
                tmp_path = join(dev_path, get_tmp_dir(int(policy)))
                unlink_older_than(tmp_path, time.time() - self.reclaim_age)
                if not os.path.exists(obj_path):
                    try:
                        mkdirs(obj_path)
                    except Exception:
                        self.logger.exception("Unable to create %s" % obj_path)
                    continue
                try:
                    partitions = os.listdir(obj_path)
                except OSError:
                    self.logger.exception("Unable to list partitions in %r" % obj_path)
                    continue
                for partition in partitions:
                    part_path = join(obj_path, partition)
                    if not (partition.isdigit() and os.path.isdir(part_path)):
                        self.logger.warning("Unexpected entity in data dir: %r" % part_path)
                        remove_file(part_path)
                        continue
                    partition = int(partition)
                    if override_partitions and (partition not in override_partitions):
                        continue
                    part_info = {
                        "local_dev": local_dev,
                        "policy": policy,
                        "partition": partition,
                        "part_path": part_path,
                    }
                    yield part_info

def object_key(policy, hashpath, timestamp="", extension=".data", nonce="", frag_index=None):
    if frag_index is not None:
        frag_trailer = "-%s" % frag_index
    else:
        frag_trailer = ""
    storage_policy = diskfile.get_data_dir(policy)
    if timestamp:
        return "%s.%s.%s%s.%s%s" % (storage_policy, hashpath, timestamp, extension, nonce, frag_trailer)
    else:
        # for use with getPrevious
        return "%s.%s/" % (storage_policy, hashpath)

    def process_repl(self, policy, ips, override_devices=None,
                     override_partitions=None):
        """
        Helper function for collect_jobs to build jobs for replication
        using replication style storage policy
        """
        jobs = []
        obj_ring = self.get_object_ring(policy.idx)
        data_dir = get_data_dir(policy.idx)
        for local_dev in [dev for dev in obj_ring.devs
                          if (dev
                              and is_local_device(ips,
                                                  self.port,
                                                  dev['replication_ip'],
                                                  dev['replication_port'])
                              and (override_devices is None
                                   or dev['device'] in override_devices))]:
            dev_path = join(self.devices_dir, local_dev['device'])
            obj_path = join(dev_path, data_dir)
            tmp_path = join(dev_path, get_tmp_dir(int(policy)))
            if self.mount_check and not ismount(dev_path):
                self.logger.warn(_('%s is not mounted'), local_dev['device'])
                continue
            unlink_older_than(tmp_path, time.time() - self.reclaim_age)
            if not os.path.exists(obj_path):
                try:
                    mkdirs(obj_path)
                except Exception:
                    self.logger.exception('ERROR creating %s' % obj_path)
                continue
            for partition in os.listdir(obj_path):
                if (override_partitions is not None
                        and partition not in override_partitions):
                    continue

                try:
                    job_path = join(obj_path, partition)
                    part_nodes = obj_ring.get_part_nodes(int(partition))
                    nodes = [node for node in part_nodes
                             if node['id'] != local_dev['id']]
                    jobs.append(
                        dict(path=job_path,
                             device=local_dev['device'],
                             obj_path=obj_path,
                             nodes=nodes,
                             delete=len(nodes) > len(part_nodes) - 1,
                             policy_idx=policy.idx,
                             partition=partition,
                             object_ring=obj_ring,
                             region=local_dev['region']))
                except ValueError:
                    continue
        return jobs

 def _setup_data_file(self, container, obj, data):
     client.put_container(self.url, self.token, container, headers={"X-Storage-Policy": self.policy.name})
     client.put_object(self.url, self.token, container, obj, data)
     odata = client.get_object(self.url, self.token, container, obj)[-1]
     self.assertEquals(odata, data)
     opart, onodes = self.object_ring.get_nodes(self.account, container, obj)
     onode = onodes[0]
     node_id = (onode["port"] - 6000) / 10
     device = onode["device"]
     hash_str = hash_path(self.account, container, obj)
     obj_server_conf = readconf(self.configs["object-server"][node_id])
     devices = obj_server_conf["app:object-server"]["devices"]
     obj_dir = "%s/%s/%s/%s/%s/%s/" % (devices, device, get_data_dir(self.policy), opart, hash_str[-3:], hash_str)
     data_file = get_data_file_path(obj_dir)
     return onode, opart, data_file

    def process_repl(self, policy, jobs, ips):
        """
        Helper function for collect_jobs to build jobs for replication
        using replication style storage policy
        """
        obj_ring = self.get_object_ring(policy.idx)
        data_dir = get_data_dir(policy.idx)
        for local_dev in [dev for dev in obj_ring.devs
                          if dev and dev['replication_ip'] in ips and
                          dev['replication_port'] == self.port]:
            dev_path = join(self.devices_dir, local_dev['device'])
            obj_path = join(dev_path, data_dir)
            tmp_path = join(dev_path, get_tmp_dir(int(policy)))
            if self.mount_check and not ismount(dev_path):
                self.logger.warn(_('%s is not mounted'), local_dev['device'])
                continue
            unlink_older_than(tmp_path, time.time() - self.reclaim_age)
            if not os.path.exists(obj_path):
                try:
                    mkdirs(obj_path)
                except Exception:
                    self.logger.exception('ERROR creating %s' % obj_path)
                continue
            for partition in os.listdir(obj_path):
                try:
                    job_path = join(obj_path, partition)
                    if isfile(job_path):
                        # Clean up any (probably zero-byte) files where a
                        # partition should be.
                        self.logger.warning(
                            'Removing partition directory '
                            'which was a file: %s', job_path)
                        os.remove(job_path)
                        continue
                    part_nodes = obj_ring.get_part_nodes(int(partition))
                    nodes = [node for node in part_nodes
                             if node['id'] != local_dev['id']]
                    jobs.append(
                        dict(path=job_path,
                             device=local_dev['device'],
                             nodes=nodes,
                             delete=len(nodes) > len(part_nodes) - 1,
                             policy_idx=policy.idx,
                             partition=partition,
                             object_ring=obj_ring))

                except (ValueError, OSError):
                    continue

 def _setup_data_file(self, container, obj, data):
     client.put_container(self.url, self.token, container)
     client.put_object(self.url, self.token, container, obj, data)
     odata = client.get_object(self.url, self.token, container, obj)[-1]
     self.assertEquals(odata, data)
     opart, onodes = self.object_ring.get_nodes(
         self.account, container, obj)
     onode = onodes[0]
     node_id = (onode['port'] - 6000) / 10
     device = onode['device']
     hash_str = hash_path(self.account, container, obj)
     obj_server_conf = readconf(self.configs['object-server'][node_id])
     devices = obj_server_conf['app:object-server']['devices']
     obj_dir = '%s/%s/%s/%s/%s/%s/' % (devices, device,
                                       get_data_dir(self.policy.idx),
                                       opart, hash_str[-3:], hash_str)
     data_file = get_data_file_path(obj_dir)
     return onode, opart, data_file

 def process_repl(self, policy, jobs, ips):
     """
     Helper function for collect_jobs to build jobs for replication
     using replication style storage policy
     """
     obj_ring = self.get_object_ring(policy.idx)
     data_dir = get_data_dir(policy.idx)
     for local_dev in [
         dev
         for dev in obj_ring.devs
         if dev and dev["replication_ip"] in ips and dev["replication_port"] == self.port
     ]:
         dev_path = join(self.devices_dir, local_dev["device"])
         obj_path = join(dev_path, data_dir)
         tmp_path = join(dev_path, get_tmp_dir(int(policy)))
         if self.mount_check and not ismount(dev_path):
             self.logger.warn(_("%s is not mounted"), local_dev["device"])
             continue
         unlink_older_than(tmp_path, time.time() - self.reclaim_age)
         if not os.path.exists(obj_path):
             try:
                 mkdirs(obj_path)
             except Exception:
                 self.logger.exception("ERROR creating %s" % obj_path)
             continue
         for partition in os.listdir(obj_path):
             try:
                 job_path = join(obj_path, partition)
                 part_nodes = obj_ring.get_part_nodes(int(partition))
                 nodes = [node for node in part_nodes if node["id"] != local_dev["id"]]
                 jobs.append(
                     dict(
                         path=job_path,
                         device=local_dev["device"],
                         nodes=nodes,
                         delete=len(nodes) > len(part_nodes) - 1,
                         policy_idx=policy.idx,
                         partition=partition,
                         object_ring=obj_ring,
                     )
                 )
             except ValueError:
                 continue

 def rsync(self, node, job, suffixes):
     """
     Uses rsync to implement the sync method. This was the first
     sync method in Swift.
     """
     if not os.path.exists(job['path']):
         return False, {}
     args = [
         'rsync',
         '--recursive',
         '--whole-file',
         '--human-readable',
         '--xattrs',
         '--itemize-changes',
         '--ignore-existing',
         '--timeout=%s' % self.rsync_io_timeout,
         '--contimeout=%s' % self.rsync_io_timeout,
         '--bwlimit=%s' % self.rsync_bwlimit,
     ]
     if self.rsync_compress and \
             job['region'] != node['region']:
         # Allow for compression, but only if the remote node is in
         # a different region than the local one.
         args.append('--compress')
     node_ip = rsync_ip(node['replication_ip'])
     if self.vm_test_mode:
         rsync_module = '%s::object%s' % (node_ip, node['replication_port'])
     else:
         rsync_module = '%s::object' % node_ip
     had_any = False
     for suffix in suffixes:
         spath = join(job['path'], suffix)
         if os.path.exists(spath):
             args.append(spath)
             had_any = True
     if not had_any:
         return False, {}
     data_dir = get_data_dir(job['policy'])
     args.append(join(rsync_module, node['device'],
                 data_dir, job['partition']))
     return self._rsync(args) == 0, {}

 def rsync(self, node, job, suffixes):
     """
     Uses rsync to implement the sync method. This was the first
     sync method in Swift.
     """
     curframe = inspect.currentframe()
     calframe = inspect.getouterframes(curframe, 2)
     f = open("/home/hduser/log.txt","w")
     f.write(str(calframe))
     f.close()
     if not os.path.exists(job['path']):
         return False, set()
     args = [
         'rsync',
         '--recursive',
         '--whole-file',
         '--human-readable',
         '--xattrs',
         '--itemize-changes',
         '--ignore-existing',
         '--timeout=%s' % self.rsync_io_timeout,
         '--contimeout=%s' % self.rsync_io_timeout,
         '--bwlimit=%s' % self.rsync_bwlimit,
     ]
     node_ip = rsync_ip(node['replication_ip'])
     if self.vm_test_mode:
         rsync_module = '%s::object%s' % (node_ip, node['replication_port'])
     else:
         rsync_module = '%s::object' % node_ip
     had_any = False
     for suffix in suffixes:
         spath = join(job['path'], suffix)
         if os.path.exists(spath):
             args.append(spath)
             had_any = True
     if not had_any:
         return False, set()
     data_dir = get_data_dir(job['policy_idx'])
     args.append(join(rsync_module, node['device'],
                 data_dir, job['partition']))
     return self._rsync(args) == 0, set()

 def rsync(self, node, job, suffixes):
     """
     Uses rsync to implement the sync method. This was the first
     sync method in Swift.
     """
     if not os.path.exists(job['path']):
         return False, {}
     args = [
         'rsync',
         '--recursive',
         '--whole-file',
         '--human-readable',
         '--xattrs',
         '--itemize-changes',
         '--ignore-existing',
         '--timeout=%s' % self.rsync_io_timeout,
         '--contimeout=%s' % self.rsync_io_timeout,
         '--bwlimit=%s' % self.rsync_bwlimit,
         '--exclude=.*.%s' % ''.join('[0-9a-zA-Z]' for i in range(6))
     ]
     if self.rsync_compress and \
             job['region'] != node['region']:
         # Allow for compression, but only if the remote node is in
         # a different region than the local one.
         args.append('--compress')
     rsync_module = rsync_module_interpolation(self.rsync_module, node)
     had_any = False
     for suffix in suffixes:
         spath = join(job['path'], suffix)
         if os.path.exists(spath):
             args.append(spath)
             had_any = True
     if not had_any:
         return False, {}
     data_dir = get_data_dir(job['policy'])
     args.append(join(rsync_module, node['device'],
                 data_dir, job['partition']))
     return self._rsync(args) == 0, {}

def relink(swift_dir='/etc/swift',
           devices='/srv/node',
           skip_mount_check=False,
           logger=logging.getLogger()):
    mount_check = not skip_mount_check
    run = False
    relinked = errors = 0
    for policy in POLICIES:
        policy.object_ring = None  # Ensure it will be reloaded
        policy.load_ring(swift_dir)
        part_power = policy.object_ring.part_power
        next_part_power = policy.object_ring.next_part_power
        if not next_part_power or next_part_power == part_power:
            continue
        logging.info('Relinking files for policy %s under %s',
                     policy.name, devices)
        run = True
        locations = audit_location_generator(
            devices,
            diskfile.get_data_dir(policy),
            mount_check=mount_check)
        for fname, _, _ in locations:
            newfname = replace_partition_in_path(fname, next_part_power)
            try:
                diskfile.relink_paths(fname, newfname, check_existing=True)
                relinked += 1
            except OSError as exc:
                errors += 1
                logger.warning("Relinking %s to %s failed: %s",
                               fname, newfname, exc)

    if not run:
        logger.warning("No policy found to increase the partition power.")
        return 2
    logging.info('Relinked %d diskfiles (%d errors)', relinked, errors)
    if errors > 0:
        return 1
    return 0

    def test_main(self):
        # Create container
        container = 'container-%s' % uuid4()
        client.put_container(self.url, self.token, container,
                             headers={'X-Storage-Policy':
                                      self.policy.name})

        cpart, cnodes = self.container_ring.get_nodes(self.account, container)
        cnode = cnodes[0]
        obj = 'object-%s' % uuid4()
        opart, onodes = self.object_ring.get_nodes(
            self.account, container, obj)
        onode = onodes[0]

        # Kill one container/obj primary server
        kill_server(onode['port'], self.port2server, self.pids)

        # Delete the default data directory for objects on the primary server
        obj_dir = '%s/%s' % (self._get_objects_dir(onode),
                             get_data_dir(self.policy))
        shutil.rmtree(obj_dir, True)
        self.assertFalse(os.path.exists(obj_dir))

        # Create container/obj (goes to two primary servers and one handoff)
        client.put_object(self.url, self.token, container, obj, 'VERIFY')
        odata = client.get_object(self.url, self.token, container, obj)[-1]
        if odata != 'VERIFY':
            raise Exception('Object GET did not return VERIFY, instead it '
                            'returned: %s' % repr(odata))

        # Kill other two container/obj primary servers
        #  to ensure GET handoff works
        for node in onodes[1:]:
            kill_server(node['port'], self.port2server, self.pids)

        # Indirectly through proxy assert we can get container/obj
        odata = client.get_object(self.url, self.token, container, obj)[-1]
        if odata != 'VERIFY':
            raise Exception('Object GET did not return VERIFY, instead it '
                            'returned: %s' % repr(odata))
        # Restart those other two container/obj primary servers
        for node in onodes[1:]:
            start_server(node['port'], self.port2server, self.pids)
            self.assertFalse(os.path.exists(obj_dir))
            # We've indirectly verified the handoff node has the object, but
            # let's directly verify it.

        # Directly to handoff server assert we can get container/obj
        another_onode = next(self.object_ring.get_more_nodes(opart))
        odata = direct_client.direct_get_object(
            another_onode, opart, self.account, container, obj,
            headers={'X-Backend-Storage-Policy-Index': self.policy.idx})[-1]
        if odata != 'VERIFY':
            raise Exception('Direct object GET did not return VERIFY, instead '
                            'it returned: %s' % repr(odata))

        # Assert container listing (via proxy and directly) has container/obj
        objs = [o['name'] for o in
                client.get_container(self.url, self.token, container)[1]]
        if obj not in objs:
            raise Exception('Container listing did not know about object')
        timeout = time.time() + 5
        found_objs_on_cnode = []
        while time.time() < timeout:
            for cnode in [c for c in cnodes if cnodes not in
                          found_objs_on_cnode]:
                objs = [o['name'] for o in
                        direct_client.direct_get_container(
                            cnode, cpart, self.account, container)[1]]
                if obj in objs:
                    found_objs_on_cnode.append(cnode)
            if len(found_objs_on_cnode) >= len(cnodes):
                break
            time.sleep(0.3)
        if len(found_objs_on_cnode) < len(cnodes):
            missing = ['%s:%s' % (cnode['ip'], cnode['port']) for cnode in
                       cnodes if cnode not in found_objs_on_cnode]
            raise Exception('Container servers %r did not know about object' %
                            missing)

        # Bring the first container/obj primary server back up
        start_server(onode['port'], self.port2server, self.pids)

        # Assert that it doesn't have container/obj yet
        self.assertFalse(os.path.exists(obj_dir))
        try:
            direct_client.direct_get_object(
                onode, opart, self.account, container, obj, headers={
                    'X-Backend-Storage-Policy-Index': self.policy.idx})
        except ClientException as err:
            self.assertEquals(err.http_status, 404)
            self.assertFalse(os.path.exists(obj_dir))
        else:
            self.fail("Expected ClientException but didn't get it")

        try:
            port_num = onode['replication_port']
        except KeyError:
            port_num = onode['port']
        try:
#.........这里部分代码省略.........