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

HTTPX stub now generates cassettes in the same format as other stubs.

Browse Source 
As part of this, I've removed the tests which inspect the
data type of the response content in the cassette. That
behaviour should be controlled via the inbuilt serializers.
This commit is contained in:
Allan Crooks
2024-01-20 17:46:52 +00:00
committed by Jair Henrique
parent 5fa7010712
commit 5cf23298ac
2 changed files with 20 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files

40
vcr/stubs/httpx_stubs.py
View File

@@ -1,3 +1,4 @@
import asyncio
import functools
import inspect
import logging
@@ -35,17 +36,17 @@ def _transform_headers(httpx_response):
return out
def _to_serialized_response(httpx_response):
try:
content = httpx_response.content.decode("utf-8")
except UnicodeDecodeError:
content = httpx_response.content
async def _to_serialized_response(resp, aread):
if aread:
await resp.aread()
else:
resp.read()
return {
"status_code": httpx_response.status_code,
"http_version": httpx_response.http_version,
"headers": _transform_headers(httpx_response),
"content": content,
"status": dict(code=resp.status_code, message=resp.reason_phrase),
"headers": _transform_headers(resp),
"body": {"string": resp.content},
}
@@ -65,17 +66,16 @@ def _from_serialized_headers(headers):
@patch("httpx.Response.read", MagicMock())
def _from_serialized_response(request, serialized_response, history=None):
# HTTPX cassette format.
# Cassette format generated for HTTPX requests by older versions of
# vcrpy. We restructure the content to resemble what a regular
# cassette looks like.
if "status_code" in serialized_response:
serialized_response = decode_response(convert_body_to_bytes({
'headers': serialized_response['headers'],
'body': {'string': serialized_response['content']},
'status': {'code': serialized_response['status_code']},
}))
# We don't store the reason phrase in this format.
extensions = None
# Cassette format that all other stubs use.
else:
extensions = {"reason_phrase": serialized_response["status"]["message"].encode()}
@@ -113,17 +113,17 @@ def _shared_vcr_send(cassette, real_send, *args, **kwargs):
return vcr_request, None
def _record_responses(cassette, vcr_request, real_response):
async def _record_responses(cassette, vcr_request, real_response, aread):
for past_real_response in real_response.history:
past_vcr_request = _make_vcr_request(past_real_response.request)
cassette.append(past_vcr_request, _to_serialized_response(past_real_response))
cassette.append(past_vcr_request, await _to_serialized_response(past_real_response, aread))
if real_response.history:
# If there was a redirection keep we want the request which will hold the
# final redirect value
vcr_request = _make_vcr_request(real_response.request)
cassette.append(vcr_request, _to_serialized_response(real_response))
cassette.append(vcr_request, await _to_serialized_response(real_response, aread))
return real_response
@@ -141,8 +141,8 @@ async def _async_vcr_send(cassette, real_send, *args, **kwargs):
return response
real_response = await real_send(*args, **kwargs)
await real_response.aread()
return _record_responses(cassette, vcr_request, real_response)
await _record_responses(cassette, vcr_request, real_response, aread=True)
return real_response
def async_vcr_send(cassette, real_send):
@@ -161,8 +161,8 @@ def _sync_vcr_send(cassette, real_send, *args, **kwargs):
return response
real_response = real_send(*args, **kwargs)
real_response.read()
return _record_responses(cassette, vcr_request, real_response)
asyncio.run(_record_responses(cassette, vcr_request, real_response, aread=False))
return real_response
def sync_vcr_send(cassette, real_send):