1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
|
#!/usr/bin/env python
"""Detect reference leaks after several unit test runs.
The script runs the unit test and counts the objects alive after the run. If
the object count differs between the last two runs, a report is printed and the
script exits with error 1.
"""
# Copyright (C) 2011-2019 Daniele Varrazzo <daniele.varrazzo@gmail.com>
#
# psycopg2 is free software: you can redistribute it and/or modify it
# under the terms of the GNU Lesser General Public License as published
# by the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# psycopg2 is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
# FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public
# License for more details.
from __future__ import print_function
import argparse
import gc
import sys
import difflib
import unittest
from pprint import pprint
from collections import defaultdict
def main():
opt = parse_args()
import tests
test = tests
if opt.suite:
test = getattr(test, opt.suite)
sys.stdout.write("test suite %s\n" % test.__name__)
for i in range(1, opt.nruns + 1):
sys.stdout.write("test suite run %d of %d\n" % (i, opt.nruns))
runner = unittest.TextTestRunner()
runner.run(test.test_suite())
dump(i, opt)
f1 = open('debug-%02d.txt' % (opt.nruns - 1)).readlines()
f2 = open('debug-%02d.txt' % opt.nruns).readlines()
for line in difflib.unified_diff(f1, f2,
"run %d" % (opt.nruns - 1), "run %d" % opt.nruns):
sys.stdout.write(line)
rv = f1 != f2 and 1 or 0
if opt.objs:
f1 = open('objs-%02d.txt' % (opt.nruns - 1)).readlines()
f2 = open('objs-%02d.txt' % opt.nruns).readlines()
for line in difflib.unified_diff(f1, f2,
"run %d" % (opt.nruns - 1), "run %d" % opt.nruns):
sys.stdout.write(line)
return rv
def parse_args():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('--nruns', type=int, metavar="N", default=3,
help="number of test suite runs [default: %(default)d]")
parser.add_argument('--suite', metavar="NAME",
help="the test suite to run (e.g. 'test_cursor'). [default: all]")
parser.add_argument('--objs', metavar="TYPE",
help="in case of leaks, print a report of object TYPE "
"(support still incomplete)")
return parser.parse_args()
def dump(i, opt):
gc.collect()
objs = gc.get_objects()
c = defaultdict(int)
for o in objs:
c[type(o)] += 1
pprint(
sorted(((v, str(k)) for k, v in c.items()), reverse=True),
stream=open("debug-%02d.txt" % i, "w"))
if opt.objs:
co = []
t = getattr(__builtins__, opt.objs)
for o in objs:
if type(o) is t:
co.append(o)
# TODO: very incomplete
if t is dict:
co.sort(key=lambda d: d.items())
else:
co.sort()
pprint(co, stream=open("objs-%02d.txt" % i, "w"))
if __name__ == '__main__':
sys.exit(main())
|
