Mirroring/vcrpy
1
0
Fork 0
mirror of https://github.com/kevin1024/vcrpy.git synced 2025-12-09 01:03:24 +00:00
Code Releases Activity

replace cached() with mark_played / play_count

Browse Source
This commit is contained in:
Kevin McCarthy
2013-08-05 22:25:45 -10:00
parent 9b665fef27
commit fb2aa89445
6 changed files with 42 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
tests/integration/test_basic.py
View File

@@ -31,4 +31,4 @@ class TestCassette(TestVCR):
# Make the same requests, and assert that we haven't served any more # Make the same requests, and assert that we haven't served any more
# requests out of cache # requests out of cache
urllib2.urlopen('http://httpbin.org/').read() urllib2.urlopen('http://httpbin.org/').read()
self.assertEqual(len(cass.cached()), 0) self.assertEqual(cass.play_count, 0)

32
tests/integration/test_requests.py
View File

@@ -26,13 +26,13 @@ class TestHTTPRequests(TestRequestsBase):
with vcr.use_cassette(self.fixture('atts.yaml')) as cass: with vcr.use_cassette(self.fixture('atts.yaml')) as cass:
# Ensure that this is empty to begin with # Ensure that this is empty to begin with
self.assertEqual(len(cass), 0) self.assertEqual(len(cass), 0)
self.assertEqual(len(cass.cached()), 0) self.assertEqual(cass.play_count, 0)
self.assertEqual( self.assertEqual(
requests.get(url).status_code, requests.get(url).status_code,
requests.get(url).status_code) requests.get(url).status_code)
# Ensure that we've now cached a single response # Ensure that we've now cached a single response
self.assertEqual(len(cass), 1) self.assertEqual(len(cass), 1)
self.assertEqual(len(cass.cached()), 1) self.assertEqual(cass.play_count, 1)
def test_headers(self): def test_headers(self):
'''Ensure that we can read the headers back''' '''Ensure that we can read the headers back'''
@@ -40,13 +40,13 @@ class TestHTTPRequests(TestRequestsBase):
with vcr.use_cassette(self.fixture('headers.yaml')) as cass: with vcr.use_cassette(self.fixture('headers.yaml')) as cass:
# Ensure that this is empty to begin with # Ensure that this is empty to begin with
self.assertEqual(len(cass), 0) self.assertEqual(len(cass), 0)
self.assertEqual(len(cass.cached()), 0) self.assertEqual(cass.play_count, 0)
self.assertEqual( self.assertEqual(
requests.get(url).headers, requests.get(url).headers,
requests.get(url).headers) requests.get(url).headers)
# Ensure that we've now cached a single response # Ensure that we've now cached a single response
self.assertEqual(len(cass), 1) self.assertEqual(len(cass), 1)
self.assertEqual(len(cass.cached()), 1) self.assertEqual(cass.play_count, 1)
def test_body(self): def test_body(self):
'''Ensure the responses are all identical enough''' '''Ensure the responses are all identical enough'''
@@ -54,13 +54,13 @@ class TestHTTPRequests(TestRequestsBase):
with vcr.use_cassette(self.fixture('body.yaml')) as cass: with vcr.use_cassette(self.fixture('body.yaml')) as cass:
# Ensure that this is empty to begin with # Ensure that this is empty to begin with
self.assertEqual(len(cass), 0) self.assertEqual(len(cass), 0)
self.assertEqual(len(cass.cached()), 0) self.assertEqual(cass.play_count, 0)
self.assertEqual( self.assertEqual(
requests.get(url).content, requests.get(url).content,
requests.get(url).content) requests.get(url).content)
# Ensure that we've now cached a single response # Ensure that we've now cached a single response
self.assertEqual(len(cass), 1) self.assertEqual(len(cass), 1)
self.assertEqual(len(cass.cached()), 1) self.assertEqual(cass.play_count, 1)
def test_auth(self): def test_auth(self):
'''Ensure that we can handle basic auth''' '''Ensure that we can handle basic auth'''
@@ -69,14 +69,14 @@ class TestHTTPRequests(TestRequestsBase):
with vcr.use_cassette(self.fixture('auth.yaml')) as cass: with vcr.use_cassette(self.fixture('auth.yaml')) as cass:
# Ensure that this is empty to begin with # Ensure that this is empty to begin with
self.assertEqual(len(cass), 0) self.assertEqual(len(cass), 0)
self.assertEqual(len(cass.cached()), 0) self.assertEqual(cass.play_count, 0)
one = requests.get(url, auth=auth) one = requests.get(url, auth=auth)
two = requests.get(url, auth=auth) two = requests.get(url, auth=auth)
self.assertEqual(one.content, two.content) self.assertEqual(one.content, two.content)
self.assertEqual(one.status_code, two.status_code) self.assertEqual(one.status_code, two.status_code)
# Ensure that we've now cached a single response # Ensure that we've now cached a single response
self.assertEqual(len(cass), 1) self.assertEqual(len(cass), 1)
self.assertEqual(len(cass.cached()), 1) self.assertEqual(cass.play_count, 1)
def test_auth_failed(self): def test_auth_failed(self):
'''Ensure that we can save failed auth statuses''' '''Ensure that we can save failed auth statuses'''
@@ -85,7 +85,7 @@ class TestHTTPRequests(TestRequestsBase):
with vcr.use_cassette(self.fixture('auth-failed.yaml')) as cass: with vcr.use_cassette(self.fixture('auth-failed.yaml')) as cass:
# Ensure that this is empty to begin with # Ensure that this is empty to begin with
self.assertEqual(len(cass), 0) self.assertEqual(len(cass), 0)
self.assertEqual(len(cass.cached()), 0) self.assertEqual(cass.play_count, 0)
one = requests.get(url, auth=auth) one = requests.get(url, auth=auth)
two = requests.get(url, auth=auth) two = requests.get(url, auth=auth)
self.assertEqual(one.content, two.content) self.assertEqual(one.content, two.content)
@@ -93,7 +93,7 @@ class TestHTTPRequests(TestRequestsBase):
self.assertNotEqual(one.status_code, 200) self.assertNotEqual(one.status_code, 200)
# Ensure that we've now cached a single response # Ensure that we've now cached a single response
self.assertEqual(len(cass), 1) self.assertEqual(len(cass), 1)
self.assertEqual(len(cass.cached()), 1) self.assertEqual(cass.play_count, 1)
def test_post(self): def test_post(self):
'''Ensure that we can post and cache the results''' '''Ensure that we can post and cache the results'''
@@ -102,13 +102,13 @@ class TestHTTPRequests(TestRequestsBase):
with vcr.use_cassette(self.fixture('redirect.yaml')) as cass: with vcr.use_cassette(self.fixture('redirect.yaml')) as cass:
# Ensure that this is empty to begin with # Ensure that this is empty to begin with
self.assertEqual(len(cass), 0) self.assertEqual(len(cass), 0)
self.assertEqual(len(cass.cached()), 0) self.assertEqual(cass.play_count, 0)
self.assertEqual( self.assertEqual(
requests.post(url, data).content, requests.post(url, data).content,
requests.post(url, data).content) requests.post(url, data).content)
# Ensure that we've now cached a single response # Ensure that we've now cached a single response
self.assertEqual(len(cass), 1) self.assertEqual(len(cass), 1)
self.assertEqual(len(cass.cached()), 1) self.assertEqual(cass.play_count, 1)
def test_redirects(self): def test_redirects(self):
'''Ensure that we can handle redirects''' '''Ensure that we can handle redirects'''
@@ -116,14 +116,14 @@ class TestHTTPRequests(TestRequestsBase):
with vcr.use_cassette(self.fixture('redirect.yaml')) as cass: with vcr.use_cassette(self.fixture('redirect.yaml')) as cass:
# Ensure that this is empty to begin with # Ensure that this is empty to begin with
self.assertEqual(len(cass), 0) self.assertEqual(len(cass), 0)
self.assertEqual(len(cass.cached()), 0) self.assertEqual(cass.play_count, 0)
self.assertEqual( self.assertEqual(
requests.get(url).content, requests.get(url).content,
requests.get(url).content) requests.get(url).content)
# Ensure that we've now cached /two/ responses. One for the redirect # Ensure that we've now cached /two/ responses. One for the redirect
# and one for the final fetch # and one for the final fetch
self.assertEqual(len(cass), 2) self.assertEqual(len(cass), 2)
self.assertEqual(len(cass.cached()), 2) self.assertEqual(cass.play_count, 2)
class TestHTTPSRequests(TestHTTPRequests): class TestHTTPSRequests(TestHTTPRequests):
@@ -139,7 +139,7 @@ class TestHTTPSRequests(TestHTTPRequests):
requests.get('https://httpbin.org/') requests.get('https://httpbin.org/')
requests.get('http://httpbin.org/') requests.get('http://httpbin.org/')
self.assertEqual(len(cass), 2) self.assertEqual(len(cass), 2)
self.assertEqual(len(cass.cached()), 0) self.assertEqual(cass.play_count, 0)
class TestWild(TestRequestsBase): class TestWild(TestRequestsBase):
@@ -160,4 +160,4 @@ class TestWild(TestRequestsBase):
requests.get(url, headers={'User-Agent': 'vcrpy-test'}) requests.get(url, headers={'User-Agent': 'vcrpy-test'})
# Ensure that we've now served two responses. One for the original # Ensure that we've now served two responses. One for the original
# redirect, and a second for the actual fetch # redirect, and a second for the actual fetch
self.assertEqual(len(cass.cached()), 2) self.assertEqual(cass.play_count, 2)

28
tests/integration/test_urllib2.py
View File

@@ -26,13 +26,13 @@ class TestUrllib2Http(TestUrllib2):
with vcr.use_cassette(self.fixture('atts.yaml')) as cass: with vcr.use_cassette(self.fixture('atts.yaml')) as cass:
# Ensure that this is empty to begin with # Ensure that this is empty to begin with
self.assertEqual(len(cass), 0) self.assertEqual(len(cass), 0)
self.assertEqual(len(cass.cached()), 0) self.assertEqual(cass.play_count, 0)
self.assertEqual( self.assertEqual(
urllib2.urlopen(url).getcode(), urllib2.urlopen(url).getcode(),
urllib2.urlopen(url).getcode()) urllib2.urlopen(url).getcode())
# Ensure that we've now cached a single response # Ensure that we've now cached a single response
self.assertEqual(len(cass), 1) self.assertEqual(len(cass), 1)
self.assertEqual(len(cass.cached()), 1) self.assertEqual(cass.play_count, 1)
def test_random_body(self): def test_random_body(self):
'''Ensure we can read the content, and that it's served from cache''' '''Ensure we can read the content, and that it's served from cache'''
@@ -40,13 +40,13 @@ class TestUrllib2Http(TestUrllib2):
with vcr.use_cassette(self.fixture('body.yaml')) as cass: with vcr.use_cassette(self.fixture('body.yaml')) as cass:
# Ensure that this is empty to begin with # Ensure that this is empty to begin with
self.assertEqual(len(cass), 0) self.assertEqual(len(cass), 0)
self.assertEqual(len(cass.cached()), 0) self.assertEqual(cass.play_count, 0)
self.assertEqual( self.assertEqual(
urllib2.urlopen(url).read(), urllib2.urlopen(url).read(),
urllib2.urlopen(url).read()) urllib2.urlopen(url).read())
# Ensure that we've now cached a single response # Ensure that we've now cached a single response
self.assertEqual(len(cass), 1) self.assertEqual(len(cass), 1)
self.assertEqual(len(cass.cached()), 1) self.assertEqual(cass.play_count, 1)
def test_response_headers(self): def test_response_headers(self):
'''Ensure we can get information from the response''' '''Ensure we can get information from the response'''
@@ -54,7 +54,7 @@ class TestUrllib2Http(TestUrllib2):
with vcr.use_cassette(self.fixture('headers.yaml')) as cass: with vcr.use_cassette(self.fixture('headers.yaml')) as cass:
# Ensure that this is empty to begin with # Ensure that this is empty to begin with
self.assertEqual(len(cass), 0) self.assertEqual(len(cass), 0)
self.assertEqual(len(cass.cached()), 0) self.assertEqual(cass.play_count, 0)
self.assertEqual( self.assertEqual(
urllib2.urlopen(url).info().items(), urllib2.urlopen(url).info().items(),
urllib2.urlopen(url).info().items()) urllib2.urlopen(url).info().items())
@@ -72,12 +72,12 @@ class TestUrllib2Http(TestUrllib2):
for index in range(len(urls)): for index in range(len(urls)):
url = urls[index] url = urls[index]
self.assertEqual(len(cass), index) self.assertEqual(len(cass), index)
self.assertEqual(len(cass.cached()), index) self.assertEqual(cass.play_count, index)
self.assertEqual( self.assertEqual(
urllib2.urlopen(url).read(), urllib2.urlopen(url).read(),
urllib2.urlopen(url).read()) urllib2.urlopen(url).read())
self.assertEqual(len(cass), index + 1) self.assertEqual(len(cass), index + 1)
self.assertEqual(len(cass.cached()), index + 1) self.assertEqual(cass.play_count, index + 1)
def test_get_data(self): def test_get_data(self):
'''Ensure that it works with query data''' '''Ensure that it works with query data'''
@@ -86,13 +86,13 @@ class TestUrllib2Http(TestUrllib2):
with vcr.use_cassette(self.fixture('get_data.yaml')) as cass: with vcr.use_cassette(self.fixture('get_data.yaml')) as cass:
# Ensure that this is empty to begin with # Ensure that this is empty to begin with
self.assertEqual(len(cass), 0) self.assertEqual(len(cass), 0)
self.assertEqual(len(cass.cached()), 0) self.assertEqual(cass.play_count, 0)
self.assertEqual( self.assertEqual(
urllib2.urlopen(url).read(), urllib2.urlopen(url).read(),
urllib2.urlopen(url).read()) urllib2.urlopen(url).read())
# Ensure that we've now cached a single response # Ensure that we've now cached a single response
self.assertEqual(len(cass), 1) self.assertEqual(len(cass), 1)
self.assertEqual(len(cass.cached()), 1) self.assertEqual(cass.play_count, 1)
def test_post_data(self): def test_post_data(self):
'''Ensure that it works when posting data''' '''Ensure that it works when posting data'''
@@ -101,13 +101,13 @@ class TestUrllib2Http(TestUrllib2):
with vcr.use_cassette(self.fixture('post_data.yaml')) as cass: with vcr.use_cassette(self.fixture('post_data.yaml')) as cass:
# Ensure that this is empty to begin with # Ensure that this is empty to begin with
self.assertEqual(len(cass), 0) self.assertEqual(len(cass), 0)
self.assertEqual(len(cass.cached()), 0) self.assertEqual(cass.play_count, 0)
self.assertEqual( self.assertEqual(
urllib2.urlopen(url, data).read(), urllib2.urlopen(url, data).read(),
urllib2.urlopen(url, data).read()) urllib2.urlopen(url, data).read())
# Ensure that we've now cached a single response # Ensure that we've now cached a single response
self.assertEqual(len(cass), 1) self.assertEqual(len(cass), 1)
self.assertEqual(len(cass.cached()), 1) self.assertEqual(cass.play_count, 1)
def test_post_unicode_data(self): def test_post_unicode_data(self):
'''Ensure that it works when posting unicode data''' '''Ensure that it works when posting unicode data'''
@@ -116,13 +116,13 @@ class TestUrllib2Http(TestUrllib2):
with vcr.use_cassette(self.fixture('post_data.yaml')) as cass: with vcr.use_cassette(self.fixture('post_data.yaml')) as cass:
# Ensure that this is empty to begin with # Ensure that this is empty to begin with
self.assertEqual(len(cass), 0) self.assertEqual(len(cass), 0)
self.assertEqual(len(cass.cached()), 0) self.assertEqual(cass.play_count, 0)
self.assertEqual( self.assertEqual(
urllib2.urlopen(url, data).read(), urllib2.urlopen(url, data).read(),
urllib2.urlopen(url, data).read()) urllib2.urlopen(url, data).read())
# Ensure that we've now cached a single response # Ensure that we've now cached a single response
self.assertEqual(len(cass), 1) self.assertEqual(len(cass), 1)
self.assertEqual(len(cass.cached()), 1) self.assertEqual(cass.play_count, 1)
class TestUrllib2Https(TestUrllib2Http): class TestUrllib2Https(TestUrllib2Http):
@@ -138,4 +138,4 @@ class TestUrllib2Https(TestUrllib2Http):
urllib2.urlopen('https://httpbin.org/') urllib2.urlopen('https://httpbin.org/')
urllib2.urlopen('http://httpbin.org/') urllib2.urlopen('http://httpbin.org/')
self.assertEqual(len(cass), 2) self.assertEqual(len(cass), 2)
self.assertEqual(len(cass.cached()), 0) self.assertEqual(cass.play_count, 0)

10
tests/unit/test_cassettes.py
View File

@@ -23,14 +23,14 @@ def test_cassette_deserialize():
assert a._requests == ['foo'] assert a._requests == ['foo']
assert a._responses == ['bar'] assert a._responses == ['bar']
def test_cassette_not_cached(): def test_cassette_not_played():
a = Cassette('test') a = Cassette('test')
assert not a.cached() assert not a.play_count
def test_cassette_cached(): def test_cassette_played():
a = Cassette('test') a = Cassette('test')
a.cached('yup') a.mark_played()
assert a.cached() assert a.play_count == 1
def test_cassette_append(): def test_cassette_append():
a = Cassette('test') a = Cassette('test')

15
vcr/cassette.py
View File

@@ -23,9 +23,9 @@ class Cassette(object):
def __init__(self, path, data=None): def __init__(self, path, data=None):
self._path = path self._path = path
self._cached = []
self._requests = [] self._requests = []
self._responses = [] self._responses = []
self.play_count = 0
if data: if data:
self.deserialize(data) self.deserialize(data)
@@ -45,18 +45,11 @@ class Cassette(object):
self._requests, self._responses = ( self._requests, self._responses = (
[r['request'] for r in source], [r['response'] for r in source]) [r['request'] for r in source], [r['response'] for r in source])
def cached(self, request=None): def mark_played(self, request=None):
''' '''
Alert the cassette of a request that's been cached, or get the Alert the cassette of a request that's been played
requests that we've cached. This is used mainly for
debugging purposes.
''' '''
# TODO: maybe dependency injection for this method since self.play_count += 1
# it's only used in tests?
if request:
self._cached.append(request)
else:
return self._cached
def append(self, request, response): def append(self, request, response):
'''Add a pair of request, response pairs to this cassette''' '''Add a pair of request, response pairs to this cassette'''

4
vcr/stubs/__init__.py
View File

@@ -67,9 +67,9 @@ class VCRConnectionMixin:
# then return it # then return it
response = self.cassette.response(self._request) response = self.cassette.response(self._request)
if response: if response:
# Alert the cassette to the fact that we've served another cached # Alert the cassette to the fact that we've served another
# response for the provided requests # response for the provided requests
self.cassette.cached(self._request) self.cassette.mark_played()
return VCRHTTPResponse(response) return VCRHTTPResponse(response)
# Otherwise, we made an actual request, and should return the response # Otherwise, we made an actual request, and should return the response