Mirroring/vcrpy
1
0
Fork 0
mirror of https://github.com/kevin1024/vcrpy.git synced 2025-12-08 16:53:23 +00:00
Code Releases Activity

Merge pull request #149 from kevin1024/pytest-httpbin

Browse Source 
WIP: add pytest-httpbin
This commit is contained in:
Kevin McCarthy
2016-05-01 14:19:54 -10:00
parent c571c932c9 4e36997e1a
commit 528c9e7b1a
20 changed files with 380 additions and 369 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
runtests.sh Executable file
View File

@@ -0,0 +1,3 @@
#!/bin/bash
REQUESTS_CA_BUNDLE=`python -m pytest_httpbin.certs` py.test $1

3
setup.py
View File

@@ -64,8 +64,7 @@ setup(
install_requires=install_requires,
extras_require=extras_require,
license='MIT',
tests_require=['pytest', 'mock', 'pytest-localserver'],
cmdclass={'test': PyTest},
tests_require=['pytest', 'mock', 'pytest-localserver', 'pytest-httpbin'],
classifiers=[
'Development Status :: 4 - Beta',
'Environment :: Console',

43
tests/integration/test_basic.py
View File

@@ -9,74 +9,63 @@ from six.moves.urllib.request import urlopen
import vcr
def test_nonexistent_directory(tmpdir):
def test_nonexistent_directory(tmpdir, httpbin):
'''If we load a cassette in a nonexistent directory, it can save ok'''
# Check to make sure directory doesnt exist
assert not os.path.exists(str(tmpdir.join('nonexistent')))
# Run VCR to create dir and cassette file
with vcr.use_cassette(str(tmpdir.join('nonexistent', 'cassette.yml'))):
urlopen('http://httpbin.org/').read()
urlopen(httpbin.url).read()
# This should have made the file and the directory
assert os.path.exists(str(tmpdir.join('nonexistent', 'cassette.yml')))
def test_unpatch(tmpdir):
def test_unpatch(tmpdir, httpbin):
'''Ensure that our cassette gets unpatched when we're done'''
with vcr.use_cassette(str(tmpdir.join('unpatch.yaml'))) as cass:
urlopen('http://httpbin.org/').read()
urlopen(httpbin.url).read()
# Make the same request, and assert that we haven't served any more
# requests out of cache
urlopen('http://httpbin.org/').read()
urlopen(httpbin.url).read()
assert cass.play_count == 0
def test_basic_use(tmpdir):
'''
Copied from the docs
'''
with vcr.use_cassette('fixtures/vcr_cassettes/synopsis.yaml'):
response = urlopen(
'http://www.iana.org/domains/reserved'
).read()
assert b'Example domains' in response
def test_basic_json_use(tmpdir):
def test_basic_json_use(tmpdir, httpbin):
'''
Ensure you can load a json serialized cassette
'''
test_fixture = 'fixtures/vcr_cassettes/synopsis.json'
test_fixture = str(tmpdir.join('synopsis.json'))
with vcr.use_cassette(test_fixture, serializer='json'):
response = urlopen('http://httpbin.org/').read()
response = urlopen(httpbin.url).read()
assert b'difficult sometimes' in response
def test_patched_content(tmpdir):
def test_patched_content(tmpdir, httpbin):
'''
Ensure that what you pull from a cassette is what came from the
request
'''
with vcr.use_cassette(str(tmpdir.join('synopsis.yaml'))) as cass:
response = urlopen('http://httpbin.org/').read()
response = urlopen(httpbin.url).read()
assert cass.play_count == 0
with vcr.use_cassette(str(tmpdir.join('synopsis.yaml'))) as cass:
response2 = urlopen('http://httpbin.org/').read()
response2 = urlopen(httpbin.url).read()
assert cass.play_count == 1
cass._save(force=True)
with vcr.use_cassette(str(tmpdir.join('synopsis.yaml'))) as cass:
response3 = urlopen('http://httpbin.org/').read()
response3 = urlopen(httpbin.url).read()
assert cass.play_count == 1
assert response == response2
assert response2 == response3
def test_patched_content_json(tmpdir):
def test_patched_content_json(tmpdir, httpbin):
'''
Ensure that what you pull from a json cassette is what came from the
request
@@ -85,16 +74,16 @@ def test_patched_content_json(tmpdir):
testfile = str(tmpdir.join('synopsis.json'))
with vcr.use_cassette(testfile) as cass:
response = urlopen('http://httpbin.org/').read()
response = urlopen(httpbin.url).read()
assert cass.play_count == 0
with vcr.use_cassette(testfile) as cass:
response2 = urlopen('http://httpbin.org/').read()
response2 = urlopen(httpbin.url).read()
assert cass.play_count == 1
cass._save(force=True)
with vcr.use_cassette(testfile) as cass:
response3 = urlopen('http://httpbin.org/').read()
response3 = urlopen(httpbin.url).read()
assert cass.play_count == 1
assert response == response2

18
tests/integration/test_config.py
View File

@@ -5,46 +5,46 @@ import vcr
from six.moves.urllib.request import urlopen
def test_set_serializer_default_config(tmpdir):
def test_set_serializer_default_config(tmpdir, httpbin):
my_vcr = vcr.VCR(serializer='json')
with my_vcr.use_cassette(str(tmpdir.join('test.json'))):
assert my_vcr.serializer == 'json'
urlopen('http://httpbin.org/get')
urlopen(httpbin.url + '/get')
with open(str(tmpdir.join('test.json'))) as f:
assert json.loads(f.read())
def test_default_set_cassette_library_dir(tmpdir):
def test_default_set_cassette_library_dir(tmpdir, httpbin):
my_vcr = vcr.VCR(cassette_library_dir=str(tmpdir.join('subdir')))
with my_vcr.use_cassette('test.json'):
urlopen('http://httpbin.org/get')
urlopen(httpbin.url + '/get')
assert os.path.exists(str(tmpdir.join('subdir').join('test.json')))
def test_override_set_cassette_library_dir(tmpdir):
def test_override_set_cassette_library_dir(tmpdir, httpbin):
my_vcr = vcr.VCR(cassette_library_dir=str(tmpdir.join('subdir')))
cld = str(tmpdir.join('subdir2'))
with my_vcr.use_cassette('test.json', cassette_library_dir=cld):
urlopen('http://httpbin.org/get')
urlopen(httpbin.url + '/get')
assert os.path.exists(str(tmpdir.join('subdir2').join('test.json')))
assert not os.path.exists(str(tmpdir.join('subdir').join('test.json')))
def test_override_match_on(tmpdir):
def test_override_match_on(tmpdir, httpbin):
my_vcr = vcr.VCR(match_on=['method'])
with my_vcr.use_cassette(str(tmpdir.join('test.json'))):
urlopen('http://httpbin.org/')
urlopen(httpbin.url)
with my_vcr.use_cassette(str(tmpdir.join('test.json'))) as cass:
urlopen('http://httpbin.org/get')
urlopen(httpbin.url + '/get')
assert len(cass) == 1
assert cass.play_count == 1

14
tests/integration/test_disksaver.py
View File

@@ -10,19 +10,19 @@ from six.moves.urllib.request import urlopen
import vcr
def test_disk_saver_nowrite(tmpdir):
def test_disk_saver_nowrite(tmpdir, httpbin):
'''
Ensure that when you close a cassette without changing it it doesn't
rewrite the file
'''
fname = str(tmpdir.join('synopsis.yaml'))
with vcr.use_cassette(fname) as cass:
urlopen('http://www.iana.org/domains/reserved').read()
urlopen(httpbin.url).read()
assert cass.play_count == 0
last_mod = os.path.getmtime(fname)
with vcr.use_cassette(fname) as cass:
urlopen('http://www.iana.org/domains/reserved').read()
urlopen(httpbin.url).read()
assert cass.play_count == 1
assert cass.dirty is False
last_mod2 = os.path.getmtime(fname)
@@ -30,14 +30,14 @@ def test_disk_saver_nowrite(tmpdir):
assert last_mod == last_mod2
def test_disk_saver_write(tmpdir):
def test_disk_saver_write(tmpdir, httpbin):
'''
Ensure that when you close a cassette after changing it it does
rewrite the file
'''
fname = str(tmpdir.join('synopsis.yaml'))
with vcr.use_cassette(fname) as cass:
urlopen('http://www.iana.org/domains/reserved').read()
urlopen(httpbin.url).read()
assert cass.play_count == 0
last_mod = os.path.getmtime(fname)
@@ -46,8 +46,8 @@ def test_disk_saver_write(tmpdir):
time.sleep(1)
with vcr.use_cassette(fname, record_mode='any') as cass:
urlopen('http://www.iana.org/domains/reserved').read()
urlopen('http://httpbin.org/').read()
urlopen(httpbin.url).read()
urlopen(httpbin.url + '/get').read()
assert cass.play_count == 1
assert cass.dirty
last_mod2 = os.path.getmtime(fname)

32
tests/integration/test_filter.py
View File

@@ -21,8 +21,8 @@ def _find_header(cassette, header):
return any(header in request.headers for request in cassette.requests)
def test_filter_basic_auth(tmpdir):
url = 'http://httpbin.org/basic-auth/user/passwd'
def test_filter_basic_auth(tmpdir, httpbin):
url = httpbin.url + '/basic-auth/user/passwd'
cass_file = str(tmpdir.join('basic_auth_filter.yaml'))
my_vcr = vcr.VCR(match_on=['uri', 'method', 'headers'])
# 2 requests, one with auth failure and one with auth success
@@ -44,8 +44,8 @@ def test_filter_basic_auth(tmpdir):
assert len(cass) == 2
def test_filter_querystring(tmpdir):
url = 'http://httpbin.org/?foo=bar'
def test_filter_querystring(tmpdir, httpbin):
url = httpbin.url + '/?foo=bar'
cass_file = str(tmpdir.join('filter_qs.yaml'))
with vcr.use_cassette(cass_file, filter_query_parameters=['foo']):
urlopen(url)
@@ -54,8 +54,8 @@ def test_filter_querystring(tmpdir):
assert 'foo' not in cass.requests[0].url
def test_filter_post_data(tmpdir):
url = 'http://httpbin.org/post'
def test_filter_post_data(tmpdir, httpbin):
url = httpbin.url + '/post'
data = urlencode({'id': 'secret', 'foo': 'bar'}).encode('utf-8')
cass_file = str(tmpdir.join('filter_pd.yaml'))
with vcr.use_cassette(cass_file, filter_post_data_parameters=['id']):
@@ -64,9 +64,9 @@ def test_filter_post_data(tmpdir):
assert b'id=secret' not in cass.requests[0].body
def test_filter_json_post_data(tmpdir):
def test_filter_json_post_data(tmpdir, httpbin):
data = json.dumps({'id': 'secret', 'foo': 'bar'}).encode('utf-8')
request = Request('http://httpbin.org/post', data=data)
request = Request(httpbin.url + '/post', data=data)
request.add_header('Content-Type', 'application/json')
cass_file = str(tmpdir.join('filter_jpd.yaml'))
@@ -76,8 +76,8 @@ def test_filter_json_post_data(tmpdir):
assert b'"id": "secret"' not in cass.requests[0].body
def test_filter_callback(tmpdir):
url = 'http://httpbin.org/get'
def test_filter_callback(tmpdir, httpbin):
url = httpbin.url + '/get'
cass_file = str(tmpdir.join('basic_auth_filter.yaml'))
def before_record_cb(request):
@@ -96,8 +96,8 @@ def test_filter_callback(tmpdir):
assert len(cass) == 0
def test_decompress_gzip(tmpdir):
url = 'http://httpbin.org/gzip'
def test_decompress_gzip(tmpdir, httpbin):
url = httpbin.url + '/gzip'
request = Request(url, headers={'Accept-Encoding': ['gzip, deflate']})
cass_file = str(tmpdir.join('gzip_response.yaml'))
with vcr.use_cassette(cass_file, decode_compressed_response=True):
@@ -108,8 +108,8 @@ def test_decompress_gzip(tmpdir):
assert_is_json(decoded_response)
def test_decompress_deflate(tmpdir):
url = 'http://httpbin.org/deflate'
def test_decompress_deflate(tmpdir, httpbin):
url = httpbin.url + '/deflate'
request = Request(url, headers={'Accept-Encoding': ['gzip, deflate']})
cass_file = str(tmpdir.join('deflate_response.yaml'))
with vcr.use_cassette(cass_file, decode_compressed_response=True):
@@ -120,9 +120,9 @@ def test_decompress_deflate(tmpdir):
assert_is_json(decoded_response)
def test_decompress_regular(tmpdir):
def test_decompress_regular(tmpdir, httpbin):
"""Test that it doesn't try to decompress content that isn't compressed"""
url = 'http://httpbin.org/get'
url = httpbin.url + '/get'
cass_file = str(tmpdir.join('noncompressed_response.yaml'))
with vcr.use_cassette(cass_file, decode_compressed_response=True):
urlopen(url)

93
tests/integration/test_httplib2.py
View File

@@ -4,6 +4,7 @@
# External imports
from six.moves.urllib_parse import urlencode
import pytest
import pytest_httpbin.certs
# Internal imports
import vcr
@@ -13,141 +14,141 @@ from assertions import assert_cassette_has_one_response
httplib2 = pytest.importorskip("httplib2")
@pytest.fixture(params=["https", "http"])
def scheme(request):
def http():
"""
Fixture that returns both http and https
Returns an httplib2 HTTP instance
with the certificate replaced by the httpbin one.
"""
return request.param
return httplib2.Http(ca_certs=pytest_httpbin.certs.where())
def test_response_code(scheme, tmpdir):
def test_response_code(tmpdir, httpbin_both):
'''Ensure we can read a response code from a fetch'''
url = scheme + '://httpbin.org/'
url = httpbin_both.url
with vcr.use_cassette(str(tmpdir.join('atts.yaml'))):
resp, _ = httplib2.Http().request(url)
resp, _ = http().request(url)
code = resp.status
with vcr.use_cassette(str(tmpdir.join('atts.yaml'))):
resp, _ = httplib2.Http().request(url)
resp, _ = http().request(url)
assert code == resp.status
def test_random_body(scheme, tmpdir):
def test_random_body(httpbin_both, tmpdir):
'''Ensure we can read the content, and that it's served from cache'''
url = scheme + '://httpbin.org/bytes/1024'
url = httpbin_both.url + '/bytes/1024'
with vcr.use_cassette(str(tmpdir.join('body.yaml'))):
_, content = httplib2.Http().request(url)
_, content = http().request(url)
body = content
with vcr.use_cassette(str(tmpdir.join('body.yaml'))):
_, content = httplib2.Http().request(url)
_, content = http().request(url)
assert body == content
def test_response_headers(scheme, tmpdir):
def test_response_headers(tmpdir, httpbin_both):
'''Ensure we can get information from the response'''
url = scheme + '://httpbin.org/'
url = httpbin_both.url
with vcr.use_cassette(str(tmpdir.join('headers.yaml'))):
resp, _ = httplib2.Http().request(url)
resp, _ = http().request(url)
headers = resp.items()
with vcr.use_cassette(str(tmpdir.join('headers.yaml'))):
resp, _ = httplib2.Http().request(url)
resp, _ = http().request(url)
assert set(headers) == set(resp.items())
def test_effective_url(scheme, tmpdir):
def test_effective_url(tmpdir, httpbin_both):
'''Ensure that the effective_url is captured'''
url = scheme + '://httpbin.org/redirect-to?url=/html'
url = httpbin_both.url + '/redirect-to?url=/html'
with vcr.use_cassette(str(tmpdir.join('headers.yaml'))):
resp, _ = httplib2.Http().request(url)
resp, _ = http().request(url)
effective_url = resp['content-location']
assert effective_url == scheme + '://httpbin.org/html'
assert effective_url == httpbin_both + '/html'
with vcr.use_cassette(str(tmpdir.join('headers.yaml'))):
resp, _ = httplib2.Http().request(url)
resp, _ = http().request(url)
assert effective_url == resp['content-location']
def test_multiple_requests(scheme, tmpdir):
def test_multiple_requests(tmpdir, httpbin_both):
'''Ensure that we can cache multiple requests'''
urls = [
scheme + '://httpbin.org/',
scheme + '://httpbin.org/',
scheme + '://httpbin.org/get',
scheme + '://httpbin.org/bytes/1024'
httpbin_both.url,
httpbin_both.url,
httpbin_both.url + '/get',
httpbin_both.url + '/bytes/1024',
]
with vcr.use_cassette(str(tmpdir.join('multiple.yaml'))) as cass:
[httplib2.Http().request(url) for url in urls]
[http().request(url) for url in urls]
assert len(cass) == len(urls)
def test_get_data(scheme, tmpdir):
def test_get_data(tmpdir, httpbin_both):
'''Ensure that it works with query data'''
data = urlencode({'some': 1, 'data': 'here'})
url = scheme + '://httpbin.org/get?' + data
url = httpbin_both.url + '/get?' + data
with vcr.use_cassette(str(tmpdir.join('get_data.yaml'))):
_, res1 = httplib2.Http().request(url)
_, res1 = http().request(url)
with vcr.use_cassette(str(tmpdir.join('get_data.yaml'))):
_, res2 = httplib2.Http().request(url)
_, res2 = http().request(url)
assert res1 == res2
def test_post_data(scheme, tmpdir):
def test_post_data(tmpdir, httpbin_both):
'''Ensure that it works when posting data'''
data = urlencode({'some': 1, 'data': 'here'})
url = scheme + '://httpbin.org/post'
url = httpbin_both.url + '/post'
with vcr.use_cassette(str(tmpdir.join('post_data.yaml'))):
_, res1 = httplib2.Http().request(url, "POST", data)
_, res1 = http().request(url, "POST", data)
with vcr.use_cassette(str(tmpdir.join('post_data.yaml'))) as cass:
_, res2 = httplib2.Http().request(url, "POST", data)
_, res2 = http().request(url, "POST", data)
assert res1 == res2
assert_cassette_has_one_response(cass)
def test_post_unicode_data(scheme, tmpdir):
def test_post_unicode_data(tmpdir, httpbin_both):
'''Ensure that it works when posting unicode data'''
data = urlencode({'snowman': u'☃'.encode('utf-8')})
url = scheme + '://httpbin.org/post'
url = httpbin_both.url + '/post'
with vcr.use_cassette(str(tmpdir.join('post_data.yaml'))):
_, res1 = httplib2.Http().request(url, "POST", data)
_, res1 = http().request(url, "POST", data)
with vcr.use_cassette(str(tmpdir.join('post_data.yaml'))) as cass:
_, res2 = httplib2.Http().request(url, "POST", data)
_, res2 = http().request(url, "POST", data)
assert res1 == res2
assert_cassette_has_one_response(cass)
def test_cross_scheme(tmpdir):
def test_cross_scheme(tmpdir, httpbin, httpbin_secure):
'''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(str(tmpdir.join('cross_scheme.yaml'))) as cass:
httplib2.Http().request('https://httpbin.org/')
httplib2.Http().request('http://httpbin.org/')
http().request(httpbin_secure.url)
http().request(httpbin.url)
assert len(cass) == 2
assert cass.play_count == 0
def test_decorator(scheme, tmpdir):
def test_decorator(tmpdir, httpbin_both):
'''Test the decorator version of VCR.py'''
url = scheme + '://httpbin.org/'
url = httpbin_both.url
@vcr.use_cassette(str(tmpdir.join('atts.yaml')))
def inner1():
resp, _ = httplib2.Http().request(url)
resp, _ = http().request(url)
return resp['status']
@vcr.use_cassette(str(tmpdir.join('atts.yaml')))
def inner2():
resp, _ = httplib2.Http().request(url)
resp, _ = http().request(url)
return resp['status']
assert inner1() == inner2()

107
tests/integration/test_ignore.py
View File

@@ -1,54 +1,73 @@
from six.moves.urllib.request import urlopen
import socket
from contextlib import contextmanager
import vcr
def test_ignore_localhost(tmpdir, httpserver):
httpserver.serve_content('Hello!')
cass_file = str(tmpdir.join('filter_qs.yaml'))
with vcr.use_cassette(cass_file, ignore_localhost=True) as cass:
urlopen(httpserver.url)
assert len(cass) == 0
urlopen('http://httpbin.org')
assert len(cass) == 1
@contextmanager
def overridden_dns(overrides):
"""
Monkeypatch socket.getaddrinfo() to override DNS lookups (name will resolve
to address)
"""
real_getaddrinfo = socket.getaddrinfo
def fake_getaddrinfo(*args, **kwargs):
if args[0] in overrides:
address = overrides[args[0]]
return [(2, 1, 6, '', (address, args[1]))]
return real_getaddrinfo(*args, **kwargs)
socket.getaddrinfo = fake_getaddrinfo
yield
socket.getaddrinfo = real_getaddrinfo
def test_ignore_httpbin(tmpdir, httpserver):
httpserver.serve_content('Hello!')
cass_file = str(tmpdir.join('filter_qs.yaml'))
with vcr.use_cassette(
cass_file,
ignore_hosts=['httpbin.org']
) as cass:
urlopen('http://httpbin.org')
assert len(cass) == 0
urlopen(httpserver.url)
assert len(cass) == 1
def test_ignore_localhost(tmpdir, httpbin):
with overridden_dns({'httpbin.org': '127.0.0.1'}):
cass_file = str(tmpdir.join('filter_qs.yaml'))
with vcr.use_cassette(cass_file, ignore_localhost=True) as cass:
urlopen('http://localhost:{0}/'.format(httpbin.port))
assert len(cass) == 0
urlopen('http://httpbin.org:{0}/'.format(httpbin.port))
assert len(cass) == 1
def test_ignore_localhost_and_httpbin(tmpdir, httpserver):
httpserver.serve_content('Hello!')
cass_file = str(tmpdir.join('filter_qs.yaml'))
with vcr.use_cassette(
cass_file,
ignore_hosts=['httpbin.org'],
ignore_localhost=True
) as cass:
urlopen('http://httpbin.org')
urlopen(httpserver.url)
assert len(cass) == 0
def test_ignore_httpbin(tmpdir, httpbin):
with overridden_dns({'httpbin.org': '127.0.0.1'}):
cass_file = str(tmpdir.join('filter_qs.yaml'))
with vcr.use_cassette(
cass_file,
ignore_hosts=['httpbin.org']
) as cass:
urlopen('http://httpbin.org:{0}/'.format(httpbin.port))
assert len(cass) == 0
urlopen('http://localhost:{0}/'.format(httpbin.port))
assert len(cass) == 1
def test_ignore_localhost_twice(tmpdir, httpserver):
httpserver.serve_content('Hello!')
cass_file = str(tmpdir.join('filter_qs.yaml'))
with vcr.use_cassette(cass_file, ignore_localhost=True) as cass:
urlopen(httpserver.url)
assert len(cass) == 0
urlopen('http://httpbin.org')
assert len(cass) == 1
with vcr.use_cassette(cass_file, ignore_localhost=True) as cass:
assert len(cass) == 1
urlopen(httpserver.url)
urlopen('http://httpbin.org')
assert len(cass) == 1
def test_ignore_localhost_and_httpbin(tmpdir, httpbin):
with overridden_dns({'httpbin.org': '127.0.0.1'}):
cass_file = str(tmpdir.join('filter_qs.yaml'))
with vcr.use_cassette(
cass_file,
ignore_hosts=['httpbin.org'],
ignore_localhost=True
) as cass:
urlopen('http://httpbin.org:{0}'.format(httpbin.port))
urlopen('http://localhost:{0}'.format(httpbin.port))
assert len(cass) == 0
def test_ignore_localhost_twice(tmpdir, httpbin):
with overridden_dns({'httpbin.org': '127.0.0.1'}):
cass_file = str(tmpdir.join('filter_qs.yaml'))
with vcr.use_cassette(cass_file, ignore_localhost=True) as cass:
urlopen('http://localhost:{0}'.format(httpbin.port))
assert len(cass) == 0
urlopen('http://httpbin.org:{0}'.format(httpbin.port))
assert len(cass) == 1
with vcr.use_cassette(cass_file, ignore_localhost=True) as cass:
assert len(cass) == 1
urlopen('http://localhost:{0}'.format(httpbin.port))
urlopen('http://httpbin.org:{0}'.format(httpbin.port))
assert len(cass) == 1

42
tests/integration/test_matchers.py
View File

@@ -6,15 +6,21 @@ from six.moves.urllib.request import urlopen
DEFAULT_URI = 'http://httpbin.org/get?p1=q1&p2=q2' # base uri for testing
def _replace_httpbin(uri, httpbin, httpbin_secure):
return uri.replace('http://httpbin.org', httpbin.url).replace('https://httpbin.org', httpbin_secure.url)
@pytest.fixture
def cassette(tmpdir):
def cassette(tmpdir, httpbin, httpbin_secure):
"""
Helper fixture used to prepare the cassete
returns path to the recorded cassette
"""
default_uri = _replace_httpbin(DEFAULT_URI, httpbin, httpbin_secure)
cassette_path = str(tmpdir.join('test.yml'))
with vcr.use_cassette(cassette_path, record_mode='all'):
urlopen(DEFAULT_URI)
urlopen(default_uri)
return cassette_path
@@ -28,9 +34,6 @@ def cassette(tmpdir):
('host',
'https://httpbin.org/post?a=b',
'http://google.com/get?p1=q1&p2=q2'),
('port',
'https://google.com:80/post?a=b',
'http://httpbin.org:5000/get?p1=q1&p2=q2'),
('path',
'https://google.com/get?a=b',
'http://httpbin.org/post?p1=q1&p2=q2'),
@@ -38,10 +41,15 @@ def cassette(tmpdir):
'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):
def test_matchers(httpbin, httpbin_secure, cassette, matcher, matching_uri, not_matching_uri):
matching_uri = _replace_httpbin(matching_uri, httpbin, httpbin_secure)
not_matching_uri = _replace_httpbin(not_matching_uri, httpbin, httpbin_secure)
default_uri = _replace_httpbin(DEFAULT_URI, httpbin, httpbin_secure)
# play cassette with default uri
with vcr.use_cassette(cassette, match_on=[matcher]) as cass:
urlopen(DEFAULT_URI)
urlopen(default_uri)
assert cass.play_count == 1
# play cassette with matching on uri
@@ -55,7 +63,9 @@ def test_matchers(cassette, matcher, matching_uri, not_matching_uri):
urlopen(not_matching_uri)
def test_method_matcher(cassette):
def test_method_matcher(cassette, httpbin, httpbin_secure):
default_uri = _replace_httpbin(DEFAULT_URI, httpbin, httpbin_secure)
# play cassette with matching on method
with vcr.use_cassette(cassette, match_on=['method']) as cass:
urlopen('https://google.com/get?a=b')
@@ -65,7 +75,7 @@ def test_method_matcher(cassette):
with pytest.raises(vcr.errors.CannotOverwriteExistingCassetteException):
with vcr.use_cassette(cassette, match_on=['method']) as cass:
# is a POST request
urlopen(DEFAULT_URI, data=b'')
urlopen(default_uri, data=b'')
@pytest.mark.parametrize("uri", [
@@ -73,7 +83,10 @@ def test_method_matcher(cassette):
'http://httpbin.org/get?p2=q2&p1=q1',
'http://httpbin.org/get?p2=q2&p1=q1',
])
def test_default_matcher_matches(cassette, uri):
def test_default_matcher_matches(cassette, uri, httpbin, httpbin_secure):
uri = _replace_httpbin(uri, httpbin, httpbin_secure)
with vcr.use_cassette(cassette) as cass:
urlopen(uri)
assert cass.play_count == 1
@@ -82,18 +95,19 @@ def test_default_matcher_matches(cassette, uri):
@pytest.mark.parametrize("uri", [
'https://httpbin.org/get?p1=q1&p2=q2',
'http://google.com/get?p1=q1&p2=q2',
'http://httpbin.org:5000/get?p1=q1&p2=q2',
'http://httpbin.org/post?p1=q1&p2=q2',
'http://httpbin.org/get?p1=q1&a=b'
])
def test_default_matcher_does_not_match(cassette, uri):
def test_default_matcher_does_not_match(cassette, uri, httpbin, httpbin_secure):
uri = _replace_httpbin(uri, httpbin, httpbin_secure)
with pytest.raises(vcr.errors.CannotOverwriteExistingCassetteException):
with vcr.use_cassette(cassette):
urlopen(uri)
def test_default_matcher_does_not_match_on_method(cassette):
def test_default_matcher_does_not_match_on_method(cassette, httpbin, httpbin_secure):
default_uri = _replace_httpbin(DEFAULT_URI, httpbin, httpbin_secure)
with pytest.raises(vcr.errors.CannotOverwriteExistingCassetteException):
with vcr.use_cassette(cassette):
# is a POST request
urlopen(DEFAULT_URI, data=b'')
urlopen(default_uri, data=b'')

12
tests/integration/test_multiple.py
View File

@@ -3,18 +3,18 @@ import vcr
from six.moves.urllib.request import urlopen
def test_making_extra_request_raises_exception(tmpdir):
def test_making_extra_request_raises_exception(tmpdir, httpbin):
# make two requests in the first request that are considered
# identical (since the match is based on method)
with vcr.use_cassette(str(tmpdir.join('test.json')), match_on=['method']):
urlopen('http://httpbin.org/status/200')
urlopen('http://httpbin.org/status/201')
urlopen(httpbin.url + '/status/200')
urlopen(httpbin.url + '/status/201')
# Now, try to make three requests. The first two should return the
# correct status codes in order, and the third should raise an
# exception.
with vcr.use_cassette(str(tmpdir.join('test.json')), match_on=['method']):
assert urlopen('http://httpbin.org/status/200').getcode() == 200
assert urlopen('http://httpbin.org/status/201').getcode() == 201
assert urlopen(httpbin.url + '/status/200').getcode() == 200
assert urlopen(httpbin.url + '/status/201').getcode() == 201
with pytest.raises(Exception):
urlopen('http://httpbin.org/status/200')
urlopen(httpbin.url + '/status/200')

56
tests/integration/test_record_mode.py
View File

@@ -3,62 +3,62 @@ import vcr
from six.moves.urllib.request import urlopen
def test_once_record_mode(tmpdir):
def test_once_record_mode(tmpdir, httpbin):
testfile = str(tmpdir.join('recordmode.yml'))
with vcr.use_cassette(testfile, record_mode="once"):
# cassette file doesn't exist, so create.
urlopen('http://httpbin.org/').read()
urlopen(httpbin.url).read()
with vcr.use_cassette(testfile, record_mode="once"):
# make the same request again
urlopen('http://httpbin.org/').read()
urlopen(httpbin.url).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):
urlopen('http://httpbin.org/get').read()
urlopen(httpbin.url + '/get').read()
def test_once_record_mode_two_times(tmpdir):
def test_once_record_mode_two_times(tmpdir, httpbin):
testfile = str(tmpdir.join('recordmode.yml'))
with vcr.use_cassette(testfile, record_mode="once"):
# get two of the same file
urlopen('http://httpbin.org/').read()
urlopen('http://httpbin.org/').read()
urlopen(httpbin.url).read()
urlopen(httpbin.url).read()
with vcr.use_cassette(testfile, record_mode="once"):
# do it again
urlopen('http://httpbin.org/').read()
urlopen('http://httpbin.org/').read()
urlopen(httpbin.url).read()
urlopen(httpbin.url).read()
def test_once_mode_three_times(tmpdir):
def test_once_mode_three_times(tmpdir, httpbin):
testfile = str(tmpdir.join('recordmode.yml'))
with vcr.use_cassette(testfile, record_mode="once"):
# get three of the same file
urlopen('http://httpbin.org/').read()
urlopen('http://httpbin.org/').read()
urlopen('http://httpbin.org/').read()
urlopen(httpbin.url).read()
urlopen(httpbin.url).read()
urlopen(httpbin.url).read()
def test_new_episodes_record_mode(tmpdir):
def test_new_episodes_record_mode(tmpdir, httpbin):
testfile = str(tmpdir.join('recordmode.yml'))
with vcr.use_cassette(testfile, record_mode="new_episodes"):
# cassette file doesn't exist, so create.
urlopen('http://httpbin.org/').read()
urlopen(httpbin.url).read()
with vcr.use_cassette(testfile, record_mode="new_episodes") as cass:
# make the same request again
urlopen('http://httpbin.org/').read()
urlopen(httpbin.url).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.
urlopen('http://httpbin.org/get').read()
urlopen(httpbin.url + '/get').read()
# one of the responses has been played
assert cass.play_count == 1
@@ -71,9 +71,9 @@ def test_new_episodes_record_mode(tmpdir):
assert len(cass.responses) == 2
def test_new_episodes_record_mode_two_times(tmpdir):
def test_new_episodes_record_mode_two_times(tmpdir, httpbin):
testfile = str(tmpdir.join('recordmode.yml'))
url = 'http://httpbin.org/bytes/1024'
url = httpbin.url + '/bytes/1024'
with vcr.use_cassette(testfile, record_mode="new_episodes"):
# cassette file doesn't exist, so create.
original_first_response = urlopen(url).read()
@@ -96,12 +96,12 @@ def test_new_episodes_record_mode_two_times(tmpdir):
urlopen(url).read()
def test_all_record_mode(tmpdir):
def test_all_record_mode(tmpdir, httpbin):
testfile = str(tmpdir.join('recordmode.yml'))
with vcr.use_cassette(testfile, record_mode="all"):
# cassette file doesn't exist, so create.
urlopen('http://httpbin.org/').read()
urlopen(httpbin.url).read()
with vcr.use_cassette(testfile, record_mode="all") as cass:
# make the same request again
@@ -109,7 +109,7 @@ def test_all_record_mode(tmpdir):
# in the "all" record mode, we can add more requests to
# a cassette without repurcussions.
urlopen('http://httpbin.org/get').read()
urlopen(httpbin.url + '/get').read()
# The cassette was never actually played, even though it existed.
# that's because, in "all" mode, the requests all go directly to
@@ -117,26 +117,26 @@ def test_all_record_mode(tmpdir):
assert cass.play_count == 0
def test_none_record_mode(tmpdir):
def test_none_record_mode(tmpdir, httpbin):
# Cassette file doesn't exist, yet we are trying to make a request.
# raise hell.
testfile = str(tmpdir.join('recordmode.yml'))
with vcr.use_cassette(testfile, record_mode="none"):
with pytest.raises(Exception):
urlopen('http://httpbin.org/').read()
urlopen(httpbin.url).read()
def test_none_record_mode_with_existing_cassette(tmpdir):
def test_none_record_mode_with_existing_cassette(tmpdir, httpbin):
# create a cassette file
testfile = str(tmpdir.join('recordmode.yml'))
with vcr.use_cassette(testfile, record_mode="all"):
urlopen('http://httpbin.org/').read()
urlopen(httpbin.url).read()
# play from cassette file
with vcr.use_cassette(testfile, record_mode="none") as cass:
urlopen('http://httpbin.org/').read()
urlopen(httpbin.url).read()
assert cass.play_count == 1
# but if I try to hit the net, raise an exception.
with pytest.raises(Exception):
urlopen('http://httpbin.org/get').read()
urlopen(httpbin.url + '/get').read()

16
tests/integration/test_register_matcher.py
View File

@@ -10,27 +10,27 @@ def false_matcher(r1, r2):
return False
def test_registered_true_matcher(tmpdir):
def test_registered_true_matcher(tmpdir, httpbin):
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']):
# These 2 different urls are stored as the same request
urlopen('http://httpbin.org/')
urlopen('https://httpbin.org/get')
urlopen(httpbin.url)
urlopen(httpbin.url + '/get')
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')
urlopen(httpbin.url + '/get')
urlopen(httpbin.url + '/get')
def test_registered_false_matcher(tmpdir):
def test_registered_false_matcher(tmpdir, httpbin):
my_vcr = vcr.VCR()
my_vcr.register_matcher('false', false_matcher)
testfile = str(tmpdir.join('test.yml'))
with my_vcr.use_cassette(testfile, match_on=['false']) as cass:
# These 2 different urls are stored as different requests
urlopen('http://httpbin.org/')
urlopen('https://httpbin.org/get')
urlopen(httpbin.url)
urlopen(httpbin.url + '/get')
assert len(cass) == 2

15
tests/integration/test_request.py
View File

@@ -2,21 +2,18 @@ import vcr
from six.moves.urllib.request import urlopen
def test_recorded_request_uri_with_redirected_request(tmpdir):
def test_recorded_request_uri_with_redirected_request(tmpdir, httpbin):
with vcr.use_cassette(str(tmpdir.join('test.yml'))) as cass:
assert len(cass) == 0
urlopen('http://httpbin.org/redirect/3')
assert cass.requests[0].uri == 'http://httpbin.org/redirect/3'
assert cass.requests[3].uri == 'http://httpbin.org/get'
urlopen(httpbin.url + '/redirect/3')
assert cass.requests[0].uri == httpbin.url + '/redirect/3'
assert cass.requests[3].uri == httpbin.url + '/get'
assert len(cass) == 4
def test_records_multiple_header_values(tmpdir, httpserver):
httpserver.serve_content('Hello!', headers=[('foo', 'bar'), ('foo', 'baz')])
def test_records_multiple_header_values(tmpdir, httpbin):
with vcr.use_cassette(str(tmpdir.join('test.yml'))) as cass:
assert len(cass) == 0
urlopen(httpserver.url)
urlopen(httpbin.url + '/response-headers?foo=bar&foo=baz')
assert len(cass) == 1
assert cass.responses[0]['headers']['foo'] == ['bar', 'baz']

92
tests/integration/test_requests.py
View File

@@ -8,15 +8,9 @@ from assertions import assert_cassette_empty, assert_is_json
requests = pytest.importorskip("requests")
@pytest.fixture(params=["https", "http"])
def scheme(request):
'''Fixture that returns both http and https.'''
return request.param
def test_status_code(scheme, tmpdir):
def test_status_code(httpbin_both, tmpdir):
'''Ensure that we can read the status code'''
url = scheme + '://httpbin.org/'
url = httpbin_both.url + '/'
with vcr.use_cassette(str(tmpdir.join('atts.yaml'))):
status_code = requests.get(url).status_code
@@ -24,9 +18,9 @@ def test_status_code(scheme, tmpdir):
assert status_code == requests.get(url).status_code
def test_headers(scheme, tmpdir):
def test_headers(httpbin_both, tmpdir):
'''Ensure that we can read the headers back'''
url = scheme + '://httpbin.org/'
url = httpbin_both + '/'
with vcr.use_cassette(str(tmpdir.join('headers.yaml'))):
headers = requests.get(url).headers
@@ -34,9 +28,9 @@ def test_headers(scheme, tmpdir):
assert headers == requests.get(url).headers
def test_body(tmpdir, scheme):
def test_body(tmpdir, httpbin_both):
'''Ensure the responses are all identical enough'''
url = scheme + '://httpbin.org/bytes/1024'
url = httpbin_both + '/bytes/1024'
with vcr.use_cassette(str(tmpdir.join('body.yaml'))):
content = requests.get(url).content
@@ -44,21 +38,21 @@ def test_body(tmpdir, scheme):
assert content == requests.get(url).content
def test_effective_url(scheme, tmpdir):
def test_effective_url(tmpdir, httpbin_both):
'''Ensure that the effective_url is captured'''
url = scheme + '://httpbin.org/redirect-to?url=/html'
url = httpbin_both.url + '/redirect-to?url=/html'
with vcr.use_cassette(str(tmpdir.join('url.yaml'))):
effective_url = requests.get(url).url
assert effective_url == scheme + '://httpbin.org/html'
assert effective_url == httpbin_both.url + '/html'
with vcr.use_cassette(str(tmpdir.join('url.yaml'))):
assert effective_url == requests.get(url).url
def test_auth(tmpdir, scheme):
def test_auth(tmpdir, httpbin_both):
'''Ensure that we can handle basic auth'''
auth = ('user', 'passwd')
url = scheme + '://httpbin.org/basic-auth/user/passwd'
url = httpbin_both + '/basic-auth/user/passwd'
with vcr.use_cassette(str(tmpdir.join('auth.yaml'))):
one = requests.get(url, auth=auth)
@@ -68,10 +62,10 @@ def test_auth(tmpdir, scheme):
assert one.status_code == two.status_code
def test_auth_failed(tmpdir, scheme):
def test_auth_failed(tmpdir, httpbin_both):
'''Ensure that we can save failed auth statuses'''
auth = ('user', 'wrongwrongwrong')
url = scheme + '://httpbin.org/basic-auth/user/passwd'
url = httpbin_both + '/basic-auth/user/passwd'
with vcr.use_cassette(str(tmpdir.join('auth-failed.yaml'))) as cass:
# Ensure that this is empty to begin with
assert_cassette_empty(cass)
@@ -81,10 +75,10 @@ def test_auth_failed(tmpdir, scheme):
assert one.status_code == two.status_code == 401
def test_post(tmpdir, scheme):
def test_post(tmpdir, httpbin_both):
'''Ensure that we can post and cache the results'''
data = {'key1': 'value1', 'key2': 'value2'}
url = scheme + '://httpbin.org/post'
url = httpbin_both + '/post'
with vcr.use_cassette(str(tmpdir.join('requests.yaml'))):
req1 = requests.post(url, data).content
@@ -94,11 +88,11 @@ def test_post(tmpdir, scheme):
assert req1 == req2
def test_post_chunked_binary(tmpdir, scheme):
def test_post_chunked_binary(tmpdir, httpbin_both):
'''Ensure that we can send chunked binary without breaking while trying to concatenate bytes with str.'''
data1 = iter([b'data', b'to', b'send'])
data2 = iter([b'data', b'to', b'send'])
url = scheme + '://httpbin.org/post'
url = httpbin_both.url + '/post'
with vcr.use_cassette(str(tmpdir.join('requests.yaml'))):
req1 = requests.post(url, data1).content
print(req1)
@@ -109,9 +103,9 @@ def test_post_chunked_binary(tmpdir, scheme):
assert req1 == req2
def test_redirects(tmpdir, scheme):
def test_redirects(tmpdir, httpbin_both):
'''Ensure that we can handle redirects'''
url = scheme + '://httpbin.org/redirect-to?url=bytes/1024'
url = httpbin_both + '/redirect-to?url=bytes/1024'
with vcr.use_cassette(str(tmpdir.join('requests.yaml'))):
content = requests.get(url).content
@@ -123,24 +117,24 @@ def test_redirects(tmpdir, scheme):
assert cass.play_count == 2
def test_cross_scheme(tmpdir, scheme):
def test_cross_scheme(tmpdir, httpbin_secure, httpbin):
'''Ensure that requests between schemes are treated separately'''
# First fetch a url under http, 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(str(tmpdir.join('cross_scheme.yaml'))) as cass:
requests.get('https://httpbin.org/')
requests.get('http://httpbin.org/')
requests.get(httpbin_secure + '/')
requests.get(httpbin + '/')
assert cass.play_count == 0
assert len(cass) == 2
def test_gzip(tmpdir, scheme):
def test_gzip(tmpdir, httpbin_both):
'''
Ensure that requests (actually urllib3) is able to automatically decompress
the response body
'''
url = scheme + '://httpbin.org/gzip'
url = httpbin_both + '/gzip'
response = requests.get(url)
with vcr.use_cassette(str(tmpdir.join('gzip.yaml'))):
@@ -151,7 +145,7 @@ def test_gzip(tmpdir, scheme):
assert_is_json(response.content)
def test_session_and_connection_close(tmpdir, scheme):
def test_session_and_connection_close(tmpdir, httpbin):
'''
This tests the issue in https://github.com/kevin1024/vcrpy/issues/48
@@ -162,29 +156,29 @@ def test_session_and_connection_close(tmpdir, scheme):
with vcr.use_cassette(str(tmpdir.join('session_connection_closed.yaml'))):
session = requests.session()
session.get('http://httpbin.org/get', headers={'Connection': 'close'})
session.get('http://httpbin.org/get', headers={'Connection': 'close'})
session.get(httpbin + '/get', headers={'Connection': 'close'})
session.get(httpbin + '/get', headers={'Connection': 'close'})
def test_https_with_cert_validation_disabled(tmpdir):
def test_https_with_cert_validation_disabled(tmpdir, httpbin_secure):
with vcr.use_cassette(str(tmpdir.join('cert_validation_disabled.yaml'))):
requests.get('https://httpbin.org', verify=False)
requests.get(httpbin_secure.url, verify=False)
def test_session_can_make_requests_after_requests_unpatched(tmpdir):
def test_session_can_make_requests_after_requests_unpatched(tmpdir, httpbin):
with vcr.use_cassette(str(tmpdir.join('test_session_after_unpatched.yaml'))):
session = requests.session()
session.get('http://httpbin.org/get')
session.get(httpbin + '/get')
with vcr.use_cassette(str(tmpdir.join('test_session_after_unpatched.yaml'))):
session = requests.session()
session.get('http://httpbin.org/get')
session.get(httpbin + '/get')
session.get('http://httpbin.org/status/200')
session.get(httpbin + '/status/200')
def test_session_created_before_use_cassette_is_patched(tmpdir, scheme):
url = scheme + '://httpbin.org/bytes/1024'
def test_session_created_before_use_cassette_is_patched(tmpdir, httpbin_both):
url = httpbin_both + '/bytes/1024'
# Record arbitrary, random data to the cassette
with vcr.use_cassette(str(tmpdir.join('session_created_outside.yaml'))):
session = requests.session()
@@ -193,20 +187,20 @@ def test_session_created_before_use_cassette_is_patched(tmpdir, scheme):
# Create a session outside of any cassette context manager
session = requests.session()
# Make a request to make sure that a connectionpool is instantiated
session.get(scheme + '://httpbin.org/get')
session.get(httpbin_both + '/get')
with vcr.use_cassette(str(tmpdir.join('session_created_outside.yaml'))):
# These should only be the same if the patching succeeded.
assert session.get(url).content == body
def test_nested_cassettes_with_session_created_before_nesting(scheme, tmpdir):
def test_nested_cassettes_with_session_created_before_nesting(httpbin_both, tmpdir):
'''
This tests ensures that a session that was created while one cassette was
active is patched to the use the responses of a second cassette when it
is enabled.
'''
url = scheme + '://httpbin.org/bytes/1024'
url = httpbin_both + '/bytes/1024'
with vcr.use_cassette(str(tmpdir.join('first_nested.yaml'))):
session = requests.session()
first_body = session.get(url).content
@@ -222,12 +216,12 @@ def test_nested_cassettes_with_session_created_before_nesting(scheme, tmpdir):
assert session.get(url).content == third_body
# Make sure that the session can now get content normally.
session.get('http://www.reddit.com')
assert 'User-agent' in session.get(httpbin_both.url + '/robots.txt').text
def test_post_file(tmpdir, scheme):
def test_post_file(tmpdir, httpbin_both):
'''Ensure that we handle posting a file.'''
url = scheme + '://httpbin.org/post'
url = httpbin_both + '/post'
with vcr.use_cassette(str(tmpdir.join('post_file.yaml'))) as cass:
# Don't use 2.7+ only style ',' separated with here because we support python 2.6
with open('tox.ini') as f:
@@ -244,7 +238,7 @@ def test_post_file(tmpdir, scheme):
assert original_response == new_response
def test_filter_post_params(tmpdir, scheme):
def test_filter_post_params(tmpdir, httpbin_both):
'''
This tests the issue in https://github.com/kevin1024/vcrpy/issues/158
@@ -252,7 +246,7 @@ def test_filter_post_params(tmpdir, scheme):
with vcr.use_cassette(cass_file, filter_post_data_parameters=['id']) as cass:
assert b'id=secret' not in cass.requests[0].body
'''
url = scheme + '://httpbin.org/post'
url = httpbin_both.url + '/post'
cass_loc = str(tmpdir.join('filter_post_params.yaml'))
with vcr.use_cassette(cass_loc, filter_post_data_parameters=['key']) as cass:
requests.post(url, data={'key': 'value'})

28
tests/integration/test_stubs.py
View File

@@ -2,45 +2,45 @@ import vcr
import six.moves.http_client as httplib
def _headers_are_case_insensitive():
conn = httplib.HTTPConnection('httpbin.org')
def _headers_are_case_insensitive(host, port):
conn = httplib.HTTPConnection(host, port)
conn.request('GET', "/cookies/set?k1=v1")
r1 = conn.getresponse()
cookie_data1 = r1.getheader('set-cookie')
conn = httplib.HTTPConnection('httpbin.org')
conn = httplib.HTTPConnection(host, port)
conn.request('GET', "/cookies/set?k1=v1")
r2 = conn.getresponse()
cookie_data2 = r2.getheader('Set-Cookie')
return cookie_data1 == cookie_data2
def test_case_insensitivity(tmpdir):
def test_case_insensitivity(tmpdir, httpbin):
testfile = str(tmpdir.join('case_insensitivity.yml'))
# check if headers are case insensitive outside of vcrpy
outside = _headers_are_case_insensitive()
host, port = httpbin.host, httpbin.port
outside = _headers_are_case_insensitive(host, port)
with vcr.use_cassette(testfile):
# check if headers are case insensitive inside of vcrpy
inside = _headers_are_case_insensitive()
inside = _headers_are_case_insensitive(host, port)
# check if headers are case insensitive after vcrpy deserializes headers
inside2 = _headers_are_case_insensitive()
inside2 = _headers_are_case_insensitive(host, port)
# 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', "/")
def _multiple_header_value(httpbin):
conn = httplib.HTTPConnection(httpbin.host, httpbin.port)
conn.request('GET', "/response-headers?foo=bar&foo=baz")
r = conn.getresponse()
return r.getheader('foo')
def test_multiple_headers(tmpdir, httpserver):
def test_multiple_headers(tmpdir, httpbin):
testfile = str(tmpdir.join('multiple_headers.yaml'))
httpserver.serve_content('Hello!', headers=[('foo', 'bar'), ('foo', 'baz')])
outside = _multiple_header_value(httpserver)
outside = _multiple_header_value(httpbin)
with vcr.use_cassette(testfile):
inside = _multiple_header_value(httpserver)
inside = _multiple_header_value(httpbin)
assert outside == inside

101
tests/integration/test_urllib2.py
View File

@@ -1,9 +1,9 @@
# -*- coding: utf-8 -*-
'''Integration tests with urllib2'''
import pytest
from six.moves.urllib.request import urlopen
from six.moves.urllib_parse import urlencode
import pytest_httpbin.certs
# Internal imports
import vcr
@@ -11,135 +11,136 @@ import vcr
from assertions import assert_cassette_has_one_response
@pytest.fixture(params=["https", "http"])
def scheme(request):
"""
Fixture that returns both http and https
"""
return request.param
def urlopen_with_cafile(*args, **kwargs):
kwargs['cafile'] = pytest_httpbin.certs.where()
try:
return urlopen(*args, **kwargs)
except TypeError:
# python2/pypi don't let us override this
del kwargs['cafile']
return urlopen(*args, **kwargs)
def test_response_code(scheme, tmpdir):
def test_response_code(httpbin_both, tmpdir):
'''Ensure we can read a response code from a fetch'''
url = scheme + '://httpbin.org/'
url = httpbin_both.url
with vcr.use_cassette(str(tmpdir.join('atts.yaml'))):
code = urlopen(url).getcode()
code = urlopen_with_cafile(url).getcode()
with vcr.use_cassette(str(tmpdir.join('atts.yaml'))):
assert code == urlopen(url).getcode()
assert code == urlopen_with_cafile(url).getcode()
def test_random_body(scheme, tmpdir):
def test_random_body(httpbin_both, tmpdir):
'''Ensure we can read the content, and that it's served from cache'''
url = scheme + '://httpbin.org/bytes/1024'
url = httpbin_both.url + '/bytes/1024'
with vcr.use_cassette(str(tmpdir.join('body.yaml'))):
body = urlopen(url).read()
body = urlopen_with_cafile(url).read()
with vcr.use_cassette(str(tmpdir.join('body.yaml'))):
assert body == urlopen(url).read()
assert body == urlopen_with_cafile(url).read()
def test_response_headers(scheme, tmpdir):
def test_response_headers(httpbin_both, tmpdir):
'''Ensure we can get information from the response'''
url = scheme + '://httpbin.org/'
url = httpbin_both.url
with vcr.use_cassette(str(tmpdir.join('headers.yaml'))):
open1 = urlopen(url).info().items()
open1 = urlopen_with_cafile(url).info().items()
with vcr.use_cassette(str(tmpdir.join('headers.yaml'))):
open2 = urlopen(url).info().items()
open2 = urlopen_with_cafile(url).info().items()
assert sorted(open1) == sorted(open2)
def test_effective_url(scheme, tmpdir):
def test_effective_url(httpbin_both, tmpdir):
'''Ensure that the effective_url is captured'''
url = scheme + '://httpbin.org/redirect-to?url=/html'
url = httpbin_both.url + '/redirect-to?url=/html'
with vcr.use_cassette(str(tmpdir.join('headers.yaml'))):
effective_url = urlopen(url).geturl()
assert effective_url == scheme + '://httpbin.org/html'
effective_url = urlopen_with_cafile(url).geturl()
assert effective_url == httpbin_both.url + '/html'
with vcr.use_cassette(str(tmpdir.join('headers.yaml'))):
assert effective_url == urlopen(url).geturl()
assert effective_url == urlopen_with_cafile(url).geturl()
def test_multiple_requests(scheme, tmpdir):
def test_multiple_requests(httpbin_both, tmpdir):
'''Ensure that we can cache multiple requests'''
urls = [
scheme + '://httpbin.org/',
scheme + '://httpbin.org/',
scheme + '://httpbin.org/get',
scheme + '://httpbin.org/bytes/1024'
httpbin_both.url,
httpbin_both.url,
httpbin_both.url + '/get',
httpbin_both.url + '/bytes/1024',
]
with vcr.use_cassette(str(tmpdir.join('multiple.yaml'))) as cass:
[urlopen(url) for url in urls]
[urlopen_with_cafile(url) for url in urls]
assert len(cass) == len(urls)
def test_get_data(scheme, tmpdir):
def test_get_data(httpbin_both, tmpdir):
'''Ensure that it works with query data'''
data = urlencode({'some': 1, 'data': 'here'})
url = scheme + '://httpbin.org/get?' + data
url = httpbin_both.url + '/get?' + data
with vcr.use_cassette(str(tmpdir.join('get_data.yaml'))):
res1 = urlopen(url).read()
res1 = urlopen_with_cafile(url).read()
with vcr.use_cassette(str(tmpdir.join('get_data.yaml'))):
res2 = urlopen(url).read()
res2 = urlopen_with_cafile(url).read()
assert res1 == res2
def test_post_data(scheme, tmpdir):
def test_post_data(httpbin_both, tmpdir):
'''Ensure that it works when posting data'''
data = urlencode({'some': 1, 'data': 'here'}).encode('utf-8')
url = scheme + '://httpbin.org/post'
url = httpbin_both.url + '/post'
with vcr.use_cassette(str(tmpdir.join('post_data.yaml'))):
res1 = urlopen(url, data).read()
res1 = urlopen_with_cafile(url, data).read()
with vcr.use_cassette(str(tmpdir.join('post_data.yaml'))) as cass:
res2 = urlopen(url, data).read()
res2 = urlopen_with_cafile(url, data).read()
assert len(cass) == 1
assert res1 == res2
assert_cassette_has_one_response(cass)
def test_post_unicode_data(scheme, tmpdir):
def test_post_unicode_data(httpbin_both, tmpdir):
'''Ensure that it works when posting unicode data'''
data = urlencode({'snowman': u'☃'.encode('utf-8')}).encode('utf-8')
url = scheme + '://httpbin.org/post'
url = httpbin_both.url + '/post'
with vcr.use_cassette(str(tmpdir.join('post_data.yaml'))):
res1 = urlopen(url, data).read()
res1 = urlopen_with_cafile(url, data).read()
with vcr.use_cassette(str(tmpdir.join('post_data.yaml'))) as cass:
res2 = urlopen(url, data).read()
res2 = urlopen_with_cafile(url, data).read()
assert len(cass) == 1
assert res1 == res2
assert_cassette_has_one_response(cass)
def test_cross_scheme(tmpdir):
def test_cross_scheme(tmpdir, httpbin_secure, httpbin):
'''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(str(tmpdir.join('cross_scheme.yaml'))) as cass:
urlopen('https://httpbin.org/')
urlopen('http://httpbin.org/')
urlopen_with_cafile(httpbin_secure.url)
urlopen_with_cafile(httpbin.url)
assert len(cass) == 2
assert cass.play_count == 0
def test_decorator(scheme, tmpdir):
def test_decorator(httpbin_both, tmpdir):
'''Test the decorator version of VCR.py'''
url = scheme + '://httpbin.org/'
url = httpbin_both.url
@vcr.use_cassette(str(tmpdir.join('atts.yaml')))
def inner1():
return urlopen(url).getcode()
return urlopen_with_cafile(url).getcode()
@vcr.use_cassette(str(tmpdir.join('atts.yaml')))
def inner2():
return urlopen(url).getcode()
return urlopen_with_cafile(url).getcode()
assert inner1() == inner2()

54
tests/integration/test_urllib3.py
View File

@@ -3,25 +3,17 @@
# coding=utf-8
import pytest
import pytest_httpbin
import vcr
from assertions import assert_cassette_empty, assert_is_json
certifi = pytest.importorskip("certifi")
urllib3 = pytest.importorskip("urllib3")
@pytest.fixture(params=["https", "http"])
def scheme(request):
"""
Fixture that returns both http and https
"""
return request.param
@pytest.fixture(scope='module')
def verify_pool_mgr():
return urllib3.PoolManager(
cert_reqs='CERT_REQUIRED', # Force certificate check.
ca_certs=certifi.where()
ca_certs=pytest_httpbin.certs.where()
)
@@ -30,9 +22,9 @@ def pool_mgr():
return urllib3.PoolManager()
def test_status_code(scheme, tmpdir, verify_pool_mgr):
def test_status_code(httpbin_both, tmpdir, verify_pool_mgr):
'''Ensure that we can read the status code'''
url = scheme + '://httpbin.org/'
url = httpbin_both.url
with vcr.use_cassette(str(tmpdir.join('atts.yaml'))):
status_code = verify_pool_mgr.request('GET', url).status
@@ -40,9 +32,9 @@ def test_status_code(scheme, tmpdir, verify_pool_mgr):
assert status_code == verify_pool_mgr.request('GET', url).status
def test_headers(scheme, tmpdir, verify_pool_mgr):
def test_headers(tmpdir, httpbin_both, verify_pool_mgr):
'''Ensure that we can read the headers back'''
url = scheme + '://httpbin.org/'
url = httpbin_both.url
with vcr.use_cassette(str(tmpdir.join('headers.yaml'))):
headers = verify_pool_mgr.request('GET', url).headers
@@ -50,9 +42,9 @@ def test_headers(scheme, tmpdir, verify_pool_mgr):
assert headers == verify_pool_mgr.request('GET', url).headers
def test_body(tmpdir, scheme, verify_pool_mgr):
def test_body(tmpdir, httpbin_both, verify_pool_mgr):
'''Ensure the responses are all identical enough'''
url = scheme + '://httpbin.org/bytes/1024'
url = httpbin_both.url + '/bytes/1024'
with vcr.use_cassette(str(tmpdir.join('body.yaml'))):
content = verify_pool_mgr.request('GET', url).data
@@ -60,11 +52,11 @@ def test_body(tmpdir, scheme, verify_pool_mgr):
assert content == verify_pool_mgr.request('GET', url).data
def test_auth(tmpdir, scheme, verify_pool_mgr):
def test_auth(tmpdir, httpbin_both, verify_pool_mgr):
'''Ensure that we can handle basic auth'''
auth = ('user', 'passwd')
headers = urllib3.util.make_headers(basic_auth='{0}:{1}'.format(*auth))
url = scheme + '://httpbin.org/basic-auth/user/passwd'
url = httpbin_both.url + '/basic-auth/user/passwd'
with vcr.use_cassette(str(tmpdir.join('auth.yaml'))):
one = verify_pool_mgr.request('GET', url, headers=headers)
@@ -74,11 +66,11 @@ def test_auth(tmpdir, scheme, verify_pool_mgr):
assert one.status == two.status
def test_auth_failed(tmpdir, scheme, verify_pool_mgr):
def test_auth_failed(tmpdir, httpbin_both, verify_pool_mgr):
'''Ensure that we can save failed auth statuses'''
auth = ('user', 'wrongwrongwrong')
headers = urllib3.util.make_headers(basic_auth='{0}:{1}'.format(*auth))
url = scheme + '://httpbin.org/basic-auth/user/passwd'
url = httpbin_both.url + '/basic-auth/user/passwd'
with vcr.use_cassette(str(tmpdir.join('auth-failed.yaml'))) as cass:
# Ensure that this is empty to begin with
assert_cassette_empty(cass)
@@ -88,10 +80,10 @@ def test_auth_failed(tmpdir, scheme, verify_pool_mgr):
assert one.status == two.status == 401
def test_post(tmpdir, scheme, verify_pool_mgr):
def test_post(tmpdir, httpbin_both, verify_pool_mgr):
'''Ensure that we can post and cache the results'''
data = {'key1': 'value1', 'key2': 'value2'}
url = scheme + '://httpbin.org/post'
url = httpbin_both.url + '/post'
with vcr.use_cassette(str(tmpdir.join('verify_pool_mgr.yaml'))):
req1 = verify_pool_mgr.request('POST', url, data).data
@@ -101,9 +93,9 @@ def test_post(tmpdir, scheme, verify_pool_mgr):
assert req1 == req2
def test_redirects(tmpdir, scheme, verify_pool_mgr):
def test_redirects(tmpdir, httpbin_both, verify_pool_mgr):
'''Ensure that we can handle redirects'''
url = scheme + '://httpbin.org/redirect-to?url=bytes/1024'
url = httpbin_both.url + '/redirect-to?url=bytes/1024'
with vcr.use_cassette(str(tmpdir.join('verify_pool_mgr.yaml'))):
content = verify_pool_mgr.request('GET', url).data
@@ -115,24 +107,24 @@ def test_redirects(tmpdir, scheme, verify_pool_mgr):
assert cass.play_count == 2
def test_cross_scheme(tmpdir, scheme, verify_pool_mgr):
def test_cross_scheme(tmpdir, httpbin, httpbin_secure, verify_pool_mgr):
'''Ensure that requests between schemes are treated separately'''
# First fetch a url under http, 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(str(tmpdir.join('cross_scheme.yaml'))) as cass:
verify_pool_mgr.request('GET', 'https://httpbin.org/')
verify_pool_mgr.request('GET', 'http://httpbin.org/')
verify_pool_mgr.request('GET', httpbin_secure.url)
verify_pool_mgr.request('GET', httpbin.url)
assert cass.play_count == 0
assert len(cass) == 2
def test_gzip(tmpdir, scheme, verify_pool_mgr):
def test_gzip(tmpdir, httpbin_both, verify_pool_mgr):
'''
Ensure that requests (actually urllib3) is able to automatically decompress
the response body
'''
url = scheme + '://httpbin.org/gzip'
url = httpbin_both.url + '/gzip'
response = verify_pool_mgr.request('GET', url)
with vcr.use_cassette(str(tmpdir.join('gzip.yaml'))):
@@ -143,6 +135,6 @@ def test_gzip(tmpdir, scheme, verify_pool_mgr):
assert_is_json(response.data)
def test_https_with_cert_validation_disabled(tmpdir, pool_mgr):
def test_https_with_cert_validation_disabled(tmpdir, httpbin_secure, pool_mgr):
with vcr.use_cassette(str(tmpdir.join('cert_validation_disabled.yaml'))):
pool_mgr.request('GET', 'https://httpbin.org')
pool_mgr.request('GET', httpbin_secure.url)

15
tests/integration/test_wild.py
View File

@@ -24,14 +24,14 @@ def test_domain_redirect():
assert len(cass) == 2
def test_flickr_multipart_upload():
def test_flickr_multipart_upload(httpbin, tmpdir):
"""
The python-flickr-api project does a multipart
upload that confuses vcrpy
"""
def _pretend_to_be_flickr_library():
content_type, body = "text/plain", "HELLO WORLD"
h = httplib.HTTPConnection("httpbin.org")
h = httplib.HTTPConnection(httpbin.host, httpbin.port)
headers = {
"Content-Type": content_type,
"content-length": str(len(body))
@@ -44,11 +44,12 @@ def test_flickr_multipart_upload():
return data
with vcr.use_cassette('fixtures/vcr_cassettes/flickr.yaml') as cass:
testfile = str(tmpdir.join('flickr.yml'))
with vcr.use_cassette(testfile) as cass:
_pretend_to_be_flickr_library()
assert len(cass) == 1
with vcr.use_cassette('fixtures/vcr_cassettes/flickr.yaml') as cass:
with vcr.use_cassette(testfile) as cass:
assert len(cass) == 1
_pretend_to_be_flickr_library()
assert cass.play_count == 1
@@ -61,13 +62,13 @@ def test_flickr_should_respond_with_200(tmpdir):
assert r.status_code == 200
def test_cookies(tmpdir):
def test_cookies(tmpdir, httpbin):
testfile = str(tmpdir.join('cookies.yml'))
with vcr.use_cassette(testfile):
s = requests.Session()
s.get("http://httpbin.org/cookies/set?k1=v1&k2=v2")
s.get(httpbin.url + "/cookies/set?k1=v1&k2=v2")
r2 = s.get("http://httpbin.org/cookies")
r2 = s.get(httpbin.url + "/cookies")
assert len(r2.json()['cookies']) == 2

4
tox.ini
View File

@@ -11,11 +11,11 @@ deps = flake8
[testenv]
commands =
py.test {posargs}
./runtests.sh {posargs}
deps =
mock
pytest
pytest-localserver
pytest-httpbin
PyYAML
requests1: requests==1.2.3
requests27: requests==2.7.0

1
vcr/stubs/httplib2_stubs.py
View File

@@ -41,6 +41,7 @@ class VCRHTTPSConnectionWithTimeout(VCRHTTPSConnection,
'strict',
'timeout',
'source_address',
'ca_certs',
))
unknown_keys = set(kwargs.keys()) - safe_keys
safe_kwargs = kwargs.copy()