dev(narugo): add support for headers and binary body in recorder

.gitignore

@@ -14,3 +14,7 @@ venv
 /.pytest_cache
 /.tox
 /.artifacts
+/test_*
+/*.yaml
+*.bin
+!/example_bins/*.bin

.pre-commit-config.yaml

@@ -1,3 +1,4 @@
+exclude: ^example_bins
 repos:
   - repo: https://github.com/psf/black
     rev: 23.3.0

example_bins/202.bin

@@ -0,0 +1 @@
+OK

example_bins/400.bin

@@ -0,0 +1 @@
+Invalid status code

example_bins/500.bin

@@ -0,0 +1 @@
+500 Internal Server Error

responses/__init__.py

@@ -2,6 +2,7 @@
 import inspect
 import json as json_module
 import logging
+import os
 import socket
 from collections import namedtuple
 from functools import partialmethod
@@ -57,7 +58,6 @@
 
 if TYPE_CHECKING:  # pragma: no cover
     # import only for linter run
-    import os
     from typing import Protocol
     from unittest.mock import _patch as _mock_patcher
 
@@ -802,14 +802,24 @@ def _parse_response_file(
 
     def _add_from_file(self, file_path: "Union[str, bytes, os.PathLike[Any]]") -> None:
         data = self._parse_response_file(file_path)
+        parent_directory = os.path.dirname(os.path.abspath(file_path))
 
         for rsp in data["responses"]:
             rsp = rsp["response"]
+            headers = dict(rsp.get("headers") or {})
+            if "Content-Type" in headers:
+                headers.pop("Content-Type")
+            if "body_file" in rsp:
+                with open(os.path.join(parent_directory, rsp["body_file"]), "rb") as f:
+                    body = f.read()
+            else:
+                body = rsp["body"]
             self.add(
                 method=rsp["method"],
                 url=rsp["url"],
-                body=rsp["body"],
+                body=body,
                 status=rsp["status"],
+                headers=headers,
                 content_type=rsp["content_type"],
                 auto_calculate_content_length=rsp["auto_calculate_content_length"],
             )

responses/_recorder.py

@@ -1,27 +1,27 @@
+import os
+import uuid
 from functools import wraps
 from typing import TYPE_CHECKING
 
 if TYPE_CHECKING:  # pragma: no cover
-    import os
-
     from typing import Any
-    from typing import BinaryIO
     from typing import Callable
     from typing import Dict
     from typing import List
     from typing import Type
     from typing import Union
+    from typing import IO
     from responses import FirstMatchRegistry
     from responses import HTTPAdapter
     from responses import PreparedRequest
     from responses import models
     from responses import _F
     from responses import BaseResponse
 
-    from io import TextIOWrapper
 
 import yaml
 
+from responses import _UNSET
 from responses import RequestsMock
 from responses import Response
 from responses import _real_send
@@ -38,19 +38,44 @@ def _remove_nones(d: "Any") -> "Any":
 
 def _dump(
     registered: "List[BaseResponse]",
-    destination: "Union[BinaryIO, TextIOWrapper]",
-    dumper: "Callable[[Union[Dict[Any, Any], List[Any]], Union[BinaryIO, TextIOWrapper]], Any]",
+    config_file: "Union[str, os.PathLike[str]]",
+    dumper: "Callable[[Union[Dict[Any, Any], List[Any]], Union[IO[Any]]], Any]",
+    dumper_mode: "str" = "w",
 ) -> None:
     data: Dict[str, Any] = {"responses": []}
+
+    # e.g. config_file = 'my/dir/responses.yaml'
+    # parent_directory = 'my/dir'
+    # binary_directory = 'my/dir/responses'
+    fname, fext = os.path.splitext(os.path.basename(config_file))
+    parent_directory = os.path.dirname(os.path.abspath(config_file))
+    binary_directory = os.path.join(
+        parent_directory, fname if fext else f"{fname}_bodies"
+    )
+
     for rsp in registered:
         try:
             content_length = rsp.auto_calculate_content_length  # type: ignore[attr-defined]
+            body = rsp.body  # type: ignore[attr-defined]
+            if isinstance(body, bytes):
+                os.makedirs(binary_directory, exist_ok=True)
+                bin_file = os.path.join(binary_directory, f"{uuid.uuid4()}.bin")
+                with open(bin_file, "wb") as bf:
+                    bf.write(body)
+
+                # make sure the stored binary file path is relative to config file
+                # or the config file and binary directory will be hard to move
+                body_file = os.path.relpath(bin_file, parent_directory)
+                body = None
+            else:
+                body_file = None
             data["responses"].append(
                 {
                     "response": {
                         "method": rsp.method,
                         "url": rsp.url,
-                        "body": rsp.body,  # type: ignore[attr-defined]
+                        "body": body,
+                        "body_file": body_file,
                         "status": rsp.status,  # type: ignore[attr-defined]
                         "headers": rsp.headers,
                         "content_type": rsp.content_type,
@@ -63,7 +88,9 @@ def _dump(
                 "Cannot dump response object."
                 "Probably you use custom Response object that is missing required attributes"
             ) from exc
-    dumper(_remove_nones(data), destination)
+
+    with open(config_file, dumper_mode) as cfile:
+        dumper(_remove_nones(data), cfile)
 
 
 class Recorder(RequestsMock):
@@ -79,7 +106,7 @@ def reset(self) -> None:
         self._registry = OrderedRegistry()
 
     def record(
-        self, *, file_path: "Union[str, bytes, os.PathLike[Any]]" = "response.yaml"
+        self, *, file_path: "Union[str, os.PathLike[str]]" = "response.yaml"
     ) -> "Union[Callable[[_F], _F], _F]":
         def deco_record(function: "_F") -> "Callable[..., Any]":
             @wraps(function)
@@ -99,11 +126,10 @@ def wrapper(*args: "Any", **kwargs: "Any") -> "Any":  # type: ignore[misc]
     def dump_to_file(
         self,
         *,
-        file_path: "Union[str, bytes, os.PathLike[Any]]",
+        file_path: "Union[str, os.PathLike[str]]",
         registered: "List[BaseResponse]",
     ) -> None:
-        with open(file_path, "w") as file:
-            _dump(registered, file, yaml.dump)
+        _dump(registered, file_path, yaml.dump)
 
     def _on_request(
         self,
@@ -116,11 +142,22 @@ def _on_request(
         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)
+        requests_headers = dict(requests_response.headers)
+        if "Content-Type" in requests_headers:
+            requests_content_type = requests_headers.pop("Content-Type")
+        else:
+            requests_content_type = _UNSET  # type: ignore[assignment]
+        # Content-Encoding should be removed to
+        # avoid 'Content-Encoding: gzip' causing the error in requests
+        if "Content-Encoding" in requests_headers:
+            requests_headers.pop("Content-Encoding")
         responses_response = Response(
             method=str(request.method),
             url=str(requests_response.request.url),
             status=requests_response.status_code,
-            body=requests_response.text,
+            headers=requests_headers,
+            body=requests_response.content,
+            content_type=requests_content_type,
         )
         self._registry.add(responses_response)
         return requests_response