Method test_max_time_ms

test/asynchronous/test_cursor.py:159–201

(self)

        self.assertEqual(False, cursor._allow_disk_use)

    async def test_max_time_ms(self):
        db = self.db
        await db.pymongo_test.drop()
        coll = db.pymongo_test
        with self.assertRaises(TypeError):
            coll.find().max_time_ms("foo")  # type: ignore[arg-type]
        await coll.insert_one({"amalia": 1})
        await coll.insert_one({"amalia": 2})

        coll.find().max_time_ms(None)
        coll.find().max_time_ms(1)

        cursor = coll.find().max_time_ms(999)
        self.assertEqual(999, cursor._max_time_ms)
        cursor = coll.find().max_time_ms(10).max_time_ms(1000)
        self.assertEqual(1000, cursor._max_time_ms)

        cursor = coll.find().max_time_ms(999)
        c2 = cursor.clone()
        self.assertEqual(999, c2._max_time_ms)
        self.assertIn("$maxTimeMS", cursor._query_spec())
        self.assertIn("$maxTimeMS", c2._query_spec())

        self.assertTrue(await coll.find_one(max_time_ms=1000))

        client = self.client
        if not async_client_context.is_mongos and async_client_context.test_commands_enabled:
            # Cursor parses server timeout error in response to initial query.
            await client.admin.command(
                "configureFailPoint", "maxTimeAlwaysTimeOut", mode="alwaysOn"
            )
            try:
                cursor = coll.find().max_time_ms(1)
                try:
                    await anext(cursor)
                except ExecutionTimeout:
                    pass
                else:
                    self.fail("ExecutionTimeout not raised")
                with self.assertRaises(ExecutionTimeout):
                    await coll.find_one(max_time_ms=1)
            finally:
                await client.admin.command("configureFailPoint", "maxTimeAlwaysTimeOut", mode="off")

    async def test_maxtime_ms_message(self):
        db = self.db