blob: 50222f277d9099f58ddb3cabe9f098d799a7495e [file] [log] [blame]
jcgregorio2d66d4f2006-02-07 05:34:14 +00001#!/usr/bin/env python2.4
2"""
3httplib2test
4
5A set of unit tests for httplib2.py.
6
7Requires Python 2.4 or later
8"""
9
10__author__ = "Joe Gregorio (joe@bitworking.org)"
11__copyright__ = "Copyright 2006, Joe Gregorio"
12__contributors__ = []
13__license__ = "MIT"
14__history__ = """ """
15__version__ = "0.1 ($Rev: 118 $)"
16
17
Joe Gregoriob6c90c42011-02-11 01:03:22 -050018import StringIO
19import base64
jcgregoriode8238d2007-03-07 19:08:26 +000020import httplib
21import httplib2
22import os
Joe Gregoriob6c90c42011-02-11 01:03:22 -050023import socket
24import sys
jcgregoriode8238d2007-03-07 19:08:26 +000025import time
Joe Gregoriob6c90c42011-02-11 01:03:22 -050026import unittest
27import urlparse
jcgregorio8421f272006-02-14 18:19:51 +000028
29# Python 2.3 support
30if not hasattr(unittest.TestCase, 'assertTrue'):
31 unittest.TestCase.assertTrue = unittest.TestCase.failUnless
32 unittest.TestCase.assertFalse = unittest.TestCase.failIf
33
jcgregorio2d66d4f2006-02-07 05:34:14 +000034# The test resources base uri
35base = 'http://bitworking.org/projects/httplib2/test/'
36#base = 'http://localhost/projects/httplib2/test/'
jcgregorio90fb4a42006-11-17 16:19:47 +000037cacheDirName = ".cache"
jcgregorio2d66d4f2006-02-07 05:34:14 +000038
jcgregoriode8238d2007-03-07 19:08:26 +000039
40class CredentialsTest(unittest.TestCase):
41 def test(self):
42 c = httplib2.Credentials()
43 c.add("joe", "password")
44 self.assertEqual(("joe", "password"), list(c.iter("bitworking.org"))[0])
45 self.assertEqual(("joe", "password"), list(c.iter(""))[0])
46 c.add("fred", "password2", "wellformedweb.org")
47 self.assertEqual(("joe", "password"), list(c.iter("bitworking.org"))[0])
48 self.assertEqual(1, len(list(c.iter("bitworking.org"))))
49 self.assertEqual(2, len(list(c.iter("wellformedweb.org"))))
50 self.assertTrue(("fred", "password2") in list(c.iter("wellformedweb.org")))
51 c.clear()
52 self.assertEqual(0, len(list(c.iter("bitworking.org"))))
53 c.add("fred", "password2", "wellformedweb.org")
54 self.assertTrue(("fred", "password2") in list(c.iter("wellformedweb.org")))
55 self.assertEqual(0, len(list(c.iter("bitworking.org"))))
56 self.assertEqual(0, len(list(c.iter(""))))
57
58
jcgregorio2d66d4f2006-02-07 05:34:14 +000059class ParserTest(unittest.TestCase):
60 def testFromStd66(self):
61 self.assertEqual( ('http', 'example.com', '', None, None ), httplib2.parse_uri("http://example.com"))
62 self.assertEqual( ('https', 'example.com', '', None, None ), httplib2.parse_uri("https://example.com"))
63 self.assertEqual( ('https', 'example.com:8080', '', None, None ), httplib2.parse_uri("https://example.com:8080"))
64 self.assertEqual( ('http', 'example.com', '/', None, None ), httplib2.parse_uri("http://example.com/"))
65 self.assertEqual( ('http', 'example.com', '/path', None, None ), httplib2.parse_uri("http://example.com/path"))
66 self.assertEqual( ('http', 'example.com', '/path', 'a=1&b=2', None ), httplib2.parse_uri("http://example.com/path?a=1&b=2"))
67 self.assertEqual( ('http', 'example.com', '/path', 'a=1&b=2', 'fred' ), httplib2.parse_uri("http://example.com/path?a=1&b=2#fred"))
68 self.assertEqual( ('http', 'example.com', '/path', 'a=1&b=2', 'fred' ), httplib2.parse_uri("http://example.com/path?a=1&b=2#fred"))
69
jcgregorio2d66d4f2006-02-07 05:34:14 +000070
jcgregorioa46fe4e2006-11-16 04:13:45 +000071class UrlNormTest(unittest.TestCase):
72 def test(self):
73 self.assertEqual( "http://example.org/", httplib2.urlnorm("http://example.org")[-1])
74 self.assertEqual( "http://example.org/", httplib2.urlnorm("http://EXAMple.org")[-1])
75 self.assertEqual( "http://example.org/?=b", httplib2.urlnorm("http://EXAMple.org?=b")[-1])
76 self.assertEqual( "http://example.org/mypath?a=b", httplib2.urlnorm("http://EXAMple.org/mypath?a=b")[-1])
77 self.assertEqual( "http://localhost:80/", httplib2.urlnorm("http://localhost:80")[-1])
jcgregoriob4e9ab02006-11-17 15:53:15 +000078 self.assertEqual( httplib2.urlnorm("http://localhost:80/"), httplib2.urlnorm("HTTP://LOCALHOST:80"))
jcgregorio132d28e2007-01-23 16:22:53 +000079 try:
80 httplib2.urlnorm("/")
81 self.fail("Non-absolute URIs should raise an exception")
82 except httplib2.RelativeURIError:
83 pass
jcgregorioa46fe4e2006-11-16 04:13:45 +000084
85class UrlSafenameTest(unittest.TestCase):
86 def test(self):
87 # Test that different URIs end up generating different safe names
88 self.assertEqual( "example.org,fred,a=b,58489f63a7a83c3b7794a6a398ee8b1f", httplib2.safename("http://example.org/fred/?a=b"))
89 self.assertEqual( "example.org,fred,a=b,8c5946d56fec453071f43329ff0be46b", httplib2.safename("http://example.org/fred?/a=b"))
90 self.assertEqual( "www.example.org,fred,a=b,499c44b8d844a011b67ea2c015116968", httplib2.safename("http://www.example.org/fred?/a=b"))
91 self.assertEqual( httplib2.safename(httplib2.urlnorm("http://www")[-1]), httplib2.safename(httplib2.urlnorm("http://WWW")[-1]))
92 self.assertEqual( "www.example.org,fred,a=b,692e843a333484ce0095b070497ab45d", httplib2.safename("https://www.example.org/fred?/a=b"))
93 self.assertNotEqual( httplib2.safename("http://www"), httplib2.safename("https://www"))
94 # Test the max length limits
95 uri = "http://" + ("w" * 200) + ".org"
96 uri2 = "http://" + ("w" * 201) + ".org"
97 self.assertNotEqual( httplib2.safename(uri2), httplib2.safename(uri))
98 # Max length should be 200 + 1 (",") + 32
99 self.assertEqual(233, len(httplib2.safename(uri2)))
100 self.assertEqual(233, len(httplib2.safename(uri)))
101 # Unicode
jcgregoriodebceec2006-12-12 20:26:02 +0000102 if sys.version_info >= (2,3):
103 self.assertEqual( "xn--http,-4y1d.org,fred,a=b,579924c35db315e5a32e3d9963388193", httplib2.safename(u"http://\u2304.org/fred/?a=b"))
jcgregorioa46fe4e2006-11-16 04:13:45 +0000104
jcgregorio14644372007-07-30 14:13:37 +0000105class _MyResponse(StringIO.StringIO):
106 def __init__(self, body, **kwargs):
107 StringIO.StringIO.__init__(self, body)
108 self.headers = kwargs
109
110 def iteritems(self):
111 return self.headers.iteritems()
112
113
114class _MyHTTPConnection(object):
115 "This class is just a mock of httplib.HTTPConnection used for testing"
116
117 def __init__(self, host, port=None, key_file=None, cert_file=None,
joe.gregoriof28536d2007-10-23 14:10:11 +0000118 strict=None, timeout=None, proxy_info=None):
jcgregorio14644372007-07-30 14:13:37 +0000119 self.host = host
120 self.port = port
121 self.timeout = timeout
122 self.log = ""
123
124 def set_debuglevel(self, level):
125 pass
126
127 def connect(self):
128 "Connect to a host on a given port."
129 pass
130
131 def close(self):
132 pass
133
134 def request(self, method, request_uri, body, headers):
135 pass
136
137 def getresponse(self):
138 return _MyResponse("the body", status="200")
jcgregorioa46fe4e2006-11-16 04:13:45 +0000139
jcgregorio90fb4a42006-11-17 16:19:47 +0000140
jcgregorio2d66d4f2006-02-07 05:34:14 +0000141class HttpTest(unittest.TestCase):
142 def setUp(self):
jcgregorio7e3608f2006-06-15 13:01:53 +0000143 if os.path.exists(cacheDirName):
144 [os.remove(os.path.join(cacheDirName, file)) for file in os.listdir(cacheDirName)]
145 self.http = httplib2.Http(cacheDirName)
jcgregorio36140b52006-06-13 02:17:52 +0000146 self.http.clear_credentials()
jcgregorio2d66d4f2006-02-07 05:34:14 +0000147
jcgregorio14644372007-07-30 14:13:37 +0000148 def testConnectionType(self):
joe.gregoriof28536d2007-10-23 14:10:11 +0000149 self.http.force_exception_to_status_code = False
jcgregorio14644372007-07-30 14:13:37 +0000150 response, content = self.http.request("http://bitworking.org", connection_type=_MyHTTPConnection)
151 self.assertEqual(response['content-location'], "http://bitworking.org")
152 self.assertEqual(content, "the body")
153
jcgregorio6a638172007-01-23 16:40:23 +0000154 def testGetUnknownServer(self):
jcgregorio07a9a4a2007-03-08 21:18:39 +0000155 self.http.force_exception_to_status_code = False
jcgregorio6a638172007-01-23 16:40:23 +0000156 try:
157 self.http.request("http://fred.bitworking.org/")
158 self.fail("An httplib2.ServerNotFoundError Exception must be thrown on an unresolvable server.")
159 except httplib2.ServerNotFoundError:
160 pass
161
jcgregorio07a9a4a2007-03-08 21:18:39 +0000162 # Now test with exceptions turned off
163 self.http.force_exception_to_status_code = True
164
165 (response, content) = self.http.request("http://fred.bitworking.org/")
166 self.assertEqual(response['content-type'], 'text/plain')
167 self.assertTrue(content.startswith("Unable to find"))
168 self.assertEqual(response.status, 400)
169
Joe Gregoriob6c90c42011-02-11 01:03:22 -0500170 def testGetConnectionRefused(self):
171 self.http.force_exception_to_status_code = False
172 try:
173 self.http.request("http://localhost:7777/")
174 self.fail("An socket.error exception must be thrown on Connection Refused.")
175 except socket.error:
176 pass
177
178 # Now test with exceptions turned off
179 self.http.force_exception_to_status_code = True
180
181 (response, content) = self.http.request("http://localhost:7777/")
182 self.assertEqual(response['content-type'], 'text/plain')
183 self.assertTrue("Connection refused" in content)
184 self.assertEqual(response.status, 400)
185
jcgregorioa898f8f2006-12-12 17:16:55 +0000186 def testGetIRI(self):
jcgregoriodebceec2006-12-12 20:26:02 +0000187 if sys.version_info >= (2,3):
188 uri = urlparse.urljoin(base, u"reflector/reflector.cgi?d=\N{CYRILLIC CAPITAL LETTER DJE}")
189 (response, content) = self.http.request(uri, "GET")
190 d = self.reflector(content)
191 self.assertTrue(d.has_key('QUERY_STRING'))
192 self.assertTrue(d['QUERY_STRING'].find('%D0%82') > 0)
jcgregorioa898f8f2006-12-12 17:16:55 +0000193
jcgregorio2d66d4f2006-02-07 05:34:14 +0000194 def testGetIsDefaultMethod(self):
195 # Test that GET is the default method
196 uri = urlparse.urljoin(base, "methods/method_reflector.cgi")
jcgregorio36140b52006-06-13 02:17:52 +0000197 (response, content) = self.http.request(uri)
jcgregorio2d66d4f2006-02-07 05:34:14 +0000198 self.assertEqual(response['x-method'], "GET")
199
200 def testDifferentMethods(self):
201 # Test that all methods can be used
202 uri = urlparse.urljoin(base, "methods/method_reflector.cgi")
203 for method in ["GET", "PUT", "DELETE", "POST"]:
jcgregorio36140b52006-06-13 02:17:52 +0000204 (response, content) = self.http.request(uri, method, body=" ")
jcgregorio2d66d4f2006-02-07 05:34:14 +0000205 self.assertEqual(response['x-method'], method)
206
Joe Gregoriob628c0b2009-07-16 12:28:04 -0400207 def testHeadRead(self):
208 # Test that we don't try to read the response of a HEAD request
209 # since httplib blocks response.read() for HEAD requests.
210 # Oddly enough this doesn't appear as a problem when doing HEAD requests
211 # against Apache servers.
212 uri = "http://www.google.com/"
213 (response, content) = self.http.request(uri, "HEAD")
214 self.assertEqual(response.status, 200)
215 self.assertEqual(content, "")
216
jcgregorio2d66d4f2006-02-07 05:34:14 +0000217 def testGetNoCache(self):
218 # Test that can do a GET w/o the cache turned on.
219 http = httplib2.Http()
220 uri = urlparse.urljoin(base, "304/test_etag.txt")
221 (response, content) = http.request(uri, "GET")
222 self.assertEqual(response.status, 200)
jcgregorioa0713ab2006-07-01 05:21:34 +0000223 self.assertEqual(response.previous, None)
jcgregorio2d66d4f2006-02-07 05:34:14 +0000224
Joe Gregorioe202d212009-07-16 14:57:52 -0400225 def testGetOnlyIfCachedCacheHit(self):
226 # Test that can do a GET with cache and 'only-if-cached'
227 uri = urlparse.urljoin(base, "304/test_etag.txt")
228 (response, content) = self.http.request(uri, "GET")
229 (response, content) = self.http.request(uri, "GET", headers={'cache-control': 'only-if-cached'})
230 self.assertEqual(response.fromcache, True)
231 self.assertEqual(response.status, 200)
232
jcgregorioe4ce13e2006-04-02 03:05:08 +0000233 def testGetOnlyIfCachedCacheMiss(self):
234 # Test that can do a GET with no cache with 'only-if-cached'
jcgregorioe4ce13e2006-04-02 03:05:08 +0000235 uri = urlparse.urljoin(base, "304/test_etag.txt")
Joe Gregorioe202d212009-07-16 14:57:52 -0400236 (response, content) = self.http.request(uri, "GET", headers={'cache-control': 'only-if-cached'})
jcgregorioe4ce13e2006-04-02 03:05:08 +0000237 self.assertEqual(response.fromcache, False)
Joe Gregorioe202d212009-07-16 14:57:52 -0400238 self.assertEqual(response.status, 504)
jcgregorioe4ce13e2006-04-02 03:05:08 +0000239
240 def testGetOnlyIfCachedNoCacheAtAll(self):
241 # Test that can do a GET with no cache with 'only-if-cached'
242 # Of course, there might be an intermediary beyond us
243 # that responds to the 'only-if-cached', so this
244 # test can't really be guaranteed to pass.
245 http = httplib2.Http()
246 uri = urlparse.urljoin(base, "304/test_etag.txt")
247 (response, content) = http.request(uri, "GET", headers={'cache-control': 'only-if-cached'})
248 self.assertEqual(response.fromcache, False)
Joe Gregorioe202d212009-07-16 14:57:52 -0400249 self.assertEqual(response.status, 504)
jcgregorioe4ce13e2006-04-02 03:05:08 +0000250
jcgregorio2d66d4f2006-02-07 05:34:14 +0000251 def testUserAgent(self):
252 # Test that we provide a default user-agent
253 uri = urlparse.urljoin(base, "user-agent/test.cgi")
jcgregorio36140b52006-06-13 02:17:52 +0000254 (response, content) = self.http.request(uri, "GET")
jcgregorio2d66d4f2006-02-07 05:34:14 +0000255 self.assertEqual(response.status, 200)
256 self.assertTrue(content.startswith("Python-httplib2/"))
257
258 def testUserAgentNonDefault(self):
259 # Test that the default user-agent can be over-ridden
joe.gregoriof28536d2007-10-23 14:10:11 +0000260
jcgregorio2d66d4f2006-02-07 05:34:14 +0000261 uri = urlparse.urljoin(base, "user-agent/test.cgi")
jcgregorio36140b52006-06-13 02:17:52 +0000262 (response, content) = self.http.request(uri, "GET", headers={'User-Agent': 'fred/1.0'})
jcgregorio2d66d4f2006-02-07 05:34:14 +0000263 self.assertEqual(response.status, 200)
264 self.assertTrue(content.startswith("fred/1.0"))
265
266 def testGet300WithLocation(self):
267 # Test the we automatically follow 300 redirects if a Location: header is provided
268 uri = urlparse.urljoin(base, "300/with-location-header.asis")
jcgregorio36140b52006-06-13 02:17:52 +0000269 (response, content) = self.http.request(uri, "GET")
jcgregorio2d66d4f2006-02-07 05:34:14 +0000270 self.assertEqual(response.status, 200)
271 self.assertEqual(content, "This is the final destination.\n")
jcgregorioa0713ab2006-07-01 05:21:34 +0000272 self.assertEqual(response.previous.status, 300)
273 self.assertEqual(response.previous.fromcache, False)
jcgregorio2d66d4f2006-02-07 05:34:14 +0000274
275 # Confirm that the intermediate 300 is not cached
jcgregorio36140b52006-06-13 02:17:52 +0000276 (response, content) = self.http.request(uri, "GET")
jcgregorio2d66d4f2006-02-07 05:34:14 +0000277 self.assertEqual(response.status, 200)
278 self.assertEqual(content, "This is the final destination.\n")
jcgregorioa0713ab2006-07-01 05:21:34 +0000279 self.assertEqual(response.previous.status, 300)
280 self.assertEqual(response.previous.fromcache, False)
jcgregorio2d66d4f2006-02-07 05:34:14 +0000281
jcgregorio2f1e1422007-05-03 13:17:33 +0000282 def testGet300WithLocationNoRedirect(self):
283 # Test the we automatically follow 300 redirects if a Location: header is provided
284 self.http.follow_redirects = False
285 uri = urlparse.urljoin(base, "300/with-location-header.asis")
286 (response, content) = self.http.request(uri, "GET")
287 self.assertEqual(response.status, 300)
288
jcgregorio2d66d4f2006-02-07 05:34:14 +0000289 def testGet300WithoutLocation(self):
290 # Not giving a Location: header in a 300 response is acceptable
291 # In which case we just return the 300 response
292 uri = urlparse.urljoin(base, "300/without-location-header.asis")
jcgregorio36140b52006-06-13 02:17:52 +0000293 (response, content) = self.http.request(uri, "GET")
jcgregorio2d66d4f2006-02-07 05:34:14 +0000294 self.assertEqual(response.status, 300)
295 self.assertTrue(response['content-type'].startswith("text/html"))
jcgregorioa0713ab2006-07-01 05:21:34 +0000296 self.assertEqual(response.previous, None)
jcgregorio2d66d4f2006-02-07 05:34:14 +0000297
298 def testGet301(self):
299 # Test that we automatically follow 301 redirects
300 # and that we cache the 301 response
301 uri = urlparse.urljoin(base, "301/onestep.asis")
jcgregorio8e300b92006-11-07 16:44:35 +0000302 destination = urlparse.urljoin(base, "302/final-destination.txt")
jcgregorio36140b52006-06-13 02:17:52 +0000303 (response, content) = self.http.request(uri, "GET")
jcgregorio2d66d4f2006-02-07 05:34:14 +0000304 self.assertEqual(response.status, 200)
jcgregorio772adc82006-11-17 21:52:34 +0000305 self.assertTrue(response.has_key('content-location'))
306 self.assertEqual(response['content-location'], destination)
jcgregorio2d66d4f2006-02-07 05:34:14 +0000307 self.assertEqual(content, "This is the final destination.\n")
jcgregorioa0713ab2006-07-01 05:21:34 +0000308 self.assertEqual(response.previous.status, 301)
309 self.assertEqual(response.previous.fromcache, False)
jcgregorio2d66d4f2006-02-07 05:34:14 +0000310
jcgregorio36140b52006-06-13 02:17:52 +0000311 (response, content) = self.http.request(uri, "GET")
jcgregorio2d66d4f2006-02-07 05:34:14 +0000312 self.assertEqual(response.status, 200)
jcgregorio772adc82006-11-17 21:52:34 +0000313 self.assertEqual(response['content-location'], destination)
jcgregorio2d66d4f2006-02-07 05:34:14 +0000314 self.assertEqual(content, "This is the final destination.\n")
jcgregorioa0713ab2006-07-01 05:21:34 +0000315 self.assertEqual(response.previous.status, 301)
316 self.assertEqual(response.previous.fromcache, True)
jcgregorio2d66d4f2006-02-07 05:34:14 +0000317
jcgregorio2f1e1422007-05-03 13:17:33 +0000318
319 def testGet301NoRedirect(self):
320 # Test that we automatically follow 301 redirects
321 # and that we cache the 301 response
322 self.http.follow_redirects = False
323 uri = urlparse.urljoin(base, "301/onestep.asis")
324 destination = urlparse.urljoin(base, "302/final-destination.txt")
325 (response, content) = self.http.request(uri, "GET")
326 self.assertEqual(response.status, 301)
327
328
jcgregorio2d66d4f2006-02-07 05:34:14 +0000329 def testGet302(self):
330 # Test that we automatically follow 302 redirects
331 # and that we DO NOT cache the 302 response
332 uri = urlparse.urljoin(base, "302/onestep.asis")
jcgregorio8e300b92006-11-07 16:44:35 +0000333 destination = urlparse.urljoin(base, "302/final-destination.txt")
jcgregorio36140b52006-06-13 02:17:52 +0000334 (response, content) = self.http.request(uri, "GET")
jcgregorio2d66d4f2006-02-07 05:34:14 +0000335 self.assertEqual(response.status, 200)
jcgregorio772adc82006-11-17 21:52:34 +0000336 self.assertEqual(response['content-location'], destination)
jcgregorio2d66d4f2006-02-07 05:34:14 +0000337 self.assertEqual(content, "This is the final destination.\n")
jcgregorioa0713ab2006-07-01 05:21:34 +0000338 self.assertEqual(response.previous.status, 302)
339 self.assertEqual(response.previous.fromcache, False)
jcgregorio2d66d4f2006-02-07 05:34:14 +0000340
341 uri = urlparse.urljoin(base, "302/onestep.asis")
jcgregorio36140b52006-06-13 02:17:52 +0000342 (response, content) = self.http.request(uri, "GET")
jcgregorio2d66d4f2006-02-07 05:34:14 +0000343 self.assertEqual(response.status, 200)
344 self.assertEqual(response.fromcache, True)
jcgregorio772adc82006-11-17 21:52:34 +0000345 self.assertEqual(response['content-location'], destination)
jcgregorio2d66d4f2006-02-07 05:34:14 +0000346 self.assertEqual(content, "This is the final destination.\n")
jcgregorioa0713ab2006-07-01 05:21:34 +0000347 self.assertEqual(response.previous.status, 302)
348 self.assertEqual(response.previous.fromcache, False)
jcgregorio772adc82006-11-17 21:52:34 +0000349 self.assertEqual(response.previous['content-location'], uri)
jcgregorio2d66d4f2006-02-07 05:34:14 +0000350
351 uri = urlparse.urljoin(base, "302/twostep.asis")
352
jcgregorio36140b52006-06-13 02:17:52 +0000353 (response, content) = self.http.request(uri, "GET")
jcgregorio2d66d4f2006-02-07 05:34:14 +0000354 self.assertEqual(response.status, 200)
355 self.assertEqual(response.fromcache, True)
356 self.assertEqual(content, "This is the final destination.\n")
jcgregorioa0713ab2006-07-01 05:21:34 +0000357 self.assertEqual(response.previous.status, 302)
358 self.assertEqual(response.previous.fromcache, False)
jcgregorio2d66d4f2006-02-07 05:34:14 +0000359
360 def testGet302RedirectionLimit(self):
361 # Test that we can set a lower redirection limit
362 # and that we raise an exception when we exceed
363 # that limit.
jcgregorio07a9a4a2007-03-08 21:18:39 +0000364 self.http.force_exception_to_status_code = False
365
jcgregorio2d66d4f2006-02-07 05:34:14 +0000366 uri = urlparse.urljoin(base, "302/twostep.asis")
367 try:
jcgregorio36140b52006-06-13 02:17:52 +0000368 (response, content) = self.http.request(uri, "GET", redirections = 1)
jcgregorio2d66d4f2006-02-07 05:34:14 +0000369 self.fail("This should not happen")
370 except httplib2.RedirectLimit:
371 pass
372 except Exception, e:
373 self.fail("Threw wrong kind of exception ")
374
jcgregorio07a9a4a2007-03-08 21:18:39 +0000375 # Re-run the test with out the exceptions
376 self.http.force_exception_to_status_code = True
377
378 (response, content) = self.http.request(uri, "GET", redirections = 1)
379 self.assertEqual(response.status, 500)
380 self.assertTrue(response.reason.startswith("Redirected more"))
381 self.assertEqual("302", response['status'])
382 self.assertTrue(content.startswith("<html>"))
383 self.assertTrue(response.previous != None)
384
jcgregorio2d66d4f2006-02-07 05:34:14 +0000385 def testGet302NoLocation(self):
386 # Test that we throw an exception when we get
387 # a 302 with no Location: header.
jcgregorio07a9a4a2007-03-08 21:18:39 +0000388 self.http.force_exception_to_status_code = False
jcgregorio2d66d4f2006-02-07 05:34:14 +0000389 uri = urlparse.urljoin(base, "302/no-location.asis")
390 try:
jcgregorio36140b52006-06-13 02:17:52 +0000391 (response, content) = self.http.request(uri, "GET")
jcgregorio2d66d4f2006-02-07 05:34:14 +0000392 self.fail("Should never reach here")
393 except httplib2.RedirectMissingLocation:
394 pass
395 except Exception, e:
396 self.fail("Threw wrong kind of exception ")
397
jcgregorio07a9a4a2007-03-08 21:18:39 +0000398 # Re-run the test with out the exceptions
399 self.http.force_exception_to_status_code = True
400
401 (response, content) = self.http.request(uri, "GET")
402 self.assertEqual(response.status, 500)
403 self.assertTrue(response.reason.startswith("Redirected but"))
404 self.assertEqual("302", response['status'])
405 self.assertTrue(content.startswith("This is content"))
406
jcgregorio2d66d4f2006-02-07 05:34:14 +0000407 def testGet302ViaHttps(self):
jcgregorioadbb4f82006-05-19 15:17:42 +0000408 # Google always redirects to http://google.com
jcgregorio36140b52006-06-13 02:17:52 +0000409 (response, content) = self.http.request("https://google.com", "GET")
jcgregorio2d66d4f2006-02-07 05:34:14 +0000410 self.assertEqual(200, response.status)
jcgregorioa0713ab2006-07-01 05:21:34 +0000411 self.assertEqual(302, response.previous.status)
jcgregorio2d66d4f2006-02-07 05:34:14 +0000412
413 def testGetViaHttps(self):
414 # Test that we can handle HTTPS
jcgregorio36140b52006-06-13 02:17:52 +0000415 (response, content) = self.http.request("https://google.com/adsense/", "GET")
jcgregorio2d66d4f2006-02-07 05:34:14 +0000416 self.assertEqual(200, response.status)
jcgregorio2d66d4f2006-02-07 05:34:14 +0000417
418 def testGetViaHttpsSpecViolationOnLocation(self):
419 # Test that we follow redirects through HTTPS
420 # even if they violate the spec by including
421 # a relative Location: header instead of an
422 # absolute one.
jcgregorio36140b52006-06-13 02:17:52 +0000423 (response, content) = self.http.request("https://google.com/adsense", "GET")
jcgregorio2d66d4f2006-02-07 05:34:14 +0000424 self.assertEqual(200, response.status)
jcgregorioa0713ab2006-07-01 05:21:34 +0000425 self.assertNotEqual(None, response.previous)
jcgregorio2d66d4f2006-02-07 05:34:14 +0000426
jcgregoriode8238d2007-03-07 19:08:26 +0000427
428 def testGetViaHttpsKeyCert(self):
jcgregorio2f1e1422007-05-03 13:17:33 +0000429 # At this point I can only test
430 # that the key and cert files are passed in
431 # correctly to httplib. It would be nice to have
432 # a real https endpoint to test against.
433 http = httplib2.Http(timeout=2)
jcgregoriode8238d2007-03-07 19:08:26 +0000434
435 http.add_certificate("akeyfile", "acertfile", "bitworking.org")
436 try:
437 (response, content) = http.request("https://bitworking.org", "GET")
438 except:
439 pass
440 self.assertEqual(http.connections["https:bitworking.org"].key_file, "akeyfile")
441 self.assertEqual(http.connections["https:bitworking.org"].cert_file, "acertfile")
442
jcgregorio2f1e1422007-05-03 13:17:33 +0000443 try:
444 (response, content) = http.request("https://notthere.bitworking.org", "GET")
445 except:
446 pass
447 self.assertEqual(http.connections["https:notthere.bitworking.org"].key_file, None)
448 self.assertEqual(http.connections["https:notthere.bitworking.org"].cert_file, None)
449
450
451
jcgregoriode8238d2007-03-07 19:08:26 +0000452
jcgregorio2d66d4f2006-02-07 05:34:14 +0000453 def testGet303(self):
454 # Do a follow-up GET on a Location: header
455 # returned from a POST that gave a 303.
456 uri = urlparse.urljoin(base, "303/303.cgi")
jcgregorio36140b52006-06-13 02:17:52 +0000457 (response, content) = self.http.request(uri, "POST", " ")
jcgregorio2d66d4f2006-02-07 05:34:14 +0000458 self.assertEqual(response.status, 200)
459 self.assertEqual(content, "This is the final destination.\n")
jcgregorioa0713ab2006-07-01 05:21:34 +0000460 self.assertEqual(response.previous.status, 303)
jcgregorio2d66d4f2006-02-07 05:34:14 +0000461
jcgregorio2f1e1422007-05-03 13:17:33 +0000462 def testGet303NoRedirect(self):
463 # Do a follow-up GET on a Location: header
464 # returned from a POST that gave a 303.
465 self.http.follow_redirects = False
466 uri = urlparse.urljoin(base, "303/303.cgi")
467 (response, content) = self.http.request(uri, "POST", " ")
468 self.assertEqual(response.status, 303)
469
jcgregorio2d66d4f2006-02-07 05:34:14 +0000470 def test303ForDifferentMethods(self):
471 # Test that all methods can be used
472 uri = urlparse.urljoin(base, "303/redirect-to-reflector.cgi")
jcgregorio2d66d4f2006-02-07 05:34:14 +0000473 for (method, method_on_303) in [("PUT", "GET"), ("DELETE", "GET"), ("POST", "GET"), ("GET", "GET"), ("HEAD", "GET")]:
jcgregorio36140b52006-06-13 02:17:52 +0000474 (response, content) = self.http.request(uri, method, body=" ")
jcgregorio2d66d4f2006-02-07 05:34:14 +0000475 self.assertEqual(response['x-method'], method_on_303)
476
477 def testGet304(self):
478 # Test that we use ETags properly to validate our cache
479 uri = urlparse.urljoin(base, "304/test_etag.txt")
jcgregorio36140b52006-06-13 02:17:52 +0000480 (response, content) = self.http.request(uri, "GET")
jcgregorio2d66d4f2006-02-07 05:34:14 +0000481 self.assertNotEqual(response['etag'], "")
482
jcgregorio36140b52006-06-13 02:17:52 +0000483 (response, content) = self.http.request(uri, "GET")
484 (response, content) = self.http.request(uri, "GET", headers = {'cache-control': 'must-revalidate'})
jcgregorio2d66d4f2006-02-07 05:34:14 +0000485 self.assertEqual(response.status, 200)
486 self.assertEqual(response.fromcache, True)
487
jcgregorio90fb4a42006-11-17 16:19:47 +0000488 cache_file_name = os.path.join(cacheDirName, httplib2.safename(httplib2.urlnorm(uri)[-1]))
489 f = open(cache_file_name, "r")
490 status_line = f.readline()
491 f.close()
492
493 self.assertTrue(status_line.startswith("status:"))
494
jcgregorio36140b52006-06-13 02:17:52 +0000495 (response, content) = self.http.request(uri, "HEAD")
jcgregorio2d66d4f2006-02-07 05:34:14 +0000496 self.assertEqual(response.status, 200)
497 self.assertEqual(response.fromcache, True)
498
jcgregorio36140b52006-06-13 02:17:52 +0000499 (response, content) = self.http.request(uri, "GET", headers = {'range': 'bytes=0-0'})
jcgregorio2d66d4f2006-02-07 05:34:14 +0000500 self.assertEqual(response.status, 206)
501 self.assertEqual(response.fromcache, False)
502
jcgregorio25185622006-10-28 05:12:34 +0000503 def testGetIgnoreEtag(self):
504 # Test that we can forcibly ignore ETags
505 uri = urlparse.urljoin(base, "reflector/reflector.cgi")
506 (response, content) = self.http.request(uri, "GET")
507 self.assertNotEqual(response['etag'], "")
508
509 (response, content) = self.http.request(uri, "GET", headers = {'cache-control': 'max-age=0'})
510 d = self.reflector(content)
511 self.assertTrue(d.has_key('HTTP_IF_NONE_MATCH'))
512
513 self.http.ignore_etag = True
514 (response, content) = self.http.request(uri, "GET", headers = {'cache-control': 'max-age=0'})
515 d = self.reflector(content)
516 self.assertEqual(response.fromcache, False)
517 self.assertFalse(d.has_key('HTTP_IF_NONE_MATCH'))
518
jcgregorio4b145e82007-01-18 19:46:34 +0000519 def testOverrideEtag(self):
520 # Test that we can forcibly ignore ETags
521 uri = urlparse.urljoin(base, "reflector/reflector.cgi")
522 (response, content) = self.http.request(uri, "GET")
523 self.assertNotEqual(response['etag'], "")
524
525 (response, content) = self.http.request(uri, "GET", headers = {'cache-control': 'max-age=0'})
526 d = self.reflector(content)
527 self.assertTrue(d.has_key('HTTP_IF_NONE_MATCH'))
528 self.assertNotEqual(d['HTTP_IF_NONE_MATCH'], "fred")
529
pilgrim00a352e2009-05-29 04:04:44 +0000530 (response, content) = self.http.request(uri, "GET", headers = {'cache-control': 'max-age=0', 'if-none-match': 'fred'})
jcgregorio4b145e82007-01-18 19:46:34 +0000531 d = self.reflector(content)
532 self.assertTrue(d.has_key('HTTP_IF_NONE_MATCH'))
533 self.assertEqual(d['HTTP_IF_NONE_MATCH'], "fred")
jcgregorio25185622006-10-28 05:12:34 +0000534
pilgrim00a352e2009-05-29 04:04:44 +0000535#MAP-commented this out because it consistently fails
536# def testGet304EndToEnd(self):
537# # Test that end to end headers get overwritten in the cache
538# uri = urlparse.urljoin(base, "304/end2end.cgi")
539# (response, content) = self.http.request(uri, "GET")
540# self.assertNotEqual(response['etag'], "")
541# old_date = response['date']
542# time.sleep(2)
543#
544# (response, content) = self.http.request(uri, "GET", headers = {'Cache-Control': 'max-age=0'})
545# # The response should be from the cache, but the Date: header should be updated.
546# new_date = response['date']
547# self.assertNotEqual(new_date, old_date)
548# self.assertEqual(response.status, 200)
549# self.assertEqual(response.fromcache, True)
jcgregorio2d66d4f2006-02-07 05:34:14 +0000550
551 def testGet304LastModified(self):
552 # Test that we can still handle a 304
553 # by only using the last-modified cache validator.
554 uri = urlparse.urljoin(base, "304/last-modified-only/last-modified-only.txt")
jcgregorio36140b52006-06-13 02:17:52 +0000555 (response, content) = self.http.request(uri, "GET")
jcgregorio2d66d4f2006-02-07 05:34:14 +0000556
557 self.assertNotEqual(response['last-modified'], "")
jcgregorio36140b52006-06-13 02:17:52 +0000558 (response, content) = self.http.request(uri, "GET")
559 (response, content) = self.http.request(uri, "GET")
jcgregorio2d66d4f2006-02-07 05:34:14 +0000560 self.assertEqual(response.status, 200)
561 self.assertEqual(response.fromcache, True)
562
563 def testGet307(self):
564 # Test that we do follow 307 redirects but
565 # do not cache the 307
566 uri = urlparse.urljoin(base, "307/onestep.asis")
jcgregorio36140b52006-06-13 02:17:52 +0000567 (response, content) = self.http.request(uri, "GET")
jcgregorio2d66d4f2006-02-07 05:34:14 +0000568 self.assertEqual(response.status, 200)
569 self.assertEqual(content, "This is the final destination.\n")
jcgregorioa0713ab2006-07-01 05:21:34 +0000570 self.assertEqual(response.previous.status, 307)
571 self.assertEqual(response.previous.fromcache, False)
jcgregorio2d66d4f2006-02-07 05:34:14 +0000572
jcgregorio36140b52006-06-13 02:17:52 +0000573 (response, content) = self.http.request(uri, "GET")
jcgregorio2d66d4f2006-02-07 05:34:14 +0000574 self.assertEqual(response.status, 200)
575 self.assertEqual(response.fromcache, True)
576 self.assertEqual(content, "This is the final destination.\n")
jcgregorioa0713ab2006-07-01 05:21:34 +0000577 self.assertEqual(response.previous.status, 307)
578 self.assertEqual(response.previous.fromcache, False)
jcgregorio2d66d4f2006-02-07 05:34:14 +0000579
580 def testGet410(self):
581 # Test that we pass 410's through
582 uri = urlparse.urljoin(base, "410/410.asis")
jcgregorio36140b52006-06-13 02:17:52 +0000583 (response, content) = self.http.request(uri, "GET")
jcgregorio2d66d4f2006-02-07 05:34:14 +0000584 self.assertEqual(response.status, 410)
585
chris dent89f15142009-12-24 14:02:57 -0600586 def testVaryHeaderSimple(self):
587 """
588 RFC 2616 13.6
589 When the cache receives a subsequent request whose Request-URI
590 specifies one or more cache entries including a Vary header field,
591 the cache MUST NOT use such a cache entry to construct a response
592 to the new request unless all of the selecting request-headers
593 present in the new request match the corresponding stored
594 request-headers in the original request.
595 """
596 # test that the vary header is sent
597 uri = urlparse.urljoin(base, "vary/accept.asis")
598 (response, content) = self.http.request(uri, "GET", headers={'Accept': 'text/plain'})
599 self.assertEqual(response.status, 200)
600 self.assertTrue(response.has_key('vary'))
601
602 # get the resource again, from the cache since accept header in this
603 # request is the same as the request
604 (response, content) = self.http.request(uri, "GET", headers={'Accept': 'text/plain'})
605 self.assertEqual(response.status, 200)
606 self.assertEqual(response.fromcache, True, msg="Should be from cache")
607
608 # get the resource again, not from cache since Accept headers does not match
609 (response, content) = self.http.request(uri, "GET", headers={'Accept': 'text/html'})
610 self.assertEqual(response.status, 200)
611 self.assertEqual(response.fromcache, False, msg="Should not be from cache")
612
613 # get the resource again, without any Accept header, so again no match
614 (response, content) = self.http.request(uri, "GET")
615 self.assertEqual(response.status, 200)
616 self.assertEqual(response.fromcache, False, msg="Should not be from cache")
617
618 def testNoVary(self):
619 # when there is no vary, a different Accept header (e.g.) should not
620 # impact if the cache is used
621 # test that the vary header is not sent
622 uri = urlparse.urljoin(base, "vary/no-vary.asis")
623 (response, content) = self.http.request(uri, "GET", headers={'Accept': 'text/plain'})
624 self.assertEqual(response.status, 200)
625 self.assertFalse(response.has_key('vary'))
626
627 (response, content) = self.http.request(uri, "GET", headers={'Accept': 'text/plain'})
628 self.assertEqual(response.status, 200)
629 self.assertEqual(response.fromcache, True, msg="Should be from cache")
630
631 (response, content) = self.http.request(uri, "GET", headers={'Accept': 'text/html'})
632 self.assertEqual(response.status, 200)
633 self.assertEqual(response.fromcache, True, msg="Should be from cache")
634
635 def testVaryHeaderDouble(self):
636 uri = urlparse.urljoin(base, "vary/accept-double.asis")
637 (response, content) = self.http.request(uri, "GET", headers={
638 'Accept': 'text/plain', 'Accept-Language': 'da, en-gb;q=0.8, en;q=0.7'})
639 self.assertEqual(response.status, 200)
640 self.assertTrue(response.has_key('vary'))
641
642 # we are from cache
643 (response, content) = self.http.request(uri, "GET", headers={
644 'Accept': 'text/plain', 'Accept-Language': 'da, en-gb;q=0.8, en;q=0.7'})
645 self.assertEqual(response.fromcache, True, msg="Should be from cache")
646
647 (response, content) = self.http.request(uri, "GET", headers={'Accept': 'text/plain'})
648 self.assertEqual(response.status, 200)
649 self.assertEqual(response.fromcache, False)
650
651 # get the resource again, not from cache, varied headers don't match exact
652 (response, content) = self.http.request(uri, "GET", headers={'Accept-Language': 'da'})
653 self.assertEqual(response.status, 200)
654 self.assertEqual(response.fromcache, False, msg="Should not be from cache")
655
jcgregorio88ef89b2010-05-13 23:42:11 -0400656 def testVaryUnusedHeader(self):
657 # A header's value is not considered to vary if it's not used at all.
658 uri = urlparse.urljoin(base, "vary/unused-header.asis")
659 (response, content) = self.http.request(uri, "GET", headers={
660 'Accept': 'text/plain'})
661 self.assertEqual(response.status, 200)
662 self.assertTrue(response.has_key('vary'))
663
664 # we are from cache
665 (response, content) = self.http.request(uri, "GET", headers={
666 'Accept': 'text/plain',})
667 self.assertEqual(response.fromcache, True, msg="Should be from cache")
668
chris dent89f15142009-12-24 14:02:57 -0600669
joe.gregorio0d4a2b82007-10-23 14:28:35 +0000670 def testHeadGZip(self):
671 # Test that we don't try to decompress a HEAD response
672 uri = urlparse.urljoin(base, "gzip/final-destination.txt")
673 (response, content) = self.http.request(uri, "HEAD")
674 self.assertEqual(response.status, 200)
675 self.assertNotEqual(int(response['content-length']), 0)
676 self.assertEqual(content, "")
677
jcgregorio2d66d4f2006-02-07 05:34:14 +0000678 def testGetGZip(self):
679 # Test that we support gzip compression
680 uri = urlparse.urljoin(base, "gzip/final-destination.txt")
jcgregorio36140b52006-06-13 02:17:52 +0000681 (response, content) = self.http.request(uri, "GET")
jcgregorio2d66d4f2006-02-07 05:34:14 +0000682 self.assertEqual(response.status, 200)
jcgregorio90fb4a42006-11-17 16:19:47 +0000683 self.assertFalse(response.has_key('content-encoding'))
joe.gregorio8b6d2312007-12-16 05:42:07 +0000684 self.assertTrue(response.has_key('-content-encoding'))
jcgregorio153f5882006-11-06 03:33:24 +0000685 self.assertEqual(int(response['content-length']), len("This is the final destination.\n"))
jcgregorio2d66d4f2006-02-07 05:34:14 +0000686 self.assertEqual(content, "This is the final destination.\n")
687
688 def testGetGZipFailure(self):
689 # Test that we raise a good exception when the gzip fails
jcgregorio07a9a4a2007-03-08 21:18:39 +0000690 self.http.force_exception_to_status_code = False
jcgregorio2d66d4f2006-02-07 05:34:14 +0000691 uri = urlparse.urljoin(base, "gzip/failed-compression.asis")
692 try:
jcgregorio36140b52006-06-13 02:17:52 +0000693 (response, content) = self.http.request(uri, "GET")
jcgregorio2d66d4f2006-02-07 05:34:14 +0000694 self.fail("Should never reach here")
695 except httplib2.FailedToDecompressContent:
696 pass
697 except Exception:
698 self.fail("Threw wrong kind of exception")
699
jcgregorio07a9a4a2007-03-08 21:18:39 +0000700 # Re-run the test with out the exceptions
701 self.http.force_exception_to_status_code = True
702
703 (response, content) = self.http.request(uri, "GET")
704 self.assertEqual(response.status, 500)
705 self.assertTrue(response.reason.startswith("Content purported"))
706
707 def testTimeout(self):
jcgregoriob2697912007-03-09 02:23:47 +0000708 self.http.force_exception_to_status_code = True
jcgregorio07a9a4a2007-03-08 21:18:39 +0000709 uri = urlparse.urljoin(base, "timeout/timeout.cgi")
710 try:
711 import socket
712 socket.setdefaulttimeout(1)
713 except:
714 # Don't run the test if we can't set the timeout
715 return
716 (response, content) = self.http.request(uri)
717 self.assertEqual(response.status, 408)
718 self.assertTrue(response.reason.startswith("Request Timeout"))
719 self.assertTrue(content.startswith("Request Timeout"))
720
jcgregoriob2697912007-03-09 02:23:47 +0000721 def testIndividualTimeout(self):
jcgregoriob2697912007-03-09 02:23:47 +0000722 uri = urlparse.urljoin(base, "timeout/timeout.cgi")
723 http = httplib2.Http(timeout=1)
joe.gregoriof28536d2007-10-23 14:10:11 +0000724 http.force_exception_to_status_code = True
jcgregoriob2697912007-03-09 02:23:47 +0000725
726 (response, content) = http.request(uri)
727 self.assertEqual(response.status, 408)
728 self.assertTrue(response.reason.startswith("Request Timeout"))
729 self.assertTrue(content.startswith("Request Timeout"))
730
jcgregorio07a9a4a2007-03-08 21:18:39 +0000731
Joe Gregorio1a7609f2009-07-16 10:59:44 -0400732 def testHTTPSInitTimeout(self):
733 c = httplib2.HTTPSConnectionWithTimeout('localhost', 80, timeout=47)
734 self.assertEqual(47, c.timeout)
735
jcgregorio2d66d4f2006-02-07 05:34:14 +0000736 def testGetDeflate(self):
737 # Test that we support deflate compression
738 uri = urlparse.urljoin(base, "deflate/deflated.asis")
jcgregorio36140b52006-06-13 02:17:52 +0000739 (response, content) = self.http.request(uri, "GET")
jcgregorio2d66d4f2006-02-07 05:34:14 +0000740 self.assertEqual(response.status, 200)
jcgregorio90fb4a42006-11-17 16:19:47 +0000741 self.assertFalse(response.has_key('content-encoding'))
jcgregorio153f5882006-11-06 03:33:24 +0000742 self.assertEqual(int(response['content-length']), len("This is the final destination."))
jcgregorio2d66d4f2006-02-07 05:34:14 +0000743 self.assertEqual(content, "This is the final destination.")
744
745 def testGetDeflateFailure(self):
746 # Test that we raise a good exception when the deflate fails
jcgregorio07a9a4a2007-03-08 21:18:39 +0000747 self.http.force_exception_to_status_code = False
748
jcgregorio2d66d4f2006-02-07 05:34:14 +0000749 uri = urlparse.urljoin(base, "deflate/failed-compression.asis")
750 try:
jcgregorio36140b52006-06-13 02:17:52 +0000751 (response, content) = self.http.request(uri, "GET")
jcgregorio2d66d4f2006-02-07 05:34:14 +0000752 self.fail("Should never reach here")
753 except httplib2.FailedToDecompressContent:
754 pass
755 except Exception:
756 self.fail("Threw wrong kind of exception")
757
jcgregorio07a9a4a2007-03-08 21:18:39 +0000758 # Re-run the test with out the exceptions
759 self.http.force_exception_to_status_code = True
760
761 (response, content) = self.http.request(uri, "GET")
762 self.assertEqual(response.status, 500)
763 self.assertTrue(response.reason.startswith("Content purported"))
764
jcgregorio2d66d4f2006-02-07 05:34:14 +0000765 def testGetDuplicateHeaders(self):
766 # Test that duplicate headers get concatenated via ','
767 uri = urlparse.urljoin(base, "duplicate-headers/multilink.asis")
jcgregorio36140b52006-06-13 02:17:52 +0000768 (response, content) = self.http.request(uri, "GET")
jcgregorio2d66d4f2006-02-07 05:34:14 +0000769 self.assertEqual(response.status, 200)
770 self.assertEqual(content, "This is content\n")
771 self.assertEqual(response['link'].split(",")[0], '<http://bitworking.org>; rel="home"; title="BitWorking"')
772
773 def testGetCacheControlNoCache(self):
774 # Test Cache-Control: no-cache on requests
775 uri = urlparse.urljoin(base, "304/test_etag.txt")
jcgregorio36140b52006-06-13 02:17:52 +0000776 (response, content) = self.http.request(uri, "GET")
jcgregorio2d66d4f2006-02-07 05:34:14 +0000777 self.assertNotEqual(response['etag'], "")
jcgregorio36140b52006-06-13 02:17:52 +0000778 (response, content) = self.http.request(uri, "GET")
jcgregorio2d66d4f2006-02-07 05:34:14 +0000779 self.assertEqual(response.status, 200)
780 self.assertEqual(response.fromcache, True)
781
jcgregorio36140b52006-06-13 02:17:52 +0000782 (response, content) = self.http.request(uri, "GET", headers={'Cache-Control': 'no-cache'})
jcgregorio2d66d4f2006-02-07 05:34:14 +0000783 self.assertEqual(response.status, 200)
784 self.assertEqual(response.fromcache, False)
785
786 def testGetCacheControlPragmaNoCache(self):
787 # Test Pragma: no-cache on requests
788 uri = urlparse.urljoin(base, "304/test_etag.txt")
jcgregorio36140b52006-06-13 02:17:52 +0000789 (response, content) = self.http.request(uri, "GET")
jcgregorio2d66d4f2006-02-07 05:34:14 +0000790 self.assertNotEqual(response['etag'], "")
jcgregorio36140b52006-06-13 02:17:52 +0000791 (response, content) = self.http.request(uri, "GET")
jcgregorio2d66d4f2006-02-07 05:34:14 +0000792 self.assertEqual(response.status, 200)
793 self.assertEqual(response.fromcache, True)
794
jcgregorio36140b52006-06-13 02:17:52 +0000795 (response, content) = self.http.request(uri, "GET", headers={'Pragma': 'no-cache'})
jcgregorio2d66d4f2006-02-07 05:34:14 +0000796 self.assertEqual(response.status, 200)
797 self.assertEqual(response.fromcache, False)
798
799 def testGetCacheControlNoStoreRequest(self):
800 # A no-store request means that the response should not be stored.
801 uri = urlparse.urljoin(base, "304/test_etag.txt")
802
jcgregorio36140b52006-06-13 02:17:52 +0000803 (response, content) = self.http.request(uri, "GET", headers={'Cache-Control': 'no-store'})
jcgregorio2d66d4f2006-02-07 05:34:14 +0000804 self.assertEqual(response.status, 200)
805 self.assertEqual(response.fromcache, False)
806
jcgregorio36140b52006-06-13 02:17:52 +0000807 (response, content) = self.http.request(uri, "GET", headers={'Cache-Control': 'no-store'})
jcgregorio2d66d4f2006-02-07 05:34:14 +0000808 self.assertEqual(response.status, 200)
809 self.assertEqual(response.fromcache, False)
810
811 def testGetCacheControlNoStoreResponse(self):
812 # A no-store response means that the response should not be stored.
813 uri = urlparse.urljoin(base, "no-store/no-store.asis")
814
jcgregorio36140b52006-06-13 02:17:52 +0000815 (response, content) = self.http.request(uri, "GET")
jcgregorio2d66d4f2006-02-07 05:34:14 +0000816 self.assertEqual(response.status, 200)
817 self.assertEqual(response.fromcache, False)
818
jcgregorio36140b52006-06-13 02:17:52 +0000819 (response, content) = self.http.request(uri, "GET")
jcgregorio2d66d4f2006-02-07 05:34:14 +0000820 self.assertEqual(response.status, 200)
821 self.assertEqual(response.fromcache, False)
jcgregorio2d66d4f2006-02-07 05:34:14 +0000822
823 def testGetCacheControlNoCacheNoStoreRequest(self):
824 # Test that a no-store, no-cache clears the entry from the cache
825 # even if it was cached previously.
826 uri = urlparse.urljoin(base, "304/test_etag.txt")
827
jcgregorio36140b52006-06-13 02:17:52 +0000828 (response, content) = self.http.request(uri, "GET")
829 (response, content) = self.http.request(uri, "GET")
jcgregorio2d66d4f2006-02-07 05:34:14 +0000830 self.assertEqual(response.fromcache, True)
jcgregorio36140b52006-06-13 02:17:52 +0000831 (response, content) = self.http.request(uri, "GET", headers={'Cache-Control': 'no-store, no-cache'})
832 (response, content) = self.http.request(uri, "GET", headers={'Cache-Control': 'no-store, no-cache'})
jcgregorio2d66d4f2006-02-07 05:34:14 +0000833 self.assertEqual(response.status, 200)
834 self.assertEqual(response.fromcache, False)
jcgregorio2d66d4f2006-02-07 05:34:14 +0000835
836 def testUpdateInvalidatesCache(self):
837 # Test that calling PUT or DELETE on a
838 # URI that is cache invalidates that cache.
839 uri = urlparse.urljoin(base, "304/test_etag.txt")
840
jcgregorio36140b52006-06-13 02:17:52 +0000841 (response, content) = self.http.request(uri, "GET")
842 (response, content) = self.http.request(uri, "GET")
jcgregorio2d66d4f2006-02-07 05:34:14 +0000843 self.assertEqual(response.fromcache, True)
jcgregorio36140b52006-06-13 02:17:52 +0000844 (response, content) = self.http.request(uri, "DELETE")
jcgregorio2d66d4f2006-02-07 05:34:14 +0000845 self.assertEqual(response.status, 405)
846
jcgregorio36140b52006-06-13 02:17:52 +0000847 (response, content) = self.http.request(uri, "GET")
jcgregorio2d66d4f2006-02-07 05:34:14 +0000848 self.assertEqual(response.fromcache, False)
849
850 def testUpdateUsesCachedETag(self):
851 # Test that we natively support http://www.w3.org/1999/04/Editing/
852 uri = urlparse.urljoin(base, "conditional-updates/test.cgi")
853
jcgregorio36140b52006-06-13 02:17:52 +0000854 (response, content) = self.http.request(uri, "GET")
jcgregorio2d66d4f2006-02-07 05:34:14 +0000855 self.assertEqual(response.status, 200)
856 self.assertEqual(response.fromcache, False)
jcgregorio36140b52006-06-13 02:17:52 +0000857 (response, content) = self.http.request(uri, "GET")
jcgregorio2d66d4f2006-02-07 05:34:14 +0000858 self.assertEqual(response.status, 200)
859 self.assertEqual(response.fromcache, True)
Joe Gregoriocd868102009-09-29 17:09:16 -0400860 (response, content) = self.http.request(uri, "PUT", body="foo")
jcgregorio2d66d4f2006-02-07 05:34:14 +0000861 self.assertEqual(response.status, 200)
Joe Gregoriocd868102009-09-29 17:09:16 -0400862 (response, content) = self.http.request(uri, "PUT", body="foo")
jcgregorio2d66d4f2006-02-07 05:34:14 +0000863 self.assertEqual(response.status, 412)
864
joe.gregorio700f04d2008-09-06 04:46:32 +0000865 def testUpdateUsesCachedETagAndOCMethod(self):
866 # Test that we natively support http://www.w3.org/1999/04/Editing/
867 uri = urlparse.urljoin(base, "conditional-updates/test.cgi")
868
869 (response, content) = self.http.request(uri, "GET")
870 self.assertEqual(response.status, 200)
871 self.assertEqual(response.fromcache, False)
872 (response, content) = self.http.request(uri, "GET")
873 self.assertEqual(response.status, 200)
874 self.assertEqual(response.fromcache, True)
875 self.http.optimistic_concurrency_methods.append("DELETE")
876 (response, content) = self.http.request(uri, "DELETE")
877 self.assertEqual(response.status, 200)
878
879
jcgregorio4b145e82007-01-18 19:46:34 +0000880 def testUpdateUsesCachedETagOverridden(self):
881 # Test that we natively support http://www.w3.org/1999/04/Editing/
882 uri = urlparse.urljoin(base, "conditional-updates/test.cgi")
883
884 (response, content) = self.http.request(uri, "GET")
885 self.assertEqual(response.status, 200)
886 self.assertEqual(response.fromcache, False)
887 (response, content) = self.http.request(uri, "GET")
888 self.assertEqual(response.status, 200)
889 self.assertEqual(response.fromcache, True)
Joe Gregoriocd868102009-09-29 17:09:16 -0400890 (response, content) = self.http.request(uri, "PUT", body="foo", headers={'if-match': 'fred'})
jcgregorio4b145e82007-01-18 19:46:34 +0000891 self.assertEqual(response.status, 412)
892
jcgregorio2d66d4f2006-02-07 05:34:14 +0000893 def testBasicAuth(self):
894 # Test Basic Authentication
895 uri = urlparse.urljoin(base, "basic/file.txt")
jcgregorio36140b52006-06-13 02:17:52 +0000896 (response, content) = self.http.request(uri, "GET")
jcgregorio2d66d4f2006-02-07 05:34:14 +0000897 self.assertEqual(response.status, 401)
898
899 uri = urlparse.urljoin(base, "basic/")
jcgregorio36140b52006-06-13 02:17:52 +0000900 (response, content) = self.http.request(uri, "GET")
jcgregorio2d66d4f2006-02-07 05:34:14 +0000901 self.assertEqual(response.status, 401)
902
jcgregorio36140b52006-06-13 02:17:52 +0000903 self.http.add_credentials('joe', 'password')
904 (response, content) = self.http.request(uri, "GET")
jcgregorio2d66d4f2006-02-07 05:34:14 +0000905 self.assertEqual(response.status, 200)
906
907 uri = urlparse.urljoin(base, "basic/file.txt")
jcgregorio36140b52006-06-13 02:17:52 +0000908 (response, content) = self.http.request(uri, "GET")
jcgregorio2d66d4f2006-02-07 05:34:14 +0000909 self.assertEqual(response.status, 200)
910
jcgregoriode8238d2007-03-07 19:08:26 +0000911 def testBasicAuthWithDomain(self):
912 # Test Basic Authentication
913 uri = urlparse.urljoin(base, "basic/file.txt")
914 (response, content) = self.http.request(uri, "GET")
915 self.assertEqual(response.status, 401)
916
917 uri = urlparse.urljoin(base, "basic/")
918 (response, content) = self.http.request(uri, "GET")
919 self.assertEqual(response.status, 401)
920
921 self.http.add_credentials('joe', 'password', "example.org")
922 (response, content) = self.http.request(uri, "GET")
923 self.assertEqual(response.status, 401)
924
925 uri = urlparse.urljoin(base, "basic/file.txt")
926 (response, content) = self.http.request(uri, "GET")
927 self.assertEqual(response.status, 401)
928
929 domain = urlparse.urlparse(base)[1]
930 self.http.add_credentials('joe', 'password', domain)
931 (response, content) = self.http.request(uri, "GET")
932 self.assertEqual(response.status, 200)
933
934 uri = urlparse.urljoin(base, "basic/file.txt")
935 (response, content) = self.http.request(uri, "GET")
936 self.assertEqual(response.status, 200)
937
938
939
940
941
942
jcgregorio2d66d4f2006-02-07 05:34:14 +0000943 def testBasicAuthTwoDifferentCredentials(self):
jcgregorioadbb4f82006-05-19 15:17:42 +0000944 # Test Basic Authentication with multiple sets of credentials
jcgregorio2d66d4f2006-02-07 05:34:14 +0000945 uri = urlparse.urljoin(base, "basic2/file.txt")
jcgregorio36140b52006-06-13 02:17:52 +0000946 (response, content) = self.http.request(uri, "GET")
jcgregorio2d66d4f2006-02-07 05:34:14 +0000947 self.assertEqual(response.status, 401)
948
949 uri = urlparse.urljoin(base, "basic2/")
jcgregorio36140b52006-06-13 02:17:52 +0000950 (response, content) = self.http.request(uri, "GET")
jcgregorio2d66d4f2006-02-07 05:34:14 +0000951 self.assertEqual(response.status, 401)
952
jcgregorio36140b52006-06-13 02:17:52 +0000953 self.http.add_credentials('fred', 'barney')
954 (response, content) = self.http.request(uri, "GET")
jcgregorio2d66d4f2006-02-07 05:34:14 +0000955 self.assertEqual(response.status, 200)
956
957 uri = urlparse.urljoin(base, "basic2/file.txt")
jcgregorio36140b52006-06-13 02:17:52 +0000958 (response, content) = self.http.request(uri, "GET")
jcgregorio2d66d4f2006-02-07 05:34:14 +0000959 self.assertEqual(response.status, 200)
960
961 def testBasicAuthNested(self):
962 # Test Basic Authentication with resources
963 # that are nested
964 uri = urlparse.urljoin(base, "basic-nested/")
jcgregorio36140b52006-06-13 02:17:52 +0000965 (response, content) = self.http.request(uri, "GET")
jcgregorio2d66d4f2006-02-07 05:34:14 +0000966 self.assertEqual(response.status, 401)
967
968 uri = urlparse.urljoin(base, "basic-nested/subdir")
jcgregorio36140b52006-06-13 02:17:52 +0000969 (response, content) = self.http.request(uri, "GET")
jcgregorio2d66d4f2006-02-07 05:34:14 +0000970 self.assertEqual(response.status, 401)
971
jcgregorioadbb4f82006-05-19 15:17:42 +0000972 # Now add in credentials one at a time and test.
jcgregorio36140b52006-06-13 02:17:52 +0000973 self.http.add_credentials('joe', 'password')
jcgregorio2d66d4f2006-02-07 05:34:14 +0000974
975 uri = urlparse.urljoin(base, "basic-nested/")
jcgregorio36140b52006-06-13 02:17:52 +0000976 (response, content) = self.http.request(uri, "GET")
jcgregorio2d66d4f2006-02-07 05:34:14 +0000977 self.assertEqual(response.status, 200)
978
979 uri = urlparse.urljoin(base, "basic-nested/subdir")
jcgregorio36140b52006-06-13 02:17:52 +0000980 (response, content) = self.http.request(uri, "GET")
jcgregorio2d66d4f2006-02-07 05:34:14 +0000981 self.assertEqual(response.status, 401)
982
jcgregorio36140b52006-06-13 02:17:52 +0000983 self.http.add_credentials('fred', 'barney')
jcgregorio2d66d4f2006-02-07 05:34:14 +0000984
985 uri = urlparse.urljoin(base, "basic-nested/")
jcgregorio36140b52006-06-13 02:17:52 +0000986 (response, content) = self.http.request(uri, "GET")
jcgregorio2d66d4f2006-02-07 05:34:14 +0000987 self.assertEqual(response.status, 200)
988
989 uri = urlparse.urljoin(base, "basic-nested/subdir")
jcgregorio36140b52006-06-13 02:17:52 +0000990 (response, content) = self.http.request(uri, "GET")
jcgregorio2d66d4f2006-02-07 05:34:14 +0000991 self.assertEqual(response.status, 200)
992
993 def testDigestAuth(self):
994 # Test that we support Digest Authentication
995 uri = urlparse.urljoin(base, "digest/")
jcgregorio36140b52006-06-13 02:17:52 +0000996 (response, content) = self.http.request(uri, "GET")
jcgregorio2d66d4f2006-02-07 05:34:14 +0000997 self.assertEqual(response.status, 401)
998
jcgregorio36140b52006-06-13 02:17:52 +0000999 self.http.add_credentials('joe', 'password')
1000 (response, content) = self.http.request(uri, "GET")
jcgregorio2d66d4f2006-02-07 05:34:14 +00001001 self.assertEqual(response.status, 200)
1002
1003 uri = urlparse.urljoin(base, "digest/file.txt")
jcgregorio36140b52006-06-13 02:17:52 +00001004 (response, content) = self.http.request(uri, "GET")
jcgregorio2d66d4f2006-02-07 05:34:14 +00001005
1006 def testDigestAuthNextNonceAndNC(self):
1007 # Test that if the server sets nextnonce that we reset
1008 # the nonce count back to 1
1009 uri = urlparse.urljoin(base, "digest/file.txt")
jcgregorio36140b52006-06-13 02:17:52 +00001010 self.http.add_credentials('joe', 'password')
1011 (response, content) = self.http.request(uri, "GET", headers = {"cache-control":"no-cache"})
jcgregorio2d66d4f2006-02-07 05:34:14 +00001012 info = httplib2._parse_www_authenticate(response, 'authentication-info')
1013 self.assertEqual(response.status, 200)
jcgregorio36140b52006-06-13 02:17:52 +00001014 (response, content) = self.http.request(uri, "GET", headers = {"cache-control":"no-cache"})
jcgregorio2d66d4f2006-02-07 05:34:14 +00001015 info2 = httplib2._parse_www_authenticate(response, 'authentication-info')
1016 self.assertEqual(response.status, 200)
1017
1018 if info.has_key('nextnonce'):
1019 self.assertEqual(info2['nc'], 1)
1020
1021 def testDigestAuthStale(self):
1022 # Test that we can handle a nonce becoming stale
1023 uri = urlparse.urljoin(base, "digest-expire/file.txt")
jcgregorio36140b52006-06-13 02:17:52 +00001024 self.http.add_credentials('joe', 'password')
1025 (response, content) = self.http.request(uri, "GET", headers = {"cache-control":"no-cache"})
jcgregorio2d66d4f2006-02-07 05:34:14 +00001026 info = httplib2._parse_www_authenticate(response, 'authentication-info')
1027 self.assertEqual(response.status, 200)
1028
1029 time.sleep(3)
1030 # Sleep long enough that the nonce becomes stale
1031
jcgregorio36140b52006-06-13 02:17:52 +00001032 (response, content) = self.http.request(uri, "GET", headers = {"cache-control":"no-cache"})
jcgregorio2d66d4f2006-02-07 05:34:14 +00001033 self.assertFalse(response.fromcache)
1034 self.assertTrue(response._stale_digest)
1035 info3 = httplib2._parse_www_authenticate(response, 'authentication-info')
1036 self.assertEqual(response.status, 200)
1037
1038 def reflector(self, content):
jcgregorio25185622006-10-28 05:12:34 +00001039 return dict( [tuple(x.split("=", 1)) for x in content.strip().split("\n")] )
jcgregorio2d66d4f2006-02-07 05:34:14 +00001040
1041 def testReflector(self):
1042 uri = urlparse.urljoin(base, "reflector/reflector.cgi")
jcgregorio36140b52006-06-13 02:17:52 +00001043 (response, content) = self.http.request(uri, "GET")
jcgregorio2d66d4f2006-02-07 05:34:14 +00001044 d = self.reflector(content)
1045 self.assertTrue(d.has_key('HTTP_USER_AGENT'))
1046
Joe Gregorio84cc10a2009-09-01 13:02:49 -04001047 def testConnectionClose(self):
1048 uri = "http://www.google.com/"
1049 (response, content) = self.http.request(uri, "GET")
1050 for c in self.http.connections.values():
1051 self.assertNotEqual(None, c.sock)
1052 (response, content) = self.http.request(uri, "GET", headers={"connection": "close"})
1053 for c in self.http.connections.values():
1054 self.assertEqual(None, c.sock)
1055
1056
jcgregorio36140b52006-06-13 02:17:52 +00001057try:
1058 import memcache
1059 class HttpTestMemCached(HttpTest):
1060 def setUp(self):
1061 self.cache = memcache.Client(['127.0.0.1:11211'], debug=0)
jcgregorio47d24672006-06-29 05:18:59 +00001062 #self.cache = memcache.Client(['10.0.0.4:11211'], debug=1)
jcgregorio36140b52006-06-13 02:17:52 +00001063 self.http = httplib2.Http(self.cache)
1064 self.cache.flush_all()
jcgregorio47d24672006-06-29 05:18:59 +00001065 # Not exactly sure why the sleep is needed here, but
1066 # if not present then some unit tests that rely on caching
1067 # fail. Memcached seems to lose some sets immediately
1068 # after a flush_all if the set is to a value that
1069 # was previously cached. (Maybe the flush is handled async?)
1070 time.sleep(1)
jcgregorio36140b52006-06-13 02:17:52 +00001071 self.http.clear_credentials()
1072except:
1073 pass
1074
1075
1076
chris dent89f15142009-12-24 14:02:57 -06001077
jcgregoriodb8dfc82006-03-31 14:59:46 +00001078# ------------------------------------------------------------------------
jcgregorio2d66d4f2006-02-07 05:34:14 +00001079
1080class HttpPrivateTest(unittest.TestCase):
1081
1082 def testParseCacheControl(self):
1083 # Test that we can parse the Cache-Control header
1084 self.assertEqual({}, httplib2._parse_cache_control({}))
1085 self.assertEqual({'no-cache': 1}, httplib2._parse_cache_control({'cache-control': ' no-cache'}))
1086 cc = httplib2._parse_cache_control({'cache-control': ' no-cache, max-age = 7200'})
1087 self.assertEqual(cc['no-cache'], 1)
1088 self.assertEqual(cc['max-age'], '7200')
1089 cc = httplib2._parse_cache_control({'cache-control': ' , '})
1090 self.assertEqual(cc[''], 1)
1091
Joe Gregorioe314e8b2009-07-16 20:11:28 -04001092 try:
1093 cc = httplib2._parse_cache_control({'cache-control': 'Max-age=3600;post-check=1800,pre-check=3600'})
1094 self.assertTrue("max-age" in cc)
1095 except:
1096 self.fail("Should not throw exception")
1097
jcgregorio2d66d4f2006-02-07 05:34:14 +00001098 def testNormalizeHeaders(self):
1099 # Test that we normalize headers to lowercase
1100 h = httplib2._normalize_headers({'Cache-Control': 'no-cache', 'Other': 'Stuff'})
1101 self.assertTrue(h.has_key('cache-control'))
1102 self.assertTrue(h.has_key('other'))
1103 self.assertEqual('Stuff', h['other'])
1104
1105 def testExpirationModelTransparent(self):
1106 # Test that no-cache makes our request TRANSPARENT
1107 response_headers = {
1108 'cache-control': 'max-age=7200'
1109 }
1110 request_headers = {
1111 'cache-control': 'no-cache'
1112 }
1113 self.assertEqual("TRANSPARENT", httplib2._entry_disposition(response_headers, request_headers))
1114
jcgregorio45865012007-01-18 16:38:22 +00001115 def testMaxAgeNonNumeric(self):
1116 # Test that no-cache makes our request TRANSPARENT
1117 response_headers = {
1118 'cache-control': 'max-age=fred, min-fresh=barney'
1119 }
1120 request_headers = {
1121 }
1122 self.assertEqual("STALE", httplib2._entry_disposition(response_headers, request_headers))
1123
1124
jcgregorio2d66d4f2006-02-07 05:34:14 +00001125 def testExpirationModelNoCacheResponse(self):
1126 # The date and expires point to an entry that should be
1127 # FRESH, but the no-cache over-rides that.
1128 now = time.time()
1129 response_headers = {
1130 'date': time.strftime("%a, %d %b %Y %H:%M:%S GMT", time.gmtime(now)),
1131 'expires': time.strftime("%a, %d %b %Y %H:%M:%S GMT", time.gmtime(now+4)),
1132 'cache-control': 'no-cache'
1133 }
1134 request_headers = {
1135 }
1136 self.assertEqual("STALE", httplib2._entry_disposition(response_headers, request_headers))
1137
1138 def testExpirationModelStaleRequestMustReval(self):
1139 # must-revalidate forces STALE
1140 self.assertEqual("STALE", httplib2._entry_disposition({}, {'cache-control': 'must-revalidate'}))
1141
1142 def testExpirationModelStaleResponseMustReval(self):
1143 # must-revalidate forces STALE
1144 self.assertEqual("STALE", httplib2._entry_disposition({'cache-control': 'must-revalidate'}, {}))
1145
1146 def testExpirationModelFresh(self):
1147 response_headers = {
1148 'date': time.strftime("%a, %d %b %Y %H:%M:%S GMT", time.gmtime()),
1149 'cache-control': 'max-age=2'
1150 }
1151 request_headers = {
1152 }
1153 self.assertEqual("FRESH", httplib2._entry_disposition(response_headers, request_headers))
1154 time.sleep(3)
1155 self.assertEqual("STALE", httplib2._entry_disposition(response_headers, request_headers))
1156
1157 def testExpirationMaxAge0(self):
1158 response_headers = {
1159 'date': time.strftime("%a, %d %b %Y %H:%M:%S GMT", time.gmtime()),
1160 'cache-control': 'max-age=0'
1161 }
1162 request_headers = {
1163 }
1164 self.assertEqual("STALE", httplib2._entry_disposition(response_headers, request_headers))
1165
1166 def testExpirationModelDateAndExpires(self):
1167 now = time.time()
1168 response_headers = {
1169 'date': time.strftime("%a, %d %b %Y %H:%M:%S GMT", time.gmtime(now)),
1170 'expires': time.strftime("%a, %d %b %Y %H:%M:%S GMT", time.gmtime(now+2)),
1171 }
1172 request_headers = {
1173 }
1174 self.assertEqual("FRESH", httplib2._entry_disposition(response_headers, request_headers))
1175 time.sleep(3)
1176 self.assertEqual("STALE", httplib2._entry_disposition(response_headers, request_headers))
1177
jcgregoriof9511052007-06-01 14:56:34 +00001178 def testExpiresZero(self):
1179 now = time.time()
1180 response_headers = {
1181 'date': time.strftime("%a, %d %b %Y %H:%M:%S GMT", time.gmtime(now)),
1182 'expires': "0",
1183 }
1184 request_headers = {
1185 }
1186 self.assertEqual("STALE", httplib2._entry_disposition(response_headers, request_headers))
1187
jcgregorio2d66d4f2006-02-07 05:34:14 +00001188 def testExpirationModelDateOnly(self):
1189 now = time.time()
1190 response_headers = {
1191 'date': time.strftime("%a, %d %b %Y %H:%M:%S GMT", time.gmtime(now+3)),
1192 }
1193 request_headers = {
1194 }
1195 self.assertEqual("STALE", httplib2._entry_disposition(response_headers, request_headers))
1196
1197 def testExpirationModelOnlyIfCached(self):
1198 response_headers = {
1199 }
1200 request_headers = {
1201 'cache-control': 'only-if-cached',
1202 }
1203 self.assertEqual("FRESH", httplib2._entry_disposition(response_headers, request_headers))
1204
1205 def testExpirationModelMaxAgeBoth(self):
1206 now = time.time()
1207 response_headers = {
1208 'date': time.strftime("%a, %d %b %Y %H:%M:%S GMT", time.gmtime(now)),
1209 'cache-control': 'max-age=2'
1210 }
1211 request_headers = {
1212 'cache-control': 'max-age=0'
1213 }
1214 self.assertEqual("STALE", httplib2._entry_disposition(response_headers, request_headers))
1215
1216 def testExpirationModelDateAndExpiresMinFresh1(self):
1217 now = time.time()
1218 response_headers = {
1219 'date': time.strftime("%a, %d %b %Y %H:%M:%S GMT", time.gmtime(now)),
1220 'expires': time.strftime("%a, %d %b %Y %H:%M:%S GMT", time.gmtime(now+2)),
1221 }
1222 request_headers = {
1223 'cache-control': 'min-fresh=2'
1224 }
1225 self.assertEqual("STALE", httplib2._entry_disposition(response_headers, request_headers))
1226
1227 def testExpirationModelDateAndExpiresMinFresh2(self):
1228 now = time.time()
1229 response_headers = {
1230 'date': time.strftime("%a, %d %b %Y %H:%M:%S GMT", time.gmtime(now)),
1231 'expires': time.strftime("%a, %d %b %Y %H:%M:%S GMT", time.gmtime(now+4)),
1232 }
1233 request_headers = {
1234 'cache-control': 'min-fresh=2'
1235 }
1236 self.assertEqual("FRESH", httplib2._entry_disposition(response_headers, request_headers))
1237
1238 def testParseWWWAuthenticateEmpty(self):
1239 res = httplib2._parse_www_authenticate({})
1240 self.assertEqual(len(res.keys()), 0)
1241
jcgregoriofd22e432006-04-27 02:00:08 +00001242 def testParseWWWAuthenticate(self):
1243 # different uses of spaces around commas
1244 res = httplib2._parse_www_authenticate({ 'www-authenticate': 'Test realm="test realm" , foo=foo ,bar="bar", baz=baz,qux=qux'})
1245 self.assertEqual(len(res.keys()), 1)
1246 self.assertEqual(len(res['test'].keys()), 5)
1247
1248 # tokens with non-alphanum
1249 res = httplib2._parse_www_authenticate({ 'www-authenticate': 'T*!%#st realm=to*!%#en, to*!%#en="quoted string"'})
1250 self.assertEqual(len(res.keys()), 1)
1251 self.assertEqual(len(res['t*!%#st'].keys()), 2)
1252
1253 # quoted string with quoted pairs
1254 res = httplib2._parse_www_authenticate({ 'www-authenticate': 'Test realm="a \\"test\\" realm"'})
1255 self.assertEqual(len(res.keys()), 1)
1256 self.assertEqual(res['test']['realm'], 'a "test" realm')
1257
1258 def testParseWWWAuthenticateStrict(self):
1259 httplib2.USE_WWW_AUTH_STRICT_PARSING = 1;
1260 self.testParseWWWAuthenticate();
1261 httplib2.USE_WWW_AUTH_STRICT_PARSING = 0;
1262
jcgregorio2d66d4f2006-02-07 05:34:14 +00001263 def testParseWWWAuthenticateBasic(self):
1264 res = httplib2._parse_www_authenticate({ 'www-authenticate': 'Basic realm="me"'})
1265 basic = res['basic']
1266 self.assertEqual('me', basic['realm'])
1267
1268 res = httplib2._parse_www_authenticate({ 'www-authenticate': 'Basic realm="me", algorithm="MD5"'})
1269 basic = res['basic']
1270 self.assertEqual('me', basic['realm'])
1271 self.assertEqual('MD5', basic['algorithm'])
1272
1273 res = httplib2._parse_www_authenticate({ 'www-authenticate': 'Basic realm="me", algorithm=MD5'})
1274 basic = res['basic']
1275 self.assertEqual('me', basic['realm'])
1276 self.assertEqual('MD5', basic['algorithm'])
1277
1278 def testParseWWWAuthenticateBasic2(self):
1279 res = httplib2._parse_www_authenticate({ 'www-authenticate': 'Basic realm="me",other="fred" '})
1280 basic = res['basic']
1281 self.assertEqual('me', basic['realm'])
1282 self.assertEqual('fred', basic['other'])
1283
1284 def testParseWWWAuthenticateBasic3(self):
1285 res = httplib2._parse_www_authenticate({ 'www-authenticate': 'Basic REAlm="me" '})
1286 basic = res['basic']
1287 self.assertEqual('me', basic['realm'])
1288
1289
1290 def testParseWWWAuthenticateDigest(self):
1291 res = httplib2._parse_www_authenticate({ 'www-authenticate':
1292 'Digest realm="testrealm@host.com", qop="auth,auth-int", nonce="dcd98b7102dd2f0e8b11d0f600bfb0c093", opaque="5ccc069c403ebaf9f0171e9517f40e41"'})
1293 digest = res['digest']
1294 self.assertEqual('testrealm@host.com', digest['realm'])
1295 self.assertEqual('auth,auth-int', digest['qop'])
1296
1297
1298 def testParseWWWAuthenticateMultiple(self):
1299 res = httplib2._parse_www_authenticate({ 'www-authenticate':
1300 'Digest realm="testrealm@host.com", qop="auth,auth-int", nonce="dcd98b7102dd2f0e8b11d0f600bfb0c093", opaque="5ccc069c403ebaf9f0171e9517f40e41" Basic REAlm="me" '})
1301 digest = res['digest']
1302 self.assertEqual('testrealm@host.com', digest['realm'])
1303 self.assertEqual('auth,auth-int', digest['qop'])
1304 self.assertEqual('dcd98b7102dd2f0e8b11d0f600bfb0c093', digest['nonce'])
1305 self.assertEqual('5ccc069c403ebaf9f0171e9517f40e41', digest['opaque'])
1306 basic = res['basic']
1307 self.assertEqual('me', basic['realm'])
1308
1309 def testParseWWWAuthenticateMultiple2(self):
1310 # Handle an added comma between challenges, which might get thrown in if the challenges were
1311 # originally sent in separate www-authenticate headers.
1312 res = httplib2._parse_www_authenticate({ 'www-authenticate':
1313 'Digest realm="testrealm@host.com", qop="auth,auth-int", nonce="dcd98b7102dd2f0e8b11d0f600bfb0c093", opaque="5ccc069c403ebaf9f0171e9517f40e41", Basic REAlm="me" '})
1314 digest = res['digest']
1315 self.assertEqual('testrealm@host.com', digest['realm'])
1316 self.assertEqual('auth,auth-int', digest['qop'])
1317 self.assertEqual('dcd98b7102dd2f0e8b11d0f600bfb0c093', digest['nonce'])
1318 self.assertEqual('5ccc069c403ebaf9f0171e9517f40e41', digest['opaque'])
1319 basic = res['basic']
1320 self.assertEqual('me', basic['realm'])
1321
1322 def testParseWWWAuthenticateMultiple3(self):
1323 # Handle an added comma between challenges, which might get thrown in if the challenges were
1324 # originally sent in separate www-authenticate headers.
1325 res = httplib2._parse_www_authenticate({ 'www-authenticate':
1326 'Digest realm="testrealm@host.com", qop="auth,auth-int", nonce="dcd98b7102dd2f0e8b11d0f600bfb0c093", opaque="5ccc069c403ebaf9f0171e9517f40e41", Basic REAlm="me", WSSE realm="foo", profile="UsernameToken"'})
1327 digest = res['digest']
1328 self.assertEqual('testrealm@host.com', digest['realm'])
1329 self.assertEqual('auth,auth-int', digest['qop'])
1330 self.assertEqual('dcd98b7102dd2f0e8b11d0f600bfb0c093', digest['nonce'])
1331 self.assertEqual('5ccc069c403ebaf9f0171e9517f40e41', digest['opaque'])
1332 basic = res['basic']
1333 self.assertEqual('me', basic['realm'])
1334 wsse = res['wsse']
1335 self.assertEqual('foo', wsse['realm'])
1336 self.assertEqual('UsernameToken', wsse['profile'])
1337
1338 def testParseWWWAuthenticateMultiple4(self):
1339 res = httplib2._parse_www_authenticate({ 'www-authenticate':
1340 'Digest realm="test-real.m@host.com", qop \t=\t"\tauth,auth-int", nonce="(*)&^&$%#",opaque="5ccc069c403ebaf9f0171e9517f40e41", Basic REAlm="me", WSSE realm="foo", profile="UsernameToken"'})
1341 digest = res['digest']
1342 self.assertEqual('test-real.m@host.com', digest['realm'])
1343 self.assertEqual('\tauth,auth-int', digest['qop'])
1344 self.assertEqual('(*)&^&$%#', digest['nonce'])
1345
1346 def testParseWWWAuthenticateMoreQuoteCombos(self):
1347 res = httplib2._parse_www_authenticate({'www-authenticate':'Digest realm="myrealm", nonce="Ygk86AsKBAA=3516200d37f9a3230352fde99977bd6d472d4306", algorithm=MD5, qop="auth", stale=true'})
1348 digest = res['digest']
1349 self.assertEqual('myrealm', digest['realm'])
1350
1351 def testDigestObject(self):
1352 credentials = ('joe', 'password')
1353 host = None
1354 request_uri = '/projects/httplib2/test/digest/'
1355 headers = {}
1356 response = {
1357 'www-authenticate': 'Digest realm="myrealm", nonce="Ygk86AsKBAA=3516200d37f9a3230352fde99977bd6d472d4306", algorithm=MD5, qop="auth"'
1358 }
1359 content = ""
1360
jcgregorio6cbab7e2006-04-21 20:35:43 +00001361 d = httplib2.DigestAuthentication(credentials, host, request_uri, headers, response, content, None)
jcgregorio2d66d4f2006-02-07 05:34:14 +00001362 d.request("GET", request_uri, headers, content, cnonce="33033375ec278a46")
1363 our_request = "Authorization: %s" % headers['Authorization']
1364 working_request = 'Authorization: Digest username="joe", realm="myrealm", nonce="Ygk86AsKBAA=3516200d37f9a3230352fde99977bd6d472d4306", uri="/projects/httplib2/test/digest/", algorithm=MD5, response="97ed129401f7cdc60e5db58a80f3ea8b", qop=auth, nc=00000001, cnonce="33033375ec278a46"'
1365 self.assertEqual(our_request, working_request)
1366
1367
1368 def testDigestObjectStale(self):
1369 credentials = ('joe', 'password')
1370 host = None
1371 request_uri = '/projects/httplib2/test/digest/'
1372 headers = {}
1373 response = httplib2.Response({ })
1374 response['www-authenticate'] = 'Digest realm="myrealm", nonce="Ygk86AsKBAA=3516200d37f9a3230352fde99977bd6d472d4306", algorithm=MD5, qop="auth", stale=true'
1375 response.status = 401
1376 content = ""
jcgregorio6cbab7e2006-04-21 20:35:43 +00001377 d = httplib2.DigestAuthentication(credentials, host, request_uri, headers, response, content, None)
jcgregorio2d66d4f2006-02-07 05:34:14 +00001378 # Returns true to force a retry
1379 self.assertTrue( d.response(response, content) )
1380
1381 def testDigestObjectAuthInfo(self):
1382 credentials = ('joe', 'password')
1383 host = None
1384 request_uri = '/projects/httplib2/test/digest/'
1385 headers = {}
1386 response = httplib2.Response({ })
1387 response['www-authenticate'] = 'Digest realm="myrealm", nonce="Ygk86AsKBAA=3516200d37f9a3230352fde99977bd6d472d4306", algorithm=MD5, qop="auth", stale=true'
1388 response['authentication-info'] = 'nextnonce="fred"'
1389 content = ""
jcgregorio6cbab7e2006-04-21 20:35:43 +00001390 d = httplib2.DigestAuthentication(credentials, host, request_uri, headers, response, content, None)
jcgregorio2d66d4f2006-02-07 05:34:14 +00001391 # Returns true to force a retry
1392 self.assertFalse( d.response(response, content) )
1393 self.assertEqual('fred', d.challenge['nonce'])
1394 self.assertEqual(1, d.challenge['nc'])
1395
1396 def testWsseAlgorithm(self):
1397 digest = httplib2._wsse_username_token("d36e316282959a9ed4c89851497a717f", "2003-12-15T14:43:07Z", "taadtaadpstcsm")
1398 expected = "quR/EWLAV4xLf9Zqyw4pDmfV9OY="
1399 self.assertEqual(expected, digest)
1400
jcgregoriodb8dfc82006-03-31 14:59:46 +00001401 def testEnd2End(self):
1402 # one end to end header
1403 response = {'content-type': 'application/atom+xml', 'te': 'deflate'}
1404 end2end = httplib2._get_end2end_headers(response)
1405 self.assertTrue('content-type' in end2end)
1406 self.assertTrue('te' not in end2end)
1407 self.assertTrue('connection' not in end2end)
1408
1409 # one end to end header that gets eliminated
1410 response = {'connection': 'content-type', 'content-type': 'application/atom+xml', 'te': 'deflate'}
1411 end2end = httplib2._get_end2end_headers(response)
1412 self.assertTrue('content-type' not in end2end)
1413 self.assertTrue('te' not in end2end)
1414 self.assertTrue('connection' not in end2end)
1415
1416 # Degenerate case of no headers
1417 response = {}
1418 end2end = httplib2._get_end2end_headers(response)
1419 self.assertEquals(0, len(end2end))
1420
1421 # Degenerate case of connection referrring to a header not passed in
1422 response = {'connection': 'content-type'}
1423 end2end = httplib2._get_end2end_headers(response)
1424 self.assertEquals(0, len(end2end))
jcgregorio2d66d4f2006-02-07 05:34:14 +00001425
chris dent89f15142009-12-24 14:02:57 -06001426if __name__ == '__main__':
1427 unittest.main()