mirror of
https://github.com/kevin1024/vcrpy.git
synced 2025-12-08 16:53:23 +00:00
8c6b1fdf38
.. with regard to: - not crashing with decode_compressed_response==True - expected cassette content for body string - expected response content, i.e. proper decompression
316 lines
12 KiB
Python
316 lines
12 KiB
Python
"""Test requests' interaction with vcr"""
|
|
import pytest
|
|
from assertions import assert_cassette_empty, assert_is_json
|
|
|
|
import vcr
|
|
|
|
requests = pytest.importorskip("requests")
|
|
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"))):
|
|
status_code = requests.get(url).status_code
|
|
|
|
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"))):
|
|
headers = requests.get(url).headers
|
|
|
|
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"))):
|
|
content = requests.get(url).content
|
|
|
|
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"}
|
|
|
|
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",)):
|
|
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"))):
|
|
effective_url = requests.get(url).url
|
|
assert effective_url == httpbin_both.url + "/html"
|
|
|
|
with vcr.use_cassette(str(tmpdir.join("url.yaml"))):
|
|
assert effective_url == requests.get(url).url
|
|
|
|
|
|
def test_auth(tmpdir, 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"))):
|
|
one = requests.get(url, auth=auth)
|
|
|
|
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 this is empty to begin with
|
|
assert_cassette_empty(cass)
|
|
one = requests.get(url, auth=auth)
|
|
two = requests.get(url, auth=auth)
|
|
assert one.content == two.content
|
|
assert one.status_code == two.status_code == 401
|
|
|
|
|
|
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"))):
|
|
req1 = requests.post(url, data).content
|
|
|
|
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"))):
|
|
req1 = requests.post(url, data1).content
|
|
|
|
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)
|
|
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"))):
|
|
req1 = requests.post(url, data1).content
|
|
print(req1)
|
|
|
|
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"))):
|
|
content = requests.get(url).content
|
|
|
|
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
|
|
assert len(cass) == 2
|
|
assert cass.play_count == 2
|
|
|
|
|
|
def test_cross_scheme(tmpdir, httpbin_secure, httpbin):
|
|
"""Ensure that requests between schemes are treated separately"""
|
|
# First fetch a url under http, and then again under https and then
|
|
# ensure that we haven't served anything out of cache, and we have two
|
|
# requests / response pairs in the cassette
|
|
with vcr.use_cassette(str(tmpdir.join("cross_scheme.yaml"))) as cass:
|
|
requests.get(httpbin_secure + "/")
|
|
requests.get(httpbin + "/")
|
|
assert cass.play_count == 0
|
|
assert len(cass) == 2
|
|
|
|
|
|
def test_gzip__decode_compressed_response_false(tmpdir, httpbin_both):
|
|
"""
|
|
Ensure that requests (actually urllib3) is able to automatically decompress
|
|
the response body
|
|
"""
|
|
for _ in range(2): # one for recording, one for re-playing
|
|
with vcr.use_cassette(str(tmpdir.join("gzip.yaml"))):
|
|
response = requests.get(httpbin_both + "/gzip")
|
|
assert response.headers["content-encoding"] == "gzip" # i.e. not removed
|
|
assert_is_json(response.content) # i.e. uncompressed bytes
|
|
|
|
|
|
def test_gzip__decode_compressed_response_true(tmpdir, httpbin_both):
|
|
url = httpbin_both + "/gzip"
|
|
|
|
expected_response = requests.get(url)
|
|
expected_content = expected_response.content
|
|
assert expected_response.headers["content-encoding"] == "gzip" # self-test
|
|
|
|
with vcr.use_cassette(
|
|
str(tmpdir.join("decode_compressed.yaml")), decode_compressed_response=True
|
|
) as cassette:
|
|
r = requests.get(url)
|
|
assert r.headers["content-encoding"] == "gzip" # i.e. not removed
|
|
assert r.content == expected_content
|
|
|
|
# Has the cassette body been decompressed?
|
|
cassette_response_body = cassette.responses[0]["body"]["string"]
|
|
assert isinstance(cassette_response_body, str)
|
|
|
|
with vcr.use_cassette(str(tmpdir.join("decode_compressed.yaml")), decode_compressed_response=True):
|
|
r = requests.get(url)
|
|
assert "content-encoding" not in r.headers # i.e. removed
|
|
assert r.content == expected_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"))):
|
|
session = requests.session()
|
|
|
|
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"))):
|
|
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"))):
|
|
session = requests.session()
|
|
session.get(httpbin + "/get")
|
|
|
|
with vcr.use_cassette(str(tmpdir.join("test_session_after_unpatched.yaml"))):
|
|
session = requests.session()
|
|
session.get(httpbin + "/get")
|
|
|
|
session.get(httpbin + "/status/200")
|
|
|
|
|
|
def test_session_created_before_use_cassette_is_patched(tmpdir, httpbin_both):
|
|
url = httpbin_both + "/bytes/1024"
|
|
# Record arbitrary, random data to the cassette
|
|
with vcr.use_cassette(str(tmpdir.join("session_created_outside.yaml"))):
|
|
session = requests.session()
|
|
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")
|
|
|
|
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"))):
|
|
session = requests.session()
|
|
first_body = session.get(url).content
|
|
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"))):
|
|
session = requests.session()
|
|
assert session.get(url).content == second_body
|
|
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
|
|
|
|
|
|
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:
|
|
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:
|
|
tox_content = f.read()
|
|
assert cass.requests[0].body.read() == tox_content
|
|
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
|
|
|
|
|
|
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"
|
|
|
|
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",)):
|
|
req2 = requests.post(url, data).content
|
|
|
|
assert req1 == req2
|