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,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
 ------

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 <https://toml.io>`_ format.
+file. Recorded data is stored in `YAML <https://yaml.org>`_ 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,63 +90,57 @@ 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")
         another()
 
 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")
 
 

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"]

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,22 +79,32 @@ 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
 
             return wrapper
 
         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",

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,30 +66,15 @@ 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
 
         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")
+        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/"