mirror of
https://github.com/kevin1024/vcrpy.git
synced 2025-12-08 16:53:23 +00:00
314 lines
10 KiB
Python
314 lines
10 KiB
Python
import pytest
|
|
|
|
import vcr
|
|
|
|
asyncio = pytest.importorskip("asyncio")
|
|
httpx = pytest.importorskip("httpx")
|
|
|
|
|
|
@pytest.fixture(params=["https", "http"])
|
|
def scheme(request):
|
|
"""Fixture that returns both http and https."""
|
|
return request.param
|
|
|
|
|
|
class BaseDoRequest:
|
|
_client_class = None
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
self._client_args = args
|
|
self._client_kwargs = kwargs
|
|
self._client_kwargs["follow_redirects"] = self._client_kwargs.get("follow_redirects", True)
|
|
|
|
def _make_client(self):
|
|
return self._client_class(*self._client_args, **self._client_kwargs)
|
|
|
|
|
|
class DoSyncRequest(BaseDoRequest):
|
|
_client_class = httpx.Client
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *args):
|
|
pass
|
|
|
|
@property
|
|
def client(self):
|
|
try:
|
|
return self._client
|
|
except AttributeError:
|
|
self._client = self._make_client()
|
|
return self._client
|
|
|
|
def __call__(self, *args, **kwargs):
|
|
return self.client.request(*args, timeout=60, **kwargs)
|
|
|
|
def stream(self, *args, **kwargs):
|
|
with self.client.stream(*args, **kwargs) as response:
|
|
return b"".join(response.iter_bytes())
|
|
|
|
|
|
class DoAsyncRequest(BaseDoRequest):
|
|
_client_class = httpx.AsyncClient
|
|
|
|
def __enter__(self):
|
|
# Need to manage both loop and client, because client's implementation
|
|
# will fail if the loop is closed before the client's end of life.
|
|
self._loop = asyncio.new_event_loop()
|
|
asyncio.set_event_loop(self._loop)
|
|
self._client = self._make_client()
|
|
self._loop.run_until_complete(self._client.__aenter__())
|
|
return self
|
|
|
|
def __exit__(self, *args):
|
|
try:
|
|
self._loop.run_until_complete(self._client.__aexit__(*args))
|
|
finally:
|
|
del self._client
|
|
self._loop.close()
|
|
del self._loop
|
|
|
|
@property
|
|
def client(self):
|
|
try:
|
|
return self._client
|
|
except AttributeError as e:
|
|
raise ValueError('To access async client, use "with do_request() as client"') from e
|
|
|
|
def __call__(self, *args, **kwargs):
|
|
if hasattr(self, "_loop"):
|
|
return self._loop.run_until_complete(self.client.request(*args, **kwargs))
|
|
|
|
# Use one-time context and dispose of the loop/client afterwards
|
|
with self:
|
|
return self._loop.run_until_complete(self.client.request(*args, **kwargs))
|
|
|
|
async def _get_stream(self, *args, **kwargs):
|
|
async with self.client.stream(*args, **kwargs) as response:
|
|
content = b""
|
|
async for c in response.aiter_bytes():
|
|
content += c
|
|
return content
|
|
|
|
def stream(self, *args, **kwargs):
|
|
if hasattr(self, "_loop"):
|
|
return self._loop.run_until_complete(self._get_stream(*args, **kwargs))
|
|
|
|
# Use one-time context and dispose of the loop/client afterwards
|
|
with self:
|
|
return self._loop.run_until_complete(self._get_stream(*args, **kwargs))
|
|
|
|
|
|
def pytest_generate_tests(metafunc):
|
|
if "do_request" in metafunc.fixturenames:
|
|
metafunc.parametrize("do_request", [DoAsyncRequest, DoSyncRequest])
|
|
|
|
|
|
@pytest.fixture
|
|
def yml(tmpdir, request):
|
|
return str(tmpdir.join(request.function.__name__ + ".yaml"))
|
|
|
|
|
|
@pytest.mark.online
|
|
def test_status(tmpdir, httpbin, do_request):
|
|
url = httpbin.url
|
|
|
|
with vcr.use_cassette(str(tmpdir.join("status.yaml"))):
|
|
response = do_request()("GET", url)
|
|
|
|
with vcr.use_cassette(str(tmpdir.join("status.yaml"))) as cassette:
|
|
cassette_response = do_request()("GET", url)
|
|
assert cassette_response.status_code == response.status_code
|
|
assert cassette.play_count == 1
|
|
|
|
|
|
@pytest.mark.online
|
|
def test_case_insensitive_headers(tmpdir, httpbin, do_request):
|
|
url = httpbin.url
|
|
|
|
with vcr.use_cassette(str(tmpdir.join("whatever.yaml"))):
|
|
do_request()("GET", url)
|
|
|
|
with vcr.use_cassette(str(tmpdir.join("whatever.yaml"))) as cassette:
|
|
cassette_response = do_request()("GET", url)
|
|
assert "Content-Type" in cassette_response.headers
|
|
assert "content-type" in cassette_response.headers
|
|
assert cassette.play_count == 1
|
|
|
|
|
|
@pytest.mark.online
|
|
def test_content(tmpdir, httpbin, do_request):
|
|
url = httpbin.url
|
|
|
|
with vcr.use_cassette(str(tmpdir.join("cointent.yaml"))):
|
|
response = do_request()("GET", url)
|
|
|
|
with vcr.use_cassette(str(tmpdir.join("cointent.yaml"))) as cassette:
|
|
cassette_response = do_request()("GET", url)
|
|
assert cassette_response.content == response.content
|
|
assert cassette.play_count == 1
|
|
|
|
|
|
@pytest.mark.online
|
|
def test_json(tmpdir, httpbin, do_request):
|
|
url = httpbin.url + "/json"
|
|
|
|
with vcr.use_cassette(str(tmpdir.join("json.yaml"))):
|
|
response = do_request()("GET", url)
|
|
|
|
with vcr.use_cassette(str(tmpdir.join("json.yaml"))) as cassette:
|
|
cassette_response = do_request()("GET", url)
|
|
assert cassette_response.json() == response.json()
|
|
assert cassette.play_count == 1
|
|
|
|
|
|
@pytest.mark.online
|
|
def test_params_same_url_distinct_params(tmpdir, httpbin, do_request):
|
|
url = httpbin.url + "/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 = do_request()("GET", url, params=params, headers=headers)
|
|
|
|
with vcr.use_cassette(str(tmpdir.join("get.yaml"))) as cassette:
|
|
cassette_response = do_request()("GET", url, params=params, headers=headers)
|
|
assert cassette_response.request.url == response.request.url
|
|
assert cassette_response.json() == response.json()
|
|
assert cassette.play_count == 1
|
|
|
|
params = {"other": "params"}
|
|
with vcr.use_cassette(str(tmpdir.join("get.yaml"))) as cassette:
|
|
with pytest.raises(vcr.errors.CannotOverwriteExistingCassetteException):
|
|
do_request()("GET", url, params=params, headers=headers)
|
|
|
|
|
|
@pytest.mark.online
|
|
def test_redirect(httpbin, yml, do_request):
|
|
url = httpbin.url + "/redirect-to"
|
|
|
|
response = do_request()("GET", url)
|
|
with vcr.use_cassette(yml):
|
|
response = do_request()("GET", url, params={"url": "./get", "status_code": 302})
|
|
|
|
with vcr.use_cassette(yml) as cassette:
|
|
cassette_response = do_request()("GET", url, params={"url": "./get", "status_code": 302})
|
|
|
|
assert cassette_response.status_code == response.status_code
|
|
assert len(cassette_response.history) == len(response.history)
|
|
assert len(cassette) == 2
|
|
assert cassette.play_count == 2
|
|
|
|
# Assert that the real response and the cassette response have a similar
|
|
# looking request_info.
|
|
assert cassette_response.request.url == response.request.url
|
|
assert cassette_response.request.method == response.request.method
|
|
assert cassette_response.request.headers.items() == response.request.headers.items()
|
|
|
|
|
|
@pytest.mark.online
|
|
def test_work_with_gzipped_data(httpbin, do_request, yml):
|
|
url = httpbin.url + "/gzip?foo=bar"
|
|
headers = {"accept-encoding": "deflate, gzip"}
|
|
|
|
with vcr.use_cassette(yml):
|
|
do_request(headers=headers)("GET", url)
|
|
|
|
with vcr.use_cassette(yml) as cassette:
|
|
cassette_response = do_request(headers=headers)("GET", url)
|
|
|
|
assert cassette_response.headers["content-encoding"] == "gzip"
|
|
assert cassette_response.read()
|
|
assert cassette.play_count == 1
|
|
|
|
|
|
@pytest.mark.online
|
|
@pytest.mark.parametrize("url", ["https://github.com/kevin1024/vcrpy/issues/" + str(i) for i in range(3, 6)])
|
|
def test_simple_fetching(do_request, yml, url):
|
|
with vcr.use_cassette(yml):
|
|
do_request()("GET", url)
|
|
|
|
with vcr.use_cassette(yml) as cassette:
|
|
cassette_response = do_request()("GET", url)
|
|
assert str(cassette_response.request.url) == url
|
|
assert cassette.play_count == 1
|
|
|
|
|
|
@pytest.mark.online
|
|
def test_cookies(tmpdir, httpbin, do_request):
|
|
def client_cookies(client):
|
|
return list(client.client.cookies)
|
|
|
|
def response_cookies(response):
|
|
return list(response.cookies)
|
|
|
|
url = httpbin.url + "/cookies/set"
|
|
params = {"k1": "v1", "k2": "v2"}
|
|
|
|
with do_request(params=params, follow_redirects=False) as client:
|
|
assert client_cookies(client) == []
|
|
|
|
testfile = str(tmpdir.join("cookies.yml"))
|
|
with vcr.use_cassette(testfile):
|
|
r1 = client("GET", url)
|
|
|
|
assert response_cookies(r1) == ["k1", "k2"]
|
|
|
|
r2 = client("GET", url)
|
|
|
|
assert response_cookies(r2) == ["k1", "k2"]
|
|
assert client_cookies(client) == ["k1", "k2"]
|
|
|
|
with do_request(params=params, follow_redirects=False) as new_client:
|
|
assert client_cookies(new_client) == []
|
|
|
|
with vcr.use_cassette(testfile) as cassette:
|
|
cassette_response = new_client("GET", url)
|
|
|
|
assert cassette.play_count == 1
|
|
assert response_cookies(cassette_response) == ["k1", "k2"]
|
|
assert client_cookies(new_client) == ["k1", "k2"]
|
|
|
|
|
|
@pytest.mark.online
|
|
def test_stream(tmpdir, httpbin, do_request):
|
|
url = httpbin.url + "/stream-bytes/512"
|
|
testfile = str(tmpdir.join("stream.yml"))
|
|
|
|
with vcr.use_cassette(testfile):
|
|
response_content = do_request().stream("GET", url)
|
|
assert len(response_content) == 512
|
|
|
|
with vcr.use_cassette(testfile) as cassette:
|
|
cassette_content = do_request().stream("GET", url)
|
|
assert cassette_content == response_content
|
|
assert len(cassette_content) == 512
|
|
assert cassette.play_count == 1
|
|
|
|
@pytest.mark.online
|
|
def test_text_content_type(tmpdir, httpbin, do_request):
|
|
url = httpbin.url + "/json"
|
|
|
|
with vcr.use_cassette(str(tmpdir.join("json_type.yaml"))):
|
|
response = do_request()("GET", url)
|
|
|
|
with vcr.use_cassette(str(tmpdir.join("json_type.yaml"))) as cassette:
|
|
cassette_response = do_request()("GET", url)
|
|
assert cassette_response.content == response.content
|
|
assert cassette.play_count == 1
|
|
assert isinstance(cassette.responses[0]['content'], str)
|
|
|
|
@pytest.mark.online
|
|
def test_binary_content_type(tmpdir, httpbin, do_request):
|
|
url = httpbin.url + "/bytes/1024"
|
|
|
|
with vcr.use_cassette(str(tmpdir.join("json_type.yaml"))):
|
|
response = do_request()("GET", url)
|
|
|
|
with vcr.use_cassette(str(tmpdir.join("json_type.yaml"))) as cassette:
|
|
cassette_response = do_request()("GET", url)
|
|
assert cassette_response.content == response.content
|
|
assert cassette.play_count == 1
|
|
assert isinstance(cassette.responses[0]['content'], bytes)
|