httplib2test.py 75 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675
  1. #!/usr/bin/env python2.4
  2. """
  3. httplib2test
  4. A set of unit tests for httplib2.py.
  5. Requires Python 2.4 or later
  6. """
  7. __author__ = "Joe Gregorio (joe@bitworking.org)"
  8. __copyright__ = "Copyright 2006, Joe Gregorio"
  9. __contributors__ = []
  10. __license__ = "MIT"
  11. __history__ = """ """
  12. __version__ = "0.1 ($Rev: 118 $)"
  13. import StringIO
  14. import base64
  15. import httplib
  16. import httplib2
  17. import os
  18. import pickle
  19. import socket
  20. import sys
  21. import time
  22. import unittest
  23. import urlparse
  24. try:
  25. import ssl
  26. except ImportError:
  27. pass
  28. # Python 2.3 support
  29. if not hasattr(unittest.TestCase, 'assertTrue'):
  30. unittest.TestCase.assertTrue = unittest.TestCase.failUnless
  31. unittest.TestCase.assertFalse = unittest.TestCase.failIf
  32. # The test resources base uri
  33. base = 'http://bitworking.org/projects/httplib2/test/'
  34. #base = 'http://localhost/projects/httplib2/test/'
  35. cacheDirName = ".cache"
  36. class CredentialsTest(unittest.TestCase):
  37. def test(self):
  38. c = httplib2.Credentials()
  39. c.add("joe", "password")
  40. self.assertEqual(("joe", "password"), list(c.iter("bitworking.org"))[0])
  41. self.assertEqual(("joe", "password"), list(c.iter(""))[0])
  42. c.add("fred", "password2", "wellformedweb.org")
  43. self.assertEqual(("joe", "password"), list(c.iter("bitworking.org"))[0])
  44. self.assertEqual(1, len(list(c.iter("bitworking.org"))))
  45. self.assertEqual(2, len(list(c.iter("wellformedweb.org"))))
  46. self.assertTrue(("fred", "password2") in list(c.iter("wellformedweb.org")))
  47. c.clear()
  48. self.assertEqual(0, len(list(c.iter("bitworking.org"))))
  49. c.add("fred", "password2", "wellformedweb.org")
  50. self.assertTrue(("fred", "password2") in list(c.iter("wellformedweb.org")))
  51. self.assertEqual(0, len(list(c.iter("bitworking.org"))))
  52. self.assertEqual(0, len(list(c.iter(""))))
  53. class ParserTest(unittest.TestCase):
  54. def testFromStd66(self):
  55. self.assertEqual( ('http', 'example.com', '', None, None ), httplib2.parse_uri("http://example.com"))
  56. self.assertEqual( ('https', 'example.com', '', None, None ), httplib2.parse_uri("https://example.com"))
  57. self.assertEqual( ('https', 'example.com:8080', '', None, None ), httplib2.parse_uri("https://example.com:8080"))
  58. self.assertEqual( ('http', 'example.com', '/', None, None ), httplib2.parse_uri("http://example.com/"))
  59. self.assertEqual( ('http', 'example.com', '/path', None, None ), httplib2.parse_uri("http://example.com/path"))
  60. self.assertEqual( ('http', 'example.com', '/path', 'a=1&b=2', None ), httplib2.parse_uri("http://example.com/path?a=1&b=2"))
  61. self.assertEqual( ('http', 'example.com', '/path', 'a=1&b=2', 'fred' ), httplib2.parse_uri("http://example.com/path?a=1&b=2#fred"))
  62. self.assertEqual( ('http', 'example.com', '/path', 'a=1&b=2', 'fred' ), httplib2.parse_uri("http://example.com/path?a=1&b=2#fred"))
  63. class UrlNormTest(unittest.TestCase):
  64. def test(self):
  65. self.assertEqual( "http://example.org/", httplib2.urlnorm("http://example.org")[-1])
  66. self.assertEqual( "http://example.org/", httplib2.urlnorm("http://EXAMple.org")[-1])
  67. self.assertEqual( "http://example.org/?=b", httplib2.urlnorm("http://EXAMple.org?=b")[-1])
  68. self.assertEqual( "http://example.org/mypath?a=b", httplib2.urlnorm("http://EXAMple.org/mypath?a=b")[-1])
  69. self.assertEqual( "http://localhost:80/", httplib2.urlnorm("http://localhost:80")[-1])
  70. self.assertEqual( httplib2.urlnorm("http://localhost:80/"), httplib2.urlnorm("HTTP://LOCALHOST:80"))
  71. try:
  72. httplib2.urlnorm("/")
  73. self.fail("Non-absolute URIs should raise an exception")
  74. except httplib2.RelativeURIError:
  75. pass
  76. class UrlSafenameTest(unittest.TestCase):
  77. def test(self):
  78. # Test that different URIs end up generating different safe names
  79. self.assertEqual( "example.org,fred,a=b,58489f63a7a83c3b7794a6a398ee8b1f", httplib2.safename("http://example.org/fred/?a=b"))
  80. self.assertEqual( "example.org,fred,a=b,8c5946d56fec453071f43329ff0be46b", httplib2.safename("http://example.org/fred?/a=b"))
  81. self.assertEqual( "www.example.org,fred,a=b,499c44b8d844a011b67ea2c015116968", httplib2.safename("http://www.example.org/fred?/a=b"))
  82. self.assertEqual( httplib2.safename(httplib2.urlnorm("http://www")[-1]), httplib2.safename(httplib2.urlnorm("http://WWW")[-1]))
  83. self.assertEqual( "www.example.org,fred,a=b,692e843a333484ce0095b070497ab45d", httplib2.safename("https://www.example.org/fred?/a=b"))
  84. self.assertNotEqual( httplib2.safename("http://www"), httplib2.safename("https://www"))
  85. # Test the max length limits
  86. uri = "http://" + ("w" * 200) + ".org"
  87. uri2 = "http://" + ("w" * 201) + ".org"
  88. self.assertNotEqual( httplib2.safename(uri2), httplib2.safename(uri))
  89. # Max length should be 200 + 1 (",") + 32
  90. self.assertEqual(233, len(httplib2.safename(uri2)))
  91. self.assertEqual(233, len(httplib2.safename(uri)))
  92. # Unicode
  93. if sys.version_info >= (2,3):
  94. self.assertEqual( "xn--http,-4y1d.org,fred,a=b,579924c35db315e5a32e3d9963388193", httplib2.safename(u"http://\u2304.org/fred/?a=b"))
  95. class _MyResponse(StringIO.StringIO):
  96. def __init__(self, body, **kwargs):
  97. StringIO.StringIO.__init__(self, body)
  98. self.headers = kwargs
  99. def iteritems(self):
  100. return self.headers.iteritems()
  101. class _MyHTTPConnection(object):
  102. "This class is just a mock of httplib.HTTPConnection used for testing"
  103. def __init__(self, host, port=None, key_file=None, cert_file=None,
  104. strict=None, timeout=None, proxy_info=None):
  105. self.host = host
  106. self.port = port
  107. self.timeout = timeout
  108. self.log = ""
  109. self.sock = None
  110. def set_debuglevel(self, level):
  111. pass
  112. def connect(self):
  113. "Connect to a host on a given port."
  114. pass
  115. def close(self):
  116. pass
  117. def request(self, method, request_uri, body, headers):
  118. pass
  119. def getresponse(self):
  120. return _MyResponse("the body", status="200")
  121. class HttpTest(unittest.TestCase):
  122. def setUp(self):
  123. if os.path.exists(cacheDirName):
  124. [os.remove(os.path.join(cacheDirName, file)) for file in os.listdir(cacheDirName)]
  125. if sys.version_info < (2, 6):
  126. disable_cert_validation = True
  127. else:
  128. disable_cert_validation = False
  129. self.http = httplib2.Http(
  130. cacheDirName,
  131. disable_ssl_certificate_validation=disable_cert_validation)
  132. self.http.clear_credentials()
  133. def testIPv6NoSSL(self):
  134. try:
  135. self.http.request("http://[::1]/")
  136. except socket.gaierror:
  137. self.fail("should get the address family right for IPv6")
  138. except socket.error:
  139. # Even if IPv6 isn't installed on a machine it should just raise socket.error
  140. pass
  141. def testIPv6SSL(self):
  142. try:
  143. self.http.request("https://[::1]/")
  144. except socket.gaierror:
  145. self.fail("should get the address family right for IPv6")
  146. except httplib2.CertificateHostnameMismatch:
  147. # We connected and verified that the certificate doesn't match
  148. # the name. Good enough.
  149. pass
  150. except socket.error:
  151. # Even if IPv6 isn't installed on a machine it should just raise socket.error
  152. pass
  153. def testConnectionType(self):
  154. self.http.force_exception_to_status_code = False
  155. response, content = self.http.request("http://bitworking.org", connection_type=_MyHTTPConnection)
  156. self.assertEqual(response['content-location'], "http://bitworking.org")
  157. self.assertEqual(content, "the body")
  158. def testGetUnknownServer(self):
  159. self.http.force_exception_to_status_code = False
  160. try:
  161. self.http.request("http://fred.bitworking.org/")
  162. self.fail("An httplib2.ServerNotFoundError Exception must be thrown on an unresolvable server.")
  163. except httplib2.ServerNotFoundError:
  164. pass
  165. # Now test with exceptions turned off
  166. self.http.force_exception_to_status_code = True
  167. (response, content) = self.http.request("http://fred.bitworking.org/")
  168. self.assertEqual(response['content-type'], 'text/plain')
  169. self.assertTrue(content.startswith("Unable to find"))
  170. self.assertEqual(response.status, 400)
  171. def testGetConnectionRefused(self):
  172. self.http.force_exception_to_status_code = False
  173. try:
  174. self.http.request("http://localhost:7777/")
  175. self.fail("An socket.error exception must be thrown on Connection Refused.")
  176. except socket.error:
  177. pass
  178. # Now test with exceptions turned off
  179. self.http.force_exception_to_status_code = True
  180. (response, content) = self.http.request("http://localhost:7777/")
  181. self.assertEqual(response['content-type'], 'text/plain')
  182. self.assertTrue("Connection refused" in content
  183. or "actively refused" in content,
  184. "Unexpected status %(content)s" % vars())
  185. self.assertEqual(response.status, 400)
  186. def testGetIRI(self):
  187. 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)
  193. def testGetIsDefaultMethod(self):
  194. # Test that GET is the default method
  195. uri = urlparse.urljoin(base, "methods/method_reflector.cgi")
  196. (response, content) = self.http.request(uri)
  197. self.assertEqual(response['x-method'], "GET")
  198. def testDifferentMethods(self):
  199. # Test that all methods can be used
  200. uri = urlparse.urljoin(base, "methods/method_reflector.cgi")
  201. for method in ["GET", "PUT", "DELETE", "POST"]:
  202. (response, content) = self.http.request(uri, method, body=" ")
  203. self.assertEqual(response['x-method'], method)
  204. def testHeadRead(self):
  205. # Test that we don't try to read the response of a HEAD request
  206. # since httplib blocks response.read() for HEAD requests.
  207. # Oddly enough this doesn't appear as a problem when doing HEAD requests
  208. # against Apache servers.
  209. uri = "http://www.google.com/"
  210. (response, content) = self.http.request(uri, "HEAD")
  211. self.assertEqual(response.status, 200)
  212. self.assertEqual(content, "")
  213. def testGetNoCache(self):
  214. # Test that can do a GET w/o the cache turned on.
  215. http = httplib2.Http()
  216. uri = urlparse.urljoin(base, "304/test_etag.txt")
  217. (response, content) = http.request(uri, "GET")
  218. self.assertEqual(response.status, 200)
  219. self.assertEqual(response.previous, None)
  220. def testGetOnlyIfCachedCacheHit(self):
  221. # Test that can do a GET with cache and 'only-if-cached'
  222. uri = urlparse.urljoin(base, "304/test_etag.txt")
  223. (response, content) = self.http.request(uri, "GET")
  224. (response, content) = self.http.request(uri, "GET", headers={'cache-control': 'only-if-cached'})
  225. self.assertEqual(response.fromcache, True)
  226. self.assertEqual(response.status, 200)
  227. def testGetOnlyIfCachedCacheMiss(self):
  228. # Test that can do a GET with no cache with 'only-if-cached'
  229. uri = urlparse.urljoin(base, "304/test_etag.txt")
  230. (response, content) = self.http.request(uri, "GET", headers={'cache-control': 'only-if-cached'})
  231. self.assertEqual(response.fromcache, False)
  232. self.assertEqual(response.status, 504)
  233. def testGetOnlyIfCachedNoCacheAtAll(self):
  234. # Test that can do a GET with no cache with 'only-if-cached'
  235. # Of course, there might be an intermediary beyond us
  236. # that responds to the 'only-if-cached', so this
  237. # test can't really be guaranteed to pass.
  238. http = httplib2.Http()
  239. uri = urlparse.urljoin(base, "304/test_etag.txt")
  240. (response, content) = http.request(uri, "GET", headers={'cache-control': 'only-if-cached'})
  241. self.assertEqual(response.fromcache, False)
  242. self.assertEqual(response.status, 504)
  243. def testUserAgent(self):
  244. # Test that we provide a default user-agent
  245. uri = urlparse.urljoin(base, "user-agent/test.cgi")
  246. (response, content) = self.http.request(uri, "GET")
  247. self.assertEqual(response.status, 200)
  248. self.assertTrue(content.startswith("Python-httplib2/"))
  249. def testUserAgentNonDefault(self):
  250. # Test that the default user-agent can be over-ridden
  251. uri = urlparse.urljoin(base, "user-agent/test.cgi")
  252. (response, content) = self.http.request(uri, "GET", headers={'User-Agent': 'fred/1.0'})
  253. self.assertEqual(response.status, 200)
  254. self.assertTrue(content.startswith("fred/1.0"))
  255. def testGet300WithLocation(self):
  256. # Test the we automatically follow 300 redirects if a Location: header is provided
  257. uri = urlparse.urljoin(base, "300/with-location-header.asis")
  258. (response, content) = self.http.request(uri, "GET")
  259. self.assertEqual(response.status, 200)
  260. self.assertEqual(content, "This is the final destination.\n")
  261. self.assertEqual(response.previous.status, 300)
  262. self.assertEqual(response.previous.fromcache, False)
  263. # Confirm that the intermediate 300 is not cached
  264. (response, content) = self.http.request(uri, "GET")
  265. self.assertEqual(response.status, 200)
  266. self.assertEqual(content, "This is the final destination.\n")
  267. self.assertEqual(response.previous.status, 300)
  268. self.assertEqual(response.previous.fromcache, False)
  269. def testGet300WithLocationNoRedirect(self):
  270. # Test the we automatically follow 300 redirects if a Location: header is provided
  271. self.http.follow_redirects = False
  272. uri = urlparse.urljoin(base, "300/with-location-header.asis")
  273. (response, content) = self.http.request(uri, "GET")
  274. self.assertEqual(response.status, 300)
  275. def testGet300WithoutLocation(self):
  276. # Not giving a Location: header in a 300 response is acceptable
  277. # In which case we just return the 300 response
  278. uri = urlparse.urljoin(base, "300/without-location-header.asis")
  279. (response, content) = self.http.request(uri, "GET")
  280. self.assertEqual(response.status, 300)
  281. self.assertTrue(response['content-type'].startswith("text/html"))
  282. self.assertEqual(response.previous, None)
  283. def testGet301(self):
  284. # Test that we automatically follow 301 redirects
  285. # and that we cache the 301 response
  286. uri = urlparse.urljoin(base, "301/onestep.asis")
  287. destination = urlparse.urljoin(base, "302/final-destination.txt")
  288. (response, content) = self.http.request(uri, "GET")
  289. self.assertEqual(response.status, 200)
  290. self.assertTrue(response.has_key('content-location'))
  291. self.assertEqual(response['content-location'], destination)
  292. self.assertEqual(content, "This is the final destination.\n")
  293. self.assertEqual(response.previous.status, 301)
  294. self.assertEqual(response.previous.fromcache, False)
  295. (response, content) = self.http.request(uri, "GET")
  296. self.assertEqual(response.status, 200)
  297. self.assertEqual(response['content-location'], destination)
  298. self.assertEqual(content, "This is the final destination.\n")
  299. self.assertEqual(response.previous.status, 301)
  300. self.assertEqual(response.previous.fromcache, True)
  301. def testHead301(self):
  302. # Test that we automatically follow 301 redirects
  303. uri = urlparse.urljoin(base, "301/onestep.asis")
  304. destination = urlparse.urljoin(base, "302/final-destination.txt")
  305. (response, content) = self.http.request(uri, "HEAD")
  306. self.assertEqual(response.status, 200)
  307. self.assertEqual(response.previous.status, 301)
  308. self.assertEqual(response.previous.fromcache, False)
  309. def testGet301NoRedirect(self):
  310. # Test that we automatically follow 301 redirects
  311. # and that we cache the 301 response
  312. self.http.follow_redirects = False
  313. uri = urlparse.urljoin(base, "301/onestep.asis")
  314. destination = urlparse.urljoin(base, "302/final-destination.txt")
  315. (response, content) = self.http.request(uri, "GET")
  316. self.assertEqual(response.status, 301)
  317. def testGet302(self):
  318. # Test that we automatically follow 302 redirects
  319. # and that we DO NOT cache the 302 response
  320. uri = urlparse.urljoin(base, "302/onestep.asis")
  321. destination = urlparse.urljoin(base, "302/final-destination.txt")
  322. (response, content) = self.http.request(uri, "GET")
  323. self.assertEqual(response.status, 200)
  324. self.assertEqual(response['content-location'], destination)
  325. self.assertEqual(content, "This is the final destination.\n")
  326. self.assertEqual(response.previous.status, 302)
  327. self.assertEqual(response.previous.fromcache, False)
  328. uri = urlparse.urljoin(base, "302/onestep.asis")
  329. (response, content) = self.http.request(uri, "GET")
  330. self.assertEqual(response.status, 200)
  331. self.assertEqual(response.fromcache, True)
  332. self.assertEqual(response['content-location'], destination)
  333. self.assertEqual(content, "This is the final destination.\n")
  334. self.assertEqual(response.previous.status, 302)
  335. self.assertEqual(response.previous.fromcache, False)
  336. self.assertEqual(response.previous['content-location'], uri)
  337. uri = urlparse.urljoin(base, "302/twostep.asis")
  338. (response, content) = self.http.request(uri, "GET")
  339. self.assertEqual(response.status, 200)
  340. self.assertEqual(response.fromcache, True)
  341. self.assertEqual(content, "This is the final destination.\n")
  342. self.assertEqual(response.previous.status, 302)
  343. self.assertEqual(response.previous.fromcache, False)
  344. def testGet302RedirectionLimit(self):
  345. # Test that we can set a lower redirection limit
  346. # and that we raise an exception when we exceed
  347. # that limit.
  348. self.http.force_exception_to_status_code = False
  349. uri = urlparse.urljoin(base, "302/twostep.asis")
  350. try:
  351. (response, content) = self.http.request(uri, "GET", redirections = 1)
  352. self.fail("This should not happen")
  353. except httplib2.RedirectLimit:
  354. pass
  355. except Exception, e:
  356. self.fail("Threw wrong kind of exception ")
  357. # Re-run the test with out the exceptions
  358. self.http.force_exception_to_status_code = True
  359. (response, content) = self.http.request(uri, "GET", redirections = 1)
  360. self.assertEqual(response.status, 500)
  361. self.assertTrue(response.reason.startswith("Redirected more"))
  362. self.assertEqual("302", response['status'])
  363. self.assertTrue(content.startswith("<html>"))
  364. self.assertTrue(response.previous != None)
  365. def testGet302NoLocation(self):
  366. # Test that we throw an exception when we get
  367. # a 302 with no Location: header.
  368. self.http.force_exception_to_status_code = False
  369. uri = urlparse.urljoin(base, "302/no-location.asis")
  370. try:
  371. (response, content) = self.http.request(uri, "GET")
  372. self.fail("Should never reach here")
  373. except httplib2.RedirectMissingLocation:
  374. pass
  375. except Exception, e:
  376. self.fail("Threw wrong kind of exception ")
  377. # Re-run the test with out the exceptions
  378. self.http.force_exception_to_status_code = True
  379. (response, content) = self.http.request(uri, "GET")
  380. self.assertEqual(response.status, 500)
  381. self.assertTrue(response.reason.startswith("Redirected but"))
  382. self.assertEqual("302", response['status'])
  383. self.assertTrue(content.startswith("This is content"))
  384. def testGet301ViaHttps(self):
  385. # Google always redirects to https://www.google.com
  386. (response, content) = self.http.request("https://code.google.com/apis/", "GET")
  387. self.assertEqual(200, response.status)
  388. self.assertEqual(301, response.previous.status)
  389. def testGetViaHttps(self):
  390. # Test that we can handle HTTPS
  391. (response, content) = self.http.request("https://www.google.com/adsense/", "GET")
  392. self.assertEqual(200, response.status)
  393. def testGetViaHttpsSpecViolationOnLocation(self):
  394. # Test that we follow redirects through HTTPS
  395. # even if they violate the spec by including
  396. # a relative Location: header instead of an
  397. # absolute one.
  398. (response, content) = self.http.request("https://www.google.com/adsense", "GET")
  399. self.assertEqual(200, response.status)
  400. self.assertNotEqual(None, response.previous)
  401. def testSslCertValidation(self):
  402. if sys.version_info >= (2, 6):
  403. # Test that we get an ssl.SSLError when specifying a non-existent CA
  404. # certs file.
  405. http = httplib2.Http(ca_certs='/nosuchfile')
  406. self.assertRaises(ssl.SSLError,
  407. http.request, "https://www.google.com/", "GET")
  408. # Test that we get a SSLHandshakeError if we try to access
  409. # https;//www.google.com, using a CA cert file that doesn't contain
  410. # the CA Gogole uses (i.e., simulating a cert that's not signed by a
  411. # trusted CA).
  412. other_ca_certs = os.path.join(
  413. os.path.dirname(os.path.abspath(httplib2.__file__ )),
  414. "test", "other_cacerts.txt")
  415. http = httplib2.Http(ca_certs=other_ca_certs)
  416. self.assertRaises(httplib2.SSLHandshakeError,
  417. http.request, "https://www.google.com/", "GET")
  418. def testSslCertValidationDoubleDots(self):
  419. if sys.version_info >= (2, 6):
  420. # Test that we get match a double dot cert
  421. try:
  422. self.http.request("https://1.www.appspot.com/", "GET")
  423. except httplib2.CertificateHostnameMismatch:
  424. self.fail('cert with *.*.appspot.com should not raise an exception.')
  425. def testSslHostnameValidation(self):
  426. pass
  427. # No longer a valid test.
  428. #if sys.version_info >= (2, 6):
  429. # The SSL server at google.com:443 returns a certificate for
  430. # 'www.google.com', which results in a host name mismatch.
  431. # Note that this test only works because the ssl module and httplib2
  432. # do not support SNI; for requests specifying a server name of
  433. # 'google.com' via SNI, a matching cert would be returned.
  434. # self.assertRaises(httplib2.CertificateHostnameMismatch,
  435. # self.http.request, "https://google.com/", "GET")
  436. def testSslCertValidationWithoutSslModuleFails(self):
  437. if sys.version_info < (2, 6):
  438. http = httplib2.Http(disable_ssl_certificate_validation=False)
  439. self.assertRaises(httplib2.CertificateValidationUnsupported,
  440. http.request, "https://www.google.com/", "GET")
  441. def testGetViaHttpsKeyCert(self):
  442. # At this point I can only test
  443. # that the key and cert files are passed in
  444. # correctly to httplib. It would be nice to have
  445. # a real https endpoint to test against.
  446. # bitworking.org presents an certificate for a non-matching host
  447. # (*.webfaction.com), so we need to disable cert checking for this test.
  448. http = httplib2.Http(timeout=2, disable_ssl_certificate_validation=True)
  449. http.add_certificate("akeyfile", "acertfile", "bitworking.org")
  450. try:
  451. (response, content) = http.request("https://bitworking.org", "GET")
  452. except:
  453. pass
  454. self.assertEqual(http.connections["https:bitworking.org"].key_file, "akeyfile")
  455. self.assertEqual(http.connections["https:bitworking.org"].cert_file, "acertfile")
  456. try:
  457. (response, content) = http.request("https://notthere.bitworking.org", "GET")
  458. except:
  459. pass
  460. self.assertEqual(http.connections["https:notthere.bitworking.org"].key_file, None)
  461. self.assertEqual(http.connections["https:notthere.bitworking.org"].cert_file, None)
  462. def testGet303(self):
  463. # Do a follow-up GET on a Location: header
  464. # returned from a POST that gave a 303.
  465. uri = urlparse.urljoin(base, "303/303.cgi")
  466. (response, content) = self.http.request(uri, "POST", " ")
  467. self.assertEqual(response.status, 200)
  468. self.assertEqual(content, "This is the final destination.\n")
  469. self.assertEqual(response.previous.status, 303)
  470. def testGet303NoRedirect(self):
  471. # Do a follow-up GET on a Location: header
  472. # returned from a POST that gave a 303.
  473. self.http.follow_redirects = False
  474. uri = urlparse.urljoin(base, "303/303.cgi")
  475. (response, content) = self.http.request(uri, "POST", " ")
  476. self.assertEqual(response.status, 303)
  477. def test303ForDifferentMethods(self):
  478. # Test that all methods can be used
  479. uri = urlparse.urljoin(base, "303/redirect-to-reflector.cgi")
  480. for (method, method_on_303) in [("PUT", "GET"), ("DELETE", "GET"), ("POST", "GET"), ("GET", "GET"), ("HEAD", "GET")]:
  481. (response, content) = self.http.request(uri, method, body=" ")
  482. self.assertEqual(response['x-method'], method_on_303)
  483. def test303AndForwardAuthorizationHeader(self):
  484. # Test that all methods can be used
  485. uri = urlparse.urljoin(base, "303/redirect-to-header-reflector.cgi")
  486. headers = {'authorization': 'Bearer foo'}
  487. response, content = self.http.request(uri, 'GET', body=" ",
  488. headers=headers)
  489. # self.assertTrue('authorization' not in content)
  490. self.http.follow_all_redirects = True
  491. self.http.forward_authorization_headers = True
  492. response, content = self.http.request(uri, 'GET', body=" ",
  493. headers=headers)
  494. # Oh, how I wish Apache didn't eat the Authorization header.
  495. # self.assertTrue('authorization' in content)
  496. def testGet304(self):
  497. # Test that we use ETags properly to validate our cache
  498. uri = urlparse.urljoin(base, "304/test_etag.txt")
  499. (response, content) = self.http.request(uri, "GET")
  500. self.assertNotEqual(response['etag'], "")
  501. (response, content) = self.http.request(uri, "GET")
  502. (response, content) = self.http.request(uri, "GET", headers = {'cache-control': 'must-revalidate'})
  503. self.assertEqual(response.status, 200)
  504. self.assertEqual(response.fromcache, True)
  505. cache_file_name = os.path.join(cacheDirName, httplib2.safename(httplib2.urlnorm(uri)[-1]))
  506. f = open(cache_file_name, "r")
  507. status_line = f.readline()
  508. f.close()
  509. self.assertTrue(status_line.startswith("status:"))
  510. (response, content) = self.http.request(uri, "HEAD")
  511. self.assertEqual(response.status, 200)
  512. self.assertEqual(response.fromcache, True)
  513. (response, content) = self.http.request(uri, "GET", headers = {'range': 'bytes=0-0'})
  514. self.assertEqual(response.status, 206)
  515. self.assertEqual(response.fromcache, False)
  516. def testGetIgnoreEtag(self):
  517. # Test that we can forcibly ignore ETags
  518. uri = urlparse.urljoin(base, "reflector/reflector.cgi")
  519. (response, content) = self.http.request(uri, "GET")
  520. self.assertNotEqual(response['etag'], "")
  521. (response, content) = self.http.request(uri, "GET", headers = {'cache-control': 'max-age=0'})
  522. d = self.reflector(content)
  523. self.assertTrue(d.has_key('HTTP_IF_NONE_MATCH'))
  524. self.http.ignore_etag = True
  525. (response, content) = self.http.request(uri, "GET", headers = {'cache-control': 'max-age=0'})
  526. d = self.reflector(content)
  527. self.assertEqual(response.fromcache, False)
  528. self.assertFalse(d.has_key('HTTP_IF_NONE_MATCH'))
  529. def testOverrideEtag(self):
  530. # Test that we can forcibly ignore ETags
  531. uri = urlparse.urljoin(base, "reflector/reflector.cgi")
  532. (response, content) = self.http.request(uri, "GET")
  533. self.assertNotEqual(response['etag'], "")
  534. (response, content) = self.http.request(uri, "GET", headers = {'cache-control': 'max-age=0'})
  535. d = self.reflector(content)
  536. self.assertTrue(d.has_key('HTTP_IF_NONE_MATCH'))
  537. self.assertNotEqual(d['HTTP_IF_NONE_MATCH'], "fred")
  538. (response, content) = self.http.request(uri, "GET", headers = {'cache-control': 'max-age=0', 'if-none-match': 'fred'})
  539. d = self.reflector(content)
  540. self.assertTrue(d.has_key('HTTP_IF_NONE_MATCH'))
  541. self.assertEqual(d['HTTP_IF_NONE_MATCH'], "fred")
  542. #MAP-commented this out because it consistently fails
  543. # def testGet304EndToEnd(self):
  544. # # Test that end to end headers get overwritten in the cache
  545. # uri = urlparse.urljoin(base, "304/end2end.cgi")
  546. # (response, content) = self.http.request(uri, "GET")
  547. # self.assertNotEqual(response['etag'], "")
  548. # old_date = response['date']
  549. # time.sleep(2)
  550. #
  551. # (response, content) = self.http.request(uri, "GET", headers = {'Cache-Control': 'max-age=0'})
  552. # # The response should be from the cache, but the Date: header should be updated.
  553. # new_date = response['date']
  554. # self.assertNotEqual(new_date, old_date)
  555. # self.assertEqual(response.status, 200)
  556. # self.assertEqual(response.fromcache, True)
  557. def testGet304LastModified(self):
  558. # Test that we can still handle a 304
  559. # by only using the last-modified cache validator.
  560. uri = urlparse.urljoin(base, "304/last-modified-only/last-modified-only.txt")
  561. (response, content) = self.http.request(uri, "GET")
  562. self.assertNotEqual(response['last-modified'], "")
  563. (response, content) = self.http.request(uri, "GET")
  564. (response, content) = self.http.request(uri, "GET")
  565. self.assertEqual(response.status, 200)
  566. self.assertEqual(response.fromcache, True)
  567. def testGet307(self):
  568. # Test that we do follow 307 redirects but
  569. # do not cache the 307
  570. uri = urlparse.urljoin(base, "307/onestep.asis")
  571. (response, content) = self.http.request(uri, "GET")
  572. self.assertEqual(response.status, 200)
  573. self.assertEqual(content, "This is the final destination.\n")
  574. self.assertEqual(response.previous.status, 307)
  575. self.assertEqual(response.previous.fromcache, False)
  576. (response, content) = self.http.request(uri, "GET")
  577. self.assertEqual(response.status, 200)
  578. self.assertEqual(response.fromcache, True)
  579. self.assertEqual(content, "This is the final destination.\n")
  580. self.assertEqual(response.previous.status, 307)
  581. self.assertEqual(response.previous.fromcache, False)
  582. def testGet410(self):
  583. # Test that we pass 410's through
  584. uri = urlparse.urljoin(base, "410/410.asis")
  585. (response, content) = self.http.request(uri, "GET")
  586. self.assertEqual(response.status, 410)
  587. def testVaryHeaderSimple(self):
  588. """
  589. RFC 2616 13.6
  590. When the cache receives a subsequent request whose Request-URI
  591. specifies one or more cache entries including a Vary header field,
  592. the cache MUST NOT use such a cache entry to construct a response
  593. to the new request unless all of the selecting request-headers
  594. present in the new request match the corresponding stored
  595. request-headers in the original request.
  596. """
  597. # test that the vary header is sent
  598. uri = urlparse.urljoin(base, "vary/accept.asis")
  599. (response, content) = self.http.request(uri, "GET", headers={'Accept': 'text/plain'})
  600. self.assertEqual(response.status, 200)
  601. self.assertTrue(response.has_key('vary'))
  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. # get the resource again, not from cache since Accept headers does not match
  608. (response, content) = self.http.request(uri, "GET", headers={'Accept': 'text/html'})
  609. self.assertEqual(response.status, 200)
  610. self.assertEqual(response.fromcache, False, msg="Should not be from cache")
  611. # get the resource again, without any Accept header, so again no match
  612. (response, content) = self.http.request(uri, "GET")
  613. self.assertEqual(response.status, 200)
  614. self.assertEqual(response.fromcache, False, msg="Should not be from cache")
  615. def testNoVary(self):
  616. pass
  617. # when there is no vary, a different Accept header (e.g.) should not
  618. # impact if the cache is used
  619. # test that the vary header is not sent
  620. # uri = urlparse.urljoin(base, "vary/no-vary.asis")
  621. # (response, content) = self.http.request(uri, "GET", headers={'Accept': 'text/plain'})
  622. # self.assertEqual(response.status, 200)
  623. # self.assertFalse(response.has_key('vary'))
  624. # (response, content) = self.http.request(uri, "GET", headers={'Accept': 'text/plain'})
  625. # self.assertEqual(response.status, 200)
  626. # self.assertEqual(response.fromcache, True, msg="Should be from cache")
  627. #
  628. # (response, content) = self.http.request(uri, "GET", headers={'Accept': 'text/html'})
  629. # self.assertEqual(response.status, 200)
  630. # self.assertEqual(response.fromcache, True, msg="Should be from cache")
  631. def testVaryHeaderDouble(self):
  632. uri = urlparse.urljoin(base, "vary/accept-double.asis")
  633. (response, content) = self.http.request(uri, "GET", headers={
  634. 'Accept': 'text/plain', 'Accept-Language': 'da, en-gb;q=0.8, en;q=0.7'})
  635. self.assertEqual(response.status, 200)
  636. self.assertTrue(response.has_key('vary'))
  637. # we are from cache
  638. (response, content) = self.http.request(uri, "GET", headers={
  639. 'Accept': 'text/plain', 'Accept-Language': 'da, en-gb;q=0.8, en;q=0.7'})
  640. self.assertEqual(response.fromcache, True, msg="Should be from cache")
  641. (response, content) = self.http.request(uri, "GET", headers={'Accept': 'text/plain'})
  642. self.assertEqual(response.status, 200)
  643. self.assertEqual(response.fromcache, False)
  644. # get the resource again, not from cache, varied headers don't match exact
  645. (response, content) = self.http.request(uri, "GET", headers={'Accept-Language': 'da'})
  646. self.assertEqual(response.status, 200)
  647. self.assertEqual(response.fromcache, False, msg="Should not be from cache")
  648. def testVaryUnusedHeader(self):
  649. # A header's value is not considered to vary if it's not used at all.
  650. uri = urlparse.urljoin(base, "vary/unused-header.asis")
  651. (response, content) = self.http.request(uri, "GET", headers={
  652. 'Accept': 'text/plain'})
  653. self.assertEqual(response.status, 200)
  654. self.assertTrue(response.has_key('vary'))
  655. # we are from cache
  656. (response, content) = self.http.request(uri, "GET", headers={
  657. 'Accept': 'text/plain',})
  658. self.assertEqual(response.fromcache, True, msg="Should be from cache")
  659. def testHeadGZip(self):
  660. # Test that we don't try to decompress a HEAD response
  661. uri = urlparse.urljoin(base, "gzip/final-destination.txt")
  662. (response, content) = self.http.request(uri, "HEAD")
  663. self.assertEqual(response.status, 200)
  664. self.assertNotEqual(int(response['content-length']), 0)
  665. self.assertEqual(content, "")
  666. def testGetGZip(self):
  667. # Test that we support gzip compression
  668. uri = urlparse.urljoin(base, "gzip/final-destination.txt")
  669. (response, content) = self.http.request(uri, "GET")
  670. self.assertEqual(response.status, 200)
  671. self.assertFalse(response.has_key('content-encoding'))
  672. self.assertTrue(response.has_key('-content-encoding'))
  673. self.assertEqual(int(response['content-length']), len("This is the final destination.\n"))
  674. self.assertEqual(content, "This is the final destination.\n")
  675. def testPostAndGZipResponse(self):
  676. uri = urlparse.urljoin(base, "gzip/post.cgi")
  677. (response, content) = self.http.request(uri, "POST", body=" ")
  678. self.assertEqual(response.status, 200)
  679. self.assertFalse(response.has_key('content-encoding'))
  680. self.assertTrue(response.has_key('-content-encoding'))
  681. def testGetGZipFailure(self):
  682. # Test that we raise a good exception when the gzip fails
  683. self.http.force_exception_to_status_code = False
  684. uri = urlparse.urljoin(base, "gzip/failed-compression.asis")
  685. try:
  686. (response, content) = self.http.request(uri, "GET")
  687. self.fail("Should never reach here")
  688. except httplib2.FailedToDecompressContent:
  689. pass
  690. except Exception:
  691. self.fail("Threw wrong kind of exception")
  692. # Re-run the test with out the exceptions
  693. self.http.force_exception_to_status_code = True
  694. (response, content) = self.http.request(uri, "GET")
  695. self.assertEqual(response.status, 500)
  696. self.assertTrue(response.reason.startswith("Content purported"))
  697. def testTimeout(self):
  698. self.http.force_exception_to_status_code = True
  699. uri = urlparse.urljoin(base, "timeout/timeout.cgi")
  700. try:
  701. import socket
  702. socket.setdefaulttimeout(1)
  703. except:
  704. # Don't run the test if we can't set the timeout
  705. return
  706. (response, content) = self.http.request(uri)
  707. self.assertEqual(response.status, 408)
  708. self.assertTrue(response.reason.startswith("Request Timeout"))
  709. self.assertTrue(content.startswith("Request Timeout"))
  710. def testIndividualTimeout(self):
  711. uri = urlparse.urljoin(base, "timeout/timeout.cgi")
  712. http = httplib2.Http(timeout=1)
  713. http.force_exception_to_status_code = True
  714. (response, content) = http.request(uri)
  715. self.assertEqual(response.status, 408)
  716. self.assertTrue(response.reason.startswith("Request Timeout"))
  717. self.assertTrue(content.startswith("Request Timeout"))
  718. def testHTTPSInitTimeout(self):
  719. c = httplib2.HTTPSConnectionWithTimeout('localhost', 80, timeout=47)
  720. self.assertEqual(47, c.timeout)
  721. def testGetDeflate(self):
  722. # Test that we support deflate compression
  723. uri = urlparse.urljoin(base, "deflate/deflated.asis")
  724. (response, content) = self.http.request(uri, "GET")
  725. self.assertEqual(response.status, 200)
  726. self.assertFalse(response.has_key('content-encoding'))
  727. self.assertEqual(int(response['content-length']), len("This is the final destination."))
  728. self.assertEqual(content, "This is the final destination.")
  729. def testGetDeflateFailure(self):
  730. # Test that we raise a good exception when the deflate fails
  731. self.http.force_exception_to_status_code = False
  732. uri = urlparse.urljoin(base, "deflate/failed-compression.asis")
  733. try:
  734. (response, content) = self.http.request(uri, "GET")
  735. self.fail("Should never reach here")
  736. except httplib2.FailedToDecompressContent:
  737. pass
  738. except Exception:
  739. self.fail("Threw wrong kind of exception")
  740. # Re-run the test with out the exceptions
  741. self.http.force_exception_to_status_code = True
  742. (response, content) = self.http.request(uri, "GET")
  743. self.assertEqual(response.status, 500)
  744. self.assertTrue(response.reason.startswith("Content purported"))
  745. def testGetDuplicateHeaders(self):
  746. # Test that duplicate headers get concatenated via ','
  747. uri = urlparse.urljoin(base, "duplicate-headers/multilink.asis")
  748. (response, content) = self.http.request(uri, "GET")
  749. self.assertEqual(response.status, 200)
  750. self.assertEqual(content, "This is content\n")
  751. self.assertEqual(response['link'].split(",")[0], '<http://bitworking.org>; rel="home"; title="BitWorking"')
  752. def testGetCacheControlNoCache(self):
  753. # Test Cache-Control: no-cache on requests
  754. uri = urlparse.urljoin(base, "304/test_etag.txt")
  755. (response, content) = self.http.request(uri, "GET")
  756. self.assertNotEqual(response['etag'], "")
  757. (response, content) = self.http.request(uri, "GET")
  758. self.assertEqual(response.status, 200)
  759. self.assertEqual(response.fromcache, True)
  760. (response, content) = self.http.request(uri, "GET", headers={'Cache-Control': 'no-cache'})
  761. self.assertEqual(response.status, 200)
  762. self.assertEqual(response.fromcache, False)
  763. def testGetCacheControlPragmaNoCache(self):
  764. # Test Pragma: no-cache on requests
  765. uri = urlparse.urljoin(base, "304/test_etag.txt")
  766. (response, content) = self.http.request(uri, "GET")
  767. self.assertNotEqual(response['etag'], "")
  768. (response, content) = self.http.request(uri, "GET")
  769. self.assertEqual(response.status, 200)
  770. self.assertEqual(response.fromcache, True)
  771. (response, content) = self.http.request(uri, "GET", headers={'Pragma': 'no-cache'})
  772. self.assertEqual(response.status, 200)
  773. self.assertEqual(response.fromcache, False)
  774. def testGetCacheControlNoStoreRequest(self):
  775. # A no-store request means that the response should not be stored.
  776. uri = urlparse.urljoin(base, "304/test_etag.txt")
  777. (response, content) = self.http.request(uri, "GET", headers={'Cache-Control': 'no-store'})
  778. self.assertEqual(response.status, 200)
  779. self.assertEqual(response.fromcache, False)
  780. (response, content) = self.http.request(uri, "GET", headers={'Cache-Control': 'no-store'})
  781. self.assertEqual(response.status, 200)
  782. self.assertEqual(response.fromcache, False)
  783. def testGetCacheControlNoStoreResponse(self):
  784. # A no-store response means that the response should not be stored.
  785. uri = urlparse.urljoin(base, "no-store/no-store.asis")
  786. (response, content) = self.http.request(uri, "GET")
  787. self.assertEqual(response.status, 200)
  788. self.assertEqual(response.fromcache, False)
  789. (response, content) = self.http.request(uri, "GET")
  790. self.assertEqual(response.status, 200)
  791. self.assertEqual(response.fromcache, False)
  792. def testGetCacheControlNoCacheNoStoreRequest(self):
  793. # Test that a no-store, no-cache clears the entry from the cache
  794. # even if it was cached previously.
  795. uri = urlparse.urljoin(base, "304/test_etag.txt")
  796. (response, content) = self.http.request(uri, "GET")
  797. (response, content) = self.http.request(uri, "GET")
  798. self.assertEqual(response.fromcache, True)
  799. (response, content) = self.http.request(uri, "GET", headers={'Cache-Control': 'no-store, no-cache'})
  800. (response, content) = self.http.request(uri, "GET", headers={'Cache-Control': 'no-store, no-cache'})
  801. self.assertEqual(response.status, 200)
  802. self.assertEqual(response.fromcache, False)
  803. def testUpdateInvalidatesCache(self):
  804. # Test that calling PUT or DELETE on a
  805. # URI that is cache invalidates that cache.
  806. uri = urlparse.urljoin(base, "304/test_etag.txt")
  807. (response, content) = self.http.request(uri, "GET")
  808. (response, content) = self.http.request(uri, "GET")
  809. self.assertEqual(response.fromcache, True)
  810. (response, content) = self.http.request(uri, "DELETE")
  811. self.assertEqual(response.status, 405)
  812. (response, content) = self.http.request(uri, "GET")
  813. self.assertEqual(response.fromcache, False)
  814. def testUpdateUsesCachedETag(self):
  815. # Test that we natively support http://www.w3.org/1999/04/Editing/
  816. uri = urlparse.urljoin(base, "conditional-updates/test.cgi")
  817. (response, content) = self.http.request(uri, "GET")
  818. self.assertEqual(response.status, 200)
  819. self.assertEqual(response.fromcache, False)
  820. (response, content) = self.http.request(uri, "GET")
  821. self.assertEqual(response.status, 200)
  822. self.assertEqual(response.fromcache, True)
  823. (response, content) = self.http.request(uri, "PUT", body="foo")
  824. self.assertEqual(response.status, 200)
  825. (response, content) = self.http.request(uri, "PUT", body="foo")
  826. self.assertEqual(response.status, 412)
  827. def testUpdatePatchUsesCachedETag(self):
  828. # Test that we natively support http://www.w3.org/1999/04/Editing/
  829. uri = urlparse.urljoin(base, "conditional-updates/test.cgi")
  830. (response, content) = self.http.request(uri, "GET")
  831. self.assertEqual(response.status, 200)
  832. self.assertEqual(response.fromcache, False)
  833. (response, content) = self.http.request(uri, "GET")
  834. self.assertEqual(response.status, 200)
  835. self.assertEqual(response.fromcache, True)
  836. (response, content) = self.http.request(uri, "PATCH", body="foo")
  837. self.assertEqual(response.status, 200)
  838. (response, content) = self.http.request(uri, "PATCH", body="foo")
  839. self.assertEqual(response.status, 412)
  840. def testUpdateUsesCachedETagAndOCMethod(self):
  841. # Test that we natively support http://www.w3.org/1999/04/Editing/
  842. uri = urlparse.urljoin(base, "conditional-updates/test.cgi")
  843. (response, content) = self.http.request(uri, "GET")
  844. self.assertEqual(response.status, 200)
  845. self.assertEqual(response.fromcache, False)
  846. (response, content) = self.http.request(uri, "GET")
  847. self.assertEqual(response.status, 200)
  848. self.assertEqual(response.fromcache, True)
  849. self.http.optimistic_concurrency_methods.append("DELETE")
  850. (response, content) = self.http.request(uri, "DELETE")
  851. self.assertEqual(response.status, 200)
  852. def testUpdateUsesCachedETagOverridden(self):
  853. # Test that we natively support http://www.w3.org/1999/04/Editing/
  854. uri = urlparse.urljoin(base, "conditional-updates/test.cgi")
  855. (response, content) = self.http.request(uri, "GET")
  856. self.assertEqual(response.status, 200)
  857. self.assertEqual(response.fromcache, False)
  858. (response, content) = self.http.request(uri, "GET")
  859. self.assertEqual(response.status, 200)
  860. self.assertEqual(response.fromcache, True)
  861. (response, content) = self.http.request(uri, "PUT", body="foo", headers={'if-match': 'fred'})
  862. self.assertEqual(response.status, 412)
  863. def testBasicAuth(self):
  864. # Test Basic Authentication
  865. uri = urlparse.urljoin(base, "basic/file.txt")
  866. (response, content) = self.http.request(uri, "GET")
  867. self.assertEqual(response.status, 401)
  868. uri = urlparse.urljoin(base, "basic/")
  869. (response, content) = self.http.request(uri, "GET")
  870. self.assertEqual(response.status, 401)
  871. self.http.add_credentials('joe', 'password')
  872. (response, content) = self.http.request(uri, "GET")
  873. self.assertEqual(response.status, 200)
  874. uri = urlparse.urljoin(base, "basic/file.txt")
  875. (response, content) = self.http.request(uri, "GET")
  876. self.assertEqual(response.status, 200)
  877. def testBasicAuthWithDomain(self):
  878. # Test Basic Authentication
  879. uri = urlparse.urljoin(base, "basic/file.txt")
  880. (response, content) = self.http.request(uri, "GET")
  881. self.assertEqual(response.status, 401)
  882. uri = urlparse.urljoin(base, "basic/")
  883. (response, content) = self.http.request(uri, "GET")
  884. self.assertEqual(response.status, 401)
  885. self.http.add_credentials('joe', 'password', "example.org")
  886. (response, content) = self.http.request(uri, "GET")
  887. self.assertEqual(response.status, 401)
  888. uri = urlparse.urljoin(base, "basic/file.txt")
  889. (response, content) = self.http.request(uri, "GET")
  890. self.assertEqual(response.status, 401)
  891. domain = urlparse.urlparse(base)[1]
  892. self.http.add_credentials('joe', 'password', domain)
  893. (response, content) = self.http.request(uri, "GET")
  894. self.assertEqual(response.status, 200)
  895. uri = urlparse.urljoin(base, "basic/file.txt")
  896. (response, content) = self.http.request(uri, "GET")
  897. self.assertEqual(response.status, 200)
  898. def testBasicAuthTwoDifferentCredentials(self):
  899. # Test Basic Authentication with multiple sets of credentials
  900. uri = urlparse.urljoin(base, "basic2/file.txt")
  901. (response, content) = self.http.request(uri, "GET")
  902. self.assertEqual(response.status, 401)
  903. uri = urlparse.urljoin(base, "basic2/")
  904. (response, content) = self.http.request(uri, "GET")
  905. self.assertEqual(response.status, 401)
  906. self.http.add_credentials('fred', 'barney')
  907. (response, content) = self.http.request(uri, "GET")
  908. self.assertEqual(response.status, 200)
  909. uri = urlparse.urljoin(base, "basic2/file.txt")
  910. (response, content) = self.http.request(uri, "GET")
  911. self.assertEqual(response.status, 200)
  912. def testBasicAuthNested(self):
  913. # Test Basic Authentication with resources
  914. # that are nested
  915. uri = urlparse.urljoin(base, "basic-nested/")
  916. (response, content) = self.http.request(uri, "GET")
  917. self.assertEqual(response.status, 401)
  918. uri = urlparse.urljoin(base, "basic-nested/subdir")
  919. (response, content) = self.http.request(uri, "GET")
  920. self.assertEqual(response.status, 401)
  921. # Now add in credentials one at a time and test.
  922. self.http.add_credentials('joe', 'password')
  923. uri = urlparse.urljoin(base, "basic-nested/")
  924. (response, content) = self.http.request(uri, "GET")
  925. self.assertEqual(response.status, 200)
  926. uri = urlparse.urljoin(base, "basic-nested/subdir")
  927. (response, content) = self.http.request(uri, "GET")
  928. self.assertEqual(response.status, 401)
  929. self.http.add_credentials('fred', 'barney')
  930. uri = urlparse.urljoin(base, "basic-nested/")
  931. (response, content) = self.http.request(uri, "GET")
  932. self.assertEqual(response.status, 200)
  933. uri = urlparse.urljoin(base, "basic-nested/subdir")
  934. (response, content) = self.http.request(uri, "GET")
  935. self.assertEqual(response.status, 200)
  936. def testDigestAuth(self):
  937. # Test that we support Digest Authentication
  938. uri = urlparse.urljoin(base, "digest/")
  939. (response, content) = self.http.request(uri, "GET")
  940. self.assertEqual(response.status, 401)
  941. self.http.add_credentials('joe', 'password')
  942. (response, content) = self.http.request(uri, "GET")
  943. self.assertEqual(response.status, 200)
  944. uri = urlparse.urljoin(base, "digest/file.txt")
  945. (response, content) = self.http.request(uri, "GET")
  946. def testDigestAuthNextNonceAndNC(self):
  947. # Test that if the server sets nextnonce that we reset
  948. # the nonce count back to 1
  949. uri = urlparse.urljoin(base, "digest/file.txt")
  950. self.http.add_credentials('joe', 'password')
  951. (response, content) = self.http.request(uri, "GET", headers = {"cache-control":"no-cache"})
  952. info = httplib2._parse_www_authenticate(response, 'authentication-info')
  953. self.assertEqual(response.status, 200)
  954. (response, content) = self.http.request(uri, "GET", headers = {"cache-control":"no-cache"})
  955. info2 = httplib2._parse_www_authenticate(response, 'authentication-info')
  956. self.assertEqual(response.status, 200)
  957. if info.has_key('nextnonce'):
  958. self.assertEqual(info2['nc'], 1)
  959. def testDigestAuthStale(self):
  960. # Test that we can handle a nonce becoming stale
  961. uri = urlparse.urljoin(base, "digest-expire/file.txt")
  962. self.http.add_credentials('joe', 'password')
  963. (response, content) = self.http.request(uri, "GET", headers = {"cache-control":"no-cache"})
  964. info = httplib2._parse_www_authenticate(response, 'authentication-info')
  965. self.assertEqual(response.status, 200)
  966. time.sleep(3)
  967. # Sleep long enough that the nonce becomes stale
  968. (response, content) = self.http.request(uri, "GET", headers = {"cache-control":"no-cache"})
  969. self.assertFalse(response.fromcache)
  970. self.assertTrue(response._stale_digest)
  971. info3 = httplib2._parse_www_authenticate(response, 'authentication-info')
  972. self.assertEqual(response.status, 200)
  973. def reflector(self, content):
  974. return dict( [tuple(x.split("=", 1)) for x in content.strip().split("\n")] )
  975. def testReflector(self):
  976. uri = urlparse.urljoin(base, "reflector/reflector.cgi")
  977. (response, content) = self.http.request(uri, "GET")
  978. d = self.reflector(content)
  979. self.assertTrue(d.has_key('HTTP_USER_AGENT'))
  980. def testConnectionClose(self):
  981. uri = "http://www.google.com/"
  982. (response, content) = self.http.request(uri, "GET")
  983. for c in self.http.connections.values():
  984. self.assertNotEqual(None, c.sock)
  985. (response, content) = self.http.request(uri, "GET", headers={"connection": "close"})
  986. for c in self.http.connections.values():
  987. self.assertEqual(None, c.sock)
  988. def testPickleHttp(self):
  989. pickled_http = pickle.dumps(self.http)
  990. new_http = pickle.loads(pickled_http)
  991. self.assertEqual(sorted(new_http.__dict__.keys()),
  992. sorted(self.http.__dict__.keys()))
  993. for key in new_http.__dict__:
  994. if key in ('certificates', 'credentials'):
  995. self.assertEqual(new_http.__dict__[key].credentials,
  996. self.http.__dict__[key].credentials)
  997. elif key == 'cache':
  998. self.assertEqual(new_http.__dict__[key].cache,
  999. self.http.__dict__[key].cache)
  1000. else:
  1001. self.assertEqual(new_http.__dict__[key],
  1002. self.http.__dict__[key])
  1003. def testPickleHttpWithConnection(self):
  1004. self.http.request('http://bitworking.org',
  1005. connection_type=_MyHTTPConnection)
  1006. pickled_http = pickle.dumps(self.http)
  1007. new_http = pickle.loads(pickled_http)
  1008. self.assertEqual(self.http.connections.keys(), ['http:bitworking.org'])
  1009. self.assertEqual(new_http.connections, {})
  1010. def testPickleCustomRequestHttp(self):
  1011. def dummy_request(*args, **kwargs):
  1012. return new_request(*args, **kwargs)
  1013. dummy_request.dummy_attr = 'dummy_value'
  1014. self.http.request = dummy_request
  1015. pickled_http = pickle.dumps(self.http)
  1016. self.assertFalse("S'request'" in pickled_http)
  1017. try:
  1018. import memcache
  1019. class HttpTestMemCached(HttpTest):
  1020. def setUp(self):
  1021. self.cache = memcache.Client(['127.0.0.1:11211'], debug=0)
  1022. #self.cache = memcache.Client(['10.0.0.4:11211'], debug=1)
  1023. self.http = httplib2.Http(self.cache)
  1024. self.cache.flush_all()
  1025. # Not exactly sure why the sleep is needed here, but
  1026. # if not present then some unit tests that rely on caching
  1027. # fail. Memcached seems to lose some sets immediately
  1028. # after a flush_all if the set is to a value that
  1029. # was previously cached. (Maybe the flush is handled async?)
  1030. time.sleep(1)
  1031. self.http.clear_credentials()
  1032. except:
  1033. pass
  1034. # ------------------------------------------------------------------------
  1035. class HttpPrivateTest(unittest.TestCase):
  1036. def testParseCacheControl(self):
  1037. # Test that we can parse the Cache-Control header
  1038. self.assertEqual({}, httplib2._parse_cache_control({}))
  1039. self.assertEqual({'no-cache': 1}, httplib2._parse_cache_control({'cache-control': ' no-cache'}))
  1040. cc = httplib2._parse_cache_control({'cache-control': ' no-cache, max-age = 7200'})
  1041. self.assertEqual(cc['no-cache'], 1)
  1042. self.assertEqual(cc['max-age'], '7200')
  1043. cc = httplib2._parse_cache_control({'cache-control': ' , '})
  1044. self.assertEqual(cc[''], 1)
  1045. try:
  1046. cc = httplib2._parse_cache_control({'cache-control': 'Max-age=3600;post-check=1800,pre-check=3600'})
  1047. self.assertTrue("max-age" in cc)
  1048. except:
  1049. self.fail("Should not throw exception")
  1050. def testNormalizeHeaders(self):
  1051. # Test that we normalize headers to lowercase
  1052. h = httplib2._normalize_headers({'Cache-Control': 'no-cache', 'Other': 'Stuff'})
  1053. self.assertTrue(h.has_key('cache-control'))
  1054. self.assertTrue(h.has_key('other'))
  1055. self.assertEqual('Stuff', h['other'])
  1056. def testExpirationModelTransparent(self):
  1057. # Test that no-cache makes our request TRANSPARENT
  1058. response_headers = {
  1059. 'cache-control': 'max-age=7200'
  1060. }
  1061. request_headers = {
  1062. 'cache-control': 'no-cache'
  1063. }
  1064. self.assertEqual("TRANSPARENT", httplib2._entry_disposition(response_headers, request_headers))
  1065. def testMaxAgeNonNumeric(self):
  1066. # Test that no-cache makes our request TRANSPARENT
  1067. response_headers = {
  1068. 'cache-control': 'max-age=fred, min-fresh=barney'
  1069. }
  1070. request_headers = {
  1071. }
  1072. self.assertEqual("STALE", httplib2._entry_disposition(response_headers, request_headers))
  1073. def testExpirationModelNoCacheResponse(self):
  1074. # The date and expires point to an entry that should be
  1075. # FRESH, but the no-cache over-rides that.
  1076. now = time.time()
  1077. response_headers = {
  1078. 'date': time.strftime("%a, %d %b %Y %H:%M:%S GMT", time.gmtime(now)),
  1079. 'expires': time.strftime("%a, %d %b %Y %H:%M:%S GMT", time.gmtime(now+4)),
  1080. 'cache-control': 'no-cache'
  1081. }
  1082. request_headers = {
  1083. }
  1084. self.assertEqual("STALE", httplib2._entry_disposition(response_headers, request_headers))
  1085. def testExpirationModelStaleRequestMustReval(self):
  1086. # must-revalidate forces STALE
  1087. self.assertEqual("STALE", httplib2._entry_disposition({}, {'cache-control': 'must-revalidate'}))
  1088. def testExpirationModelStaleResponseMustReval(self):
  1089. # must-revalidate forces STALE
  1090. self.assertEqual("STALE", httplib2._entry_disposition({'cache-control': 'must-revalidate'}, {}))
  1091. def testExpirationModelFresh(self):
  1092. response_headers = {
  1093. 'date': time.strftime("%a, %d %b %Y %H:%M:%S GMT", time.gmtime()),
  1094. 'cache-control': 'max-age=2'
  1095. }
  1096. request_headers = {
  1097. }
  1098. self.assertEqual("FRESH", httplib2._entry_disposition(response_headers, request_headers))
  1099. time.sleep(3)
  1100. self.assertEqual("STALE", httplib2._entry_disposition(response_headers, request_headers))
  1101. def testExpirationMaxAge0(self):
  1102. response_headers = {
  1103. 'date': time.strftime("%a, %d %b %Y %H:%M:%S GMT", time.gmtime()),
  1104. 'cache-control': 'max-age=0'
  1105. }
  1106. request_headers = {
  1107. }
  1108. self.assertEqual("STALE", httplib2._entry_disposition(response_headers, request_headers))
  1109. def testExpirationModelDateAndExpires(self):
  1110. now = time.time()
  1111. response_headers = {
  1112. 'date': time.strftime("%a, %d %b %Y %H:%M:%S GMT", time.gmtime(now)),
  1113. 'expires': time.strftime("%a, %d %b %Y %H:%M:%S GMT", time.gmtime(now+2)),
  1114. }
  1115. request_headers = {
  1116. }
  1117. self.assertEqual("FRESH", httplib2._entry_disposition(response_headers, request_headers))
  1118. time.sleep(3)
  1119. self.assertEqual("STALE", httplib2._entry_disposition(response_headers, request_headers))
  1120. def testExpiresZero(self):
  1121. now = time.time()
  1122. response_headers = {
  1123. 'date': time.strftime("%a, %d %b %Y %H:%M:%S GMT", time.gmtime(now)),
  1124. 'expires': "0",
  1125. }
  1126. request_headers = {
  1127. }
  1128. self.assertEqual("STALE", httplib2._entry_disposition(response_headers, request_headers))
  1129. def testExpirationModelDateOnly(self):
  1130. now = time.time()
  1131. response_headers = {
  1132. 'date': time.strftime("%a, %d %b %Y %H:%M:%S GMT", time.gmtime(now+3)),
  1133. }
  1134. request_headers = {
  1135. }
  1136. self.assertEqual("STALE", httplib2._entry_disposition(response_headers, request_headers))
  1137. def testExpirationModelOnlyIfCached(self):
  1138. response_headers = {
  1139. }
  1140. request_headers = {
  1141. 'cache-control': 'only-if-cached',
  1142. }
  1143. self.assertEqual("FRESH", httplib2._entry_disposition(response_headers, request_headers))
  1144. def testExpirationModelMaxAgeBoth(self):
  1145. now = time.time()
  1146. response_headers = {
  1147. 'date': time.strftime("%a, %d %b %Y %H:%M:%S GMT", time.gmtime(now)),
  1148. 'cache-control': 'max-age=2'
  1149. }
  1150. request_headers = {
  1151. 'cache-control': 'max-age=0'
  1152. }
  1153. self.assertEqual("STALE", httplib2._entry_disposition(response_headers, request_headers))
  1154. def testExpirationModelDateAndExpiresMinFresh1(self):
  1155. now = time.time()
  1156. response_headers = {
  1157. 'date': time.strftime("%a, %d %b %Y %H:%M:%S GMT", time.gmtime(now)),
  1158. 'expires': time.strftime("%a, %d %b %Y %H:%M:%S GMT", time.gmtime(now+2)),
  1159. }
  1160. request_headers = {
  1161. 'cache-control': 'min-fresh=2'
  1162. }
  1163. self.assertEqual("STALE", httplib2._entry_disposition(response_headers, request_headers))
  1164. def testExpirationModelDateAndExpiresMinFresh2(self):
  1165. now = time.time()
  1166. response_headers = {
  1167. 'date': time.strftime("%a, %d %b %Y %H:%M:%S GMT", time.gmtime(now)),
  1168. 'expires': time.strftime("%a, %d %b %Y %H:%M:%S GMT", time.gmtime(now+4)),
  1169. }
  1170. request_headers = {
  1171. 'cache-control': 'min-fresh=2'
  1172. }
  1173. self.assertEqual("FRESH", httplib2._entry_disposition(response_headers, request_headers))
  1174. def testParseWWWAuthenticateEmpty(self):
  1175. res = httplib2._parse_www_authenticate({})
  1176. self.assertEqual(len(res.keys()), 0)
  1177. def testParseWWWAuthenticate(self):
  1178. # different uses of spaces around commas
  1179. res = httplib2._parse_www_authenticate({ 'www-authenticate': 'Test realm="test realm" , foo=foo ,bar="bar", baz=baz,qux=qux'})
  1180. self.assertEqual(len(res.keys()), 1)
  1181. self.assertEqual(len(res['test'].keys()), 5)
  1182. # tokens with non-alphanum
  1183. res = httplib2._parse_www_authenticate({ 'www-authenticate': 'T*!%#st realm=to*!%#en, to*!%#en="quoted string"'})
  1184. self.assertEqual(len(res.keys()), 1)
  1185. self.assertEqual(len(res['t*!%#st'].keys()), 2)
  1186. # quoted string with quoted pairs
  1187. res = httplib2._parse_www_authenticate({ 'www-authenticate': 'Test realm="a \\"test\\" realm"'})
  1188. self.assertEqual(len(res.keys()), 1)
  1189. self.assertEqual(res['test']['realm'], 'a "test" realm')
  1190. def testParseWWWAuthenticateStrict(self):
  1191. httplib2.USE_WWW_AUTH_STRICT_PARSING = 1;
  1192. self.testParseWWWAuthenticate();
  1193. httplib2.USE_WWW_AUTH_STRICT_PARSING = 0;
  1194. def testParseWWWAuthenticateBasic(self):
  1195. res = httplib2._parse_www_authenticate({ 'www-authenticate': 'Basic realm="me"'})
  1196. basic = res['basic']
  1197. self.assertEqual('me', basic['realm'])
  1198. res = httplib2._parse_www_authenticate({ 'www-authenticate': 'Basic realm="me", algorithm="MD5"'})
  1199. basic = res['basic']
  1200. self.assertEqual('me', basic['realm'])
  1201. self.assertEqual('MD5', basic['algorithm'])
  1202. res = httplib2._parse_www_authenticate({ 'www-authenticate': 'Basic realm="me", algorithm=MD5'})
  1203. basic = res['basic']
  1204. self.assertEqual('me', basic['realm'])
  1205. self.assertEqual('MD5', basic['algorithm'])
  1206. def testParseWWWAuthenticateBasic2(self):
  1207. res = httplib2._parse_www_authenticate({ 'www-authenticate': 'Basic realm="me",other="fred" '})
  1208. basic = res['basic']
  1209. self.assertEqual('me', basic['realm'])
  1210. self.assertEqual('fred', basic['other'])
  1211. def testParseWWWAuthenticateBasic3(self):
  1212. res = httplib2._parse_www_authenticate({ 'www-authenticate': 'Basic REAlm="me" '})
  1213. basic = res['basic']
  1214. self.assertEqual('me', basic['realm'])
  1215. def testParseWWWAuthenticateDigest(self):
  1216. res = httplib2._parse_www_authenticate({ 'www-authenticate':
  1217. 'Digest realm="testrealm@host.com", qop="auth,auth-int", nonce="dcd98b7102dd2f0e8b11d0f600bfb0c093", opaque="5ccc069c403ebaf9f0171e9517f40e41"'})
  1218. digest = res['digest']
  1219. self.assertEqual('testrealm@host.com', digest['realm'])
  1220. self.assertEqual('auth,auth-int', digest['qop'])
  1221. def testParseWWWAuthenticateMultiple(self):
  1222. res = httplib2._parse_www_authenticate({ 'www-authenticate':
  1223. 'Digest realm="testrealm@host.com", qop="auth,auth-int", nonce="dcd98b7102dd2f0e8b11d0f600bfb0c093", opaque="5ccc069c403ebaf9f0171e9517f40e41" Basic REAlm="me" '})
  1224. digest = res['digest']
  1225. self.assertEqual('testrealm@host.com', digest['realm'])
  1226. self.assertEqual('auth,auth-int', digest['qop'])
  1227. self.assertEqual('dcd98b7102dd2f0e8b11d0f600bfb0c093', digest['nonce'])
  1228. self.assertEqual('5ccc069c403ebaf9f0171e9517f40e41', digest['opaque'])
  1229. basic = res['basic']
  1230. self.assertEqual('me', basic['realm'])
  1231. def testParseWWWAuthenticateMultiple2(self):
  1232. # Handle an added comma between challenges, which might get thrown in if the challenges were
  1233. # originally sent in separate www-authenticate headers.
  1234. res = httplib2._parse_www_authenticate({ 'www-authenticate':
  1235. 'Digest realm="testrealm@host.com", qop="auth,auth-int", nonce="dcd98b7102dd2f0e8b11d0f600bfb0c093", opaque="5ccc069c403ebaf9f0171e9517f40e41", Basic REAlm="me" '})
  1236. digest = res['digest']
  1237. self.assertEqual('testrealm@host.com', digest['realm'])
  1238. self.assertEqual('auth,auth-int', digest['qop'])
  1239. self.assertEqual('dcd98b7102dd2f0e8b11d0f600bfb0c093', digest['nonce'])
  1240. self.assertEqual('5ccc069c403ebaf9f0171e9517f40e41', digest['opaque'])
  1241. basic = res['basic']
  1242. self.assertEqual('me', basic['realm'])
  1243. def testParseWWWAuthenticateMultiple3(self):
  1244. # Handle an added comma between challenges, which might get thrown in if the challenges were
  1245. # originally sent in separate www-authenticate headers.
  1246. res = httplib2._parse_www_authenticate({ 'www-authenticate':
  1247. 'Digest realm="testrealm@host.com", qop="auth,auth-int", nonce="dcd98b7102dd2f0e8b11d0f600bfb0c093", opaque="5ccc069c403ebaf9f0171e9517f40e41", Basic REAlm="me", WSSE realm="foo", profile="UsernameToken"'})
  1248. digest = res['digest']
  1249. self.assertEqual('testrealm@host.com', digest['realm'])
  1250. self.assertEqual('auth,auth-int', digest['qop'])
  1251. self.assertEqual('dcd98b7102dd2f0e8b11d0f600bfb0c093', digest['nonce'])
  1252. self.assertEqual('5ccc069c403ebaf9f0171e9517f40e41', digest['opaque'])
  1253. basic = res['basic']
  1254. self.assertEqual('me', basic['realm'])
  1255. wsse = res['wsse']
  1256. self.assertEqual('foo', wsse['realm'])
  1257. self.assertEqual('UsernameToken', wsse['profile'])
  1258. def testParseWWWAuthenticateMultiple4(self):
  1259. res = httplib2._parse_www_authenticate({ 'www-authenticate':
  1260. 'Digest realm="test-real.m@host.com", qop \t=\t"\tauth,auth-int", nonce="(*)&^&$%#",opaque="5ccc069c403ebaf9f0171e9517f40e41", Basic REAlm="me", WSSE realm="foo", profile="UsernameToken"'})
  1261. digest = res['digest']
  1262. self.assertEqual('test-real.m@host.com', digest['realm'])
  1263. self.assertEqual('\tauth,auth-int', digest['qop'])
  1264. self.assertEqual('(*)&^&$%#', digest['nonce'])
  1265. def testParseWWWAuthenticateMoreQuoteCombos(self):
  1266. res = httplib2._parse_www_authenticate({'www-authenticate':'Digest realm="myrealm", nonce="Ygk86AsKBAA=3516200d37f9a3230352fde99977bd6d472d4306", algorithm=MD5, qop="auth", stale=true'})
  1267. digest = res['digest']
  1268. self.assertEqual('myrealm', digest['realm'])
  1269. def testParseWWWAuthenticateMalformed(self):
  1270. try:
  1271. res = httplib2._parse_www_authenticate({'www-authenticate':'OAuth "Facebook Platform" "invalid_token" "Invalid OAuth access token."'})
  1272. self.fail("should raise an exception")
  1273. except httplib2.MalformedHeader:
  1274. pass
  1275. def testDigestObject(self):
  1276. credentials = ('joe', 'password')
  1277. host = None
  1278. request_uri = '/projects/httplib2/test/digest/'
  1279. headers = {}
  1280. response = {
  1281. 'www-authenticate': 'Digest realm="myrealm", nonce="Ygk86AsKBAA=3516200d37f9a3230352fde99977bd6d472d4306", algorithm=MD5, qop="auth"'
  1282. }
  1283. content = ""
  1284. d = httplib2.DigestAuthentication(credentials, host, request_uri, headers, response, content, None)
  1285. d.request("GET", request_uri, headers, content, cnonce="33033375ec278a46")
  1286. our_request = "authorization: %s" % headers['authorization']
  1287. 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"'
  1288. self.assertEqual(our_request, working_request)
  1289. def testDigestObjectWithOpaque(self):
  1290. credentials = ('joe', 'password')
  1291. host = None
  1292. request_uri = '/projects/httplib2/test/digest/'
  1293. headers = {}
  1294. response = {
  1295. 'www-authenticate': 'Digest realm="myrealm", nonce="Ygk86AsKBAA=3516200d37f9a3230352fde99977bd6d472d4306", algorithm=MD5, qop="auth", opaque="atestopaque"'
  1296. }
  1297. content = ""
  1298. d = httplib2.DigestAuthentication(credentials, host, request_uri, headers, response, content, None)
  1299. d.request("GET", request_uri, headers, content, cnonce="33033375ec278a46")
  1300. our_request = "authorization: %s" % headers['authorization']
  1301. 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", opaque="atestopaque"'
  1302. self.assertEqual(our_request, working_request)
  1303. def testDigestObjectStale(self):
  1304. credentials = ('joe', 'password')
  1305. host = None
  1306. request_uri = '/projects/httplib2/test/digest/'
  1307. headers = {}
  1308. response = httplib2.Response({ })
  1309. response['www-authenticate'] = 'Digest realm="myrealm", nonce="Ygk86AsKBAA=3516200d37f9a3230352fde99977bd6d472d4306", algorithm=MD5, qop="auth", stale=true'
  1310. response.status = 401
  1311. content = ""
  1312. d = httplib2.DigestAuthentication(credentials, host, request_uri, headers, response, content, None)
  1313. # Returns true to force a retry
  1314. self.assertTrue( d.response(response, content) )
  1315. def testDigestObjectAuthInfo(self):
  1316. credentials = ('joe', 'password')
  1317. host = None
  1318. request_uri = '/projects/httplib2/test/digest/'
  1319. headers = {}
  1320. response = httplib2.Response({ })
  1321. response['www-authenticate'] = 'Digest realm="myrealm", nonce="Ygk86AsKBAA=3516200d37f9a3230352fde99977bd6d472d4306", algorithm=MD5, qop="auth", stale=true'
  1322. response['authentication-info'] = 'nextnonce="fred"'
  1323. content = ""
  1324. d = httplib2.DigestAuthentication(credentials, host, request_uri, headers, response, content, None)
  1325. # Returns true to force a retry
  1326. self.assertFalse( d.response(response, content) )
  1327. self.assertEqual('fred', d.challenge['nonce'])
  1328. self.assertEqual(1, d.challenge['nc'])
  1329. def testWsseAlgorithm(self):
  1330. digest = httplib2._wsse_username_token("d36e316282959a9ed4c89851497a717f", "2003-12-15T14:43:07Z", "taadtaadpstcsm")
  1331. expected = "quR/EWLAV4xLf9Zqyw4pDmfV9OY="
  1332. self.assertEqual(expected, digest)
  1333. def testEnd2End(self):
  1334. # one end to end header
  1335. response = {'content-type': 'application/atom+xml', 'te': 'deflate'}
  1336. end2end = httplib2._get_end2end_headers(response)
  1337. self.assertTrue('content-type' in end2end)
  1338. self.assertTrue('te' not in end2end)
  1339. self.assertTrue('connection' not in end2end)
  1340. # one end to end header that gets eliminated
  1341. response = {'connection': 'content-type', 'content-type': 'application/atom+xml', 'te': 'deflate'}
  1342. end2end = httplib2._get_end2end_headers(response)
  1343. self.assertTrue('content-type' not in end2end)
  1344. self.assertTrue('te' not in end2end)
  1345. self.assertTrue('connection' not in end2end)
  1346. # Degenerate case of no headers
  1347. response = {}
  1348. end2end = httplib2._get_end2end_headers(response)
  1349. self.assertEquals(0, len(end2end))
  1350. # Degenerate case of connection referrring to a header not passed in
  1351. response = {'connection': 'content-type'}
  1352. end2end = httplib2._get_end2end_headers(response)
  1353. self.assertEquals(0, len(end2end))
  1354. class TestProxyInfo(unittest.TestCase):
  1355. def setUp(self):
  1356. self.orig_env = dict(os.environ)
  1357. def tearDown(self):
  1358. os.environ.clear()
  1359. os.environ.update(self.orig_env)
  1360. def test_from_url(self):
  1361. pi = httplib2.proxy_info_from_url('http://myproxy.example.com')
  1362. self.assertEquals(pi.proxy_host, 'myproxy.example.com')
  1363. self.assertEquals(pi.proxy_port, 80)
  1364. self.assertEquals(pi.proxy_user, None)
  1365. def test_from_url_ident(self):
  1366. pi = httplib2.proxy_info_from_url('http://zoidberg:fish@someproxy:99')
  1367. self.assertEquals(pi.proxy_host, 'someproxy')
  1368. self.assertEquals(pi.proxy_port, 99)
  1369. self.assertEquals(pi.proxy_user, 'zoidberg')
  1370. self.assertEquals(pi.proxy_pass, 'fish')
  1371. def test_from_env(self):
  1372. os.environ['http_proxy'] = 'http://myproxy.example.com:8080'
  1373. pi = httplib2.proxy_info_from_environment()
  1374. self.assertEquals(pi.proxy_host, 'myproxy.example.com')
  1375. self.assertEquals(pi.proxy_port, 8080)
  1376. self.assertEquals(pi.bypass_hosts, [])
  1377. def test_from_env_no_proxy(self):
  1378. os.environ['http_proxy'] = 'http://myproxy.example.com:80'
  1379. os.environ['https_proxy'] = 'http://myproxy.example.com:81'
  1380. os.environ['no_proxy'] = 'localhost,otherhost.domain.local'
  1381. pi = httplib2.proxy_info_from_environment('https')
  1382. self.assertEquals(pi.proxy_host, 'myproxy.example.com')
  1383. self.assertEquals(pi.proxy_port, 81)
  1384. self.assertEquals(pi.bypass_hosts, ['localhost',
  1385. 'otherhost.domain.local'])
  1386. def test_from_env_none(self):
  1387. os.environ.clear()
  1388. pi = httplib2.proxy_info_from_environment()
  1389. self.assertEquals(pi, None)
  1390. def test_applies_to(self):
  1391. os.environ['http_proxy'] = 'http://myproxy.example.com:80'
  1392. os.environ['https_proxy'] = 'http://myproxy.example.com:81'
  1393. os.environ['no_proxy'] = 'localhost,otherhost.domain.local,example.com'
  1394. pi = httplib2.proxy_info_from_environment()
  1395. self.assertFalse(pi.applies_to('localhost'))
  1396. self.assertTrue(pi.applies_to('www.google.com'))
  1397. self.assertFalse(pi.applies_to('www.example.com'))
  1398. def test_no_proxy_star(self):
  1399. os.environ['http_proxy'] = 'http://myproxy.example.com:80'
  1400. os.environ['NO_PROXY'] = '*'
  1401. pi = httplib2.proxy_info_from_environment()
  1402. for host in ('localhost', '169.254.38.192', 'www.google.com'):
  1403. self.assertFalse(pi.applies_to(host))
  1404. if __name__ == '__main__':
  1405. unittest.main()