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

Format project with black (#467)

Browse Source 
Format with line length 110 to match flake8

make black part of linting check

Update travis spec for updated black requirements

Add diff output for black on failure

update changelog
This commit is contained in:
Josh Peak
2019-08-24 11:36:35 +10:00
committed by GitHub
parent 75969de601
commit 7caf29735a
70 changed files with 2000 additions and 2217 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
tests/integration/aiohttp_utils.py
View File

@@ -5,19 +5,19 @@ import aiohttp
from aiohttp.test_utils import TestClient
async def aiohttp_request(loop, method, url, output='text', encoding='utf-8', content_type=None, **kwargs):
async def aiohttp_request(loop, method, url, output="text", encoding="utf-8", content_type=None, **kwargs):
session = aiohttp.ClientSession(loop=loop)
response_ctx = session.request(method, url, **kwargs)
response = await response_ctx.__aenter__()
if output == 'text':
if output == "text":
content = await response.text()
elif output == 'json':
content_type = content_type or 'application/json'
elif output == "json":
content_type = content_type or "application/json"
content = await response.json(encoding=encoding, content_type=content_type)
elif output == 'raw':
elif output == "raw":
content = await response.read()
elif output == 'stream':
elif output == "stream":
content = await response.content.read()
response_ctx._resp.close()
@@ -28,7 +28,7 @@ async def aiohttp_request(loop, method, url, output='text', encoding='utf-8', co
def aiohttp_app():
async def hello(request):
return aiohttp.web.Response(text='hello')
return aiohttp.web.Response(text="hello")
async def json(request):
return aiohttp.web.json_response({})
@@ -37,7 +37,7 @@ def aiohttp_app():
return aiohttp.web.json_response()
app = aiohttp.web.Application()
app.router.add_get('/', hello)
app.router.add_get('/json', json)
app.router.add_get('/json/empty', json_empty_body)
app.router.add_get("/", hello)
app.router.add_get("/json", json)
app.router.add_get("/json/empty", json_empty_body)
return app

162
tests/integration/test_aiohttp.py
View File

@@ -2,6 +2,7 @@ import contextlib
import logging
import pytest
asyncio = pytest.importorskip("asyncio")
aiohttp = pytest.importorskip("aiohttp")
@@ -16,33 +17,33 @@ def run_in_loop(fn):
return loop.run_until_complete(task)
def request(method, url, output='text', **kwargs):
def request(method, url, output="text", **kwargs):
def run(loop):
return aiohttp_request(loop, method, url, output=output, **kwargs)
return run_in_loop(run)
def get(url, output='text', **kwargs):
return request('GET', url, output=output, **kwargs)
def get(url, output="text", **kwargs):
return request("GET", url, output=output, **kwargs)
def post(url, output='text', **kwargs):
return request('POST', url, output='text', **kwargs)
def post(url, output="text", **kwargs):
return request("POST", url, output="text", **kwargs)
@pytest.fixture(params=["https", "http"])
def scheme(request):
'''Fixture that returns both http and https.'''
"""Fixture that returns both http and https."""
return request.param
def test_status(tmpdir, scheme):
url = scheme + '://httpbin.org'
with vcr.use_cassette(str(tmpdir.join('status.yaml'))):
url = scheme + "://httpbin.org"
with vcr.use_cassette(str(tmpdir.join("status.yaml"))):
response, _ = get(url)
with vcr.use_cassette(str(tmpdir.join('status.yaml'))) as cassette:
with vcr.use_cassette(str(tmpdir.join("status.yaml"))) as cassette:
cassette_response, _ = get(url)
assert cassette_response.status == response.status
assert cassette.play_count == 1
@@ -50,27 +51,27 @@ def test_status(tmpdir, scheme):
@pytest.mark.parametrize("auth", [None, aiohttp.BasicAuth("vcrpy", "test")])
def test_headers(tmpdir, scheme, auth):
url = scheme + '://httpbin.org'
with vcr.use_cassette(str(tmpdir.join('headers.yaml'))):
url = scheme + "://httpbin.org"
with vcr.use_cassette(str(tmpdir.join("headers.yaml"))):
response, _ = get(url, auth=auth)
with vcr.use_cassette(str(tmpdir.join('headers.yaml'))) as cassette:
with vcr.use_cassette(str(tmpdir.join("headers.yaml"))) as cassette:
if auth is not None:
request = cassette.requests[0]
assert "AUTHORIZATION" in request.headers
cassette_response, _ = get(url, auth=auth)
assert dict(cassette_response.headers) == dict(response.headers)
assert cassette.play_count == 1
assert 'istr' not in cassette.data[0]
assert 'yarl.URL' not in cassette.data[0]
assert "istr" not in cassette.data[0]
assert "yarl.URL" not in cassette.data[0]
def test_case_insensitive_headers(tmpdir, scheme):
url = scheme + '://httpbin.org'
with vcr.use_cassette(str(tmpdir.join('whatever.yaml'))):
url = scheme + "://httpbin.org"
with vcr.use_cassette(str(tmpdir.join("whatever.yaml"))):
_, _ = get(url)
with vcr.use_cassette(str(tmpdir.join('whatever.yaml'))) as cassette:
with vcr.use_cassette(str(tmpdir.join("whatever.yaml"))) as cassette:
cassette_response, _ = get(url)
assert "Content-Type" in cassette_response.headers
assert "content-type" in cassette_response.headers
@@ -78,61 +79,61 @@ def test_case_insensitive_headers(tmpdir, scheme):
def test_text(tmpdir, scheme):
url = scheme + '://httpbin.org'
with vcr.use_cassette(str(tmpdir.join('text.yaml'))):
url = scheme + "://httpbin.org"
with vcr.use_cassette(str(tmpdir.join("text.yaml"))):
_, response_text = get(url)
with vcr.use_cassette(str(tmpdir.join('text.yaml'))) as cassette:
with vcr.use_cassette(str(tmpdir.join("text.yaml"))) as cassette:
_, cassette_response_text = get(url)
assert cassette_response_text == response_text
assert cassette.play_count == 1
def test_json(tmpdir, scheme):
url = scheme + '://httpbin.org/get'
headers = {'Content-Type': 'application/json'}
url = scheme + "://httpbin.org/get"
headers = {"Content-Type": "application/json"}
with vcr.use_cassette(str(tmpdir.join('json.yaml'))):
_, response_json = get(url, output='json', headers=headers)
with vcr.use_cassette(str(tmpdir.join("json.yaml"))):
_, response_json = get(url, output="json", headers=headers)
with vcr.use_cassette(str(tmpdir.join('json.yaml'))) as cassette:
_, cassette_response_json = get(url, output='json', headers=headers)
with vcr.use_cassette(str(tmpdir.join("json.yaml"))) as cassette:
_, cassette_response_json = get(url, output="json", headers=headers)
assert cassette_response_json == response_json
assert cassette.play_count == 1
def test_binary(tmpdir, scheme):
url = scheme + '://httpbin.org/image/png'
with vcr.use_cassette(str(tmpdir.join('binary.yaml'))):
_, response_binary = get(url, output='raw')
url = scheme + "://httpbin.org/image/png"
with vcr.use_cassette(str(tmpdir.join("binary.yaml"))):
_, response_binary = get(url, output="raw")
with vcr.use_cassette(str(tmpdir.join('binary.yaml'))) as cassette:
_, cassette_response_binary = get(url, output='raw')
with vcr.use_cassette(str(tmpdir.join("binary.yaml"))) as cassette:
_, cassette_response_binary = get(url, output="raw")
assert cassette_response_binary == response_binary
assert cassette.play_count == 1
def test_stream(tmpdir, scheme):
url = scheme + '://httpbin.org/get'
url = scheme + "://httpbin.org/get"
with vcr.use_cassette(str(tmpdir.join('stream.yaml'))):
resp, body = get(url, output='raw') # Do not use stream here, as the stream is exhausted by vcr
with vcr.use_cassette(str(tmpdir.join("stream.yaml"))):
resp, body = get(url, output="raw") # Do not use stream here, as the stream is exhausted by vcr
with vcr.use_cassette(str(tmpdir.join('stream.yaml'))) as cassette:
cassette_resp, cassette_body = get(url, output='stream')
with vcr.use_cassette(str(tmpdir.join("stream.yaml"))) as cassette:
cassette_resp, cassette_body = get(url, output="stream")
assert cassette_body == body
assert cassette.play_count == 1
@pytest.mark.parametrize('body', ['data', 'json'])
@pytest.mark.parametrize("body", ["data", "json"])
def test_post(tmpdir, scheme, body, caplog):
caplog.set_level(logging.INFO)
data = {'key1': 'value1', 'key2': 'value2'}
url = scheme + '://httpbin.org/post'
with vcr.use_cassette(str(tmpdir.join('post.yaml'))):
data = {"key1": "value1", "key2": "value2"}
url = scheme + "://httpbin.org/post"
with vcr.use_cassette(str(tmpdir.join("post.yaml"))):
_, response_json = post(url, **{body: data})
with vcr.use_cassette(str(tmpdir.join('post.yaml'))) as cassette:
with vcr.use_cassette(str(tmpdir.join("post.yaml"))) as cassette:
request = cassette.requests[0]
assert request.body == data
_, cassette_response_json = post(url, **{body: data})
@@ -143,58 +144,57 @@ def test_post(tmpdir, scheme, body, caplog):
(
log
for log in caplog.records
if log.getMessage()
== '<Request (POST) {}> not in cassette, sending to real server'.format(url)
if log.getMessage() == "<Request (POST) {}> not in cassette, sending to real server".format(url)
),
None,
), 'Log message not found.'
), "Log message not found."
def test_params(tmpdir, scheme):
url = scheme + '://httpbin.org/get'
headers = {'Content-Type': 'application/json'}
params = {'a': 1, 'b': False, 'c': 'c'}
url = scheme + "://httpbin.org/get"
headers = {"Content-Type": "application/json"}
params = {"a": 1, "b": False, "c": "c"}
with vcr.use_cassette(str(tmpdir.join('get.yaml'))) as cassette:
_, response_json = get(url, output='json', params=params, headers=headers)
with vcr.use_cassette(str(tmpdir.join("get.yaml"))) as cassette:
_, response_json = get(url, output="json", params=params, headers=headers)
with vcr.use_cassette(str(tmpdir.join('get.yaml'))) as cassette:
_, cassette_response_json = get(url, output='json', params=params, headers=headers)
with vcr.use_cassette(str(tmpdir.join("get.yaml"))) as cassette:
_, cassette_response_json = get(url, output="json", params=params, headers=headers)
assert cassette_response_json == response_json
assert cassette.play_count == 1
def test_params_same_url_distinct_params(tmpdir, scheme):
url = scheme + '://httpbin.org/get'
headers = {'Content-Type': 'application/json'}
params = {'a': 1, 'b': False, 'c': 'c'}
url = scheme + "://httpbin.org/get"
headers = {"Content-Type": "application/json"}
params = {"a": 1, "b": False, "c": "c"}
with vcr.use_cassette(str(tmpdir.join('get.yaml'))) as cassette:
_, response_json = get(url, output='json', params=params, headers=headers)
with vcr.use_cassette(str(tmpdir.join("get.yaml"))) as cassette:
_, response_json = get(url, output="json", params=params, headers=headers)
with vcr.use_cassette(str(tmpdir.join('get.yaml'))) as cassette:
_, cassette_response_json = get(url, output='json', params=params, headers=headers)
with vcr.use_cassette(str(tmpdir.join("get.yaml"))) as cassette:
_, cassette_response_json = get(url, output="json", params=params, headers=headers)
assert cassette_response_json == response_json
assert cassette.play_count == 1
other_params = {'other': 'params'}
with vcr.use_cassette(str(tmpdir.join('get.yaml'))) as cassette:
response, cassette_response_text = get(url, output='text', params=other_params)
assert 'No match for the request' in cassette_response_text
other_params = {"other": "params"}
with vcr.use_cassette(str(tmpdir.join("get.yaml"))) as cassette:
response, cassette_response_text = get(url, output="text", params=other_params)
assert "No match for the request" in cassette_response_text
assert response.status == 599
def test_params_on_url(tmpdir, scheme):
url = scheme + '://httpbin.org/get?a=1&b=foo'
headers = {'Content-Type': 'application/json'}
url = scheme + "://httpbin.org/get?a=1&b=foo"
headers = {"Content-Type": "application/json"}
with vcr.use_cassette(str(tmpdir.join('get.yaml'))) as cassette:
_, response_json = get(url, output='json', headers=headers)
with vcr.use_cassette(str(tmpdir.join("get.yaml"))) as cassette:
_, response_json = get(url, output="json", headers=headers)
request = cassette.requests[0]
assert request.url == url
with vcr.use_cassette(str(tmpdir.join('get.yaml'))) as cassette:
_, cassette_response_json = get(url, output='json', headers=headers)
with vcr.use_cassette(str(tmpdir.join("get.yaml"))) as cassette:
_, cassette_response_json = get(url, output="json", headers=headers)
request = cassette.requests[0]
assert request.url == url
assert cassette_response_json == response_json
@@ -204,42 +204,42 @@ def test_params_on_url(tmpdir, scheme):
def test_aiohttp_test_client(aiohttp_client, tmpdir):
loop = asyncio.get_event_loop()
app = aiohttp_app()
url = '/'
url = "/"
client = loop.run_until_complete(aiohttp_client(app))
with vcr.use_cassette(str(tmpdir.join('get.yaml'))):
with vcr.use_cassette(str(tmpdir.join("get.yaml"))):
response = loop.run_until_complete(client.get(url))
assert response.status == 200
response_text = loop.run_until_complete(response.text())
assert response_text == 'hello'
response_text = loop.run_until_complete(response.text(errors='replace'))
assert response_text == 'hello'
assert response_text == "hello"
response_text = loop.run_until_complete(response.text(errors="replace"))
assert response_text == "hello"
with vcr.use_cassette(str(tmpdir.join('get.yaml'))) as cassette:
with vcr.use_cassette(str(tmpdir.join("get.yaml"))) as cassette:
response = loop.run_until_complete(client.get(url))
request = cassette.requests[0]
assert request.url == str(client.make_url(url))
response_text = loop.run_until_complete(response.text())
assert response_text == 'hello'
assert response_text == "hello"
assert cassette.play_count == 1
def test_aiohttp_test_client_json(aiohttp_client, tmpdir):
loop = asyncio.get_event_loop()
app = aiohttp_app()
url = '/json/empty'
url = "/json/empty"
client = loop.run_until_complete(aiohttp_client(app))
with vcr.use_cassette(str(tmpdir.join('get.yaml'))):
with vcr.use_cassette(str(tmpdir.join("get.yaml"))):
response = loop.run_until_complete(client.get(url))
assert response.status == 200
response_json = loop.run_until_complete(response.json())
assert response_json is None
with vcr.use_cassette(str(tmpdir.join('get.yaml'))) as cassette:
with vcr.use_cassette(str(tmpdir.join("get.yaml"))) as cassette:
response = loop.run_until_complete(client.get(url))
request = cassette.requests[0]
@@ -250,12 +250,12 @@ def test_aiohttp_test_client_json(aiohttp_client, tmpdir):
def test_redirect(aiohttp_client, tmpdir):
url = 'https://httpbin.org/redirect/2'
url = "https://httpbin.org/redirect/2"
with vcr.use_cassette(str(tmpdir.join('redirect.yaml'))):
with vcr.use_cassette(str(tmpdir.join("redirect.yaml"))):
response, _ = get(url)
with vcr.use_cassette(str(tmpdir.join('redirect.yaml'))) as cassette:
with vcr.use_cassette(str(tmpdir.join("redirect.yaml"))) as cassette:
cassette_response, _ = get(url)
assert cassette_response.status == response.status

40
tests/integration/test_basic.py
View File

@@ -1,5 +1,5 @@
# -*- coding: utf-8 -*-
'''Basic tests for cassettes'''
"""Basic tests for cassettes"""
# External imports
import os
@@ -10,21 +10,21 @@ import vcr
def test_nonexistent_directory(tmpdir, httpbin):
'''If we load a cassette in a nonexistent directory, it can save ok'''
"""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')))
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'))):
with vcr.use_cassette(str(tmpdir.join("nonexistent", "cassette.yml"))):
urlopen(httpbin.url).read()
# This should have made the file and the directory
assert os.path.exists(str(tmpdir.join('nonexistent', 'cassette.yml')))
assert os.path.exists(str(tmpdir.join("nonexistent", "cassette.yml")))
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:
"""Ensure that our cassette gets unpatched when we're done"""
with vcr.use_cassette(str(tmpdir.join("unpatch.yaml"))) as cass:
urlopen(httpbin.url).read()
# Make the same request, and assert that we haven't served any more
@@ -34,30 +34,30 @@ def test_unpatch(tmpdir, httpbin):
def test_basic_json_use(tmpdir, httpbin):
'''
"""
Ensure you can load a json serialized cassette
'''
test_fixture = str(tmpdir.join('synopsis.json'))
with vcr.use_cassette(test_fixture, serializer='json'):
"""
test_fixture = str(tmpdir.join("synopsis.json"))
with vcr.use_cassette(test_fixture, serializer="json"):
response = urlopen(httpbin.url).read()
assert b'difficult sometimes' in response
assert b"difficult sometimes" in response
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:
"""
with vcr.use_cassette(str(tmpdir.join("synopsis.yaml"))) as cass:
response = urlopen(httpbin.url).read()
assert cass.play_count == 0
with vcr.use_cassette(str(tmpdir.join('synopsis.yaml'))) as cass:
with vcr.use_cassette(str(tmpdir.join("synopsis.yaml"))) as cass:
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:
with vcr.use_cassette(str(tmpdir.join("synopsis.yaml"))) as cass:
response3 = urlopen(httpbin.url).read()
assert cass.play_count == 1
@@ -66,12 +66,12 @@ def test_patched_content(tmpdir, httpbin):
def test_patched_content_json(tmpdir, httpbin):
'''
"""
Ensure that what you pull from a json cassette is what came from the
request
'''
"""
testfile = str(tmpdir.join('synopsis.json'))
testfile = str(tmpdir.join("synopsis.json"))
with vcr.use_cassette(testfile) as cass:
response = urlopen(httpbin.url).read()

57
tests/integration/test_boto.py
View File

@@ -1,4 +1,5 @@
import pytest
boto = pytest.importorskip("boto")
import boto # NOQA
@@ -6,6 +7,7 @@ import boto.iam # NOQA
from boto.s3.connection import S3Connection # NOQA
from boto.s3.key import Key # NOQA
import vcr # NOQA
try: # NOQA
from ConfigParser import DuplicateSectionError # NOQA
except ImportError: # NOQA
@@ -14,68 +16,69 @@ except ImportError: # NOQA
def test_boto_stubs(tmpdir):
with vcr.use_cassette(str(tmpdir.join('boto-stubs.yml'))):
with vcr.use_cassette(str(tmpdir.join("boto-stubs.yml"))):
# Perform the imports within the patched context so that
# CertValidatingHTTPSConnection refers to the patched version.
from boto.https_connection import CertValidatingHTTPSConnection
from vcr.stubs.boto_stubs import VCRCertValidatingHTTPSConnection
# Prove that the class was patched by the stub and that we can instantiate it.
assert issubclass(CertValidatingHTTPSConnection, VCRCertValidatingHTTPSConnection)
CertValidatingHTTPSConnection('hostname.does.not.matter')
CertValidatingHTTPSConnection("hostname.does.not.matter")
def test_boto_without_vcr():
s3_conn = S3Connection()
s3_bucket = s3_conn.get_bucket('boto-demo-1394171994') # a bucket you can access
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')
k.key = "test.txt"
k.set_contents_from_string("hello world i am a string")
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'))):
s3_bucket = s3_conn.get_bucket("boto-demo-1394171994") # a bucket you can access
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')
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'))):
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')
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'))):
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
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')
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'))):
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
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')
k.key = "test.txt"
k.set_contents_from_string("hello world i am a string")
def test_boto_iam(tmpdir):
try:
boto.config.add_section('Boto')
boto.config.add_section("Boto")
except DuplicateSectionError:
pass
# Ensure that boto uses HTTPS
boto.config.set('Boto', 'is_secure', 'true')
boto.config.set("Boto", "is_secure", "true")
# Ensure that boto uses CertValidatingHTTPSConnection
boto.config.set('Boto', 'https_validate_certificates', 'true')
boto.config.set("Boto", "https_validate_certificates", "true")
with vcr.use_cassette(str(tmpdir.join('boto-iam.yml'))):
iam_conn = boto.iam.connect_to_region('universal')
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'))):
iam_conn = boto.iam.connect_to_region('universal')
with vcr.use_cassette(str(tmpdir.join("boto-iam.yml"))):
iam_conn = boto.iam.connect_to_region("universal")
iam_conn.get_all_users()

63
tests/integration/test_boto3.py
View File

@@ -18,13 +18,13 @@ except ImportError:
# https://github.com/boto/botocore/pull/1495
boto3_skip_vendored_requests = pytest.mark.skipif(
botocore_awsrequest,
reason='botocore version {ver} does not use vendored requests anymore.'.format(
ver=botocore.__version__))
reason="botocore version {ver} does not use vendored requests anymore.".format(ver=botocore.__version__),
)
boto3_skip_awsrequest = pytest.mark.skipif(
not botocore_awsrequest,
reason='botocore version {ver} still uses vendored requests.'.format(
ver=botocore.__version__))
reason="botocore version {ver} still uses vendored requests.".format(ver=botocore.__version__),
)
IAM_USER_NAME = "vcrpy"
@@ -34,12 +34,13 @@ def iam_client():
def _iam_client(boto3_session=None):
if boto3_session is None:
boto3_session = boto3.Session(
aws_access_key_id=os.environ.get('AWS_ACCESS_KEY_ID', "default"),
aws_secret_access_key=os.environ.get('AWS_SECRET_ACCESS_KEY', "default"),
aws_access_key_id=os.environ.get("AWS_ACCESS_KEY_ID", "default"),
aws_secret_access_key=os.environ.get("AWS_SECRET_ACCESS_KEY", "default"),
aws_session_token=None,
region_name=os.environ.get('AWS_DEFAULT_REGION', "default"),
region_name=os.environ.get("AWS_DEFAULT_REGION", "default"),
)
return boto3_session.client('iam')
return boto3_session.client("iam")
return _iam_client
@@ -50,39 +51,43 @@ def get_user(iam_client):
# Default client set with fixture `iam_client`
client = iam_client()
return client.get_user(UserName=user_name)
return _get_user
@boto3_skip_vendored_requests
def test_boto_vendored_stubs(tmpdir):
with vcr.use_cassette(str(tmpdir.join('boto3-stubs.yml'))):
with vcr.use_cassette(str(tmpdir.join("boto3-stubs.yml"))):
# Perform the imports within the patched context so that
# HTTPConnection, VerifiedHTTPSConnection refers to the patched version.
from botocore.vendored.requests.packages.urllib3.connectionpool import \
HTTPConnection, VerifiedHTTPSConnection
from botocore.vendored.requests.packages.urllib3.connectionpool import (
HTTPConnection,
VerifiedHTTPSConnection,
)
from vcr.stubs.boto3_stubs import VCRRequestsHTTPConnection, VCRRequestsHTTPSConnection
# Prove that the class was patched by the stub and that we can instantiate it.
assert issubclass(HTTPConnection, VCRRequestsHTTPConnection)
assert issubclass(VerifiedHTTPSConnection, VCRRequestsHTTPSConnection)
HTTPConnection('hostname.does.not.matter')
VerifiedHTTPSConnection('hostname.does.not.matter')
HTTPConnection("hostname.does.not.matter")
VerifiedHTTPSConnection("hostname.does.not.matter")
@pytest.mark.skipif(
os.environ.get("TRAVIS_PULL_REQUEST") != "false",
reason="Encrypted Environment Variables from Travis Repository Settings"
" are disabled on PRs from forks. "
"https://docs.travis-ci.com/user/pull-requests/#pull-requests-and-security-restrictions"
"https://docs.travis-ci.com/user/pull-requests/#pull-requests-and-security-restrictions",
)
def test_boto_medium_difficulty(tmpdir, get_user):
with vcr.use_cassette(str(tmpdir.join('boto3-medium.yml'))):
with vcr.use_cassette(str(tmpdir.join("boto3-medium.yml"))):
response = get_user()
assert response['User']['UserName'] == IAM_USER_NAME
assert response["User"]["UserName"] == IAM_USER_NAME
with vcr.use_cassette(str(tmpdir.join('boto3-medium.yml'))) as cass:
with vcr.use_cassette(str(tmpdir.join("boto3-medium.yml"))) as cass:
response = get_user()
assert response['User']['UserName'] == IAM_USER_NAME
assert response["User"]["UserName"] == IAM_USER_NAME
assert cass.all_played
@@ -90,28 +95,28 @@ def test_boto_medium_difficulty(tmpdir, get_user):
os.environ.get("TRAVIS_PULL_REQUEST") != "false",
reason="Encrypted Environment Variables from Travis Repository Settings"
" are disabled on PRs from forks. "
"https://docs.travis-ci.com/user/pull-requests/#pull-requests-and-security-restrictions"
"https://docs.travis-ci.com/user/pull-requests/#pull-requests-and-security-restrictions",
)
def test_boto_hardcore_mode(tmpdir, iam_client, get_user):
with vcr.use_cassette(str(tmpdir.join('boto3-hardcore.yml'))):
with vcr.use_cassette(str(tmpdir.join("boto3-hardcore.yml"))):
ses = boto3.Session(
aws_access_key_id=os.environ.get('AWS_ACCESS_KEY_ID'),
aws_secret_access_key=os.environ.get('AWS_SECRET_ACCESS_KEY'),
region_name=os.environ.get('AWS_DEFAULT_REGION'),
aws_access_key_id=os.environ.get("AWS_ACCESS_KEY_ID"),
aws_secret_access_key=os.environ.get("AWS_SECRET_ACCESS_KEY"),
region_name=os.environ.get("AWS_DEFAULT_REGION"),
)
client = iam_client(ses)
response = get_user(client=client)
assert response['User']['UserName'] == IAM_USER_NAME
assert response["User"]["UserName"] == IAM_USER_NAME
with vcr.use_cassette(str(tmpdir.join('boto3-hardcore.yml'))) as cass:
with vcr.use_cassette(str(tmpdir.join("boto3-hardcore.yml"))) as cass:
ses = boto3.Session(
aws_access_key_id=os.environ.get('AWS_ACCESS_KEY_ID'),
aws_secret_access_key=os.environ.get('AWS_SECRET_ACCESS_KEY'),
aws_access_key_id=os.environ.get("AWS_ACCESS_KEY_ID"),
aws_secret_access_key=os.environ.get("AWS_SECRET_ACCESS_KEY"),
aws_session_token=None,
region_name=os.environ.get('AWS_DEFAULT_REGION'),
region_name=os.environ.get("AWS_DEFAULT_REGION"),
)
client = iam_client(ses)
response = get_user(client=client)
assert response['User']['UserName'] == IAM_USER_NAME
assert response["User"]["UserName"] == IAM_USER_NAME
assert cass.all_played

40
tests/integration/test_config.py
View File

@@ -6,45 +6,45 @@ from six.moves.urllib.request import urlopen
def test_set_serializer_default_config(tmpdir, httpbin):
my_vcr = vcr.VCR(serializer='json')
my_vcr = vcr.VCR(serializer="json")
with my_vcr.use_cassette(str(tmpdir.join('test.json'))):
assert my_vcr.serializer == 'json'
urlopen(httpbin.url + '/get')
with my_vcr.use_cassette(str(tmpdir.join("test.json"))):
assert my_vcr.serializer == "json"
urlopen(httpbin.url + "/get")
with open(str(tmpdir.join('test.json'))) as f:
with open(str(tmpdir.join("test.json"))) as f:
assert json.loads(f.read())
def test_default_set_cassette_library_dir(tmpdir, httpbin):
my_vcr = vcr.VCR(cassette_library_dir=str(tmpdir.join('subdir')))
my_vcr = vcr.VCR(cassette_library_dir=str(tmpdir.join("subdir")))
with my_vcr.use_cassette('test.json'):
urlopen(httpbin.url + '/get')
with my_vcr.use_cassette("test.json"):
urlopen(httpbin.url + "/get")
assert os.path.exists(str(tmpdir.join('subdir').join('test.json')))
assert os.path.exists(str(tmpdir.join("subdir").join("test.json")))
def test_override_set_cassette_library_dir(tmpdir, httpbin):
my_vcr = vcr.VCR(cassette_library_dir=str(tmpdir.join('subdir')))
my_vcr = vcr.VCR(cassette_library_dir=str(tmpdir.join("subdir")))
cld = str(tmpdir.join('subdir2'))
cld = str(tmpdir.join("subdir2"))
with my_vcr.use_cassette('test.json', cassette_library_dir=cld):
urlopen(httpbin.url + '/get')
with my_vcr.use_cassette("test.json", cassette_library_dir=cld):
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')))
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, httpbin):
my_vcr = vcr.VCR(match_on=['method'])
my_vcr = vcr.VCR(match_on=["method"])
with my_vcr.use_cassette(str(tmpdir.join('test.json'))):
with my_vcr.use_cassette(str(tmpdir.join("test.json"))):
urlopen(httpbin.url)
with my_vcr.use_cassette(str(tmpdir.join('test.json'))) as cass:
urlopen(httpbin.url + '/get')
with my_vcr.use_cassette(str(tmpdir.join("test.json"))) as cass:
urlopen(httpbin.url + "/get")
assert len(cass) == 1
assert cass.play_count == 1
@@ -54,5 +54,5 @@ def test_missing_matcher():
my_vcr = vcr.VCR()
my_vcr.register_matcher("awesome", object)
with pytest.raises(KeyError):
with my_vcr.use_cassette("test.yaml", match_on=['notawesome']):
with my_vcr.use_cassette("test.yaml", match_on=["notawesome"]):
pass

18
tests/integration/test_disksaver.py
View File

@@ -1,5 +1,5 @@
# -*- coding: utf-8 -*-
'''Basic tests about save behavior'''
"""Basic tests about save behavior"""
# External imports
import os
@@ -11,11 +11,11 @@ import vcr
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'))
"""
fname = str(tmpdir.join("synopsis.yaml"))
with vcr.use_cassette(fname) as cass:
urlopen(httpbin.url).read()
assert cass.play_count == 0
@@ -31,11 +31,11 @@ def test_disk_saver_nowrite(tmpdir, httpbin):
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'))
"""
fname = str(tmpdir.join("synopsis.yaml"))
with vcr.use_cassette(fname) as cass:
urlopen(httpbin.url).read()
assert cass.play_count == 0
@@ -45,9 +45,9 @@ def test_disk_saver_write(tmpdir, httpbin):
# the mtime doesn't change
time.sleep(1)
with vcr.use_cassette(fname, record_mode='any') as cass:
with vcr.use_cassette(fname, record_mode="any") as cass:
urlopen(httpbin.url).read()
urlopen(httpbin.url + '/get').read()
urlopen(httpbin.url + "/get").read()
assert cass.play_count == 1
assert cass.dirty
last_mod2 = os.path.getmtime(fname)

84
tests/integration/test_filter.py
View File

@@ -10,9 +10,7 @@ from assertions import assert_cassette_has_one_response, assert_is_json
def _request_with_auth(url, username, password):
request = Request(url)
base64string = base64.b64encode(
username.encode('ascii') + b':' + password.encode('ascii')
)
base64string = base64.b64encode(username.encode("ascii") + b":" + password.encode("ascii"))
request.add_header(b"Authorization", b"Basic " + base64string)
return urlopen(request)
@@ -22,84 +20,84 @@ def _find_header(cassette, header):
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'])
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
with my_vcr.use_cassette(cass_file, filter_headers=['authorization']):
with my_vcr.use_cassette(cass_file, filter_headers=["authorization"]):
with pytest.raises(HTTPError):
resp = _request_with_auth(url, 'user', 'wrongpasswd')
resp = _request_with_auth(url, "user", "wrongpasswd")
assert resp.getcode() == 401
resp = _request_with_auth(url, 'user', 'passwd')
resp = _request_with_auth(url, "user", "passwd")
assert resp.getcode() == 200
# make same 2 requests, this time both served from cassette.
with my_vcr.use_cassette(cass_file, filter_headers=['authorization']) as cass:
with my_vcr.use_cassette(cass_file, filter_headers=["authorization"]) as cass:
with pytest.raises(HTTPError):
resp = _request_with_auth(url, 'user', 'wrongpasswd')
resp = _request_with_auth(url, "user", "wrongpasswd")
assert resp.getcode() == 401
resp = _request_with_auth(url, 'user', 'passwd')
resp = _request_with_auth(url, "user", "passwd")
assert resp.getcode() == 200
# authorization header should not have been recorded
assert not _find_header(cass, 'authorization')
assert not _find_header(cass, "authorization")
assert len(cass) == 2
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']):
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)
with vcr.use_cassette(cass_file, filter_query_parameters=['foo']) as cass:
with vcr.use_cassette(cass_file, filter_query_parameters=["foo"]) as cass:
urlopen(url)
assert 'foo' not in cass.requests[0].url
assert "foo" not in cass.requests[0].url
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']):
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"]):
urlopen(url, data)
with vcr.use_cassette(cass_file, filter_post_data_parameters=['id']) as cass:
assert b'id=secret' not in cass.requests[0].body
with vcr.use_cassette(cass_file, filter_post_data_parameters=["id"]) as cass:
assert b"id=secret" not in cass.requests[0].body
def test_filter_json_post_data(tmpdir, httpbin):
data = json.dumps({'id': 'secret', 'foo': 'bar'}).encode('utf-8')
request = Request(httpbin.url + '/post', data=data)
request.add_header('Content-Type', 'application/json')
data = json.dumps({"id": "secret", "foo": "bar"}).encode("utf-8")
request = Request(httpbin.url + "/post", data=data)
request.add_header("Content-Type", "application/json")
cass_file = str(tmpdir.join('filter_jpd.yaml'))
with vcr.use_cassette(cass_file, filter_post_data_parameters=['id']):
cass_file = str(tmpdir.join("filter_jpd.yaml"))
with vcr.use_cassette(cass_file, filter_post_data_parameters=["id"]):
urlopen(request)
with vcr.use_cassette(cass_file, filter_post_data_parameters=['id']) as cass:
with vcr.use_cassette(cass_file, filter_post_data_parameters=["id"]) as cass:
assert b'"id": "secret"' not in cass.requests[0].body
def test_filter_callback(tmpdir, httpbin):
url = httpbin.url + '/get'
cass_file = str(tmpdir.join('basic_auth_filter.yaml'))
url = httpbin.url + "/get"
cass_file = str(tmpdir.join("basic_auth_filter.yaml"))
def before_record_cb(request):
if request.path != '/get':
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:
with my_vcr.use_cassette(cass_file, filter_headers=["authorization"]) as cass:
urlopen(url)
assert len(cass) == 0
my_vcr = vcr.VCR(before_record_request=before_record_cb)
with my_vcr.use_cassette(cass_file, filter_headers=['authorization']) as cass:
with my_vcr.use_cassette(cass_file, filter_headers=["authorization"]) as cass:
urlopen(url)
assert len(cass) == 0
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'))
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):
urlopen(request)
with vcr.use_cassette(cass_file) as cass:
@@ -109,9 +107,9 @@ def test_decompress_gzip(tmpdir, httpbin):
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'))
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):
urlopen(request)
with vcr.use_cassette(cass_file) as cass:
@@ -122,8 +120,8 @@ def test_decompress_deflate(tmpdir, httpbin):
def test_decompress_regular(tmpdir, httpbin):
"""Test that it doesn't try to decompress content that isn't compressed"""
url = httpbin.url + '/get'
cass_file = str(tmpdir.join('noncompressed_response.yaml'))
url = httpbin.url + "/get"
cass_file = str(tmpdir.join("noncompressed_response.yaml"))
with vcr.use_cassette(cass_file, decode_compressed_response=True):
urlopen(url)
with vcr.use_cassette(cass_file) as cass:

97
tests/integration/test_httplib2.py
View File

@@ -1,5 +1,5 @@
# -*- coding: utf-8 -*-
'''Integration tests with httplib2'''
"""Integration tests with httplib2"""
import sys
@@ -19,97 +19,90 @@ def http():
Returns an httplib2 HTTP instance
with the certificate replaced by the httpbin one.
"""
kwargs = {
'ca_certs': pytest_httpbin.certs.where()
}
kwargs = {"ca_certs": pytest_httpbin.certs.where()}
if sys.version_info[:2] in [(2, 7), (3, 7)]:
kwargs['disable_ssl_certificate_validation'] = True
kwargs["disable_ssl_certificate_validation"] = True
return httplib2.Http(**kwargs)
def test_response_code(tmpdir, httpbin_both):
'''Ensure we can read a response code from a fetch'''
"""Ensure we can read a response code from a fetch"""
url = httpbin_both.url
with vcr.use_cassette(str(tmpdir.join('atts.yaml'))):
with vcr.use_cassette(str(tmpdir.join("atts.yaml"))):
resp, _ = http().request(url)
code = resp.status
with vcr.use_cassette(str(tmpdir.join('atts.yaml'))):
with vcr.use_cassette(str(tmpdir.join("atts.yaml"))):
resp, _ = http().request(url)
assert code == resp.status
def test_random_body(httpbin_both, tmpdir):
'''Ensure we can read the content, and that it's served from cache'''
url = httpbin_both.url + '/bytes/1024'
with vcr.use_cassette(str(tmpdir.join('body.yaml'))):
"""Ensure we can read the content, and that it's served from cache"""
url = httpbin_both.url + "/bytes/1024"
with vcr.use_cassette(str(tmpdir.join("body.yaml"))):
_, content = http().request(url)
body = content
with vcr.use_cassette(str(tmpdir.join('body.yaml'))):
with vcr.use_cassette(str(tmpdir.join("body.yaml"))):
_, content = http().request(url)
assert body == content
def test_response_headers(tmpdir, httpbin_both):
'''Ensure we can get information from the response'''
"""Ensure we can get information from the response"""
url = httpbin_both.url
with vcr.use_cassette(str(tmpdir.join('headers.yaml'))):
with vcr.use_cassette(str(tmpdir.join("headers.yaml"))):
resp, _ = http().request(url)
headers = resp.items()
with vcr.use_cassette(str(tmpdir.join('headers.yaml'))):
with vcr.use_cassette(str(tmpdir.join("headers.yaml"))):
resp, _ = http().request(url)
assert set(headers) == set(resp.items())
def test_effective_url(tmpdir, httpbin_both):
'''Ensure that the effective_url is captured'''
url = httpbin_both.url + '/redirect-to?url=/html'
with vcr.use_cassette(str(tmpdir.join('headers.yaml'))):
"""Ensure that the effective_url is captured"""
url = httpbin_both.url + "/redirect-to?url=/html"
with vcr.use_cassette(str(tmpdir.join("headers.yaml"))):
resp, _ = http().request(url)
effective_url = resp['content-location']
assert effective_url == httpbin_both + '/html'
effective_url = resp["content-location"]
assert effective_url == httpbin_both + "/html"
with vcr.use_cassette(str(tmpdir.join('headers.yaml'))):
with vcr.use_cassette(str(tmpdir.join("headers.yaml"))):
resp, _ = http().request(url)
assert effective_url == resp['content-location']
assert effective_url == resp["content-location"]
def test_multiple_requests(tmpdir, httpbin_both):
'''Ensure that we can cache multiple requests'''
urls = [
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:
"""Ensure that we can cache multiple requests"""
urls = [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:
[http().request(url) for url in urls]
assert len(cass) == len(urls)
def test_get_data(tmpdir, httpbin_both):
'''Ensure that it works with query data'''
data = urlencode({'some': 1, 'data': 'here'})
url = httpbin_both.url + '/get?' + data
with vcr.use_cassette(str(tmpdir.join('get_data.yaml'))):
"""Ensure that it works with query data"""
data = urlencode({"some": 1, "data": "here"})
url = httpbin_both.url + "/get?" + data
with vcr.use_cassette(str(tmpdir.join("get_data.yaml"))):
_, res1 = http().request(url)
with vcr.use_cassette(str(tmpdir.join('get_data.yaml'))):
with vcr.use_cassette(str(tmpdir.join("get_data.yaml"))):
_, res2 = http().request(url)
assert res1 == res2
def test_post_data(tmpdir, httpbin_both):
'''Ensure that it works when posting data'''
data = urlencode({'some': 1, 'data': 'here'})
url = httpbin_both.url + '/post'
with vcr.use_cassette(str(tmpdir.join('post_data.yaml'))):
"""Ensure that it works when posting data"""
data = urlencode({"some": 1, "data": "here"})
url = httpbin_both.url + "/post"
with vcr.use_cassette(str(tmpdir.join("post_data.yaml"))):
_, res1 = http().request(url, "POST", data)
with vcr.use_cassette(str(tmpdir.join('post_data.yaml'))) as cass:
with vcr.use_cassette(str(tmpdir.join("post_data.yaml"))) as cass:
_, res2 = http().request(url, "POST", data)
assert res1 == res2
@@ -117,13 +110,13 @@ def test_post_data(tmpdir, httpbin_both):
def test_post_unicode_data(tmpdir, httpbin_both):
'''Ensure that it works when posting unicode data'''
data = urlencode({'snowman': u'☃'.encode('utf-8')})
url = httpbin_both.url + '/post'
with vcr.use_cassette(str(tmpdir.join('post_data.yaml'))):
"""Ensure that it works when posting unicode data"""
data = urlencode({"snowman": u"☃".encode("utf-8")})
url = httpbin_both.url + "/post"
with vcr.use_cassette(str(tmpdir.join("post_data.yaml"))):
_, res1 = http().request(url, "POST", data)
with vcr.use_cassette(str(tmpdir.join('post_data.yaml'))) as cass:
with vcr.use_cassette(str(tmpdir.join("post_data.yaml"))) as cass:
_, res2 = http().request(url, "POST", data)
assert res1 == res2
@@ -131,11 +124,11 @@ def test_post_unicode_data(tmpdir, httpbin_both):
def test_cross_scheme(tmpdir, httpbin, httpbin_secure):
'''Ensure that requests between schemes are treated separately'''
"""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:
with vcr.use_cassette(str(tmpdir.join("cross_scheme.yaml"))) as cass:
http().request(httpbin_secure.url)
http().request(httpbin.url)
assert len(cass) == 2
@@ -143,17 +136,17 @@ def test_cross_scheme(tmpdir, httpbin, httpbin_secure):
def test_decorator(tmpdir, httpbin_both):
'''Test the decorator version of VCR.py'''
"""Test the decorator version of VCR.py"""
url = httpbin_both.url
@vcr.use_cassette(str(tmpdir.join('atts.yaml')))
@vcr.use_cassette(str(tmpdir.join("atts.yaml")))
def inner1():
resp, _ = http().request(url)
return resp['status']
return resp["status"]
@vcr.use_cassette(str(tmpdir.join('atts.yaml')))
@vcr.use_cassette(str(tmpdir.join("atts.yaml")))
def inner2():
resp, _ = http().request(url)
return resp['status']
return resp["status"]
assert inner1() == inner2()

50
tests/integration/test_ignore.py
View File

@@ -15,59 +15,53 @@ def overridden_dns(overrides):
def fake_getaddrinfo(*args, **kwargs):
if args[0] in overrides:
address = overrides[args[0]]
return [(2, 1, 6, '', (address, args[1]))]
return [(2, 1, 6, "", (address, args[1]))]
return real_getaddrinfo(*args, **kwargs)
socket.getaddrinfo = fake_getaddrinfo
yield
socket.getaddrinfo = real_getaddrinfo
def test_ignore_localhost(tmpdir, httpbin):
with overridden_dns({'httpbin.org': '127.0.0.1'}):
cass_file = str(tmpdir.join('filter_qs.yaml'))
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:{}/'.format(httpbin.port))
urlopen("http://localhost:{}/".format(httpbin.port))
assert len(cass) == 0
urlopen('http://httpbin.org:{}/'.format(httpbin.port))
urlopen("http://httpbin.org:{}/".format(httpbin.port))
assert len(cass) == 1
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:{}/'.format(httpbin.port))
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:{}/".format(httpbin.port))
assert len(cass) == 0
urlopen('http://localhost:{}/'.format(httpbin.port))
urlopen("http://localhost:{}/".format(httpbin.port))
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:{}'.format(httpbin.port))
urlopen('http://localhost:{}'.format(httpbin.port))
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:{}".format(httpbin.port))
urlopen("http://localhost:{}".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 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:{}'.format(httpbin.port))
urlopen("http://localhost:{}".format(httpbin.port))
assert len(cass) == 0
urlopen('http://httpbin.org:{}'.format(httpbin.port))
urlopen("http://httpbin.org:{}".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:{}'.format(httpbin.port))
urlopen('http://httpbin.org:{}'.format(httpbin.port))
urlopen("http://localhost:{}".format(httpbin.port))
urlopen("http://httpbin.org:{}".format(httpbin.port))
assert len(cass) == 1

68
tests/integration/test_matchers.py
View File

@@ -3,11 +3,11 @@ import pytest
from six.moves.urllib.request import urlopen
DEFAULT_URI = 'http://httpbin.org/get?p1=q1&p2=q2' # base uri for testing
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)
return uri.replace("http://httpbin.org", httpbin.url).replace("https://httpbin.org", httpbin_secure.url)
@pytest.fixture
@@ -18,29 +18,22 @@ def cassette(tmpdir, httpbin, httpbin_secure):
"""
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'):
cassette_path = str(tmpdir.join("test.yml"))
with vcr.use_cassette(cassette_path, record_mode="all"):
urlopen(default_uri)
return cassette_path
@pytest.mark.parametrize("matcher, matching_uri, not_matching_uri", [
('uri',
'http://httpbin.org/get?p1=q1&p2=q2',
'http://httpbin.org/get?p2=q2&p1=q1'),
('scheme',
'http://google.com/post?a=b',
'https://httpbin.org/get?p1=q1&p2=q2'),
('host',
'https://httpbin.org/post?a=b',
'http://google.com/get?p1=q1&p2=q2'),
('path',
'https://google.com/get?a=b',
'http://httpbin.org/post?p1=q1&p2=q2'),
('query',
'https://google.com/get?p2=q2&p1=q1',
'http://httpbin.org/get?p1=q1&a=b')
])
@pytest.mark.parametrize(
"matcher, matching_uri, not_matching_uri",
[
("uri", "http://httpbin.org/get?p1=q1&p2=q2", "http://httpbin.org/get?p2=q2&p1=q1"),
("scheme", "http://google.com/post?a=b", "https://httpbin.org/get?p1=q1&p2=q2"),
("host", "https://httpbin.org/post?a=b", "http://google.com/get?p1=q1&p2=q2"),
("path", "https://google.com/get?a=b", "http://httpbin.org/post?p1=q1&p2=q2"),
("query", "https://google.com/get?p2=q2&p1=q1", "http://httpbin.org/get?p1=q1&a=b"),
],
)
def test_matchers(httpbin, httpbin_secure, cassette, matcher, matching_uri, not_matching_uri):
matching_uri = _replace_httpbin(matching_uri, httpbin, httpbin_secure)
@@ -67,22 +60,20 @@ 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')
with vcr.use_cassette(cassette, match_on=["method"]) as cass:
urlopen("https://google.com/get?a=b")
assert cass.play_count == 1
# should fail if method does not match
with pytest.raises(vcr.errors.CannotOverwriteExistingCassetteException):
with vcr.use_cassette(cassette, match_on=['method']) as cass:
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", [
DEFAULT_URI,
'http://httpbin.org/get?p2=q2&p1=q1',
'http://httpbin.org/get?p2=q2&p1=q1',
])
@pytest.mark.parametrize(
"uri", [DEFAULT_URI, "http://httpbin.org/get?p2=q2&p1=q1", "http://httpbin.org/get?p2=q2&p1=q1"]
)
def test_default_matcher_matches(cassette, uri, httpbin, httpbin_secure):
uri = _replace_httpbin(uri, httpbin, httpbin_secure)
@@ -92,12 +83,15 @@ def test_default_matcher_matches(cassette, uri, httpbin, httpbin_secure):
assert cass.play_count == 1
@pytest.mark.parametrize("uri", [
'https://httpbin.org/get?p1=q1&p2=q2',
'http://google.com/get?p1=q1&p2=q2',
'http://httpbin.org/post?p1=q1&p2=q2',
'http://httpbin.org/get?p1=q1&a=b'
])
@pytest.mark.parametrize(
"uri",
[
"https://httpbin.org/get?p1=q1&p2=q2",
"http://google.com/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, httpbin, httpbin_secure):
uri = _replace_httpbin(uri, httpbin, httpbin_secure)
with pytest.raises(vcr.errors.CannotOverwriteExistingCassetteException):
@@ -110,4 +104,4 @@ def test_default_matcher_does_not_match_on_method(cassette, httpbin, httpbin_sec
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"")

14
tests/integration/test_multiple.py
View File

@@ -6,15 +6,15 @@ from six.moves.urllib.request import urlopen
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(httpbin.url + '/status/200')
urlopen(httpbin.url + '/status/201')
with vcr.use_cassette(str(tmpdir.join("test.json")), match_on=["method"]):
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(httpbin.url + '/status/200').getcode() == 200
assert urlopen(httpbin.url + '/status/201').getcode() == 201
with vcr.use_cassette(str(tmpdir.join("test.json")), match_on=["method"]):
assert urlopen(httpbin.url + "/status/200").getcode() == 200
assert urlopen(httpbin.url + "/status/201").getcode() == 201
with pytest.raises(Exception):
urlopen(httpbin.url + '/status/200')
urlopen(httpbin.url + "/status/200")

27
tests/integration/test_proxy.py
View File

@@ -1,5 +1,5 @@
# -*- coding: utf-8 -*-
'''Test using a proxy.'''
"""Test using a proxy."""
# External imports
import multiprocessing
@@ -16,11 +16,12 @@ requests = pytest.importorskip("requests")
class Proxy(SimpleHTTPServer.SimpleHTTPRequestHandler):
'''
"""
Simple proxy server.
(Inspired by: http://effbot.org/librarybook/simplehttpserver.htm).
'''
"""
def do_GET(self):
upstream_response = urlopen(self.path)
try:
@@ -37,24 +38,22 @@ class Proxy(SimpleHTTPServer.SimpleHTTPRequestHandler):
self.copyfile(upstream_response, self.wfile)
@pytest.yield_fixture(scope='session')
@pytest.yield_fixture(scope="session")
def proxy_server():
httpd = socketserver.ThreadingTCPServer(('', 0), Proxy)
proxy_process = multiprocessing.Process(
target=httpd.serve_forever,
)
httpd = socketserver.ThreadingTCPServer(("", 0), Proxy)
proxy_process = multiprocessing.Process(target=httpd.serve_forever)
proxy_process.start()
yield 'http://{}:{}'.format(*httpd.server_address)
yield "http://{}:{}".format(*httpd.server_address)
proxy_process.terminate()
def test_use_proxy(tmpdir, httpbin, proxy_server):
'''Ensure that it works with a proxy.'''
with vcr.use_cassette(str(tmpdir.join('proxy.yaml'))):
response = requests.get(httpbin.url, proxies={'http': proxy_server})
"""Ensure that it works with a proxy."""
with vcr.use_cassette(str(tmpdir.join("proxy.yaml"))):
response = requests.get(httpbin.url, proxies={"http": proxy_server})
with vcr.use_cassette(str(tmpdir.join('proxy.yaml'))) as cassette:
cassette_response = requests.get(httpbin.url, proxies={'http': proxy_server})
with vcr.use_cassette(str(tmpdir.join("proxy.yaml"))) as cassette:
cassette_response = requests.get(httpbin.url, proxies={"http": proxy_server})
assert cassette_response.headers == response.headers
assert cassette.play_count == 1

26
tests/integration/test_record_mode.py
View File

@@ -4,7 +4,7 @@ from six.moves.urllib.request import urlopen
def test_once_record_mode(tmpdir, httpbin):
testfile = str(tmpdir.join('recordmode.yml'))
testfile = str(tmpdir.join("recordmode.yml"))
with vcr.use_cassette(testfile, record_mode="once"):
# cassette file doesn't exist, so create.
urlopen(httpbin.url).read()
@@ -17,11 +17,11 @@ def test_once_record_mode(tmpdir, httpbin):
# but, try to access something else from the same cassette, and an
# exception is raised.
with pytest.raises(Exception):
urlopen(httpbin.url + '/get').read()
urlopen(httpbin.url + "/get").read()
def test_once_record_mode_two_times(tmpdir, httpbin):
testfile = str(tmpdir.join('recordmode.yml'))
testfile = str(tmpdir.join("recordmode.yml"))
with vcr.use_cassette(testfile, record_mode="once"):
# get two of the same file
urlopen(httpbin.url).read()
@@ -34,7 +34,7 @@ def test_once_record_mode_two_times(tmpdir, httpbin):
def test_once_mode_three_times(tmpdir, httpbin):
testfile = str(tmpdir.join('recordmode.yml'))
testfile = str(tmpdir.join("recordmode.yml"))
with vcr.use_cassette(testfile, record_mode="once"):
# get three of the same file
urlopen(httpbin.url).read()
@@ -43,7 +43,7 @@ def test_once_mode_three_times(tmpdir, httpbin):
def test_new_episodes_record_mode(tmpdir, httpbin):
testfile = str(tmpdir.join('recordmode.yml'))
testfile = str(tmpdir.join("recordmode.yml"))
with vcr.use_cassette(testfile, record_mode="new_episodes"):
# cassette file doesn't exist, so create.
@@ -58,7 +58,7 @@ def test_new_episodes_record_mode(tmpdir, httpbin):
# in the "new_episodes" record mode, we can add more requests to
# a cassette without repurcussions.
urlopen(httpbin.url + '/get').read()
urlopen(httpbin.url + "/get").read()
# one of the responses has been played
assert cass.play_count == 1
@@ -72,8 +72,8 @@ def test_new_episodes_record_mode(tmpdir, httpbin):
def test_new_episodes_record_mode_two_times(tmpdir, httpbin):
testfile = str(tmpdir.join('recordmode.yml'))
url = httpbin.url + '/bytes/1024'
testfile = str(tmpdir.join("recordmode.yml"))
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()
@@ -97,7 +97,7 @@ def test_new_episodes_record_mode_two_times(tmpdir, httpbin):
def test_all_record_mode(tmpdir, httpbin):
testfile = str(tmpdir.join('recordmode.yml'))
testfile = str(tmpdir.join("recordmode.yml"))
with vcr.use_cassette(testfile, record_mode="all"):
# cassette file doesn't exist, so create.
@@ -109,7 +109,7 @@ def test_all_record_mode(tmpdir, httpbin):
# in the "all" record mode, we can add more requests to
# a cassette without repurcussions.
urlopen(httpbin.url + '/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
@@ -120,7 +120,7 @@ def test_all_record_mode(tmpdir, httpbin):
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'))
testfile = str(tmpdir.join("recordmode.yml"))
with vcr.use_cassette(testfile, record_mode="none"):
with pytest.raises(Exception):
urlopen(httpbin.url).read()
@@ -128,7 +128,7 @@ def test_none_record_mode(tmpdir, httpbin):
def test_none_record_mode_with_existing_cassette(tmpdir, httpbin):
# create a cassette file
testfile = str(tmpdir.join('recordmode.yml'))
testfile = str(tmpdir.join("recordmode.yml"))
with vcr.use_cassette(testfile, record_mode="all"):
urlopen(httpbin.url).read()
@@ -139,4 +139,4 @@ def test_none_record_mode_with_existing_cassette(tmpdir, httpbin):
assert cass.play_count == 1
# but if I try to hit the net, raise an exception.
with pytest.raises(Exception):
urlopen(httpbin.url + '/get').read()
urlopen(httpbin.url + "/get").read()

22
tests/integration/test_register_matcher.py
View File

@@ -12,25 +12,25 @@ def false_matcher(r1, r2):
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']):
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(httpbin.url)
urlopen(httpbin.url + '/get')
urlopen(httpbin.url + "/get")
with my_vcr.use_cassette(testfile, match_on=['true']):
with my_vcr.use_cassette(testfile, match_on=["true"]):
# I can get the response twice even though I only asked for it once
urlopen(httpbin.url + '/get')
urlopen(httpbin.url + '/get')
urlopen(httpbin.url + "/get")
urlopen(httpbin.url + "/get")
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:
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(httpbin.url)
urlopen(httpbin.url + '/get')
urlopen(httpbin.url + "/get")
assert len(cass) == 2

32
tests/integration/test_register_persister.py
View File

@@ -1,5 +1,5 @@
# -*- coding: utf-8 -*-
'''Tests for cassettes with custom persistence'''
"""Tests for cassettes with custom persistence"""
# External imports
import os
@@ -11,45 +11,45 @@ from vcr.persisters.filesystem import FilesystemPersister
class CustomFilesystemPersister(object):
'''Behaves just like default FilesystemPersister but adds .test extension
to the cassette file'''
"""Behaves just like default FilesystemPersister but adds .test extension
to the cassette file"""
@staticmethod
def load_cassette(cassette_path, serializer):
cassette_path += '.test'
cassette_path += ".test"
return FilesystemPersister.load_cassette(cassette_path, serializer)
@staticmethod
def save_cassette(cassette_path, cassette_dict, serializer):
cassette_path += '.test'
FilesystemPersister.save_cassette(cassette_path, cassette_dict,
serializer)
cassette_path += ".test"
FilesystemPersister.save_cassette(cassette_path, cassette_dict, serializer)
def test_save_cassette_with_custom_persister(tmpdir, httpbin):
'''Ensure you can save a cassette using custom persister'''
"""Ensure you can save a cassette using custom persister"""
my_vcr = vcr.VCR()
my_vcr.register_persister(CustomFilesystemPersister)
# Check to make sure directory doesnt exist
assert not os.path.exists(str(tmpdir.join('nonexistent')))
assert not os.path.exists(str(tmpdir.join("nonexistent")))
# Run VCR to create dir and cassette file using new save_cassette callback
with my_vcr.use_cassette(str(tmpdir.join('nonexistent', 'cassette.yml'))):
with my_vcr.use_cassette(str(tmpdir.join("nonexistent", "cassette.yml"))):
urlopen(httpbin.url).read()
# Callback should have made the file and the directory
assert os.path.exists(str(tmpdir.join('nonexistent', 'cassette.yml.test')))
assert os.path.exists(str(tmpdir.join("nonexistent", "cassette.yml.test")))
def test_load_cassette_with_custom_persister(tmpdir, httpbin):
'''
"""
Ensure you can load a cassette using custom persister
'''
"""
my_vcr = vcr.VCR()
my_vcr.register_persister(CustomFilesystemPersister)
test_fixture = str(tmpdir.join('synopsis.json.test'))
test_fixture = str(tmpdir.join("synopsis.json.test"))
with my_vcr.use_cassette(test_fixture, serializer='json'):
with my_vcr.use_cassette(test_fixture, serializer="json"):
response = urlopen(httpbin.url).read()
assert b'difficult sometimes' in response
assert b"difficult sometimes" in response

10
tests/integration/test_register_serializer.py
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
@@ -20,13 +20,13 @@ class MockSerializer(object):
def test_registered_serializer(tmpdir):
ms = MockSerializer()
my_vcr = vcr.VCR()
my_vcr.register_serializer('mock', ms)
tmpdir.join('test.mock').write('test_data')
with my_vcr.use_cassette(str(tmpdir.join('test.mock')), serializer='mock'):
my_vcr.register_serializer("mock", ms)
tmpdir.join("test.mock").write("test_data")
with my_vcr.use_cassette(str(tmpdir.join("test.mock")), serializer="mock"):
# Serializer deserialized once
assert ms.serialize_count == 1
# and serialized the test data string
assert ms.cassette_string == 'test_data'
assert ms.cassette_string == "test_data"
# and hasn't serialized yet
assert ms.deserialize_count == 0

14
tests/integration/test_request.py
View File

@@ -3,17 +3,17 @@ from six.moves.urllib.request import urlopen
def test_recorded_request_uri_with_redirected_request(tmpdir, httpbin):
with vcr.use_cassette(str(tmpdir.join('test.yml'))) as cass:
with vcr.use_cassette(str(tmpdir.join("test.yml"))) as cass:
assert len(cass) == 0
urlopen(httpbin.url + '/redirect/3')
assert cass.requests[0].uri == httpbin.url + '/redirect/3'
assert cass.requests[3].uri == httpbin.url + '/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, httpbin):
with vcr.use_cassette(str(tmpdir.join('test.yml'))) as cass:
with vcr.use_cassette(str(tmpdir.join("test.yml"))) as cass:
assert len(cass) == 0
urlopen(httpbin.url + '/response-headers?foo=bar&foo=baz')
urlopen(httpbin.url + "/response-headers?foo=bar&foo=baz")
assert len(cass) == 1
assert cass.responses[0]['headers']['foo'] == ['bar', 'baz']
assert cass.responses[0]["headers"]["foo"] == ["bar", "baz"]

221
tests/integration/test_requests.py
View File

@@ -1,5 +1,5 @@
# -*- coding: utf-8 -*-
'''Test requests' interaction with vcr'''
"""Test requests' interaction with vcr"""
import platform
import pytest
import sys
@@ -11,76 +11,76 @@ from requests.exceptions import ConnectionError # noqa E402
def test_status_code(httpbin_both, tmpdir):
'''Ensure that we can read the status code'''
url = httpbin_both.url + '/'
with vcr.use_cassette(str(tmpdir.join('atts.yaml'))):
"""Ensure that we can read the status code"""
url = httpbin_both.url + "/"
with vcr.use_cassette(str(tmpdir.join("atts.yaml"))):
status_code = requests.get(url).status_code
with vcr.use_cassette(str(tmpdir.join('atts.yaml'))):
with vcr.use_cassette(str(tmpdir.join("atts.yaml"))):
assert status_code == requests.get(url).status_code
def test_headers(httpbin_both, tmpdir):
'''Ensure that we can read the headers back'''
url = httpbin_both + '/'
with vcr.use_cassette(str(tmpdir.join('headers.yaml'))):
"""Ensure that we can read the headers back"""
url = httpbin_both + "/"
with vcr.use_cassette(str(tmpdir.join("headers.yaml"))):
headers = requests.get(url).headers
with vcr.use_cassette(str(tmpdir.join('headers.yaml'))):
with vcr.use_cassette(str(tmpdir.join("headers.yaml"))):
assert headers == requests.get(url).headers
def test_body(tmpdir, httpbin_both):
'''Ensure the responses are all identical enough'''
url = httpbin_both + '/bytes/1024'
with vcr.use_cassette(str(tmpdir.join('body.yaml'))):
"""Ensure the responses are all identical enough"""
url = httpbin_both + "/bytes/1024"
with vcr.use_cassette(str(tmpdir.join("body.yaml"))):
content = requests.get(url).content
with vcr.use_cassette(str(tmpdir.join('body.yaml'))):
with vcr.use_cassette(str(tmpdir.join("body.yaml"))):
assert content == requests.get(url).content
def test_get_empty_content_type_json(tmpdir, httpbin_both):
'''Ensure GET with application/json content-type and empty request body doesn't crash'''
url = httpbin_both + '/status/200'
headers = {'Content-Type': 'application/json'}
"""Ensure GET with application/json content-type and empty request body doesn't crash"""
url = httpbin_both + "/status/200"
headers = {"Content-Type": "application/json"}
with vcr.use_cassette(str(tmpdir.join('get_empty_json.yaml')), match_on=('body',)):
with vcr.use_cassette(str(tmpdir.join("get_empty_json.yaml")), match_on=("body",)):
status = requests.get(url, headers=headers).status_code
with vcr.use_cassette(str(tmpdir.join('get_empty_json.yaml')), match_on=('body',)):
with vcr.use_cassette(str(tmpdir.join("get_empty_json.yaml")), match_on=("body",)):
assert status == requests.get(url, headers=headers).status_code
def test_effective_url(tmpdir, httpbin_both):
'''Ensure that the effective_url is captured'''
url = httpbin_both.url + '/redirect-to?url=/html'
with vcr.use_cassette(str(tmpdir.join('url.yaml'))):
"""Ensure that the effective_url is captured"""
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 == httpbin_both.url + '/html'
assert effective_url == httpbin_both.url + "/html"
with vcr.use_cassette(str(tmpdir.join('url.yaml'))):
with vcr.use_cassette(str(tmpdir.join("url.yaml"))):
assert effective_url == requests.get(url).url
def test_auth(tmpdir, httpbin_both):
'''Ensure that we can handle basic auth'''
auth = ('user', 'passwd')
url = httpbin_both + '/basic-auth/user/passwd'
with vcr.use_cassette(str(tmpdir.join('auth.yaml'))):
"""Ensure that we can handle basic auth"""
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)
with vcr.use_cassette(str(tmpdir.join('auth.yaml'))):
with vcr.use_cassette(str(tmpdir.join("auth.yaml"))):
two = requests.get(url, auth=auth)
assert one.content == two.content
assert one.status_code == two.status_code
def test_auth_failed(tmpdir, httpbin_both):
'''Ensure that we can save failed auth statuses'''
auth = ('user', 'wrongwrongwrong')
url = httpbin_both + '/basic-auth/user/passwd'
with vcr.use_cassette(str(tmpdir.join('auth-failed.yaml'))) as cass:
"""Ensure that we can save failed auth statuses"""
auth = ("user", "wrongwrongwrong")
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)
one = requests.get(url, auth=auth)
@@ -90,58 +90,59 @@ def test_auth_failed(tmpdir, httpbin_both):
def test_post(tmpdir, httpbin_both):
'''Ensure that we can post and cache the results'''
data = {'key1': 'value1', 'key2': 'value2'}
url = httpbin_both + '/post'
with vcr.use_cassette(str(tmpdir.join('requests.yaml'))):
"""Ensure that we can post and cache the results"""
data = {"key1": "value1", "key2": "value2"}
url = httpbin_both + "/post"
with vcr.use_cassette(str(tmpdir.join("requests.yaml"))):
req1 = requests.post(url, data).content
with vcr.use_cassette(str(tmpdir.join('requests.yaml'))):
with vcr.use_cassette(str(tmpdir.join("requests.yaml"))):
req2 = requests.post(url, data).content
assert req1 == req2
def test_post_chunked_binary(tmpdir, httpbin):
'''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 = httpbin.url + '/post'
with vcr.use_cassette(str(tmpdir.join('requests.yaml'))):
"""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 = httpbin.url + "/post"
with vcr.use_cassette(str(tmpdir.join("requests.yaml"))):
req1 = requests.post(url, data1).content
with vcr.use_cassette(str(tmpdir.join('requests.yaml'))):
with vcr.use_cassette(str(tmpdir.join("requests.yaml"))):
req2 = requests.post(url, data2).content
assert req1 == req2
@pytest.mark.skipif('sys.version_info >= (3, 6)', strict=True, raises=ConnectionError)
@pytest.mark.skipif((3, 5) < sys.version_info < (3, 6) and
platform.python_implementation() == 'CPython',
reason='Fails on CPython 3.5')
@pytest.mark.skipif("sys.version_info >= (3, 6)", strict=True, raises=ConnectionError)
@pytest.mark.skipif(
(3, 5) < sys.version_info < (3, 6) and platform.python_implementation() == "CPython",
reason="Fails on CPython 3.5",
)
def test_post_chunked_binary_secure(tmpdir, httpbin_secure):
'''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 = httpbin_secure.url + '/post'
with vcr.use_cassette(str(tmpdir.join('requests.yaml'))):
"""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 = httpbin_secure.url + "/post"
with vcr.use_cassette(str(tmpdir.join("requests.yaml"))):
req1 = requests.post(url, data1).content
print(req1)
with vcr.use_cassette(str(tmpdir.join('requests.yaml'))):
with vcr.use_cassette(str(tmpdir.join("requests.yaml"))):
req2 = requests.post(url, data2).content
assert req1 == req2
def test_redirects(tmpdir, httpbin_both):
'''Ensure that we can handle redirects'''
url = httpbin_both + '/redirect-to?url=bytes/1024'
with vcr.use_cassette(str(tmpdir.join('requests.yaml'))):
"""Ensure that we can handle redirects"""
url = httpbin_both + "/redirect-to?url=bytes/1024"
with vcr.use_cassette(str(tmpdir.join("requests.yaml"))):
content = requests.get(url).content
with vcr.use_cassette(str(tmpdir.join('requests.yaml'))) as cass:
with vcr.use_cassette(str(tmpdir.join("requests.yaml"))) as cass:
assert content == requests.get(url).content
# Ensure that we've now cached *two* responses. One for the redirect
# and one for the final fetch
@@ -150,149 +151,151 @@ def test_redirects(tmpdir, httpbin_both):
def test_cross_scheme(tmpdir, httpbin_secure, httpbin):
'''Ensure that requests between schemes are treated separately'''
"""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(httpbin_secure + '/')
requests.get(httpbin + '/')
with vcr.use_cassette(str(tmpdir.join("cross_scheme.yaml"))) as cass:
requests.get(httpbin_secure + "/")
requests.get(httpbin + "/")
assert cass.play_count == 0
assert len(cass) == 2
def test_gzip(tmpdir, httpbin_both):
'''
"""
Ensure that requests (actually urllib3) is able to automatically decompress
the response body
'''
url = httpbin_both + '/gzip'
"""
url = httpbin_both + "/gzip"
response = requests.get(url)
with vcr.use_cassette(str(tmpdir.join('gzip.yaml'))):
with vcr.use_cassette(str(tmpdir.join("gzip.yaml"))):
response = requests.get(url)
assert_is_json(response.content)
with vcr.use_cassette(str(tmpdir.join('gzip.yaml'))):
with vcr.use_cassette(str(tmpdir.join("gzip.yaml"))):
assert_is_json(response.content)
def test_session_and_connection_close(tmpdir, httpbin):
'''
"""
This tests the issue in https://github.com/kevin1024/vcrpy/issues/48
If you use a requests.session and the connection is closed, then an
exception is raised in the urllib3 module vendored into requests:
`AttributeError: 'NoneType' object has no attribute 'settimeout'`
'''
with vcr.use_cassette(str(tmpdir.join('session_connection_closed.yaml'))):
"""
with vcr.use_cassette(str(tmpdir.join("session_connection_closed.yaml"))):
session = requests.session()
session.get(httpbin + '/get', headers={'Connection': 'close'})
session.get(httpbin + '/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, httpbin_secure):
with vcr.use_cassette(str(tmpdir.join('cert_validation_disabled.yaml'))):
with vcr.use_cassette(str(tmpdir.join("cert_validation_disabled.yaml"))):
requests.get(httpbin_secure.url, verify=False)
def test_session_can_make_requests_after_requests_unpatched(tmpdir, httpbin):
with vcr.use_cassette(str(tmpdir.join('test_session_after_unpatched.yaml'))):
with vcr.use_cassette(str(tmpdir.join("test_session_after_unpatched.yaml"))):
session = requests.session()
session.get(httpbin + '/get')
session.get(httpbin + "/get")
with vcr.use_cassette(str(tmpdir.join('test_session_after_unpatched.yaml'))):
with vcr.use_cassette(str(tmpdir.join("test_session_after_unpatched.yaml"))):
session = requests.session()
session.get(httpbin + '/get')
session.get(httpbin + "/get")
session.get(httpbin + '/status/200')
session.get(httpbin + "/status/200")
def test_session_created_before_use_cassette_is_patched(tmpdir, httpbin_both):
url = httpbin_both + '/bytes/1024'
url = httpbin_both + "/bytes/1024"
# Record arbitrary, random data to the cassette
with vcr.use_cassette(str(tmpdir.join('session_created_outside.yaml'))):
with vcr.use_cassette(str(tmpdir.join("session_created_outside.yaml"))):
session = requests.session()
body = session.get(url).content
# 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(httpbin_both + '/get')
session.get(httpbin_both + "/get")
with vcr.use_cassette(str(tmpdir.join('session_created_outside.yaml'))):
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(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 = httpbin_both + '/bytes/1024'
with vcr.use_cassette(str(tmpdir.join('first_nested.yaml'))):
"""
url = httpbin_both + "/bytes/1024"
with vcr.use_cassette(str(tmpdir.join("first_nested.yaml"))):
session = requests.session()
first_body = session.get(url).content
with vcr.use_cassette(str(tmpdir.join('second_nested.yaml'))):
with vcr.use_cassette(str(tmpdir.join("second_nested.yaml"))):
second_body = session.get(url).content
third_body = requests.get(url).content
with vcr.use_cassette(str(tmpdir.join('second_nested.yaml'))):
with vcr.use_cassette(str(tmpdir.join("second_nested.yaml"))):
session = requests.session()
assert session.get(url).content == second_body
with vcr.use_cassette(str(tmpdir.join('first_nested.yaml'))):
with vcr.use_cassette(str(tmpdir.join("first_nested.yaml"))):
assert session.get(url).content == first_body
assert session.get(url).content == third_body
# Make sure that the session can now get content normally.
assert 'User-agent' in session.get(httpbin_both.url + '/robots.txt').text
assert "User-agent" in session.get(httpbin_both.url + "/robots.txt").text
def test_post_file(tmpdir, httpbin_both):
'''Ensure that we handle posting a file.'''
url = httpbin_both + '/post'
with vcr.use_cassette(str(tmpdir.join('post_file.yaml'))) as cass, open('tox.ini', 'rb') as f:
"""Ensure that we handle posting a file."""
url = httpbin_both + "/post"
with vcr.use_cassette(str(tmpdir.join("post_file.yaml"))) as cass, open("tox.ini", "rb") as f:
original_response = requests.post(url, f).content
# This also tests that we do the right thing with matching the body when they are files.
with vcr.use_cassette(str(tmpdir.join('post_file.yaml')),
match_on=('method', 'scheme', 'host', 'port', 'path', 'query', 'body')) as cass:
with open('tox.ini', 'rb') as f:
with vcr.use_cassette(
str(tmpdir.join("post_file.yaml")),
match_on=("method", "scheme", "host", "port", "path", "query", "body"),
) as cass:
with open("tox.ini", "rb") as f:
tox_content = f.read()
assert cass.requests[0].body.read() == tox_content
with open('tox.ini', 'rb') as f:
with open("tox.ini", "rb") as f:
new_response = requests.post(url, f).content
assert original_response == new_response
def test_filter_post_params(tmpdir, httpbin_both):
'''
"""
This tests the issue in https://github.com/kevin1024/vcrpy/issues/158
Ensure that a post request made through requests can still be filtered.
with vcr.use_cassette(cass_file, filter_post_data_parameters=['id']) as cass:
assert b'id=secret' not in cass.requests[0].body
'''
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'})
with vcr.use_cassette(cass_loc, filter_post_data_parameters=['key']) as cass:
assert b'key=value' not in cass.requests[0].body
"""
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"})
with vcr.use_cassette(cass_loc, filter_post_data_parameters=["key"]) as cass:
assert b"key=value" not in cass.requests[0].body
def test_post_unicode_match_on_body(tmpdir, httpbin_both):
'''Ensure that matching on POST body that contains Unicode characters works.'''
data = {'key1': 'value1', '●‿●': '٩(●̮̮̃•̃)۶'}
url = httpbin_both + '/post'
"""Ensure that matching on POST body that contains Unicode characters works."""
data = {"key1": "value1", "●‿●": "٩(●̮̮̃•̃)۶"}
url = httpbin_both + "/post"
with vcr.use_cassette(str(tmpdir.join('requests.yaml')), additional_matchers=('body',)):
with vcr.use_cassette(str(tmpdir.join("requests.yaml")), additional_matchers=("body",)):
req1 = requests.post(url, data).content
with vcr.use_cassette(str(tmpdir.join('requests.yaml')), additional_matchers=('body',)):
with vcr.use_cassette(str(tmpdir.join("requests.yaml")), additional_matchers=("body",)):
req2 = requests.post(url, data).content
assert req1 == req2

61
tests/integration/test_stubs.py
View File

@@ -8,18 +8,18 @@ from assertions import assert_is_json
def _headers_are_case_insensitive(host, port):
conn = httplib.HTTPConnection(host, port)
conn.request('GET', "/cookies/set?k1=v1")
conn.request("GET", "/cookies/set?k1=v1")
r1 = conn.getresponse()
cookie_data1 = r1.getheader('set-cookie')
cookie_data1 = r1.getheader("set-cookie")
conn = httplib.HTTPConnection(host, port)
conn.request('GET', "/cookies/set?k1=v1")
conn.request("GET", "/cookies/set?k1=v1")
r2 = conn.getresponse()
cookie_data2 = r2.getheader('Set-Cookie')
cookie_data2 = r2.getheader("Set-Cookie")
return cookie_data1 == cookie_data2
def test_case_insensitivity(tmpdir, httpbin):
testfile = str(tmpdir.join('case_insensitivity.yml'))
testfile = str(tmpdir.join("case_insensitivity.yml"))
# check if headers are case insensitive outside of vcrpy
host, port = httpbin.host, httpbin.port
outside = _headers_are_case_insensitive(host, port)
@@ -35,13 +35,13 @@ def test_case_insensitivity(tmpdir, httpbin):
def _multiple_header_value(httpbin):
conn = httplib.HTTPConnection(httpbin.host, httpbin.port)
conn.request('GET', "/response-headers?foo=bar&foo=baz")
conn.request("GET", "/response-headers?foo=bar&foo=baz")
r = conn.getresponse()
return r.getheader('foo')
return r.getheader("foo")
def test_multiple_headers(tmpdir, httpbin):
testfile = str(tmpdir.join('multiple_headers.yaml'))
testfile = str(tmpdir.join("multiple_headers.yaml"))
outside = _multiple_header_value(httpbin)
with vcr.use_cassette(testfile):
@@ -51,83 +51,84 @@ def test_multiple_headers(tmpdir, httpbin):
def test_original_decoded_response_is_not_modified(tmpdir, httpbin):
testfile = str(tmpdir.join('decoded_response.yml'))
testfile = str(tmpdir.join("decoded_response.yml"))
host, port = httpbin.host, httpbin.port
conn = httplib.HTTPConnection(host, port)
conn.request('GET', '/gzip')
conn.request("GET", "/gzip")
outside = conn.getresponse()
with vcr.use_cassette(testfile, decode_compressed_response=True):
conn = httplib.HTTPConnection(host, port)
conn.request('GET', '/gzip')
conn.request("GET", "/gzip")
inside = conn.getresponse()
# Assert that we do not modify the original response while appending
# to the casssette.
assert 'gzip' == inside.headers['content-encoding']
assert "gzip" == inside.headers["content-encoding"]
# They should effectively be the same response.
inside_headers = (h for h in inside.headers.items() if h[0].lower() != 'date')
outside_headers = (h for h in outside.getheaders() if h[0].lower() != 'date')
inside_headers = (h for h in inside.headers.items() if h[0].lower() != "date")
outside_headers = (h for h in outside.getheaders() if h[0].lower() != "date")
assert set(inside_headers) == set(outside_headers)
inside = zlib.decompress(inside.read(), 16+zlib.MAX_WBITS)
outside = zlib.decompress(outside.read(), 16+zlib.MAX_WBITS)
inside = zlib.decompress(inside.read(), 16 + zlib.MAX_WBITS)
outside = zlib.decompress(outside.read(), 16 + zlib.MAX_WBITS)
assert inside == outside
# Even though the above are raw bytes, the JSON data should have been
# decoded and saved to the cassette.
with vcr.use_cassette(testfile):
conn = httplib.HTTPConnection(host, port)
conn.request('GET', '/gzip')
conn.request("GET", "/gzip")
inside = conn.getresponse()
assert 'content-encoding' not in inside.headers
assert "content-encoding" not in inside.headers
assert_is_json(inside.read())
def _make_before_record_response(fields, replacement='[REDACTED]'):
def _make_before_record_response(fields, replacement="[REDACTED]"):
def before_record_response(response):
string_body = response['body']['string'].decode('utf8')
string_body = response["body"]["string"].decode("utf8")
body = json.loads(string_body)
for field in fields:
if field in body:
body[field] = replacement
response['body']['string'] = json.dumps(body).encode()
response["body"]["string"] = json.dumps(body).encode()
return response
return before_record_response
def test_original_response_is_not_modified_by_before_filter(tmpdir, httpbin):
testfile = str(tmpdir.join('sensitive_data_scrubbed_response.yml'))
testfile = str(tmpdir.join("sensitive_data_scrubbed_response.yml"))
host, port = httpbin.host, httpbin.port
field_to_scrub = 'url'
replacement = '[YOU_CANT_HAVE_THE_MANGO]'
field_to_scrub = "url"
replacement = "[YOU_CANT_HAVE_THE_MANGO]"
conn = httplib.HTTPConnection(host, port)
conn.request('GET', '/get')
conn.request("GET", "/get")
outside = conn.getresponse()
callback = _make_before_record_response([field_to_scrub], replacement)
with vcr.use_cassette(testfile, before_record_response=callback):
conn = httplib.HTTPConnection(host, port)
conn.request('GET', '/get')
conn.request("GET", "/get")
inside = conn.getresponse()
# The scrubbed field should be the same, because no cassette existed.
# Furthermore, the responses should be identical.
inside_body = json.loads(inside.read().decode('utf-8'))
outside_body = json.loads(outside.read().decode('utf-8'))
inside_body = json.loads(inside.read().decode("utf-8"))
outside_body = json.loads(outside.read().decode("utf-8"))
assert not inside_body[field_to_scrub] == replacement
assert inside_body[field_to_scrub] == outside_body[field_to_scrub]
# Ensure that when a cassette exists, the scrubbed response is returned.
with vcr.use_cassette(testfile, before_record_response=callback):
conn = httplib.HTTPConnection(host, port)
conn.request('GET', '/get')
conn.request("GET", "/get")
inside = conn.getresponse()
inside_body = json.loads(inside.read().decode('utf-8'))
inside_body = json.loads(inside.read().decode("utf-8"))
assert inside_body[field_to_scrub] == replacement

259
tests/integration/test_tornado.py
View File

@@ -1,5 +1,5 @@
# -*- coding: utf-8 -*-
'''Test requests' interaction with vcr'''
"""Test requests' interaction with vcr"""
import json
@@ -17,105 +17,99 @@ http = pytest.importorskip("tornado.httpclient")
supports_raise_error = tornado.version_info >= (4,)
@pytest.fixture(params=['simple', 'curl', 'default'])
@pytest.fixture(params=["simple", "curl", "default"])
def get_client(request):
if request.param == 'simple':
if request.param == "simple":
from tornado import simple_httpclient as simple
return (lambda: simple.SimpleAsyncHTTPClient())
elif request.param == 'curl':
return lambda: simple.SimpleAsyncHTTPClient()
elif request.param == "curl":
curl = pytest.importorskip("tornado.curl_httpclient")
return (lambda: curl.CurlAsyncHTTPClient())
return lambda: curl.CurlAsyncHTTPClient()
else:
return (lambda: http.AsyncHTTPClient())
return lambda: http.AsyncHTTPClient()
def get(client, url, **kwargs):
fetch_kwargs = {}
if supports_raise_error:
fetch_kwargs['raise_error'] = kwargs.pop('raise_error', True)
fetch_kwargs["raise_error"] = kwargs.pop("raise_error", True)
return client.fetch(
http.HTTPRequest(url, method='GET', **kwargs),
**fetch_kwargs
)
return client.fetch(http.HTTPRequest(url, method="GET", **kwargs), **fetch_kwargs)
def post(client, url, data=None, **kwargs):
if data:
kwargs['body'] = json.dumps(data)
return client.fetch(http.HTTPRequest(url, method='POST', **kwargs))
kwargs["body"] = json.dumps(data)
return client.fetch(http.HTTPRequest(url, method="POST", **kwargs))
@pytest.fixture(params=["https", "http"])
def scheme(request):
'''Fixture that returns both http and https.'''
"""Fixture that returns both http and https."""
return request.param
@pytest.mark.gen_test
def test_status_code(get_client, scheme, tmpdir):
'''Ensure that we can read the status code'''
url = scheme + '://httpbin.org/'
with vcr.use_cassette(str(tmpdir.join('atts.yaml'))):
"""Ensure that we can read the status code"""
url = scheme + "://httpbin.org/"
with vcr.use_cassette(str(tmpdir.join("atts.yaml"))):
status_code = (yield get(get_client(), url)).code
with vcr.use_cassette(str(tmpdir.join('atts.yaml'))) as cass:
with vcr.use_cassette(str(tmpdir.join("atts.yaml"))) as cass:
assert status_code == (yield get(get_client(), url)).code
assert 1 == cass.play_count
@pytest.mark.gen_test
def test_headers(get_client, scheme, tmpdir):
'''Ensure that we can read the headers back'''
url = scheme + '://httpbin.org/'
with vcr.use_cassette(str(tmpdir.join('headers.yaml'))):
"""Ensure that we can read the headers back"""
url = scheme + "://httpbin.org/"
with vcr.use_cassette(str(tmpdir.join("headers.yaml"))):
headers = (yield get(get_client(), url)).headers
with vcr.use_cassette(str(tmpdir.join('headers.yaml'))) as cass:
with vcr.use_cassette(str(tmpdir.join("headers.yaml"))) as cass:
assert headers == (yield get(get_client(), url)).headers
assert 1 == cass.play_count
@pytest.mark.gen_test
def test_body(get_client, tmpdir, scheme):
'''Ensure the responses are all identical enough'''
"""Ensure the responses are all identical enough"""
url = scheme + '://httpbin.org/bytes/1024'
with vcr.use_cassette(str(tmpdir.join('body.yaml'))):
url = scheme + "://httpbin.org/bytes/1024"
with vcr.use_cassette(str(tmpdir.join("body.yaml"))):
content = (yield get(get_client(), url)).body
with vcr.use_cassette(str(tmpdir.join('body.yaml'))) as cass:
with vcr.use_cassette(str(tmpdir.join("body.yaml"))) as cass:
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'''
url = scheme + '://httpbin.org/redirect-to?url=/html'
with vcr.use_cassette(str(tmpdir.join('url.yaml'))):
"""Ensure that the effective_url is captured"""
url = scheme + "://httpbin.org/redirect-to?url=/html"
with vcr.use_cassette(str(tmpdir.join("url.yaml"))):
effective_url = (yield get(get_client(), url)).effective_url
assert effective_url == scheme + '://httpbin.org/html'
assert effective_url == scheme + "://httpbin.org/html"
with vcr.use_cassette(str(tmpdir.join('url.yaml'))) as cass:
with vcr.use_cassette(str(tmpdir.join("url.yaml"))) as cass:
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'''
auth = ('user', 'passwd')
url = scheme + '://httpbin.org/basic-auth/user/passwd'
with vcr.use_cassette(str(tmpdir.join('auth.yaml'))):
one = yield get(
get_client(), url, auth_username=auth[0], auth_password=auth[1]
)
"""Ensure that we can handle basic auth"""
auth = ("user", "passwd")
url = scheme + "://httpbin.org/basic-auth/user/passwd"
with vcr.use_cassette(str(tmpdir.join("auth.yaml"))):
one = yield get(get_client(), url, auth_username=auth[0], auth_password=auth[1])
with vcr.use_cassette(str(tmpdir.join('auth.yaml'))) as cass:
two = yield get(
get_client(), url, auth_username=auth[0], auth_password=auth[1]
)
with vcr.use_cassette(str(tmpdir.join("auth.yaml"))) as cass:
two = yield get(get_client(), url, auth_username=auth[0], auth_password=auth[1])
assert one.body == two.body
assert one.code == two.code
assert 1 == cass.play_count
@@ -123,30 +117,20 @@ def test_auth(get_client, tmpdir, scheme):
@pytest.mark.gen_test
def test_auth_failed(get_client, tmpdir, scheme):
'''Ensure that we can save failed auth statuses'''
auth = ('user', 'wrongwrongwrong')
url = scheme + '://httpbin.org/basic-auth/user/passwd'
with vcr.use_cassette(str(tmpdir.join('auth-failed.yaml'))) as cass:
"""Ensure that we can save failed auth statuses"""
auth = ("user", "wrongwrongwrong")
url = scheme + "://httpbin.org/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)
with pytest.raises(http.HTTPError) as exc_info:
yield get(
get_client(),
url,
auth_username=auth[0],
auth_password=auth[1],
)
yield get(get_client(), url, auth_username=auth[0], auth_password=auth[1])
one = exc_info.value.response
assert exc_info.value.code == 401
with vcr.use_cassette(str(tmpdir.join('auth-failed.yaml'))) as cass:
with vcr.use_cassette(str(tmpdir.join("auth-failed.yaml"))) as cass:
with pytest.raises(http.HTTPError) as exc_info:
two = yield get(
get_client(),
url,
auth_username=auth[0],
auth_password=auth[1],
)
two = yield get(get_client(), url, auth_username=auth[0], auth_password=auth[1])
two = exc_info.value.response
assert exc_info.value.code == 401
assert one.body == two.body
@@ -156,13 +140,13 @@ def test_auth_failed(get_client, tmpdir, scheme):
@pytest.mark.gen_test
def test_post(get_client, tmpdir, scheme):
'''Ensure that we can post and cache the results'''
data = {'key1': 'value1', 'key2': 'value2'}
url = scheme + '://httpbin.org/post'
with vcr.use_cassette(str(tmpdir.join('requests.yaml'))):
"""Ensure that we can post and cache the results"""
data = {"key1": "value1", "key2": "value2"}
url = scheme + "://httpbin.org/post"
with vcr.use_cassette(str(tmpdir.join("requests.yaml"))):
req1 = (yield post(get_client(), url, data)).body
with vcr.use_cassette(str(tmpdir.join('requests.yaml'))) as cass:
with vcr.use_cassette(str(tmpdir.join("requests.yaml"))) as cass:
req2 = (yield post(get_client(), url, data)).body
assert req1 == req2
@@ -171,55 +155,55 @@ def test_post(get_client, tmpdir, scheme):
@pytest.mark.gen_test
def test_redirects(get_client, tmpdir, scheme):
'''Ensure that we can handle redirects'''
url = scheme + '://httpbin.org/redirect-to?url=bytes/1024'
with vcr.use_cassette(str(tmpdir.join('requests.yaml'))):
"""Ensure that we can handle redirects"""
url = scheme + "://httpbin.org/redirect-to?url=bytes/1024"
with vcr.use_cassette(str(tmpdir.join("requests.yaml"))):
content = (yield get(get_client(), url)).body
with vcr.use_cassette(str(tmpdir.join('requests.yaml'))) as cass:
with vcr.use_cassette(str(tmpdir.join("requests.yaml"))) as cass:
assert content == (yield get(get_client(), url)).body
assert cass.play_count == 1
@pytest.mark.gen_test
def test_cross_scheme(get_client, tmpdir, scheme):
'''Ensure that requests between schemes are treated separately'''
"""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:
yield get(get_client(), 'https://httpbin.org/')
yield get(get_client(), 'http://httpbin.org/')
with vcr.use_cassette(str(tmpdir.join("cross_scheme.yaml"))) as cass:
yield get(get_client(), "https://httpbin.org/")
yield get(get_client(), "http://httpbin.org/")
assert cass.play_count == 0
assert len(cass) == 2
# Then repeat the same requests and ensure both were replayed.
with vcr.use_cassette(str(tmpdir.join('cross_scheme.yaml'))) as cass:
yield get(get_client(), 'https://httpbin.org/')
yield get(get_client(), 'http://httpbin.org/')
with vcr.use_cassette(str(tmpdir.join("cross_scheme.yaml"))) as cass:
yield get(get_client(), "https://httpbin.org/")
yield get(get_client(), "http://httpbin.org/")
assert cass.play_count == 2
@pytest.mark.gen_test
def test_gzip(get_client, tmpdir, scheme):
'''
"""
Ensure that httpclient is able to automatically decompress the response
body
'''
url = scheme + '://httpbin.org/gzip'
"""
url = scheme + "://httpbin.org/gzip"
# use_gzip was renamed to decompress_response in 4.0
kwargs = {}
if tornado.version_info < (4,):
kwargs['use_gzip'] = True
kwargs["use_gzip"] = True
else:
kwargs['decompress_response'] = True
kwargs["decompress_response"] = True
with vcr.use_cassette(str(tmpdir.join('gzip.yaml'))):
with vcr.use_cassette(str(tmpdir.join("gzip.yaml"))):
response = yield get(get_client(), url, **kwargs)
assert_is_json(response.body)
with vcr.use_cassette(str(tmpdir.join('gzip.yaml'))) as cass:
with vcr.use_cassette(str(tmpdir.join("gzip.yaml"))) as cass:
response = yield get(get_client(), url, **kwargs)
assert_is_json(response.body)
assert 1 == cass.play_count
@@ -227,28 +211,26 @@ def test_gzip(get_client, tmpdir, scheme):
@pytest.mark.gen_test
def test_https_with_cert_validation_disabled(get_client, tmpdir):
cass_path = str(tmpdir.join('cert_validation_disabled.yaml'))
cass_path = str(tmpdir.join("cert_validation_disabled.yaml"))
with vcr.use_cassette(cass_path):
yield get(get_client(), 'https://httpbin.org', validate_cert=False)
yield get(get_client(), "https://httpbin.org", validate_cert=False)
with vcr.use_cassette(cass_path) as cass:
yield get(get_client(), 'https://httpbin.org', validate_cert=False)
yield get(get_client(), "https://httpbin.org", validate_cert=False)
assert 1 == cass.play_count
@pytest.mark.gen_test
def test_unsupported_features_raises_in_future(get_client, tmpdir):
'''Ensure that the exception for an AsyncHTTPClient feature not being
supported is raised inside the future.'''
"""Ensure that the exception for an AsyncHTTPClient feature not being
supported is raised inside the future."""
def callback(chunk):
assert False, "Did not expect to be called."
with vcr.use_cassette(str(tmpdir.join('invalid.yaml'))):
future = get(
get_client(), 'http://httpbin.org', streaming_callback=callback
)
with vcr.use_cassette(str(tmpdir.join("invalid.yaml"))):
future = get(get_client(), "http://httpbin.org", streaming_callback=callback)
with pytest.raises(Exception) as excinfo:
yield future
@@ -256,24 +238,18 @@ def test_unsupported_features_raises_in_future(get_client, tmpdir):
assert "not yet supported by VCR" in str(excinfo)
@pytest.mark.skipif(
not supports_raise_error,
reason='raise_error unavailable in tornado <= 3',
)
@pytest.mark.skipif(not supports_raise_error, reason="raise_error unavailable in tornado <= 3")
@pytest.mark.gen_test
def test_unsupported_features_raise_error_disabled(get_client, tmpdir):
'''Ensure that the exception for an AsyncHTTPClient feature not being
supported is not raised if raise_error=False.'''
"""Ensure that the exception for an AsyncHTTPClient feature not being
supported is not raised if raise_error=False."""
def callback(chunk):
assert False, "Did not expect to be called."
with vcr.use_cassette(str(tmpdir.join('invalid.yaml'))):
with vcr.use_cassette(str(tmpdir.join("invalid.yaml"))):
response = yield get(
get_client(),
'http://httpbin.org',
streaming_callback=callback,
raise_error=False,
get_client(), "http://httpbin.org", streaming_callback=callback, raise_error=False
)
assert "not yet supported by VCR" in str(response.error)
@@ -281,60 +257,51 @@ def test_unsupported_features_raise_error_disabled(get_client, tmpdir):
@pytest.mark.gen_test
def test_cannot_overwrite_cassette_raises_in_future(get_client, tmpdir):
'''Ensure that CannotOverwriteExistingCassetteException is raised inside
the future.'''
"""Ensure that CannotOverwriteExistingCassetteException is raised inside
the future."""
with vcr.use_cassette(str(tmpdir.join('overwrite.yaml'))):
yield get(get_client(), 'http://httpbin.org/get')
with vcr.use_cassette(str(tmpdir.join("overwrite.yaml"))):
yield get(get_client(), "http://httpbin.org/get")
with vcr.use_cassette(str(tmpdir.join('overwrite.yaml'))):
future = get(get_client(), 'http://httpbin.org/headers')
with vcr.use_cassette(str(tmpdir.join("overwrite.yaml"))):
future = get(get_client(), "http://httpbin.org/headers")
with pytest.raises(CannotOverwriteExistingCassetteException):
yield future
@pytest.mark.skipif(
not supports_raise_error,
reason='raise_error unavailable in tornado <= 3',
)
@pytest.mark.skipif(not supports_raise_error, reason="raise_error unavailable in tornado <= 3")
@pytest.mark.gen_test
def test_cannot_overwrite_cassette_raise_error_disabled(get_client, tmpdir):
'''Ensure that CannotOverwriteExistingCassetteException is not raised if
raise_error=False in the fetch() call.'''
"""Ensure that CannotOverwriteExistingCassetteException is not raised if
raise_error=False in the fetch() call."""
with vcr.use_cassette(str(tmpdir.join('overwrite.yaml'))):
yield get(
get_client(), 'http://httpbin.org/get', raise_error=False
)
with vcr.use_cassette(str(tmpdir.join("overwrite.yaml"))):
yield get(get_client(), "http://httpbin.org/get", raise_error=False)
with vcr.use_cassette(str(tmpdir.join('overwrite.yaml'))):
response = yield get(
get_client(), 'http://httpbin.org/headers', raise_error=False
)
with vcr.use_cassette(str(tmpdir.join("overwrite.yaml"))):
response = yield get(get_client(), "http://httpbin.org/headers", raise_error=False)
assert isinstance(response.error, CannotOverwriteExistingCassetteException)
@pytest.mark.gen_test
@vcr.use_cassette(path_transformer=vcr.default_vcr.ensure_suffix('.yaml'))
@vcr.use_cassette(path_transformer=vcr.default_vcr.ensure_suffix(".yaml"))
def test_tornado_with_decorator_use_cassette(get_client):
response = yield get_client().fetch(
http.HTTPRequest('http://www.google.com/', method='GET')
)
assert response.body.decode('utf-8') == "not actually google"
response = yield get_client().fetch(http.HTTPRequest("http://www.google.com/", method="GET"))
assert response.body.decode("utf-8") == "not actually google"
@pytest.mark.gen_test
@vcr.use_cassette(path_transformer=vcr.default_vcr.ensure_suffix('.yaml'))
@vcr.use_cassette(path_transformer=vcr.default_vcr.ensure_suffix(".yaml"))
def test_tornado_exception_can_be_caught(get_client):
try:
yield get(get_client(), 'http://httpbin.org/status/500')
yield get(get_client(), "http://httpbin.org/status/500")
except http.HTTPError as e:
assert e.code == 500
try:
yield get(get_client(), 'http://httpbin.org/status/404')
yield get(get_client(), "http://httpbin.org/status/404")
except http.HTTPError as e:
assert e.code == 404
@@ -343,41 +310,41 @@ def test_tornado_exception_can_be_caught(get_client):
def test_existing_references_get_patched(tmpdir):
from tornado.httpclient import AsyncHTTPClient
with vcr.use_cassette(str(tmpdir.join('data.yaml'))):
with vcr.use_cassette(str(tmpdir.join("data.yaml"))):
client = AsyncHTTPClient()
yield get(client, 'http://httpbin.org/get')
yield get(client, "http://httpbin.org/get")
with vcr.use_cassette(str(tmpdir.join('data.yaml'))) as cass:
yield get(client, 'http://httpbin.org/get')
with vcr.use_cassette(str(tmpdir.join("data.yaml"))) as cass:
yield get(client, "http://httpbin.org/get")
assert cass.play_count == 1
@pytest.mark.gen_test
def test_existing_instances_get_patched(get_client, tmpdir):
'''Ensure that existing instances of AsyncHTTPClient get patched upon
entering VCR context.'''
"""Ensure that existing instances of AsyncHTTPClient get patched upon
entering VCR context."""
client = get_client()
with vcr.use_cassette(str(tmpdir.join('data.yaml'))):
yield get(client, 'http://httpbin.org/get')
with vcr.use_cassette(str(tmpdir.join("data.yaml"))):
yield get(client, "http://httpbin.org/get")
with vcr.use_cassette(str(tmpdir.join('data.yaml'))) as cass:
yield get(client, 'http://httpbin.org/get')
with vcr.use_cassette(str(tmpdir.join("data.yaml"))) as cass:
yield get(client, "http://httpbin.org/get")
assert cass.play_count == 1
@pytest.mark.gen_test
def test_request_time_is_set(get_client, tmpdir):
'''Ensures that the request_time on HTTPResponses is set.'''
"""Ensures that the request_time on HTTPResponses is set."""
with vcr.use_cassette(str(tmpdir.join('data.yaml'))):
with vcr.use_cassette(str(tmpdir.join("data.yaml"))):
client = get_client()
response = yield get(client, 'http://httpbin.org/get')
response = yield get(client, "http://httpbin.org/get")
assert response.request_time is not None
with vcr.use_cassette(str(tmpdir.join('data.yaml'))) as cass:
with vcr.use_cassette(str(tmpdir.join("data.yaml"))) as cass:
client = get_client()
response = yield get(client, 'http://httpbin.org/get')
response = yield get(client, "http://httpbin.org/get")
assert response.request_time is not None
assert cass.play_count == 1

87
tests/integration/test_urllib2.py
View File

@@ -1,5 +1,5 @@
# -*- coding: utf-8 -*-
'''Integration tests with urllib2'''
"""Integration tests with urllib2"""
import ssl
from six.moves.urllib.request import urlopen
@@ -15,91 +15,86 @@ from assertions import assert_cassette_has_one_response
def urlopen_with_cafile(*args, **kwargs):
context = ssl.create_default_context(cafile=pytest_httpbin.certs.where())
context.check_hostname = False
kwargs['context'] = context
kwargs["context"] = context
try:
return urlopen(*args, **kwargs)
except TypeError:
# python2/pypi don't let us override this
del kwargs['cafile']
del kwargs["cafile"]
return urlopen(*args, **kwargs)
def test_response_code(httpbin_both, tmpdir):
'''Ensure we can read a response code from a fetch'''
"""Ensure we can read a response code from a fetch"""
url = httpbin_both.url
with vcr.use_cassette(str(tmpdir.join('atts.yaml'))):
with vcr.use_cassette(str(tmpdir.join("atts.yaml"))):
code = urlopen_with_cafile(url).getcode()
with vcr.use_cassette(str(tmpdir.join('atts.yaml'))):
with vcr.use_cassette(str(tmpdir.join("atts.yaml"))):
assert code == urlopen_with_cafile(url).getcode()
def test_random_body(httpbin_both, tmpdir):
'''Ensure we can read the content, and that it's served from cache'''
url = httpbin_both.url + '/bytes/1024'
with vcr.use_cassette(str(tmpdir.join('body.yaml'))):
"""Ensure we can read the content, and that it's served from cache"""
url = httpbin_both.url + "/bytes/1024"
with vcr.use_cassette(str(tmpdir.join("body.yaml"))):
body = urlopen_with_cafile(url).read()
with vcr.use_cassette(str(tmpdir.join('body.yaml'))):
with vcr.use_cassette(str(tmpdir.join("body.yaml"))):
assert body == urlopen_with_cafile(url).read()
def test_response_headers(httpbin_both, tmpdir):
'''Ensure we can get information from the response'''
"""Ensure we can get information from the response"""
url = httpbin_both.url
with vcr.use_cassette(str(tmpdir.join('headers.yaml'))):
with vcr.use_cassette(str(tmpdir.join("headers.yaml"))):
open1 = urlopen_with_cafile(url).info().items()
with vcr.use_cassette(str(tmpdir.join('headers.yaml'))):
with vcr.use_cassette(str(tmpdir.join("headers.yaml"))):
open2 = urlopen_with_cafile(url).info().items()
assert sorted(open1) == sorted(open2)
def test_effective_url(httpbin_both, tmpdir):
'''Ensure that the effective_url is captured'''
url = httpbin_both.url + '/redirect-to?url=/html'
with vcr.use_cassette(str(tmpdir.join('headers.yaml'))):
"""Ensure that the effective_url is captured"""
url = httpbin_both.url + "/redirect-to?url=/html"
with vcr.use_cassette(str(tmpdir.join("headers.yaml"))):
effective_url = urlopen_with_cafile(url).geturl()
assert effective_url == httpbin_both.url + '/html'
assert effective_url == httpbin_both.url + "/html"
with vcr.use_cassette(str(tmpdir.join('headers.yaml'))):
with vcr.use_cassette(str(tmpdir.join("headers.yaml"))):
assert effective_url == urlopen_with_cafile(url).geturl()
def test_multiple_requests(httpbin_both, tmpdir):
'''Ensure that we can cache multiple requests'''
urls = [
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:
"""Ensure that we can cache multiple requests"""
urls = [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_with_cafile(url) for url in urls]
assert len(cass) == len(urls)
def test_get_data(httpbin_both, tmpdir):
'''Ensure that it works with query data'''
data = urlencode({'some': 1, 'data': 'here'})
url = httpbin_both.url + '/get?' + data
with vcr.use_cassette(str(tmpdir.join('get_data.yaml'))):
"""Ensure that it works with query data"""
data = urlencode({"some": 1, "data": "here"})
url = httpbin_both.url + "/get?" + data
with vcr.use_cassette(str(tmpdir.join("get_data.yaml"))):
res1 = urlopen_with_cafile(url).read()
with vcr.use_cassette(str(tmpdir.join('get_data.yaml'))):
with vcr.use_cassette(str(tmpdir.join("get_data.yaml"))):
res2 = urlopen_with_cafile(url).read()
assert res1 == res2
def test_post_data(httpbin_both, tmpdir):
'''Ensure that it works when posting data'''
data = urlencode({'some': 1, 'data': 'here'}).encode('utf-8')
url = httpbin_both.url + '/post'
with vcr.use_cassette(str(tmpdir.join('post_data.yaml'))):
"""Ensure that it works when posting data"""
data = urlencode({"some": 1, "data": "here"}).encode("utf-8")
url = httpbin_both.url + "/post"
with vcr.use_cassette(str(tmpdir.join("post_data.yaml"))):
res1 = urlopen_with_cafile(url, data).read()
with vcr.use_cassette(str(tmpdir.join('post_data.yaml'))) as cass:
with vcr.use_cassette(str(tmpdir.join("post_data.yaml"))) as cass:
res2 = urlopen_with_cafile(url, data).read()
assert len(cass) == 1
@@ -108,13 +103,13 @@ def test_post_data(httpbin_both, 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 = httpbin_both.url + '/post'
with vcr.use_cassette(str(tmpdir.join('post_data.yaml'))):
"""Ensure that it works when posting unicode data"""
data = urlencode({"snowman": u"☃".encode("utf-8")}).encode("utf-8")
url = httpbin_both.url + "/post"
with vcr.use_cassette(str(tmpdir.join("post_data.yaml"))):
res1 = urlopen_with_cafile(url, data).read()
with vcr.use_cassette(str(tmpdir.join('post_data.yaml'))) as cass:
with vcr.use_cassette(str(tmpdir.join("post_data.yaml"))) as cass:
res2 = urlopen_with_cafile(url, data).read()
assert len(cass) == 1
@@ -123,11 +118,11 @@ def test_post_unicode_data(httpbin_both, tmpdir):
def test_cross_scheme(tmpdir, httpbin_secure, httpbin):
'''Ensure that requests between schemes are treated separately'''
"""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:
with vcr.use_cassette(str(tmpdir.join("cross_scheme.yaml"))) as cass:
urlopen_with_cafile(httpbin_secure.url)
urlopen_with_cafile(httpbin.url)
assert len(cass) == 2
@@ -135,14 +130,14 @@ def test_cross_scheme(tmpdir, httpbin_secure, httpbin):
def test_decorator(httpbin_both, tmpdir):
'''Test the decorator version of VCR.py'''
"""Test the decorator version of VCR.py"""
url = httpbin_both.url
@vcr.use_cassette(str(tmpdir.join('atts.yaml')))
@vcr.use_cassette(str(tmpdir.join("atts.yaml")))
def inner1():
return urlopen_with_cafile(url).getcode()
@vcr.use_cassette(str(tmpdir.join('atts.yaml')))
@vcr.use_cassette(str(tmpdir.join("atts.yaml")))
def inner2():
return urlopen_with_cafile(url).getcode()

128
tests/integration/test_urllib3.py
View File

@@ -1,4 +1,4 @@
'''Integration tests with urllib3'''
"""Integration tests with urllib3"""
# coding=utf-8
@@ -7,101 +7,101 @@ import pytest_httpbin
import vcr
from vcr.patch import force_reset
from assertions import assert_cassette_empty, assert_is_json
urllib3 = pytest.importorskip("urllib3")
@pytest.fixture(scope='module')
@pytest.fixture(scope="module")
def verify_pool_mgr():
return urllib3.PoolManager(
cert_reqs='CERT_REQUIRED', # Force certificate check.
ca_certs=pytest_httpbin.certs.where()
cert_reqs="CERT_REQUIRED", ca_certs=pytest_httpbin.certs.where() # Force certificate check.
)
@pytest.fixture(scope='module')
@pytest.fixture(scope="module")
def pool_mgr():
return urllib3.PoolManager(cert_reqs='CERT_NONE')
return urllib3.PoolManager(cert_reqs="CERT_NONE")
def test_status_code(httpbin_both, tmpdir, verify_pool_mgr):
'''Ensure that we can read the status code'''
"""Ensure that we can read the status code"""
url = httpbin_both.url
with vcr.use_cassette(str(tmpdir.join('atts.yaml'))):
status_code = verify_pool_mgr.request('GET', url).status
with vcr.use_cassette(str(tmpdir.join("atts.yaml"))):
status_code = verify_pool_mgr.request("GET", url).status
with vcr.use_cassette(str(tmpdir.join('atts.yaml'))):
assert status_code == verify_pool_mgr.request('GET', url).status
with vcr.use_cassette(str(tmpdir.join("atts.yaml"))):
assert status_code == verify_pool_mgr.request("GET", url).status
def test_headers(tmpdir, httpbin_both, verify_pool_mgr):
'''Ensure that we can read the headers back'''
"""Ensure that we can read the headers back"""
url = httpbin_both.url
with vcr.use_cassette(str(tmpdir.join('headers.yaml'))):
headers = verify_pool_mgr.request('GET', url).headers
with vcr.use_cassette(str(tmpdir.join("headers.yaml"))):
headers = verify_pool_mgr.request("GET", url).headers
with vcr.use_cassette(str(tmpdir.join('headers.yaml'))):
assert headers == verify_pool_mgr.request('GET', url).headers
with vcr.use_cassette(str(tmpdir.join("headers.yaml"))):
assert headers == verify_pool_mgr.request("GET", url).headers
def test_body(tmpdir, httpbin_both, verify_pool_mgr):
'''Ensure the responses are all identical enough'''
url = httpbin_both.url + '/bytes/1024'
with vcr.use_cassette(str(tmpdir.join('body.yaml'))):
content = verify_pool_mgr.request('GET', url).data
"""Ensure the responses are all identical enough"""
url = httpbin_both.url + "/bytes/1024"
with vcr.use_cassette(str(tmpdir.join("body.yaml"))):
content = verify_pool_mgr.request("GET", url).data
with vcr.use_cassette(str(tmpdir.join('body.yaml'))):
assert content == verify_pool_mgr.request('GET', url).data
with vcr.use_cassette(str(tmpdir.join("body.yaml"))):
assert content == verify_pool_mgr.request("GET", url).data
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='{}:{}'.format(*auth))
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)
"""Ensure that we can handle basic auth"""
auth = ("user", "passwd")
headers = urllib3.util.make_headers(basic_auth="{}:{}".format(*auth))
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)
with vcr.use_cassette(str(tmpdir.join('auth.yaml'))):
two = verify_pool_mgr.request('GET', url, headers=headers)
with vcr.use_cassette(str(tmpdir.join("auth.yaml"))):
two = verify_pool_mgr.request("GET", url, headers=headers)
assert one.data == two.data
assert one.status == two.status
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='{}:{}'.format(*auth))
url = httpbin_both.url + '/basic-auth/user/passwd'
with vcr.use_cassette(str(tmpdir.join('auth-failed.yaml'))) as cass:
"""Ensure that we can save failed auth statuses"""
auth = ("user", "wrongwrongwrong")
headers = urllib3.util.make_headers(basic_auth="{}:{}".format(*auth))
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)
one = verify_pool_mgr.request('GET', url, headers=headers)
two = verify_pool_mgr.request('GET', url, headers=headers)
one = verify_pool_mgr.request("GET", url, headers=headers)
two = verify_pool_mgr.request("GET", url, headers=headers)
assert one.data == two.data
assert one.status == two.status == 401
def test_post(tmpdir, httpbin_both, verify_pool_mgr):
'''Ensure that we can post and cache the results'''
data = {'key1': 'value1', 'key2': 'value2'}
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
"""Ensure that we can post and cache the results"""
data = {"key1": "value1", "key2": "value2"}
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
with vcr.use_cassette(str(tmpdir.join('verify_pool_mgr.yaml'))):
req2 = verify_pool_mgr.request('POST', url, data).data
with vcr.use_cassette(str(tmpdir.join("verify_pool_mgr.yaml"))):
req2 = verify_pool_mgr.request("POST", url, data).data
assert req1 == req2
def test_redirects(tmpdir, httpbin_both, verify_pool_mgr):
'''Ensure that we can handle redirects'''
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
"""Ensure that we can handle redirects"""
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
with vcr.use_cassette(str(tmpdir.join('verify_pool_mgr.yaml'))) as cass:
assert content == verify_pool_mgr.request('GET', url).data
with vcr.use_cassette(str(tmpdir.join("verify_pool_mgr.yaml"))) as cass:
assert content == verify_pool_mgr.request("GET", url).data
# Ensure that we've now cached *two* responses. One for the redirect
# and one for the final fetch
assert len(cass) == 2
@@ -109,36 +109,36 @@ def test_redirects(tmpdir, httpbin_both, verify_pool_mgr):
def test_cross_scheme(tmpdir, httpbin, httpbin_secure, verify_pool_mgr):
'''Ensure that requests between schemes are treated separately'''
"""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', httpbin_secure.url)
verify_pool_mgr.request('GET', httpbin.url)
with vcr.use_cassette(str(tmpdir.join("cross_scheme.yaml"))) as cass:
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, httpbin_both, verify_pool_mgr):
'''
"""
Ensure that requests (actually urllib3) is able to automatically decompress
the response body
'''
url = httpbin_both.url + '/gzip'
response = verify_pool_mgr.request('GET', url)
"""
url = httpbin_both.url + "/gzip"
response = verify_pool_mgr.request("GET", url)
with vcr.use_cassette(str(tmpdir.join('gzip.yaml'))):
response = verify_pool_mgr.request('GET', url)
with vcr.use_cassette(str(tmpdir.join("gzip.yaml"))):
response = verify_pool_mgr.request("GET", url)
assert_is_json(response.data)
with vcr.use_cassette(str(tmpdir.join('gzip.yaml'))):
with vcr.use_cassette(str(tmpdir.join("gzip.yaml"))):
assert_is_json(response.data)
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', httpbin_secure.url)
with vcr.use_cassette(str(tmpdir.join("cert_validation_disabled.yaml"))):
pool_mgr.request("GET", httpbin_secure.url)
def test_urllib3_force_reset():
@@ -146,7 +146,7 @@ def test_urllib3_force_reset():
http_original = cpool.HTTPConnection
https_original = cpool.HTTPSConnection
verified_https_original = cpool.VerifiedHTTPSConnection
with vcr.use_cassette(path='test'):
with vcr.use_cassette(path="test"):
first_cassette_HTTPConnection = cpool.HTTPConnection
first_cassette_HTTPSConnection = cpool.HTTPSConnection
first_cassette_VerifiedHTTPSConnection = cpool.VerifiedHTTPSConnection

43
tests/integration/test_wild.py
View File

@@ -13,13 +13,13 @@ except ImportError:
def test_domain_redirect():
'''Ensure that redirects across domains are considered unique'''
"""Ensure that redirects across domains are considered unique"""
# In this example, seomoz.org redirects to moz.com, and if those
# requests are considered identical, then we'll be stuck in a redirect
# loop.
url = 'http://seomoz.org/'
with vcr.use_cassette('tests/fixtures/wild/domain_redirect.yaml') as cass:
requests.get(url, headers={'User-Agent': 'vcrpy-test'})
url = "http://seomoz.org/"
with vcr.use_cassette("tests/fixtures/wild/domain_redirect.yaml") as cass:
requests.get(url, headers={"User-Agent": "vcrpy-test"})
# Ensure that we've now served two responses. One for the original
# redirect, and a second for the actual fetch
assert len(cass) == 2
@@ -30,13 +30,11 @@ 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.host, httpbin.port)
headers = {
"Content-Type": content_type,
"content-length": str(len(body))
}
headers = {"Content-Type": content_type, "content-length": str(len(body))}
h.request("POST", "/post/", headers=headers)
h.send(body)
r = h.getresponse()
@@ -45,7 +43,7 @@ def test_flickr_multipart_upload(httpbin, tmpdir):
return data
testfile = str(tmpdir.join('flickr.yml'))
testfile = str(tmpdir.join("flickr.yml"))
with vcr.use_cassette(testfile) as cass:
_pretend_to_be_flickr_library()
assert len(cass) == 1
@@ -57,44 +55,41 @@ def test_flickr_multipart_upload(httpbin, tmpdir):
def test_flickr_should_respond_with_200(tmpdir):
testfile = str(tmpdir.join('flickr.yml'))
testfile = str(tmpdir.join("flickr.yml"))
with vcr.use_cassette(testfile):
r = requests.post("https://api.flickr.com/services/upload", verify=False)
assert r.status_code == 200
def test_cookies(tmpdir, httpbin):
testfile = str(tmpdir.join('cookies.yml'))
testfile = str(tmpdir.join("cookies.yml"))
with vcr.use_cassette(testfile):
s = requests.Session()
s.get(httpbin.url + "/cookies/set?k1=v1&k2=v2")
r2 = s.get(httpbin.url + "/cookies")
assert len(r2.json()['cookies']) == 2
assert len(r2.json()["cookies"]) == 2
def test_amazon_doctype(tmpdir):
# amazon gzips its homepage. For some reason, in requests 2.7, it's not
# getting gunzipped.
with vcr.use_cassette(str(tmpdir.join('amz.yml'))):
r = requests.get('http://www.amazon.com', verify=False)
assert 'html' in r.text
with vcr.use_cassette(str(tmpdir.join("amz.yml"))):
r = requests.get("http://www.amazon.com", verify=False)
assert "html" in r.text
def start_rpc_server(q):
httpd = xmlrpc_server.SimpleXMLRPCServer(('127.0.0.1', 0))
httpd = xmlrpc_server.SimpleXMLRPCServer(("127.0.0.1", 0))
httpd.register_function(pow)
q.put('http://{}:{}'.format(*httpd.server_address))
q.put("http://{}:{}".format(*httpd.server_address))
httpd.serve_forever()
@pytest.yield_fixture(scope='session')
@pytest.yield_fixture(scope="session")
def rpc_server():
q = multiprocessing.Queue()
proxy_process = multiprocessing.Process(
target=start_rpc_server,
args=(q,)
)
proxy_process = multiprocessing.Process(target=start_rpc_server, args=(q,))
try:
proxy_process.start()
yield q.get()
@@ -103,11 +98,11 @@ def rpc_server():
def test_xmlrpclib(tmpdir, rpc_server):
with vcr.use_cassette(str(tmpdir.join('xmlrpcvideo.yaml'))):
with vcr.use_cassette(str(tmpdir.join("xmlrpcvideo.yaml"))):
roundup_server = xmlrpc_client.ServerProxy(rpc_server, allow_none=True)
original_schema = roundup_server.pow(2, 4)
with vcr.use_cassette(str(tmpdir.join('xmlrpcvideo.yaml'))):
with vcr.use_cassette(str(tmpdir.join("xmlrpcvideo.yaml"))):
roundup_server = xmlrpc_client.ServerProxy(rpc_server, allow_none=True)
second_schema = roundup_server.pow(2, 4)