Ticket #13320: python_db_test.py

#!/usr/bin/python

from MythTV import MythDB, dbmodule
from pprint import pprint

import time
import platform
import sys


# See https://mariadb.com/kb/en/library/mysql_ping/
# And help on MySQLdb ping():
"""
>>> import MySQLdb
>>> help(MySQLdb._mysql.connection.ping)
Help on method_descriptor:

ping(...)
    Checks whether or not the connection to the server is
    working. If it has gone down, an automatic reconnection is
    attempted.

    This function can be used by clients that remain idle for a
    long while, to check whether or not the server has closed the
    connection and reconnect if necessary.

    New in 1.2.2: Accepts an optional reconnect parameter. If True,
    then the client will attempt reconnection. Note that this setting
    is persistent. By default, this is on in MySQL<5.0.3, and off
    thereafter.

    Non-standard. You should assume that ping() performs an
    implicit rollback; use only when starting a new transaction.
    You have been warned
"""


print("Python Version  : %s" %platform.python_version())
print("SQL Driver Version: %s" %repr(dbmodule.__version__))

print("This test run in 2 steps:")
print(" - tc #1 uses the old version of MysqlDB.ping: '_ping121(self): self._get_db().ping(True)'")
print(" - tc #2 uses the new version of MysqlDB.ping: '_ping122(self): self._get_db().ping()'") 
print("Each test connects to mysql via the MythTV Python Bindings and checks connection "
      "recovery after a timeout occured.")

for tc in (1,2):
    print("\nConnect to MythDB (tc = %d) :" %tc)
    mydb = MythDB()
    print("  Allow only one connection from Python Bindings to mysql: MythDB.db.resizePool(1)")
    mydb.db.resizePool(1)
    with mydb as cursor:
        cursor.execute("SHOW VARIABLES LIKE 'version'")
        print("  MYSQL version : %s" %cursor.fetchone()[1])

        cursor.execute("select CONNECTION_ID()")
        print("    Connection_ID : %d" %cursor.fetchone()[0])

        print("  current SQL timeouts for this session: ")
        # 'wait_timeout' : The number of seconds the server waits for activity
        #  on a noninteractive connection before closing it. 
        cursor.execute("select @@session.wait_timeout")
        print("    session.wait_timeout : %d" %cursor.fetchone()[0])

        # 'interactive_timeout' : The number of seconds the server waits for activity
        #  on an interactive connection before closing it. 
        cursor.execute("select @@session.interactive_timeout")
        print("    session.interactive_timeout : %d" %cursor.fetchone()[0])

        cursor.execute("select CONNECTION_ID()")
        print("    Connection_ID : %d" %cursor.fetchone()[0])

        print("  set session timeouts to 60 secs: ")
        cursor.execute("select CONNECTION_ID()")
        print("    Connection_ID : %d" %cursor.fetchone()[0])

        cursor.execute("set @@session.wait_timeout = 60")
        print("    session.wait_timeout set to 60")

        cursor.execute("set @@session.interactive_timeout = 60")
        print("    session.interactive_timeout set to 60")

        print("  check again current timeouts: ")
        cursor.execute("select @@session.wait_timeout")
        print("    session.wait_timeout : %d" %cursor.fetchone()[0])

        cursor.execute("select @@session.interactive_timeout")
        print("    session.interactive_timeout : %d" %cursor.fetchone()[0])

        cursor.execute("select CONNECTION_ID()")
        connection_id_1 = cursor.fetchone()[0]
        print("    Connection_ID : %d" %connection_id_1)

        print("  sleep 50 secs (should not time out) ...")
        time.sleep(50)
        if tc == 1:
            print("\tping db connection with _ping121()")
            cursor._ping121()
        else:
            print("\tping db connection with _ping122()")
            cursor._ping122()

        print("  sleep 50 secs (should not time out) ...")
        time.sleep(50)
        if tc == 1:
            print("\tping db connection with _ping121()")
            cursor._ping121()
        else:
            print("\tping db connection with _ping122()")
            cursor._ping122()

        print("  sleep 80 secs (should time out) ...")
        time.sleep(80)
        if tc == 1:
            print("\tping db connection with _ping121()")
            cursor._ping121()
        else:
            print("\tping db connection with _ping122()")
            cursor._ping122()
        cursor.execute("select CONNECTION_ID()")
        connection_id_2 = cursor.fetchone()[0]
        print("    Connection_ID : %d" %connection_id_2)
        print("    Is this Connection_ID ident to the original one ( %d <> %d ) ?"
                %(connection_id_1, connection_id_2) )


print("Tests finished, thank you for providing the data!")