MySQL Connector/Python reference leak test script

mysql_connector_leak_test.py

#!/usr/bin/env python3

"""
Schema used:

SET sql_mode = 'NO_ENGINE_SUBSTITUTION,ANSI,MAXDB,TRADITIONAL,NO_AUTO_VALUE_ON_ZERO,ONLY_FULL_GROUP_BY';

CREATE DATABASE "leak_test";
USE "leak_test"

CREATE TABLE "parent" (
    "uuid" BINARY(16) PRIMARY KEY
) ENGINE=INNODB;

CREATE TABLE "child" (
    "uuid" BINARY(16) PRIMARY KEY,
    "parent_uuid" BINARY(16),
    "date" DATETIME NOT NULL,
    "num" BIGINT UNSIGNED NOT NULL,
    CONSTRAINT "fk_c_parent" FOREIGN KEY ("uuid") REFERENCES "parent"("uuid") ON UPDATE CASCADE ON DELETE CASCADE
) ENGINE=INNODB;

"""

from sys import argv, getrefcount
from functools import partial
from contextlib import closing
from datetime import datetime
import tracemalloc

from mysql.connector import connect
from mysql.connector.errors import IntegrityError


def leak_test(do_connect, loops=1):
    query = 'INSERT INTO "child" ("uuid", "parent_uuid", "date", "num") VALUES (%s, %s, %s, %s)'
    args = (b'0'*16, b'1'*16, datetime.utcnow(), 2**63 - 1)

    for _ in range(loops):
        try:
            with closing(do_connect()) as conn:
                with closing(conn.cursor(buffered=True)) as cursor:
                    cursor.execute(query, params=args)
                    cursor.fetchall()
        except IntegrityError:
            # Expecting FK error for parent_uuid as part of test
            pass


def main():
    try:
        loops = int(argv[1])
    except:
        loops = 10
    print('Will run %d loops' % loops)

    do_connect = partial(
        connect,
        host='127.0.0.1',
        port=3306,
        user='leak_test',
        password='p',
        database='leak_test',
        use_unicode=True,
        use_pure=False,
        compress=False,
        sql_mode='NO_ENGINE_SUBSTITUTION,ANSI,MAXDB,TRADITIONAL,NO_AUTO_VALUE_ON_ZERO,ONLY_FULL_GROUP_BY',
        charset='utf8'
    )

    # Exclude one-off references by running loop once beforehand
    leak_test(do_connect)

    refs = {item: getrefcount(item) for item in (True, False, None)}
    tracemalloc.start()
    snap1 = tracemalloc.take_snapshot()

    leak_test(do_connect, loops=loops)

    snap2 = tracemalloc.take_snapshot()
    tracemalloc.stop()

    print('** Reference changes (built-in types):')
    print('\n'.join('%r: %d' % (item, getrefcount(item) - count) for item, count in refs.items()))
    del refs

    print('\n** Tracemalloc top 50')
    for stat in snap2.compare_to(snap1, 'lineno')[:50]:
        print(stat)


if __name__ == '__main__':
    main()

Output including mysql/mysql-connector-python#41 and mysql/mysql-connector-python#42

Will run 100 loops
** Reference changes (built-in types):
False: 3
True: 3
None: 3

** Tracemalloc top 50
/usr/lib/python3.5/tracemalloc.py:349: size=528 B (+528 B), count=3 (+3), average=176 B
/usr/lib/python3.5/tracemalloc.py:487: size=504 B (+504 B), count=2 (+2), average=252 B
/usr/lib/python3.5/tracemalloc.py:277: size=32 B (+32 B), count=1 (+1), average=32 B
dbtest/raw.py:76: size=448 B (+0 B), count=1 (+0), average=448 B