mirror of
https://github.com/kevin1024/vcrpy.git
synced 2025-12-09 09:13:23 +00:00
133423ce94
Because the recoded requests are using original value. Changing the value on redirect prevents finding the matching response.
243 lines
8.5 KiB
Python
243 lines
8.5 KiB
Python
from unittest.mock import MagicMock
|
|
import pytest
|
|
import contextlib
|
|
import os
|
|
|
|
asyncio = pytest.importorskip("asyncio")
|
|
httpx = pytest.importorskip("httpx")
|
|
|
|
import vcr # noqa: E402
|
|
|
|
|
|
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):
|
|
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", ["https://github.com/kevin1024/vcrpy/issues/" + str(i) for i in range(3, 6)])
|
|
def test_simple_fetching(tmpdir, 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
|
|
|
|
|
|
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)
|
|
assert str(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
|
|
|
|
|
|
def test_cookies(tmpdir, scheme, do_request):
|
|
def client_cookies(client):
|
|
return [c for c in client._client.cookies]
|
|
def response_cookies(response):
|
|
return [c for c in response.cookies]
|
|
|
|
client = do_request()
|
|
assert client_cookies(client) == []
|
|
|
|
url = scheme + "://httpbin.org"
|
|
testfile = str(tmpdir.join("cookies.yml"))
|
|
with vcr.use_cassette(testfile):
|
|
r1 = client("GET", url + "/cookies/set?k1=v1&k2=v2")
|
|
assert response_cookies(r1.history[0]) == ['k1', 'k2']
|
|
assert response_cookies(r1) == []
|
|
|
|
r2 = client("GET", url + "/cookies")
|
|
assert len(r2.json()["cookies"]) == 2
|
|
|
|
assert client_cookies(client) == ['k1', 'k2']
|
|
|
|
|
|
new_client = do_request()
|
|
assert client_cookies(new_client) == []
|
|
|
|
with vcr.use_cassette(testfile) as cassette:
|
|
cassette_response = new_client("GET", url + "/cookies/set?k1=v1&k2=v2")
|
|
assert response_cookies(cassette_response.history[0]) == ['k1', 'k2']
|
|
assert response_cookies(cassette_response) == []
|
|
|
|
assert cassette.play_count == 2
|
|
assert client_cookies(new_client) == ['k1', 'k2']
|
|
|
|
|
|
def test_relative_redirects(tmpdir, scheme, do_request):
|
|
url = scheme + "://httpbin.org"
|
|
testfile = str(tmpdir.join("relative_redirects.yml"))
|
|
with vcr.use_cassette(testfile):
|
|
response = do_request()("GET", url + "/redirect-to?url=/redirect-to?url=/get")
|
|
assert len(response.history) == 2
|
|
assert response.json()["url"].endswith("get")
|
|
|
|
with vcr.use_cassette(testfile) as cassette:
|
|
response = do_request()("GET", url + "/redirect-to?url=/redirect-to?url=/get")
|
|
assert len(response.history) == 2
|
|
assert response.json()["url"].endswith("get")
|
|
|
|
assert cassette.play_count == 3
|
|
|