mirror of
https://github.com/kevin1024/vcrpy.git
synced 2025-12-09 01:03:24 +00:00
refactor tests a bit
This commit is contained in:
7
tests/assertions.py
Normal file
7
tests/assertions.py
Normal file
@@ -0,0 +1,7 @@
|
|||||||
|
def assert_cassette_empty(cass):
|
||||||
|
assert len(cass) == 0
|
||||||
|
assert cass.play_count == 0
|
||||||
|
|
||||||
|
def assert_cassette_has_one_response(cass):
|
||||||
|
assert len(cass) == 1
|
||||||
|
assert cass.play_count == 1
|
||||||
@@ -9,6 +9,8 @@ import pytest
|
|||||||
|
|
||||||
import vcr
|
import vcr
|
||||||
|
|
||||||
|
from assertions import assert_cassette_empty, assert_cassette_has_one_response
|
||||||
|
|
||||||
requests = pytest.importorskip("requests")
|
requests = pytest.importorskip("requests")
|
||||||
|
|
||||||
@pytest.fixture(params=["https","http"])
|
@pytest.fixture(params=["https","http"])
|
||||||
@@ -24,36 +26,30 @@ def test_status_code(scheme, tmpdir):
|
|||||||
url = scheme + '://httpbin.org/'
|
url = scheme + '://httpbin.org/'
|
||||||
with vcr.use_cassette(str(tmpdir.join('atts.yaml'))) as cass:
|
with vcr.use_cassette(str(tmpdir.join('atts.yaml'))) as cass:
|
||||||
# Ensure that this is empty to begin with
|
# Ensure that this is empty to begin with
|
||||||
assert len(cass) == 0
|
assert_cassette_empty(cass)
|
||||||
assert cass.play_count == 0
|
|
||||||
assert requests.get(url).status_code == requests.get(url).status_code
|
assert requests.get(url).status_code == requests.get(url).status_code
|
||||||
# Ensure that we've now cached a single response
|
# Ensure that we've now cached a single response
|
||||||
assert len(cass) == 1
|
assert_cassette_has_one_response(cass)
|
||||||
assert cass.play_count == 1
|
|
||||||
|
|
||||||
def test_headers(scheme, tmpdir):
|
def test_headers(scheme, tmpdir):
|
||||||
'''Ensure that we can read the headers back'''
|
'''Ensure that we can read the headers back'''
|
||||||
url = scheme + '://httpbin.org/'
|
url = scheme + '://httpbin.org/'
|
||||||
with vcr.use_cassette(str(tmpdir.join('headers.yaml'))) as cass:
|
with vcr.use_cassette(str(tmpdir.join('headers.yaml'))) as cass:
|
||||||
# Ensure that this is empty to begin with
|
# Ensure that this is empty to begin with
|
||||||
assert len(cass) == 0
|
assert_cassette_empty(cass)
|
||||||
assert cass.play_count == 0
|
|
||||||
assert requests.get(url).headers == requests.get(url).headers
|
assert requests.get(url).headers == requests.get(url).headers
|
||||||
# Ensure that we've now cached a single response
|
# Ensure that we've now cached a single response
|
||||||
assert len(cass) == 1
|
assert_cassette_has_one_response(cass)
|
||||||
assert cass.play_count == 1
|
|
||||||
|
|
||||||
def test_body(tmpdir, scheme):
|
def test_body(tmpdir, scheme):
|
||||||
'''Ensure the responses are all identical enough'''
|
'''Ensure the responses are all identical enough'''
|
||||||
url = scheme + '://httpbin.org/bytes/1024'
|
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'))) as cass:
|
||||||
# Ensure that this is empty to begin with
|
# Ensure that this is empty to begin with
|
||||||
assert len(cass) == 0
|
assert_cassette_empty(cass)
|
||||||
assert cass.play_count == 0
|
|
||||||
assert requests.get(url).content == requests.get(url).content
|
assert requests.get(url).content == requests.get(url).content
|
||||||
# Ensure that we've now cached a single response
|
# Ensure that we've now cached a single response
|
||||||
assert len(cass) == 1
|
assert_cassette_has_one_response(cass)
|
||||||
assert cass.play_count == 1
|
|
||||||
|
|
||||||
def test_auth(tmpdir, scheme):
|
def test_auth(tmpdir, scheme):
|
||||||
'''Ensure that we can handle basic auth'''
|
'''Ensure that we can handle basic auth'''
|
||||||
@@ -61,15 +57,13 @@ def test_auth(tmpdir, scheme):
|
|||||||
url = scheme + '://httpbin.org/basic-auth/user/passwd'
|
url = scheme + '://httpbin.org/basic-auth/user/passwd'
|
||||||
with vcr.use_cassette(str(tmpdir.join('auth.yaml'))) as cass:
|
with vcr.use_cassette(str(tmpdir.join('auth.yaml'))) as cass:
|
||||||
# Ensure that this is empty to begin with
|
# Ensure that this is empty to begin with
|
||||||
assert len(cass) == 0
|
assert_cassette_empty(cass)
|
||||||
assert cass.play_count == 0
|
|
||||||
one = requests.get(url, auth=auth)
|
one = requests.get(url, auth=auth)
|
||||||
two = requests.get(url, auth=auth)
|
two = requests.get(url, auth=auth)
|
||||||
assert one.content == two.content
|
assert one.content == two.content
|
||||||
assert one.status_code == two.status_code
|
assert one.status_code == two.status_code
|
||||||
# Ensure that we've now cached a single response
|
# Ensure that we've now cached a single response
|
||||||
assert len(cass) == 1
|
assert_cassette_has_one_response(cass)
|
||||||
assert cass.play_count == 1
|
|
||||||
|
|
||||||
def test_auth_failed(tmpdir, scheme):
|
def test_auth_failed(tmpdir, scheme):
|
||||||
'''Ensure that we can save failed auth statuses'''
|
'''Ensure that we can save failed auth statuses'''
|
||||||
@@ -77,15 +71,13 @@ def test_auth_failed(tmpdir, scheme):
|
|||||||
url = scheme + '://httpbin.org/basic-auth/user/passwd'
|
url = scheme + '://httpbin.org/basic-auth/user/passwd'
|
||||||
with vcr.use_cassette(str(tmpdir.join('auth-failed.yaml'))) as cass:
|
with vcr.use_cassette(str(tmpdir.join('auth-failed.yaml'))) as cass:
|
||||||
# Ensure that this is empty to begin with
|
# Ensure that this is empty to begin with
|
||||||
assert len(cass) == 0
|
assert_cassette_empty(cass)
|
||||||
assert cass.play_count == 0
|
|
||||||
one = requests.get(url, auth=auth)
|
one = requests.get(url, auth=auth)
|
||||||
two = requests.get(url, auth=auth)
|
two = requests.get(url, auth=auth)
|
||||||
assert one.content == two.content
|
assert one.content == two.content
|
||||||
assert one.status_code == two.status_code == 401
|
assert one.status_code == two.status_code == 401
|
||||||
# Ensure that we've now cached a single response
|
# Ensure that we've now cached a single response
|
||||||
assert len(cass) == 1
|
assert_cassette_has_one_response(cass)
|
||||||
assert cass.play_count == 1
|
|
||||||
|
|
||||||
def test_post(tmpdir, scheme):
|
def test_post(tmpdir, scheme):
|
||||||
'''Ensure that we can post and cache the results'''
|
'''Ensure that we can post and cache the results'''
|
||||||
@@ -93,22 +85,19 @@ def test_post(tmpdir, scheme):
|
|||||||
url = scheme + '://httpbin.org/post'
|
url = scheme + '://httpbin.org/post'
|
||||||
with vcr.use_cassette(str(tmpdir.join('redirect.yaml'))) as cass:
|
with vcr.use_cassette(str(tmpdir.join('redirect.yaml'))) as cass:
|
||||||
# Ensure that this is empty to begin with
|
# Ensure that this is empty to begin with
|
||||||
assert len(cass) == 0
|
assert_cassette_empty(cass)
|
||||||
assert cass.play_count == 0
|
|
||||||
assert requests.post(url, data).content == requests.post(url, data).content
|
assert requests.post(url, data).content == requests.post(url, data).content
|
||||||
# Ensure that we've now cached a single response
|
# Ensure that we've now cached a single response
|
||||||
assert len(cass) == 1
|
assert_cassette_has_one_response(cass)
|
||||||
assert cass.play_count == 1
|
|
||||||
|
|
||||||
def test_redirects(tmpdir, scheme):
|
def test_redirects(tmpdir, scheme):
|
||||||
'''Ensure that we can handle redirects'''
|
'''Ensure that we can handle redirects'''
|
||||||
url = scheme + '://httpbin.org/redirect-to?url=bytes/1024'
|
url = scheme + '://httpbin.org/redirect-to?url=bytes/1024'
|
||||||
with vcr.use_cassette(str(tmpdir.join('redirect.yaml'))) as cass:
|
with vcr.use_cassette(str(tmpdir.join('redirect.yaml'))) as cass:
|
||||||
# Ensure that this is empty to begin with
|
# Ensure that this is empty to begin with
|
||||||
assert len(cass) == 0
|
assert_cassette_empty(cass)
|
||||||
assert cass.play_count == 0
|
|
||||||
assert requests.get(url).content == requests.get(url).content
|
assert requests.get(url).content == requests.get(url).content
|
||||||
# Ensure that we've now cached /two/ responses. One for the redirect
|
# Ensure that we've now cached *two* responses. One for the redirect
|
||||||
# and one for the final fetch
|
# and one for the final fetch
|
||||||
assert len(cass) == 2
|
assert len(cass) == 2
|
||||||
assert cass.play_count == 2
|
assert cass.play_count == 2
|
||||||
@@ -121,6 +110,6 @@ def test_cross_scheme(tmpdir, scheme):
|
|||||||
with vcr.use_cassette(str(tmpdir.join('cross_scheme.yaml'))) as cass:
|
with vcr.use_cassette(str(tmpdir.join('cross_scheme.yaml'))) as cass:
|
||||||
requests.get('https://httpbin.org/')
|
requests.get('https://httpbin.org/')
|
||||||
requests.get('http://httpbin.org/')
|
requests.get('http://httpbin.org/')
|
||||||
assert len(cass) == 2
|
|
||||||
assert cass.play_count == 0
|
assert cass.play_count == 0
|
||||||
|
assert len(cass) == 2
|
||||||
|
|
||||||
|
|||||||
@@ -10,6 +10,7 @@ import pytest
|
|||||||
# Internal imports
|
# Internal imports
|
||||||
import vcr
|
import vcr
|
||||||
|
|
||||||
|
from assertions import assert_cassette_empty, assert_cassette_has_one_response
|
||||||
|
|
||||||
@pytest.fixture(params=["https","http"])
|
@pytest.fixture(params=["https","http"])
|
||||||
def scheme(request):
|
def scheme(request):
|
||||||
@@ -23,35 +24,30 @@ def test_response_code(scheme, tmpdir):
|
|||||||
url = scheme + '://httpbin.org/'
|
url = scheme + '://httpbin.org/'
|
||||||
with vcr.use_cassette(str(tmpdir.join('atts.yaml'))) as cass:
|
with vcr.use_cassette(str(tmpdir.join('atts.yaml'))) as cass:
|
||||||
# Ensure that this is empty to begin with
|
# Ensure that this is empty to begin with
|
||||||
assert len(cass) == 0
|
assert_cassette_empty(cass)
|
||||||
assert cass.play_count == 0
|
|
||||||
assert urllib2.urlopen(url).getcode() == urllib2.urlopen(url).getcode()
|
assert urllib2.urlopen(url).getcode() == urllib2.urlopen(url).getcode()
|
||||||
# Ensure that we've now cached a single response
|
# Ensure that we've now cached a single response
|
||||||
assert len(cass) == 1
|
assert_cassette_has_one_response(cass)
|
||||||
assert cass.play_count == 1
|
|
||||||
|
|
||||||
def test_random_body(scheme, tmpdir):
|
def test_random_body(scheme, tmpdir):
|
||||||
'''Ensure we can read the content, and that it's served from cache'''
|
'''Ensure we can read the content, and that it's served from cache'''
|
||||||
url = scheme + '://httpbin.org/bytes/1024'
|
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'))) as cass:
|
||||||
# Ensure that this is empty to begin with
|
# Ensure that this is empty to begin with
|
||||||
assert len(cass) == 0
|
assert_cassette_empty(cass)
|
||||||
assert cass.play_count == 0
|
|
||||||
assert urllib2.urlopen(url).read() == urllib2.urlopen(url).read()
|
assert urllib2.urlopen(url).read() == urllib2.urlopen(url).read()
|
||||||
# Ensure that we've now cached a single response
|
# Ensure that we've now cached a single response
|
||||||
assert len(cass) == 1
|
assert_cassette_has_one_response(cass)
|
||||||
assert cass.play_count == 1
|
|
||||||
|
|
||||||
def test_response_headers(scheme, tmpdir):
|
def test_response_headers(scheme, tmpdir):
|
||||||
'''Ensure we can get information from the response'''
|
'''Ensure we can get information from the response'''
|
||||||
url = scheme + '://httpbin.org/'
|
url = scheme + '://httpbin.org/'
|
||||||
with vcr.use_cassette(str(tmpdir.join('headers.yaml'))) as cass:
|
with vcr.use_cassette(str(tmpdir.join('headers.yaml'))) as cass:
|
||||||
# Ensure that this is empty to begin with
|
# Ensure that this is empty to begin with
|
||||||
assert len(cass) == 0
|
assert_cassette_empty(cass)
|
||||||
assert cass.play_count == 0
|
|
||||||
assert urllib2.urlopen(url).info().items() == urllib2.urlopen(url).info().items()
|
assert urllib2.urlopen(url).info().items() == urllib2.urlopen(url).info().items()
|
||||||
# Ensure that we've now cached a single response
|
# Ensure that we've now cached a single response
|
||||||
assert len(cass) == 1
|
assert_cassette_has_one_response(cass)
|
||||||
|
|
||||||
def test_multiple_requests(scheme, tmpdir):
|
def test_multiple_requests(scheme, tmpdir):
|
||||||
'''Ensure that we can cache multiple requests'''
|
'''Ensure that we can cache multiple requests'''
|
||||||
@@ -75,8 +71,7 @@ def test_get_data(scheme, tmpdir):
|
|||||||
url = scheme + '://httpbin.org/get?' + data
|
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'))) as cass:
|
||||||
# Ensure that this is empty to begin with
|
# Ensure that this is empty to begin with
|
||||||
assert len(cass) == 0
|
assert_cassette_empty(cass)
|
||||||
assert cass.play_count == 0
|
|
||||||
res1 = urllib2.urlopen(url).read()
|
res1 = urllib2.urlopen(url).read()
|
||||||
res2 = urllib2.urlopen(url).read()
|
res2 = urllib2.urlopen(url).read()
|
||||||
assert res1 == res2
|
assert res1 == res2
|
||||||
@@ -90,14 +85,12 @@ def test_post_data(scheme, tmpdir):
|
|||||||
url = scheme + '://httpbin.org/post'
|
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'))) as cass:
|
||||||
# Ensure that this is empty to begin with
|
# Ensure that this is empty to begin with
|
||||||
assert len(cass) == 0
|
assert_cassette_empty(cass)
|
||||||
assert cass.play_count == 0
|
|
||||||
res1 = urllib2.urlopen(url, data).read()
|
res1 = urllib2.urlopen(url, data).read()
|
||||||
res2 = urllib2.urlopen(url, data).read()
|
res2 = urllib2.urlopen(url, data).read()
|
||||||
assert res1 == res2
|
assert res1 == res2
|
||||||
# Ensure that we've now cached a single response
|
# Ensure that we've now cached a single response
|
||||||
assert len(cass) == 1
|
assert_cassette_has_one_response(cass)
|
||||||
assert cass.play_count == 1
|
|
||||||
|
|
||||||
def test_post_unicode_data(scheme, tmpdir):
|
def test_post_unicode_data(scheme, tmpdir):
|
||||||
'''Ensure that it works when posting unicode data'''
|
'''Ensure that it works when posting unicode data'''
|
||||||
@@ -105,14 +98,12 @@ def test_post_unicode_data(scheme, tmpdir):
|
|||||||
url = scheme + '://httpbin.org/post'
|
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'))) as cass:
|
||||||
# Ensure that this is empty to begin with
|
# Ensure that this is empty to begin with
|
||||||
assert len(cass) == 0
|
assert_cassette_empty(cass)
|
||||||
assert cass.play_count == 0
|
|
||||||
res1 = urllib2.urlopen(url, data).read()
|
res1 = urllib2.urlopen(url, data).read()
|
||||||
res2 = urllib2.urlopen(url, data).read()
|
res2 = urllib2.urlopen(url, data).read()
|
||||||
assert res1 == res2
|
assert res1 == res2
|
||||||
# Ensure that we've now cached a single response
|
# Ensure that we've now cached a single response
|
||||||
assert len(cass) == 1
|
assert_cassette_has_one_response(cass)
|
||||||
assert cass.play_count == 1
|
|
||||||
|
|
||||||
def test_cross_scheme(tmpdir):
|
def test_cross_scheme(tmpdir):
|
||||||
'''Ensure that requests between schemes are treated separately'''
|
'''Ensure that requests between schemes are treated separately'''
|
||||||
|
|||||||
Reference in New Issue
Block a user