Method test_execution_time_limit

pymysql/tests/test_SSCursor.py:115–199

(self)

        cursor.close()

    def test_execution_time_limit(self):
        # this method is similarly implemented in test_cursor

        conn = self.connect()

        # table creation and filling is SSCursor only as it's not provided by self.setUp()
        self.safe_create_table(
            conn,
            "test",
            "create table test (data varchar(10))",
        )
        with conn.cursor() as cur:
            cur.execute(
                "insert into test (data) values "
                "('row1'), ('row2'), ('row3'), ('row4'), ('row5')"
            )
            conn.commit()

        db_type = self.get_mysql_vendor(conn)

        with conn.cursor(pymysql.cursors.SSCursor) as cur:
            # MySQL MAX_EXECUTION_TIME takes ms
            # MariaDB max_statement_time takes seconds as int/float, introduced in 10.1

            # this will sleep 0.01 seconds per row
            if db_type == "mysql":
                sql = (
                    "SELECT /*+ MAX_EXECUTION_TIME(2000) */ data, sleep(0.01) FROM test"
                )
            else:
                sql = "SET STATEMENT max_statement_time=2 FOR SELECT data, sleep(0.01) FROM test"

            cur.execute(sql)
            # unlike Cursor, SSCursor returns a list of tuples here
            self.assertEqual(
                cur.fetchall(),
                [
                    ("row1", 0),
                    ("row2", 0),
                    ("row3", 0),
                    ("row4", 0),
                    ("row5", 0),
                ],
            )

            if db_type == "mysql":
                sql = (
                    "SELECT /*+ MAX_EXECUTION_TIME(2000) */ data, sleep(0.01) FROM test"
                )
            else:
                sql = "SET STATEMENT max_statement_time=2 FOR SELECT data, sleep(0.01) FROM test"
            cur.execute(sql)
            self.assertEqual(cur.fetchone(), ("row1", 0))

            # this discards the previous unfinished query and raises an
            # incomplete unbuffered query warning
            with pytest.warns(UserWarning):
                cur.execute("SELECT 1")