httplib2 support

tests/integration/test_httplib2.py

@@ -0,0 +1,140 @@
+'''Integration tests with httplib2'''
+# coding=utf-8
+
+# External imports
+from urllib import urlencode
+import pytest
+
+# Internal imports
+import vcr
+
+from assertions import assert_cassette_has_one_response
+
+httplib2 = pytest.importorskip("httplib2")
+
+
+@pytest.fixture(params=["https", "http"])
+def scheme(request):
+    """
+    Fixture that returns both http and https
+    """
+    return request.param
+
+
+def test_response_code(scheme, tmpdir):
+    '''Ensure we can read a response code from a fetch'''
+    url = scheme + '://httpbin.org/'
+    with vcr.use_cassette(str(tmpdir.join('atts.yaml'))) as cass:
+        resp, _ = httplib2.Http().request(url)
+        code = resp.status
+
+    with vcr.use_cassette(str(tmpdir.join('atts.yaml'))) as cass:
+        resp, _ = httplib2.Http().request(url)
+        assert code == resp.status
+
+
+def test_random_body(scheme, tmpdir):
+    '''Ensure we can read the content, and that it's served from cache'''
+    url = scheme + '://httpbin.org/bytes/1024'
+    with vcr.use_cassette(str(tmpdir.join('body.yaml'))) as cass:
+        _, content = httplib2.Http().request(url)
+        body = content
+
+    with vcr.use_cassette(str(tmpdir.join('body.yaml'))) as cass:
+        _, content = httplib2.Http().request(url)
+        assert body == content
+
+
+def test_response_headers(scheme, tmpdir):
+    '''Ensure we can get information from the response'''
+    url = scheme + '://httpbin.org/'
+    with vcr.use_cassette(str(tmpdir.join('headers.yaml'))) as cass:
+        resp, _ = httplib2.Http().request(url)
+        headers = resp.items()
+
+    with vcr.use_cassette(str(tmpdir.join('headers.yaml'))) as cass:
+        resp, _ = httplib2.Http().request(url)
+        assert headers == resp.items()
+
+
+def test_multiple_requests(scheme, tmpdir):
+    '''Ensure that we can cache multiple requests'''
+    urls = [
+        scheme + '://httpbin.org/',
+        scheme + '://httpbin.org/',
+        scheme + '://httpbin.org/get',
+        scheme + '://httpbin.org/bytes/1024'
+    ]
+    with vcr.use_cassette(str(tmpdir.join('multiple.yaml'))) as cass:
+        [httplib2.Http().request(url) for url in urls]
+    assert len(cass) == len(urls)
+
+
+def test_get_data(scheme, tmpdir):
+    '''Ensure that it works with query data'''
+    data = urlencode({'some': 1, 'data': 'here'})
+    url = scheme + '://httpbin.org/get?' + data
+    with vcr.use_cassette(str(tmpdir.join('get_data.yaml'))) as cass:
+        _, res1 = httplib2.Http().request(url)
+
+    with vcr.use_cassette(str(tmpdir.join('get_data.yaml'))) as cass:
+        _, res2 = httplib2.Http().request(url)
+
+    assert res1 == res2
+
+
+def test_post_data(scheme, tmpdir):
+    '''Ensure that it works when posting data'''
+    data = urlencode({'some': 1, 'data': 'here'})
+    url = scheme + '://httpbin.org/post'
+    with vcr.use_cassette(str(tmpdir.join('post_data.yaml'))) as cass:
+        _, res1 = httplib2.Http().request(url, "POST", data)
+
+    with vcr.use_cassette(str(tmpdir.join('post_data.yaml'))) as cass:
+        _, res2 = httplib2.Http().request(url, "POST", data)
+
+    assert res1 == res2
+    assert_cassette_has_one_response(cass)
+
+
+def test_post_unicode_data(scheme, tmpdir):
+    '''Ensure that it works when posting unicode data'''
+    data = urlencode({'snowman': u'☃'.encode('utf-8')})
+    url = scheme + '://httpbin.org/post'
+    with vcr.use_cassette(str(tmpdir.join('post_data.yaml'))) as cass:
+        _, res1 = httplib2.Http().request(url, "POST", data)
+
+    with vcr.use_cassette(str(tmpdir.join('post_data.yaml'))) as cass:
+        _, res2 = httplib2.Http().request(url, "POST", data)
+
+    assert res1 == res2
+    assert_cassette_has_one_response(cass)
+
+
+def test_cross_scheme(tmpdir):
+    '''Ensure that requests between schemes are treated separately'''
+    # First fetch a url under https, and then again under https and then
+    # ensure that we haven't served anything out of cache, and we have two
+    # requests / response pairs in the cassette
+    with vcr.use_cassette(str(tmpdir.join('cross_scheme.yaml'))) as cass:
+        httplib2.Http().request('https://httpbin.org/')
+        httplib2.Http().request('http://httpbin.org/')
+        assert len(cass) == 2
+        assert cass.play_count == 0
+
+
+def test_decorator(scheme, tmpdir):
+    '''Test the decorator version of VCR.py'''
+    url = scheme + '://httpbin.org/'
+
+    @vcr.use_cassette(str(tmpdir.join('atts.yaml')))
+    def inner1():
+        resp, _ = httplib2.Http().request(url)
+        return resp['status']
+
+    @vcr.use_cassette(str(tmpdir.join('atts.yaml')))
+    def inner2():
+        resp, _ = httplib2.Http().request(url)
+        return resp['status']
+
+    assert inner1() == inner2()

tox.ini

@@ -4,7 +4,8 @@
 # and then run "tox" from this directory.
 
 [tox]
-envlist = py26, py27, pypy, py26requests, py27requests, pypyrequests, py26oldrequests, py27oldrequests, pypyoldrequests
+#envlist = py26, py27, pypy, py26requests, py27requests, pypyrequests, py26oldrequests, py27oldrequests, pypyoldrequests
+envlist = py26, py27, pypy, py26httplib2, py27httplib2, pypyhttplib2
 
 [testenv]
 commands =
@@ -61,3 +62,24 @@ deps =
     pytest
     PyYAML
     requests
+
+[testenv:py26httplib2]
+basepython = python2.6
+deps =
+    pytest
+    PyYAML
+    httplib2
+
+[testenv:py27httplib2]
+basepython = python2.7
+deps =
+    pytest
+    PyYAML
+    httplib2
+
+[testenv:pypyhttplib2]
+basepython = pypy
+deps =
+    pytest
+    PyYAML
+    httplib2

vcr/patch.py

@@ -24,6 +24,15 @@ try:
 except ImportError:  # pragma: no cover
     pass
 
+try:
+    # Try to save the original types for httplib2
+    import httplib2
+    _HTTPConnectionWithTimeout = httplib2.HTTPConnectionWithTimeout
+    _HTTPSConnectionWithTimeout = httplib2.HTTPSConnectionWithTimeout
+    _SCHEME_TO_CONNECTION = httplib2.SCHEME_TO_CONNECTION
+except ImportError:  # pragma: no cover
+    pass
+
 
 def install(cassette):
     """
@@ -64,6 +73,20 @@ def install(cassette):
     except ImportError:  # pragma: no cover
         pass
 
+    # patch httplib2
+    try:
+        import httplib2 as cpool
+        from .stubs.httplib2_stubs import VCRHTTPConnectionWithTimeout
+        from .stubs.httplib2_stubs import VCRHTTPSConnectionWithTimeout
+        cpool.HTTPConnectionWithTimeout = VCRHTTPConnectionWithTimeout
+        cpool.HTTPSConnectionWithTimeout = VCRHTTPSConnectionWithTimeout
+        cpool.SCHEME_TO_CONNECTION = {
+            'http': VCRHTTPConnectionWithTimeout,
+            'https': VCRHTTPSConnectionWithTimeout
+        }
+    except ImportError:  # pragma: no cover
+        pass
+
 
 def reset():
     '''Undo all the patching'''
@@ -91,3 +114,11 @@ def reset():
         cpool.HTTPSConnectionPool.ConnectionCls = _HTTPSConnection
     except ImportError:  # pragma: no cover
         pass
+
+    try:
+        import httplib2 as cpool
+        cpool.HTTPConnectionWithTimeout = _HTTPConnectionWithTimeout
+        cpool.HTTPSConnectionWithTimeout = _HTTPSConnectionWithTimeout
+        cpool.SCHEME_TO_CONNECTION = _SCHEME_TO_CONNECTION
+    except ImportError:  # pragma: no cover
+        pass

vcr/stubs/__init__.py

@@ -1,6 +1,6 @@
 '''Stubs for patching HTTP and HTTPS requests'''
 
-from httplib import HTTPConnection, HTTPSConnection, HTTPMessage
+from httplib import HTTPConnection, HTTPSConnection, HTTPMessage, HTTPResponse
 from cStringIO import StringIO
 
 from vcr.request import Request
@@ -31,7 +31,7 @@ def parse_headers(header_list):
     return msg
 
 
-class VCRHTTPResponse(object):
+class VCRHTTPResponse(HTTPResponse):
     """
     Stub reponse class that gets returned instead of a HTTPResponse
     """
@@ -175,6 +175,28 @@ class VCRConnection:
     def set_debuglevel(self, *args, **kwargs):
         self.real_connection.set_debuglevel(*args, **kwargs)
 
+    def connect(self):
+        """
+        httplib2 uses this.  Connects to the server I'm assuming.
+
+        Only pass to the baseclass if we don't have a recorded response
+        and are not write-protected.
+        """
+
+        if hasattr(self, '_vcr_request') and \
+                self._vcr_request in self.cassette and \
+                self.cassette.record_mode != "all" and \
+                self.cassette.rewound:
+            # We already have a response we are going to play, don't
+            # actually connect
+            return 
+        
+        if self.cassette.write_protected:
+            # Cassette is write-protected, don't actually connect
+            return
+
+        return self.real_connection.connect(self)
+
     def __init__(self, *args, **kwargs):
         # need to temporarily reset here because the real connection
         # inherits from the thing that we are mocking out.  Take out

vcr/stubs/httplib2_stubs.py

@@ -0,0 +1,51 @@
+'''Stubs for httplib2'''
+
+from httplib2 import HTTPConnectionWithTimeout, HTTPSConnectionWithTimeout
+from ..stubs import VCRHTTPConnection, VCRHTTPSConnection
+
+
+class VCRHTTPConnectionWithTimeout(VCRHTTPConnection,
+                                   HTTPConnectionWithTimeout):
+    _baseclass = HTTPConnectionWithTimeout
+
+    def __init__(self, *args, **kwargs):
+        '''I overrode the init because I need to clean kwargs before calling
+        HTTPConnection.__init__.'''
+
+        # Delete the keyword arguments that HTTPConnection would not recognize
+        safe_keys = set(('host', 'port', 'strict', 'timeout', 'source_address'))
+        unknown_keys = set(kwargs.keys()) - safe_keys
+        safe_kwargs = kwargs.copy()
+        for kw in unknown_keys:
+            del safe_kwargs[kw]
+
+        self.proxy_info = kwargs.pop('proxy_info', None)
+        VCRHTTPConnection.__init__(self, *args, **safe_kwargs)
+
+
+class VCRHTTPSConnectionWithTimeout(VCRHTTPSConnection,
+                                    HTTPSConnectionWithTimeout):
+    _baseclass = HTTPSConnectionWithTimeout
+
+    def __init__(self, *args, **kwargs):
+
+        # Delete the keyword arguments that HTTPSConnection would not recognize
+        safe_keys = set(('host', 'port', 'key_file', 'cert_file', 'strict',
+                     'timeout', 'source_address'))
+        unknown_keys = set(kwargs.keys()) - safe_keys
+        safe_kwargs = kwargs.copy()
+        for kw in unknown_keys:
+            del safe_kwargs[kw]
+        self.proxy_info = kwargs.pop('proxy_info', None)
+        if not 'ca_certs' in kwargs or kwargs['ca_certs'] is None:
+            try:
+                import httplib2
+                self.ca_certs = httplib2.CA_CERTS
+            except ImportError:
+                self.ca_certs = None
+        else:
+            self.ca_certs = kwargs['ca_certs']
+ 
+        self.disable_ssl_certificate_validation = kwargs.pop(
+            'disable_ssl_certificate_validation', None)
+        VCRHTTPSConnection.__init__(self, *args, **safe_kwargs)