def test_scope_queue_name(self):
        self.assertEqual(utils.scope_queue_name('my-q'), '/my-q')
        self.assertEqual(utils.scope_queue_name('my-q', None), '/my-q')
        self.assertEqual(utils.scope_queue_name('my-q', '123'), '123/my-q')

        self.assertEqual(utils.scope_queue_name(None), '/')
        self.assertEqual(utils.scope_queue_name(None, '123'), '123/')

    def _count(self, queue_name, project=None, include_claimed=False):
        """Return total number of messages in a queue.

        This method is designed to very quickly count the number
        of messages in a given queue. Expired messages are not
        counted, of course. If the queue does not exist, the
        count will always be 0.

        Note: Some expired messages may be included in the count if
            they haven't been GC'd yet. This is done for performance.
        """
        query = {
            # Messages must belong to this queue and project.
            PROJ_QUEUE: utils.scope_queue_name(queue_name, project),

            # NOTE(kgriffs): Messages must be finalized (i.e., must not
            # be part of an unfinalized transaction).
            #
            # See also the note wrt 'tx' within the definition
            # of ACTIVE_INDEX_FIELDS.
            'tx': None,
        }

        if not include_claimed:
            # Exclude messages that are claimed
            query['c.e'] = {'$lte': timeutils.utcnow_ts()}

        collection = self._collection(queue_name, project)
        return collection.find(query).hint(COUNTING_INDEX_FIELDS).count()

    def _claimed(self, queue_name, claim_id,
                 expires=None, limit=None, project=None):

        if claim_id is None:
            claim_id = {'$ne': None}

        query = {
            PROJ_QUEUE: utils.scope_queue_name(queue_name, project),
            'c.id': claim_id,
            'c.e': {'$gt': expires or timeutils.utcnow_ts()},
        }

        # NOTE(kgriffs): Claimed messages bust be queried from
        # the primary to avoid a race condition caused by the
        # multi-phased "create claim" algorithm.
        preference = pymongo.read_preferences.ReadPreference.PRIMARY
        collection = self._collection(queue_name, project)
        msgs = collection.find(query, sort=[('k', 1)],
                               read_preference=preference).hint(
                                   CLAIMED_INDEX_FIELDS
                               )

        if limit is not None:
            msgs = msgs.limit(limit)

        now = timeutils.utcnow_ts()

        def denormalizer(msg):
            doc = _basic_message(msg, now)
            doc['claim'] = msg['c']

            return doc

        return utils.HookedCursor(msgs, denormalizer)

    def bulk_delete(self, queue_name, message_ids, project=None):
        message_ids = [mid for mid in map(utils.to_oid, message_ids) if mid]
        query = {
            '_id': {'$in': message_ids},
            PROJ_QUEUE: utils.scope_queue_name(queue_name, project),
        }

        collection = self._collection(queue_name, project)
        collection.remove(query, w=0)

    def get(self, project, queue):
        fields = {'_id': 0}
        key = utils.scope_queue_name(queue, project)
        entry = self._col.find_one({PRIMARY_KEY: key},
                                   fields=fields)

        if entry is None:
            raise errors.QueueNotMapped(queue, project)

        return _normalize(entry)

    def _purge_queue(self, queue_name, project=None):
        """Removes all messages from the queue.

        Warning: Only use this when deleting the queue; otherwise
        you can cause a side-effect of reseting the marker counter
        which can cause clients to miss tons of messages.

        If the queue does not exist, this method fails silently.

        :param queue_name: name of the queue to purge
        :param project: ID of the project to which the queue belongs
        """
        scope = utils.scope_queue_name(queue_name, project)
        collection = self._collection(queue_name, project)
        collection.remove({PROJ_QUEUE: scope}, w=0)

    def create(self, name, project=None):
        try:
            # NOTE(kgriffs): Start counting at 1, and assume the first
            # message ever posted will succeed and set t to a UNIX
            # "modified at" timestamp.
            counter = {'v': 1, 't': 0}

            scoped_name = utils.scope_queue_name(name, project)
            self._collection.insert({'p_q': scoped_name, 'm': {},
                                     'c': counter})

        except pymongo.errors.DuplicateKeyError:
            return False
        else:
            return True

    def delete(self, queue_name, message_id, project=None, claim=None):
        # NOTE(cpp-cabrera): return early - this is an invalid message
        # id so we won't be able to find it any way
        mid = utils.to_oid(message_id)
        if mid is None:
            return

        collection = self._collection(queue_name, project)

        query = {
            '_id': mid,
            PROJ_QUEUE: utils.scope_queue_name(queue_name, project),
        }

        # NOTE(cpp-cabrera): return early - the user gaves us an
        # invalid claim id and that renders the rest of this
        # request moot
        cid = utils.to_oid(claim)
        if cid is None:
            return

        now = timeutils.utcnow_ts()
        cursor = collection.find(query).hint(ID_INDEX_FIELDS)

        try:
            message = next(cursor)
        except StopIteration:
            return

        is_claimed = (message['c']['id'] is not None and
                      message['c']['e'] > now)

        if claim is None:
            if is_claimed:
                raise errors.MessageIsClaimed(message_id)

        else:
            if message['c']['id'] != cid:
                # NOTE(kgriffs): Read from primary in case the message
                # was just barely claimed, and claim hasn't made it to
                # the secondary.
                pref = pymongo.read_preferences.ReadPreference.PRIMARY
                message = collection.find_one(query, read_preference=pref)

                if message['c']['id'] != cid:
                    raise errors.MessageIsClaimedBy(message_id, claim)

        collection.remove(query['_id'], w=0)

    def get(self, queue_name, message_id, project=None):
        mid = utils.to_oid(message_id)
        if mid is None:
            raise errors.MessageDoesNotExist(message_id, queue_name, project)

        now = timeutils.utcnow_ts()

        query = {"_id": mid, PROJ_QUEUE: utils.scope_queue_name(queue_name, project)}

        collection = self._collection(queue_name, project)
        message = list(collection.find(query).limit(1).hint(ID_INDEX_FIELDS))

        if not message:
            raise errors.MessageDoesNotExist(message_id, queue_name, project)

        return _basic_message(message[0], now)

    def _unclaim(self, queue_name, claim_id, project=None):
        cid = utils.to_oid(claim_id)

        # NOTE(cpp-cabrera): early abort - avoid a DB query if we're handling
        # an invalid ID
        if cid is None:
            return

        # NOTE(cpp-cabrera):  unclaim by setting the claim ID to None
        # and the claim expiration time to now
        now = timeutils.utcnow_ts()
        scope = utils.scope_queue_name(queue_name, project)
        collection = self._collection(queue_name, project)

        collection.update({PROJ_QUEUE: scope, 'c.id': cid},
                          {'$set': {'c': {'id': None, 'e': now}}},
                          upsert=False, multi=True)

    def pop(self, queue_name, limit, project=None):
        query = {PROJ_QUEUE: utils.scope_queue_name(queue_name, project)}

        # Only include messages that are not part of
        # any claim, or are part of an expired claim.
        now = timeutils.utcnow_ts()
        query["c.e"] = {"$lte": now}

        collection = self._collection(queue_name, project)
        fields = {"_id": 1, "t": 1, "b": 1}

        messages = (collection.find_and_modify(query, fields=fields, remove=True) for _ in range(limit))

        messages = itertools.ifilter(None, messages)

        final_messages = [_basic_message(message, now) for message in messages]

        return final_messages

    def bulk_get(self, queue_name, message_ids, project=None):
        message_ids = [mid for mid in map(utils.to_oid, message_ids) if mid]
        if not message_ids:
            return iter([])

        now = timeutils.utcnow_ts()

        # Base query, always check expire time
        query = {"_id": {"$in": message_ids}, PROJ_QUEUE: utils.scope_queue_name(queue_name, project)}

        collection = self._collection(queue_name, project)

        # NOTE(flaper87): Should this query
        # be sorted?
        messages = collection.find(query).hint(ID_INDEX_FIELDS)

        def denormalizer(msg):
            return _basic_message(msg, now)

        return utils.HookedCursor(messages, denormalizer)

    def update(self, queue, claim_id, metadata, project=None):
        cid = utils.to_oid(claim_id)
        if cid is None:
            raise errors.ClaimDoesNotExist(claim_id, queue, project)

        now = timeutils.utcnow_ts()
        ttl = int(metadata.get('ttl', 60))
        expires = now + ttl

        msg_ctrl = self.driver.message_controller
        claimed = msg_ctrl._claimed(queue, cid, expires=now,
                                    limit=1, project=project)

        try:
            next(claimed)
        except StopIteration:
            raise errors.ClaimDoesNotExist(claim_id, queue, project)

        meta = {
            'id': cid,
            't': ttl,
            'e': expires,
        }

        # TODO(kgriffs): Create methods for these so we don't interact
        # with the messages collection directly (loose coupling)
        scope = utils.scope_queue_name(queue, project)
        collection = msg_ctrl._collection(queue, project)
        collection.update({'p_q': scope, 'c.id': cid},
                          {'$set': {'c': meta}},
                          upsert=False, multi=True)

        # NOTE(flaper87): Dirty hack!
        # This sets the expiration time to
        # `expires` on messages that would
        # expire before claim.
        collection.update({'p_q': scope,
                           'e': {'$lt': expires},
                           'c.id': cid},
                          {'$set': {'e': expires, 't': ttl}},
                          upsert=False, multi=True)

    def _count(self, queue_name, project=None, include_claimed=False):
        """Return total number of messages in a queue.

        This method is designed to very quickly count the number
        of messages in a given queue. Expired messages are not
        counted, of course. If the queue does not exist, the
        count will always be 0.

        Note: Some expired messages may be included in the count if
            they haven't been GC'd yet. This is done for performance.
        """
        query = {
            # Messages must belong to this queue
            PROJ_QUEUE: utils.scope_queue_name(queue_name, project)
        }

        if not include_claimed:
            # Exclude messages that are claimed
            query["c.e"] = {"$lte": timeutils.utcnow_ts()}

        collection = self._collection(queue_name, project)
        return collection.find(query).hint(COUNTING_INDEX_FIELDS).count()

    def delete(self, queue_name, message_id, project=None, claim=None):
        # NOTE(cpp-cabrera): return early - this is an invalid message
        # id so we won't be able to find it any way
        mid = utils.to_oid(message_id)
        if mid is None:
            return

        collection = self._collection(queue_name, project)

        query = {
            '_id': mid,
            PROJ_QUEUE: utils.scope_queue_name(queue_name, project),
        }

        # NOTE(cpp-cabrera): return early - the user gaves us an
        # invalid claim id and that renders the rest of this
        # request moot
        cid = utils.to_oid(claim)
        if cid is None:
            return

        now = timeutils.utcnow_ts()
        message = collection.find_one(query)

        if message is None:
            return

        is_claimed = (message['c']['id'] is not None and
                      message['c']['e'] > now)

        if claim is None:
            if is_claimed:
                raise errors.MessageIsClaimed(message_id)

        else:
            if message['c']['id'] != cid:
                raise errors.MessageIsClaimedBy(message_id, claim)

        collection.remove(query['_id'], w=0)

    def list(self, project=None, marker=None,
             limit=None, detailed=False):

        if limit is None:
            limit = self.driver.limits_conf.default_queue_paging

        query = {}
        scoped_name = utils.scope_queue_name(marker, project)

        if not scoped_name.startswith('/'):
            # NOTE(kgriffs): scoped queue, e.g., 'project-id/queue-name'
            project_prefix = '^' + project + '/'
            query['p_q'] = {'$regex': project_prefix, '$gt': scoped_name}
        elif scoped_name == '/':
            # NOTE(kgriffs): list global queues, but exclude scoped ones
            query['p_q'] = {'$regex': '^/'}
        else:
            # NOTE(kgriffs): unscoped queue, e.g., '/my-global-queue'
            query['p_q'] = {'$regex': '^/', '$gt': scoped_name}

        fields = {'p_q': 1, '_id': 0}
        if detailed:
            fields['m'] = 1

        cursor = self._collection.find(query, fields=fields)
        cursor = cursor.limit(limit).sort('p_q')
        marker_name = {}

        def normalizer(record):
            queue = {'name': utils.descope_queue_name(record['p_q'])}
            marker_name['next'] = queue['name']
            if detailed:
                queue['metadata'] = record['m']
            return queue

        yield utils.HookedCursor(cursor, normalizer)
        yield marker_name and marker_name['next']

    def create(self, name, project=None):
        # NOTE(flaper87): If the connection fails after it was called
        # and we retry to insert the queue, we could end up returning
        # `False` because of the `DuplicatedKeyError` although the
        # queue was indeed created by this API call.
        #
        # TODO(kgriffs): Commented out `retries_on_autoreconnect` for
        # now due to the above issue, since creating a queue is less
        # important to make super HA.

        try:
            # NOTE(kgriffs): Start counting at 1, and assume the first
            # message ever posted will succeed and set t to a UNIX
            # "modified at" timestamp.
            counter = {'v': 1, 't': 0}

            scoped_name = utils.scope_queue_name(name, project)
            self._collection.insert({'p_q': scoped_name, 'm': {},
                                     'c': counter})

        except pymongo.errors.DuplicateKeyError:
            return False
        else:
            return True

 def delete(self, project, queue):
     self._col.remove({PRIMARY_KEY: utils.scope_queue_name(queue, project)},
                      w=0)

    def create(self, queue, metadata, project=None,
               limit=storage.DEFAULT_MESSAGES_PER_CLAIM):
        """Creates a claim.

        This implementation was done in a best-effort fashion.
        In order to create a claim we need to get a list
        of messages that can be claimed. Once we have that
        list we execute a query filtering by the ids returned
        by the previous query.

        Since there's a lot of space for race conditions here,
        we'll check if the number of updated records is equal to
        the max number of messages to claim. If the number of updated
        messages is lower than limit we'll try to claim the remaining
        number of messages.

        This 2 queries are required because there's no way, as for the
        time being, to execute an update on a limited number of records.
        """
        msg_ctrl = self.driver.message_controller

        ttl = metadata['ttl']
        grace = metadata['grace']
        oid = objectid.ObjectId()

        now = timeutils.utcnow_ts()
        claim_expires = now + ttl

        message_ttl = ttl + grace
        message_expiration = datetime.datetime.utcfromtimestamp(
            claim_expires + grace)

        meta = {
            'id': oid,
            't': ttl,
            'e': claim_expires,
        }

        # Get a list of active, not claimed nor expired
        # messages that could be claimed.
        msgs = msg_ctrl._active(queue, fields={'_id': 1}, project=project,
                                limit=limit)

        messages = iter([])
        ids = [msg['_id'] for msg in msgs]

        if len(ids) == 0:
            return (None, messages)

        now = timeutils.utcnow_ts()

        # NOTE(kgriffs): Set the claim field for
        # the active message batch, while also
        # filtering out any messages that happened
        # to get claimed just now by one or more
        # parallel requests.
        #
        # Filtering by just 'c.e' works because
        # new messages have that field initialized
        # to the current time when the message is
        # posted. There is no need to check whether
        # 'c' exists or 'c.id' is None.
        collection = msg_ctrl._collection(queue, project)
        updated = collection.update({'_id': {'$in': ids},
                                     'c.e': {'$lte': now}},
                                    {'$set': {'c': meta}},
                                    upsert=False,
                                    multi=True)['n']

        # NOTE(flaper87): Dirty hack!
        # This sets the expiration time to
        # `expires` on messages that would
        # expire before claim.
        new_values = {'e': message_expiration, 't': message_ttl}
        collection.update({'p_q': utils.scope_queue_name(queue, project),
                           'e': {'$lt': message_expiration},
                           'c.id': oid},
                          {'$set': new_values},
                          upsert=False, multi=True)

        if updated != 0:
            # NOTE(kgriffs): This extra step is necessary because
            # in between having gotten a list of active messages
            # and updating them, some of them may have been
            # claimed by a parallel request. Therefore, we need
            # to find out which messages were actually tagged
            # with the claim ID successfully.
            claim, messages = self.get(queue, oid, project=project)

        return (str(oid), messages)

 def exists(self, project, queue):
     key = utils.scope_queue_name(queue, project)
     return self._col.find_one({PRIMARY_KEY: key}) is not None