summaryrefslogtreecommitdiff
path: root/cherrypy/test/helper.py
diff options
context:
space:
mode:
authorJason R. Coombs <jaraco@jaraco.com>2012-03-22 11:55:00 -0700
committerJason R. Coombs <jaraco@jaraco.com>2012-03-22 11:55:00 -0700
commita448af2f5ae9ccb5887a5b60b0c30aae22875a61 (patch)
treea83d1ae23d0c6a69aef9c566ffb428f797da24e3 /cherrypy/test/helper.py
parenta0342a182fd0b5d9d39913c7db4b4a5b58acdcc2 (diff)
downloadcherrypy-a448af2f5ae9ccb5887a5b60b0c30aae22875a61.tar.gz
Re-saved with excess whitespace removed. Removed unused import.
Diffstat (limited to 'cherrypy/test/helper.py')
-rw-r--r--cherrypy/test/helper.py70
1 files changed, 35 insertions, 35 deletions
diff --git a/cherrypy/test/helper.py b/cherrypy/test/helper.py
index e3006a52..ac19c210 100644
--- a/cherrypy/test/helper.py
+++ b/cherrypy/test/helper.py
@@ -110,9 +110,9 @@ class LocalSupervisor(Supervisor):
td = getattr(self, 'teardown', None)
if td:
td()
-
+
cherrypy.engine.exit()
-
+
for name, server in copyitems(getattr(cherrypy, 'servers', {})):
server.unsubscribe()
del cherrypy.servers[name]
@@ -147,7 +147,7 @@ class LocalWSGISupervisor(LocalSupervisor):
"""Obtain a new (decorated) WSGI app to hook into the origin server."""
if app is None:
app = cherrypy.tree
-
+
if self.conquer:
try:
import wsgiconq
@@ -155,7 +155,7 @@ class LocalWSGISupervisor(LocalSupervisor):
warnings.warn("Error importing wsgiconq. pyconquer will not run.")
else:
app = wsgiconq.WSGILogger(app, c_calls=True)
-
+
if self.validate:
try:
from wsgiref import validate
@@ -164,7 +164,7 @@ class LocalWSGISupervisor(LocalSupervisor):
else:
#wraps the app in the validator
app = validate.validator(app)
-
+
return app
@@ -199,7 +199,7 @@ def get_wsgi_u_supervisor(**options):
class CPWebCase(webtest.WebCase):
-
+
script_name = ""
scheme = "http"
@@ -213,7 +213,7 @@ class CPWebCase(webtest.WebCase):
'modfastcgi': get_modfastcgi_supervisor,
}
default_server = "wsgi"
-
+
def _setup_server(cls, supervisor, conf):
v = sys.version.split()[0]
log.info("Python version used to run this test script: %s" % v)
@@ -257,7 +257,7 @@ class CPWebCase(webtest.WebCase):
webtest.WebCase.HTTP_CONN = HTTPSConnection
return baseconf
_setup_server = classmethod(_setup_server)
-
+
def setup_class(cls):
''
#Creates a server
@@ -293,9 +293,9 @@ class CPWebCase(webtest.WebCase):
if hasattr(cls, 'setup_server'):
cls.supervisor.stop()
teardown_class = classmethod(teardown_class)
-
+
do_gc_test = False
-
+
def test_gc(self):
if self.do_gc_test:
self.getPage("/gc/stats")
@@ -303,41 +303,41 @@ class CPWebCase(webtest.WebCase):
# Tell nose to run this last in each class.
# Prefer sys.maxint for Python 2.3, which didn't have float('inf')
test_gc.compat_co_firstlineno = getattr(sys, 'maxint', None) or float('inf')
-
+
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 getPage(self, url, headers=None, method="GET", body=None, protocol=None):
"""Open the url. Return status, headers, body."""
if self.script_name:
url = httputil.urljoin(self.script_name, url)
return webtest.WebCase.getPage(self, url, headers, method, body, protocol)
-
+
def skip(self, msg='skipped '):
raise nose.SkipTest(msg)
-
+
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
@@ -348,7 +348,7 @@ class CPWebCase(webtest.WebCase):
if not m:
self._handlewebError('Error page does not match; expected:\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.
@@ -360,14 +360,14 @@ class CPWebCase(webtest.WebCase):
m.group(1))):
msg = 'Error page does not contain %s in traceback'
self._handlewebError(msg % repr(pattern))
-
+
date_tolerance = 2
-
+
def assertEqualDates(self, dt1, dt2, seconds=None):
"""Assert abs(dt1 - dt2) is within Y seconds."""
if seconds is None:
seconds = self.date_tolerance
-
+
if dt1 > dt2:
diff = dt1 - dt2
else:
@@ -388,7 +388,7 @@ def setup_client():
class CPProcess(object):
-
+
pid_file = os.path.join(thisdir, 'test.pid')
config_file = os.path.join(thisdir, 'test.conf')
config_template = """[global]
@@ -403,14 +403,14 @@ log.access_file: r'%(access_log)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, socket_host=None, socket_port=None):
self.wait = wait
self.daemonize = daemonize
self.ssl = ssl
self.host = socket_host or cherrypy.server.socket_host
self.port = socket_port or cherrypy.server.socket_port
-
+
def write_conf(self, extra=""):
if self.ssl:
serverpem = os.path.join(thisdir, 'test.pem')
@@ -420,7 +420,7 @@ server.ssl_private_key: r'%s'
""" % (serverpem, serverpem)
else:
ssl = ""
-
+
conf = self.config_template % {
'host': self.host,
'port': self.port,
@@ -432,21 +432,21 @@ server.ssl_private_key: r'%s'
f = open(self.config_file, 'wb')
f.write(ntob(conf, 'utf-8'))
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')
@@ -462,16 +462,16 @@ server.ssl_private_key: r'%s'
else:
os.spawnve(os.P_NOWAIT, sys.executable, args, env)
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: