def test_enable(self):
        broker = self._make_broker()
        broker.update_metadata({'X-Container-Sysmeta-Sharding':
                                (True, Timestamp.now().internal)})
        # no shard ranges
        out = StringIO()
        err = StringIO()
        with self.assertRaises(SystemExit):
            with mock.patch('sys.stdout', out), mock.patch('sys.stderr', err):
                main([broker.db_file, 'enable'])
        expected = ["WARNING: invalid shard ranges: ['No shard ranges.'].",
                    'Aborting.']
        self.assertEqual(expected, out.getvalue().splitlines())
        self.assertEqual(['Loaded db broker for a/c.'],
                         err.getvalue().splitlines())

        # success
        shard_ranges = []
        for data in self.shard_data:
            path = ShardRange.make_path(
                '.shards_a', 'c', 'c', Timestamp.now(), data['index'])
            shard_ranges.append(
                ShardRange(path, Timestamp.now(), data['lower'],
                           data['upper'], data['object_count']))
        broker.merge_shard_ranges(shard_ranges)
        out = StringIO()
        err = StringIO()
        with mock.patch('sys.stdout', out), mock.patch('sys.stderr', err):
            with mock_timestamp_now() as now:
                main([broker.db_file, 'enable'])
        expected = [
            "Container moved to state 'sharding' with epoch %s." %
            now.internal,
            'Run container-sharder on all nodes to shard the container.']
        self.assertEqual(expected, out.getvalue().splitlines())
        self.assertEqual(['Loaded db broker for a/c.'],
                         err.getvalue().splitlines())
        self._assert_enabled(broker, now)

        # already enabled
        out = StringIO()
        err = StringIO()
        with mock.patch('sys.stdout', out), mock.patch('sys.stderr', err):
            main([broker.db_file, 'enable'])
        expected = [
            "Container already in state 'sharding' with epoch %s." %
            now.internal,
            'No action required.',
            'Run container-sharder on all nodes to shard the container.']
        self.assertEqual(expected, out.getvalue().splitlines())
        self.assertEqual(['Loaded db broker for a/c.'],
                         err.getvalue().splitlines())
        self._assert_enabled(broker, now)

def main():
    parser = argparse.ArgumentParser(
        description=__doc__,
        formatter_class=argparse.RawTextHelpFormatter)
    parser.add_argument('--config', default='/etc/swift/internal-client.conf',
                        help=('internal-client config file '
                              '(default: /etc/swift/internal-client.conf'))
    parser.add_argument('--request-tries', type=int, default=3,
                        help='(default: 3)')
    parser.add_argument('account', help='account from which to delete')
    parser.add_argument('container', help='container from which to delete')
    parser.add_argument(
        '--prefix', default='',
        help='only delete objects with this prefix (default: none)')
    parser.add_argument(
        '--marker', default='',
        help='only delete objects after this marker (default: none)')
    parser.add_argument(
        '--end-marker', default='',
        help='only delete objects before this end-marker (default: none)')
    parser.add_argument(
        '--timestamp', type=Timestamp, default=Timestamp.now(),
        help='delete all objects as of this time (default: now)')
    args = parser.parse_args()

    swift = InternalClient(
        args.config, 'Swift Container Deleter', args.request_tries)
    for deleted, marker in mark_for_deletion(
            swift, args.account, args.container,
            args.marker, args.end_marker, args.prefix, args.timestamp):
        if marker is None:
            print('Finished. Marked %d objects for deletion.' % deleted)
        else:
            print('Marked %d objects for deletion, through %r' % (
                deleted, marker))

 def get_info(self):
     now = Timestamp.now().internal
     return {'container_count': 0,
             'object_count': 0,
             'bytes_used': 0,
             'created_at': now,
             'put_timestamp': now}

    def iter_task_to_expire(self, task_account_container_list,
                            my_index, divisor):
        """
        Yields task expire info dict which consists of task_account,
        task_container, task_object, timestamp_to_delete, and target_path
        """
        for task_account, task_container in task_account_container_list:
            for o in self.swift.iter_objects(task_account, task_container):
                task_object = o['name'].encode('utf8')
                try:
                    delete_timestamp, target_account, target_container, \
                        target_object = self.parse_task_obj(task_object)
                except ValueError:
                    self.logger.exception('Unexcepted error handling task %r' %
                                          task_object)
                    continue
                if delete_timestamp > Timestamp.now():
                    # we shouldn't yield the object that doesn't reach
                    # the expiration date yet.
                    break

                # Only one expirer daemon assigned for one task
                if self.hash_mod('%s/%s' % (task_container, task_object),
                                 divisor) != my_index:
                    continue

                yield {'task_account': task_account,
                       'task_container': task_container,
                       'task_object': task_object,
                       'target_path': '/'.join([
                           target_account, target_container, target_object]),
                       'delete_timestamp': delete_timestamp}

    def create_account_stat_table(self, conn, put_timestamp):
        """
        Create account_stat table which is specific to the account DB.
        Not a part of Pluggable Back-ends, internal to the baseline code.

        :param conn: DB connection object
        :param put_timestamp: put timestamp
        """
        conn.executescript("""
            CREATE TABLE account_stat (
                account TEXT,
                created_at TEXT,
                put_timestamp TEXT DEFAULT '0',
                delete_timestamp TEXT DEFAULT '0',
                container_count INTEGER,
                object_count INTEGER DEFAULT 0,
                bytes_used INTEGER DEFAULT 0,
                hash TEXT default '00000000000000000000000000000000',
                id TEXT,
                status TEXT DEFAULT '',
                status_changed_at TEXT DEFAULT '0',
                metadata TEXT DEFAULT ''
            );

            INSERT INTO account_stat (container_count) VALUES (0);
        """)

        conn.execute('''
            UPDATE account_stat SET account = ?, created_at = ?, id = ?,
                   put_timestamp = ?, status_changed_at = ?
            ''', (self.account, Timestamp.now().internal, str(uuid4()),
                  put_timestamp, put_timestamp))

 def _make_diskfile(self, device='dev', partition='9',
                    account='a', container='c', obj='o', body='test',
                    extra_metadata=None, policy=None,
                    frag_index=None, timestamp=None, df_mgr=None,
                    commit=True, verify=True):
     policy = policy or POLICIES.legacy
     object_parts = account, container, obj
     timestamp = Timestamp.now() if timestamp is None else timestamp
     if df_mgr is None:
         df_mgr = self.daemon._df_router[policy]
     df = df_mgr.get_diskfile(
         device, partition, *object_parts, policy=policy,
         frag_index=frag_index)
     write_diskfile(df, timestamp, data=body, extra_metadata=extra_metadata,
                    commit=commit)
     if commit and verify:
         # when we write and commit stub data, sanity check it's readable
         # and not quarantined because of any validation check
         with df.open():
             self.assertEqual(b''.join(df.reader()), body)
         # sanity checks
         listing = os.listdir(df._datadir)
         self.assertTrue(listing)
         for filename in listing:
             self.assertTrue(filename.startswith(timestamp.internal))
     return df

 def test_get_shard_ranges_invalid_shard_range(self):
     sr = ShardRange('a/c', Timestamp.now())
     bad_sr_data = dict(sr, name='bad_name')
     body = json.dumps([bad_sr_data]).encode('ascii')
     error_lines = self._check_get_shard_ranges_bad_data(body)
     self.assertIn('Failed to get shard ranges', error_lines[0])
     self.assertIn('ValueError', error_lines[0])
     self.assertFalse(error_lines[1:])

 def iter_task_containers_to_expire(self, task_account):
     """
     Yields task_container names under the task_account if the delete at
     timestamp of task_container is past.
     """
     for c in self.swift.iter_containers(task_account,
                                         prefix=self.task_container_prefix):
         task_container = str(c['name'])
         timestamp = self.delete_at_time_of_task_container(task_container)
         if timestamp > Timestamp.now():
             break
         yield task_container

 def test_get_shard_ranges_missing_record_type(self):
     base = Controller(self.app)
     req = Request.blank('/v1/a/c/o', method='PUT')
     sr = ShardRange('a/c', Timestamp.now())
     body = json.dumps([dict(sr)]).encode('ascii')
     with mocked_http_conn(
             200, 200, body_iter=iter([b'', body])):
         actual = base._get_shard_ranges(req, 'a', 'c', '1_test')
     self.assertIsNone(actual)
     error_lines = self.app.logger.get_lines_for_level('error')
     self.assertIn('Failed to get shard ranges', error_lines[0])
     self.assertIn('unexpected record type', error_lines[0])
     self.assertIn('/a/c', error_lines[0])
     self.assertFalse(error_lines[1:])

def gen_headers(hdrs_in=None, add_ts=True):
    """
    Get the headers ready for a request. All requests should have a User-Agent
    string, but if one is passed in don't over-write it. Not all requests will
    need an X-Timestamp, but if one is passed in do not over-write it.

    :param headers: dict or None, base for HTTP headers
    :param add_ts: boolean, should be True for any "unsafe" HTTP request

    :returns: HeaderKeyDict based on headers and ready for the request
    """
    hdrs_out = HeaderKeyDict(hdrs_in) if hdrs_in else HeaderKeyDict()
    if add_ts and 'X-Timestamp' not in hdrs_out:
        hdrs_out['X-Timestamp'] = Timestamp.now().internal
    if 'user-agent' not in hdrs_out:
        hdrs_out['User-Agent'] = 'direct-client %s' % os.getpid()
    return hdrs_out

    def test_direct_head_container_deleted(self):
        important_timestamp = Timestamp.now().internal
        headers = HeaderKeyDict({'X-Backend-Important-Timestamp':
                                 important_timestamp})

        with mocked_http_conn(404, headers) as conn:
            with self.assertRaises(ClientException) as raised:
                direct_client.direct_head_container(
                    self.node, self.part, self.account, self.container)
            self.assertEqual(conn.host, self.node['ip'])
            self.assertEqual(conn.port, self.node['port'])
            self.assertEqual(conn.method, 'HEAD')
            self.assertEqual(conn.path, self.container_path)

        self.assertEqual(conn.req_headers['user-agent'], self.user_agent)
        self.assertEqual(raised.exception.http_status, 404)
        self.assertEqual(raised.exception.http_headers, headers)

    def test_direct_head_object_not_found(self):
        important_timestamp = Timestamp.now().internal
        stub_headers = {'X-Backend-Important-Timestamp': important_timestamp}
        with mocked_http_conn(404, headers=stub_headers) as conn:
            with self.assertRaises(ClientException) as raised:
                direct_client.direct_head_object(
                    self.node, self.part, self.account, self.container,
                    self.obj)
            self.assertEqual(conn.host, self.node['ip'])
            self.assertEqual(conn.port, self.node['port'])
            self.assertEqual(conn.method, 'HEAD')
            self.assertEqual(conn.path, self.obj_path)

        self.assertEqual(raised.exception.http_status, 404)
        self.assertEqual(
            raised.exception.http_headers['x-backend-important-timestamp'],
            important_timestamp)

 def _handle_sync_response(self, node, response, info, broker, http,
                           different_region):
     parent = super(ContainerReplicator, self)
     if is_success(response.status):
         remote_info = json.loads(response.data)
         if incorrect_policy_index(info, remote_info):
             status_changed_at = Timestamp.now()
             broker.set_storage_policy_index(
                 remote_info['storage_policy_index'],
                 timestamp=status_changed_at.internal)
         sync_timestamps = ('created_at', 'put_timestamp',
                            'delete_timestamp')
         if any(info[key] != remote_info[key] for key in sync_timestamps):
             broker.merge_timestamps(*(remote_info[key] for key in
                                       sync_timestamps))
     rv = parent._handle_sync_response(
         node, response, info, broker, http, different_region)
     return rv

    def _get_synced_replication_info(self, broker, remote_info):
        """
        Sync the remote_info storage_policy_index if needed and return the
        newly synced replication info.

        :param broker: the database broker
        :param remote_info: the remote replication info

        :returns: local broker replication info
        """
        info = broker.get_replication_info()
        if incorrect_policy_index(info, remote_info):
            status_changed_at = Timestamp.now().internal
            broker.set_storage_policy_index(
                remote_info['storage_policy_index'],
                timestamp=status_changed_at)
            info = broker.get_replication_info()
        return info

    def merge_timestamps(self, created_at, put_timestamp, delete_timestamp):
        """
        Used in replication to handle updating timestamps.

        :param created_at: create timestamp
        :param put_timestamp: put timestamp
        :param delete_timestamp: delete timestamp
        """
        with self.get() as conn:
            old_status = self._is_deleted(conn)
            conn.execute('''
                UPDATE %s_stat SET created_at=MIN(?, created_at),
                                   put_timestamp=MAX(?, put_timestamp),
                                   delete_timestamp=MAX(?, delete_timestamp)
            ''' % self.db_type, (created_at, put_timestamp, delete_timestamp))
            if old_status != self._is_deleted(conn):
                timestamp = Timestamp.now()
                self._update_status_changed_at(conn, timestamp.internal)

            conn.commit()

    def test_direct_head_container_deleted(self):
        important_timestamp = Timestamp.now().internal
        headers = HeaderKeyDict({'X-Backend-Important-Timestamp':
                                 important_timestamp})

        with mocked_http_conn(404, headers) as conn:
            try:
                direct_client.direct_head_container(
                    self.node, self.part, self.account, self.container)
            except Exception as err:
                self.assertTrue(isinstance(err, ClientException))
            else:
                self.fail('ClientException not raised')
            self.assertEqual(conn.host, self.node['ip'])
            self.assertEqual(conn.port, self.node['port'])
            self.assertEqual(conn.method, 'HEAD')
            self.assertEqual(conn.path, self.container_path)

        self.assertEqual(conn.req_headers['user-agent'], self.user_agent)
        self.assertEqual(err.http_status, 404)
        self.assertEqual(err.http_headers, headers)

    def _handle_sync_response(self, node, response, info, broker, http,
                              different_region=False):
        if is_success(response.status):
            remote_info = json.loads(response.data)
            if incorrect_policy_index(info, remote_info):
                status_changed_at = Timestamp.now()
                broker.set_storage_policy_index(
                    remote_info['storage_policy_index'],
                    timestamp=status_changed_at.internal)
            sync_timestamps = ('created_at', 'put_timestamp',
                               'delete_timestamp')
            if any(info[key] != remote_info[key] for key in sync_timestamps):
                broker.merge_timestamps(*(remote_info[key] for key in
                                          sync_timestamps))

            if 'shard_max_row' in remote_info:
                # Grab remote's shard ranges, too
                self._fetch_and_merge_shard_ranges(http, broker)

        return super(ContainerReplicator, self)._handle_sync_response(
            node, response, info, broker, http, different_region)

 def test_replace(self):
     broker = self._make_broker()
     broker.update_metadata({'X-Container-Sysmeta-Sharding':
                             (True, Timestamp.now().internal)})
     input_file = os.path.join(self.testdir, 'shards')
     with open(input_file, 'wb') as fd:
         json.dump(self.shard_data, fd)
     out = StringIO()
     err = StringIO()
     with mock.patch('sys.stdout', out), mock.patch('sys.stderr', err):
         main([broker.db_file, 'replace', input_file])
     expected = [
         'No shard ranges found to delete.',
         'Injected 10 shard ranges.',
         'Run container-replicator to replicate them to other nodes.',
         'Use the enable sub-command to enable sharding.']
     self.assertEqual(expected, out.getvalue().splitlines())
     self.assertEqual(['Loaded db broker for a/c.'],
                      err.getvalue().splitlines())
     self.assertEqual(
         [(data['lower'], data['upper']) for data in self.shard_data],
         [(sr.lower_str, sr.upper_str) for sr in broker.get_shard_ranges()])

    def _backend_requests(self, req, n_outgoing, account_partition, accounts,
                          policy_index=None):
        additional = {'X-Timestamp': Timestamp.now().internal}
        if policy_index is None:
            additional['X-Backend-Storage-Policy-Default'] = \
                int(POLICIES.default)
        else:
            additional['X-Backend-Storage-Policy-Index'] = str(policy_index)
        headers = [self.generate_request_headers(req, transfer=True,
                                                 additional=additional)
                   for _junk in range(n_outgoing)]

        for i, account in enumerate(accounts):
            i = i % len(headers)

            headers[i]['X-Account-Partition'] = account_partition
            headers[i]['X-Account-Host'] = csv_append(
                headers[i].get('X-Account-Host'),
                '%(ip)s:%(port)s' % account)
            headers[i]['X-Account-Device'] = csv_append(
                headers[i].get('X-Account-Device'),
                account['device'])

        return headers

    def reap_object(self, account, container, container_partition,
                    container_nodes, obj, policy_index):
        """
        Deletes the given object by issuing a delete request to each node for
        the object. The format of the delete request is such that each object
        server will update a corresponding container server, removing the
        object from the container's listing.

        This function returns nothing and should raise no exception but only
        update various self.stats_* values for what occurs.

        :param account: The name of the account for the object.
        :param container: The name of the container for the object.
        :param container_partition: The partition for the container on the
                                    container ring.
        :param container_nodes: The primary node dicts for the container.
        :param obj: The name of the object to delete.
        :param policy_index: The storage policy index of the object's container

        * See also: :func:`swift.common.ring.Ring.get_nodes` for a description
          of the container node dicts.
        """
        cnodes = itertools.cycle(container_nodes)
        try:
            ring = self.get_object_ring(policy_index)
        except PolicyError:
            self.stats_objects_remaining += 1
            self.logger.increment('objects_remaining')
            return
        part, nodes = ring.get_nodes(account, container, obj)
        successes = 0
        failures = 0
        timestamp = Timestamp.now()

        for node in nodes:
            cnode = next(cnodes)
            try:
                direct_delete_object(
                    node, part, account, container, obj,
                    conn_timeout=self.conn_timeout,
                    response_timeout=self.node_timeout,
                    headers={'X-Container-Host': '%(ip)s:%(port)s' % cnode,
                             'X-Container-Partition': str(container_partition),
                             'X-Container-Device': cnode['device'],
                             'X-Backend-Storage-Policy-Index': policy_index,
                             'X-Timestamp': timestamp.internal})
                successes += 1
                self.stats_return_codes[2] = \
                    self.stats_return_codes.get(2, 0) + 1
                self.logger.increment('return_codes.2')
            except ClientException as err:
                if self.logger.getEffectiveLevel() <= DEBUG:
                    self.logger.exception(
                        _('Exception with %(ip)s:%(port)s/%(device)s'), node)
                failures += 1
                self.logger.increment('objects_failures')
                self.stats_return_codes[err.http_status // 100] = \
                    self.stats_return_codes.get(err.http_status // 100, 0) + 1
                self.logger.increment(
                    'return_codes.%d' % (err.http_status // 100,))
            except (Timeout, socket.error) as err:
                failures += 1
                self.logger.increment('objects_failures')
                self.logger.error(
                    _('Timeout Exception with %(ip)s:%(port)s/%(device)s'),
                    node)
            if successes > failures:
                self.stats_objects_deleted += 1
                self.logger.increment('objects_deleted')
            elif not successes:
                self.stats_objects_remaining += 1
                self.logger.increment('objects_remaining')
            else:
                self.stats_objects_possibly_remaining += 1
                self.logger.increment('objects_possibly_remaining')