Add Support for Requests Library

Added support and tests for the Requests library. At least basic support seems to be working. The tests should run fine if you don't have requests installed; it just won't verify the requests- specific tests.
2025-12-09 01:03:24 +00:00 · 2012-09-05 20:43:39 -10:00
parent e19d82aa03
commit 41da9ddbab
8 changed files with 173 additions and 79 deletions
--- a/setup.py
+++ b/setup.py
@@ -8,4 +8,4 @@ setup(name='VCR.py',
      author='Kevin McCarthy',
      author_email='mc@kevinmccarthy.org',
      packages=['vcr'],
-     )
+      )
--- a/test.py
+++ b/test.py
@@ -5,9 +5,9 @@ import vcr
 from vcr.cassette import Cassette
 import urllib2
 from urllib import urlencode
-import requests

-TEST_CASSETTE_FILE = 'test/test_req.yaml'
+TEST_CASSETTE_FILE = 'cassettes/test_req.yaml'
+

 class TestHttpRequest(unittest.TestCase):

@@ -98,79 +98,10 @@ class TestCassette(unittest.TestCase):
        self.assertEqual(c1.requests, c2.requests)
        self.assertEqual(c1.responses, c2.responses)

-class TestRequestsGet(unittest.TestCase):
-
-    def setUp(self):
-        self.unmolested_response = requests.get('http://httpbin.org/')
-        with vcr.use_cassette(TEST_CASSETTE_FILE):
-            self.initial_response = requests.get('http://httpbin.org/')
-            self.cached_response = requests.get('http://httpbin.org/')
-
-    def tearDown(self):
-        try:
-            os.remove(TEST_CASSETTE_FILE)
-        except OSError:
-            pass
-
-    def test_initial_response_code(self):
-        self.assertEqual(self.unmolested_response.status_code, self.initial_response.status_code)
-
-    def test_cached_response_code(self):
-        self.assertEqual(self.unmolested_response.status_code, self.cached_response.status_code)
-
-    def test_initial_response_headers(self):
-        self.assertEqual(self.unmolested_response.headers['content-type'], self.initial_response.headers['content-type'])
-
-    def test_cached_response_headers(self):
-        self.assertEqual(self.unmolested_response.headers['content-type'], self.cached_response.headers['content-type'])
-
-    def test_initial_response_text(self):
-        self.assertEqual(self.unmolested_response.text, self.initial_response.text)
-
-    def test_cached_response_text(self):
-        self.assertEqual(self.unmolested_response.text, self.cached_response.text)
-
-class TestRequestsPost(unittest.TestCase):
-    def setUp(self):
-        payload = {'key1': 'value1', 'key2': 'value2'}
-        self.unmolested_response = requests.post('http://httpbin.org/post', payload)
-        with vcr.use_cassette(TEST_CASSETTE_FILE):
-            self.initial_response = requests.post('http://httpbin.org/post', payload)
-            self.cached_response = requests.post('http://httpbin.org/post', payload)
-
-    def tearDown(self):
-        try:
-            os.remove(TEST_CASSETTE_FILE)
-        except OSError:
-            pass
-
-    def test_initial_post_response_text(self):
-        self.assertEqual(self.unmolested_response.text, self.initial_response.text)
-
-    def test_cached_post_response_text(self):
-        self.assertEqual(self.unmolested_response.text, self.cached_response.text)
-
-class TestRequestsHTTPS(unittest.TestCase):
-    def setUp(self):
-        self.unmolested_response = requests.get('https://github.com')
-        with vcr.use_cassette(TEST_CASSETTE_FILE):
-            self.initial_response = requests.get('https://github.com')
-            self.cached_response = requests.get('https://github.com')
-
-    def tearDown(self):
-        try:
-            os.remove(TEST_CASSETTE_FILE)
-        except OSError:
-            pass
-
-    def test_initial_https_response_text(self):
-        self.assertEqual(self.unmolested_response.text, self.initial_response.text)
-
-    def test_cached_https_response_text(self):
-        self.assertEqual(self.unmolested_response.text, self.cached_response.text)
-
-    
-
+try:
+    from test_requests import *
+except ImportError:
+    pass

 if __name__ == '__main__':
    unittest.main()
--- a/test_requests.py
+++ b/test_requests.py
@@ -0,0 +1,109 @@
+# coding=utf-8
+import os
+import unittest
+import vcr
+import requests
+
+TEST_CASSETTE_FILE = 'cassettes/test_req.yaml'
+
+
+class TestRequestsGet(unittest.TestCase):
+
+    def setUp(self):
+        self.unmolested_response = requests.get('http://httpbin.org/')
+        with vcr.use_cassette(TEST_CASSETTE_FILE):
+            self.initial_response = requests.get('http://httpbin.org/')
+            self.cached_response = requests.get('http://httpbin.org/')
+
+    def tearDown(self):
+        try:
+            os.remove(TEST_CASSETTE_FILE)
+        except OSError:
+            pass
+
+    def test_initial_response_code(self):
+        self.assertEqual(self.unmolested_response.status_code, self.initial_response.status_code)
+
+    def test_cached_response_code(self):
+        self.assertEqual(self.unmolested_response.status_code, self.cached_response.status_code)
+
+    def test_initial_response_headers(self):
+        self.assertEqual(self.unmolested_response.headers['content-type'], self.initial_response.headers['content-type'])
+
+    def test_cached_response_headers(self):
+        self.assertEqual(self.unmolested_response.headers['content-type'], self.cached_response.headers['content-type'])
+
+    def test_initial_response_text(self):
+        self.assertEqual(self.unmolested_response.text, self.initial_response.text)
+
+    def test_cached_response_text(self):
+        self.assertEqual(self.unmolested_response.text, self.cached_response.text)
+
+
+class TestRequestsAuth(unittest.TestCase):
+
+    def setUp(self):
+        self.unmolested_response = requests.get('https://httpbin.org/basic-auth/user/passwd', auth=('user', 'passwd'))
+        with vcr.use_cassette(TEST_CASSETTE_FILE):
+            self.initial_response = requests.get('https://httpbin.org/basic-auth/user/passwd', auth=('user', 'passwd'))
+            self.cached_response = requests.get('https://httpbin.org/basic-auth/user/passwd', auth=('user', 'passwd'))
+
+    def tearDown(self):
+        try:
+            os.remove(TEST_CASSETTE_FILE)
+        except OSError:
+            pass
+
+    def test_initial_response_code(self):
+        self.assertEqual(self.unmolested_response.status_code, self.initial_response.status_code)
+
+    def test_cached_response_code(self):
+        self.assertEqual(self.unmolested_response.status_code, self.cached_response.status_code)
+
+    def test_cached_response_auth_can_fail(self):
+        auth_fail_cached = requests.get('https://httpbin.org/basic-auth/user/passwd', auth=('user', 'passwdzzz'))
+        self.assertNotEqual(self.unmolested_response.status_code, auth_fail_cached.status_code)
+
+
+class TestRequestsPost(unittest.TestCase):
+
+    def setUp(self):
+        payload = {'key1': 'value1', 'key2': 'value2'}
+        self.unmolested_response = requests.post('http://httpbin.org/post', payload)
+        with vcr.use_cassette(TEST_CASSETTE_FILE):
+            self.initial_response = requests.post('http://httpbin.org/post', payload)
+            self.cached_response = requests.post('http://httpbin.org/post', payload)
+
+    def tearDown(self):
+        try:
+            os.remove(TEST_CASSETTE_FILE)
+        except OSError:
+            pass
+
+    def test_initial_post_response_text(self):
+        self.assertEqual(self.unmolested_response.text, self.initial_response.text)
+
+    def test_cached_post_response_text(self):
+        self.assertEqual(self.unmolested_response.text, self.cached_response.text)
+
+
+class TestRequestsHTTPS(unittest.TestCase):
+    maxDiff = None
+
+    def setUp(self):
+        self.unmolested_response = requests.get('https://httpbin.org/get')
+        with vcr.use_cassette(TEST_CASSETTE_FILE):
+            self.initial_response = requests.get('https://httpbin.org/get')
+            self.cached_response = requests.get('https://httpbin.org/get')
+
+    def tearDown(self):
+        try:
+            os.remove(TEST_CASSETTE_FILE)
+        except OSError:
+            pass
+
+    def test_initial_https_response_text(self):
+        self.assertEqual(self.unmolested_response.text, self.initial_response.text)
+
+    def test_cached_https_response_text(self):
+        self.assertEqual(self.unmolested_response.text, self.cached_response.text)
--- a/test_requirements.txt
+++ b/test_requirements.txt
@@ -1,2 +0,0 @@
-PyYAML
-requests
--- a/tox.ini
+++ b/tox.ini
@@ -4,11 +4,28 @@
 # and then run "tox" from this directory.

 [tox]
-envlist = py26, py27, pypy
+envlist = py26, py27, pypy, py26requests, py27requests, pypyrequests

 [testenv]
 commands =
    python test.py
+deps =
+    PyYAML
+
+[testenv:py26requests]
+basepython = python2.6
+deps =
+    PyYAML
+    requests
+
+[testenv:py27requests]
+basepython = python2.7
+deps =
+    PyYAML
+    requests
+
+[testenv:pypyrequests]
+basepython = pypy
 deps =
    PyYAML
    requests
--- a/vcr/patch.py
+++ b/vcr/patch.py
@@ -2,21 +2,40 @@ import httplib
 from contextlib import contextmanager
 from .stubs import VCRHTTPConnection, VCRHTTPSConnection

+
 _HTTPConnection = httplib.HTTPConnection
 _HTTPSConnection = httplib.HTTPSConnection

+try:
+    import requests.packages.urllib3.connectionpool
+    _VerifiedHTTPSConnection = requests.packages.urllib3.connectionpool.VerifiedHTTPSConnection 
+except ImportError:
+    pass
+

 def install(cassette_path):
    httplib.HTTPConnection = httplib.HTTP._connection_class = VCRHTTPConnection
    httplib.HTTPSConnection = httplib.HTTPS._connection_class = VCRHTTPSConnection
    httplib.HTTPConnection._vcr_cassette_path = cassette_path
    httplib.HTTPSConnection._vcr_cassette_path = cassette_path
+    try:
+        import requests.packages.urllib3.connectionpool
+        from .requests_stubs import VCRVerifiedHTTPSConnection
+        requests.packages.urllib3.connectionpool.VerifiedHTTPSConnection = VCRVerifiedHTTPSConnection
+        requests.packages.urllib3.connectionpool.VerifiedHTTPSConnection._vcr_cassette_path = cassette_path
+    except ImportError:
+        pass


 def reset():
    httplib.HTTPConnection = httplib.HTTP._connection_class = _HTTPConnection
    httplib.HTTPSConnection = httplib.HTTPS._connection_class = \
            _HTTPSConnection
+    try:
+        import requests.packages.urllib3.connectionpool
+        requests.packages.urllib3.connectionpool.VerifiedHTTPSConnection = _VerifiedHTTPSConnection
+    except ImportError:
+        pass


@contextmanager
--- a/vcr/requests_stubs.py
+++ b/vcr/requests_stubs.py
@@ -0,0 +1,7 @@
+from requests.packages.urllib3.connectionpool import VerifiedHTTPSConnection
+from .stubs import VCRHTTPSConnection
+
+class VCRVerifiedHTTPSConnection(VCRHTTPSConnection, VerifiedHTTPSConnection):
+
+    _baseclass = VerifiedHTTPSConnection
+
--- a/vcr/stubs.py
+++ b/vcr/stubs.py
@@ -12,15 +12,28 @@ class VCRHTTPResponse(object):
        self.recorded_response = recorded_response
        self.reason = recorded_response['status']['message']
        self.status = recorded_response['status']['code']
+        self.version = None
        self._content = StringIO(self.recorded_response['body']['string'])

        self.msg = HTTPMessage(StringIO(''))
        for k, v in self.recorded_response['headers'].iteritems():
            self.msg.addheader(k, v)

+        self.length = self.msg.getheader('content-length') or None
+
    def read(self, chunked=False):
+        # Note: I'm pretty much ignoring any chunking stuff because
+        # I don't really understand what it is or how it works.
        return self._content.read()

+    def isclosed(self):
+        # Urllib3 seems to call this because it actually uses
+        # the weird chunking support in httplib
+        return True
+
+    def getheaders(self):
+        return self.recorded_response['headers'].iteritems()
+

 class VCRConnectionMixin:
    def _load_old_response(self):