Mirroring/vcrpy
1
0
Fork 0
mirror of https://github.com/kevin1024/vcrpy.git synced 2025-12-08 16:53:23 +00:00
Code Releases Activity

Run linters.

Browse Source
This commit is contained in:
Allan Crooks
2024-01-21 15:50:46 +00:00
committed by Jair Henrique
parent c5487384ee
commit 54bc6467eb
2 changed files with 17 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
tests/integration/test_httpx.py
View File

@@ -1,11 +1,14 @@
import pytest
import os import os
import pytest
import vcr import vcr
from ..assertions import assert_is_json_bytes
asyncio = pytest.importorskip("asyncio") asyncio = pytest.importorskip("asyncio")
httpx = pytest.importorskip("httpx") httpx = pytest.importorskip("httpx")
from ..assertions import assert_is_json_bytes
@pytest.fixture(params=["https", "http"]) @pytest.fixture(params=["https", "http"])
def scheme(request): def scheme(request):

21
vcr/stubs/httpx_stubs.py
View File

@@ -7,8 +7,8 @@ from unittest.mock import MagicMock, patch
import httpx import httpx
from vcr.errors import CannotOverwriteExistingCassetteException from vcr.errors import CannotOverwriteExistingCassetteException
from vcr.request import Request as VcrRequest
from vcr.filters import decode_response from vcr.filters import decode_response
from vcr.request import Request as VcrRequest
from vcr.serializers.compat import convert_body_to_bytes from vcr.serializers.compat import convert_body_to_bytes
_httpx_signature = inspect.signature(httpx.Client.request) _httpx_signature = inspect.signature(httpx.Client.request)
@@ -37,7 +37,6 @@ def _transform_headers(httpx_response):
async def _to_serialized_response(resp, aread): async def _to_serialized_response(resp, aread):
# The content shouldn't already have been read in by HTTPX. # The content shouldn't already have been read in by HTTPX.
assert not hasattr(resp, "_decoder") assert not hasattr(resp, "_decoder")
@@ -49,7 +48,7 @@ async def _to_serialized_response(resp, aread):
resp.read() resp.read()
result = { result = {
"status": dict(code=resp.status_code, message=resp.reason_phrase), "status": {"code": resp.status_code, "message": resp.reason_phrase},
"headers": _transform_headers(resp), "headers": _transform_headers(resp),
"body": {"string": resp.content}, "body": {"string": resp.content},
} }
@@ -60,6 +59,7 @@ async def _to_serialized_response(resp, aread):
resp._content = resp._get_content_decoder().decode(resp.content) resp._content = resp._get_content_decoder().decode(resp.content)
return result return result
def _from_serialized_headers(headers): def _from_serialized_headers(headers):
""" """
httpx accepts headers as list of tuples of header key and value. httpx accepts headers as list of tuples of header key and value.
@@ -75,16 +75,19 @@ def _from_serialized_headers(headers):
@patch("httpx.Response.close", MagicMock()) @patch("httpx.Response.close", MagicMock())
@patch("httpx.Response.read", MagicMock()) @patch("httpx.Response.read", MagicMock())
def _from_serialized_response(request, serialized_response, history=None): def _from_serialized_response(request, serialized_response, history=None):
# Cassette format generated for HTTPX requests by older versions of # Cassette format generated for HTTPX requests by older versions of
# vcrpy. We restructure the content to resemble what a regular # vcrpy. We restructure the content to resemble what a regular
# cassette looks like. # cassette looks like.
if "status_code" in serialized_response: if "status_code" in serialized_response:
serialized_response = decode_response(convert_body_to_bytes({ serialized_response = decode_response(
'headers': serialized_response['headers'], convert_body_to_bytes(
'body': {'string': serialized_response['content']}, {
'status': {'code': serialized_response['status_code']}, "headers": serialized_response["headers"],
})) "body": {"string": serialized_response["content"]},
"status": {"code": serialized_response["status_code"]},
},
),
)
extensions = None extensions = None
else: else:
extensions = {"reason_phrase": serialized_response["status"]["message"].encode()} extensions = {"reason_phrase": serialized_response["status"]["message"].encode()}