~ubuntu-branches/debian/jessie/cherrypy3/jessie

« back to all changes in this revision

Viewing changes to cherrypy/test/test_tools.py

  • Committer: Package Import Robot
  • Author(s): Gustavo Noronha Silva, JCF Ploemen, Stéphane Graber, Gustavo Noronha
  • Date: 2012-01-06 10:13:27 UTC
  • mfrom: (1.1.4) (7.1.2 sid)
  • Revision ID: package-import@ubuntu.com-20120106101327-smxnhguqs14ubl7e
Tags: 3.2.2-1
[ JCF Ploemen ]
* New upstream release (Closes: #571196).
* Bumped Standards-Version to 3.8.4 (no changes needed).
* Removing patch 02: no longer needed, incorporated upstream.
* Updating patch 00 to match release.
* Install cherryd man page via debian/manpages.
* debian/copyright:
  + Added notice for cherrypy/lib/httpauth.py.
  + Fixed years.
* debian/watch:
  + Don't hit on the -py3 release by blocking '-' from the version.
  + Mangle upstream version, inserting a tilde for beta/rc.

[ Stéphane Graber <stgraber@ubuntu.com> ]
 * Convert from python-support to dh_python2 (#654375)
  - debian/pyversions: Removed (no longer needed)
  - debian/rules
   + Replace call to dh_pysupport by dh_python2
   + Add --with=python2 to all dh calls
  - debian/control
   + Drop build-depends on python-support
   + Bump build-depends on python-all to >= 2.6.6-3~
   + Replace XS-Python-Version by X-Python-Version
   + Remove XB-Python-Version from binary package

[ Gustavo Noronha ]
* debian/control, debian/rules, debian/manpages:
 - use help2man to generate a manpage for cherryd at build time, since
  one is no longer shipped along with the source code
* debian/control:
- add python-nose to Build-Depends, since it's used during the
  documentation build for cross-reference generation

Show diffs side-by-side

added added

removed removed

Lines of Context:
1
1
"""Test the various means of instantiating and invoking tools."""
2
2
 
3
3
import gzip
4
 
import StringIO
5
4
import sys
6
 
from httplib import IncompleteRead
 
5
from cherrypy._cpcompat import BytesIO, copyitems, itervalues
 
6
from cherrypy._cpcompat import IncompleteRead, ntob, ntou, py3k, xrange
7
7
import time
8
8
timeout = 0.2
9
 
 
10
9
import types
11
 
from cherrypy.test import test
12
 
test.prefer_parent_path()
13
10
 
14
11
import cherrypy
15
12
from cherrypy import tools
16
13
 
17
14
 
18
 
europoundUnicode = u'\x80\xa3'
19
 
 
20
 
def setup_server():
21
 
    
22
 
    # Put check_access in a custom toolbox with its own namespace
23
 
    myauthtools = cherrypy._cptools.Toolbox("myauth")
24
 
    
25
 
    def check_access(default=False):
26
 
        if not getattr(cherrypy.request, "userid", default):
27
 
            raise cherrypy.HTTPError(401)
28
 
    myauthtools.check_access = cherrypy.Tool('before_request_body', check_access)
29
 
    
30
 
    def numerify():
31
 
        def number_it(body):
32
 
            for chunk in body:
33
 
                for k, v in cherrypy.request.numerify_map:
34
 
                    chunk = chunk.replace(k, v)
35
 
                yield chunk
36
 
        cherrypy.response.body = number_it(cherrypy.response.body)
37
 
    
38
 
    class NumTool(cherrypy.Tool):
39
 
        def _setup(self):
40
 
            def makemap():
41
 
                m = self._merged_args().get("map", {})
42
 
                cherrypy.request.numerify_map = m.items()
43
 
            cherrypy.request.hooks.attach('on_start_resource', makemap)
44
 
            
45
 
            def critical():
46
 
                cherrypy.request.error_response = cherrypy.HTTPError(502).set_response
47
 
            critical.failsafe = True
48
 
            
49
 
            cherrypy.request.hooks.attach('on_start_resource', critical)
50
 
            cherrypy.request.hooks.attach(self._point, self.callable)
51
 
    
52
 
    tools.numerify = NumTool('before_finalize', numerify)
53
 
    
54
 
    # It's not mandatory to inherit from cherrypy.Tool.
55
 
    class NadsatTool:
56
 
        
57
 
        def __init__(self):
58
 
            self.ended = {}
59
 
            self._name = "nadsat"
60
 
        
61
 
        def nadsat(self):
62
 
            def nadsat_it_up(body):
63
 
                for chunk in body:
64
 
                    chunk = chunk.replace("good", "horrorshow")
65
 
                    chunk = chunk.replace("piece", "lomtick")
66
 
                    yield chunk
67
 
            cherrypy.response.body = nadsat_it_up(cherrypy.response.body)
68
 
        nadsat.priority = 0
69
 
        
70
 
        def cleanup(self):
71
 
            # This runs after the request has been completely written out.
72
 
            cherrypy.response.body = "razdrez"
73
 
            id = cherrypy.request.params.get("id")
74
 
            if id:
75
 
                self.ended[id] = True
76
 
        cleanup.failsafe = True
77
 
        
78
 
        def _setup(self):
79
 
            cherrypy.request.hooks.attach('before_finalize', self.nadsat)
80
 
            cherrypy.request.hooks.attach('on_end_request', self.cleanup)
81
 
    tools.nadsat = NadsatTool()
82
 
    
83
 
    def pipe_body():
84
 
        cherrypy.request.process_request_body = False
85
 
        clen = int(cherrypy.request.headers['Content-Length'])
86
 
        cherrypy.request.body = cherrypy.request.rfile.read(clen)
87
 
    
88
 
    # Assert that we can use a callable object instead of a function.
89
 
    class Rotator(object):
90
 
        def __call__(self, scale):
91
 
            r = cherrypy.response
92
 
            r.collapse_body()
93
 
            r.body = [chr(ord(x) + scale) for x in r.body]
94
 
    cherrypy.tools.rotator = cherrypy.Tool('before_finalize', Rotator())
95
 
    
96
 
    def stream_handler(next_handler, *args, **kwargs):
97
 
        cherrypy.response.output = o = StringIO.StringIO()
98
 
        try:
99
 
            response = next_handler(*args, **kwargs)
100
 
            # Ignore the response and return our accumulated output instead.
101
 
            return o.getvalue()
102
 
        finally:
103
 
            o.close()
104
 
    cherrypy.tools.streamer = cherrypy._cptools.HandlerWrapperTool(stream_handler)
105
 
    
106
 
    class Root:
107
 
        def index(self):
108
 
            return "Howdy earth!"
109
 
        index.exposed = True
110
 
        
111
 
        def tarfile(self):
112
 
            cherrypy.response.output.write('I am ')
113
 
            cherrypy.response.output.write('a tarfile')
114
 
        tarfile.exposed = True
115
 
        tarfile._cp_config = {'tools.streamer.on': True}
116
 
        
117
 
        def euro(self):
118
 
            hooks = list(cherrypy.request.hooks['before_finalize'])
119
 
            hooks.sort()
120
 
            assert [x.callback.__name__ for x in hooks] == ['encode', 'gzip']
121
 
            assert [x.priority for x in hooks] == [70, 80]
122
 
            yield u"Hello,"
123
 
            yield u"world"
124
 
            yield europoundUnicode
125
 
        euro.exposed = True
126
 
        
127
 
        # Bare hooks
128
 
        def pipe(self):
129
 
            return cherrypy.request.body
130
 
        pipe.exposed = True
131
 
        pipe._cp_config = {'hooks.before_request_body': pipe_body}
132
 
        
133
 
        # Multiple decorators; include kwargs just for fun.
134
 
        # Note that encode must run before gzip.
135
 
        def decorated_euro(self, *vpath):
136
 
            yield u"Hello,"
137
 
            yield u"world"
138
 
            yield europoundUnicode
139
 
        decorated_euro.exposed = True
140
 
        decorated_euro = tools.gzip(compress_level=6)(decorated_euro)
141
 
        decorated_euro = tools.encode(errors='ignore')(decorated_euro)
142
 
    
143
 
    root = Root()
144
 
    
145
 
    
146
 
    class TestType(type):
147
 
        """Metaclass which automatically exposes all functions in each subclass,
148
 
        and adds an instance of the subclass as an attribute of root.
149
 
        """
150
 
        def __init__(cls, name, bases, dct):
151
 
            type.__init__(cls, name, bases, dct)
152
 
            for value in dct.itervalues():
153
 
                if isinstance(value, types.FunctionType):
154
 
                    value.exposed = True
155
 
            setattr(root, name.lower(), cls())
156
 
    class Test(object):
157
 
        __metaclass__ = TestType
158
 
    
159
 
    
160
 
    # METHOD ONE:
161
 
    # Declare Tools in _cp_config
162
 
    class Demo(Test):
163
 
        
164
 
        _cp_config = {"tools.nadsat.on": True}
165
 
        
166
 
        def index(self, id=None):
167
 
            return "A good piece of cherry pie"
168
 
        
169
 
        def ended(self, id):
170
 
            return repr(tools.nadsat.ended[id])
171
 
        
172
 
        def err(self, id=None):
173
 
            raise ValueError()
174
 
        
175
 
        def errinstream(self, id=None):
176
 
            yield "nonconfidential"
177
 
            raise ValueError()
178
 
            yield "confidential"
179
 
        
180
 
        # METHOD TWO: decorator using Tool()
181
 
        # We support Python 2.3, but the @-deco syntax would look like this:
182
 
        # @tools.check_access()
183
 
        def restricted(self):
184
 
            return "Welcome!"
185
 
        restricted = myauthtools.check_access()(restricted)
186
 
        userid = restricted
187
 
        
188
 
        def err_in_onstart(self):
189
 
            return "success!"
190
 
        
191
 
        def stream(self, id=None):
192
 
            for x in xrange(100000000):
193
 
                yield str(x)
194
 
        stream._cp_config = {'response.stream': True}
195
 
    
196
 
    
197
 
    cherrypy.config.update({'environment': 'test_suite'})
198
 
    
199
 
    conf = {
200
 
        # METHOD THREE:
201
 
        # Declare Tools in detached config
202
 
        '/demo': {
203
 
            'tools.numerify.on': True,
204
 
            'tools.numerify.map': {"pie": "3.14159"},
205
 
        },
206
 
        '/demo/restricted': {
207
 
            'request.show_tracebacks': False,
208
 
        },
209
 
        '/demo/userid': {
210
 
            'request.show_tracebacks': False,
211
 
            'myauth.check_access.default': True,
212
 
        },
213
 
        '/demo/errinstream': {
214
 
            'response.stream': True,
215
 
        },
216
 
        '/demo/err_in_onstart': {
217
 
            # Because this isn't a dict, on_start_resource will error.
218
 
            'tools.numerify.map': "pie->3.14159"
219
 
        },
220
 
        # Combined tools
221
 
        '/euro': {
222
 
            'tools.gzip.on': True,
223
 
            'tools.encode.on': True,
224
 
        },
225
 
        # Priority specified in config
226
 
        '/decorated_euro/subpath': {
227
 
            'tools.gzip.priority': 10,
228
 
        },
229
 
        # Handler wrappers
230
 
        '/tarfile': {'tools.streamer.on': True}
231
 
    }
232
 
    app = cherrypy.tree.mount(root, config=conf)
233
 
    app.request_class.namespaces['myauth'] = myauthtools
234
 
 
235
 
    if sys.version_info >= (2, 5):
236
 
        from cherrypy.test import py25
237
 
        root.tooldecs = py25.ToolExamples()
 
15
europoundUnicode = ntou('\x80\xa3')
238
16
 
239
17
 
240
18
#                             Client-side code                             #
243
21
 
244
22
 
245
23
class ToolTests(helper.CPWebCase):
246
 
    
 
24
    def setup_server():
 
25
        
 
26
        # Put check_access in a custom toolbox with its own namespace
 
27
        myauthtools = cherrypy._cptools.Toolbox("myauth")
 
28
        
 
29
        def check_access(default=False):
 
30
            if not getattr(cherrypy.request, "userid", default):
 
31
                raise cherrypy.HTTPError(401)
 
32
        myauthtools.check_access = cherrypy.Tool('before_request_body', check_access)
 
33
        
 
34
        def numerify():
 
35
            def number_it(body):
 
36
                for chunk in body:
 
37
                    for k, v in cherrypy.request.numerify_map:
 
38
                        chunk = chunk.replace(k, v)
 
39
                    yield chunk
 
40
            cherrypy.response.body = number_it(cherrypy.response.body)
 
41
        
 
42
        class NumTool(cherrypy.Tool):
 
43
            def _setup(self):
 
44
                def makemap():
 
45
                    m = self._merged_args().get("map", {})
 
46
                    cherrypy.request.numerify_map = copyitems(m)
 
47
                cherrypy.request.hooks.attach('on_start_resource', makemap)
 
48
                
 
49
                def critical():
 
50
                    cherrypy.request.error_response = cherrypy.HTTPError(502).set_response
 
51
                critical.failsafe = True
 
52
                
 
53
                cherrypy.request.hooks.attach('on_start_resource', critical)
 
54
                cherrypy.request.hooks.attach(self._point, self.callable)
 
55
        
 
56
        tools.numerify = NumTool('before_finalize', numerify)
 
57
        
 
58
        # It's not mandatory to inherit from cherrypy.Tool.
 
59
        class NadsatTool:
 
60
            
 
61
            def __init__(self):
 
62
                self.ended = {}
 
63
                self._name = "nadsat"
 
64
            
 
65
            def nadsat(self):
 
66
                def nadsat_it_up(body):
 
67
                    for chunk in body:
 
68
                        chunk = chunk.replace(ntob("good"), ntob("horrorshow"))
 
69
                        chunk = chunk.replace(ntob("piece"), ntob("lomtick"))
 
70
                        yield chunk
 
71
                cherrypy.response.body = nadsat_it_up(cherrypy.response.body)
 
72
            nadsat.priority = 0
 
73
            
 
74
            def cleanup(self):
 
75
                # This runs after the request has been completely written out.
 
76
                cherrypy.response.body = [ntob("razdrez")]
 
77
                id = cherrypy.request.params.get("id")
 
78
                if id:
 
79
                    self.ended[id] = True
 
80
            cleanup.failsafe = True
 
81
            
 
82
            def _setup(self):
 
83
                cherrypy.request.hooks.attach('before_finalize', self.nadsat)
 
84
                cherrypy.request.hooks.attach('on_end_request', self.cleanup)
 
85
        tools.nadsat = NadsatTool()
 
86
        
 
87
        def pipe_body():
 
88
            cherrypy.request.process_request_body = False
 
89
            clen = int(cherrypy.request.headers['Content-Length'])
 
90
            cherrypy.request.body = cherrypy.request.rfile.read(clen)
 
91
        
 
92
        # Assert that we can use a callable object instead of a function.
 
93
        class Rotator(object):
 
94
            def __call__(self, scale):
 
95
                r = cherrypy.response
 
96
                r.collapse_body()
 
97
                if py3k:
 
98
                    r.body = [bytes([(x + scale) % 256 for x in r.body[0]])]
 
99
                else:
 
100
                    r.body = [chr((ord(x) + scale) % 256) for x in r.body[0]]
 
101
        cherrypy.tools.rotator = cherrypy.Tool('before_finalize', Rotator())
 
102
        
 
103
        def stream_handler(next_handler, *args, **kwargs):
 
104
            cherrypy.response.output = o = BytesIO()
 
105
            try:
 
106
                response = next_handler(*args, **kwargs)
 
107
                # Ignore the response and return our accumulated output instead.
 
108
                return o.getvalue()
 
109
            finally:
 
110
                o.close()
 
111
        cherrypy.tools.streamer = cherrypy._cptools.HandlerWrapperTool(stream_handler)
 
112
        
 
113
        class Root:
 
114
            def index(self):
 
115
                return "Howdy earth!"
 
116
            index.exposed = True
 
117
            
 
118
            def tarfile(self):
 
119
                cherrypy.response.output.write(ntob('I am '))
 
120
                cherrypy.response.output.write(ntob('a tarfile'))
 
121
            tarfile.exposed = True
 
122
            tarfile._cp_config = {'tools.streamer.on': True}
 
123
            
 
124
            def euro(self):
 
125
                hooks = list(cherrypy.request.hooks['before_finalize'])
 
126
                hooks.sort()
 
127
                cbnames = [x.callback.__name__ for x in hooks]
 
128
                assert cbnames == ['gzip'], cbnames
 
129
                priorities = [x.priority for x in hooks]
 
130
                assert priorities == [80], priorities
 
131
                yield ntou("Hello,")
 
132
                yield ntou("world")
 
133
                yield europoundUnicode
 
134
            euro.exposed = True
 
135
            
 
136
            # Bare hooks
 
137
            def pipe(self):
 
138
                return cherrypy.request.body
 
139
            pipe.exposed = True
 
140
            pipe._cp_config = {'hooks.before_request_body': pipe_body}
 
141
            
 
142
            # Multiple decorators; include kwargs just for fun.
 
143
            # Note that rotator must run before gzip.
 
144
            def decorated_euro(self, *vpath):
 
145
                yield ntou("Hello,")
 
146
                yield ntou("world")
 
147
                yield europoundUnicode
 
148
            decorated_euro.exposed = True
 
149
            decorated_euro = tools.gzip(compress_level=6)(decorated_euro)
 
150
            decorated_euro = tools.rotator(scale=3)(decorated_euro)
 
151
        
 
152
        root = Root()
 
153
        
 
154
        
 
155
        class TestType(type):
 
156
            """Metaclass which automatically exposes all functions in each subclass,
 
157
            and adds an instance of the subclass as an attribute of root.
 
158
            """
 
159
            def __init__(cls, name, bases, dct):
 
160
                type.__init__(cls, name, bases, dct)
 
161
                for value in itervalues(dct):
 
162
                    if isinstance(value, types.FunctionType):
 
163
                        value.exposed = True
 
164
                setattr(root, name.lower(), cls())
 
165
        Test = TestType('Test', (object,), {})
 
166
        
 
167
        
 
168
        # METHOD ONE:
 
169
        # Declare Tools in _cp_config
 
170
        class Demo(Test):
 
171
            
 
172
            _cp_config = {"tools.nadsat.on": True}
 
173
            
 
174
            def index(self, id=None):
 
175
                return "A good piece of cherry pie"
 
176
            
 
177
            def ended(self, id):
 
178
                return repr(tools.nadsat.ended[id])
 
179
            
 
180
            def err(self, id=None):
 
181
                raise ValueError()
 
182
            
 
183
            def errinstream(self, id=None):
 
184
                yield "nonconfidential"
 
185
                raise ValueError()
 
186
                yield "confidential"
 
187
            
 
188
            # METHOD TWO: decorator using Tool()
 
189
            # We support Python 2.3, but the @-deco syntax would look like this:
 
190
            # @tools.check_access()
 
191
            def restricted(self):
 
192
                return "Welcome!"
 
193
            restricted = myauthtools.check_access()(restricted)
 
194
            userid = restricted
 
195
            
 
196
            def err_in_onstart(self):
 
197
                return "success!"
 
198
            
 
199
            def stream(self, id=None):
 
200
                for x in xrange(100000000):
 
201
                    yield str(x)
 
202
            stream._cp_config = {'response.stream': True}
 
203
        
 
204
        
 
205
        conf = {
 
206
            # METHOD THREE:
 
207
            # Declare Tools in detached config
 
208
            '/demo': {
 
209
                'tools.numerify.on': True,
 
210
                'tools.numerify.map': {ntob("pie"): ntob("3.14159")},
 
211
            },
 
212
            '/demo/restricted': {
 
213
                'request.show_tracebacks': False,
 
214
            },
 
215
            '/demo/userid': {
 
216
                'request.show_tracebacks': False,
 
217
                'myauth.check_access.default': True,
 
218
            },
 
219
            '/demo/errinstream': {
 
220
                'response.stream': True,
 
221
            },
 
222
            '/demo/err_in_onstart': {
 
223
                # Because this isn't a dict, on_start_resource will error.
 
224
                'tools.numerify.map': "pie->3.14159"
 
225
            },
 
226
            # Combined tools
 
227
            '/euro': {
 
228
                'tools.gzip.on': True,
 
229
                'tools.encode.on': True,
 
230
            },
 
231
            # Priority specified in config
 
232
            '/decorated_euro/subpath': {
 
233
                'tools.gzip.priority': 10,
 
234
            },
 
235
            # Handler wrappers
 
236
            '/tarfile': {'tools.streamer.on': True}
 
237
        }
 
238
        app = cherrypy.tree.mount(root, config=conf)
 
239
        app.request_class.namespaces['myauth'] = myauthtools
 
240
        
 
241
        if sys.version_info >= (2, 5):
 
242
            from cherrypy.test import _test_decorators
 
243
            root.tooldecs = _test_decorators.ToolExamples()
 
244
    setup_server = staticmethod(setup_server)
 
245
 
247
246
    def testHookErrors(self):
248
247
        self.getPage("/demo/?id=1")
249
248
        # If body is "razdrez", then on_end_request is being called too early.
295
294
            httpserver = cherrypy.server.httpserver
296
295
            old_timeout = httpserver.timeout
297
296
        except (AttributeError, IndexError):
298
 
            print "skipped ",
299
 
            return
 
297
            return self.skip()
300
298
        
301
299
        try:
302
300
            httpserver.timeout = timeout
334
332
        self.assertInBody("AttributeError: 'str' object has no attribute 'items'")
335
333
    
336
334
    def testCombinedTools(self):
337
 
        expectedResult = (u"Hello,world" + europoundUnicode).encode('utf-8')
338
 
        zbuf = StringIO.StringIO()
 
335
        expectedResult = (ntou("Hello,world") + europoundUnicode).encode('utf-8')
 
336
        zbuf = BytesIO()
339
337
        zfile = gzip.GzipFile(mode='wb', fileobj=zbuf, compresslevel=9)
340
338
        zfile.write(expectedResult)
341
339
        zfile.close()
344
342
                                        ("Accept-Charset", "ISO-8859-1,utf-8;q=0.7,*;q=0.7")])
345
343
        self.assertInBody(zbuf.getvalue()[:3])
346
344
        
347
 
        zbuf = StringIO.StringIO()
 
345
        zbuf = BytesIO()
348
346
        zfile = gzip.GzipFile(mode='wb', fileobj=zbuf, compresslevel=6)
349
347
        zfile.write(expectedResult)
350
348
        zfile.close()
352
350
        self.getPage("/decorated_euro", headers=[("Accept-Encoding", "gzip")])
353
351
        self.assertInBody(zbuf.getvalue()[:3])
354
352
        
355
 
        # This should break because gzip's priority was lowered in conf.
 
353
        # This returns a different value because gzip's priority was
 
354
        # lowered in conf, allowing the rotator to run after gzip.
356
355
        # Of course, we don't want breakage in production apps,
357
356
        # but it proves the priority was changed.
358
357
        self.getPage("/decorated_euro/subpath",
359
358
                     headers=[("Accept-Encoding", "gzip")])
360
 
        self.assertErrorPage(500, pattern='UnicodeEncodeError')
 
359
        if py3k:
 
360
            self.assertInBody(bytes([(x + 3) % 256 for x in zbuf.getvalue()]))
 
361
        else:
 
362
            self.assertInBody(''.join([chr((ord(x) + 3) % 256) for x in zbuf.getvalue()]))
361
363
    
362
364
    def testBareHooks(self):
363
365
        content = "bit of a pain in me gulliver"
364
366
        self.getPage("/pipe",
365
 
                     headers=[("Content-Length", len(content)),
 
367
                     headers=[("Content-Length", str(len(content))),
366
368
                              ("Content-Type", "text/plain")],
367
369
                     method="POST", body=content)
368
370
        self.assertBody(content)
373
375
    
374
376
    def testToolWithConfig(self):
375
377
        if not sys.version_info >= (2, 5):
376
 
            print "skipped (Python 2.5+ only)",
377
 
            return
 
378
            return self.skip("skipped (Python 2.5+ only)")
378
379
        
379
380
        self.getPage('/tooldecs/blah')
380
381
        self.assertHeader('Content-Type', 'application/data')
381
 
 
382
 
 
383
 
if __name__ == '__main__':
384
 
    setup_server()
385
 
    helper.testmain()
 
382
    
 
383
    def testWarnToolOn(self):
 
384
        # get
 
385
        try:
 
386
            numon = cherrypy.tools.numerify.on
 
387
        except AttributeError:
 
388
            pass
 
389
        else:
 
390
            raise AssertionError("Tool.on did not error as it should have.")
 
391
        
 
392
        # set
 
393
        try:
 
394
            cherrypy.tools.numerify.on = True
 
395
        except AttributeError:
 
396
            pass
 
397
        else:
 
398
            raise AssertionError("Tool.on did not error as it should have.")
386
399