-
-
Notifications
You must be signed in to change notification settings - Fork 346
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
beta - Add request recording to fixture files (#545)
Add the ability to record requests that are made by application or test code and record them into a TOML based fixture file. In the future we may also provide APIs to load fixture files into `responses`. Co-authored-by: Mark Story <mark@mark-story.com>
- Loading branch information
1 parent
27872a4
commit f431d49
Showing
4 changed files
with
205 additions
and
0 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,108 @@ | ||
from functools import wraps | ||
from typing import TYPE_CHECKING | ||
|
||
if TYPE_CHECKING: # pragma: no cover | ||
import io | ||
import os | ||
|
||
from typing import Any | ||
from typing import Callable | ||
from typing import Dict | ||
from typing import List | ||
from typing import Type | ||
from typing import Union | ||
from responses import FirstMatchRegistry | ||
from responses import HTTPAdapter | ||
from responses import PreparedRequest | ||
from responses import models | ||
from responses import _F | ||
from responses import BaseResponse | ||
|
||
import toml as _toml | ||
|
||
from responses import RequestsMock | ||
from responses import Response | ||
from responses import _real_send | ||
from responses.registries import OrderedRegistry | ||
|
||
|
||
def _dump(registered: "List[BaseResponse]", destination: "io.IOBase") -> None: | ||
data: Dict[str, Any] = {"responses": []} | ||
for rsp in registered: | ||
try: | ||
content_length = rsp.auto_calculate_content_length # type: ignore[attr-defined] | ||
data["responses"].append( | ||
{ | ||
"response": { | ||
"method": rsp.method, | ||
"url": rsp.url, | ||
"body": rsp.body, # type: ignore[attr-defined] | ||
"status": rsp.status, # type: ignore[attr-defined] | ||
"headers": rsp.headers, | ||
"content_type": rsp.content_type, | ||
"auto_calculate_content_length": content_length, | ||
} | ||
} | ||
) | ||
except AttributeError as exc: # pragma: no cover | ||
raise AttributeError( | ||
"Cannot dump response object." | ||
"Probably you use custom Response object that is missing required attributes" | ||
) from exc | ||
_toml.dump(data, destination) | ||
|
||
|
||
class Recorder(RequestsMock): | ||
def __init__( | ||
self, | ||
target: str = "requests.adapters.HTTPAdapter.send", | ||
registry: "Type[FirstMatchRegistry]" = OrderedRegistry, | ||
) -> None: | ||
super().__init__(target=target, registry=registry) | ||
|
||
def reset(self) -> None: | ||
self._registry = OrderedRegistry() | ||
|
||
def record( | ||
self, *, file_path: "Union[str, bytes, os.PathLike[Any]]" = "response.toml" | ||
) -> "Union[Callable[[_F], _F], _F]": | ||
def deco_record(function: "_F") -> "Callable[..., Any]": | ||
@wraps(function) | ||
def wrapper(*args: "Any", **kwargs: "Any") -> "Any": # type: ignore[misc] | ||
with self: | ||
ret = function(*args, **kwargs) | ||
with open(file_path, "w") as file: | ||
_dump(self.get_registry().registered, file) | ||
|
||
return ret | ||
|
||
return wrapper | ||
|
||
return deco_record | ||
|
||
def _on_request( | ||
self, | ||
adapter: "HTTPAdapter", | ||
request: "PreparedRequest", | ||
**kwargs: "Any", | ||
) -> "models.Response": | ||
# add attributes params and req_kwargs to 'request' object for further match comparison | ||
# original request object does not have these attributes | ||
request.params = self._parse_request_params(request.path_url) # type: ignore[attr-defined] | ||
request.req_kwargs = kwargs # type: ignore[attr-defined] | ||
requests_response = _real_send(adapter, request, **kwargs) | ||
responses_response = Response( | ||
method=str(request.method), | ||
url=str(requests_response.request.url), | ||
status=requests_response.status_code, | ||
body=requests_response.text, | ||
) | ||
self._registry.add(responses_response) | ||
return requests_response | ||
|
||
def stop(self, allow_assert: bool = True) -> None: | ||
super().stop(allow_assert=False) | ||
|
||
|
||
recorder = Recorder() | ||
record = recorder.record |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,94 @@ | ||
from pathlib import Path | ||
|
||
import requests | ||
import toml | ||
|
||
from responses import _recorder | ||
|
||
|
||
class TestRecord: | ||
def setup(self): | ||
self.out_file = Path("out.toml") | ||
if self.out_file.exists(): | ||
self.out_file.unlink() # pragma: no cover | ||
|
||
assert not self.out_file.exists() | ||
|
||
def test_recorder(self, httpserver): | ||
|
||
httpserver.expect_request("/500").respond_with_data( | ||
"500 Internal Server Error", status=500, content_type="text/plain" | ||
) | ||
httpserver.expect_request("/202").respond_with_data( | ||
"OK", status=202, content_type="text/plain" | ||
) | ||
httpserver.expect_request("/404").respond_with_data( | ||
"404 Not Found", status=404, content_type="text/plain" | ||
) | ||
httpserver.expect_request("/status/wrong").respond_with_data( | ||
"Invalid status code", status=400, content_type="text/plain" | ||
) | ||
url500 = httpserver.url_for("/500") | ||
url202 = httpserver.url_for("/202") | ||
url404 = httpserver.url_for("/404") | ||
url400 = httpserver.url_for("/status/wrong") | ||
|
||
def another(): | ||
requests.get(url500) | ||
requests.get(url202) | ||
|
||
@_recorder.record(file_path=self.out_file) | ||
def run(): | ||
requests.get(url404) | ||
requests.get(url400) | ||
another() | ||
|
||
run() | ||
|
||
with open(self.out_file) as file: | ||
data = toml.load(file) | ||
|
||
assert data == { | ||
"responses": [ | ||
{ | ||
"response": { | ||
"method": "GET", | ||
"url": f"http://{httpserver.host}:{httpserver.port}/404", | ||
"body": "404 Not Found", | ||
"status": 404, | ||
"content_type": "text/plain", | ||
"auto_calculate_content_length": False, | ||
} | ||
}, | ||
{ | ||
"response": { | ||
"method": "GET", | ||
"url": f"http://{httpserver.host}:{httpserver.port}/status/wrong", | ||
"body": "Invalid status code", | ||
"status": 400, | ||
"content_type": "text/plain", | ||
"auto_calculate_content_length": False, | ||
} | ||
}, | ||
{ | ||
"response": { | ||
"method": "GET", | ||
"url": f"http://{httpserver.host}:{httpserver.port}/500", | ||
"body": "500 Internal Server Error", | ||
"status": 500, | ||
"content_type": "text/plain", | ||
"auto_calculate_content_length": False, | ||
} | ||
}, | ||
{ | ||
"response": { | ||
"method": "GET", | ||
"url": f"http://{httpserver.host}:{httpserver.port}/202", | ||
"body": "OK", | ||
"status": 202, | ||
"content_type": "text/plain", | ||
"auto_calculate_content_length": False, | ||
} | ||
}, | ||
] | ||
} |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters