From e85804ae020d030c65a71d71570b286d258fd5b0 Mon Sep 17 00:00:00 2001 From: Maksim Beliaev Date: Mon, 6 Mar 2023 21:37:44 +0100 Subject: [PATCH] Change default recorder format to YAML (#624) * Default recorder format was changed to YAML. Added `responses.RequestsMock._parse_response_file` and `responses._recorder.Recorder.dump_to_file` methods that allow users to override default parser to eg toml, json --- CHANGES | 3 +- README.rst | 86 ++++++++++++------------ responses/__init__.py | 14 +++- responses/_recorder.py | 29 +++++++-- responses/tests/test_recorder.py | 108 +++++++++++++++++++++++-------- setup.py | 7 +- 6 files changed, 163 insertions(+), 84 deletions(-) diff --git a/CHANGES b/CHANGES index 325a0448..575ee0f4 100644 --- a/CHANGES +++ b/CHANGES @@ -3,9 +3,10 @@ * Add Python 3.11 support * Fix type annotations of `CallList`. See #593 -* Replaced toml with tomli and tomli-w. * `request` object is attached to any custom exception provided as `Response` `body` argument. See #588 * Fixed mocked responses leaking between tests when `assert_all_requests_are_fired` and a request was not fired. +* [BETA] Default recorder format was changed to YAML. Added `responses.RequestsMock._parse_response_file` and + `responses._recorder.Recorder.dump_to_file` methods that allow users to override default parser to eg toml, json 0.22.0 ------ diff --git a/README.rst b/README.rst index 2a693d5f..cfdf0b20 100644 --- a/README.rst +++ b/README.rst @@ -72,10 +72,10 @@ Record Responses to files ^^^^^^^^^^^^^^^^^^^^^^^^^ You can perform real requests to the server and ``responses`` will automatically record the output to the -file. Recorded data is stored in `toml `_ format. +file. Recorded data is stored in `YAML `_ format. -Apply ``@responses._recorder.record(file_path="out.toml")`` decorator to any function where you perform -requests to record responses to ``out.toml`` file. +Apply ``@responses._recorder.record(file_path="out.yaml")`` decorator to any function where you perform +requests to record responses to ``out.yaml`` file. Following code @@ -90,7 +90,7 @@ Following code rsp = requests.get("https://httpstat.us/202") - @_recorder.record(file_path="out.toml") + @_recorder.record(file_path="out.yaml") def test_recorder(): rsp = requests.get("https://httpstat.us/404") rsp = requests.get("https://httpbin.org/status/wrong") @@ -98,55 +98,49 @@ Following code will produce next output: -.. code-block:: toml - - [[responses]] - - [responses.response] - method = "GET" - url = "https://httpstat.us/404" - body = "404 Not Found" - status = 404 - content_type = "text/plain" - auto_calculate_content_length = false - [[responses]] - - [responses.response] - method = "GET" - url = "https://httpbin.org/status/wrong" - body = "Invalid status code" - status = 400 - content_type = "text/plain" - auto_calculate_content_length = false - [[responses]] - - [responses.response] - method = "GET" - url = "https://httpstat.us/500" - body = "500 Internal Server Error" - status = 500 - content_type = "text/plain" - auto_calculate_content_length = false - [[responses]] - - [responses.response] - method = "GET" - url = "https://httpstat.us/202" - body = "202 Accepted" - status = 202 - content_type = "text/plain" - auto_calculate_content_length = false +.. code-block:: yaml + + responses: + - response: + auto_calculate_content_length: false + body: 404 Not Found + content_type: text/plain + method: GET + status: 404 + url: https://httpstat.us/404 + - response: + auto_calculate_content_length: false + body: Invalid status code + content_type: text/plain + method: GET + status: 400 + url: https://httpbin.org/status/wrong + - response: + auto_calculate_content_length: false + body: 500 Internal Server Error + content_type: text/plain + method: GET + status: 500 + url: https://httpstat.us/500 + - response: + auto_calculate_content_length: false + body: 202 Accepted + content_type: text/plain + method: GET + status: 202 + url: https://httpstat.us/202 + Replay responses (populate registry) from files ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ -You can populate your active registry from a ``toml`` file with recorded responses. +You can populate your active registry from a ``yaml`` file with recorded responses. (See `Record Responses to files`_ to understand how to obtain a file). -To do that you need to execute ``responses._add_from_file(file_path="out.toml")`` within +To do that you need to execute ``responses._add_from_file(file_path="out.yaml")`` within an activated decorator or a context manager. The following code example registers a ``patch`` response, then all responses present in -``out.toml`` file and a ``post`` response at the end. +``out.yaml`` file and a ``post`` response at the end. .. code-block:: python @@ -156,7 +150,7 @@ The following code example registers a ``patch`` response, then all responses pr @responses.activate def run(): responses.patch("http://httpbin.org") - responses._add_from_file(file_path="out.toml") + responses._add_from_file(file_path="out.yaml") responses.post("http://httpbin.org/form") diff --git a/responses/__init__.py b/responses/__init__.py index a24e12cb..f9f721ec 100644 --- a/responses/__init__.py +++ b/responses/__init__.py @@ -25,11 +25,13 @@ from typing import overload from warnings import warn +import yaml + try: import tomli as _toml except ImportError: # python 3.11 - import tomllib as _toml # type: ignore[no-redef] + import tomllib as _toml # type: ignore[no-redef] # noqa: F401 from requests.adapters import HTTPAdapter from requests.adapters import MaxRetryError @@ -783,9 +785,15 @@ def add( post = partialmethod(add, POST) put = partialmethod(add, PUT) + def _parse_response_file( + self, file_path: "Union[str, bytes, os.PathLike[Any]]" + ) -> "Dict[str, Any]": + with open(file_path, "r") as file: + data = yaml.safe_load(file) + return data + def _add_from_file(self, file_path: "Union[str, bytes, os.PathLike[Any]]") -> None: - with open(file_path, "rb") as file: - data = _toml.load(file) + data = self._parse_response_file(file_path) for rsp in data["responses"]: rsp = rsp["response"] diff --git a/responses/_recorder.py b/responses/_recorder.py index 7142691f..93603ff7 100644 --- a/responses/_recorder.py +++ b/responses/_recorder.py @@ -18,7 +18,9 @@ from responses import _F from responses import BaseResponse -import tomli_w as _toml_w + from io import TextIOWrapper + +import yaml from responses import RequestsMock from responses import Response @@ -34,7 +36,11 @@ def _remove_nones(d: "Any") -> "Any": return d -def _dump(registered: "List[BaseResponse]", destination: "BinaryIO") -> None: +def _dump( + registered: "List[BaseResponse]", + destination: "Union[BinaryIO, TextIOWrapper]", + dumper: "Callable[[Union[Dict[Any, Any], List[Any]], Union[BinaryIO, TextIOWrapper]], Any]", +) -> None: data: Dict[str, Any] = {"responses": []} for rsp in registered: try: @@ -57,12 +63,13 @@ def _dump(registered: "List[BaseResponse]", destination: "BinaryIO") -> None: "Cannot dump response object." "Probably you use custom Response object that is missing required attributes" ) from exc - _toml_w.dump(_remove_nones(data), destination) + dumper(_remove_nones(data), destination) class Recorder(RequestsMock): def __init__( self, + *, target: str = "requests.adapters.HTTPAdapter.send", registry: "Type[FirstMatchRegistry]" = OrderedRegistry, ) -> None: @@ -72,15 +79,16 @@ def reset(self) -> None: self._registry = OrderedRegistry() def record( - self, *, file_path: "Union[str, bytes, os.PathLike[Any]]" = "response.toml" + self, *, file_path: "Union[str, bytes, os.PathLike[Any]]" = "response.yaml" ) -> "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, "wb") as file: - _dump(self.get_registry().registered, file) + self.dump_to_file( + file_path=file_path, registered=self.get_registry().registered + ) return ret @@ -88,6 +96,15 @@ def wrapper(*args: "Any", **kwargs: "Any") -> "Any": # type: ignore[misc] return deco_record + def dump_to_file( + self, + *, + file_path: "Union[str, bytes, os.PathLike[Any]]", + registered: "List[BaseResponse]", + ) -> None: + with open(file_path, "w") as file: + _dump(registered, file, yaml.dump) + def _on_request( self, adapter: "HTTPAdapter", diff --git a/responses/tests/test_recorder.py b/responses/tests/test_recorder.py index 30134d70..31402988 100644 --- a/responses/tests/test_recorder.py +++ b/responses/tests/test_recorder.py @@ -1,11 +1,19 @@ from pathlib import Path +import pytest import requests import tomli_w +import yaml import responses from responses import _recorder -from responses import _toml +from responses._recorder import _dump + +try: + import tomli as _toml +except ImportError: + # python 3.11 + import tomllib as _toml # type: ignore[no-redef] def get_data(host, port): @@ -58,7 +66,7 @@ def get_data(host, port): class TestRecord: def setup(self): - self.out_file = Path("out.toml") + self.out_file = Path("response_record") if self.out_file.exists(): self.out_file.unlink() # pragma: no cover @@ -66,22 +74,7 @@ def setup(self): 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") + url202, url400, url404, url500 = self.prepare_server(httpserver) def another(): requests.get(url500) @@ -95,28 +88,91 @@ def run(): run() + with open(self.out_file, "r") as file: + data = yaml.safe_load(file) + + assert data == get_data(httpserver.host, httpserver.port) + + def test_recorder_toml(self, httpserver): + custom_recorder = _recorder.Recorder() + + def dump_to_file(file_path, registered): + with open(file_path, "wb") as file: + _dump(registered, file, tomli_w.dump) + + custom_recorder.dump_to_file = dump_to_file + + url202, url400, url404, url500 = self.prepare_server(httpserver) + + def another(): + requests.get(url500) + requests.put(url202) + + @custom_recorder.record(file_path=self.out_file) + def run(): + requests.get(url404) + requests.get(url400) + another() + + run() + with open(self.out_file, "rb") as file: data = _toml.load(file) assert data == get_data(httpserver.host, httpserver.port) + def prepare_server(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") + return url202, url400, url404, url500 + class TestReplay: + def setup(self): + self.out_file = Path("response_record") + def teardown(self): - out_file = Path("out.toml") - if out_file.exists(): - out_file.unlink() + if self.out_file.exists(): + self.out_file.unlink() - assert not out_file.exists() + assert not self.out_file.exists() - def test_add_from_file(self): - with open("out.toml", "wb") as file: - tomli_w.dump(get_data("example.com", "8080"), file) + @pytest.mark.parametrize("parser", (yaml, tomli_w)) + def test_add_from_file(self, parser): + if parser == yaml: + with open(self.out_file, "w") as file: + parser.dump(get_data("example.com", "8080"), file) + else: + with open(self.out_file, "wb") as file: + parser.dump(get_data("example.com", "8080"), file) @responses.activate def run(): responses.patch("http://httpbin.org") - responses._add_from_file(file_path="out.toml") + if parser == tomli_w: + + def _parse_response_file(file_path): + with open(file_path, "rb") as file: + data = _toml.load(file) + return data + + responses.mock._parse_response_file = _parse_response_file + + responses._add_from_file(file_path=self.out_file) responses.post("http://httpbin.org/form") assert responses.registered()[0].url == "http://httpbin.org/" diff --git a/setup.py b/setup.py index 07a37149..06243432 100644 --- a/setup.py +++ b/setup.py @@ -19,8 +19,8 @@ install_requires = [ "requests>=2.22.0,<3.0", "urllib3>=1.25.10", - "tomli; python_version < '3.11'", - "tomli-w", + "pyyaml", + "types-PyYAML", "typing_extensions; python_version < '3.8'", ] @@ -33,6 +33,9 @@ "flake8", "types-requests", "mypy", + # for check of different parsers in recorder + "tomli; python_version < '3.11'", + "tomli-w", ] if "test" in sys.argv: