1
0
mirror of https://github.com/kevin1024/vcrpy.git synced 2025-12-09 09:13:23 +00:00

Fix pyflakes and pep8 errors

Use extra asserts to use previously unused variables in tests,
such as `cass` and `response`.

Fix only pyflakes errors in docs/conf.py
This commit is contained in:
John Vandenberg
2015-11-25 08:44:33 +11:00
parent 6ae1b00207
commit dc9cd4229b
27 changed files with 145 additions and 97 deletions

View File

@@ -31,26 +31,26 @@ def test_boto_without_vcr():
def test_boto_medium_difficulty(tmpdir):
s3_conn = S3Connection()
s3_bucket = s3_conn.get_bucket('boto-demo-1394171994') # a bucket you can access
with vcr.use_cassette(str(tmpdir.join('boto-medium.yml'))) as cass:
with vcr.use_cassette(str(tmpdir.join('boto-medium.yml'))):
k = Key(s3_bucket)
k.key = 'test.txt'
k.set_contents_from_string('hello world i am a string')
with vcr.use_cassette(str(tmpdir.join('boto-medium.yml'))) as cass:
with vcr.use_cassette(str(tmpdir.join('boto-medium.yml'))):
k = Key(s3_bucket)
k.key = 'test.txt'
k.set_contents_from_string('hello world i am a string')
def test_boto_hardcore_mode(tmpdir):
with vcr.use_cassette(str(tmpdir.join('boto-hardcore.yml'))) as cass:
with vcr.use_cassette(str(tmpdir.join('boto-hardcore.yml'))):
s3_conn = S3Connection()
s3_bucket = s3_conn.get_bucket('boto-demo-1394171994') # a bucket you can access
k = Key(s3_bucket)
k.key = 'test.txt'
k.set_contents_from_string('hello world i am a string')
with vcr.use_cassette(str(tmpdir.join('boto-hardcore.yml'))) as cass:
with vcr.use_cassette(str(tmpdir.join('boto-hardcore.yml'))):
s3_conn = S3Connection()
s3_bucket = s3_conn.get_bucket('boto-demo-1394171994') # a bucket you can access
k = Key(s3_bucket)
@@ -68,10 +68,10 @@ def test_boto_iam(tmpdir):
# Ensure that boto uses CertValidatingHTTPSConnection
boto.config.set('Boto', 'https_validate_certificates', 'true')
with vcr.use_cassette(str(tmpdir.join('boto-iam.yml'))) as cass:
with vcr.use_cassette(str(tmpdir.join('boto-iam.yml'))):
iam_conn = boto.iam.connect_to_region('universal')
iam_conn.get_all_users()
with vcr.use_cassette(str(tmpdir.join('boto-iam.yml'))) as cass:
with vcr.use_cassette(str(tmpdir.join('boto-iam.yml'))):
iam_conn = boto.iam.connect_to_region('universal')
iam_conn.get_all_users()

View File

@@ -78,9 +78,11 @@ def test_filter_json_post_data(tmpdir):
def test_filter_callback(tmpdir):
url = 'http://httpbin.org/get'
cass_file = str(tmpdir.join('basic_auth_filter.yaml'))
def before_record_cb(request):
if request.path != '/get':
return request
# Test the legacy keyword.
my_vcr = vcr.VCR(before_record=before_record_cb)
with my_vcr.use_cassette(cass_file, filter_headers=['authorization']) as cass:

View File

@@ -24,11 +24,11 @@ def scheme(request):
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:
with vcr.use_cassette(str(tmpdir.join('atts.yaml'))):
resp, _ = httplib2.Http().request(url)
code = resp.status
with vcr.use_cassette(str(tmpdir.join('atts.yaml'))) as cass:
with vcr.use_cassette(str(tmpdir.join('atts.yaml'))):
resp, _ = httplib2.Http().request(url)
assert code == resp.status
@@ -36,11 +36,11 @@ def test_response_code(scheme, tmpdir):
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:
with vcr.use_cassette(str(tmpdir.join('body.yaml'))):
_, content = httplib2.Http().request(url)
body = content
with vcr.use_cassette(str(tmpdir.join('body.yaml'))) as cass:
with vcr.use_cassette(str(tmpdir.join('body.yaml'))):
_, content = httplib2.Http().request(url)
assert body == content
@@ -48,26 +48,28 @@ def test_random_body(scheme, tmpdir):
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:
with vcr.use_cassette(str(tmpdir.join('headers.yaml'))):
resp, _ = httplib2.Http().request(url)
headers = resp.items()
with vcr.use_cassette(str(tmpdir.join('headers.yaml'))) as cass:
with vcr.use_cassette(str(tmpdir.join('headers.yaml'))):
resp, _ = httplib2.Http().request(url)
assert set(headers) == set(resp.items())
def test_effective_url(scheme, tmpdir):
'''Ensure that the effective_url is captured'''
url = scheme + '://httpbin.org/redirect-to?url=/html'
with vcr.use_cassette(str(tmpdir.join('headers.yaml'))) as cass:
with vcr.use_cassette(str(tmpdir.join('headers.yaml'))):
resp, _ = httplib2.Http().request(url)
effective_url = resp['content-location']
assert effective_url == scheme + '://httpbin.org/html'
with vcr.use_cassette(str(tmpdir.join('headers.yaml'))) as cass:
with vcr.use_cassette(str(tmpdir.join('headers.yaml'))):
resp, _ = httplib2.Http().request(url)
assert effective_url == resp['content-location']
def test_multiple_requests(scheme, tmpdir):
'''Ensure that we can cache multiple requests'''
urls = [
@@ -85,10 +87,10 @@ 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:
with vcr.use_cassette(str(tmpdir.join('get_data.yaml'))):
_, res1 = httplib2.Http().request(url)
with vcr.use_cassette(str(tmpdir.join('get_data.yaml'))) as cass:
with vcr.use_cassette(str(tmpdir.join('get_data.yaml'))):
_, res2 = httplib2.Http().request(url)
assert res1 == res2
@@ -98,7 +100,7 @@ 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:
with vcr.use_cassette(str(tmpdir.join('post_data.yaml'))):
_, res1 = httplib2.Http().request(url, "POST", data)
with vcr.use_cassette(str(tmpdir.join('post_data.yaml'))) as cass:
@@ -112,7 +114,7 @@ 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:
with vcr.use_cassette(str(tmpdir.join('post_data.yaml'))):
_, res1 = httplib2.Http().request(url, "POST", data)
with vcr.use_cassette(str(tmpdir.join('post_data.yaml'))) as cass:

View File

@@ -1,7 +1,5 @@
import base64
import pytest
from six.moves.urllib.request import urlopen, Request
from six.moves.urllib.error import HTTPError
from six.moves.urllib.request import urlopen
import vcr
@@ -40,6 +38,7 @@ def test_ignore_localhost_and_httpbin(tmpdir, httpserver):
urlopen(httpserver.url)
assert len(cass) == 0
def test_ignore_localhost_twice(tmpdir, httpserver):
httpserver.serve_content('Hello!')
cass_file = str(tmpdir.join('filter_qs.yaml'))

View File

@@ -37,7 +37,7 @@ def cassette(tmpdir):
('query',
'https://google.com/get?p2=q2&p1=q1',
'http://httpbin.org/get?p1=q1&a=b')
])
])
def test_matchers(cassette, matcher, matching_uri, not_matching_uri):
# play cassette with default uri
with vcr.use_cassette(cassette, match_on=[matcher]) as cass:

View File

@@ -1,4 +1,3 @@
import os
import pytest
import vcr
from six.moves.urllib.request import urlopen
@@ -8,39 +7,39 @@ def test_once_record_mode(tmpdir):
testfile = str(tmpdir.join('recordmode.yml'))
with vcr.use_cassette(testfile, record_mode="once"):
# cassette file doesn't exist, so create.
response = urlopen('http://httpbin.org/').read()
urlopen('http://httpbin.org/').read()
with vcr.use_cassette(testfile, record_mode="once") as cass:
with vcr.use_cassette(testfile, record_mode="once"):
# make the same request again
response = urlopen('http://httpbin.org/').read()
urlopen('http://httpbin.org/').read()
# the first time, it's played from the cassette.
# but, try to access something else from the same cassette, and an
# exception is raised.
with pytest.raises(Exception):
response = urlopen('http://httpbin.org/get').read()
urlopen('http://httpbin.org/get').read()
def test_once_record_mode_two_times(tmpdir):
testfile = str(tmpdir.join('recordmode.yml'))
with vcr.use_cassette(testfile, record_mode="once"):
# get two of the same file
response1 = urlopen('http://httpbin.org/').read()
response2 = urlopen('http://httpbin.org/').read()
urlopen('http://httpbin.org/').read()
urlopen('http://httpbin.org/').read()
with vcr.use_cassette(testfile, record_mode="once") as cass:
with vcr.use_cassette(testfile, record_mode="once"):
# do it again
response = urlopen('http://httpbin.org/').read()
response = urlopen('http://httpbin.org/').read()
urlopen('http://httpbin.org/').read()
urlopen('http://httpbin.org/').read()
def test_once_mode_three_times(tmpdir):
testfile = str(tmpdir.join('recordmode.yml'))
with vcr.use_cassette(testfile, record_mode="once"):
# get three of the same file
response1 = urlopen('http://httpbin.org/').read()
response2 = urlopen('http://httpbin.org/').read()
response2 = urlopen('http://httpbin.org/').read()
urlopen('http://httpbin.org/').read()
urlopen('http://httpbin.org/').read()
urlopen('http://httpbin.org/').read()
def test_new_episodes_record_mode(tmpdir):
@@ -48,18 +47,18 @@ def test_new_episodes_record_mode(tmpdir):
with vcr.use_cassette(testfile, record_mode="new_episodes"):
# cassette file doesn't exist, so create.
response = urlopen('http://httpbin.org/').read()
urlopen('http://httpbin.org/').read()
with vcr.use_cassette(testfile, record_mode="new_episodes") as cass:
# make the same request again
response = urlopen('http://httpbin.org/').read()
urlopen('http://httpbin.org/').read()
# all responses have been played
assert cass.all_played
# in the "new_episodes" record mode, we can add more requests to
# a cassette without repurcussions.
response = urlopen('http://httpbin.org/get').read()
urlopen('http://httpbin.org/get').read()
# one of the responses has been played
assert cass.play_count == 1
@@ -102,15 +101,15 @@ def test_all_record_mode(tmpdir):
with vcr.use_cassette(testfile, record_mode="all"):
# cassette file doesn't exist, so create.
response = urlopen('http://httpbin.org/').read()
urlopen('http://httpbin.org/').read()
with vcr.use_cassette(testfile, record_mode="all") as cass:
# make the same request again
response = urlopen('http://httpbin.org/').read()
urlopen('http://httpbin.org/').read()
# in the "all" record mode, we can add more requests to
# a cassette without repurcussions.
response = urlopen('http://httpbin.org/get').read()
urlopen('http://httpbin.org/get').read()
# The cassette was never actually played, even though it existed.
# that's because, in "all" mode, the requests all go directly to
@@ -124,7 +123,7 @@ def test_none_record_mode(tmpdir):
testfile = str(tmpdir.join('recordmode.yml'))
with vcr.use_cassette(testfile, record_mode="none"):
with pytest.raises(Exception):
response = urlopen('http://httpbin.org/').read()
urlopen('http://httpbin.org/').read()
def test_none_record_mode_with_existing_cassette(tmpdir):
@@ -132,12 +131,12 @@ def test_none_record_mode_with_existing_cassette(tmpdir):
testfile = str(tmpdir.join('recordmode.yml'))
with vcr.use_cassette(testfile, record_mode="all"):
response = urlopen('http://httpbin.org/').read()
urlopen('http://httpbin.org/').read()
# play from cassette file
with vcr.use_cassette(testfile, record_mode="none") as cass:
response = urlopen('http://httpbin.org/').read()
urlopen('http://httpbin.org/').read()
assert cass.play_count == 1
# but if I try to hit the net, raise an exception.
with pytest.raises(Exception):
response = urlopen('http://httpbin.org/get').read()
urlopen('http://httpbin.org/get').read()

View File

@@ -14,12 +14,12 @@ def test_registered_true_matcher(tmpdir):
my_vcr = vcr.VCR()
my_vcr.register_matcher('true', true_matcher)
testfile = str(tmpdir.join('test.yml'))
with my_vcr.use_cassette(testfile, match_on=['true']) as cass:
with my_vcr.use_cassette(testfile, match_on=['true']):
# These 2 different urls are stored as the same request
urlopen('http://httpbin.org/')
urlopen('https://httpbin.org/get')
with my_vcr.use_cassette(testfile, match_on=['true']) as cass:
with my_vcr.use_cassette(testfile, match_on=['true']):
# I can get the response twice even though I only asked for it once
urlopen('http://httpbin.org/get')
urlopen('https://httpbin.org/get')

View File

@@ -10,7 +10,7 @@ class MockSerializer(object):
def deserialize(self, cassette_string):
self.serialize_count += 1
self.cassette_string = cassette_string
return {'interactions':[]}
return {'interactions': []}
def serialize(self, cassette_dict):
self.deserialize_count += 1

View File

@@ -44,6 +44,7 @@ def test_body(tmpdir, scheme):
with vcr.use_cassette(str(tmpdir.join('body.yaml'))):
assert content == requests.get(url).content
def test_effective_url(scheme, tmpdir):
'''Ensure that the effective_url is captured'''
url = scheme + '://httpbin.org/redirect-to?url=/html'
@@ -54,6 +55,7 @@ def test_effective_url(scheme, tmpdir):
with vcr.use_cassette(str(tmpdir.join('url.yaml'))):
assert effective_url == requests.get(url).url
def test_auth(tmpdir, scheme):
'''Ensure that we can handle basic auth'''
auth = ('user', 'passwd')
@@ -242,4 +244,3 @@ def test_filter_post_params(tmpdir, scheme):
requests.post(url, data={'key': 'value'})
with vcr.use_cassette(cass_loc, filter_post_data_parameters=['key']) as cass:
assert b'key=value' not in cass.requests[0].body

View File

@@ -1,6 +1,7 @@
import vcr
import six.moves.http_client as httplib
def _headers_are_case_insensitive():
conn = httplib.HTTPConnection('httpbin.org')
conn.request('GET', "/cookies/set?k1=v1")
@@ -12,6 +13,7 @@ def _headers_are_case_insensitive():
cookie_data2 = r2.getheader('Set-Cookie')
return cookie_data1 == cookie_data2
def test_case_insensitivity(tmpdir):
testfile = str(tmpdir.join('case_insensitivity.yml'))
# check if headers are case insensitive outside of vcrpy
@@ -25,12 +27,14 @@ def test_case_insensitivity(tmpdir):
# behavior should be the same both inside and outside
assert outside == inside == inside2
def _multiple_header_value(httpserver):
conn = httplib.HTTPConnection('%s:%s' % httpserver.server_address)
conn.request('GET', "/")
r = conn.getresponse()
return r.getheader('foo')
def test_multiple_headers(tmpdir, httpserver):
testfile = str(tmpdir.join('multiple_headers.yaml'))
httpserver.serve_content('Hello!', headers=[('foo', 'bar'), ('foo', 'baz')])

View File

@@ -81,6 +81,7 @@ def test_body(get_client, tmpdir, scheme):
assert content == (yield get(get_client(), url)).body
assert 1 == cass.play_count
@pytest.mark.gen_test
def test_effective_url(get_client, scheme, tmpdir):
'''Ensure that the effective_url is captured'''
@@ -93,6 +94,7 @@ def test_effective_url(get_client, scheme, tmpdir):
assert effective_url == (yield get(get_client(), url)).effective_url
assert 1 == cass.play_count
@pytest.mark.gen_test
def test_auth(get_client, tmpdir, scheme):
'''Ensure that we can handle basic auth'''

View File

@@ -22,43 +22,46 @@ def scheme(request):
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:
with vcr.use_cassette(str(tmpdir.join('atts.yaml'))):
code = urlopen(url).getcode()
with vcr.use_cassette(str(tmpdir.join('atts.yaml'))) as cass:
with vcr.use_cassette(str(tmpdir.join('atts.yaml'))):
assert code == urlopen(url).getcode()
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:
with vcr.use_cassette(str(tmpdir.join('body.yaml'))):
body = urlopen(url).read()
with vcr.use_cassette(str(tmpdir.join('body.yaml'))) as cass:
with vcr.use_cassette(str(tmpdir.join('body.yaml'))):
assert body == urlopen(url).read()
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:
with vcr.use_cassette(str(tmpdir.join('headers.yaml'))):
open1 = urlopen(url).info().items()
with vcr.use_cassette(str(tmpdir.join('headers.yaml'))) as cass:
with vcr.use_cassette(str(tmpdir.join('headers.yaml'))):
open2 = urlopen(url).info().items()
assert sorted(open1) == sorted(open2)
def test_effective_url(scheme, tmpdir):
'''Ensure that the effective_url is captured'''
url = scheme + '://httpbin.org/redirect-to?url=/html'
with vcr.use_cassette(str(tmpdir.join('headers.yaml'))) as cass:
with vcr.use_cassette(str(tmpdir.join('headers.yaml'))):
effective_url = urlopen(url).geturl()
assert effective_url == scheme + '://httpbin.org/html'
with vcr.use_cassette(str(tmpdir.join('headers.yaml'))) as cass:
with vcr.use_cassette(str(tmpdir.join('headers.yaml'))):
assert effective_url == urlopen(url).geturl()
def test_multiple_requests(scheme, tmpdir):
'''Ensure that we can cache multiple requests'''
urls = [
@@ -76,10 +79,10 @@ 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:
with vcr.use_cassette(str(tmpdir.join('get_data.yaml'))):
res1 = urlopen(url).read()
with vcr.use_cassette(str(tmpdir.join('get_data.yaml'))) as cass:
with vcr.use_cassette(str(tmpdir.join('get_data.yaml'))):
res2 = urlopen(url).read()
assert res1 == res2
@@ -89,11 +92,12 @@ def test_post_data(scheme, tmpdir):
'''Ensure that it works when posting data'''
data = urlencode({'some': 1, 'data': 'here'}).encode('utf-8')
url = scheme + '://httpbin.org/post'
with vcr.use_cassette(str(tmpdir.join('post_data.yaml'))) as cass:
with vcr.use_cassette(str(tmpdir.join('post_data.yaml'))):
res1 = urlopen(url, data).read()
with vcr.use_cassette(str(tmpdir.join('post_data.yaml'))) as cass:
res2 = urlopen(url, data).read()
assert len(cass) == 1
assert res1 == res2
assert_cassette_has_one_response(cass)
@@ -103,10 +107,13 @@ def test_post_unicode_data(scheme, tmpdir):
'''Ensure that it works when posting unicode data'''
data = urlencode({'snowman': u''.encode('utf-8')}).encode('utf-8')
url = scheme + '://httpbin.org/post'
with vcr.use_cassette(str(tmpdir.join('post_data.yaml'))) as cass:
with vcr.use_cassette(str(tmpdir.join('post_data.yaml'))):
res1 = urlopen(url, data).read()
with vcr.use_cassette(str(tmpdir.join('post_data.yaml'))) as cass:
res2 = urlopen(url, data).read()
assert len(cass) == 1
assert res1 == res2
assert_cassette_has_one_response(cass)
@@ -122,6 +129,7 @@ def test_cross_scheme(tmpdir):
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/'

View File

@@ -20,7 +20,7 @@ def scheme(request):
@pytest.fixture(scope='module')
def verify_pool_mgr():
return urllib3.PoolManager(
cert_reqs='CERT_REQUIRED', # Force certificate check.
cert_reqs='CERT_REQUIRED', # Force certificate check.
ca_certs=certifi.where()
)

View File

@@ -42,6 +42,8 @@ def test_flickr_multipart_upload():
data = r.read()
h.close()
return data
with vcr.use_cassette('fixtures/vcr_cassettes/flickr.yaml') as cass:
_pretend_to_be_flickr_library()
assert len(cass) == 1
@@ -63,7 +65,8 @@ def test_cookies(tmpdir):
testfile = str(tmpdir.join('cookies.yml'))
with vcr.use_cassette(testfile):
s = requests.Session()
r1 = s.get("http://httpbin.org/cookies/set?k1=v1&k2=v2")
s.get("http://httpbin.org/cookies/set?k1=v1&k2=v2")
r2 = s.get("http://httpbin.org/cookies")
assert len(r2.json()['cookies']) == 2
@@ -81,9 +84,8 @@ def test_xmlrpclib(tmpdir):
roundup_server = xmlrpc_client.ServerProxy('http://bugs.python.org/xmlrpc', allow_none=True)
original_schema = roundup_server.schema()
with vcr.use_cassette(str(tmpdir.join('xmlrpcvideo.yaml'))) as cassette:
with vcr.use_cassette(str(tmpdir.join('xmlrpcvideo.yaml'))):
roundup_server = xmlrpc_client.ServerProxy('http://bugs.python.org/xmlrpc', allow_none=True)
second_schema = roundup_server.schema()
assert original_schema == second_schema

View File

@@ -89,7 +89,7 @@ def make_get_request():
def test_function_decorated_with_use_cassette_can_be_invoked_multiple_times(*args):
decorated_function = Cassette.use(path='test')(make_get_request)
for i in range(4):
decorated_function()
decorated_function()
def test_arg_getter_functionality():
@@ -160,7 +160,7 @@ def test_nesting_cassette_context_managers(*args):
with contextlib.ExitStack() as exit_stack:
first_cassette = exit_stack.enter_context(Cassette.use(path='test'))
exit_stack.enter_context(mock.patch.object(first_cassette, 'play_response',
return_value=first_response))
return_value=first_response))
assert_get_response_body_is('first_response')
# Make sure a second cassette can supercede the first
@@ -200,16 +200,17 @@ def test_custom_patchers():
custom_patches=((Test, 'attribute', VCRHTTPSConnection),)):
assert issubclass(Test.attribute, VCRHTTPSConnection)
assert VCRHTTPSConnection is not Test.attribute
assert Test.attribute is not old_attribute
assert Test.attribute is not old_attribute
assert issubclass(Test.attribute, VCRHTTPSConnection)
assert VCRHTTPSConnection is not Test.attribute
assert Test.attribute is old_attribute
assert Test.attribute is old_attribute
def test_decorated_functions_are_reentrant():
info = {"second": False}
original_conn = httplib.HTTPConnection
@Cassette.use(path='whatever', inject=True)
def test_function(cassette):
if info['second']:
@@ -219,6 +220,7 @@ def test_decorated_functions_are_reentrant():
info['second'] = True
test_function()
assert httplib.HTTPConnection is info['first_conn']
test_function()
assert httplib.HTTPConnection is original_conn
@@ -232,9 +234,11 @@ def test_cassette_use_called_without_path_uses_function_to_generate_path():
def test_path_transformer_with_function_path():
path_transformer = lambda path: os.path.join('a', path)
@Cassette.use(inject=True, path_transformer=path_transformer)
def function_name(cassette):
assert cassette._path == os.path.join('a', 'function_name')
function_name()
@@ -255,15 +259,18 @@ def test_path_transformer_None():
def test_func_path_generator():
def generator(function):
return os.path.join(os.path.dirname(inspect.getfile(function)),
function.__name__)
function.__name__)
@Cassette.use(inject=True, func_path_generator=generator)
def function_name(cassette):
assert cassette._path == os.path.join(os.path.dirname(__file__), 'function_name')
function_name()
def test_use_as_decorator_on_coroutine():
original_http_connetion = httplib.HTTPConnection
@Cassette.use(inject=True)
def test_function(cassette):
assert httplib.HTTPConnection.cassette is cassette
@@ -274,6 +281,7 @@ def test_use_as_decorator_on_coroutine():
assert httplib.HTTPConnection is not original_http_connetion
value = yield 2
assert value == 2
coroutine = test_function()
value = next(coroutine)
while True:
@@ -285,6 +293,7 @@ def test_use_as_decorator_on_coroutine():
def test_use_as_decorator_on_generator():
original_http_connetion = httplib.HTTPConnection
@Cassette.use(inject=True)
def test_function(cassette):
assert httplib.HTTPConnection.cassette is cassette
@@ -293,4 +302,5 @@ def test_use_as_decorator_on_generator():
assert httplib.HTTPConnection.cassette is cassette
assert httplib.HTTPConnection is not original_http_connetion
yield 2
assert list(test_function()) == [1, 2]

View File

@@ -29,9 +29,9 @@ def test_add_header_deprecated():
('https://go.com/', 443),
('https://go.com:443/', 443),
('https://go.com:3000/', 3000),
])
])
def test_port(uri, expected_port):
req = Request('GET', uri, '', {})
req = Request('GET', uri, '', {})
assert req.port == expected_port

View File

@@ -1,6 +1,8 @@
from vcr.stubs import VCRHTTPSConnection
class TestVCRConnection(object):
def test_setting_of_attributes_get_propogated_to_real_connection(self):
vcr_connection = VCRHTTPSConnection('www.examplehost.com')
vcr_connection.ssl_version = 'example_ssl_version'