import sqlite3 import gc def get_vsize(): with open("/proc/self/stat") as f: return f.read().split()[22] class TEST: def log(self, msg=""): pass def test_create_function(): con = sqlite3.connect(":memory:", isolation_level=None) cur = con.cursor() t = TEST() con.create_function("log", 1, t.log) con.create_function("log", 1, t.log) cur.execute("SELECT log(1)") con.close() def test_set_trace_callback(): con = sqlite3.connect(":memory:", isolation_level=None) cur = con.cursor() t = TEST() con.set_trace_callback(t.log) con.set_trace_callback(t.log) cur.execute("SELECT 1") con.close() def test_set_trace_callback_with_func(): con = sqlite3.connect(":memory:", isolation_level=None) cur = con.cursor() con.set_trace_callback(t.log) con.set_trace_callback(t.log) cur.execute("SELECT 1") con.close() def test_set_progress_handler(): con = sqlite3.connect(":memory:", isolation_level=None) cur = con.cursor() t = TEST() con.set_progress_handler(t.log, 1) con.set_progress_handler(t.log, 1) cur.execute("SELECT 1") con.close() def test_set_authorizer(): con = sqlite3.connect(":memory:", isolation_level=None) cur = con.cursor() t = TEST() con.set_authorizer(t.log) con.set_authorizer(t.log) try: cur.execute("SELECT 1") except sqlite3.DatabaseError: pass con.close() def test_create_aggregate(): con = sqlite3.connect(":memory:", isolation_level=None) cur = con.cursor() t = TEST() con.create_aggregate("mysum", 1, t.log) con.create_aggregate("mysum", 1, t.log) try: cur.execute("SELECT mysum(1)") except sqlite3.OperationalError: pass con.close() def test_leak(): con = sqlite3.connect(":memory:", isolation_level=None) prev = get_vsize() i = 0 while True: i += 1 vsize = get_vsize() if prev != vsize: print(vsize) prev = vsize m = type("obj%s" % i, (), {"log": lambda self: ""}) con.set_trace_callback(m.log) gc.collect() con.close() test_create_function() test_set_authorizer() test_create_aggregate() test_set_progress_handler() test_leak() test_set_trace_callback() test_set_progress_handler()