NovaBryan/EVE_Dump
Log | Files | 📦 下载最新源码 (.zip) | Refs | README

test_refleaks.py (3728B)

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