mirror of
https://github.com/kevin1024/vcrpy.git
synced 2025-12-09 17:15:35 +00:00
214 lines
7.4 KiB
Python
214 lines
7.4 KiB
Python
from unittest.mock import MagicMock
|
|
import pytest
|
|
import contextlib
|
|
import os
|
|
import vcr # noqa: E402
|
|
|
|
from vcr.stubs.httpx_stubs import _get_next_url
|
|
|
|
|
|
asyncio = pytest.importorskip("asyncio")
|
|
httpx = pytest.importorskip("httpx")
|
|
|
|
|
|
class BaseDoRequest:
|
|
_client_class = None
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
self._client = self._client_class(*args, **kwargs)
|
|
|
|
|
|
class DoSyncRequest(BaseDoRequest):
|
|
_client_class = httpx.Client
|
|
|
|
def __call__(self, *args, **kwargs):
|
|
return self._client.request(*args, timeout=60, **kwargs)
|
|
|
|
|
|
class DoAsyncRequest(BaseDoRequest):
|
|
_client_class = httpx.AsyncClient
|
|
|
|
@staticmethod
|
|
def run_in_loop(coroutine):
|
|
with contextlib.closing(asyncio.new_event_loop()) as loop:
|
|
asyncio.set_event_loop(loop)
|
|
task = loop.create_task(coroutine)
|
|
return loop.run_until_complete(task)
|
|
|
|
def __call__(self, *args, **kwargs):
|
|
async def _request():
|
|
async with self._client as c:
|
|
return await c.request(*args, **kwargs)
|
|
|
|
return DoAsyncRequest.run_in_loop(_request())
|
|
|
|
|
|
def pytest_generate_tests(metafunc):
|
|
if "do_request" in metafunc.fixturenames:
|
|
metafunc.parametrize("do_request", [DoAsyncRequest, DoSyncRequest])
|
|
if "scheme" in metafunc.fixturenames:
|
|
metafunc.parametrize("scheme", ["http", "https"])
|
|
|
|
|
|
@pytest.fixture
|
|
def yml(tmpdir, request):
|
|
return str(tmpdir.join(request.function.__name__ + ".yaml"))
|
|
|
|
|
|
def test_status(tmpdir, scheme, do_request):
|
|
url = scheme + "://httpbin.org"
|
|
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
|
|
|
|
|
|
def test_case_insensitive_headers(tmpdir, scheme, do_request):
|
|
url = scheme + "://httpbin.org"
|
|
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
|
|
|
|
|
|
def test_content(tmpdir, scheme, do_request):
|
|
url = scheme + "://httpbin.org"
|
|
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
|
|
|
|
|
|
def test_json(tmpdir, scheme, do_request):
|
|
url = scheme + "://httpbin.org/get"
|
|
headers = {"Content-Type": "application/json"}
|
|
|
|
with vcr.use_cassette(str(tmpdir.join("json.yaml"))):
|
|
response = do_request(headers=headers)("GET", url)
|
|
|
|
with vcr.use_cassette(str(tmpdir.join("json.yaml"))) as cassette:
|
|
cassette_response = do_request(headers=headers)("GET", url)
|
|
assert cassette_response.json() == response.json()
|
|
assert cassette.play_count == 1
|
|
|
|
|
|
def test_params_same_url_distinct_params(tmpdir, scheme, do_request):
|
|
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 = 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)
|
|
|
|
|
|
def test_redirect(tmpdir, do_request, yml):
|
|
url = "https://httpbin.org/redirect/2"
|
|
|
|
response = do_request()("GET", url)
|
|
with vcr.use_cassette(yml):
|
|
response = do_request()("GET", url)
|
|
|
|
with vcr.use_cassette(yml) as cassette:
|
|
cassette_response = do_request()("GET", url)
|
|
|
|
assert cassette_response.status_code == response.status_code
|
|
assert len(cassette_response.history) == len(response.history)
|
|
assert len(cassette) == 3
|
|
assert cassette.play_count == 3
|
|
|
|
# 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 {k: v for k, v in cassette_response.request.headers.items()} == {
|
|
k: v for k, v in response.request.headers.items()
|
|
}
|
|
|
|
|
|
def test_work_with_gzipped_data(tmpdir, do_request, yml):
|
|
with vcr.use_cassette(yml):
|
|
response = do_request()("GET", "https://httpbin.org/gzip")
|
|
|
|
with vcr.use_cassette(yml) as cassette:
|
|
cassette_response = do_request()("GET", "https://httpbin.org/gzip")
|
|
|
|
assert "gzip" in cassette_response.json()["headers"]["Accept-Encoding"]
|
|
assert cassette_response.read()
|
|
assert cassette.play_count == 1
|
|
|
|
|
|
@pytest.mark.parametrize("url", [f"http://github.com/kevin1024/vcrpy/issues/{i}" for i in range(3, 6)])
|
|
def test_simple_fetching(tmpdir, do_request, yml, url):
|
|
with vcr.use_cassette(yml):
|
|
response = do_request()("GET", url)
|
|
|
|
with vcr.use_cassette(yml) as cassette:
|
|
cassette_response = do_request()("GET", url)
|
|
cassette_response.request.url == url
|
|
assert cassette.play_count == 1
|
|
|
|
|
|
class TestGetNextUrl:
|
|
def test_relative_location(self):
|
|
response = MagicMock()
|
|
response.url = "http://github.com/"
|
|
response.headers = {"location": "relative"}
|
|
assert str(_get_next_url(response)) == "http://github.com/relative"
|
|
|
|
def test_absolute_location(self):
|
|
response = MagicMock()
|
|
response.url = "http://github.com/"
|
|
response.headers = {"location": "http://google.com"}
|
|
assert str(_get_next_url(response)) == "http://google.com"
|
|
|
|
def test_no_location(self):
|
|
response = MagicMock()
|
|
response.url = "http://github.com/"
|
|
response.headers = {}
|
|
assert _get_next_url(response) is None
|
|
|
|
|
|
def test_behind_proxy(do_request):
|
|
# This is recorded because otherwise we should have a live proxy somewhere.
|
|
yml = (
|
|
os.path.dirname(os.path.realpath(__file__))
|
|
+ "/cassettes/"
|
|
+ do_request.__name__
|
|
+ "test_httpx_test_test_behind_proxy.yml"
|
|
)
|
|
url = "https://httpbin.org/headers"
|
|
proxy = "http://localhost:8080"
|
|
proxies = {"http": proxy, "https": proxy}
|
|
|
|
with vcr.use_cassette(yml):
|
|
response = do_request(proxies=proxies, verify=False)("GET", url)
|
|
|
|
with vcr.use_cassette(yml) as cassette:
|
|
cassette_response = do_request(proxies=proxies, verify=False)("GET", url)
|
|
cassette_response.request.url == url
|
|
assert cassette.play_count == 1
|
|
|
|
assert cassette_response.headers["Via"] == "my_own_proxy", str(cassette_response.headers)
|
|
assert cassette_response.request.url == response.request.url
|