CherryPy / cherrypy / test /

"""Tests for refleaks."""

from cherrypy.test import test

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!" = 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.
            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()]
                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) = True
    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)
                c = httplib.HTTPConnection(host)
                c.putrequest('GET', '/')
                response = c.getresponse()
                body =
                self.assertEqual(response.status, 200)
                self.assertEqual(body, "Hello world!")
        ITERATIONS = 25
        ts = []
        for _ in range(ITERATIONS):
            t = threading.Thread(target=getpage)
        for t in ts:
        self.assertEqual(len(success), ITERATIONS)

if __name__ == '__main__':
    helper.testmain({'server.socket_queue_size': 10})
Tip: Filter by directory path e.g. /media app.js to search for public/media/app.js.
Tip: Use camelCasing e.g. ProjME to search for
Tip: Filter by extension type e.g. /repo .js to search for all .js files in the /repo directory.
Tip: Separate your search with spaces e.g. /ssh pom.xml to search for src/ssh/pom.xml.
Tip: Use ↑ and ↓ arrow keys to navigate and return to view the file.
Tip: You can also navigate files with Ctrl+j (next) and Ctrl+k (previous) and view the file with Ctrl+o.
Tip: You can also navigate files with Alt+j (next) and Alt+k (previous) and view the file with Alt+o.