Method test_command_max_time_ms

test/test_database.py:647–670

(self)

    @client_context.require_test_commands
    @client_context.require_no_mongos
    def test_command_max_time_ms(self):
        self.client.admin.command("configureFailPoint", "maxTimeAlwaysTimeOut", mode="alwaysOn")
        try:
            db = self.client.pymongo_test
            db.command("count", "test")
            with self.assertRaises(ExecutionTimeout):
                db.command("count", "test", maxTimeMS=1)
            pipeline = [{"$project": {"name": 1, "count": 1}}]
            # Database command helper.
            db.command("aggregate", "test", pipeline=pipeline, cursor={})
            with self.assertRaises(ExecutionTimeout):
                db.command(
                    "aggregate",
                    "test",
                    pipeline=pipeline,
                    cursor={},
                    maxTimeMS=1,
                )
            # Collection helper.
            db.test.aggregate(pipeline=pipeline)
            with self.assertRaises(ExecutionTimeout):
                db.test.aggregate(pipeline, maxTimeMS=1)
        finally:
            self.client.admin.command("configureFailPoint", "maxTimeAlwaysTimeOut", mode="off")

    def test_with_options(self):
        codec_options = DECIMAL_CODECOPTS