Package cherrypy :: Package test :: Module test_refleaks
[hide private]
[frames] | no frames]

Source Code for Module cherrypy.test.test_refleaks

  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   
16 -def get_instances(cls):
17 return [x for x in gc.get_objects() if isinstance(x, cls)]
18
19 -def setup_server():
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 # Uncollectable garbage 31 32 # gc_collect isn't perfectly synchronous, because it may 33 # break reference cycles that then take time to fully 34 # finalize. Call it twice and hope for the best. 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 # Request references 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 # Response references 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
85 -class ReferenceTests(helper.CPWebCase):
86
87 - def test_threadlocal_garbage(self):
88 success = [] 89 90 def getpage(): 91 host = '%s:%s' % (self.interface(), self.PORT) 92 if self.scheme == 'https': 93 c = httplib.HTTPSConnection(host) 94 else: 95 c = httplib.HTTPConnection(host) 96 try: 97 c.putrequest('GET', '/') 98 c.endheaders() 99 response = c.getresponse() 100 body = response.read() 101 self.assertEqual(response.status, 200) 102 self.assertEqual(body, "Hello world!") 103 finally: 104 c.close() 105 success.append(True)
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