test_cursor.py 文件源码

def test_comment(self, mongo, test_db, mongo_version):
        connection = await mongo.get_connection()
        if connection.is_mongos:
            pytest.skip('Not supported via mongos')
            return

        # MongoDB 3.1.5 changed the ns for commands.
        regex = {'$regex': '{}.(\$cmd|test)'.format(test_db.name)}

        if mongo_version.at_least(3, 1, 8, -1):
            query_key = 'query.comment'
        else:
            query_key = 'query.$comment'

        await test_db.set_profiling_level(ALL)
        try:
            list(await test_db.test.find().comment('foo').to_list())
            op = test_db.system.profile.find({
                'ns': '{}.test'.format(test_db.name),
                'op': 'query',
                query_key: 'foo'
            })
            assert await op.count() == 1

            await test_db.test.find().comment('foo').count()
            op = test_db.system.profile.find({
                'ns': regex,
                'op': 'command',
                'command.count': 'test',
                'command.$comment': 'foo'
            })
            assert await op.count() == 1

            await test_db.test.find().comment('foo').distinct('type')
            op = test_db.system.profile.find({
                'ns': regex,
                'op': 'command',
                'command.distinct': 'test',
                'command.$comment': 'foo'
            })
            assert await op.count() == 1
        finally:
            await test_db.set_profiling_level(OFF)
            await test_db.system.profile.drop()

        await test_db.test.insert_many([{}, {}])
        cursor = test_db.test.find()
        await self._next(cursor)
        with pytest.raises(InvalidOperation):
            cursor.comment('hello')