Skip to content

HTTPS clone URL

Subversion checkout URL

You can clone with HTTPS or Subversion.

Download ZIP
Browse files

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.
  • Loading branch information...
commit 41da9ddbabc87f81e2ea2c7d1836517497f21426 1 parent e19d82a
@kevin1024 authored
View
2  setup.py
@@ -8,4 +8,4 @@
author='Kevin McCarthy',
author_email='mc@kevinmccarthy.org',
packages=['vcr'],
- )
+ )
View
81 test.py
@@ -5,9 +5,9 @@
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 @@ def test_serialize_cassette(self):
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()
View
109 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)
View
2  test_requirements.txt
@@ -1,2 +0,0 @@
-PyYAML
-requests
View
19 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
View
19 vcr/patch.py
@@ -2,21 +2,40 @@
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
View
7 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
+
View
13 vcr/stubs.py
@@ -12,15 +12,28 @@ def __init__(self, recorded_response):
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):
Please sign in to comment.
Something went wrong with that request. Please try again.