Source

CherryPyFixTicketUrls / cherrypy / test / helper.py

Full commit
Robert Brewer 9809b07 
Remco Boerma e0ff4c0 
Robert Brewer 9809b07 

Remco Boerma e0ff4c0 
Robert Brewer cbdc8bf 







Remco Boerma e0ff4c0 

Robert Brewer b302243 




Robert Brewer 292d7ef 

Robert Brewer 7f808ef 


Robert Brewer 292d7ef 
Robert Brewer 9fd9bfa 
Robert Brewer 22c69e1 

Robert Brewer f27cfd2 
Robert Brewer 3f1e91a 
Michael Robinson 66ad903 
Robert Brewer d4787cc 
Robert Brewer 09e3214 

Robert Brewer 1a1da0f 
Robert Brewer 222e698 
Robert Brewer 1dee819 

Robert Brewer 1a1da0f 
Robert Brewer 1dee819 
Robert Brewer b07e12b 





Robert Brewer 5a5d9c7 
Robert Brewer b234be4 
Robert Brewer b07e12b 

Robert Brewer 7f808ef 


Robert Brewer 4a5a616 


Robert Brewer 2c2f9a9 
Robert Brewer dacf95a 
Robert Brewer 1a1da0f 
Robert Brewer 996d1b3 
Lakin Wecker 4e609a6 
Robert Brewer 53b652b 

Robert Brewer 1a1da0f 
Robert Brewer 53b652b 
Robert Brewer 1a1da0f 

Sylvain Hellegou… fc57202 
Robert Brewer 1a1da0f 
Robert Brewer a06d6fc 
Robert Brewer 53b652b 


Robert Brewer 755b09d 


Robert Brewer 1a1da0f 
Sylvain Hellegou… fc57202 
Robert Brewer 755b09d 
Robert Brewer 53b652b 
Robert Brewer 86615a3 
Robert Brewer 53b652b 







Robert Brewer e8fd4b8 
Robert Brewer 53b652b 

Michael Robinson 66ad903 
Robert Brewer 42e583c 


Robert Brewer a3bda71 
Robert Brewer e0c0132 
Robert Brewer 1a1da0f 
Robert Brewer e0c0132 

Robert Brewer 1a1da0f 
Robert Brewer e0c0132 

Robert Brewer 42e583c 
Robert Brewer a3bda71 
Robert Brewer 42e583c 
Robert Brewer 348ad81 
Robert Brewer 79367e6 



Robert Brewer 1a1da0f 
Robert Brewer a3bda71 
Robert Brewer 707501c 




Robert Brewer 4ba5967 


Robert Brewer 707501c 




Robert Brewer 4ba5967 


Robert Brewer 7f808ef 
Robert Brewer cdd7334 
Robert Brewer 0e3c2fc 
















Robert Brewer cb027e9 










Robert Brewer 7fd8e8c 
Robert Brewer 7f808ef 
Robert Brewer f03a210 




























Robert Brewer 42e583c 
Robert Brewer 1a1da0f 
Robert Brewer 79367e6 
Robert Brewer 292d7ef 
Robert Brewer 8967137 

Robert Brewer 589af3c 

Robert Brewer 8967137 
Robert Brewer 292d7ef 





Robert Brewer 8967137 













Robert Brewer 589af3c 

Robert Brewer 8967137 
Robert Brewer 292d7ef 

Robert Brewer 7f808ef 

Robert Brewer 42e583c 
Robert Brewer 8336de3 
Robert Brewer 42e583c 

Robert Brewer f03a210 
Robert Brewer 42e583c 
Robert Brewer 292d7ef 


































































































"""A library of helper functions for the CherryPy test suite.

The actual script that runs the entire CP test suite is called
"test.py" (in this folder); test.py calls this module as a library.

Usage
=====
Each individual test_*.py module imports this module (helper),
usually to make an instance of CPWebCase, and then call testmain().

The CP test suite script (test.py) imports this module and calls
run_test_suite, possibly more than once. CP applications may also
import test.py (to use TestHarness), which then calls helper.py.
"""

# GREAT CARE has been taken to separate this module from test.py,
# because different consumers of each have mutually-exclusive import
# requirements. So don't go moving functions from here into test.py,
# or vice-versa, unless you *really* know what you're doing.

import os
thisdir = os.path.abspath(os.path.dirname(__file__))
import re
import sys
import thread
import time
import warnings

import cherrypy
from cherrypy.lib import http, profiler
from cherrypy.test import webtest


class CPWebCase(webtest.WebCase):
    
    script_name = ""
    scheme = "http"
    
    def prefix(self):
        return self.script_name.rstrip("/")
    
    def base(self):
        if ((self.scheme == "http" and self.PORT == 80) or
            (self.scheme == "https" and self.PORT == 443)):
            port = ""
        else:
            port = ":%s" % self.PORT
        
        return "%s://%s%s%s" % (self.scheme, self.HOST, port,
                                self.script_name.rstrip("/"))
    
    def exit(self):
        sys.exit()
    
    def tearDown(self):
        pass
    
    def getPage(self, url, headers=None, method="GET", body=None, protocol=None):
        """Open the url. Return status, headers, body."""
        if self.script_name:
            url = http.urljoin(self.script_name, url)
        return webtest.WebCase.getPage(self, url, headers, method, body, protocol)
    
    def assertErrorPage(self, status, message=None, pattern=''):
        """Compare the response body with a built in error page.
        
        The function will optionally look for the regexp pattern,
        within the exception embedded in the error page."""

        # This will never contain a traceback
        page = cherrypy._cperror.get_error_page(status, message=message)
        
        # First, test the response body without checking the traceback.
        # Stick a match-all group (.*) in to grab the traceback.
        esc = re.escape
        epage = esc(page)
        epage = epage.replace(esc('<pre id="traceback"></pre>'),
                              esc('<pre id="traceback">') + '(.*)' + esc('</pre>'))
        
        m = re.match(epage, self.body, re.DOTALL)
        if not m:
            self._handlewebError('Error page does not match\n' + page)
            return
        
        # Now test the pattern against the traceback
        if pattern is None:
            # Special-case None to mean that there should be *no* traceback.
            if m and m.group(1):
                self._handlewebError('Error page contains traceback')
        else:
            if (m is None) or (not re.search(re.escape(pattern), m.group(1))):
                msg = 'Error page does not contain %s in traceback'
                self._handlewebError(msg % repr(pattern))


CPTestLoader = webtest.ReloadingTestLoader()
CPTestRunner = webtest.TerseTestRunner(verbosity=2)

def setConfig(conf):
    """Set the global config using a copy of conf."""
    if isinstance(conf, basestring):
        # assume it's a filename
        cherrypy.config.update(conf)
    else:
        cherrypy.config.update(conf.copy())


def run_test_suite(moduleNames, server, conf):
    """Run the given test modules using the given server and [global] conf.
    
    The server is started and stopped once, regardless of the number
    of test modules. The config, however, is reset for each module.
    """
    cherrypy.config.reset()
    setConfig(conf)
    engine = cherrypy.engine
    if hasattr(engine, "signal_handler"):
        engine.signal_handler.subscribe()
    if hasattr(engine, "console_control_handler"):
        engine.console_control_handler.subscribe()
    # The Pybots automatic testing system needs the suite to exit
    # with a non-zero value if there were any problems.
    # Might as well stick it in the engine... :/
    engine.test_success = True
    engine.start_with_callback(_run_test_suite_thread,
                               args=(moduleNames, conf))
    engine.block()
    if engine.test_success:
        return 0
    else:
        return 1

def sync_apps(profile=False, validate=False, conquer=False):
    app = cherrypy.tree
    if profile:
        app = profiler.make_app(app, aggregate=False)
    if conquer:
        try:
            import wsgiconq
        except ImportError:
            warnings.warn("Error importing wsgiconq. pyconquer will not run.")
        else:
            app = wsgiconq.WSGILogger(app)
    if validate:
        try:
            from wsgiref import validate
        except ImportError:
            warnings.warn("Error importing wsgiref. The validator will not run.")
        else:
            app = validate.validator(app)
    
    h = cherrypy.server.httpserver
    if hasattr(h, 'wsgi_app'):
        # CherryPy's wsgiserver
        h.wsgi_app = app
    elif hasattr(h, 'fcgiserver'):
        # flup's WSGIServer
        h.fcgiserver.application = app
    elif hasattr(h, 'scgiserver'):
        # flup's WSGIServer
        h.scgiserver.application = app

def _run_test_suite_thread(moduleNames, conf):
    try:
        for testmod in moduleNames:
            # Must run each module in a separate suite,
            # because each module uses/overwrites cherrypy globals.
            cherrypy.tree = cherrypy._cptree.Tree()
            cherrypy.config.reset()
            setConfig(conf)
            
            m = __import__(testmod, globals(), locals())
            setup = getattr(m, "setup_server", None)
            if setup:
                setup()
            
            # The setup functions probably mounted new apps.
            # Tell our server about them.
            sync_apps(profile=conf.get("profiling.on", False),
                      validate=conf.get("validator.on", False),
                      conquer=conf.get("conquer.on", False),
                      )
            
            suite = CPTestLoader.loadTestsFromName(testmod)
            result = CPTestRunner.run(suite)
            cherrypy.engine.test_success &= result.wasSuccessful()
            
            teardown = getattr(m, "teardown_server", None)
            if teardown:
                teardown()
    finally:
        cherrypy.engine.exit()

def testmain(conf=None):
    """Run __main__ as a test module, with webtest debugging."""
    engine = cherrypy.engine
    if '--server' in sys.argv:
        # Run the test module server-side only; wait for Ctrl-C to break.
        conf = conf or {}
        conf['server.socket_host'] = '0.0.0.0'
        setConfig(conf)
        if hasattr(engine, "signal_handler"):
            engine.signal_handler.subscribe()
        if hasattr(engine, "console_control_handler"):
            engine.console_control_handler.subscribe()
        engine.start()
        engine.block()
    else:
        for arg in sys.argv:
            if arg.startswith('--client='):
                # Run the test module client-side only.
                sys.argv.remove(arg)
                conf = conf or {}
                conf['server.socket_host'] = host = arg.split('=', 1)[1].strip()
                setConfig(conf)
                webtest.WebCase.HOST = host
                webtest.WebCase.PORT = cherrypy.server.socket_port
                webtest.main()
                break
        else:
            # Run normally (both server and client in same process).
            conf = conf or {}
            conf['server.socket_host'] = '127.0.0.1'
            setConfig(conf)
            engine.start_with_callback(_test_main_thread)
            engine.block()

def _test_main_thread():
    try:
        webtest.WebCase.PORT = cherrypy.server.socket_port
        webtest.main()
    finally:
        cherrypy.engine.exit()



# --------------------------- Spawning helpers --------------------------- #


class CPProcess(object):
    
    pid_file = os.path.join(thisdir, 'test.pid')
    config_file = os.path.join(thisdir, 'test.conf')
    config_template = """[global]
server.socket_host: '%(host)s'
server.socket_port: %(port)s
log.screen: False
log.error_file: r'%(error_log)s'
log.access_file: r'%(access_log)s'
%(ssl)s
%(extra)s
"""
    error_log = os.path.join(thisdir, 'test.error.log')
    access_log = os.path.join(thisdir, 'test.access.log')
    
    def __init__(self, wait=False, daemonize=False, ssl=False):
        self.wait = wait
        self.daemonize = daemonize
        self.ssl = ssl
        self.host = cherrypy.server.socket_host
        self.port = cherrypy.server.socket_port
    
    def write_conf(self, extra=""):
        if self.ssl:
            serverpem = os.path.join(thisdir, 'test.pem')
            ssl = """
server.ssl_certificate: r'%s'
server.ssl_private_key: r'%s'
""" % (serverpem, serverpem)
        else:
            ssl = ""
        
        f = open(self.config_file, 'wb')
        f.write(self.config_template %
                {'host': self.host,
                 'port': self.port,
                 'error_log': self.error_log,
                 'access_log': self.access_log,
                 'ssl': ssl,
                 'extra': extra,
                 })
        f.close()
    
    def start(self, imports=None):
        """Start cherryd in a subprocess."""
        cherrypy._cpserver.wait_for_free_port(self.host, self.port)
        
        args = [sys.executable, os.path.join(thisdir, '..', 'cherryd'),
                '-c', self.config_file, '-p', self.pid_file]
        
        if not isinstance(imports, (list, tuple)):
            imports = [imports]
        for i in imports:
            if i:
                args.append('-i')
                args.append(i)
        
        if self.daemonize:
            args.append('-d')
        
        if self.wait:
            self.exit_code = os.spawnl(os.P_WAIT, sys.executable, *args)
        else:
            os.spawnl(os.P_NOWAIT, sys.executable, *args)
            cherrypy._cpserver.wait_for_occupied_port(self.host, self.port)
        
        # Give the engine a wee bit more time to finish STARTING
        if self.daemonize:
            time.sleep(2)
        else:
            time.sleep(1)
    
    def get_pid(self):
        return int(open(self.pid_file, 'rb').read())
    
    def join(self):
        """Wait for the process to exit."""
        try:
            try:
                # Mac, UNIX
                os.wait()
            except AttributeError:
                # Windows
                try:
                    pid = self.get_pid()
                except IOError:
                    # Assume the subprocess deleted the pidfile on shutdown.
                    pass
                else:
                    os.waitpid(pid, 0)
        except OSError, x:
            if x.args != (10, 'No child processes'):
                raise