| 12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541154215431544154515461547154815491550155115521553155415551556155715581559156015611562156315641565156615671568156915701571157215731574157515761577157815791580 |
- #!/usr/bin/env python3
- """
- httplib2test
- A set of unit tests for httplib2.py.
- Requires Python 3.0 or later
- """
- __author__ = "Joe Gregorio (joe@bitworking.org)"
- __copyright__ = "Copyright 2006, Joe Gregorio"
- __contributors__ = ["Mark Pilgrim"]
- __license__ = "MIT"
- __history__ = """ """
- __version__ = "0.2 ($Rev: 118 $)"
- import base64
- import http.client
- import httplib2
- import io
- import os
- import pickle
- import socket
- import ssl
- import sys
- import time
- import unittest
- import urllib.parse
- # The test resources base uri
- base = 'http://bitworking.org/projects/httplib2/test/'
- #base = 'http://localhost/projects/httplib2/test/'
- cacheDirName = ".cache"
- class CredentialsTest(unittest.TestCase):
- def test(self):
- c = httplib2.Credentials()
- c.add("joe", "password")
- self.assertEqual(("joe", "password"), list(c.iter("bitworking.org"))[0])
- self.assertEqual(("joe", "password"), list(c.iter(""))[0])
- c.add("fred", "password2", "wellformedweb.org")
- self.assertEqual(("joe", "password"), list(c.iter("bitworking.org"))[0])
- self.assertEqual(1, len(list(c.iter("bitworking.org"))))
- self.assertEqual(2, len(list(c.iter("wellformedweb.org"))))
- self.assertTrue(("fred", "password2") in list(c.iter("wellformedweb.org")))
- c.clear()
- self.assertEqual(0, len(list(c.iter("bitworking.org"))))
- c.add("fred", "password2", "wellformedweb.org")
- self.assertTrue(("fred", "password2") in list(c.iter("wellformedweb.org")))
- self.assertEqual(0, len(list(c.iter("bitworking.org"))))
- self.assertEqual(0, len(list(c.iter(""))))
- class ParserTest(unittest.TestCase):
- def testFromStd66(self):
- self.assertEqual( ('http', 'example.com', '', None, None ), httplib2.parse_uri("http://example.com"))
- self.assertEqual( ('https', 'example.com', '', None, None ), httplib2.parse_uri("https://example.com"))
- self.assertEqual( ('https', 'example.com:8080', '', None, None ), httplib2.parse_uri("https://example.com:8080"))
- self.assertEqual( ('http', 'example.com', '/', None, None ), httplib2.parse_uri("http://example.com/"))
- self.assertEqual( ('http', 'example.com', '/path', None, None ), httplib2.parse_uri("http://example.com/path"))
- self.assertEqual( ('http', 'example.com', '/path', 'a=1&b=2', None ), httplib2.parse_uri("http://example.com/path?a=1&b=2"))
- self.assertEqual( ('http', 'example.com', '/path', 'a=1&b=2', 'fred' ), httplib2.parse_uri("http://example.com/path?a=1&b=2#fred"))
- self.assertEqual( ('http', 'example.com', '/path', 'a=1&b=2', 'fred' ), httplib2.parse_uri("http://example.com/path?a=1&b=2#fred"))
- class UrlNormTest(unittest.TestCase):
- def test(self):
- self.assertEqual( "http://example.org/", httplib2.urlnorm("http://example.org")[-1])
- self.assertEqual( "http://example.org/", httplib2.urlnorm("http://EXAMple.org")[-1])
- self.assertEqual( "http://example.org/?=b", httplib2.urlnorm("http://EXAMple.org?=b")[-1])
- self.assertEqual( "http://example.org/mypath?a=b", httplib2.urlnorm("http://EXAMple.org/mypath?a=b")[-1])
- self.assertEqual( "http://localhost:80/", httplib2.urlnorm("http://localhost:80")[-1])
- self.assertEqual( httplib2.urlnorm("http://localhost:80/"), httplib2.urlnorm("HTTP://LOCALHOST:80"))
- try:
- httplib2.urlnorm("/")
- self.fail("Non-absolute URIs should raise an exception")
- except httplib2.RelativeURIError:
- pass
- class UrlSafenameTest(unittest.TestCase):
- def test(self):
- # Test that different URIs end up generating different safe names
- self.assertEqual( "example.org,fred,a=b,58489f63a7a83c3b7794a6a398ee8b1f", httplib2.safename("http://example.org/fred/?a=b"))
- self.assertEqual( "example.org,fred,a=b,8c5946d56fec453071f43329ff0be46b", httplib2.safename("http://example.org/fred?/a=b"))
- self.assertEqual( "www.example.org,fred,a=b,499c44b8d844a011b67ea2c015116968", httplib2.safename("http://www.example.org/fred?/a=b"))
- self.assertEqual( httplib2.safename(httplib2.urlnorm("http://www")[-1]), httplib2.safename(httplib2.urlnorm("http://WWW")[-1]))
- self.assertEqual( "www.example.org,fred,a=b,692e843a333484ce0095b070497ab45d", httplib2.safename("https://www.example.org/fred?/a=b"))
- self.assertNotEqual( httplib2.safename("http://www"), httplib2.safename("https://www"))
- # Test the max length limits
- uri = "http://" + ("w" * 200) + ".org"
- uri2 = "http://" + ("w" * 201) + ".org"
- self.assertNotEqual( httplib2.safename(uri2), httplib2.safename(uri))
- # Max length should be 200 + 1 (",") + 32
- self.assertEqual(233, len(httplib2.safename(uri2)))
- self.assertEqual(233, len(httplib2.safename(uri)))
- # Unicode
- if sys.version_info >= (2,3):
- self.assertEqual( "xn--http,-4y1d.org,fred,a=b,579924c35db315e5a32e3d9963388193", httplib2.safename("http://\u2304.org/fred/?a=b"))
- class _MyResponse(io.BytesIO):
- def __init__(self, body, **kwargs):
- io.BytesIO.__init__(self, body)
- self.headers = kwargs
- def items(self):
- return self.headers.items()
- def iteritems(self):
- return iter(self.headers.items())
- class _MyHTTPConnection(object):
- "This class is just a mock of httplib.HTTPConnection used for testing"
- def __init__(self, host, port=None, key_file=None, cert_file=None,
- strict=None, timeout=None, proxy_info=None):
- self.host = host
- self.port = port
- self.timeout = timeout
- self.log = ""
- self.sock = None
- def set_debuglevel(self, level):
- pass
- def connect(self):
- "Connect to a host on a given port."
- pass
- def close(self):
- pass
- def request(self, method, request_uri, body, headers):
- pass
- def getresponse(self):
- return _MyResponse(b"the body", status="200")
- class HttpTest(unittest.TestCase):
- def setUp(self):
- if os.path.exists(cacheDirName):
- [os.remove(os.path.join(cacheDirName, file)) for file in os.listdir(cacheDirName)]
- self.http = httplib2.Http(cacheDirName)
- self.http.clear_credentials()
- def testIPv6NoSSL(self):
- try:
- self.http.request("http://[::1]/")
- except socket.gaierror:
- self.fail("should get the address family right for IPv6")
- except socket.error:
- # Even if IPv6 isn't installed on a machine it should just raise socket.error
- pass
- def testIPv6SSL(self):
- try:
- self.http.request("https://[::1]/")
- except socket.gaierror:
- self.fail("should get the address family right for IPv6")
- except socket.error:
- # Even if IPv6 isn't installed on a machine it should just raise socket.error
- pass
- def testConnectionType(self):
- self.http.force_exception_to_status_code = False
- response, content = self.http.request("http://bitworking.org", connection_type=_MyHTTPConnection)
- self.assertEqual(response['content-location'], "http://bitworking.org")
- self.assertEqual(content, b"the body")
- def testGetUnknownServer(self):
- self.http.force_exception_to_status_code = False
- try:
- self.http.request("http://fred.bitworking.org/")
- self.fail("An httplib2.ServerNotFoundError Exception must be thrown on an unresolvable server.")
- except httplib2.ServerNotFoundError:
- pass
- # Now test with exceptions turned off
- self.http.force_exception_to_status_code = True
- (response, content) = self.http.request("http://fred.bitworking.org/")
- self.assertEqual(response['content-type'], 'text/plain')
- self.assertTrue(content.startswith(b"Unable to find"))
- self.assertEqual(response.status, 400)
- def testGetConnectionRefused(self):
- self.http.force_exception_to_status_code = False
- try:
- self.http.request("http://localhost:7777/")
- self.fail("An socket.error exception must be thrown on Connection Refused.")
- except socket.error:
- pass
- # Now test with exceptions turned off
- self.http.force_exception_to_status_code = True
- (response, content) = self.http.request("http://localhost:7777/")
- self.assertEqual(response['content-type'], 'text/plain')
- self.assertTrue(b"Connection refused" in content)
- self.assertEqual(response.status, 400)
- def testGetIRI(self):
- if sys.version_info >= (2,3):
- uri = urllib.parse.urljoin(base, "reflector/reflector.cgi?d=\N{CYRILLIC CAPITAL LETTER DJE}")
- (response, content) = self.http.request(uri, "GET")
- d = self.reflector(content)
- self.assertTrue('QUERY_STRING' in d)
- self.assertTrue(d['QUERY_STRING'].find('%D0%82') > 0)
- def testGetIsDefaultMethod(self):
- # Test that GET is the default method
- uri = urllib.parse.urljoin(base, "methods/method_reflector.cgi")
- (response, content) = self.http.request(uri)
- self.assertEqual(response['x-method'], "GET")
- def testDifferentMethods(self):
- # Test that all methods can be used
- uri = urllib.parse.urljoin(base, "methods/method_reflector.cgi")
- for method in ["GET", "PUT", "DELETE", "POST"]:
- (response, content) = self.http.request(uri, method, body=b" ")
- self.assertEqual(response['x-method'], method)
- def testHeadRead(self):
- # Test that we don't try to read the response of a HEAD request
- # since httplib blocks response.read() for HEAD requests.
- # Oddly enough this doesn't appear as a problem when doing HEAD requests
- # against Apache servers.
- uri = "http://www.google.com/"
- (response, content) = self.http.request(uri, "HEAD")
- self.assertEqual(response.status, 200)
- self.assertEqual(content, b"")
- def testGetNoCache(self):
- # Test that can do a GET w/o the cache turned on.
- http = httplib2.Http()
- uri = urllib.parse.urljoin(base, "304/test_etag.txt")
- (response, content) = http.request(uri, "GET")
- self.assertEqual(response.status, 200)
- self.assertEqual(response.previous, None)
- def testGetOnlyIfCachedCacheHit(self):
- # Test that can do a GET with cache and 'only-if-cached'
- uri = urllib.parse.urljoin(base, "304/test_etag.txt")
- (response, content) = self.http.request(uri, "GET")
- (response, content) = self.http.request(uri, "GET", headers={'cache-control': 'only-if-cached'})
- self.assertEqual(response.fromcache, True)
- self.assertEqual(response.status, 200)
- def testGetOnlyIfCachedCacheMiss(self):
- # Test that can do a GET with no cache with 'only-if-cached'
- uri = urllib.parse.urljoin(base, "304/test_etag.txt")
- (response, content) = self.http.request(uri, "GET", headers={'cache-control': 'only-if-cached'})
- self.assertEqual(response.fromcache, False)
- self.assertEqual(response.status, 504)
- def testGetOnlyIfCachedNoCacheAtAll(self):
- # Test that can do a GET with no cache with 'only-if-cached'
- # Of course, there might be an intermediary beyond us
- # that responds to the 'only-if-cached', so this
- # test can't really be guaranteed to pass.
- http = httplib2.Http()
- uri = urllib.parse.urljoin(base, "304/test_etag.txt")
- (response, content) = http.request(uri, "GET", headers={'cache-control': 'only-if-cached'})
- self.assertEqual(response.fromcache, False)
- self.assertEqual(response.status, 504)
- def testUserAgent(self):
- # Test that we provide a default user-agent
- uri = urllib.parse.urljoin(base, "user-agent/test.cgi")
- (response, content) = self.http.request(uri, "GET")
- self.assertEqual(response.status, 200)
- self.assertTrue(content.startswith(b"Python-httplib2/"))
- def testUserAgentNonDefault(self):
- # Test that the default user-agent can be over-ridden
- uri = urllib.parse.urljoin(base, "user-agent/test.cgi")
- (response, content) = self.http.request(uri, "GET", headers={'User-Agent': 'fred/1.0'})
- self.assertEqual(response.status, 200)
- self.assertTrue(content.startswith(b"fred/1.0"))
- def testGet300WithLocation(self):
- # Test the we automatically follow 300 redirects if a Location: header is provided
- uri = urllib.parse.urljoin(base, "300/with-location-header.asis")
- (response, content) = self.http.request(uri, "GET")
- self.assertEqual(response.status, 200)
- self.assertEqual(content, b"This is the final destination.\n")
- self.assertEqual(response.previous.status, 300)
- self.assertEqual(response.previous.fromcache, False)
- # Confirm that the intermediate 300 is not cached
- (response, content) = self.http.request(uri, "GET")
- self.assertEqual(response.status, 200)
- self.assertEqual(content, b"This is the final destination.\n")
- self.assertEqual(response.previous.status, 300)
- self.assertEqual(response.previous.fromcache, False)
- def testGet300WithLocationNoRedirect(self):
- # Test the we automatically follow 300 redirects if a Location: header is provided
- self.http.follow_redirects = False
- uri = urllib.parse.urljoin(base, "300/with-location-header.asis")
- (response, content) = self.http.request(uri, "GET")
- self.assertEqual(response.status, 300)
- def testGet300WithoutLocation(self):
- # Not giving a Location: header in a 300 response is acceptable
- # In which case we just return the 300 response
- uri = urllib.parse.urljoin(base, "300/without-location-header.asis")
- (response, content) = self.http.request(uri, "GET")
- self.assertEqual(response.status, 300)
- self.assertTrue(response['content-type'].startswith("text/html"))
- self.assertEqual(response.previous, None)
- def testGet301(self):
- # Test that we automatically follow 301 redirects
- # and that we cache the 301 response
- uri = urllib.parse.urljoin(base, "301/onestep.asis")
- destination = urllib.parse.urljoin(base, "302/final-destination.txt")
- (response, content) = self.http.request(uri, "GET")
- self.assertEqual(response.status, 200)
- self.assertTrue('content-location' in response)
- self.assertEqual(response['content-location'], destination)
- self.assertEqual(content, b"This is the final destination.\n")
- self.assertEqual(response.previous.status, 301)
- self.assertEqual(response.previous.fromcache, False)
- (response, content) = self.http.request(uri, "GET")
- self.assertEqual(response.status, 200)
- self.assertEqual(response['content-location'], destination)
- self.assertEqual(content, b"This is the final destination.\n")
- self.assertEqual(response.previous.status, 301)
- self.assertEqual(response.previous.fromcache, True)
- def testHead301(self):
- # Test that we automatically follow 301 redirects
- uri = urllib.parse.urljoin(base, "301/onestep.asis")
- (response, content) = self.http.request(uri, "HEAD")
- self.assertEqual(response.status, 200)
- self.assertEqual(response.previous.status, 301)
- self.assertEqual(response.previous.fromcache, False)
- def testGet301NoRedirect(self):
- # Test that we automatically follow 301 redirects
- # and that we cache the 301 response
- self.http.follow_redirects = False
- uri = urllib.parse.urljoin(base, "301/onestep.asis")
- destination = urllib.parse.urljoin(base, "302/final-destination.txt")
- (response, content) = self.http.request(uri, "GET")
- self.assertEqual(response.status, 301)
- def testGet302(self):
- # Test that we automatically follow 302 redirects
- # and that we DO NOT cache the 302 response
- uri = urllib.parse.urljoin(base, "302/onestep.asis")
- destination = urllib.parse.urljoin(base, "302/final-destination.txt")
- (response, content) = self.http.request(uri, "GET")
- self.assertEqual(response.status, 200)
- self.assertEqual(response['content-location'], destination)
- self.assertEqual(content, b"This is the final destination.\n")
- self.assertEqual(response.previous.status, 302)
- self.assertEqual(response.previous.fromcache, False)
- uri = urllib.parse.urljoin(base, "302/onestep.asis")
- (response, content) = self.http.request(uri, "GET")
- self.assertEqual(response.status, 200)
- self.assertEqual(response.fromcache, True)
- self.assertEqual(response['content-location'], destination)
- self.assertEqual(content, b"This is the final destination.\n")
- self.assertEqual(response.previous.status, 302)
- self.assertEqual(response.previous.fromcache, False)
- self.assertEqual(response.previous['content-location'], uri)
- uri = urllib.parse.urljoin(base, "302/twostep.asis")
- (response, content) = self.http.request(uri, "GET")
- self.assertEqual(response.status, 200)
- self.assertEqual(response.fromcache, True)
- self.assertEqual(content, b"This is the final destination.\n")
- self.assertEqual(response.previous.status, 302)
- self.assertEqual(response.previous.fromcache, False)
- def testGet302RedirectionLimit(self):
- # Test that we can set a lower redirection limit
- # and that we raise an exception when we exceed
- # that limit.
- self.http.force_exception_to_status_code = False
- uri = urllib.parse.urljoin(base, "302/twostep.asis")
- try:
- (response, content) = self.http.request(uri, "GET", redirections = 1)
- self.fail("This should not happen")
- except httplib2.RedirectLimit:
- pass
- except Exception as e:
- self.fail("Threw wrong kind of exception ")
- # Re-run the test with out the exceptions
- self.http.force_exception_to_status_code = True
- (response, content) = self.http.request(uri, "GET", redirections = 1)
- self.assertEqual(response.status, 500)
- self.assertTrue(response.reason.startswith("Redirected more"))
- self.assertEqual("302", response['status'])
- self.assertTrue(content.startswith(b"<html>"))
- self.assertTrue(response.previous != None)
- def testGet302NoLocation(self):
- # Test that we throw an exception when we get
- # a 302 with no Location: header.
- self.http.force_exception_to_status_code = False
- uri = urllib.parse.urljoin(base, "302/no-location.asis")
- try:
- (response, content) = self.http.request(uri, "GET")
- self.fail("Should never reach here")
- except httplib2.RedirectMissingLocation:
- pass
- except Exception as e:
- self.fail("Threw wrong kind of exception ")
- # Re-run the test with out the exceptions
- self.http.force_exception_to_status_code = True
- (response, content) = self.http.request(uri, "GET")
- self.assertEqual(response.status, 500)
- self.assertTrue(response.reason.startswith("Redirected but"))
- self.assertEqual("302", response['status'])
- self.assertTrue(content.startswith(b"This is content"))
- def testGet301ViaHttps(self):
- # Google always redirects to http://google.com
- (response, content) = self.http.request("https://code.google.com/apis/", "GET")
- self.assertEqual(200, response.status)
- self.assertEqual(301, response.previous.status)
- def testGetViaHttps(self):
- # Test that we can handle HTTPS
- (response, content) = self.http.request("https://google.com/adsense/", "GET")
- self.assertEqual(200, response.status)
- def testGetViaHttpsSpecViolationOnLocation(self):
- # Test that we follow redirects through HTTPS
- # even if they violate the spec by including
- # a relative Location: header instead of an
- # absolute one.
- (response, content) = self.http.request("https://google.com/adsense", "GET")
- self.assertEqual(200, response.status)
- self.assertNotEqual(None, response.previous)
- def testGetViaHttpsKeyCert(self):
- # At this point I can only test
- # that the key and cert files are passed in
- # correctly to httplib. It would be nice to have
- # a real https endpoint to test against.
- http = httplib2.Http(timeout=2)
- http.add_certificate("akeyfile", "acertfile", "bitworking.org")
- try:
- (response, content) = http.request("https://bitworking.org", "GET")
- except AttributeError:
- self.assertEqual(http.connections["https:bitworking.org"].key_file, "akeyfile")
- self.assertEqual(http.connections["https:bitworking.org"].cert_file, "acertfile")
- except IOError:
- # Skip on 3.2
- pass
- try:
- (response, content) = http.request("https://notthere.bitworking.org", "GET")
- except httplib2.ServerNotFoundError:
- self.assertEqual(http.connections["https:notthere.bitworking.org"].key_file, None)
- self.assertEqual(http.connections["https:notthere.bitworking.org"].cert_file, None)
- except IOError:
- # Skip on 3.2
- pass
- def testSslCertValidation(self):
- # Test that we get an ssl.SSLError when specifying a non-existent CA
- # certs file.
- http = httplib2.Http(ca_certs='/nosuchfile')
- self.assertRaises(IOError,
- http.request, "https://www.google.com/", "GET")
- # Test that we get a SSLHandshakeError if we try to access
- # https://www.google.com, using a CA cert file that doesn't contain
- # the CA Gogole uses (i.e., simulating a cert that's not signed by a
- # trusted CA).
- other_ca_certs = os.path.join(
- os.path.dirname(os.path.abspath(httplib2.__file__ )),
- "test", "other_cacerts.txt")
- http = httplib2.Http(ca_certs=other_ca_certs)
- self.assertRaises(ssl.SSLError,
- http.request,"https://www.google.com/", "GET")
- def testSniHostnameValidation(self):
- self.http.request("https://google.com/", method="GET")
- def testGet303(self):
- # Do a follow-up GET on a Location: header
- # returned from a POST that gave a 303.
- uri = urllib.parse.urljoin(base, "303/303.cgi")
- (response, content) = self.http.request(uri, "POST", " ")
- self.assertEqual(response.status, 200)
- self.assertEqual(content, b"This is the final destination.\n")
- self.assertEqual(response.previous.status, 303)
- def testGet303NoRedirect(self):
- # Do a follow-up GET on a Location: header
- # returned from a POST that gave a 303.
- self.http.follow_redirects = False
- uri = urllib.parse.urljoin(base, "303/303.cgi")
- (response, content) = self.http.request(uri, "POST", " ")
- self.assertEqual(response.status, 303)
- def test303ForDifferentMethods(self):
- # Test that all methods can be used
- uri = urllib.parse.urljoin(base, "303/redirect-to-reflector.cgi")
- for (method, method_on_303) in [("PUT", "GET"), ("DELETE", "GET"), ("POST", "GET"), ("GET", "GET"), ("HEAD", "GET")]:
- (response, content) = self.http.request(uri, method, body=b" ")
- self.assertEqual(response['x-method'], method_on_303)
- def testGet304(self):
- # Test that we use ETags properly to validate our cache
- uri = urllib.parse.urljoin(base, "304/test_etag.txt")
- (response, content) = self.http.request(uri, "GET")
- self.assertNotEqual(response['etag'], "")
- (response, content) = self.http.request(uri, "GET")
- (response, content) = self.http.request(uri, "GET", headers = {'cache-control': 'must-revalidate'})
- self.assertEqual(response.status, 200)
- self.assertEqual(response.fromcache, True)
- cache_file_name = os.path.join(cacheDirName, httplib2.safename(httplib2.urlnorm(uri)[-1]))
- f = open(cache_file_name, "r")
- status_line = f.readline()
- f.close()
- self.assertTrue(status_line.startswith("status:"))
- (response, content) = self.http.request(uri, "HEAD")
- self.assertEqual(response.status, 200)
- self.assertEqual(response.fromcache, True)
- (response, content) = self.http.request(uri, "GET", headers = {'range': 'bytes=0-0'})
- self.assertEqual(response.status, 206)
- self.assertEqual(response.fromcache, False)
- def testGetIgnoreEtag(self):
- # Test that we can forcibly ignore ETags
- uri = urllib.parse.urljoin(base, "reflector/reflector.cgi")
- (response, content) = self.http.request(uri, "GET")
- self.assertNotEqual(response['etag'], "")
- (response, content) = self.http.request(uri, "GET", headers = {'cache-control': 'max-age=0'})
- d = self.reflector(content)
- self.assertTrue('HTTP_IF_NONE_MATCH' in d)
- self.http.ignore_etag = True
- (response, content) = self.http.request(uri, "GET", headers = {'cache-control': 'max-age=0'})
- d = self.reflector(content)
- self.assertEqual(response.fromcache, False)
- self.assertFalse('HTTP_IF_NONE_MATCH' in d)
- def testOverrideEtag(self):
- # Test that we can forcibly ignore ETags
- uri = urllib.parse.urljoin(base, "reflector/reflector.cgi")
- (response, content) = self.http.request(uri, "GET")
- self.assertNotEqual(response['etag'], "")
- (response, content) = self.http.request(uri, "GET", headers = {'cache-control': 'max-age=0'})
- d = self.reflector(content)
- self.assertTrue('HTTP_IF_NONE_MATCH' in d)
- self.assertNotEqual(d['HTTP_IF_NONE_MATCH'], "fred")
- (response, content) = self.http.request(uri, "GET", headers = {'cache-control': 'max-age=0', 'if-none-match': 'fred'})
- d = self.reflector(content)
- self.assertTrue('HTTP_IF_NONE_MATCH' in d)
- self.assertEqual(d['HTTP_IF_NONE_MATCH'], "fred")
- #MAP-commented this out because it consistently fails
- # def testGet304EndToEnd(self):
- # # Test that end to end headers get overwritten in the cache
- # uri = urllib.parse.urljoin(base, "304/end2end.cgi")
- # (response, content) = self.http.request(uri, "GET")
- # self.assertNotEqual(response['etag'], "")
- # old_date = response['date']
- # time.sleep(2)
- #
- # (response, content) = self.http.request(uri, "GET", headers = {'Cache-Control': 'max-age=0'})
- # # The response should be from the cache, but the Date: header should be updated.
- # new_date = response['date']
- # self.assertNotEqual(new_date, old_date)
- # self.assertEqual(response.status, 200)
- # self.assertEqual(response.fromcache, True)
- def testGet304LastModified(self):
- # Test that we can still handle a 304
- # by only using the last-modified cache validator.
- uri = urllib.parse.urljoin(base, "304/last-modified-only/last-modified-only.txt")
- (response, content) = self.http.request(uri, "GET")
- self.assertNotEqual(response['last-modified'], "")
- (response, content) = self.http.request(uri, "GET")
- (response, content) = self.http.request(uri, "GET")
- self.assertEqual(response.status, 200)
- self.assertEqual(response.fromcache, True)
- def testGet307(self):
- # Test that we do follow 307 redirects but
- # do not cache the 307
- uri = urllib.parse.urljoin(base, "307/onestep.asis")
- (response, content) = self.http.request(uri, "GET")
- self.assertEqual(response.status, 200)
- self.assertEqual(content, b"This is the final destination.\n")
- self.assertEqual(response.previous.status, 307)
- self.assertEqual(response.previous.fromcache, False)
- (response, content) = self.http.request(uri, "GET")
- self.assertEqual(response.status, 200)
- self.assertEqual(response.fromcache, True)
- self.assertEqual(content, b"This is the final destination.\n")
- self.assertEqual(response.previous.status, 307)
- self.assertEqual(response.previous.fromcache, False)
- def testGet410(self):
- # Test that we pass 410's through
- uri = urllib.parse.urljoin(base, "410/410.asis")
- (response, content) = self.http.request(uri, "GET")
- self.assertEqual(response.status, 410)
- def testVaryHeaderSimple(self):
- """
- RFC 2616 13.6
- When the cache receives a subsequent request whose Request-URI
- specifies one or more cache entries including a Vary header field,
- the cache MUST NOT use such a cache entry to construct a response
- to the new request unless all of the selecting request-headers
- present in the new request match the corresponding stored
- request-headers in the original request.
- """
- # test that the vary header is sent
- uri = urllib.parse.urljoin(base, "vary/accept.asis")
- (response, content) = self.http.request(uri, "GET", headers={'Accept': 'text/plain'})
- self.assertEqual(response.status, 200)
- self.assertTrue('vary' in response)
- # get the resource again, from the cache since accept header in this
- # request is the same as the request
- (response, content) = self.http.request(uri, "GET", headers={'Accept': 'text/plain'})
- self.assertEqual(response.status, 200)
- self.assertEqual(response.fromcache, True, msg="Should be from cache")
- # get the resource again, not from cache since Accept headers does not match
- (response, content) = self.http.request(uri, "GET", headers={'Accept': 'text/html'})
- self.assertEqual(response.status, 200)
- self.assertEqual(response.fromcache, False, msg="Should not be from cache")
- # get the resource again, without any Accept header, so again no match
- (response, content) = self.http.request(uri, "GET")
- self.assertEqual(response.status, 200)
- self.assertEqual(response.fromcache, False, msg="Should not be from cache")
- def testNoVary(self):
- pass
- # when there is no vary, a different Accept header (e.g.) should not
- # impact if the cache is used
- # test that the vary header is not sent
- # uri = urllib.parse.urljoin(base, "vary/no-vary.asis")
- # (response, content) = self.http.request(uri, "GET", headers={'Accept': 'text/plain'})
- # self.assertEqual(response.status, 200)
- # self.assertFalse('vary' in response)
- #
- # (response, content) = self.http.request(uri, "GET", headers={'Accept': 'text/plain'})
- # self.assertEqual(response.status, 200)
- # self.assertEqual(response.fromcache, True, msg="Should be from cache")
- #
- # (response, content) = self.http.request(uri, "GET", headers={'Accept': 'text/html'})
- # self.assertEqual(response.status, 200)
- # self.assertEqual(response.fromcache, True, msg="Should be from cache")
- def testVaryHeaderDouble(self):
- uri = urllib.parse.urljoin(base, "vary/accept-double.asis")
- (response, content) = self.http.request(uri, "GET", headers={
- 'Accept': 'text/plain', 'Accept-Language': 'da, en-gb;q=0.8, en;q=0.7'})
- self.assertEqual(response.status, 200)
- self.assertTrue('vary' in response)
- # we are from cache
- (response, content) = self.http.request(uri, "GET", headers={
- 'Accept': 'text/plain', 'Accept-Language': 'da, en-gb;q=0.8, en;q=0.7'})
- self.assertEqual(response.fromcache, True, msg="Should be from cache")
- (response, content) = self.http.request(uri, "GET", headers={'Accept': 'text/plain'})
- self.assertEqual(response.status, 200)
- self.assertEqual(response.fromcache, False)
- # get the resource again, not from cache, varied headers don't match exact
- (response, content) = self.http.request(uri, "GET", headers={'Accept-Language': 'da'})
- self.assertEqual(response.status, 200)
- self.assertEqual(response.fromcache, False, msg="Should not be from cache")
- def testVaryUnusedHeader(self):
- # A header's value is not considered to vary if it's not used at all.
- uri = urllib.parse.urljoin(base, "vary/unused-header.asis")
- (response, content) = self.http.request(uri, "GET", headers={
- 'Accept': 'text/plain'})
- self.assertEqual(response.status, 200)
- self.assertTrue('vary' in response)
- # we are from cache
- (response, content) = self.http.request(uri, "GET", headers={
- 'Accept': 'text/plain',})
- self.assertEqual(response.fromcache, True, msg="Should be from cache")
- def testHeadGZip(self):
- # Test that we don't try to decompress a HEAD response
- uri = urllib.parse.urljoin(base, "gzip/final-destination.txt")
- (response, content) = self.http.request(uri, "HEAD")
- self.assertEqual(response.status, 200)
- self.assertNotEqual(int(response['content-length']), 0)
- self.assertEqual(content, b"")
- def testGetGZip(self):
- # Test that we support gzip compression
- uri = urllib.parse.urljoin(base, "gzip/final-destination.txt")
- (response, content) = self.http.request(uri, "GET")
- self.assertEqual(response.status, 200)
- self.assertFalse('content-encoding' in response)
- self.assertTrue('-content-encoding' in response)
- self.assertEqual(int(response['content-length']), len(b"This is the final destination.\n"))
- self.assertEqual(content, b"This is the final destination.\n")
- def testPostAndGZipResponse(self):
- uri = urllib.parse.urljoin(base, "gzip/post.cgi")
- (response, content) = self.http.request(uri, "POST", body=" ")
- self.assertEqual(response.status, 200)
- self.assertFalse('content-encoding' in response)
- self.assertTrue('-content-encoding' in response)
- def testGetGZipFailure(self):
- # Test that we raise a good exception when the gzip fails
- self.http.force_exception_to_status_code = False
- uri = urllib.parse.urljoin(base, "gzip/failed-compression.asis")
- try:
- (response, content) = self.http.request(uri, "GET")
- self.fail("Should never reach here")
- except httplib2.FailedToDecompressContent:
- pass
- except Exception:
- self.fail("Threw wrong kind of exception")
- # Re-run the test with out the exceptions
- self.http.force_exception_to_status_code = True
- (response, content) = self.http.request(uri, "GET")
- self.assertEqual(response.status, 500)
- self.assertTrue(response.reason.startswith("Content purported"))
- def testIndividualTimeout(self):
- uri = urllib.parse.urljoin(base, "timeout/timeout.cgi")
- http = httplib2.Http(timeout=1)
- http.force_exception_to_status_code = True
- (response, content) = http.request(uri)
- self.assertEqual(response.status, 408)
- self.assertTrue(response.reason.startswith("Request Timeout"))
- self.assertTrue(content.startswith(b"Request Timeout"))
- def testGetDeflate(self):
- # Test that we support deflate compression
- uri = urllib.parse.urljoin(base, "deflate/deflated.asis")
- (response, content) = self.http.request(uri, "GET")
- self.assertEqual(response.status, 200)
- self.assertFalse('content-encoding' in response)
- self.assertEqual(int(response['content-length']), len("This is the final destination."))
- self.assertEqual(content, b"This is the final destination.")
- def testGetDeflateFailure(self):
- # Test that we raise a good exception when the deflate fails
- self.http.force_exception_to_status_code = False
- uri = urllib.parse.urljoin(base, "deflate/failed-compression.asis")
- try:
- (response, content) = self.http.request(uri, "GET")
- self.fail("Should never reach here")
- except httplib2.FailedToDecompressContent:
- pass
- except Exception:
- self.fail("Threw wrong kind of exception")
- # Re-run the test with out the exceptions
- self.http.force_exception_to_status_code = True
- (response, content) = self.http.request(uri, "GET")
- self.assertEqual(response.status, 500)
- self.assertTrue(response.reason.startswith("Content purported"))
- def testGetDuplicateHeaders(self):
- # Test that duplicate headers get concatenated via ','
- uri = urllib.parse.urljoin(base, "duplicate-headers/multilink.asis")
- (response, content) = self.http.request(uri, "GET")
- self.assertEqual(response.status, 200)
- self.assertEqual(content, b"This is content\n")
- self.assertEqual(response['link'].split(",")[0], '<http://bitworking.org>; rel="home"; title="BitWorking"')
- def testGetCacheControlNoCache(self):
- # Test Cache-Control: no-cache on requests
- uri = urllib.parse.urljoin(base, "304/test_etag.txt")
- (response, content) = self.http.request(uri, "GET")
- self.assertNotEqual(response['etag'], "")
- (response, content) = self.http.request(uri, "GET")
- self.assertEqual(response.status, 200)
- self.assertEqual(response.fromcache, True)
- (response, content) = self.http.request(uri, "GET", headers={'Cache-Control': 'no-cache'})
- self.assertEqual(response.status, 200)
- self.assertEqual(response.fromcache, False)
- def testGetCacheControlPragmaNoCache(self):
- # Test Pragma: no-cache on requests
- uri = urllib.parse.urljoin(base, "304/test_etag.txt")
- (response, content) = self.http.request(uri, "GET")
- self.assertNotEqual(response['etag'], "")
- (response, content) = self.http.request(uri, "GET")
- self.assertEqual(response.status, 200)
- self.assertEqual(response.fromcache, True)
- (response, content) = self.http.request(uri, "GET", headers={'Pragma': 'no-cache'})
- self.assertEqual(response.status, 200)
- self.assertEqual(response.fromcache, False)
- def testGetCacheControlNoStoreRequest(self):
- # A no-store request means that the response should not be stored.
- uri = urllib.parse.urljoin(base, "304/test_etag.txt")
- (response, content) = self.http.request(uri, "GET", headers={'Cache-Control': 'no-store'})
- self.assertEqual(response.status, 200)
- self.assertEqual(response.fromcache, False)
- (response, content) = self.http.request(uri, "GET", headers={'Cache-Control': 'no-store'})
- self.assertEqual(response.status, 200)
- self.assertEqual(response.fromcache, False)
- def testGetCacheControlNoStoreResponse(self):
- # A no-store response means that the response should not be stored.
- uri = urllib.parse.urljoin(base, "no-store/no-store.asis")
- (response, content) = self.http.request(uri, "GET")
- self.assertEqual(response.status, 200)
- self.assertEqual(response.fromcache, False)
- (response, content) = self.http.request(uri, "GET")
- self.assertEqual(response.status, 200)
- self.assertEqual(response.fromcache, False)
- def testGetCacheControlNoCacheNoStoreRequest(self):
- # Test that a no-store, no-cache clears the entry from the cache
- # even if it was cached previously.
- uri = urllib.parse.urljoin(base, "304/test_etag.txt")
- (response, content) = self.http.request(uri, "GET")
- (response, content) = self.http.request(uri, "GET")
- self.assertEqual(response.fromcache, True)
- (response, content) = self.http.request(uri, "GET", headers={'Cache-Control': 'no-store, no-cache'})
- (response, content) = self.http.request(uri, "GET", headers={'Cache-Control': 'no-store, no-cache'})
- self.assertEqual(response.status, 200)
- self.assertEqual(response.fromcache, False)
- def testUpdateInvalidatesCache(self):
- # Test that calling PUT or DELETE on a
- # URI that is cache invalidates that cache.
- uri = urllib.parse.urljoin(base, "304/test_etag.txt")
- (response, content) = self.http.request(uri, "GET")
- (response, content) = self.http.request(uri, "GET")
- self.assertEqual(response.fromcache, True)
- (response, content) = self.http.request(uri, "DELETE")
- self.assertEqual(response.status, 405)
- (response, content) = self.http.request(uri, "GET")
- self.assertEqual(response.fromcache, False)
- def testUpdateUsesCachedETag(self):
- # Test that we natively support http://www.w3.org/1999/04/Editing/
- uri = urllib.parse.urljoin(base, "conditional-updates/test.cgi")
- (response, content) = self.http.request(uri, "GET")
- self.assertEqual(response.status, 200)
- self.assertEqual(response.fromcache, False)
- (response, content) = self.http.request(uri, "GET")
- self.assertEqual(response.status, 200)
- self.assertEqual(response.fromcache, True)
- (response, content) = self.http.request(uri, "PUT", body="foo")
- self.assertEqual(response.status, 200)
- (response, content) = self.http.request(uri, "PUT", body="foo")
- self.assertEqual(response.status, 412)
- def testUpdatePatchUsesCachedETag(self):
- # Test that we natively support http://www.w3.org/1999/04/Editing/
- uri = urllib.parse.urljoin(base, "conditional-updates/test.cgi")
- (response, content) = self.http.request(uri, "GET")
- self.assertEqual(response.status, 200)
- self.assertEqual(response.fromcache, False)
- (response, content) = self.http.request(uri, "GET")
- self.assertEqual(response.status, 200)
- self.assertEqual(response.fromcache, True)
- (response, content) = self.http.request(uri, "PATCH", body="foo")
- self.assertEqual(response.status, 200)
- (response, content) = self.http.request(uri, "PATCH", body="foo")
- self.assertEqual(response.status, 412)
- def testUpdateUsesCachedETagAndOCMethod(self):
- # Test that we natively support http://www.w3.org/1999/04/Editing/
- uri = urllib.parse.urljoin(base, "conditional-updates/test.cgi")
- (response, content) = self.http.request(uri, "GET")
- self.assertEqual(response.status, 200)
- self.assertEqual(response.fromcache, False)
- (response, content) = self.http.request(uri, "GET")
- self.assertEqual(response.status, 200)
- self.assertEqual(response.fromcache, True)
- self.http.optimistic_concurrency_methods.append("DELETE")
- (response, content) = self.http.request(uri, "DELETE")
- self.assertEqual(response.status, 200)
- def testUpdateUsesCachedETagOverridden(self):
- # Test that we natively support http://www.w3.org/1999/04/Editing/
- uri = urllib.parse.urljoin(base, "conditional-updates/test.cgi")
- (response, content) = self.http.request(uri, "GET")
- self.assertEqual(response.status, 200)
- self.assertEqual(response.fromcache, False)
- (response, content) = self.http.request(uri, "GET")
- self.assertEqual(response.status, 200)
- self.assertEqual(response.fromcache, True)
- (response, content) = self.http.request(uri, "PUT", body="foo", headers={'if-match': 'fred'})
- self.assertEqual(response.status, 412)
- def testBasicAuth(self):
- # Test Basic Authentication
- uri = urllib.parse.urljoin(base, "basic/file.txt")
- (response, content) = self.http.request(uri, "GET")
- self.assertEqual(response.status, 401)
- uri = urllib.parse.urljoin(base, "basic/")
- (response, content) = self.http.request(uri, "GET")
- self.assertEqual(response.status, 401)
- self.http.add_credentials('joe', 'password')
- (response, content) = self.http.request(uri, "GET")
- self.assertEqual(response.status, 200)
- uri = urllib.parse.urljoin(base, "basic/file.txt")
- (response, content) = self.http.request(uri, "GET")
- self.assertEqual(response.status, 200)
- def testBasicAuthWithDomain(self):
- # Test Basic Authentication
- uri = urllib.parse.urljoin(base, "basic/file.txt")
- (response, content) = self.http.request(uri, "GET")
- self.assertEqual(response.status, 401)
- uri = urllib.parse.urljoin(base, "basic/")
- (response, content) = self.http.request(uri, "GET")
- self.assertEqual(response.status, 401)
- self.http.add_credentials('joe', 'password', "example.org")
- (response, content) = self.http.request(uri, "GET")
- self.assertEqual(response.status, 401)
- uri = urllib.parse.urljoin(base, "basic/file.txt")
- (response, content) = self.http.request(uri, "GET")
- self.assertEqual(response.status, 401)
- domain = urllib.parse.urlparse(base)[1]
- self.http.add_credentials('joe', 'password', domain)
- (response, content) = self.http.request(uri, "GET")
- self.assertEqual(response.status, 200)
- uri = urllib.parse.urljoin(base, "basic/file.txt")
- (response, content) = self.http.request(uri, "GET")
- self.assertEqual(response.status, 200)
- def testBasicAuthTwoDifferentCredentials(self):
- # Test Basic Authentication with multiple sets of credentials
- uri = urllib.parse.urljoin(base, "basic2/file.txt")
- (response, content) = self.http.request(uri, "GET")
- self.assertEqual(response.status, 401)
- uri = urllib.parse.urljoin(base, "basic2/")
- (response, content) = self.http.request(uri, "GET")
- self.assertEqual(response.status, 401)
- self.http.add_credentials('fred', 'barney')
- (response, content) = self.http.request(uri, "GET")
- self.assertEqual(response.status, 200)
- uri = urllib.parse.urljoin(base, "basic2/file.txt")
- (response, content) = self.http.request(uri, "GET")
- self.assertEqual(response.status, 200)
- def testBasicAuthNested(self):
- # Test Basic Authentication with resources
- # that are nested
- uri = urllib.parse.urljoin(base, "basic-nested/")
- (response, content) = self.http.request(uri, "GET")
- self.assertEqual(response.status, 401)
- uri = urllib.parse.urljoin(base, "basic-nested/subdir")
- (response, content) = self.http.request(uri, "GET")
- self.assertEqual(response.status, 401)
- # Now add in credentials one at a time and test.
- self.http.add_credentials('joe', 'password')
- uri = urllib.parse.urljoin(base, "basic-nested/")
- (response, content) = self.http.request(uri, "GET")
- self.assertEqual(response.status, 200)
- uri = urllib.parse.urljoin(base, "basic-nested/subdir")
- (response, content) = self.http.request(uri, "GET")
- self.assertEqual(response.status, 401)
- self.http.add_credentials('fred', 'barney')
- uri = urllib.parse.urljoin(base, "basic-nested/")
- (response, content) = self.http.request(uri, "GET")
- self.assertEqual(response.status, 200)
- uri = urllib.parse.urljoin(base, "basic-nested/subdir")
- (response, content) = self.http.request(uri, "GET")
- self.assertEqual(response.status, 200)
- def testDigestAuth(self):
- # Test that we support Digest Authentication
- uri = urllib.parse.urljoin(base, "digest/")
- (response, content) = self.http.request(uri, "GET")
- self.assertEqual(response.status, 401)
- self.http.add_credentials('joe', 'password')
- (response, content) = self.http.request(uri, "GET")
- self.assertEqual(response.status, 200)
- uri = urllib.parse.urljoin(base, "digest/file.txt")
- (response, content) = self.http.request(uri, "GET")
- def testDigestAuthNextNonceAndNC(self):
- # Test that if the server sets nextnonce that we reset
- # the nonce count back to 1
- uri = urllib.parse.urljoin(base, "digest/file.txt")
- self.http.add_credentials('joe', 'password')
- (response, content) = self.http.request(uri, "GET", headers = {"cache-control":"no-cache"})
- info = httplib2._parse_www_authenticate(response, 'authentication-info')
- self.assertEqual(response.status, 200)
- (response, content) = self.http.request(uri, "GET", headers = {"cache-control":"no-cache"})
- info2 = httplib2._parse_www_authenticate(response, 'authentication-info')
- self.assertEqual(response.status, 200)
- if 'nextnonce' in info:
- self.assertEqual(info2['nc'], 1)
- def testDigestAuthStale(self):
- # Test that we can handle a nonce becoming stale
- uri = urllib.parse.urljoin(base, "digest-expire/file.txt")
- self.http.add_credentials('joe', 'password')
- (response, content) = self.http.request(uri, "GET", headers = {"cache-control":"no-cache"})
- info = httplib2._parse_www_authenticate(response, 'authentication-info')
- self.assertEqual(response.status, 200)
- time.sleep(3)
- # Sleep long enough that the nonce becomes stale
- (response, content) = self.http.request(uri, "GET", headers = {"cache-control":"no-cache"})
- self.assertFalse(response.fromcache)
- self.assertTrue(response._stale_digest)
- info3 = httplib2._parse_www_authenticate(response, 'authentication-info')
- self.assertEqual(response.status, 200)
- def reflector(self, content):
- return dict( [tuple(x.split("=", 1)) for x in content.decode('utf-8').strip().split("\n")] )
- def testReflector(self):
- uri = urllib.parse.urljoin(base, "reflector/reflector.cgi")
- (response, content) = self.http.request(uri, "GET")
- d = self.reflector(content)
- self.assertTrue('HTTP_USER_AGENT' in d)
- def testConnectionClose(self):
- uri = "http://www.google.com/"
- (response, content) = self.http.request(uri, "GET")
- for c in self.http.connections.values():
- self.assertNotEqual(None, c.sock)
- (response, content) = self.http.request(uri, "GET", headers={"connection": "close"})
- for c in self.http.connections.values():
- self.assertEqual(None, c.sock)
- def testPickleHttp(self):
- pickled_http = pickle.dumps(self.http)
- new_http = pickle.loads(pickled_http)
- self.assertEqual(sorted(new_http.__dict__.keys()),
- sorted(self.http.__dict__.keys()))
- for key in new_http.__dict__:
- if key in ('certificates', 'credentials'):
- self.assertEqual(new_http.__dict__[key].credentials,
- self.http.__dict__[key].credentials)
- elif key == 'cache':
- self.assertEqual(new_http.__dict__[key].cache,
- self.http.__dict__[key].cache)
- else:
- self.assertEqual(new_http.__dict__[key],
- self.http.__dict__[key])
- def testPickleHttpWithConnection(self):
- self.http.request('http://bitworking.org',
- connection_type=_MyHTTPConnection)
- pickled_http = pickle.dumps(self.http)
- new_http = pickle.loads(pickled_http)
- self.assertEqual(list(self.http.connections.keys()),
- ['http:bitworking.org'])
- self.assertEqual(new_http.connections, {})
- def testPickleCustomRequestHttp(self):
- def dummy_request(*args, **kwargs):
- return new_request(*args, **kwargs)
- dummy_request.dummy_attr = 'dummy_value'
- self.http.request = dummy_request
- pickled_http = pickle.dumps(self.http)
- self.assertFalse(b"S'request'" in pickled_http)
- try:
- import memcache
- class HttpTestMemCached(HttpTest):
- def setUp(self):
- self.cache = memcache.Client(['127.0.0.1:11211'], debug=0)
- #self.cache = memcache.Client(['10.0.0.4:11211'], debug=1)
- self.http = httplib2.Http(self.cache)
- self.cache.flush_all()
- # Not exactly sure why the sleep is needed here, but
- # if not present then some unit tests that rely on caching
- # fail. Memcached seems to lose some sets immediately
- # after a flush_all if the set is to a value that
- # was previously cached. (Maybe the flush is handled async?)
- time.sleep(1)
- self.http.clear_credentials()
- except:
- pass
- # ------------------------------------------------------------------------
- class HttpPrivateTest(unittest.TestCase):
- def testParseCacheControl(self):
- # Test that we can parse the Cache-Control header
- self.assertEqual({}, httplib2._parse_cache_control({}))
- self.assertEqual({'no-cache': 1}, httplib2._parse_cache_control({'cache-control': ' no-cache'}))
- cc = httplib2._parse_cache_control({'cache-control': ' no-cache, max-age = 7200'})
- self.assertEqual(cc['no-cache'], 1)
- self.assertEqual(cc['max-age'], '7200')
- cc = httplib2._parse_cache_control({'cache-control': ' , '})
- self.assertEqual(cc[''], 1)
- try:
- cc = httplib2._parse_cache_control({'cache-control': 'Max-age=3600;post-check=1800,pre-check=3600'})
- self.assertTrue("max-age" in cc)
- except:
- self.fail("Should not throw exception")
- def testNormalizeHeaders(self):
- # Test that we normalize headers to lowercase
- h = httplib2._normalize_headers({'Cache-Control': 'no-cache', 'Other': 'Stuff'})
- self.assertTrue('cache-control' in h)
- self.assertTrue('other' in h)
- self.assertEqual('Stuff', h['other'])
- def testExpirationModelTransparent(self):
- # Test that no-cache makes our request TRANSPARENT
- response_headers = {
- 'cache-control': 'max-age=7200'
- }
- request_headers = {
- 'cache-control': 'no-cache'
- }
- self.assertEqual("TRANSPARENT", httplib2._entry_disposition(response_headers, request_headers))
- def testMaxAgeNonNumeric(self):
- # Test that no-cache makes our request TRANSPARENT
- response_headers = {
- 'cache-control': 'max-age=fred, min-fresh=barney'
- }
- request_headers = {
- }
- self.assertEqual("STALE", httplib2._entry_disposition(response_headers, request_headers))
- def testExpirationModelNoCacheResponse(self):
- # The date and expires point to an entry that should be
- # FRESH, but the no-cache over-rides that.
- now = time.time()
- response_headers = {
- 'date': time.strftime("%a, %d %b %Y %H:%M:%S GMT", time.gmtime(now)),
- 'expires': time.strftime("%a, %d %b %Y %H:%M:%S GMT", time.gmtime(now+4)),
- 'cache-control': 'no-cache'
- }
- request_headers = {
- }
- self.assertEqual("STALE", httplib2._entry_disposition(response_headers, request_headers))
- def testExpirationModelStaleRequestMustReval(self):
- # must-revalidate forces STALE
- self.assertEqual("STALE", httplib2._entry_disposition({}, {'cache-control': 'must-revalidate'}))
- def testExpirationModelStaleResponseMustReval(self):
- # must-revalidate forces STALE
- self.assertEqual("STALE", httplib2._entry_disposition({'cache-control': 'must-revalidate'}, {}))
- def testExpirationModelFresh(self):
- response_headers = {
- 'date': time.strftime("%a, %d %b %Y %H:%M:%S GMT", time.gmtime()),
- 'cache-control': 'max-age=2'
- }
- request_headers = {
- }
- self.assertEqual("FRESH", httplib2._entry_disposition(response_headers, request_headers))
- time.sleep(3)
- self.assertEqual("STALE", httplib2._entry_disposition(response_headers, request_headers))
- def testExpirationMaxAge0(self):
- response_headers = {
- 'date': time.strftime("%a, %d %b %Y %H:%M:%S GMT", time.gmtime()),
- 'cache-control': 'max-age=0'
- }
- request_headers = {
- }
- self.assertEqual("STALE", httplib2._entry_disposition(response_headers, request_headers))
- def testExpirationModelDateAndExpires(self):
- now = time.time()
- response_headers = {
- 'date': time.strftime("%a, %d %b %Y %H:%M:%S GMT", time.gmtime(now)),
- 'expires': time.strftime("%a, %d %b %Y %H:%M:%S GMT", time.gmtime(now+2)),
- }
- request_headers = {
- }
- self.assertEqual("FRESH", httplib2._entry_disposition(response_headers, request_headers))
- time.sleep(3)
- self.assertEqual("STALE", httplib2._entry_disposition(response_headers, request_headers))
- def testExpiresZero(self):
- now = time.time()
- response_headers = {
- 'date': time.strftime("%a, %d %b %Y %H:%M:%S GMT", time.gmtime(now)),
- 'expires': "0",
- }
- request_headers = {
- }
- self.assertEqual("STALE", httplib2._entry_disposition(response_headers, request_headers))
- def testExpirationModelDateOnly(self):
- now = time.time()
- response_headers = {
- 'date': time.strftime("%a, %d %b %Y %H:%M:%S GMT", time.gmtime(now+3)),
- }
- request_headers = {
- }
- self.assertEqual("STALE", httplib2._entry_disposition(response_headers, request_headers))
- def testExpirationModelOnlyIfCached(self):
- response_headers = {
- }
- request_headers = {
- 'cache-control': 'only-if-cached',
- }
- self.assertEqual("FRESH", httplib2._entry_disposition(response_headers, request_headers))
- def testExpirationModelMaxAgeBoth(self):
- now = time.time()
- response_headers = {
- 'date': time.strftime("%a, %d %b %Y %H:%M:%S GMT", time.gmtime(now)),
- 'cache-control': 'max-age=2'
- }
- request_headers = {
- 'cache-control': 'max-age=0'
- }
- self.assertEqual("STALE", httplib2._entry_disposition(response_headers, request_headers))
- def testExpirationModelDateAndExpiresMinFresh1(self):
- now = time.time()
- response_headers = {
- 'date': time.strftime("%a, %d %b %Y %H:%M:%S GMT", time.gmtime(now)),
- 'expires': time.strftime("%a, %d %b %Y %H:%M:%S GMT", time.gmtime(now+2)),
- }
- request_headers = {
- 'cache-control': 'min-fresh=2'
- }
- self.assertEqual("STALE", httplib2._entry_disposition(response_headers, request_headers))
- def testExpirationModelDateAndExpiresMinFresh2(self):
- now = time.time()
- response_headers = {
- 'date': time.strftime("%a, %d %b %Y %H:%M:%S GMT", time.gmtime(now)),
- 'expires': time.strftime("%a, %d %b %Y %H:%M:%S GMT", time.gmtime(now+4)),
- }
- request_headers = {
- 'cache-control': 'min-fresh=2'
- }
- self.assertEqual("FRESH", httplib2._entry_disposition(response_headers, request_headers))
- def testParseWWWAuthenticateEmpty(self):
- res = httplib2._parse_www_authenticate({})
- self.assertEqual(len(list(res.keys())), 0)
- def testParseWWWAuthenticate(self):
- # different uses of spaces around commas
- res = httplib2._parse_www_authenticate({ 'www-authenticate': 'Test realm="test realm" , foo=foo ,bar="bar", baz=baz,qux=qux'})
- self.assertEqual(len(list(res.keys())), 1)
- self.assertEqual(len(list(res['test'].keys())), 5)
- # tokens with non-alphanum
- res = httplib2._parse_www_authenticate({ 'www-authenticate': 'T*!%#st realm=to*!%#en, to*!%#en="quoted string"'})
- self.assertEqual(len(list(res.keys())), 1)
- self.assertEqual(len(list(res['t*!%#st'].keys())), 2)
- # quoted string with quoted pairs
- res = httplib2._parse_www_authenticate({ 'www-authenticate': 'Test realm="a \\"test\\" realm"'})
- self.assertEqual(len(list(res.keys())), 1)
- self.assertEqual(res['test']['realm'], 'a "test" realm')
- def testParseWWWAuthenticateStrict(self):
- httplib2.USE_WWW_AUTH_STRICT_PARSING = 1;
- self.testParseWWWAuthenticate();
- httplib2.USE_WWW_AUTH_STRICT_PARSING = 0;
- def testParseWWWAuthenticateBasic(self):
- res = httplib2._parse_www_authenticate({ 'www-authenticate': 'Basic realm="me"'})
- basic = res['basic']
- self.assertEqual('me', basic['realm'])
- res = httplib2._parse_www_authenticate({ 'www-authenticate': 'Basic realm="me", algorithm="MD5"'})
- basic = res['basic']
- self.assertEqual('me', basic['realm'])
- self.assertEqual('MD5', basic['algorithm'])
- res = httplib2._parse_www_authenticate({ 'www-authenticate': 'Basic realm="me", algorithm=MD5'})
- basic = res['basic']
- self.assertEqual('me', basic['realm'])
- self.assertEqual('MD5', basic['algorithm'])
- def testParseWWWAuthenticateBasic2(self):
- res = httplib2._parse_www_authenticate({ 'www-authenticate': 'Basic realm="me",other="fred" '})
- basic = res['basic']
- self.assertEqual('me', basic['realm'])
- self.assertEqual('fred', basic['other'])
- def testParseWWWAuthenticateBasic3(self):
- res = httplib2._parse_www_authenticate({ 'www-authenticate': 'Basic REAlm="me" '})
- basic = res['basic']
- self.assertEqual('me', basic['realm'])
- def testParseWWWAuthenticateDigest(self):
- res = httplib2._parse_www_authenticate({ 'www-authenticate':
- 'Digest realm="testrealm@host.com", qop="auth,auth-int", nonce="dcd98b7102dd2f0e8b11d0f600bfb0c093", opaque="5ccc069c403ebaf9f0171e9517f40e41"'})
- digest = res['digest']
- self.assertEqual('testrealm@host.com', digest['realm'])
- self.assertEqual('auth,auth-int', digest['qop'])
- def testParseWWWAuthenticateMultiple(self):
- res = httplib2._parse_www_authenticate({ 'www-authenticate':
- 'Digest realm="testrealm@host.com", qop="auth,auth-int", nonce="dcd98b7102dd2f0e8b11d0f600bfb0c093", opaque="5ccc069c403ebaf9f0171e9517f40e41" Basic REAlm="me" '})
- digest = res['digest']
- self.assertEqual('testrealm@host.com', digest['realm'])
- self.assertEqual('auth,auth-int', digest['qop'])
- self.assertEqual('dcd98b7102dd2f0e8b11d0f600bfb0c093', digest['nonce'])
- self.assertEqual('5ccc069c403ebaf9f0171e9517f40e41', digest['opaque'])
- basic = res['basic']
- self.assertEqual('me', basic['realm'])
- def testParseWWWAuthenticateMultiple2(self):
- # Handle an added comma between challenges, which might get thrown in if the challenges were
- # originally sent in separate www-authenticate headers.
- res = httplib2._parse_www_authenticate({ 'www-authenticate':
- 'Digest realm="testrealm@host.com", qop="auth,auth-int", nonce="dcd98b7102dd2f0e8b11d0f600bfb0c093", opaque="5ccc069c403ebaf9f0171e9517f40e41", Basic REAlm="me" '})
- digest = res['digest']
- self.assertEqual('testrealm@host.com', digest['realm'])
- self.assertEqual('auth,auth-int', digest['qop'])
- self.assertEqual('dcd98b7102dd2f0e8b11d0f600bfb0c093', digest['nonce'])
- self.assertEqual('5ccc069c403ebaf9f0171e9517f40e41', digest['opaque'])
- basic = res['basic']
- self.assertEqual('me', basic['realm'])
- def testParseWWWAuthenticateMultiple3(self):
- # Handle an added comma between challenges, which might get thrown in if the challenges were
- # originally sent in separate www-authenticate headers.
- res = httplib2._parse_www_authenticate({ 'www-authenticate':
- 'Digest realm="testrealm@host.com", qop="auth,auth-int", nonce="dcd98b7102dd2f0e8b11d0f600bfb0c093", opaque="5ccc069c403ebaf9f0171e9517f40e41", Basic REAlm="me", WSSE realm="foo", profile="UsernameToken"'})
- digest = res['digest']
- self.assertEqual('testrealm@host.com', digest['realm'])
- self.assertEqual('auth,auth-int', digest['qop'])
- self.assertEqual('dcd98b7102dd2f0e8b11d0f600bfb0c093', digest['nonce'])
- self.assertEqual('5ccc069c403ebaf9f0171e9517f40e41', digest['opaque'])
- basic = res['basic']
- self.assertEqual('me', basic['realm'])
- wsse = res['wsse']
- self.assertEqual('foo', wsse['realm'])
- self.assertEqual('UsernameToken', wsse['profile'])
- def testParseWWWAuthenticateMultiple4(self):
- res = httplib2._parse_www_authenticate({ 'www-authenticate':
- 'Digest realm="test-real.m@host.com", qop \t=\t"\tauth,auth-int", nonce="(*)&^&$%#",opaque="5ccc069c403ebaf9f0171e9517f40e41", Basic REAlm="me", WSSE realm="foo", profile="UsernameToken"'})
- digest = res['digest']
- self.assertEqual('test-real.m@host.com', digest['realm'])
- self.assertEqual('\tauth,auth-int', digest['qop'])
- self.assertEqual('(*)&^&$%#', digest['nonce'])
- def testParseWWWAuthenticateMoreQuoteCombos(self):
- res = httplib2._parse_www_authenticate({'www-authenticate':'Digest realm="myrealm", nonce="Ygk86AsKBAA=3516200d37f9a3230352fde99977bd6d472d4306", algorithm=MD5, qop="auth", stale=true'})
- digest = res['digest']
- self.assertEqual('myrealm', digest['realm'])
- def testParseWWWAuthenticateMalformed(self):
- try:
- res = httplib2._parse_www_authenticate({'www-authenticate':'OAuth "Facebook Platform" "invalid_token" "Invalid OAuth access token."'})
- self.fail("should raise an exception")
- except httplib2.MalformedHeader:
- pass
- def testDigestObject(self):
- credentials = ('joe', 'password')
- host = None
- request_uri = '/projects/httplib2/test/digest/'
- headers = {}
- response = {
- 'www-authenticate': 'Digest realm="myrealm", nonce="Ygk86AsKBAA=3516200d37f9a3230352fde99977bd6d472d4306", algorithm=MD5, qop="auth"'
- }
- content = b""
- d = httplib2.DigestAuthentication(credentials, host, request_uri, headers, response, content, None)
- d.request("GET", request_uri, headers, content, cnonce="33033375ec278a46")
- our_request = "authorization: %s" % headers['authorization']
- 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"'
- self.assertEqual(our_request, working_request)
- def testDigestObjectWithOpaque(self):
- credentials = ('joe', 'password')
- host = None
- request_uri = '/projects/httplib2/test/digest/'
- headers = {}
- response = {
- 'www-authenticate': 'Digest realm="myrealm", nonce="Ygk86AsKBAA=3516200d37f9a3230352fde99977bd6d472d4306", algorithm=MD5, qop="auth", opaque="atestopaque"'
- }
- content = ""
- d = httplib2.DigestAuthentication(credentials, host, request_uri, headers, response, content, None)
- d.request("GET", request_uri, headers, content, cnonce="33033375ec278a46")
- our_request = "authorization: %s" % headers['authorization']
- 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"'
- self.assertEqual(our_request, working_request)
- def testDigestObjectStale(self):
- credentials = ('joe', 'password')
- host = None
- request_uri = '/projects/httplib2/test/digest/'
- headers = {}
- response = httplib2.Response({ })
- response['www-authenticate'] = 'Digest realm="myrealm", nonce="Ygk86AsKBAA=3516200d37f9a3230352fde99977bd6d472d4306", algorithm=MD5, qop="auth", stale=true'
- response.status = 401
- content = b""
- d = httplib2.DigestAuthentication(credentials, host, request_uri, headers, response, content, None)
- # Returns true to force a retry
- self.assertTrue( d.response(response, content) )
- def testDigestObjectAuthInfo(self):
- credentials = ('joe', 'password')
- host = None
- request_uri = '/projects/httplib2/test/digest/'
- headers = {}
- response = httplib2.Response({ })
- response['www-authenticate'] = 'Digest realm="myrealm", nonce="Ygk86AsKBAA=3516200d37f9a3230352fde99977bd6d472d4306", algorithm=MD5, qop="auth", stale=true'
- response['authentication-info'] = 'nextnonce="fred"'
- content = b""
- d = httplib2.DigestAuthentication(credentials, host, request_uri, headers, response, content, None)
- # Returns true to force a retry
- self.assertFalse( d.response(response, content) )
- self.assertEqual('fred', d.challenge['nonce'])
- self.assertEqual(1, d.challenge['nc'])
- def testWsseAlgorithm(self):
- digest = httplib2._wsse_username_token("d36e316282959a9ed4c89851497a717f", "2003-12-15T14:43:07Z", "taadtaadpstcsm")
- expected = b"quR/EWLAV4xLf9Zqyw4pDmfV9OY="
- self.assertEqual(expected, digest)
- def testEnd2End(self):
- # one end to end header
- response = {'content-type': 'application/atom+xml', 'te': 'deflate'}
- end2end = httplib2._get_end2end_headers(response)
- self.assertTrue('content-type' in end2end)
- self.assertTrue('te' not in end2end)
- self.assertTrue('connection' not in end2end)
- # one end to end header that gets eliminated
- response = {'connection': 'content-type', 'content-type': 'application/atom+xml', 'te': 'deflate'}
- end2end = httplib2._get_end2end_headers(response)
- self.assertTrue('content-type' not in end2end)
- self.assertTrue('te' not in end2end)
- self.assertTrue('connection' not in end2end)
- # Degenerate case of no headers
- response = {}
- end2end = httplib2._get_end2end_headers(response)
- self.assertEqual(0, len(end2end))
- # Degenerate case of connection referrring to a header not passed in
- response = {'connection': 'content-type'}
- end2end = httplib2._get_end2end_headers(response)
- self.assertEqual(0, len(end2end))
- class TestProxyInfo(unittest.TestCase):
- def setUp(self):
- self.orig_env = dict(os.environ)
- def tearDown(self):
- os.environ.clear()
- os.environ.update(self.orig_env)
- def test_from_url(self):
- pi = httplib2.proxy_info_from_url('http://myproxy.example.com')
- self.assertEqual(pi.proxy_host, 'myproxy.example.com')
- self.assertEqual(pi.proxy_port, 80)
- self.assertEqual(pi.proxy_user, None)
- def test_from_url_ident(self):
- pi = httplib2.proxy_info_from_url('http://zoidberg:fish@someproxy:99')
- self.assertEqual(pi.proxy_host, 'someproxy')
- self.assertEqual(pi.proxy_port, 99)
- self.assertEqual(pi.proxy_user, 'zoidberg')
- self.assertEqual(pi.proxy_pass, 'fish')
- def test_from_env(self):
- os.environ['http_proxy'] = 'http://myproxy.example.com:8080'
- pi = httplib2.proxy_info_from_environment()
- self.assertEqual(pi.proxy_host, 'myproxy.example.com')
- self.assertEqual(pi.proxy_port, 8080)
- def test_from_env_no_proxy(self):
- os.environ['http_proxy'] = 'http://myproxy.example.com:80'
- os.environ['https_proxy'] = 'http://myproxy.example.com:81'
- pi = httplib2.proxy_info_from_environment('https')
- self.assertEqual(pi.proxy_host, 'myproxy.example.com')
- self.assertEqual(pi.proxy_port, 81)
- def test_from_env_none(self):
- os.environ.clear()
- pi = httplib2.proxy_info_from_environment()
- self.assertEqual(pi, None)
- if __name__ == '__main__':
- unittest.main()
|