[BETA] replay responses from toml files. (#581)

Added beta option for replaying responses from toml files.

Co-authored-by: Mark Story <[email protected]>

CHANGES

@@ -3,6 +3,8 @@
 
 * Update `requests` dependency to the version of 2.22.0 or higher. See #584.
 * [BETA] Added possibility to record responses to TOML files via `@_recorder.record(file_path="out.toml")` decorator.
+* [BETA] Added possibility to replay responses (populate registry) from TOML files
+  via `responses._add_from_file(file_path="out.toml")` method.
 * Fix type for the `mock`'s patcher object. See #556
 * Fix type annotation for `CallList`
 * Add `passthrough` argument to `BaseResponse` object. See #557

responses/__init__.py

@@ -24,6 +24,7 @@
 from typing import overload
 from warnings import warn
 
+import toml as _toml
 from requests.adapters import HTTPAdapter
 from requests.adapters import MaxRetryError
 from requests.exceptions import ConnectionError
@@ -64,6 +65,7 @@
 
 if TYPE_CHECKING:  # pragma: no cover
     # import only for linter run
+    import os
     from unittest.mock import _patch as _mock_patcher
 
     from requests import PreparedRequest
@@ -769,6 +771,21 @@ def add(
         response = Response(method=method, url=url, body=body, **kwargs)
         return self._registry.add(response)
 
+    def _add_from_file(self, file_path: "Union[str, bytes, os.PathLike[Any]]") -> None:
+        with open(file_path) as file:
+            data = _toml.load(file)
+
+        for rsp in data["responses"]:
+            rsp = rsp["response"]
+            self.add(
+                method=rsp["method"],
+                url=rsp["url"],
+                body=rsp["body"],
+                status=rsp["status"],
+                content_type=rsp["content_type"],
+                auto_calculate_content_length=rsp["auto_calculate_content_length"],
+            )
+
     def delete(self, *args: Any, **kwargs: Any) -> BaseResponse:
         return self.add(DELETE, *args, **kwargs)
 
@@ -1146,6 +1163,7 @@ def assert_call_count(self, url: str, count: int) -> bool:
     # Exposed by the RequestsMock class:
     "activate",
     "add",
+    "_add_from_file",
     "add_callback",
     "add_passthru",
     "_deprecated_assert_all_requests_are_fired",
@@ -1180,6 +1198,7 @@ def assert_call_count(self, url: str, count: int) -> bool:
 # expose only methods and/or read-only methods
 activate = _default_mock.activate
 add = _default_mock.add
+_add_from_file = _default_mock._add_from_file
 add_callback = _default_mock.add_callback
 add_passthru = _default_mock.add_passthru
 _deprecated_assert_all_requests_are_fired = _default_mock.assert_all_requests_are_fired

responses/tests/test_recorder.py

@@ -3,9 +3,58 @@
 import requests
 import toml
 
+import responses
 from responses import _recorder
 
 
+def get_data(host, port):
+    data = {
+        "responses": [
+            {
+                "response": {
+                    "method": "GET",
+                    "url": f"http://{host}:{port}/404",
+                    "body": "404 Not Found",
+                    "status": 404,
+                    "content_type": "text/plain",
+                    "auto_calculate_content_length": False,
+                }
+            },
+            {
+                "response": {
+                    "method": "GET",
+                    "url": f"http://{host}:{port}/status/wrong",
+                    "body": "Invalid status code",
+                    "status": 400,
+                    "content_type": "text/plain",
+                    "auto_calculate_content_length": False,
+                }
+            },
+            {
+                "response": {
+                    "method": "GET",
+                    "url": f"http://{host}:{port}/500",
+                    "body": "500 Internal Server Error",
+                    "status": 500,
+                    "content_type": "text/plain",
+                    "auto_calculate_content_length": False,
+                }
+            },
+            {
+                "response": {
+                    "method": "PUT",
+                    "url": f"http://{host}:{port}/202",
+                    "body": "OK",
+                    "status": 202,
+                    "content_type": "text/plain",
+                    "auto_calculate_content_length": False,
+                }
+            },
+        ]
+    }
+    return data
+
+
 class TestRecord:
     def setup(self):
         self.out_file = Path("out.toml")
@@ -35,7 +84,7 @@ def test_recorder(self, httpserver):
 
         def another():
             requests.get(url500)
-            requests.get(url202)
+            requests.put(url202)
 
         @_recorder.record(file_path=self.out_file)
         def run():
@@ -48,47 +97,46 @@ def 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,
-                    }
-                },
-            ]
-        }
+        assert data == get_data(httpserver.host, httpserver.port)
+
+
+class TestReplay:
+    def teardown(self):
+        out_file = Path("out.toml")
+        if out_file.exists():
+            out_file.unlink()
+
+        assert not out_file.exists()
+
+    def test_add_from_file(self):
+        with open("out.toml", "w") as file:
+            toml.dump(get_data("example.com", "8080"), file)
+
+        @responses.activate
+        def run():
+            responses.patch("http://httpbin.org")
+            responses._add_from_file(file_path="out.toml")
+            responses.post("http://httpbin.org/form")
+
+            assert responses.registered()[0].url == "http://httpbin.org/"
+            assert responses.registered()[1].url == "http://example.com:8080/404"
+            assert (
+                responses.registered()[2].url == "http://example.com:8080/status/wrong"
+            )
+            assert responses.registered()[3].url == "http://example.com:8080/500"
+            assert responses.registered()[4].url == "http://example.com:8080/202"
+            assert responses.registered()[5].url == "http://httpbin.org/form"
+
+            assert responses.registered()[0].method == "PATCH"
+            assert responses.registered()[2].method == "GET"
+            assert responses.registered()[4].method == "PUT"
+            assert responses.registered()[5].method == "POST"
+
+            assert responses.registered()[2].status == 400
+            assert responses.registered()[3].status == 500
+
+            assert responses.registered()[3].body == "500 Internal Server Error"
+
+            assert responses.registered()[3].content_type == "text/plain"
+
+        run()