1
# Copyright (c) 2001-2010 Twisted Matrix Laboratories.
2
# See LICENSE for details.
8
from urlparse import urlparse, urlunsplit, clear_cache
9
import random, urllib, cgi
11
from twisted.python.compat import set
12
from twisted.python.failure import Failure
13
from twisted.trial import unittest
14
from twisted.trial.unittest import TestCase
15
from twisted.web import http, http_headers
16
from twisted.web.http import PotentialDataLoss, _DataLoss
17
from twisted.web.http import _IdentityTransferDecoder
18
from twisted.protocols import loopback
19
from twisted.internet.task import Clock
20
from twisted.internet.error import ConnectionLost
21
from twisted.test.proto_helpers import StringTransport
22
from twisted.test.test_internet import DummyProducer
23
from twisted.web.test.test_web import DummyChannel
26
class DateTimeTest(unittest.TestCase):
27
"""Test date parsing functions."""
29
def testRoundtrip(self):
30
for i in range(10000):
31
time = random.randint(0, 2000000000)
32
timestr = http.datetimeToString(time)
33
time2 = http.stringToDatetime(timestr)
34
self.assertEquals(time, time2)
37
class DummyHTTPHandler(http.Request):
40
self.content.seek(0, 0)
41
data = self.content.read()
42
length = self.getHeader('content-length')
43
request = "'''\n"+str(length)+"\n"+data+"'''\n"
44
self.setResponseCode(200)
45
self.setHeader("Request", self.uri)
46
self.setHeader("Command", self.method)
47
self.setHeader("Version", self.clientproto)
48
self.setHeader("Content-Length", len(request))
53
class LoopbackHTTPClient(http.HTTPClient):
55
def connectionMade(self):
56
self.sendCommand("GET", "/foo/bar")
57
self.sendHeader("Content-Length", 10)
59
self.transport.write("0123456789")
62
class ResponseTestMixin(object):
64
A mixin that provides a simple means of comparing an actual response string
65
to an expected response string by performing the minimal parsing.
68
def assertResponseEquals(self, responses, expected):
70
Assert that the C{responses} matches the C{expected} responses.
72
@type responses: C{str}
73
@param responses: The bytes sent in response to one or more requests.
75
@type expected: C{list} of C{tuple} of C{str}
76
@param expected: The expected values for the responses. Each tuple
77
element of the list represents one response. Each string element
78
of the tuple is a full header line without delimiter, except for
79
the last element which gives the full response body.
81
for response in expected:
82
expectedHeaders, expectedContent = response[:-1], response[-1]
83
headers, rest = responses.split('\r\n\r\n', 1)
84
headers = headers.splitlines()
85
self.assertEqual(set(headers), set(expectedHeaders))
86
content = rest[:len(expectedContent)]
87
responses = rest[len(expectedContent):]
88
self.assertEqual(content, expectedContent)
92
class HTTP1_0TestCase(unittest.TestCase, ResponseTestMixin):
97
"Accept: text/html\r\n"
100
expected_response = [
105
"Content-Length: 13",
108
def test_buffer(self):
110
Send requests over a channel and check responses match what is expected.
112
b = StringTransport()
113
a = http.HTTPChannel()
114
a.requestFactory = DummyHTTPHandler
116
# one byte at a time, to stress it.
117
for byte in self.requests:
119
a.connectionLost(IOError("all one"))
121
self.assertResponseEquals(value, self.expected_response)
124
def test_requestBodyTimeout(self):
126
L{HTTPChannel} resets its timeout whenever data from a request body is
130
transport = StringTransport()
131
protocol = http.HTTPChannel()
132
protocol.timeOut = 100
133
protocol.callLater = clock.callLater
134
protocol.makeConnection(transport)
135
protocol.dataReceived('POST / HTTP/1.0\r\nContent-Length: 2\r\n\r\n')
137
self.assertFalse(transport.disconnecting)
138
protocol.dataReceived('x')
140
self.assertFalse(transport.disconnecting)
141
protocol.dataReceived('x')
142
self.assertEqual(len(protocol.requests), 1)
146
class HTTP1_1TestCase(HTTP1_0TestCase):
150
"Accept: text/html\r\n"
152
"POST / HTTP/1.1\r\n"
153
"Content-Length: 10\r\n"
155
"0123456789POST / HTTP/1.1\r\n"
156
"Content-Length: 10\r\n"
158
"0123456789HEAD / HTTP/1.1\r\n"
161
expected_response = [
166
"Content-Length: 13",
172
"Content-Length: 21",
173
"'''\n10\n0123456789'''\n"),
178
"Content-Length: 21",
179
"'''\n10\n0123456789'''\n"),
184
"Content-Length: 13",
189
class HTTP1_1_close_TestCase(HTTP1_0TestCase):
193
"Accept: text/html\r\n"
194
"Connection: close\r\n"
199
expected_response = [
205
"Content-Length: 13",
210
class HTTP0_9TestCase(HTTP1_0TestCase):
215
expected_response = "HTTP/1.1 400 Bad Request\r\n\r\n"
218
def assertResponseEquals(self, response, expectedResponse):
219
self.assertEquals(response, expectedResponse)
222
class HTTPLoopbackTestCase(unittest.TestCase):
224
expectedHeaders = {'request' : '/foo/bar',
226
'version' : 'HTTP/1.0',
227
'content-length' : '21'}
233
def _handleStatus(self, version, status, message):
235
self.assertEquals(version, "HTTP/1.0")
236
self.assertEquals(status, "200")
238
def _handleResponse(self, data):
240
self.assertEquals(data, "'''\n10\n0123456789'''\n")
242
def _handleHeader(self, key, value):
243
self.numHeaders = self.numHeaders + 1
244
self.assertEquals(self.expectedHeaders[key.lower()], value)
246
def _handleEndHeaders(self):
247
self.gotEndHeaders = 1
248
self.assertEquals(self.numHeaders, 4)
250
def testLoopback(self):
251
server = http.HTTPChannel()
252
server.requestFactory = DummyHTTPHandler
253
client = LoopbackHTTPClient()
254
client.handleResponse = self._handleResponse
255
client.handleHeader = self._handleHeader
256
client.handleEndHeaders = self._handleEndHeaders
257
client.handleStatus = self._handleStatus
258
d = loopback.loopbackAsync(server, client)
259
d.addCallback(self._cbTestLoopback)
262
def _cbTestLoopback(self, ignored):
263
if not (self.gotStatus and self.gotResponse and self.gotEndHeaders):
265
"didn't got all callbacks %s"
266
% [self.gotStatus, self.gotResponse, self.gotEndHeaders])
267
del self.gotEndHeaders
274
def _prequest(**headers):
276
Make a request with the given request headers for the persistence tests.
278
request = http.Request(DummyChannel(), None)
279
for k, v in headers.iteritems():
280
request.requestHeaders.setRawHeaders(k, v)
284
class PersistenceTestCase(unittest.TestCase):
286
Tests for persistent HTTP connections.
289
ptests = [#(PRequest(connection="Keep-Alive"), "HTTP/1.0", 1, {'connection' : 'Keep-Alive'}),
290
(_prequest(), "HTTP/1.0", 0, {'connection': None}),
291
(_prequest(connection=["close"]), "HTTP/1.1", 0, {'connection' : ['close']}),
292
(_prequest(), "HTTP/1.1", 1, {'connection': None}),
293
(_prequest(), "HTTP/0.9", 0, {'connection': None}),
297
def testAlgorithm(self):
298
c = http.HTTPChannel()
299
for req, version, correctResult, resultHeaders in self.ptests:
300
result = c.checkPersistence(req, version)
301
self.assertEquals(result, correctResult)
302
for header in resultHeaders.keys():
303
self.assertEquals(req.responseHeaders.getRawHeaders(header, None), resultHeaders[header])
307
class IdentityTransferEncodingTests(TestCase):
309
Tests for L{_IdentityTransferDecoder}.
313
Create an L{_IdentityTransferDecoder} with callbacks hooked up so that
314
calls to them can be inspected.
318
self.contentLength = 10
319
self.decoder = _IdentityTransferDecoder(
320
self.contentLength, self.data.append, self.finish.append)
323
def test_exactAmountReceived(self):
325
If L{_IdentityTransferDecoder.dataReceived} is called with a string
326
with length equal to the content length passed to
327
L{_IdentityTransferDecoder}'s initializer, the data callback is invoked
328
with that string and the finish callback is invoked with a zero-length
331
self.decoder.dataReceived('x' * self.contentLength)
332
self.assertEqual(self.data, ['x' * self.contentLength])
333
self.assertEqual(self.finish, [''])
336
def test_shortStrings(self):
338
If L{_IdentityTransferDecoder.dataReceived} is called multiple times
339
with strings which, when concatenated, are as long as the content
340
length provided, the data callback is invoked with each string and the
341
finish callback is invoked only after the second call.
343
self.decoder.dataReceived('x')
344
self.assertEqual(self.data, ['x'])
345
self.assertEqual(self.finish, [])
346
self.decoder.dataReceived('y' * (self.contentLength - 1))
347
self.assertEqual(self.data, ['x', 'y' * (self.contentLength - 1)])
348
self.assertEqual(self.finish, [''])
351
def test_longString(self):
353
If L{_IdentityTransferDecoder.dataReceived} is called with a string
354
with length greater than the provided content length, only the prefix
355
of that string up to the content length is passed to the data callback
356
and the remainder is passed to the finish callback.
358
self.decoder.dataReceived('x' * self.contentLength + 'y')
359
self.assertEqual(self.data, ['x' * self.contentLength])
360
self.assertEqual(self.finish, ['y'])
363
def test_rejectDataAfterFinished(self):
365
If data is passed to L{_IdentityTransferDecoder.dataReceived} after the
366
finish callback has been invoked, L{RuntimeError} is raised.
371
decoder.dataReceived('foo')
373
failures.append(Failure())
374
decoder = _IdentityTransferDecoder(5, self.data.append, finish)
375
decoder.dataReceived('x' * 4)
376
self.assertEqual(failures, [])
377
decoder.dataReceived('y')
378
failures[0].trap(RuntimeError)
380
str(failures[0].value),
381
"_IdentityTransferDecoder cannot decode data after finishing")
384
def test_unknownContentLength(self):
386
If L{_IdentityTransferDecoder} is constructed with C{None} for the
387
content length, it passes all data delivered to it through to the data
392
decoder = _IdentityTransferDecoder(None, data.append, finish.append)
393
decoder.dataReceived('x')
394
self.assertEqual(data, ['x'])
395
decoder.dataReceived('y')
396
self.assertEqual(data, ['x', 'y'])
397
self.assertEqual(finish, [])
400
def _verifyCallbacksUnreferenced(self, decoder):
402
Check the decoder's data and finish callbacks and make sure they are
403
None in order to help avoid references cycles.
405
self.assertIdentical(decoder.dataCallback, None)
406
self.assertIdentical(decoder.finishCallback, None)
409
def test_earlyConnectionLose(self):
411
L{_IdentityTransferDecoder.noMoreData} raises L{_DataLoss} if it is
412
called and the content length is known but not enough bytes have been
415
self.decoder.dataReceived('x' * (self.contentLength - 1))
416
self.assertRaises(_DataLoss, self.decoder.noMoreData)
417
self._verifyCallbacksUnreferenced(self.decoder)
420
def test_unknownContentLengthConnectionLose(self):
422
L{_IdentityTransferDecoder.noMoreData} calls the finish callback and
423
raises L{PotentialDataLoss} if it is called and the content length is
428
decoder = _IdentityTransferDecoder(None, body.append, finished.append)
429
self.assertRaises(PotentialDataLoss, decoder.noMoreData)
430
self.assertEqual(body, [])
431
self.assertEqual(finished, [''])
432
self._verifyCallbacksUnreferenced(decoder)
435
def test_finishedConnectionLose(self):
437
L{_IdentityTransferDecoder.noMoreData} does not raise any exception if
438
it is called when the content length is known and that many bytes have
441
self.decoder.dataReceived('x' * self.contentLength)
442
self.decoder.noMoreData()
443
self._verifyCallbacksUnreferenced(self.decoder)
447
class ChunkedTransferEncodingTests(unittest.TestCase):
449
Tests for L{_ChunkedTransferDecoder}, which turns a byte stream encoded
450
using HTTP I{chunked} C{Transfer-Encoding} back into the original byte
453
def test_decoding(self):
455
L{_ChunkedTransferDecoder.dataReceived} decodes chunked-encoded data
456
and passes the result to the specified callback.
459
p = http._ChunkedTransferDecoder(L.append, None)
460
p.dataReceived('3\r\nabc\r\n5\r\n12345\r\n')
461
p.dataReceived('a\r\n0123456789\r\n')
462
self.assertEqual(L, ['abc', '12345', '0123456789'])
465
def test_short(self):
467
L{_ChunkedTransferDecoder.dataReceived} decodes chunks broken up and
468
delivered in multiple calls.
472
p = http._ChunkedTransferDecoder(L.append, finished.append)
473
for s in '3\r\nabc\r\n5\r\n12345\r\n0\r\n\r\n':
475
self.assertEqual(L, ['a', 'b', 'c', '1', '2', '3', '4', '5'])
476
self.assertEqual(finished, [''])
479
def test_newlines(self):
481
L{_ChunkedTransferDecoder.dataReceived} doesn't treat CR LF pairs
482
embedded in chunk bodies specially.
485
p = http._ChunkedTransferDecoder(L.append, None)
486
p.dataReceived('2\r\n\r\n\r\n')
487
self.assertEqual(L, ['\r\n'])
490
def test_extensions(self):
492
L{_ChunkedTransferDecoder.dataReceived} disregards chunk-extension
496
p = http._ChunkedTransferDecoder(L.append, None)
497
p.dataReceived('3; x-foo=bar\r\nabc\r\n')
498
self.assertEqual(L, ['abc'])
501
def test_finish(self):
503
L{_ChunkedTransferDecoder.dataReceived} interprets a zero-length
504
chunk as the end of the chunked data stream and calls the completion
508
p = http._ChunkedTransferDecoder(None, finished.append)
509
p.dataReceived('0\r\n\r\n')
510
self.assertEqual(finished, [''])
513
def test_extra(self):
515
L{_ChunkedTransferDecoder.dataReceived} passes any bytes which come
516
after the terminating zero-length chunk to the completion callback.
519
p = http._ChunkedTransferDecoder(None, finished.append)
520
p.dataReceived('0\r\n\r\nhello')
521
self.assertEqual(finished, ['hello'])
524
def test_afterFinished(self):
526
L{_ChunkedTransferDecoder.dataReceived} raises L{RuntimeError} if it
527
is called after it has seen the last chunk.
529
p = http._ChunkedTransferDecoder(None, lambda bytes: None)
530
p.dataReceived('0\r\n\r\n')
531
self.assertRaises(RuntimeError, p.dataReceived, 'hello')
534
def test_earlyConnectionLose(self):
536
L{_ChunkedTransferDecoder.noMoreData} raises L{_DataLoss} if it is
537
called and the end of the last trailer has not yet been received.
539
parser = http._ChunkedTransferDecoder(None, lambda bytes: None)
540
parser.dataReceived('0\r\n\r')
541
exc = self.assertRaises(_DataLoss, parser.noMoreData)
544
"Chunked decoder in 'trailer' state, still expecting more data "
545
"to get to finished state.")
548
def test_finishedConnectionLose(self):
550
L{_ChunkedTransferDecoder.noMoreData} does not raise any exception if
551
it is called after the terminal zero length chunk is received.
553
parser = http._ChunkedTransferDecoder(None, lambda bytes: None)
554
parser.dataReceived('0\r\n\r\n')
558
def test_reentrantFinishedNoMoreData(self):
560
L{_ChunkedTransferDecoder.noMoreData} can be called from the finished
561
callback without raising an exception.
569
errors.append(Failure())
571
successes.append(True)
572
parser = http._ChunkedTransferDecoder(None, finished)
573
parser.dataReceived('0\r\n\r\n')
574
self.assertEqual(errors, [])
575
self.assertEqual(successes, [True])
579
class ChunkingTestCase(unittest.TestCase):
581
strings = ["abcv", "", "fdfsd423", "Ffasfas\r\n",
582
"523523\n\rfsdf", "4234"]
584
def testChunks(self):
585
for s in self.strings:
586
self.assertEquals((s, ''), http.fromChunk(''.join(http.toChunk(s))))
587
self.assertRaises(ValueError, http.fromChunk, '-5\r\nmalformed!\r\n')
589
def testConcatenatedChunks(self):
590
chunked = ''.join([''.join(http.toChunk(t)) for t in self.strings])
596
data, buffer = http.fromChunk(buffer)
600
self.assertEquals(result, self.strings)
604
class ParsingTestCase(unittest.TestCase):
606
Tests for protocol parsing in L{HTTPChannel}.
608
def runRequest(self, httpRequest, requestClass, success=1):
609
httpRequest = httpRequest.replace("\n", "\r\n")
610
b = StringTransport()
611
a = http.HTTPChannel()
612
a.requestFactory = requestClass
614
# one byte at a time, to stress it.
615
for byte in httpRequest:
616
if a.transport.disconnecting:
619
a.connectionLost(IOError("all done"))
621
self.assertEquals(self.didRequest, 1)
624
self.assert_(not hasattr(self, "didRequest"))
628
def test_basicAuth(self):
630
L{HTTPChannel} provides username and password information supplied in
631
an I{Authorization} header to the L{Request} which makes it available
632
via its C{getUser} and C{getPassword} methods.
635
class Request(http.Request):
638
testcase.assertEquals(self.getUser(), self.l[0])
639
testcase.assertEquals(self.getPassword(), self.l[1])
640
for u, p in [("foo", "bar"), ("hello", "there:z")]:
641
Request.l[:] = [u, p]
643
f = "GET / HTTP/1.0\nAuthorization: Basic %s\n\n" % (s.encode("base64").strip(), )
644
self.runRequest(f, Request, 0)
647
def test_headers(self):
649
Headers received by L{HTTPChannel} in a request are made available to
653
class MyRequest(http.Request):
655
processed.append(self)
666
self.runRequest('\n'.join(requestLines), MyRequest, 0)
667
[request] = processed
669
request.requestHeaders.getRawHeaders('foo'), ['bar'])
671
request.requestHeaders.getRawHeaders('bAz'), ['Quux', 'quux'])
674
def test_tooManyHeaders(self):
676
L{HTTPChannel} enforces a limit of C{HTTPChannel.maxHeaders} on the
677
number of headers received per request.
680
class MyRequest(http.Request):
682
processed.append(self)
684
requestLines = ["GET / HTTP/1.0"]
685
for i in range(http.HTTPChannel.maxHeaders + 2):
686
requestLines.append("%s: foo" % (i,))
687
requestLines.extend(["", ""])
689
channel = self.runRequest("\n".join(requestLines), MyRequest, 0)
690
self.assertEqual(processed, [])
692
channel.transport.value(),
693
"HTTP/1.1 400 Bad Request\r\n\r\n")
696
def test_headerLimitPerRequest(self):
698
L{HTTPChannel} enforces the limit of C{HTTPChannel.maxHeaders} per
699
request so that headers received in an earlier request do not count
700
towards the limit when processing a later request.
703
class MyRequest(http.Request):
705
processed.append(self)
708
self.patch(http.HTTPChannel, 'maxHeaders', 1)
719
channel = self.runRequest("\n".join(requestLines), MyRequest, 0)
720
[first, second] = processed
721
self.assertEqual(first.getHeader('foo'), 'bar')
722
self.assertEqual(second.getHeader('bar'), 'baz')
724
channel.transport.value(),
725
'HTTP/1.1 200 OK\r\n'
726
'Transfer-Encoding: chunked\r\n'
730
'HTTP/1.1 200 OK\r\n'
731
'Transfer-Encoding: chunked\r\n'
737
def testCookies(self):
739
Test cookies parsing and reading.
743
Cookie: rabbit="eat carrot"; ninja=secret; spam="hey 1=1!"
748
class MyRequest(http.Request):
750
testcase.assertEquals(self.getCookie('rabbit'), '"eat carrot"')
751
testcase.assertEquals(self.getCookie('ninja'), 'secret')
752
testcase.assertEquals(self.getCookie('spam'), '"hey 1=1!"')
753
testcase.didRequest = 1
756
self.runRequest(httpRequest, MyRequest)
760
GET /?key=value&multiple=two+words&multiple=more%20words&empty= HTTP/1.0
764
class MyRequest(http.Request):
766
testcase.assertEquals(self.method, "GET")
767
testcase.assertEquals(self.args["key"], ["value"])
768
testcase.assertEquals(self.args["empty"], [""])
769
testcase.assertEquals(self.args["multiple"], ["two words", "more words"])
770
testcase.didRequest = 1
773
self.runRequest(httpRequest, MyRequest)
776
def test_extraQuestionMark(self):
778
While only a single '?' is allowed in an URL, several other servers
779
allow several and pass all after the first through as part of the
780
query arguments. Test that we emulate this behavior.
782
httpRequest = 'GET /foo?bar=?&baz=quux HTTP/1.0\n\n'
785
class MyRequest(http.Request):
787
testcase.assertEqual(self.method, 'GET')
788
testcase.assertEqual(self.path, '/foo')
789
testcase.assertEqual(self.args['bar'], ['?'])
790
testcase.assertEqual(self.args['baz'], ['quux'])
791
testcase.didRequest = 1
794
self.runRequest(httpRequest, MyRequest)
797
def test_formPOSTRequest(self):
799
The request body of a I{POST} request with a I{Content-Type} header
800
of I{application/x-www-form-urlencoded} is parsed according to that
801
content type and made available in the C{args} attribute of the
802
request object. The original bytes of the request may still be read
803
from the C{content} attribute.
805
query = 'key=value&multiple=two+words&multiple=more%20words&empty='
809
Content-Type: application/x-www-form-urlencoded
811
%s''' % (len(query), query)
814
class MyRequest(http.Request):
816
testcase.assertEquals(self.method, "POST")
817
testcase.assertEquals(self.args["key"], ["value"])
818
testcase.assertEquals(self.args["empty"], [""])
819
testcase.assertEquals(self.args["multiple"], ["two words", "more words"])
821
# Reading from the content file-like must produce the entire
823
testcase.assertEquals(self.content.read(), query)
824
testcase.didRequest = 1
827
self.runRequest(httpRequest, MyRequest)
829
def testMissingContentDisposition(self):
832
Content-Type: multipart/form-data; boundary=AaB03x
836
Content-Type: text/plain
837
Content-Transfer-Encoding: quoted-printable
842
self.runRequest(req, http.Request, success=False)
844
def test_chunkedEncoding(self):
846
If a request uses the I{chunked} transfer encoding, the request body is
847
decoded accordingly before it is made available on the request.
851
Content-Type: text/plain
852
Transfer-Encoding: chunked
862
class MyRequest(http.Request):
864
# The tempfile API used to create content returns an
865
# instance of a different type depending on what platform
866
# we're running on. The point here is to verify that the
867
# request body is in a file that's on the filesystem.
868
# Having a fileno method that returns an int is a somewhat
869
# close approximation of this. -exarkun
870
testcase.assertIsInstance(self.content.fileno(), int)
871
testcase.assertEqual(self.method, 'GET')
872
testcase.assertEqual(self.path, '/')
873
content = self.content.read()
874
testcase.assertEqual(content, 'Hello, spam,eggs spam spam')
875
testcase.assertIdentical(self.channel._transferDecoder, None)
876
testcase.didRequest = 1
879
self.runRequest(httpRequest, MyRequest)
883
class QueryArgumentsTestCase(unittest.TestCase):
884
def testUnquote(self):
886
from twisted.protocols import _c_urlarg
888
raise unittest.SkipTest("_c_urlarg module is not available")
889
# work exactly like urllib.unquote, including stupid things
890
# % followed by a non-hexdigit in the middle and in the end
891
self.failUnlessEqual(urllib.unquote("%notreally%n"),
892
_c_urlarg.unquote("%notreally%n"))
893
# % followed by hexdigit, followed by non-hexdigit
894
self.failUnlessEqual(urllib.unquote("%1quite%1"),
895
_c_urlarg.unquote("%1quite%1"))
896
# unquoted text, followed by some quoted chars, ends in a trailing %
897
self.failUnlessEqual(urllib.unquote("blah%21%40%23blah%"),
898
_c_urlarg.unquote("blah%21%40%23blah%"))
900
self.failUnlessEqual(urllib.unquote(""), _c_urlarg.unquote(""))
902
def testParseqs(self):
903
self.failUnlessEqual(cgi.parse_qs("a=b&d=c;+=f"),
904
http.parse_qs("a=b&d=c;+=f"))
905
self.failUnlessRaises(ValueError, http.parse_qs, "blah",
907
self.failUnlessEqual(cgi.parse_qs("a=&b=c", keep_blank_values = 1),
908
http.parse_qs("a=&b=c", keep_blank_values = 1))
909
self.failUnlessEqual(cgi.parse_qs("a=&b=c"),
910
http.parse_qs("a=&b=c"))
913
def test_urlparse(self):
915
For a given URL, L{http.urlparse} should behave the same as
916
L{urlparse}, except it should always return C{str}, never C{unicode}.
919
for scheme in ('http', 'https'):
920
for host in ('example.com',):
921
for port in (None, 100):
922
for path in ('', 'path'):
924
host = host + ':' + str(port)
925
yield urlunsplit((scheme, host, path, '', ''))
928
def assertSameParsing(url, decode):
930
Verify that C{url} is parsed into the same objects by both
931
L{http.urlparse} and L{urlparse}.
933
urlToStandardImplementation = url
935
urlToStandardImplementation = url.decode('ascii')
936
standardResult = urlparse(urlToStandardImplementation)
937
scheme, netloc, path, params, query, fragment = http.urlparse(url)
939
(scheme, netloc, path, params, query, fragment),
941
self.assertTrue(isinstance(scheme, str))
942
self.assertTrue(isinstance(netloc, str))
943
self.assertTrue(isinstance(path, str))
944
self.assertTrue(isinstance(params, str))
945
self.assertTrue(isinstance(query, str))
946
self.assertTrue(isinstance(fragment, str))
948
# With caching, unicode then str
951
assertSameParsing(url, True)
952
assertSameParsing(url, False)
954
# With caching, str then unicode
957
assertSameParsing(url, False)
958
assertSameParsing(url, True)
963
assertSameParsing(url, True)
965
assertSameParsing(url, False)
968
def test_urlparseRejectsUnicode(self):
970
L{http.urlparse} should reject unicode input early.
972
self.assertRaises(TypeError, http.urlparse, u'http://example.org/path')
975
def testEscchar(self):
977
from twisted.protocols import _c_urlarg
979
raise unittest.SkipTest("_c_urlarg module is not available")
980
self.failUnlessEqual("!@#+b",
981
_c_urlarg.unquote("+21+40+23+b", "+"))
983
class ClientDriver(http.HTTPClient):
984
def handleStatus(self, version, status, message):
985
self.version = version
987
self.message = message
989
class ClientStatusParsing(unittest.TestCase):
990
def testBaseline(self):
992
c.lineReceived('HTTP/1.0 201 foo')
993
self.failUnlessEqual(c.version, 'HTTP/1.0')
994
self.failUnlessEqual(c.status, '201')
995
self.failUnlessEqual(c.message, 'foo')
997
def testNoMessage(self):
999
c.lineReceived('HTTP/1.0 201')
1000
self.failUnlessEqual(c.version, 'HTTP/1.0')
1001
self.failUnlessEqual(c.status, '201')
1002
self.failUnlessEqual(c.message, '')
1004
def testNoMessage_trailingSpace(self):
1006
c.lineReceived('HTTP/1.0 201 ')
1007
self.failUnlessEqual(c.version, 'HTTP/1.0')
1008
self.failUnlessEqual(c.status, '201')
1009
self.failUnlessEqual(c.message, '')
1013
class RequestTests(unittest.TestCase, ResponseTestMixin):
1015
Tests for L{http.Request}
1017
def _compatHeadersTest(self, oldName, newName):
1019
Verify that each of two different attributes which are associated with
1020
the same state properly reflect changes made through the other.
1022
This is used to test that the C{headers}/C{responseHeaders} and
1023
C{received_headers}/C{requestHeaders} pairs interact properly.
1025
req = http.Request(DummyChannel(), None)
1026
getattr(req, newName).setRawHeaders("test", ["lemur"])
1027
self.assertEqual(getattr(req, oldName)["test"], "lemur")
1028
setattr(req, oldName, {"foo": "bar"})
1030
list(getattr(req, newName).getAllRawHeaders()),
1032
setattr(req, newName, http_headers.Headers())
1033
self.assertEqual(getattr(req, oldName), {})
1036
def test_received_headers(self):
1038
L{Request.received_headers} is a backwards compatible API which
1039
accesses and allows mutation of the state at L{Request.requestHeaders}.
1041
self._compatHeadersTest('received_headers', 'requestHeaders')
1044
def test_headers(self):
1046
L{Request.headers} is a backwards compatible API which accesses and
1047
allows mutation of the state at L{Request.responseHeaders}.
1049
self._compatHeadersTest('headers', 'responseHeaders')
1052
def test_getHeader(self):
1054
L{http.Request.getHeader} returns the value of the named request
1057
req = http.Request(DummyChannel(), None)
1058
req.requestHeaders.setRawHeaders("test", ["lemur"])
1059
self.assertEquals(req.getHeader("test"), "lemur")
1062
def test_getHeaderReceivedMultiples(self):
1064
When there are multiple values for a single request header,
1065
L{http.Request.getHeader} returns the last value.
1067
req = http.Request(DummyChannel(), None)
1068
req.requestHeaders.setRawHeaders("test", ["lemur", "panda"])
1069
self.assertEquals(req.getHeader("test"), "panda")
1072
def test_getHeaderNotFound(self):
1074
L{http.Request.getHeader} returns C{None} when asked for the value of a
1075
request header which is not present.
1077
req = http.Request(DummyChannel(), None)
1078
self.assertEquals(req.getHeader("test"), None)
1081
def test_getAllHeaders(self):
1083
L{http.Request.getAllheaders} returns a C{dict} mapping all request
1084
header names to their corresponding values.
1086
req = http.Request(DummyChannel(), None)
1087
req.requestHeaders.setRawHeaders("test", ["lemur"])
1088
self.assertEquals(req.getAllHeaders(), {"test": "lemur"})
1091
def test_getAllHeadersNoHeaders(self):
1093
L{http.Request.getAllHeaders} returns an empty C{dict} if there are no
1096
req = http.Request(DummyChannel(), None)
1097
self.assertEquals(req.getAllHeaders(), {})
1100
def test_getAllHeadersMultipleHeaders(self):
1102
When there are multiple values for a single request header,
1103
L{http.Request.getAllHeaders} returns only the last value.
1105
req = http.Request(DummyChannel(), None)
1106
req.requestHeaders.setRawHeaders("test", ["lemur", "panda"])
1107
self.assertEquals(req.getAllHeaders(), {"test": "panda"})
1110
def test_setResponseCode(self):
1112
L{http.Request.setResponseCode} takes a status code and causes it to be
1113
used as the response status.
1115
channel = DummyChannel()
1116
req = http.Request(channel, None)
1117
req.setResponseCode(201)
1120
channel.transport.written.getvalue().splitlines()[0],
1121
'%s 201 Created' % (req.clientproto,))
1124
def test_setResponseCodeAndMessage(self):
1126
L{http.Request.setResponseCode} takes a status code and a message and
1127
causes them to be used as the response status.
1129
channel = DummyChannel()
1130
req = http.Request(channel, None)
1131
req.setResponseCode(202, "happily accepted")
1134
channel.transport.written.getvalue().splitlines()[0],
1135
'%s 202 happily accepted' % (req.clientproto,))
1138
def test_setResponseCodeAcceptsIntegers(self):
1140
L{http.Request.setResponseCode} accepts C{int} or C{long} for the code
1141
parameter and raises L{TypeError} if passed anything else.
1143
req = http.Request(DummyChannel(), None)
1144
req.setResponseCode(1)
1145
req.setResponseCode(1L)
1146
self.assertRaises(TypeError, req.setResponseCode, "1")
1149
def test_setHost(self):
1151
L{http.Request.setHost} sets the value of the host request header.
1153
req = http.Request(DummyChannel(), None)
1154
req.setHost("example.com", 443)
1156
req.requestHeaders.getRawHeaders("host"), ["example.com"])
1159
def test_setHeader(self):
1161
L{http.Request.setHeader} sets the value of the given response header.
1163
req = http.Request(DummyChannel(), None)
1164
req.setHeader("test", "lemur")
1165
self.assertEquals(req.responseHeaders.getRawHeaders("test"), ["lemur"])
1168
def test_firstWrite(self):
1170
For an HTTP 1.0 request, L{http.Request.write} sends an HTTP 1.0
1171
Response-Line and whatever response headers are set.
1173
req = http.Request(DummyChannel(), None)
1174
trans = StringTransport()
1176
req.transport = trans
1178
req.setResponseCode(200)
1179
req.clientproto = "HTTP/1.0"
1180
req.responseHeaders.setRawHeaders("test", ["lemur"])
1183
self.assertResponseEquals(
1185
[("HTTP/1.0 200 OK",
1190
def test_firstWriteHTTP11Chunked(self):
1192
For an HTTP 1.1 request, L{http.Request.write} sends an HTTP 1.1
1193
Response-Line, whatever response headers are set, and uses chunked
1194
encoding for the response body.
1196
req = http.Request(DummyChannel(), None)
1197
trans = StringTransport()
1199
req.transport = trans
1201
req.setResponseCode(200)
1202
req.clientproto = "HTTP/1.1"
1203
req.responseHeaders.setRawHeaders("test", ["lemur"])
1207
self.assertResponseEquals(
1209
[("HTTP/1.1 200 OK",
1211
"Transfer-Encoding: chunked",
1212
"5\r\nHello\r\n6\r\nWorld!\r\n")])
1215
def test_firstWriteLastModified(self):
1217
For an HTTP 1.0 request for a resource with a known last modified time,
1218
L{http.Request.write} sends an HTTP Response-Line, whatever response
1219
headers are set, and a last-modified header with that time.
1221
req = http.Request(DummyChannel(), None)
1222
trans = StringTransport()
1224
req.transport = trans
1226
req.setResponseCode(200)
1227
req.clientproto = "HTTP/1.0"
1228
req.lastModified = 0
1229
req.responseHeaders.setRawHeaders("test", ["lemur"])
1232
self.assertResponseEquals(
1234
[("HTTP/1.0 200 OK",
1236
"Last-Modified: Thu, 01 Jan 1970 00:00:00 GMT",
1240
def test_parseCookies(self):
1242
L{http.Request.parseCookies} extracts cookies from C{requestHeaders}
1243
and adds them to C{received_cookies}.
1245
req = http.Request(DummyChannel(), None)
1246
req.requestHeaders.setRawHeaders(
1247
"cookie", ['test="lemur"; test2="panda"'])
1249
self.assertEquals(req.received_cookies, {"test": '"lemur"',
1250
"test2": '"panda"'})
1253
def test_parseCookiesMultipleHeaders(self):
1255
L{http.Request.parseCookies} can extract cookies from multiple Cookie
1258
req = http.Request(DummyChannel(), None)
1259
req.requestHeaders.setRawHeaders(
1260
"cookie", ['test="lemur"', 'test2="panda"'])
1262
self.assertEquals(req.received_cookies, {"test": '"lemur"',
1263
"test2": '"panda"'})
1266
def test_connectionLost(self):
1268
L{http.Request.connectionLost} closes L{Request.content} and drops the
1269
reference to the L{HTTPChannel} to assist with garbage collection.
1271
req = http.Request(DummyChannel(), None)
1273
# Cause Request.content to be created at all.
1276
# Grab a reference to content in case the Request drops it later on.
1277
content = req.content
1279
# Put some bytes into it
1280
req.handleContentChunk("hello")
1282
# Then something goes wrong and content should get closed.
1283
req.connectionLost(Failure(ConnectionLost("Finished")))
1284
self.assertTrue(content.closed)
1285
self.assertIdentical(req.channel, None)
1288
def test_registerProducerTwiceFails(self):
1290
Calling L{Request.registerProducer} when a producer is already
1291
registered raises ValueError.
1293
req = http.Request(DummyChannel(), None)
1294
req.registerProducer(DummyProducer(), True)
1296
ValueError, req.registerProducer, DummyProducer(), True)
1299
def test_registerProducerWhenQueuedPausesPushProducer(self):
1301
Calling L{Request.registerProducer} with an IPushProducer when the
1302
request is queued pauses the producer.
1304
req = http.Request(DummyChannel(), True)
1305
producer = DummyProducer()
1306
req.registerProducer(producer, True)
1307
self.assertEquals(['pause'], producer.events)
1310
def test_registerProducerWhenQueuedDoesntPausePullProducer(self):
1312
Calling L{Request.registerProducer} with an IPullProducer when the
1313
request is queued does not pause the producer, because it doesn't make
1314
sense to pause a pull producer.
1316
req = http.Request(DummyChannel(), True)
1317
producer = DummyProducer()
1318
req.registerProducer(producer, False)
1319
self.assertEquals([], producer.events)
1322
def test_registerProducerWhenQueuedDoesntRegisterPushProducer(self):
1324
Calling L{Request.registerProducer} with an IPushProducer when the
1325
request is queued does not register the producer on the request's
1328
self.assertIdentical(
1329
None, getattr(http.StringTransport, 'registerProducer', None),
1330
"StringTransport cannot implement registerProducer for this test "
1332
req = http.Request(DummyChannel(), True)
1333
producer = DummyProducer()
1334
req.registerProducer(producer, True)
1335
# This is a roundabout assertion: http.StringTransport doesn't
1336
# implement registerProducer, so Request.registerProducer can't have
1337
# tried to call registerProducer on the transport.
1338
self.assertIsInstance(req.transport, http.StringTransport)
1341
def test_registerProducerWhenQueuedDoesntRegisterPullProducer(self):
1343
Calling L{Request.registerProducer} with an IPullProducer when the
1344
request is queued does not register the producer on the request's
1347
self.assertIdentical(
1348
None, getattr(http.StringTransport, 'registerProducer', None),
1349
"StringTransport cannot implement registerProducer for this test "
1351
req = http.Request(DummyChannel(), True)
1352
producer = DummyProducer()
1353
req.registerProducer(producer, False)
1354
# This is a roundabout assertion: http.StringTransport doesn't
1355
# implement registerProducer, so Request.registerProducer can't have
1356
# tried to call registerProducer on the transport.
1357
self.assertIsInstance(req.transport, http.StringTransport)
1360
def test_registerProducerWhenNotQueuedRegistersPushProducer(self):
1362
Calling L{Request.registerProducer} with an IPushProducer when the
1363
request is not queued registers the producer as a push producer on the
1364
request's transport.
1366
req = http.Request(DummyChannel(), False)
1367
producer = DummyProducer()
1368
req.registerProducer(producer, True)
1369
self.assertEquals([(producer, True)], req.transport.producers)
1372
def test_registerProducerWhenNotQueuedRegistersPullProducer(self):
1374
Calling L{Request.registerProducer} with an IPullProducer when the
1375
request is not queued registers the producer as a pull producer on the
1376
request's transport.
1378
req = http.Request(DummyChannel(), False)
1379
producer = DummyProducer()
1380
req.registerProducer(producer, False)
1381
self.assertEquals([(producer, False)], req.transport.producers)
1384
def test_connectionLostNotification(self):
1386
L{Request.connectionLost} triggers all finish notification Deferreds
1387
and cleans up per-request state.
1390
request = http.Request(d, True)
1391
finished = request.notifyFinish()
1392
request.connectionLost(Failure(ConnectionLost("Connection done")))
1393
self.assertIdentical(request.channel, None)
1394
return self.assertFailure(finished, ConnectionLost)
1397
def test_finishNotification(self):
1399
L{Request.finish} triggers all finish notification Deferreds.
1401
request = http.Request(DummyChannel(), False)
1402
finished = request.notifyFinish()
1403
# Force the request to have a non-None content attribute. This is
1404
# probably a bug in Request.
1405
request.gotLength(1)
1410
def test_finishAfterConnectionLost(self):
1412
Calling L{Request.finish} after L{Request.connectionLost} has been
1413
called results in a L{RuntimeError} being raised.
1415
channel = DummyChannel()
1416
transport = channel.transport
1417
req = http.Request(channel, False)
1418
req.connectionLost(Failure(ConnectionLost("The end.")))
1419
self.assertRaises(RuntimeError, req.finish)
1423
class MultilineHeadersTestCase(unittest.TestCase):
1425
Tests to exercise handling of multiline headers by L{HTTPClient}. RFCs 1945
1426
(HTTP 1.0) and 2616 (HTTP 1.1) state that HTTP message header fields can
1427
span multiple lines if each extra line is preceded by at least one space or
1432
Initialize variables used to verify that the header-processing functions
1435
self.handleHeaderCalled = False
1436
self.handleEndHeadersCalled = False
1438
# Dictionary of sample complete HTTP header key/value pairs, including
1439
# multiline headers.
1440
expectedHeaders = {'Content-Length': '10',
1441
'X-Multiline' : 'line-0\tline-1',
1442
'X-Multiline2' : 'line-2 line-3'}
1444
def ourHandleHeader(self, key, val):
1446
Dummy implementation of L{HTTPClient.handleHeader}.
1448
self.handleHeaderCalled = True
1449
self.assertEquals(val, self.expectedHeaders[key])
1452
def ourHandleEndHeaders(self):
1454
Dummy implementation of L{HTTPClient.handleEndHeaders}.
1456
self.handleEndHeadersCalled = True
1459
def test_extractHeader(self):
1461
A header isn't processed by L{HTTPClient.extractHeader} until it is
1462
confirmed in L{HTTPClient.lineReceived} that the header has been
1463
received completely.
1466
c.handleHeader = self.ourHandleHeader
1467
c.handleEndHeaders = self.ourHandleEndHeaders
1469
c.lineReceived('HTTP/1.0 201')
1470
c.lineReceived('Content-Length: 10')
1471
self.assertIdentical(c.length, None)
1472
self.assertFalse(self.handleHeaderCalled)
1473
self.assertFalse(self.handleEndHeadersCalled)
1475
# Signal end of headers.
1477
self.assertTrue(self.handleHeaderCalled)
1478
self.assertTrue(self.handleEndHeadersCalled)
1480
self.assertEquals(c.length, 10)
1483
def test_noHeaders(self):
1485
An HTTP request with no headers will not cause any calls to
1486
L{handleHeader} but will cause L{handleEndHeaders} to be called on
1487
L{HTTPClient} subclasses.
1490
c.handleHeader = self.ourHandleHeader
1491
c.handleEndHeaders = self.ourHandleEndHeaders
1492
c.lineReceived('HTTP/1.0 201')
1494
# Signal end of headers.
1496
self.assertFalse(self.handleHeaderCalled)
1497
self.assertTrue(self.handleEndHeadersCalled)
1499
self.assertEquals(c.version, 'HTTP/1.0')
1500
self.assertEquals(c.status, '201')
1503
def test_multilineHeaders(self):
1505
L{HTTPClient} parses multiline headers by buffering header lines until
1506
an empty line or a line that does not start with whitespace hits
1507
lineReceived, confirming that the header has been received completely.
1510
c.handleHeader = self.ourHandleHeader
1511
c.handleEndHeaders = self.ourHandleEndHeaders
1513
c.lineReceived('HTTP/1.0 201')
1514
c.lineReceived('X-Multiline: line-0')
1515
self.assertFalse(self.handleHeaderCalled)
1516
# Start continuing line with a tab.
1517
c.lineReceived('\tline-1')
1518
c.lineReceived('X-Multiline2: line-2')
1519
# The previous header must be complete, so now it can be processed.
1520
self.assertTrue(self.handleHeaderCalled)
1521
# Start continuing line with a space.
1522
c.lineReceived(' line-3')
1523
c.lineReceived('Content-Length: 10')
1525
# Signal end of headers.
1527
self.assertTrue(self.handleEndHeadersCalled)
1529
self.assertEquals(c.version, 'HTTP/1.0')
1530
self.assertEquals(c.status, '201')
1531
self.assertEquals(c.length, 10)
added
removed
vendor/Twisted-10.0.0/twisted/web/test/test_http.py
