#!/usr/bin/env python2.4 """A set of unit tests for httplib2.py.""" __author__ = "Joe Gregorio (joe@bitworking.org)" __copyright__ = "Copyright 2006, Joe Gregorio" __contributors__ = [] __license__ = "MIT" __version__ = "0.1 ($Rev: 118 $)" import base64 import httplib import httplib2 import os import pickle import socket import StringIO import sys import time import unittest import urlparse try: import ssl except ImportError: pass # Python 2.3 support if not hasattr(unittest.TestCase, "assertTrue"): unittest.TestCase.assertTrue = unittest.TestCase.failUnless unittest.TestCase.assertFalse = unittest.TestCase.failIf # 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(u"http://\u2304.org/fred/?a=b"), ) class _MyResponse(StringIO.StringIO): def __init__(self, body, **kwargs): StringIO.StringIO.__init__(self, body) self.headers = kwargs def iteritems(self): return self.headers.iteritems() 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("the body", status="200") class _MyHTTPBadStatusConnection(object): "Mock of httplib.HTTPConnection that raises BadStatusLine." num_calls = 0 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 _MyHTTPBadStatusConnection.num_calls = 0 def set_debuglevel(self, level): pass def connect(self): pass def close(self): pass def request(self, method, request_uri, body, headers): pass def getresponse(self): _MyHTTPBadStatusConnection.num_calls += 1 raise httplib.BadStatusLine("") 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) ] if sys.version_info < (2, 6): disable_cert_validation = True else: disable_cert_validation = False self.http = httplib2.Http( cacheDirName, disable_ssl_certificate_validation=disable_cert_validation ) 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 httplib2.CertificateHostnameMismatch: # We connected and verified that the certificate doesn't match # the name. Good enough. pass 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, "the body") def testBadStatusLineRetry(self): old_retries = httplib2.RETRIES httplib2.RETRIES = 1 self.http.force_exception_to_status_code = False try: response, content = self.http.request( "http://bitworking.org", connection_type=_MyHTTPBadStatusConnection ) except httplib.BadStatusLine: self.assertEqual(2, _MyHTTPBadStatusConnection.num_calls) httplib2.RETRIES = old_retries 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("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( "Connection refused" in content or "actively refused" in content, "Unexpected status %(content)s" % vars(), ) self.assertEqual(response.status, 400) def testGetIRI(self): if sys.version_info >= (2, 3): uri = urlparse.urljoin( base, u"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 = urlparse.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 = urlparse.urljoin(base, "methods/method_reflector.cgi") for method in ["GET", "PUT", "DELETE", "POST"]: (response, content) = self.http.request(uri, method, body=" ") 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, "") def testGetNoCache(self): # Test that can do a GET w/o the cache turned on. http = httplib2.Http() uri = urlparse.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 = urlparse.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 = urlparse.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 = urlparse.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 = urlparse.urljoin(base, "user-agent/test.cgi") (response, content) = self.http.request(uri, "GET") self.assertEqual(response.status, 200) self.assertTrue(content.startswith("Python-httplib2/")) def testUserAgentNonDefault(self): # Test that the default user-agent can be over-ridden uri = urlparse.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("fred/1.0")) def testGet300WithLocation(self): # Test the we automatically follow 300 redirects if a Location: header is provided uri = urlparse.urljoin(base, "300/with-location-header.asis") (response, content) = self.http.request(uri, "GET") self.assertEqual(response.status, 200) self.assertEqual(content, "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, "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 = urlparse.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 = urlparse.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 = urlparse.urljoin(base, "301/onestep.asis") destination = urlparse.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, "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, "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 = urlparse.urljoin(base, "301/onestep.asis") destination = urlparse.urljoin(base, "302/final-destination.txt") (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 = urlparse.urljoin(base, "301/onestep.asis") destination = urlparse.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 = urlparse.urljoin(base, "302/onestep.asis") destination = urlparse.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, "This is the final destination.\n") self.assertEqual(response.previous.status, 302) self.assertEqual(response.previous.fromcache, False) uri = urlparse.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, "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 = urlparse.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, "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 = urlparse.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("")) 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 = urlparse.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("This is content")) def testGet301ViaHttps(self): # Google always redirects to https://www.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://www.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://www.google.com/adsense", "GET") self.assertEqual(200, response.status) self.assertNotEqual(None, response.previous) def testSslCertValidationDoubleDots(self): pass # No longer a valid test. # if sys.version_info >= (2, 6): # Test that we get match a double dot cert # try: # self.http.request("https://www.appspot.com/", "GET") # except httplib2.CertificateHostnameMismatch: # self.fail('cert with *.*.appspot.com should not raise an exception.') def testSslHostnameValidation(self): pass # No longer a valid test. # if sys.version_info >= (2, 6): # The SSL server at google.com:443 returns a certificate for # 'www.google.com', which results in a host name mismatch. # Note that this test only works because the ssl module and httplib2 # do not support SNI; for requests specifying a server name of # 'google.com' via SNI, a matching cert would be returned. # self.assertRaises(httplib2.CertificateHostnameMismatch, # self.http.request, "https://google.com/", "GET") def testSslCertValidationWithoutSslModuleFails(self): if sys.version_info < (2, 6): http = httplib2.Http(disable_ssl_certificate_validation=False) self.assertRaises( httplib2.CertificateValidationUnsupported, http.request, "https://www.google.com/", "GET", ) 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. # bitworking.org presents an certificate for a non-matching host # (*.webfaction.com), so we need to disable cert checking for this test. http = httplib2.Http(timeout=2, disable_ssl_certificate_validation=True) http.add_certificate("akeyfile", "acertfile", "bitworking.org") try: (response, content) = http.request("https://bitworking.org", "GET") except: pass self.assertEqual(http.connections["https:bitworking.org"].key_file, "akeyfile") self.assertEqual( http.connections["https:bitworking.org"].cert_file, "acertfile" ) try: (response, content) = http.request("https://notthere.bitworking.org", "GET") except: pass self.assertEqual( http.connections["https:notthere.bitworking.org"].key_file, None ) self.assertEqual( http.connections["https:notthere.bitworking.org"].cert_file, None ) def testGet303(self): # Do a follow-up GET on a Location: header # returned from a POST that gave a 303. uri = urlparse.urljoin(base, "303/303.cgi") (response, content) = self.http.request(uri, "POST", " ") self.assertEqual(response.status, 200) self.assertEqual(content, "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 = urlparse.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 = urlparse.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=" ") self.assertEqual(response["x-method"], method_on_303) def test303AndForwardAuthorizationHeader(self): # Test that all methods can be used uri = urlparse.urljoin(base, "303/redirect-to-header-reflector.cgi") headers = {"authorization": "Bearer foo"} response, content = self.http.request(uri, "GET", body=" ", headers=headers) # self.assertTrue('authorization' not in content) self.http.follow_all_redirects = True self.http.forward_authorization_headers = True response, content = self.http.request(uri, "GET", body=" ", headers=headers) # Oh, how I wish Apache didn't eat the Authorization header. # self.assertTrue('authorization' in content) def testGet304(self): # Test that we use ETags properly to validate our cache uri = urlparse.urljoin(base, "304/test_etag.txt") (response, content) = self.http.request( uri, "GET", headers={"accept-encoding": "identity"} ) 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 = urlparse.urljoin(base, "reflector/reflector.cgi") (response, content) = self.http.request( uri, "GET", headers={"accept-encoding": "identity"} ) self.assertNotEqual(response["etag"], "") (response, content) = self.http.request( uri, "GET", headers={"accept-encoding": "identity", "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={"accept-encoding": "identity", "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 = urlparse.urljoin(base, "reflector/reflector.cgi") (response, content) = self.http.request( uri, "GET", headers={"accept-encoding": "identity"} ) self.assertNotEqual(response["etag"], "") (response, content) = self.http.request( uri, "GET", headers={"accept-encoding": "identity", "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={ "accept-encoding": "identity", "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 = urlparse.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 = urlparse.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 = urlparse.urljoin(base, "307/onestep.asis") (response, content) = self.http.request(uri, "GET") self.assertEqual(response.status, 200) self.assertEqual(content, "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, "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 = urlparse.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 = urlparse.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 = urlparse.urljoin(base, "vary/no-vary.asis") # (response, content) = self.http.request(uri, "GET", headers={'Accept': 'text/plain'}) # self.assertEqual(response.status, 200) # self.assertFalse(response.has_key('vary')) # (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 = urlparse.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 = urlparse.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 = urlparse.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, "") def testGetGZip(self): # Test that we support gzip compression uri = urlparse.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("This is the final destination.\n") ) self.assertEqual(content, "This is the final destination.\n") def testPostAndGZipResponse(self): uri = urlparse.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 = urlparse.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 testTimeout(self): self.http.force_exception_to_status_code = True uri = urlparse.urljoin(base, "timeout/timeout.cgi") try: import socket socket.setdefaulttimeout(1) except: # Don't run the test if we can't set the timeout return (response, content) = self.http.request(uri) self.assertEqual(response.status, 408) self.assertTrue(response.reason.startswith("Request Timeout")) self.assertTrue(content.startswith("Request Timeout")) def testIndividualTimeout(self): uri = urlparse.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("Request Timeout")) def testHTTPSInitTimeout(self): c = httplib2.HTTPSConnectionWithTimeout("localhost", 80, timeout=47) self.assertEqual(47, c.timeout) def testGetDeflate(self): # Test that we support deflate compression uri = urlparse.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, "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 = urlparse.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 = urlparse.urljoin(base, "duplicate-headers/multilink.asis") (response, content) = self.http.request(uri, "GET") self.assertEqual(response.status, 200) self.assertEqual(content, "This is content\n") self.assertEqual( response["link"].split(",")[0], '; rel="home"; title="BitWorking"', ) def testGetCacheControlNoCache(self): # Test Cache-Control: no-cache on requests uri = urlparse.urljoin(base, "304/test_etag.txt") (response, content) = self.http.request( uri, "GET", headers={"accept-encoding": "identity"} ) self.assertNotEqual(response["etag"], "") (response, content) = self.http.request( uri, "GET", headers={"accept-encoding": "identity"} ) self.assertEqual(response.status, 200) self.assertEqual(response.fromcache, True) (response, content) = self.http.request( uri, "GET", headers={"accept-encoding": "identity", "Cache-Control": "no-cache"}, ) self.assertEqual(response.status, 200) self.assertEqual(response.fromcache, False) def testGetCacheControlPragmaNoCache(self): # Test Pragma: no-cache on requests uri = urlparse.urljoin(base, "304/test_etag.txt") (response, content) = self.http.request( uri, "GET", headers={"accept-encoding": "identity"} ) self.assertNotEqual(response["etag"], "") (response, content) = self.http.request( uri, "GET", headers={"accept-encoding": "identity"} ) self.assertEqual(response.status, 200) self.assertEqual(response.fromcache, True) (response, content) = self.http.request( uri, "GET", headers={"accept-encoding": "identity", "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 = urlparse.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 = urlparse.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 = urlparse.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 = urlparse.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 = urlparse.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 = urlparse.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 = urlparse.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 = urlparse.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 = urlparse.urljoin(base, "basic/file.txt") (response, content) = self.http.request(uri, "GET") self.assertEqual(response.status, 401) uri = urlparse.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 = urlparse.urljoin(base, "basic/file.txt") (response, content) = self.http.request(uri, "GET") self.assertEqual(response.status, 200) def testBasicAuthWithDomain(self): # Test Basic Authentication uri = urlparse.urljoin(base, "basic/file.txt") (response, content) = self.http.request(uri, "GET") self.assertEqual(response.status, 401) uri = urlparse.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 = urlparse.urljoin(base, "basic/file.txt") (response, content) = self.http.request(uri, "GET") self.assertEqual(response.status, 401) domain = urlparse.urlparse(base)[1] self.http.add_credentials("joe", "password", domain) (response, content) = self.http.request(uri, "GET") self.assertEqual(response.status, 200) uri = urlparse.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 = urlparse.urljoin(base, "basic2/file.txt") (response, content) = self.http.request(uri, "GET") self.assertEqual(response.status, 401) uri = urlparse.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 = urlparse.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 = urlparse.urljoin(base, "basic-nested/") (response, content) = self.http.request(uri, "GET") self.assertEqual(response.status, 401) uri = urlparse.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 = urlparse.urljoin(base, "basic-nested/") (response, content) = self.http.request(uri, "GET") self.assertEqual(response.status, 200) uri = urlparse.urljoin(base, "basic-nested/subdir") (response, content) = self.http.request(uri, "GET") self.assertEqual(response.status, 401) self.http.add_credentials("fred", "barney") uri = urlparse.urljoin(base, "basic-nested/") (response, content) = self.http.request(uri, "GET") self.assertEqual(response.status, 200) uri = urlparse.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 = urlparse.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 = urlparse.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 = urlparse.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 = urlparse.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.strip().split("\n")]) def testReflector(self): uri = urlparse.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(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("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(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(res.keys()), 1) self.assertEqual(len(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(res.keys()), 1) self.assertEqual(len(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(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 = "" 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 = "" 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 = "" 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 = "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.assertEquals(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.assertEquals(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.assertEquals(pi.proxy_host, "myproxy.example.com") self.assertEquals(pi.proxy_port, 80) self.assertEquals(pi.proxy_user, None) def test_from_url_ident(self): pi = httplib2.proxy_info_from_url("http://zoidberg:fish@someproxy:99") self.assertEquals(pi.proxy_host, "someproxy") self.assertEquals(pi.proxy_port, 99) self.assertEquals(pi.proxy_user, "zoidberg") self.assertEquals(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.assertEquals(pi.proxy_host, "myproxy.example.com") self.assertEquals(pi.proxy_port, 8080) self.assertEquals(pi.bypass_hosts, []) 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" os.environ["no_proxy"] = "localhost,otherhost.domain.local" pi = httplib2.proxy_info_from_environment("https") self.assertEquals(pi.proxy_host, "myproxy.example.com") self.assertEquals(pi.proxy_port, 81) self.assertEquals(pi.bypass_hosts, ["localhost", "otherhost.domain.local"]) def test_from_env_none(self): os.environ.clear() pi = httplib2.proxy_info_from_environment() self.assertEquals(pi, None) def test_applies_to(self): os.environ["http_proxy"] = "http://myproxy.example.com:80" os.environ["https_proxy"] = "http://myproxy.example.com:81" os.environ["no_proxy"] = "localhost,otherhost.domain.local,example.com" pi = httplib2.proxy_info_from_environment() self.assertFalse(pi.applies_to("localhost")) self.assertTrue(pi.applies_to("www.google.com")) self.assertFalse(pi.applies_to("www.example.com")) def test_no_proxy_star(self): os.environ["http_proxy"] = "http://myproxy.example.com:80" os.environ["NO_PROXY"] = "*" pi = httplib2.proxy_info_from_environment() for host in ("localhost", "169.254.38.192", "www.google.com"): self.assertFalse(pi.applies_to(host)) def test_proxy_headers(self): headers = {"key0": "val0", "key1": "val1"} pi = httplib2.ProxyInfo( httplib2.socks.PROXY_TYPE_HTTP, "localhost", 1234, proxy_headers=headers ) self.assertEquals(pi.proxy_headers, headers) if __name__ == "__main__": unittest.main()