mirror of
https://github.com/kevin1024/vcrpy.git
synced 2025-12-09 09:13:23 +00:00
7d2d29de12
This is required because previous extraction code is now patched out by vcpy. Also handle headers with same key in responses.
247 lines
8.6 KiB
Python
247 lines
8.6 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
|
|
from vcr.stubs.httpx_stubs import _get_next_url # 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
|
|
|
|
|
|
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)
|
|
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']
|