Ticket #13320: python_db_test.py

File python_db_test.py, 4.7 KB (added by rcrdnalor, 17 months ago)

Python testscript for checking timeout

Line 
1#!/usr/bin/python
2
3from MythTV import MythDB, dbmodule
4from pprint import pprint
5
6import time
7import platform
8import sys
9
10
11# See https://mariadb.com/kb/en/library/mysql_ping/
12# And help on MySQLdb ping():
13"""
14>>> import MySQLdb
15>>> help(MySQLdb._mysql.connection.ping)
16Help on method_descriptor:
17
18ping(...)
19    Checks whether or not the connection to the server is
20    working. If it has gone down, an automatic reconnection is
21    attempted.
22
23    This function can be used by clients that remain idle for a
24    long while, to check whether or not the server has closed the
25    connection and reconnect if necessary.
26
27    New in 1.2.2: Accepts an optional reconnect parameter. If True,
28    then the client will attempt reconnection. Note that this setting
29    is persistent. By default, this is on in MySQL<5.0.3, and off
30    thereafter.
31
32    Non-standard. You should assume that ping() performs an
33    implicit rollback; use only when starting a new transaction.
34    You have been warned
35"""
36
37
38print("Python Version  : %s" %platform.python_version())
39print("SQL Driver Version: %s" %repr(dbmodule.__version__))
40
41print("This test run in 2 steps:")
42print(" - tc #1 uses the old version of MysqlDB.ping: '_ping121(self): self._get_db().ping(True)'")
43print(" - tc #2 uses the new version of MysqlDB.ping: '_ping122(self): self._get_db().ping()'")
44print("Each test connects to mysql via the MythTV Python Bindings and checks connection "
45      "recovery after a timeout occured.")
46
47for tc in (1,2):
48    print("\nConnect to MythDB (tc = %d) :" %tc)
49    mydb = MythDB()
50    print("  Allow only one connection from Python Bindings to mysql: MythDB.db.resizePool(1)")
51    mydb.db.resizePool(1)
52    with mydb as cursor:
53        cursor.execute("SHOW VARIABLES LIKE 'version'")
54        print("  MYSQL version : %s" %cursor.fetchone()[1])
55
56        cursor.execute("select CONNECTION_ID()")
57        print("    Connection_ID : %d" %cursor.fetchone()[0])
58
59        print("  current SQL timeouts for this session: ")
60        # 'wait_timeout' : The number of seconds the server waits for activity
61        #  on a noninteractive connection before closing it.
62        cursor.execute("select @@session.wait_timeout")
63        print("    session.wait_timeout : %d" %cursor.fetchone()[0])
64
65        # 'interactive_timeout' : The number of seconds the server waits for activity
66        #  on an interactive connection before closing it.
67        cursor.execute("select @@session.interactive_timeout")
68        print("    session.interactive_timeout : %d" %cursor.fetchone()[0])
69
70        cursor.execute("select CONNECTION_ID()")
71        print("    Connection_ID : %d" %cursor.fetchone()[0])
72
73        print("  set session timeouts to 60 secs: ")
74        cursor.execute("select CONNECTION_ID()")
75        print("    Connection_ID : %d" %cursor.fetchone()[0])
76
77        cursor.execute("set @@session.wait_timeout = 60")
78        print("    session.wait_timeout set to 60")
79
80        cursor.execute("set @@session.interactive_timeout = 60")
81        print("    session.interactive_timeout set to 60")
82
83        print("  check again current timeouts: ")
84        cursor.execute("select @@session.wait_timeout")
85        print("    session.wait_timeout : %d" %cursor.fetchone()[0])
86
87        cursor.execute("select @@session.interactive_timeout")
88        print("    session.interactive_timeout : %d" %cursor.fetchone()[0])
89
90        cursor.execute("select CONNECTION_ID()")
91        connection_id_1 = cursor.fetchone()[0]
92        print("    Connection_ID : %d" %connection_id_1)
93
94        print("  sleep 50 secs (should not time out) ...")
95        time.sleep(50)
96        if tc == 1:
97            print("\tping db connection with _ping121()")
98            cursor._ping121()
99        else:
100            print("\tping db connection with _ping122()")
101            cursor._ping122()
102
103        print("  sleep 50 secs (should not time out) ...")
104        time.sleep(50)
105        if tc == 1:
106            print("\tping db connection with _ping121()")
107            cursor._ping121()
108        else:
109            print("\tping db connection with _ping122()")
110            cursor._ping122()
111
112        print("  sleep 80 secs (should time out) ...")
113        time.sleep(80)
114        if tc == 1:
115            print("\tping db connection with _ping121()")
116            cursor._ping121()
117        else:
118            print("\tping db connection with _ping122()")
119            cursor._ping122()
120        cursor.execute("select CONNECTION_ID()")
121        connection_id_2 = cursor.fetchone()[0]
122        print("    Connection_ID : %d" %connection_id_2)
123        print("    Is this Connection_ID ident to the original one ( %d <> %d ) ?"
124                %(connection_id_1, connection_id_2) )
125
126
127print("Tests finished, thank you for providing the data!")
128
129