/usr/share/pyshared/cherrypy/test/test_custom_filters.py is in python-cherrypy 2.3.0-3.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 | """Test the various means of instantiating and invoking filters."""
import time
import types
import test
test.prefer_parent_path()
import cherrypy
from cherrypy import filters
from cherrypy.filters.basefilter import BaseFilter
class AccessFilter(BaseFilter):
def before_request_body(self):
if not cherrypy.config.get("access_filter.on", False):
return
if not getattr(cherrypy.request, "login", None):
raise cherrypy.HTTPError(401)
def setup_server():
class Numerify(BaseFilter):
def on_start_resource(self):
m = cherrypy.config.get("numerify_filter.map", {})
cherrypy.request.numerify_map = m.items()
def before_finalize(self):
if not cherrypy.config.get("numerify_filter.on", False):
return
def number_it(body):
for chunk in body:
for k, v in cherrypy.request.numerify_map:
chunk = chunk.replace(k, v)
yield chunk
cherrypy.response.body = number_it(cherrypy.response.body)
# It's not mandatory to inherit from BaseFilter.
class NadsatFilter:
def __init__(self):
self.counter = 0
self.ended = {}
def before_main(self):
cherrypy.request.counter = self.counter = self.counter + 1
self.ended[cherrypy.request.counter] = False
def before_finalize(self):
def nadsat_it_up(body):
for chunk in body:
chunk = chunk.replace("good", "horrorshow")
chunk = chunk.replace("piece", "lomtick")
yield chunk
cherrypy.response.body = nadsat_it_up(cherrypy.response.body)
def on_end_request(self):
# This runs after the request has been completely written out.
cherrypy.response.body = "razdrez"
self.ended[cherrypy.request.counter] = True
class Root:
def index(self):
return "Howdy earth!"
index.exposed = True
cherrypy.root = Root()
class TestType(type):
"""Metaclass which automatically exposes all functions in each subclass,
and adds an instance of the subclass as an attribute of cherrypy.root.
"""
def __init__(cls, name, bases, dct):
type.__init__(name, bases, dct)
for value in dct.itervalues():
if isinstance(value, types.FunctionType):
value.exposed = True
setattr(cherrypy.root, name.lower(), cls())
class Test(object):
__metaclass__ = TestType
class CPFilterList(Test):
# METHOD ONE:
# Use _cp_filters (old name: _cpFilterList)
_cp_filters = [NadsatFilter()]
def index(self):
return "A good piece of cherry pie"
def ended(self, id):
return repr(self._cp_filters[0].ended[int(id)])
def err(self):
raise ValueError()
def stream(self):
for i in xrange(100000000):
yield str(i)
def errinstream(self):
raise ValueError()
yield "confidential"
def restricted(self):
return "Welcome!"
def err_in_onstart(self):
return "success!"
cherrypy.config.update({
'global': {
# METHOD TWO:
# Declare a classname in server.input_filters.
'server.input_filters': ["cherrypy.test.test_custom_filters.AccessFilter"],
'server.log_to_screen': False,
'server.environment': 'production',
'server.show_tracebacks': True,
},
'/cpfilterlist': {
'numerify_filter.on': True,
'numerify_filter.map': {"pie": "3.14159"}
},
'/cpfilterlist/restricted': {
'access_filter.on': True,
'server.show_tracebacks': False,
},
'/cpfilterlist/errinstream': {
'stream_response': True,
},
'/cpfilterlist/stream': {
'stream_response': True,
},
'/cpfilterlist/err_in_onstart': {
# Because this isn't a dict, on_start_resource will error.
'numerify_filter.map': "pie->3.14159"
},
})
# METHOD THREE:
# Insert a class directly into the filters.output_filters chain.
# You can also insert a string, but we're effectively testing
# using-a-string via the config file.
filters.input_filters.insert(0, Numerify)
filters.output_filters.insert(0, Numerify)
# We have to call filters.init() here (if we want methods #2 and #3
# to work), because the test suite may already have run server.start()
# (which is where filters.init() is usually called).
filters.init()
# Client-side code #
import helper
class FilterTests(helper.CPWebCase):
def testCPFilterList(self):
self.getPage("/cpfilterlist/")
# If body is "razdrez", then on_end_request is being called too early.
self.assertBody("A horrorshow lomtick of cherry 3.14159")
# If this fails, then on_end_request isn't being called at all.
self.getPage("/cpfilterlist/ended/1")
self.assertBody("True")
valerr = '\n raise ValueError()\nValueError'
self.getPage("/cpfilterlist/err")
# If body is "razdrez", then on_end_request is being called too early.
self.assertErrorPage(500, pattern=valerr)
# If this fails, then on_end_request isn't being called at all.
self.getPage("/cpfilterlist/ended/3")
self.assertBody("True")
# If body is "razdrez", then on_end_request is being called too early.
self.getPage("/cpfilterlist/errinstream")
# Because this error is raised after the response body has
# started, the status should not change to an error status.
self.assertStatus("200 OK")
self.assertBody("Unrecoverable error in the server.")
# If this fails, then on_end_request isn't being called at all.
self.getPage("/cpfilterlist/ended/5")
self.assertBody("True")
# Test that on_end_request is called even if the client drops.
self.persistent = True
try:
conn = self.HTTP_CONN
conn.putrequest('GET', '/cpfilterlist/stream', skip_host=True)
conn.putheader('Host', self.HOST)
conn.endheaders()
# Skip the rest of the request and close the conn. This will
# cause the server's active socket to error, which *should*
# result in the request being aborted, and request.close being
# called all the way up the stack (including WSGI middleware),
# eventually calling our on_end_request hook.
finally:
self.persistent = False
time.sleep(0.1)
# on_end_request should have been called
self.getPage('/cpfilterlist/ended/7')
self.assertBody("True")
# Test the config method.
self.getPage("/cpfilterlist/restricted")
self.assertErrorPage(401)
def testGuaranteedFilters(self):
# The on_start_resource and on_end_request filter methods are all
# guaranteed to run, even if there are failures in other on_start
# or on_end methods. This is NOT true of the other filter methods.
# Here, we have set up a failure in NumerifyFilter.on_start_resource,
# but because that failure is logged and passed over, the error
# page we obtain in the user agent should be from before_finalize.
self.getPage("/cpfilterlist/err_in_onstart")
self.assertErrorPage(500)
self.assertInBody("AttributeError: 'Request' object has no "
"attribute 'numerify_map'")
if __name__ == '__main__':
setup_server()
helper.testmain()
|