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

Use pytest-httpbin

Browse Source 
This will help the test flakiness and speed up test runs.
This commit is contained in:
Kevin McCarthy
2015-04-11 13:07:17 -10:00
parent cfc483a08d
commit 4e36997e1a
20 changed files with 380 additions and 369 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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'})