1 """Tests for refleaks."""
2
3 from cherrypy.test import test
4 test.prefer_parent_path()
5
6 import gc
7 import httplib
8 import threading
9
10 import cherrypy
11 from cherrypy import _cprequest
12
13
14 data = object()
15
17 return [x for x in gc.get_objects() if isinstance(x, cls)]
18
20
21 class Root:
22 def index(self, *args, **kwargs):
23 cherrypy.request.thing = data
24 return "Hello world!"
25 index.exposed = True
26
27 def gc_stats(self):
28 output = ["Statistics:"]
29
30
31
32
33
34
35 gc.collect()
36 unreachable = gc.collect()
37 if unreachable:
38 output.append("\n%s unreachable objects:" % unreachable)
39 trash = {}
40 for x in gc.garbage:
41 trash[type(x)] = trash.get(type(x), 0) + 1
42 trash = [(v, k) for k, v in trash.iteritems()]
43 trash.sort()
44 for pair in trash:
45 output.append(" " + repr(pair))
46
47
48 reqs = get_instances(_cprequest.Request)
49 lenreqs = len(reqs)
50 if lenreqs < 2:
51 output.append("\nMissing Request reference. Should be 1 in "
52 "this request thread and 1 in the main thread.")
53 elif lenreqs > 2:
54 output.append("\nToo many Request references (%r)." % lenreqs)
55 for req in reqs:
56 output.append("Referrers for %s:" % repr(req))
57 for ref in gc.get_referrers(req):
58 if ref is not reqs:
59 output.append(" %s" % repr(ref))
60
61
62 resps = get_instances(_cprequest.Response)
63 lenresps = len(resps)
64 if lenresps < 2:
65 output.append("\nMissing Response reference. Should be 1 in "
66 "this request thread and 1 in the main thread.")
67 elif lenresps > 2:
68 output.append("\nToo many Response references (%r)." % lenresps)
69 for resp in resps:
70 output.append("Referrers for %s:" % repr(resp))
71 for ref in gc.get_referrers(resp):
72 if ref is not resps:
73 output.append(" %s" % repr(ref))
74
75 return "\n".join(output)
76 gc_stats.exposed = True
77
78 cherrypy.tree.mount(Root())
79 cherrypy.config.update({'environment': 'test_suite'})
80
81
82 from cherrypy.test import helper
83
84
86
106
107 ITERATIONS = 25
108 ts = []
109 for _ in range(ITERATIONS):
110 t = threading.Thread(target=getpage)
111 ts.append(t)
112 t.start()
113
114 for t in ts:
115 t.join()
116
117 self.assertEqual(len(success), ITERATIONS)
118
119 self.getPage("/gc_stats")
120 self.assertBody("Statistics:")
121
122
123 if __name__ == '__main__':
124 setup_server()
125 helper.testmain({'server.socket_queue_size': 10})
126