← Back to branch summary

~ubuntu-branches/ubuntu/lucid/twisted-web2/lucid

« back to all changes in this revision

Viewing changes to twisted/web2/test/test_server.py

  • Committer: Bazaar Package Importer
  • Author(s): Matthias Klose
  • Date: 2008-03-30 22:08:20 UTC
  • mfrom: (0.1.5 upstream)
  • Revision ID: james.westby@ubuntu.com-20080330220820-m2cd4jon35wz2efx
Tags: 8.0.1-1
New upstream version.

Show diffs side-by-side

added added

removed removed

Lines of Context:
twisted/web2/test/test_server.py
 
1
# Copyright (c) 2001-2007 Twisted Matrix Laboratories.
 
2
# See LICENSE for details.
 
3
 
1
4
"""
2
5
A test harness for the twisted.web2 server.
3
6
"""
7
10
from twisted.python import components
8
11
from twisted.web2 import http, http_headers, iweb, server
9
12
from twisted.web2 import resource, stream, compat
10
 
from twisted.trial import unittest, util
11
 
from twisted.internet import reactor, defer, address, error as ti_error
 
13
from twisted.trial import unittest
 
14
from twisted.internet import reactor, defer, address
 
15
 
12
16
 
13
17
 
14
18
class NotResource(object):
132
136
 
133
137
 
134
138
 
135
 
 
136
139
class SimpleRequest(server.Request):
137
140
    """I can be used in cases where a Request object is necessary
138
141
    but it is benificial to bypass the chanRequest
139
142
    """
140
143
 
141
144
    clientproto = (1,1)
142
 
    
 
145
 
143
146
    def __init__(self, site, method, uri, headers=None, content=None):
144
147
        if not headers:
145
148
            headers = http_headers.Headers(headers)
146
 
            
 
149
 
147
150
        super(SimpleRequest, self).__init__(
148
151
            site=site,
149
152
            chanRequest=None,
170
173
    hostInfo = address.IPv4Address('TCP', 'host', 80), False
171
174
    remoteHost = address.IPv4Address('TCP', 'remotehost', 34567)
172
175
 
173
 
    
 
176
 
174
177
    def __init__(self, site, method, prepath, uri, length=None,
175
178
                 headers=None, version=(1,1), content=None):
176
179
        self.site = site
190
193
                                      self.headers,
191
194
                                      site=self.site,
192
195
                                      prepathuri=self.prepath)
193
 
        
 
196
 
194
197
        if content is not None:
195
198
            self.request.handleContentChunk(content)
196
199
            self.request.handleContentComplete()
197
 
            
 
200
 
198
201
        self.code = None
199
202
        self.responseHeaders = None
200
203
        self.data = ''
202
205
 
203
206
    def writeIntermediateResponse(code, headers=None):
204
207
        pass
205
 
    
 
208
 
206
209
    def writeHeaders(self, code, headers):
207
210
        self.responseHeaders = headers
208
211
        self.code = code
209
 
        
 
212
 
210
213
    def write(self, data):
211
214
        self.data += data
212
215
 
217
220
 
218
221
    def abortConnection(self):
219
222
        self.finish(failed=True)
220
 
        
 
223
 
221
224
    def registerProducer(self, producer, streaming):
222
225
        pass
223
226
 
229
232
 
230
233
    def getRemoteHost(self):
231
234
        return self.remoteHost
232
 
    
 
235
 
233
236
 
234
237
class BaseTestResource(resource.Resource):
235
238
    responseCode = 200
236
239
    responseText = 'This is a fake resource.'
237
240
    responseHeaders = {}
238
241
    addSlash = False
239
 
    
 
242
 
240
243
    def __init__(self, children=[]):
241
244
        """
242
245
        @type children: C{list} of C{tuple}
252
255
    def responseStream(self):
253
256
        return stream.MemoryStream(self.responseText)
254
257
 
 
258
 
 
259
 
255
260
_unset = object()
256
261
class BaseCase(unittest.TestCase):
257
262
    """
258
263
    Base class for test cases that involve testing the result
259
264
    of arbitrary HTTP(S) queries.
260
265
    """
261
 
    
 
266
 
262
267
    method = 'GET'
263
268
    version = (1, 1)
264
269
    wait_timeout = 5.0
265
 
    
 
270
 
266
271
    def chanrequest(self, root, uri, length, headers, method, version, prepath, content):
267
272
        site = server.Site(root)
268
273
        return TestChanRequest(site, method, prepath, uri, length, headers, version, content)
276
281
                length = len(content)
277
282
            else:
278
283
                length = 0
279
 
            
 
284
 
280
285
        if method is None:
281
286
            method = self.method
282
287
        if version is None:
299
304
        """
300
305
        d = self.getResponseFor(*request_data)
301
306
        d.addCallback(self._cbGotResponse, expected_response, failure)
302
 
        
 
307
 
303
308
        return d
304
309
 
305
310
    def _cbGotResponse(self, (code, headers, data, failed), expected_response, expectedfailure=False):
311
316
            self.assertEquals(headers.getHeader(key), value)
312
317
        self.assertEquals(failed, expectedfailure)
313
318
 
 
319
 
 
320
 
314
321
class SampleWebTest(BaseCase):
315
322
    class SampleTestResource(BaseTestResource):
316
323
        addSlash = True
376
383
        return self.assertResponse(
377
384
            (redirectResource, 'http://localhost/'),
378
385
            (301, {'location': 'https://localhost/foo?bar=baz'}, None))
379
 
    
 
386
 
380
387
 
381
388
class URLParsingTest(BaseCase):
382
389
    class TestResource(resource.LeafResource):
383
390
        def render(self, req):
384
391
            return http.Response(stream="Host:%s, Path:%s"%(req.host, req.path))
385
 
            
 
392
 
386
393
    def setUp(self):
387
394
        self.root = self.TestResource()
388
395
 
407
414
            (self.root, 'http://host//path'),
408
415
            (200, {}, 'Host:host, Path://path'))
409
416
 
 
417
 
 
418
 
410
419
class TestDeferredRendering(BaseCase):
411
420
    class ResourceWithDeferreds(BaseTestResource):
412
421
        addSlash=True
421
430
            d = defer.Deferred()
422
431
            reactor.callLater(0, d.callback, BaseTestResource())
423
432
            return d
424
 
        
 
433
 
425
434
    def test_deferredRootResource(self):
426
435
        return self.assertResponse(
427
436
            (self.ResourceWithDeferreds(), 'http://host/'),
432
441
            (self.ResourceWithDeferreds(), 'http://host/deferred'),
433
442
            (200, {}, 'This is a fake resource.'))
434
443
 
 
444
 
 
445
 
435
446
class RedirectResourceTest(BaseCase):
436
447
    def html(url):
437
448
        return "<html><head><title>Moved Permanently</title></head><body><h1>Moved Permanently</h1><p>Document moved to %s.</p></body></html>" % (url,)
476
487
        return defer.DeferredList(ds, fireOnOneErrback=True)
477
488
 
478
489
 
 
490
 
479
491
class EmptyResource(resource.Resource):
480
492
    def __init__(self, test):
481
493
        self.test = test
485
497
        return 201
486
498
 
487
499
 
 
500
 
488
501
class RememberURIs(BaseCase):
489
502
    """
490
503
    Tests for URI memory and lookup mechanism in server.Request.
624
637
        d = request.locateResource("/foo")
625
638
        d.addCallback(gotResource)
626
639
        return d
 
640
 
 
641
 
 
642
 
 
643
class ParsePostDataTests(unittest.TestCase):
 
644
    """
 
645
    Tests for L{server.parsePOSTData}.
 
646
    """
 
647
 
 
648
    def test_noData(self):
 
649
        """
 
650
        Parsing a request without data should succeed but should not fill the
 
651
        C{args} and C{files} attributes of the request.
 
652
        """
 
653
        root = resource.Resource()
 
654
        request = SimpleRequest(server.Site(root), "GET", "/")
 
655
        def cb(ign):
 
656
            self.assertEquals(request.args, {})
 
657
            self.assertEquals(request.files, {})
 
658
        return server.parsePOSTData(request).addCallback(cb)
 
659
 
 
660
 
 
661
    def test_noContentType(self):
 
662
        """
 
663
        Parsing a request without content-type should succeed but should not
 
664
        fill the C{args} and C{files} attributes of the request.
 
665
        """
 
666
        root = resource.Resource()
 
667
        request = SimpleRequest(server.Site(root), "GET", "/", content="foo")
 
668
        def cb(ign):
 
669
            self.assertEquals(request.args, {})
 
670
            self.assertEquals(request.files, {})
 
671
        return server.parsePOSTData(request).addCallback(cb)
 
672
 
 
673
 
 
674
    def test_urlencoded(self):
 
675
        """
 
676
        Test parsing data in urlencoded format: it should end in the C{args}
 
677
        attribute.
 
678
        """
 
679
        ctype = http_headers.MimeType('application', 'x-www-form-urlencoded')
 
680
        content = "key=value&multiple=two+words&multiple=more%20words"
 
681
        root = resource.Resource()
 
682
        request = SimpleRequest(server.Site(root), "GET", "/",
 
683
                http_headers.Headers({'content-type': ctype}), content)
 
684
        def cb(ign):
 
685
            self.assertEquals(request.files, {})
 
686
            self.assertEquals(request.args,
 
687
                {'multiple': ['two words', 'more words'], 'key': ['value']})
 
688
        return server.parsePOSTData(request).addCallback(cb)
 
689
 
 
690
 
 
691
    def test_multipart(self):
 
692
        """
 
693
        Test parsing data in multipart format: it should fill the C{files}
 
694
        attribute.
 
695
        """
 
696
        ctype = http_headers.MimeType('multipart', 'form-data',
 
697
                                      (('boundary', '---weeboundary'),))
 
698
        content="""-----weeboundary\r
 
699
Content-Disposition: form-data; name="FileNameOne"; filename="myfilename"\r
 
700
Content-Type: text/html\r
 
701
\r
 
702
my great content wooo\r
 
703
-----weeboundary--\r
 
704
"""
 
705
        root = resource.Resource()
 
706
        request = SimpleRequest(server.Site(root), "GET", "/",
 
707
                http_headers.Headers({'content-type': ctype}), content)
 
708
        def cb(ign):
 
709
            self.assertEquals(request.args, {})
 
710
            self.assertEquals(request.files.keys(), ['FileNameOne'])
 
711
            self.assertEquals(request.files.values()[0][0][:2],
 
712
                  ('myfilename', http_headers.MimeType('text', 'html', {})))
 
713
            f = request.files.values()[0][0][2]
 
714
            self.assertEquals(f.read(), "my great content wooo")
 
715
        return server.parsePOSTData(request).addCallback(cb)
 
716
 
 
717
 
 
718
    def test_multipartWithNoBoundary(self):
 
719
        """
 
720
        If the boundary type is not specified, parsing should fail with a
 
721
        C{http.HTTPError}.
 
722
        """
 
723
        ctype = http_headers.MimeType('multipart', 'form-data')
 
724
        content="""-----weeboundary\r
 
725
Content-Disposition: form-data; name="FileNameOne"; filename="myfilename"\r
 
726
Content-Type: text/html\r
 
727
\r
 
728
my great content wooo\r
 
729
-----weeboundary--\r
 
730
"""
 
731
        root = resource.Resource()
 
732
        request = SimpleRequest(server.Site(root), "GET", "/",
 
733
                http_headers.Headers({'content-type': ctype}), content)
 
734
        return self.assertFailure(server.parsePOSTData(request),
 
735
            http.HTTPError)
 
736
 
 
737
 
 
738
    def test_wrongContentType(self):
 
739
        """
 
740
        Check that a content-type not handled raise a C{http.HTTPError}.
 
741
        """
 
742
        ctype = http_headers.MimeType('application', 'foobar')
 
743
        content = "key=value&multiple=two+words&multiple=more%20words"
 
744
        root = resource.Resource()
 
745
        request = SimpleRequest(server.Site(root), "GET", "/",
 
746
                http_headers.Headers({'content-type': ctype}), content)
 
747
        return self.assertFailure(server.parsePOSTData(request),
 
748
            http.HTTPError)
 
749
 
 
750
 
 
751
    def test_mimeParsingError(self):
 
752
        """
 
753
        A malformed content should result in a C{http.HTTPError}.
 
754
        
 
755
        The tested content has an invalid closing boundary.
 
756
        """
 
757
        ctype = http_headers.MimeType('multipart', 'form-data',
 
758
                                      (('boundary', '---weeboundary'),))
 
759
        content="""-----weeboundary\r
 
760
Content-Disposition: form-data; name="FileNameOne"; filename="myfilename"\r
 
761
Content-Type: text/html\r
 
762
\r
 
763
my great content wooo\r
 
764
-----weeoundary--\r
 
765
"""
 
766
        root = resource.Resource()
 
767
        request = SimpleRequest(server.Site(root), "GET", "/",
 
768
                http_headers.Headers({'content-type': ctype}), content)
 
769
        return self.assertFailure(server.parsePOSTData(request),
 
770
            http.HTTPError)
 
771
 
 
772
 
 
773
    def test_multipartMaxMem(self):
 
774
        """
 
775
        Check that the C{maxMem} parameter makes the parsing raise an
 
776
        exception if the value is reached.
 
777
        """
 
778
        ctype = http_headers.MimeType('multipart', 'form-data',
 
779
                                      (('boundary', '---weeboundary'),))
 
780
        content="""-----weeboundary\r
 
781
Content-Disposition: form-data; name="FileNameOne"\r
 
782
Content-Type: text/html\r
 
783
\r
 
784
my great content wooo
 
785
and even more and more\r
 
786
-----weeboundary--\r
 
787
"""
 
788
        root = resource.Resource()
 
789
        request = SimpleRequest(server.Site(root), "GET", "/",
 
790
                http_headers.Headers({'content-type': ctype}), content)
 
791
        def cb(res):
 
792
            self.assertEquals(res.response.description,
 
793
                "Maximum length of 10 bytes exceeded.")
 
794
        return self.assertFailure(server.parsePOSTData(request, maxMem=10),
 
795
            http.HTTPError).addCallback(cb)
 
796
 
 
797
 
 
798
    def test_multipartMaxSize(self):
 
799
        """
 
800
        Check that the C{maxSize} parameter makes the parsing raise an
 
801
        exception if the data is too big.
 
802
        """
 
803
        ctype = http_headers.MimeType('multipart', 'form-data',
 
804
                                      (('boundary', '---weeboundary'),))
 
805
        content="""-----weeboundary\r
 
806
Content-Disposition: form-data; name="FileNameOne"; filename="myfilename"\r
 
807
Content-Type: text/html\r
 
808
\r
 
809
my great content wooo
 
810
and even more and more\r
 
811
-----weeboundary--\r
 
812
"""
 
813
        root = resource.Resource()
 
814
        request = SimpleRequest(server.Site(root), "GET", "/",
 
815
                http_headers.Headers({'content-type': ctype}), content)
 
816
        def cb(res):
 
817
            self.assertEquals(res.response.description,
 
818
                "Maximum length of 10 bytes exceeded.")
 
819
        return self.assertFailure(server.parsePOSTData(request, maxSize=10),
 
820
            http.HTTPError).addCallback(cb)
 
821
 
 
822
 
 
823
    def test_maxFields(self):
 
824
        """
 
825
        Check that the C{maxSize} parameter makes the parsing raise an
 
826
        exception if the data contains too many fields.
 
827
        """
 
828
        ctype = http_headers.MimeType('multipart', 'form-data',
 
829
                                      (('boundary', '---xyz'),))
 
830
        content = """-----xyz\r
 
831
Content-Disposition: form-data; name="foo"\r
 
832
\r
 
833
Foo Bar\r
 
834
-----xyz\r
 
835
Content-Disposition: form-data; name="foo"\r
 
836
\r
 
837
Baz\r
 
838
-----xyz\r
 
839
Content-Disposition: form-data; name="file"; filename="filename"\r
 
840
Content-Type: text/html\r
 
841
\r
 
842
blah\r
 
843
-----xyz\r
 
844
Content-Disposition: form-data; name="file"; filename="filename"\r
 
845
Content-Type: text/plain\r
 
846
\r
 
847
bleh\r
 
848
-----xyz--\r
 
849
"""
 
850
        root = resource.Resource()
 
851
        request = SimpleRequest(server.Site(root), "GET", "/",
 
852
                http_headers.Headers({'content-type': ctype}), content)
 
853
        def cb(res):
 
854
            self.assertEquals(res.response.description,
 
855
                "Maximum number of fields 3 exceeded")
 
856
        return self.assertFailure(server.parsePOSTData(request, maxFields=3),
 
857
            http.HTTPError).addCallback(cb)
 
858
 
 
859
 
 
860
    def test_otherErrors(self):
 
861
        """
 
862
        Test that errors durign parsing other than C{MimeFormatError} are
 
863
        propagated.
 
864
        """
 
865
        ctype = http_headers.MimeType('multipart', 'form-data',
 
866
                                      (('boundary', '---weeboundary'),))
 
867
        # XXX: maybe this is not a good example
 
868
        # parseContentDispositionFormData could handle this problem
 
869
        content="""-----weeboundary\r
 
870
Content-Disposition: form-data; name="FileNameOne"; filename="myfilename and invalid data \r
 
871
-----weeboundary--\r
 
872
"""
 
873
        root = resource.Resource()
 
874
        request = SimpleRequest(server.Site(root), "GET", "/",
 
875
                http_headers.Headers({'content-type': ctype}), content)
 
876
        return self.assertFailure(server.parsePOSTData(request),
 
877
            ValueError)
 
878