mbeliaev/recorder

CHANGES

@@ -1,6 +1,7 @@
 0.22.0
 ------
 
+* [BETA] Added possibility to record responses to TOML files via `@_recorder.record(file_path="out.toml")` decorator.
 * Fix type for the `mock`'s patcher object. See #556
 * Add `passthrough` argument to `BaseResponse` object. See #557
 * Fix `registries` leak. See #563

responses/_recorder.py

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

responses/tests/test_recorder.py

@@ -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,
+                    }
+                },
+            ]
+        }

setup.py

@@ -19,6 +19,8 @@
 install_requires = [
     "requests>=2.0,<3.0",
     "urllib3>=1.25.10",
+    "toml",
+    "types-toml",
     "typing_extensions; python_version < '3.8'",
 ]