lib/cherrypy/test/test_refleaks.py

   1 """Tests for refleaks."""
   2
   3 import gc
   4 from cherrypy._cpcompat import HTTPConnection, HTTPSConnection, ntob
   5 import threading
   6
   7 import cherrypy
   8 from cherrypy import _cprequest
   9
  10
  11 data = object()
  12
  13 def get_instances(cls):
  14     return [x for x in gc.get_objects() if isinstance(x, cls)]
  15
  16
  17 from cherrypy.test import helper
  18
  19
  20 class ReferenceTests(helper.CPWebCase):
  21
  22     def setup_server():
  23
  24         class Root:
  25             def index(self, *args, **kwargs):
  26                 cherrypy.request.thing = data
  27                 return "Hello world!"
  28             index.exposed = True
  29
  30             def gc_stats(self):
  31                 output = ["Statistics:"]
  32
  33                 # Uncollectable garbage
  34
  35                 # gc_collect isn't perfectly synchronous, because it may
  36                 # break reference cycles that then take time to fully
  37                 # finalize. Call it twice and hope for the best.
  38                 gc.collect()
  39                 unreachable = gc.collect()
  40                 if unreachable:
  41                     output.append("\n%s unreachable objects:" % unreachable)
  42                     trash = {}
  43                     for x in gc.garbage:
  44                         trash[type(x)] = trash.get(type(x), 0) + 1
  45                     trash = [(v, k) for k, v in trash.items()]
  46                     trash.sort()
  47                     for pair in trash:
  48                         output.append("    " + repr(pair))
  49
  50                 # Request references
  51                 reqs = get_instances(_cprequest.Request)
  52                 lenreqs = len(reqs)
  53                 if lenreqs < 2:
  54                     output.append("\nMissing Request reference. Should be 1 in "
  55                                   "this request thread and 1 in the main thread.")
  56                 elif lenreqs > 2:
  57                     output.append("\nToo many Request references (%r)." % lenreqs)
  58                     for req in reqs:
  59                         output.append("Referrers for %s:" % repr(req))
  60                         for ref in gc.get_referrers(req):
  61                             if ref is not reqs:
  62                                 output.append("    %s" % repr(ref))
  63
  64                 # Response references
  65                 resps = get_instances(_cprequest.Response)
  66                 lenresps = len(resps)
  67                 if lenresps < 2:
  68                     output.append("\nMissing Response reference. Should be 1 in "
  69                                   "this request thread and 1 in the main thread.")
  70                 elif lenresps > 2:
  71                     output.append("\nToo many Response references (%r)." % lenresps)
  72                     for resp in resps:
  73                         output.append("Referrers for %s:" % repr(resp))
  74                         for ref in gc.get_referrers(resp):
  75                             if ref is not resps:
  76                                 output.append("    %s" % repr(ref))
  77
  78                 return "\n".join(output)
  79             gc_stats.exposed = True
  80
  81         cherrypy.tree.mount(Root())
  82     setup_server = staticmethod(setup_server)
  83
  84
  85     def test_threadlocal_garbage(self):
  86         success = []
  87
  88         def getpage():
  89             host = '%s:%s' % (self.interface(), self.PORT)
  90             if self.scheme == 'https':
  91                 c = HTTPSConnection(host)
  92             else:
  93                 c = HTTPConnection(host)
  94             try:
  95                 c.putrequest('GET', '/')
  96                 c.endheaders()
  97                 response = c.getresponse()
  98                 body = response.read()
  99                 self.assertEqual(response.status, 200)
 100                 self.assertEqual(body, ntob("Hello world!"))
 101             finally:
 102                 c.close()
 103             success.append(True)
 104
 105         ITERATIONS = 25
 106         ts = []
 107         for _ in range(ITERATIONS):
 108             t = threading.Thread(target=getpage)
 109             ts.append(t)
 110             t.start()
 111
 112         for t in ts:
 113             t.join()
 114
 115         self.assertEqual(len(success), ITERATIONS)
 116
 117         self.getPage("/gc_stats")
 118         self.assertBody("Statistics:")
 119