mirror of
https://github.com/kevin1024/vcrpy.git
synced 2025-12-09 17:15:35 +00:00
Drops Python 3.9 support
This commit is contained in:
@@ -264,12 +264,6 @@ def test_aiohttp_test_client_json(aiohttp_client, tmpdir):
|
||||
assert cassette.play_count == 1
|
||||
|
||||
|
||||
def test_cleanup_from_pytest_asyncio():
|
||||
# work around https://github.com/pytest-dev/pytest-asyncio/issues/724
|
||||
asyncio.get_event_loop().close()
|
||||
asyncio.set_event_loop(None)
|
||||
|
||||
|
||||
@pytest.mark.online
|
||||
def test_redirect(tmpdir, httpbin):
|
||||
url = httpbin.url + "/redirect/2"
|
||||
|
||||
@@ -42,17 +42,13 @@ def scheme(request):
|
||||
return request.param
|
||||
|
||||
|
||||
@pytest.fixture(params=["simple", "curl", "default"])
|
||||
@pytest.fixture(params=["curl", "default"])
|
||||
def get_client(request):
|
||||
if request.param == "simple":
|
||||
from tornado import simple_httpclient as simple
|
||||
|
||||
return lambda: simple.SimpleAsyncHTTPClient()
|
||||
elif request.param == "curl":
|
||||
if request.param == "curl":
|
||||
curl = pytest.importorskip("tornado.curl_httpclient")
|
||||
return lambda: curl.CurlAsyncHTTPClient()
|
||||
else:
|
||||
return lambda: http.AsyncHTTPClient()
|
||||
|
||||
return lambda: http.AsyncHTTPClient()
|
||||
|
||||
|
||||
def get(client, url, **kwargs):
|
||||
@@ -192,7 +188,7 @@ def test_redirects(get_client, tmpdir, httpbin):
|
||||
|
||||
@pytest.mark.online
|
||||
@gen_test
|
||||
def test_cross_scheme(get_client, tmpdir, scheme):
|
||||
def test_cross_scheme(get_client, tmpdir):
|
||||
"""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
|
||||
|
||||
@@ -1,147 +0,0 @@
|
||||
"""Integration tests with urllib2"""
|
||||
|
||||
import ssl
|
||||
from urllib.parse import urlencode
|
||||
from urllib.request import urlopen
|
||||
|
||||
import pytest_httpbin.certs
|
||||
from pytest import mark
|
||||
|
||||
# Internal imports
|
||||
import vcr
|
||||
|
||||
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
|
||||
try:
|
||||
return urlopen(*args, **kwargs)
|
||||
except TypeError:
|
||||
# python2/pypi don't let us override this
|
||||
del kwargs["cafile"]
|
||||
return urlopen(*args, **kwargs)
|
||||
|
||||
|
||||
def test_response_code(httpbin_both, tmpdir):
|
||||
"""Ensure we can read a response code from a fetch"""
|
||||
url = httpbin_both.url
|
||||
with vcr.use_cassette(str(tmpdir.join("atts.yaml"))):
|
||||
code = urlopen_with_cafile(url).getcode()
|
||||
|
||||
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"))):
|
||||
body = urlopen_with_cafile(url).read()
|
||||
|
||||
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"""
|
||||
url = httpbin_both.url
|
||||
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"))):
|
||||
open2 = urlopen_with_cafile(url).info().items()
|
||||
|
||||
assert sorted(open1) == sorted(open2)
|
||||
|
||||
|
||||
@mark.online
|
||||
def test_effective_url(tmpdir, httpbin):
|
||||
"""Ensure that the effective_url is captured"""
|
||||
url = httpbin.url + "/redirect-to?url=.%2F&status_code=301"
|
||||
|
||||
with vcr.use_cassette(str(tmpdir.join("headers.yaml"))):
|
||||
effective_url = urlopen_with_cafile(url).geturl()
|
||||
assert effective_url == httpbin.url + "/"
|
||||
|
||||
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:
|
||||
[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"))):
|
||||
res1 = urlopen_with_cafile(url).read()
|
||||
|
||||
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"))):
|
||||
res1 = urlopen_with_cafile(url, data).read()
|
||||
|
||||
with vcr.use_cassette(str(tmpdir.join("post_data.yaml"))) as cass:
|
||||
res2 = urlopen_with_cafile(url, data).read()
|
||||
assert len(cass) == 1
|
||||
|
||||
assert res1 == res2
|
||||
assert_cassette_has_one_response(cass)
|
||||
|
||||
|
||||
def test_post_unicode_data(httpbin_both, tmpdir):
|
||||
"""Ensure that it works when posting unicode data"""
|
||||
data = urlencode({"snowman": "☃".encode()}).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:
|
||||
res2 = urlopen_with_cafile(url, data).read()
|
||||
assert len(cass) == 1
|
||||
|
||||
assert res1 == res2
|
||||
assert_cassette_has_one_response(cass)
|
||||
|
||||
|
||||
def test_cross_scheme(tmpdir, httpbin_secure, httpbin):
|
||||
"""Ensure that requests between schemes are treated separately"""
|
||||
# First fetch a url under https, and then again under https and then
|
||||
# ensure that we haven't served anything out of cache, and we have two
|
||||
# requests / response pairs in the cassette
|
||||
with vcr.use_cassette(str(tmpdir.join("cross_scheme.yaml"))) as cass:
|
||||
urlopen_with_cafile(httpbin_secure.url)
|
||||
urlopen_with_cafile(httpbin.url)
|
||||
assert len(cass) == 2
|
||||
assert cass.play_count == 0
|
||||
|
||||
|
||||
def test_decorator(httpbin_both, tmpdir):
|
||||
"""Test the decorator version of VCR.py"""
|
||||
url = httpbin_both.url
|
||||
|
||||
@vcr.use_cassette(str(tmpdir.join("atts.yaml")))
|
||||
def inner1():
|
||||
return urlopen_with_cafile(url).getcode()
|
||||
|
||||
@vcr.use_cassette(str(tmpdir.join("atts.yaml")))
|
||||
def inner2():
|
||||
return urlopen_with_cafile(url).getcode()
|
||||
|
||||
assert inner1() == inner2()
|
||||
@@ -76,7 +76,7 @@ def test_deserialize_py2py3_yaml_cassette(tmpdir, req_body, expect):
|
||||
cfile = tmpdir.join("test_cassette.yaml")
|
||||
cfile.write(REQBODY_TEMPLATE.format(req_body=req_body))
|
||||
with open(str(cfile)) as f:
|
||||
(requests, responses) = deserialize(f.read(), yamlserializer)
|
||||
(requests, _) = deserialize(f.read(), yamlserializer)
|
||||
assert requests[0].body == expect
|
||||
|
||||
|
||||
|
||||
@@ -178,7 +178,7 @@ def test_testcase_playback(tmpdir):
|
||||
return str(cassette_dir)
|
||||
|
||||
test = run_testcase(MyTest)[0][0]
|
||||
assert b"illustrative examples" in test.response
|
||||
assert b"Example Domain" in test.response
|
||||
assert len(test.cassette.requests) == 1
|
||||
assert test.cassette.play_count == 0
|
||||
|
||||
@@ -186,7 +186,7 @@ def test_testcase_playback(tmpdir):
|
||||
|
||||
test2 = run_testcase(MyTest)[0][0]
|
||||
assert test.cassette is not test2.cassette
|
||||
assert b"illustrative examples" in test.response
|
||||
assert b"Example Domain" in test.response
|
||||
assert len(test2.cassette.requests) == 1
|
||||
assert test2.cassette.play_count == 1
|
||||
|
||||
|
||||
Reference in New Issue
Block a user