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()