httplib2test.py 72 KB

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