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 threading 
  8  import cherrypy 
  9  from cherrypy import _cprequest 
 10   
 11   
 12  data = object() 
 13   
 14 -def get_instances(cls): 
 15      return [x for x in gc.get_objects() if isinstance(x, cls)] 
 16   
 17 -def setup_server(): 
 18       
 19      class Root: 
 20          def index(self, *args, **kwargs): 
 21              cherrypy.request.thing = data 
 22              return "Hello world!" 
 23          index.exposed = True 
 24           
 25          def gc_stats(self): 
 26              output = [] 
 27               
 28              # Uncollectable garbage 
 29               
 30              # gc_collect isn't perfectly synchronous, because it may 
 31              # break reference cycles that then take time to fully 
 32              # finalize. Call it twice and hope for the best. 
 33              gc.collect() 
 34              unreachable = gc.collect() 
 35              if unreachable: 
 36                  output.append("\n%s unreachable objects:" % unreachable) 
 37                  trash = {} 
 38                  for x in gc.garbage: 
 39                      trash[type(x)] = trash.get(type(x), 0) + 1 
 40                  trash = [(v, k) for k, v in trash.iteritems()] 
 41                  trash.sort() 
 42                  for pair in trash: 
 43                      output.append("    " + repr(pair)) 
 44               
 45              # Request references 
 46              reqs = get_instances(_cprequest.Request) 
 47              lenreqs = len(reqs) 
 48              if lenreqs < 2: 
 49                  output.append("\nMissing Request reference. Should be 1 in " 
 50                                "this request thread and 1 in the main thread.") 
 51              elif lenreqs > 2: 
 52                  output.append("\nToo many Request references (%r)." % lenreqs) 
 53                  for req in reqs: 
 54                      output.append("Referrers for %s:" % repr(req)) 
 55                      for ref in gc.get_referrers(req): 
 56                          if ref is not reqs: 
 57                              output.append("    %s" % repr(ref)) 
 58               
 59              # Response references 
 60              resps = get_instances(_cprequest.Response) 
 61              lenresps = len(resps) 
 62              if lenresps < 2: 
 63                  output.append("\nMissing Response reference. Should be 1 in " 
 64                                "this request thread and 1 in the main thread.") 
 65              elif lenresps > 2: 
 66                  output.append("\nToo many Response references (%r)." % lenresps) 
 67                  for resp in resps: 
 68                      output.append("Referrers for %s:" % repr(resp)) 
 69                      for ref in gc.get_referrers(resp): 
 70                          if ref is not resps: 
 71                              output.append("    %s" % repr(ref)) 
 72               
 73              return "\n".join(output) 
 74          gc_stats.exposed = True 
 75       
 76      cherrypy.tree.mount(Root()) 
 77      cherrypy.config.update({'environment': 'test_suite'}) 
 78   
 79   
 80  from cherrypy.test import helper 
 81   
 82   
 83 -class ReferenceTests(helper.CPWebCase): 
 84       
 85 -    def test_threadlocal_garbage(self): 
 86          def getpage(): 
 87              self.getPage('/') 
 88              self.assertBody("Hello world!") 
 89           
 90          ts = [] 
 91          for _ in range(25): 
 92              t = threading.Thread(target=getpage) 
 93              ts.append(t) 
 94              t.start() 
 95           
 96          for t in ts: 
 97              t.join() 
 98           
 99          self.getPage("/gc_stats") 
100          self.assertBody("") 
101   
102   
103  if __name__ == '__main__': 
104      setup_server() 
105      helper.testmain() 
106