diff --git a/.gitignore b/.gitignore
index 4141e94..4c946c7 100644
--- a/.gitignore
+++ b/.gitignore
@@ -1,2 +1,7 @@
*.pyc
.tox
+build/
+dist/
+*.egg/
+.coverage
+*.egg-info/
\ No newline at end of file
diff --git a/.travis.yml b/.travis.yml
index 03297a8..2f57a07 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -10,4 +10,4 @@ python:
install:
- pip install PyYAML pytest --use-mirrors
- if [ $WITH_REQUESTS = "True" ] ; then pip install requests; fi
-script: py.test
+script: pwd; py.test --basetemp=.
diff --git a/setup.py b/setup.py
index 1bb3c0b..5b345c0 100644
--- a/setup.py
+++ b/setup.py
@@ -19,12 +19,17 @@ class PyTest(TestCommand):
sys.exit(errno)
setup(name='vcrpy',
- version='0.0.4',
+ version='0.0.5',
description="A Python port of Ruby's VCR to make mocking HTTP easier",
author='Kevin McCarthy',
author_email='me@kevinmccarthy.org',
url='https://github.com/kevin1024/vcrpy',
- packages=['vcr'],
+ packages=[
+ 'vcr',
+ 'vcr.stubs'],
+ package_dir={
+ 'vcr': 'vcr',
+ 'vcr.stubs': 'vcr/stubs'},
install_requires=['PyYAML'],
license='MIT',
tests_require=['pytest'],
diff --git a/tests/common.py b/tests/common.py
new file mode 100644
index 0000000..b978267
--- /dev/null
+++ b/tests/common.py
@@ -0,0 +1,30 @@
+# coding=utf-8
+
+import os
+import json
+import shutil
+import unittest
+
+
+class TestVCR(unittest.TestCase):
+ fixtures = os.path.join('does', 'not', 'exist')
+
+ def tearDown(self):
+ # Remove th urllib2 fixtures if they exist
+ if os.path.exists(self.fixtures):
+ shutil.rmtree(self.fixtures)
+
+ def fixture(self, *names):
+ '''Return a path to the provided fixture'''
+ return os.path.join(self.fixtures, *names)
+
+ def assertBodiesEqual(self, one, two):
+ """
+ httpbin.org returns a different `origin` header
+ each time, so strip this out since it makes testing
+ difficult.
+ """
+ one, two = json.loads(one), json.loads(two)
+ del one['origin']
+ del two['origin']
+ self.assertEqual(one, two)
diff --git a/tests/fixtures/wild/domain_redirect.yaml b/tests/fixtures/wild/domain_redirect.yaml
new file mode 100644
index 0000000..aaf4134
--- /dev/null
+++ b/tests/fixtures/wild/domain_redirect.yaml
@@ -0,0 +1,37 @@
+- request:
+ body: null
+ headers: !!python/object:requests.structures.CaseInsensitiveDict
+ _store:
+ accept: !!python/tuple [Accept, '*/*']
+ accept-encoding: !!python/tuple [Accept-Encoding, 'gzip, deflate, compress']
+ user-agent: !!python/tuple [User-Agent, vcrpy-test]
+ host: seomoz.org
+ method: GET
+ port: 80
+ url: /
+ response:
+ body: {string: "\r\n301 Moved Permanently\r\n\r\n301 Moved Permanently
\r\n
nginx\r\n\r\n\r\n"}
+ headers: {accept-ranges: bytes, age: '0', connection: keep-alive, content-length: '178',
+ content-type: text/html, date: 'Mon, 05 Aug 2013 22:05:23 GMT', location: 'http://moz.com/',
+ server: nginx, server-name: dalmozwww03.dal.moz.com, via: 1.1 varnish, x-varnish: '836826051'}
+ status: {code: 301, message: Moved Permanently}
+- request:
+ body: null
+ headers: !!python/object:requests.structures.CaseInsensitiveDict
+ _store:
+ accept: !!python/tuple [Accept, '*/*']
+ accept-encoding: !!python/tuple [Accept-Encoding, 'gzip, deflate, compress']
+ user-agent: !!python/tuple [User-Agent, vcrpy-test]
+ host: moz.com
+ method: GET
+ port: 80
+ url: /
+ response:
+ body: {string: ''}
+ headers: {accept-ranges: bytes, age: '2763', cache-control: 'no-cache, must-revalidate,
+ s-maxage=3600', connection: keep-alive, content-encoding: gzip, content-length: '0',
+ content-type: text/html, date: 'Mon, 05 Aug 2013 22:05:23 GMT', expires: 'Fri,
+ 15 Oct 2004 12:00:00 GMT', server: nginx, server-name: dalmozwww02.dal.moz.com,
+ vary: Accept-Encoding, via: 1.1 varnish, x-varnish: 836826061 836743142}
+ status: {code: 200, message: OK}
diff --git a/tests/test_basic.py b/tests/test_basic.py
index 41b698e..dc3382c 100644
--- a/tests/test_basic.py
+++ b/tests/test_basic.py
@@ -1,107 +1,34 @@
+'''Basic tests about cassettes'''
# coding=utf-8
-import os
-import unittest
+
+# Internal imports
import vcr
-from vcr.cassette import Cassette
+from .common import TestVCR
+
+# External imports
+import os
import urllib2
-from urllib import urlencode
-from utils import assert_httpbin_responses_equal
-
-TEST_CASSETTE_FILE = 'cassettes/test_req.yaml'
-
-class TestHttpRequest(unittest.TestCase):
-
- def tearDown(self):
- try:
- os.remove(TEST_CASSETTE_FILE)
- except OSError:
- pass
-
- def test_response_code(self):
- code = urllib2.urlopen('http://httpbin.org/').getcode()
- with vcr.use_cassette(TEST_CASSETTE_FILE):
- self.assertEqual(code, urllib2.urlopen('http://httpbin.org/').getcode())
- self.assertEqual(code, urllib2.urlopen('http://httpbin.org/').getcode())
-
- def test_response_body(self):
- body = urllib2.urlopen('http://httpbin.org/').read()
- with vcr.use_cassette(TEST_CASSETTE_FILE):
- self.assertEqual(body, urllib2.urlopen('http://httpbin.org/').read())
- self.assertEqual(body, urllib2.urlopen('http://httpbin.org/').read())
-
- def test_response_headers(self):
- with vcr.use_cassette(TEST_CASSETTE_FILE):
- headers = urllib2.urlopen('http://httpbin.org/').info().items()
- self.assertEqual(headers, urllib2.urlopen('http://httpbin.org/').info().items())
-
- def test_multiple_requests(self):
- body1 = urllib2.urlopen('http://httpbin.org/').read()
- body2 = urllib2.urlopen('http://httpbin.org/get').read()
- with vcr.use_cassette(TEST_CASSETTE_FILE):
- self.assertEqual(body1, urllib2.urlopen('http://httpbin.org/').read())
- new_body2 = urllib2.urlopen('http://httpbin.org/get').read()
-
- assert_httpbin_responses_equal(body2, new_body2)
-
- self.assertEqual(body1, urllib2.urlopen('http://httpbin.org/').read())
- new_body2 = urllib2.urlopen('http://httpbin.org/get').read()
-
- assert_httpbin_responses_equal(body2, new_body2)
-
-class TestHttps(unittest.TestCase):
-
- def tearDown(self):
- try:
- os.remove(TEST_CASSETTE_FILE)
- except OSError:
- pass
-
- def test_response_code(self):
- code = urllib2.urlopen('https://httpbin.org/').getcode()
- with vcr.use_cassette(TEST_CASSETTE_FILE):
- self.assertEqual(code, urllib2.urlopen('https://httpbin.org/').getcode())
- self.assertEqual(code, urllib2.urlopen('https://httpbin.org/').getcode())
-
- def test_response_body(self):
- body = urllib2.urlopen('https://httpbin.org/').read()
- with vcr.use_cassette(TEST_CASSETTE_FILE):
- self.assertEqual(body, urllib2.urlopen('https://httpbin.org/').read())
- self.assertEqual(body, urllib2.urlopen('https://httpbin.org/').read())
-
- def test_response_headers(self):
- with vcr.use_cassette(TEST_CASSETTE_FILE):
- headers = urllib2.urlopen('https://httpbin.org/').info().items()
- self.assertEqual(headers, urllib2.urlopen('https://httpbin.org/').info().items())
-
- def test_get_data(self):
- TEST_DATA = urlencode({'some': 1, 'data': 'here'})
- with vcr.use_cassette(TEST_CASSETTE_FILE):
- body = urllib2.urlopen('https://httpbin.org/get?' + TEST_DATA).read()
- self.assertEqual(body, urllib2.urlopen('https://httpbin.org/get?' + TEST_DATA).read())
-
- def test_post_data(self):
- TEST_DATA = urlencode({'some': 1, 'data': 'here'})
- with vcr.use_cassette(TEST_CASSETTE_FILE):
- body = urllib2.urlopen('https://httpbin.org/post', TEST_DATA).read()
- self.assertEqual(body, urllib2.urlopen('https://httpbin.org/post', TEST_DATA).read())
-
- def test_post_unicode(self):
- TEST_DATA = urlencode({'snowman': u'☃'.encode('utf-8')})
- with vcr.use_cassette(TEST_CASSETTE_FILE):
- body = urllib2.urlopen('https://httpbin.org/post', TEST_DATA).read()
- self.assertEqual(body, urllib2.urlopen('https://httpbin.org/post', TEST_DATA).read())
-class TestCassette(unittest.TestCase):
- def test_serialize_cassette(self):
- c1 = Cassette()
- c1.requests = ['a', 'b', 'c']
- c1.responses = ['d', 'e', 'f']
- ser = c1.serialize()
- c2 = Cassette(ser)
- self.assertEqual(c1.requests, c2.requests)
- self.assertEqual(c1.responses, c2.responses)
+class TestCassette(TestVCR):
+ '''We should be able to save a cassette'''
+ fixtures = os.path.join('tests', 'fixtures', 'basic')
+ def test_nonexistent_directory(self):
+ '''If we load a cassette in a nonexistent directory, it can save ok'''
+ self.assertFalse(os.path.exists(self.fixture('nonexistent')))
+ with vcr.use_cassette(self.fixture('nonexistent', 'cass.yaml')):
+ urllib2.urlopen('http://httpbin.org/').read()
+ # This should have made the file and the directory
+ self.assertTrue(
+ os.path.exists(self.fixture('nonexistent', 'cass.yaml')))
-if __name__ == '__main__':
- unittest.main()
+ def test_unpatch(self):
+ '''Ensure that our cassette gets unpatched when we're done'''
+ with vcr.use_cassette(self.fixture('unpatch.yaml')) as cass:
+ urllib2.urlopen('http://httpbin.org/').read()
+
+ # Make the same requests, and assert that we haven't served any more
+ # requests out of cache
+ urllib2.urlopen('http://httpbin.org/').read()
+ self.assertEqual(len(cass.cached()), 0)
diff --git a/tests/test_requests.py b/tests/test_requests.py
index db90ed6..52d2ef6 100644
--- a/tests/test_requests.py
+++ b/tests/test_requests.py
@@ -1,110 +1,163 @@
-# coding=utf-8
-import os
-import unittest
-import vcr
-import pytest
-from utils import assert_httpbin_responses_equal
+'''Test requests' interaction with vcr'''
+# coding=utf-8
+
+# Internal imports
+import vcr
+from .common import TestVCR
+
+import os
+import pytest
requests = pytest.importorskip("requests")
-TEST_CASSETTE_FILE = 'cassettes/test_req.yaml'
-class TestRequestsGet(unittest.TestCase):
+class TestRequestsBase(TestVCR):
+ '''Some utility for running Requests tests'''
+ fixtures = os.path.join('tests', 'fixtures', 'requests')
- def setUp(self):
- self.unmolested_response = requests.get('http://httpbin.org/get')
- with vcr.use_cassette(TEST_CASSETTE_FILE):
- self.initial_response = requests.get('http://httpbin.org/get')
- self.cached_response = requests.get('http://httpbin.org/get')
+
+class TestHTTPRequests(TestRequestsBase):
+ '''Some tests using requests and http'''
+ scheme = 'http'
+
+ def test_status_code(self):
+ '''Ensure that we can read the status code'''
+ url = self.scheme + '://httpbin.org/'
+ with vcr.use_cassette(self.fixture('atts.yaml')) as cass:
+ # Ensure that this is empty to begin with
+ self.assertEqual(len(cass), 0)
+ self.assertEqual(len(cass.cached()), 0)
+ self.assertEqual(
+ requests.get(url).status_code,
+ requests.get(url).status_code)
+ # Ensure that we've now cached a single response
+ self.assertEqual(len(cass), 1)
+ self.assertEqual(len(cass.cached()), 1)
+
+ def test_headers(self):
+ '''Ensure that we can read the headers back'''
+ url = self.scheme + '://httpbin.org/'
+ with vcr.use_cassette(self.fixture('headers.yaml')) as cass:
+ # Ensure that this is empty to begin with
+ self.assertEqual(len(cass), 0)
+ self.assertEqual(len(cass.cached()), 0)
+ self.assertEqual(
+ requests.get(url).headers,
+ requests.get(url).headers)
+ # Ensure that we've now cached a single response
+ self.assertEqual(len(cass), 1)
+ self.assertEqual(len(cass.cached()), 1)
+
+ def test_body(self):
+ '''Ensure the responses are all identical enough'''
+ url = self.scheme + '://httpbin.org/bytes/1024'
+ with vcr.use_cassette(self.fixture('body.yaml')) as cass:
+ # Ensure that this is empty to begin with
+ self.assertEqual(len(cass), 0)
+ self.assertEqual(len(cass.cached()), 0)
+ self.assertEqual(
+ requests.get(url).content,
+ requests.get(url).content)
+ # Ensure that we've now cached a single response
+ self.assertEqual(len(cass), 1)
+ self.assertEqual(len(cass.cached()), 1)
+
+ def test_auth(self):
+ '''Ensure that we can handle basic auth'''
+ auth = ('user', 'passwd')
+ url = self.scheme + '://httpbin.org/basic-auth/user/passwd'
+ with vcr.use_cassette(self.fixture('auth.yaml')) as cass:
+ # Ensure that this is empty to begin with
+ self.assertEqual(len(cass), 0)
+ self.assertEqual(len(cass.cached()), 0)
+ one = requests.get(url, auth=auth)
+ two = requests.get(url, auth=auth)
+ self.assertEqual(one.content, two.content)
+ self.assertEqual(one.status_code, two.status_code)
+ # Ensure that we've now cached a single response
+ self.assertEqual(len(cass), 1)
+ self.assertEqual(len(cass.cached()), 1)
+
+ def test_auth_failed(self):
+ '''Ensure that we can save failed auth statuses'''
+ auth = ('user', 'wrongwrongwrong')
+ url = self.scheme + '://httpbin.org/basic-auth/user/passwd'
+ with vcr.use_cassette(self.fixture('auth-failed.yaml')) as cass:
+ # Ensure that this is empty to begin with
+ self.assertEqual(len(cass), 0)
+ self.assertEqual(len(cass.cached()), 0)
+ one = requests.get(url, auth=auth)
+ two = requests.get(url, auth=auth)
+ self.assertEqual(one.content, two.content)
+ self.assertEqual(one.status_code, two.status_code)
+ self.assertNotEqual(one.status_code, 200)
+ # Ensure that we've now cached a single response
+ self.assertEqual(len(cass), 1)
+ self.assertEqual(len(cass.cached()), 1)
+
+ def test_post(self):
+ '''Ensure that we can post and cache the results'''
+ data = {'key1': 'value1', 'key2': 'value2'}
+ url = self.scheme + '://httpbin.org/post'
+ with vcr.use_cassette(self.fixture('redirect.yaml')) as cass:
+ # Ensure that this is empty to begin with
+ self.assertEqual(len(cass), 0)
+ self.assertEqual(len(cass.cached()), 0)
+ self.assertEqual(
+ requests.post(url, data).content,
+ requests.post(url, data).content)
+ # Ensure that we've now cached a single response
+ self.assertEqual(len(cass), 1)
+ self.assertEqual(len(cass.cached()), 1)
+
+ def test_redirects(self):
+ '''Ensure that we can handle redirects'''
+ url = self.scheme + '://httpbin.org/redirect-to?url=bytes/1024'
+ with vcr.use_cassette(self.fixture('redirect.yaml')) as cass:
+ # Ensure that this is empty to begin with
+ self.assertEqual(len(cass), 0)
+ self.assertEqual(len(cass.cached()), 0)
+ self.assertEqual(
+ requests.get(url).content,
+ requests.get(url).content)
+ # Ensure that we've now cached /two/ responses. One for the redirect
+ # and one for the final fetch
+ self.assertEqual(len(cass), 2)
+ self.assertEqual(len(cass.cached()), 2)
+
+
+class TestHTTPSRequests(TestHTTPRequests):
+ '''Same as above, now in https'''
+ scheme = 'https'
+
+ def test_cross_scheme(self):
+ '''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(self.fixture('cross_scheme.yaml')) as cass:
+ requests.get('https://httpbin.org/')
+ requests.get('http://httpbin.org/')
+ self.assertEqual(len(cass), 2)
+ self.assertEqual(len(cass.cached()), 0)
+
+
+class TestWild(TestRequestsBase):
+ '''Test some examples from the wild'''
+ fixtures = os.path.join('tests', 'fixtures', 'wild')
def tearDown(self):
- try:
- os.remove(TEST_CASSETTE_FILE)
- except OSError:
- pass
+ # No deleting our directory, and ensure that it exists
+ self.assertTrue(os.path.exists(self.fixture()))
- 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):
- assert_httpbin_responses_equal(self.unmolested_response.text, self.initial_response.text)
-
- def test_cached_response_text(self):
- assert_httpbin_responses_equal(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):
- assert_httpbin_responses_equal(self.unmolested_response.text, self.initial_response.text)
-
- def test_cached_post_response_text(self):
- assert_httpbin_responses_equal(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):
- assert_httpbin_responses_equal(self.unmolested_response.text, self.initial_response.text)
-
- def test_cached_https_response_text(self):
- assert_httpbin_responses_equal(self.unmolested_response.text, self.cached_response.text)
+ def test_domain_redirect(self):
+ '''Ensure that redirects across domains are considered unique'''
+ # In this example, seomoz.org redirects to moz.com, and if those
+ # requests are considered identical, then we'll be stuck in a redirect
+ # loop.
+ url = 'http://seomoz.org/'
+ with vcr.use_cassette(self.fixture('domain_redirect.yaml')) as cass:
+ requests.get(url, headers={'User-Agent': 'vcrpy-test'})
+ # Ensure that we've now served two responses. One for the original
+ # redirect, and a second for the actual fetch
+ self.assertEqual(len(cass.cached()), 2)
diff --git a/tests/test_urllib2.py b/tests/test_urllib2.py
new file mode 100644
index 0000000..f930333
--- /dev/null
+++ b/tests/test_urllib2.py
@@ -0,0 +1,141 @@
+'''Integration tests with urllib2'''
+# coding=utf-8
+
+# Internal imports
+import vcr
+from .common import TestVCR
+
+# Testing urllib2 fetching
+import os
+import urllib2
+from urllib import urlencode
+
+
+class TestUrllib2(TestVCR):
+ '''Base class for tests for urllib2'''
+ fixtures = os.path.join('tests', 'fixtures', 'urllib2')
+
+
+class TestUrllib2Http(TestUrllib2):
+ '''Tests for integration with urllib2'''
+ scheme = 'http'
+
+ def test_response_code(self):
+ '''Ensure we can read a response code from a fetch'''
+ url = self.scheme + '://httpbin.org/'
+ with vcr.use_cassette(self.fixture('atts.yaml')) as cass:
+ # Ensure that this is empty to begin with
+ self.assertEqual(len(cass), 0)
+ self.assertEqual(len(cass.cached()), 0)
+ self.assertEqual(
+ urllib2.urlopen(url).getcode(),
+ urllib2.urlopen(url).getcode())
+ # Ensure that we've now cached a single response
+ self.assertEqual(len(cass), 1)
+ self.assertEqual(len(cass.cached()), 1)
+
+ def test_random_body(self):
+ '''Ensure we can read the content, and that it's served from cache'''
+ url = self.scheme + '://httpbin.org/bytes/1024'
+ with vcr.use_cassette(self.fixture('body.yaml')) as cass:
+ # Ensure that this is empty to begin with
+ self.assertEqual(len(cass), 0)
+ self.assertEqual(len(cass.cached()), 0)
+ self.assertEqual(
+ urllib2.urlopen(url).read(),
+ urllib2.urlopen(url).read())
+ # Ensure that we've now cached a single response
+ self.assertEqual(len(cass), 1)
+ self.assertEqual(len(cass.cached()), 1)
+
+ def test_response_headers(self):
+ '''Ensure we can get information from the response'''
+ url = self.scheme + '://httpbin.org/'
+ with vcr.use_cassette(self.fixture('headers.yaml')) as cass:
+ # Ensure that this is empty to begin with
+ self.assertEqual(len(cass), 0)
+ self.assertEqual(len(cass.cached()), 0)
+ self.assertEqual(
+ urllib2.urlopen(url).info().items(),
+ urllib2.urlopen(url).info().items())
+ # Ensure that we've now cached a single response
+ self.assertEqual(len(cass), 1)
+
+ def test_multiple_requests(self):
+ '''Ensure that we can cache multiple requests'''
+ urls = [
+ self.scheme + '://httpbin.org/',
+ self.scheme + '://httpbin.org/get',
+ self.scheme + '://httpbin.org/bytes/1024'
+ ]
+ with vcr.use_cassette(self.fixture('multiple.yaml')) as cass:
+ for index in range(len(urls)):
+ url = urls[index]
+ self.assertEqual(len(cass), index)
+ self.assertEqual(len(cass.cached()), index)
+ self.assertEqual(
+ urllib2.urlopen(url).read(),
+ urllib2.urlopen(url).read())
+ self.assertEqual(len(cass), index + 1)
+ self.assertEqual(len(cass.cached()), index + 1)
+
+ def test_get_data(self):
+ '''Ensure that it works with query data'''
+ data = urlencode({'some': 1, 'data': 'here'})
+ url = self.scheme + '://httpbin.org/get?' + data
+ with vcr.use_cassette(self.fixture('get_data.yaml')) as cass:
+ # Ensure that this is empty to begin with
+ self.assertEqual(len(cass), 0)
+ self.assertEqual(len(cass.cached()), 0)
+ self.assertEqual(
+ urllib2.urlopen(url).read(),
+ urllib2.urlopen(url).read())
+ # Ensure that we've now cached a single response
+ self.assertEqual(len(cass), 1)
+ self.assertEqual(len(cass.cached()), 1)
+
+ def test_post_data(self):
+ '''Ensure that it works when posting data'''
+ data = urlencode({'some': 1, 'data': 'here'})
+ url = self.scheme + '://httpbin.org/post'
+ with vcr.use_cassette(self.fixture('post_data.yaml')) as cass:
+ # Ensure that this is empty to begin with
+ self.assertEqual(len(cass), 0)
+ self.assertEqual(len(cass.cached()), 0)
+ self.assertEqual(
+ urllib2.urlopen(url, data).read(),
+ urllib2.urlopen(url, data).read())
+ # Ensure that we've now cached a single response
+ self.assertEqual(len(cass), 1)
+ self.assertEqual(len(cass.cached()), 1)
+
+ def test_post_unicode_data(self):
+ '''Ensure that it works when posting unicode data'''
+ data = urlencode({'snowman': u'☃'.encode('utf-8')})
+ url = self.scheme + '://httpbin.org/post'
+ with vcr.use_cassette(self.fixture('post_data.yaml')) as cass:
+ # Ensure that this is empty to begin with
+ self.assertEqual(len(cass), 0)
+ self.assertEqual(len(cass.cached()), 0)
+ self.assertEqual(
+ urllib2.urlopen(url, data).read(),
+ urllib2.urlopen(url, data).read())
+ # Ensure that we've now cached a single response
+ self.assertEqual(len(cass), 1)
+ self.assertEqual(len(cass.cached()), 1)
+
+
+class TestUrllib2Https(TestUrllib2Http):
+ '''Similar tests but for http status codes'''
+ scheme = 'https'
+
+ def test_cross_scheme(self):
+ '''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(self.fixture('cross_scheme.yaml')) as cass:
+ urllib2.urlopen('https://httpbin.org/')
+ urllib2.urlopen('http://httpbin.org/')
+ self.assertEqual(len(cass), 2)
+ self.assertEqual(len(cass.cached()), 0)
diff --git a/vcr/__init__.py b/vcr/__init__.py
index becd6b6..34c261f 100644
--- a/vcr/__init__.py
+++ b/vcr/__init__.py
@@ -1 +1,6 @@
-from .patch import use_cassette
+# Import Cassette to make it available at the top level
+from .cassette import Cassette
+
+# Also, make a 'load' function available
+def use_cassette(path):
+ return Cassette.load(path)
diff --git a/vcr/cassette.py b/vcr/cassette.py
index e101fba..7c07b01 100644
--- a/vcr/cassette.py
+++ b/vcr/cassette.py
@@ -1,27 +1,99 @@
+'''The container for recorded requests and responses'''
+
+import os
+import yaml
+import tempfile
+try:
+ # Use the libYAML versions if possible. They're much faster than the pure
+ # python implemenations
+ from yaml import CLoader as Loader, CDumper as Dumper
+except ImportError: # pragma: no cover
+ from yaml import Loader, Dumper
+
+# Internal imports
+from .patch import install, reset
+
+
class Cassette(object):
- def __init__(self, ser_cassette=None):
- self.requests = []
- self.responses = []
- if ser_cassette:
- self._unserialize(ser_cassette)
+ '''A container for recorded requests and responses'''
+ @classmethod
+ def load(cls, path):
+ '''Load in the cassette stored at the provided path'''
+ try:
+ with open(path) as fin:
+ return cls(path, yaml.load(fin, Loader=Loader))
+ except IOError:
+ return cls(path)
+
+ def __init__(self, path, data=None):
+ self._path = path
+ self._cached = []
+ self._requests = []
+ self._responses = []
+ if data:
+ self.deserialize(data)
+
+ def save(self, path):
+ '''Save this cassette to a path'''
+ dirname, filename = os.path.split(path)
+ if not os.path.exists(dirname):
+ os.makedirs(dirname)
+ # We'll overwrite the old version securely by writing out a temporary
+ # version and then moving it to replace the old version
+ fd, name = tempfile.mkstemp(dir=dirname, prefix=filename)
+ with os.fdopen(fd, 'w') as fout:
+ fout.write(yaml.dump(self.serialize(), Dumper=Dumper))
+ os.rename(name, path)
def serialize(self):
+ '''Return a serializable version of the cassette'''
return ([{
'request': req,
'response': res,
- } for req, res in zip(self.requests, self.responses)])
+ } for req, res in zip(self._requests, self._responses)])
- def _unserialize(self, source):
- self.requests, self.responses = [r['request'] for r in source], [r['response'] for r in source]
+ def deserialize(self, source):
+ '''Given a seritalized version, load the requests'''
+ self._requests, self._responses = (
+ [r['request'] for r in source], [r['response'] for r in source])
- def get_request(self, match):
+ def cached(self, request=None):
+ '''Alert the cassete of a request that's been cached, or get the
+ requests that we've cached'''
+ if request:
+ self._cached.append(request)
+ else:
+ return self._cached
+
+ def append(self, request, response):
+ '''Add a pair of request, response pairs to this cassette'''
+ self._requests.append(request)
+ self._responses.append(response)
+
+ def __len__(self):
+ '''Return the number of request / response pairs stored in here'''
+ return len(self._requests)
+
+ def __contains__(self, request):
+ '''Return whether or not a request has been stored'''
try:
- return self.requests[self.requests.index(match)]
+ self._requests.index(request)
+ return True
+ except ValueError:
+ return False
+
+ def response(self, request):
+ '''Find the response corresponding to a request'''
+ try:
+ return self._responses[self._requests.index(request)]
except ValueError:
return None
- def get_response(self, match):
- try:
- return self.responses[self.requests.index(match)]
- except ValueError:
- return None
+ def __enter__(self):
+ '''Patch the fetching libraries we know about'''
+ install(self)
+ return self
+
+ def __exit__(self, typ, value, traceback):
+ self.save(self._path)
+ reset()
diff --git a/vcr/files.py b/vcr/files.py
deleted file mode 100644
index 95c785f..0000000
--- a/vcr/files.py
+++ /dev/null
@@ -1,26 +0,0 @@
-import os
-import yaml
-from .cassette import Cassette
-
-# Use the libYAML versions if possible
-try:
- from yaml import CLoader as Loader, CDumper as Dumper
-except ImportError:
- from yaml import Loader, Dumper
-
-
-def load_cassette(cassette_path):
- try:
- pc = yaml.load(open(cassette_path), Loader=Loader)
- cassette = Cassette(pc)
- return cassette
- except IOError:
- return None
-
-
-def save_cassette(cassette_path, cassette):
- dirname, filename = os.path.split(cassette_path)
- if not os.path.exists(dirname):
- os.makedirs(dirname)
- with open(cassette_path, 'a') as cassette_file:
- cassette_file.write(yaml.dump(cassette.serialize(), Dumper=Dumper))
diff --git a/vcr/patch.py b/vcr/patch.py
index 82bc3d8..b8d4139 100644
--- a/vcr/patch.py
+++ b/vcr/patch.py
@@ -1,75 +1,75 @@
+'''Utilities for patching in cassettes'''
+
import httplib
-from contextlib import contextmanager
from .stubs import VCRHTTPConnection, VCRHTTPSConnection
+# Save some of the original types for the purposes of unpatching
_HTTPConnection = httplib.HTTPConnection
_HTTPSConnection = httplib.HTTPSConnection
try:
- import requests.packages.urllib3.connectionpool
- _VerifiedHTTPSConnection = requests.packages.urllib3.connectionpool.VerifiedHTTPSConnection
- _HTTPConnection = requests.packages.urllib3.connectionpool.HTTPConnection
-except ImportError:
+ # Try to save the original types for requests
+ import requests.packages.urllib3.connectionpool as cpool
+ _VerifiedHTTPSConnection = cpool.VerifiedHTTPSConnection
+ _HTTPConnection = cpool.HTTPConnection
+except ImportError: # pragma: no cover
pass
try:
+ # Try to save the original types for urllib3
import urllib3
- _VerifiedHTTPSConnection = urllib3.connectionpool.VerifiedHTTPSConnection
-except ImportError:
+ _VerifiedHTTPSConnection = urllib3.connectionpool.VerifiedHTTPSConnection
+except ImportError: # pragma: no cover
pass
-def install(cassette_path):
+def install(cassette):
+ '''Install a cassette in lieu of actuall fetching'''
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
+ httplib.HTTPSConnection = httplib.HTTPS._connection_class = (
+ VCRHTTPSConnection)
+ httplib.HTTPConnection.cassette = cassette
+ httplib.HTTPSConnection.cassette = cassette
- #patch requests
+ # patch requests
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
- requests.packages.urllib3.connectionpool.HTTPConnection = VCRHTTPConnection
- requests.packages.urllib3.connectionpool.HTTPConnection._vcr_cassette_path = cassette_path
- except ImportError:
+ import requests.packages.urllib3.connectionpool as cpool
+ from .stubs.requests_stubs import VCRVerifiedHTTPSConnection
+ cpool.VerifiedHTTPSConnection = VCRVerifiedHTTPSConnection
+ cpool.VerifiedHTTPSConnection.cassette = cassette
+ cpool.HTTPConnection = VCRHTTPConnection
+ cpool.HTTPConnection.cassette = cassette
+ except ImportError: # pragma: no cover
pass
- #patch urllib3
+ # patch urllib3
try:
- import urllib3.connectionpool
- from .urllib3_stubs import VCRVerifiedHTTPSConnection
- urllib3.connectionpool.VerifiedHTTPSConnection = VCRVerifiedHTTPSConnection
- urllib3.connectionpool.VerifiedHTTPSConnection._vcr_cassette_path = cassette_path
- urllib3.connectionpool.HTTPConnection = VCRHTTPConnection
- urllib3.connectionpool.HTTPConnection._vcr_cassette_path = cassette_path
- except ImportError:
+ import urllib3.connectionpool as cpool
+ from .stubs.urllib3_stubs import VCRVerifiedHTTPSConnection
+ cpool.VerifiedHTTPSConnection = VCRVerifiedHTTPSConnection
+ cpool.VerifiedHTTPSConnection.cassette = cassette
+ cpool.HTTPConnection = VCRHTTPConnection
+ cpool.HTTPConnection.cassette = cassette
+ except ImportError: # pragma: no cover
pass
def reset():
+ '''Unto all the patching'''
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
- requests.packages.urllib3.connectionpool.HTTPConnection = _HTTPConnection
- except ImportError:
+ import requests.packages.urllib3.connectionpool as cpool
+ cpool.VerifiedHTTPSConnection = _VerifiedHTTPSConnection
+ cpool.HTTPConnection = _HTTPConnection
+ except ImportError: # pragma: no cover
pass
try:
- import urllib3.connectionpool
- urllib3.connectionpool.VerifiedHTTPSConnection = _VerifiedHTTPSConnection
- urllib3.connectionpool.HTTPConnection = _HTTPConnection
- except ImportError:
+ import urllib3.connectionpool as cpool
+ cpool.VerifiedHTTPSConnection = _VerifiedHTTPSConnection
+ cpool.HTTPConnection = _HTTPConnection
+ except ImportError: # pragma: no cover
pass
-
-
-@contextmanager
-def use_cassette(cassette_path):
- install(cassette_path)
- yield
- reset()
diff --git a/vcr/stubs.py b/vcr/stubs.py
deleted file mode 100644
index 898e350..0000000
--- a/vcr/stubs.py
+++ /dev/null
@@ -1,106 +0,0 @@
-from httplib import HTTPConnection, HTTPSConnection, HTTPMessage
-from cStringIO import StringIO
-from .files import save_cassette, load_cassette
-from .cassette import Cassette
-
-
-class VCRHTTPResponse(object):
- """
- Stub reponse class that gets returned instead of a HTTPResponse
- """
- 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, *args, **kwargs):
- # 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(*args, **kwargs)
-
- def close(self):
- return True
-
- 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):
- old_cassette = load_cassette(self._vcr_cassette_path)
- if old_cassette:
- return old_cassette.get_response(self._vcr)
-
- def request(self, method, url, body=None, headers={}):
- """
- Persist the request metadata in self._vcr
- """
- self._vcr = {
- 'method': method,
- 'url': url,
- 'body': body,
- 'headers': headers,
- }
- old_cassette = load_cassette(self._vcr_cassette_path)
- if old_cassette and old_cassette.get_request(self._vcr):
- return
- self._cassette.requests.append(dict(
- method=method,
- url=url,
- body=body,
- headers=headers
- ))
- self._baseclass.request(self, method, url, body=body, headers=headers)
-
- def getresponse(self, buffering=False):
- old_response = self._load_old_response()
- if not old_response:
- response = HTTPConnection.getresponse(self)
- self._cassette.responses.append({
- 'status': {'code': response.status, 'message': response.reason},
- 'headers': dict(response.getheaders()),
- 'body': {'string': response.read()},
- })
- save_cassette(self._vcr_cassette_path, self._cassette)
- old_response = self._load_old_response()
- return VCRHTTPResponse(old_response)
-
-
-class VCRHTTPConnection(VCRConnectionMixin, HTTPConnection):
-
- # Can't use super since this is an old-style class
- _baseclass = HTTPConnection
-
- def __init__(self, *args, **kwargs):
- self._cassette = Cassette()
- HTTPConnection.__init__(self, *args, **kwargs)
-
-
-class VCRHTTPSConnection(VCRConnectionMixin, HTTPSConnection):
-
- _baseclass = HTTPSConnection
-
- def __init__(self, *args, **kwargs):
- """
- I overrode the init and copied a lot of the code from the parent
- class because HTTPConnection when this happens has been replaced
- by VCRHTTPConnection, but doing it here lets us use the original
- one.
- """
- HTTPConnection.__init__(self, *args, **kwargs)
- self.key_file = kwargs.pop('key_file', None)
- self.cert_file = kwargs.pop('cert_file', None)
- self._cassette = Cassette()
diff --git a/vcr/stubs/__init__.py b/vcr/stubs/__init__.py
new file mode 100644
index 0000000..955c66b
--- /dev/null
+++ b/vcr/stubs/__init__.py
@@ -0,0 +1,106 @@
+'''Stubs for patching HTTP and HTTPS requests'''
+
+from httplib import HTTPConnection, HTTPSConnection, HTTPMessage
+from cStringIO import StringIO
+
+
+class VCRHTTPResponse(object):
+ """
+ Stub reponse class that gets returned instead of a HTTPResponse
+ """
+ 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 key, val in self.recorded_response['headers'].iteritems():
+ self.msg.addheader(key, val)
+
+ self.length = self.msg.getheader('content-length') or None
+
+ def read(self, *args, **kwargs):
+ # 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(*args, **kwargs)
+
+ def close(self):
+ return True
+
+ 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:
+ # A reference to the cassette that's currently being patched in
+ cassette = None
+
+ def request(self, method, url, body=None, headers=None):
+ '''Persist the request metadata in self._vcr'''
+ self._request = {
+ 'host': self.host,
+ 'port': self.port,
+ 'method': method,
+ 'url': url,
+ 'body': body,
+ 'headers': headers or {},
+ }
+ # Check if we have a cassette set, and if we have a response saved.
+ # If so, there's no need to keep processing and we can bail
+ if self.cassette and self._request in self.cassette:
+ return
+
+ # Otherwise, we should submit the request
+ self._baseclass.request(
+ self, method, url, body=body, headers=headers or {})
+
+ def getresponse(self, _=False):
+ '''Retrieve a the response'''
+ # Check to see if the cassette has a response for this request. If so,
+ # then return it
+ response = self.cassette.response(self._request)
+ if response:
+ # Alert the cassette to the fact that we've served another cached
+ # response for the provided requests
+ self.cassette.cached(self._request)
+ return VCRHTTPResponse(response)
+
+ # Otherwise, we made an actual request, and should return the response
+ # we got from the actual connection
+ response = HTTPConnection.getresponse(self)
+ response = {
+ 'status': {'code': response.status, 'message': response.reason},
+ 'headers': dict(response.getheaders()),
+ 'body': {'string': response.read()},
+ }
+ self.cassette.append(self._request, response)
+ return VCRHTTPResponse(response)
+
+
+class VCRHTTPConnection(VCRConnectionMixin, HTTPConnection):
+ '''A Mocked class for HTTP requests'''
+ # Can't use super since this is an old-style class
+ _baseclass = HTTPConnection
+
+ def __init__(self, *args, **kwargs):
+ HTTPConnection.__init__(self, *args, **kwargs)
+
+
+class VCRHTTPSConnection(VCRConnectionMixin, HTTPSConnection):
+ '''A Mocked class for HTTPS requests'''
+ _baseclass = HTTPSConnection
+
+ def __init__(self, *args, **kwargs):
+ '''I overrode the init and copied a lot of the code from the parent
+ class because HTTPConnection when this happens has been replaced by
+ VCRHTTPConnection, but doing it here lets us use the original one.'''
+ HTTPConnection.__init__(self, *args, **kwargs)
+ self.key_file = kwargs.pop('key_file', None)
+ self.cert_file = kwargs.pop('cert_file', None)
diff --git a/vcr/requests_stubs.py b/vcr/stubs/requests_stubs.py
similarity index 75%
rename from vcr/requests_stubs.py
rename to vcr/stubs/requests_stubs.py
index 1dba319..f026fdc 100644
--- a/vcr/requests_stubs.py
+++ b/vcr/stubs/requests_stubs.py
@@ -1,7 +1,8 @@
+'''Stubs for requests'''
+
from requests.packages.urllib3.connectionpool import VerifiedHTTPSConnection
-from .stubs import VCRHTTPSConnection
+from ..stubs import VCRHTTPSConnection
+
class VCRVerifiedHTTPSConnection(VCRHTTPSConnection, VerifiedHTTPSConnection):
-
_baseclass = VerifiedHTTPSConnection
-
diff --git a/vcr/urllib3_stubs.py b/vcr/stubs/urllib3_stubs.py
similarity index 74%
rename from vcr/urllib3_stubs.py
rename to vcr/stubs/urllib3_stubs.py
index 0a8f5a8..d91c8f6 100644
--- a/vcr/urllib3_stubs.py
+++ b/vcr/stubs/urllib3_stubs.py
@@ -1,7 +1,8 @@
+'''Stubs for urllib3'''
+
from urllib3.connectionpool import VerifiedHTTPSConnection
-from .stubs import VCRHTTPSConnection
+from ..stubs import VCRHTTPSConnection
+
class VCRVerifiedHTTPSConnection(VCRHTTPSConnection, VerifiedHTTPSConnection):
-
_baseclass = VerifiedHTTPSConnection
-