"""Tests for refleaks."""
from cherrypy.test import test
test.prefer_parent_path()
import gc
import httplib
import threading
import cherrypy
from cherrypy import _cprequest
data = object()
def get_instances(cls):
return [x for x in gc.get_objects() if isinstance(x, cls)]
def setup_server():
class Root:
def index(self, *args, **kwargs):
cherrypy.request.thing = data
return "Hello world!"
index.exposed = True
def gc_stats(self):
output = ["Statistics:"]
# Uncollectable garbage
# gc_collect isn't perfectly synchronous, because it may
# break reference cycles that then take time to fully
# finalize. Call it twice and hope for the best.
gc.collect()
unreachable = gc.collect()
if unreachable:
output.append("\n%s unreachable objects:" % unreachable)
trash = {}
for x in gc.garbage:
trash[type(x)] = trash.get(type(x), 0) + 1
trash = [(v, k) for k, v in trash.iteritems()]
trash.sort()
for pair in trash:
output.append(" " + repr(pair))
# Request references
reqs = get_instances(_cprequest.Request)
lenreqs = len(reqs)
if lenreqs < 2:
output.append("\nMissing Request reference. Should be 1 in "
"this request thread and 1 in the main thread.")
elif lenreqs > 2:
output.append("\nToo many Request references (%r)." % lenreqs)
for req in reqs:
output.append("Referrers for %s:" % repr(req))
for ref in gc.get_referrers(req):
if ref is not reqs:
output.append(" %s" % repr(ref))
# Response references
resps = get_instances(_cprequest.Response)
lenresps = len(resps)
if lenresps < 2:
output.append("\nMissing Response reference. Should be 1 in "
"this request thread and 1 in the main thread.")
elif lenresps > 2:
output.append("\nToo many Response references (%r)." % lenresps)
for resp in resps:
output.append("Referrers for %s:" % repr(resp))
for ref in gc.get_referrers(resp):
if ref is not resps:
output.append(" %s" % repr(ref))
return "\n".join(output)
gc_stats.exposed = True
cherrypy.tree.mount(Root())
cherrypy.config.update({'environment': 'test_suite'})
from cherrypy.test import helper
class ReferenceTests(helper.CPWebCase):
def test_threadlocal_garbage(self):
success = []
def getpage():
host = '%s:%s' % (self.interface(), self.PORT)
if self.scheme == 'https':
c = httplib.HTTPSConnection(host)
else:
c = httplib.HTTPConnection(host)
try:
c.putrequest('GET', '/')
c.endheaders()
response = c.getresponse()
body = response.read()
self.assertEqual(response.status, 200)
self.assertEqual(body, "Hello world!")
finally:
c.close()
success.append(True)
ITERATIONS = 25
ts = []
for _ in range(ITERATIONS):
t = threading.Thread(target=getpage)
ts.append(t)
t.start()
for t in ts:
t.join()
self.assertEqual(len(success), ITERATIONS)
self.getPage("/gc_stats")
self.assertBody("Statistics:")
if __name__ == '__main__':
setup_server()
helper.testmain({'server.socket_queue_size': 10})
|