Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions CHANGES
Original file line number Diff line number Diff line change
@@ -1,6 +1,10 @@
0.26.3
------

* Added a ``keep_headers`` argument to ``_recorder.record`` and
``Recorder.dump_to_file`` to preserve named headers (for example ``Date``)
that are otherwise stripped as verbose defaults, so a signed response can be
recorded and later verified. Matching is case-insensitive. See #763
* Fixed the element type exposed by `CallList` so static type checkers infer
`Call` values when iterating, indexing, or filtering recorded calls. See #722
* Fixed `query_string_matcher` (and the query matching auto-applied to a
Expand Down
11 changes: 11 additions & 0 deletions README.rst
Original file line number Diff line number Diff line change
Expand Up @@ -1401,6 +1401,17 @@ will produce next output:
status: 202
url: https://httpstat.us/202

Common headers such as ``Content-Type``, ``Date`` and ``Server`` are stripped
from the recording to keep the file terse. If you need to keep one of them, for
example a ``Date`` value that is part of a signed response you later verify,
pass its name in ``keep_headers`` (matched case-insensitively):

.. code-block:: python

@_recorder.record(file_path="out.yaml", keep_headers=["Date"])
def test_recorder():
...

If you are in the REPL, you can also activate the recorder for all following responses:

.. code-block:: python
Expand Down
43 changes: 36 additions & 7 deletions responses/_recorder.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@
from typing import BinaryIO
from typing import Callable
from typing import Dict
from typing import Iterable
from typing import List
from typing import Optional
from typing import Type
Expand Down Expand Up @@ -38,11 +39,22 @@ def _remove_nones(d: "Any") -> "Any":
return d


def _remove_default_headers(data: "Any") -> "Any":
def _remove_default_headers(
data: "Any", keep_headers: "Optional[Iterable[str]]" = None
) -> "Any":
"""
It would be too verbose to store these headers in the file generated by the
record functionality.

Header names passed in ``keep_headers`` are excluded from the default
stripping, so a caller can preserve an otherwise-removed header (for
example a ``Date`` value that is part of a signed response). Matching is
case-insensitive, consistent with the rest of this function.
"""
# A bare string is iterable over its characters, which would silently keep
# nothing useful; treat it as a single header name.
if isinstance(keep_headers, str):
keep_headers = [keep_headers]
if isinstance(data, dict):
keys_to_remove = [
"Content-Length",
Expand All @@ -54,7 +66,8 @@ def _remove_default_headers(data: "Any") -> "Any":
]
# HTTP header names are case-insensitive, and HTTP/2 servers send them
# lowercase, so match without regard to case.
keys_to_remove_lower = {key.lower() for key in keys_to_remove}
keep_lower = {key.lower() for key in keep_headers or ()}
keys_to_remove_lower = {key.lower() for key in keys_to_remove} - keep_lower
for i, response in enumerate(data["responses"]):
headers = data["responses"][i]["response"]["headers"]
for key in list(headers):
Expand All @@ -69,6 +82,7 @@ def _dump(
registered: "List[BaseResponse]",
destination: "Union[BinaryIO, TextIOWrapper]",
dumper: "Callable[[Union[Dict[Any, Any], List[Any]], Union[BinaryIO, TextIOWrapper]], Any]",
keep_headers: "Optional[Iterable[str]]" = None,
) -> None:
data: Dict[str, Any] = {"responses": []}
for rsp in registered:
Expand All @@ -93,7 +107,10 @@ def _dump(
"Probably you use custom Response object that is missing required attributes"
) from exc

dumper(_remove_default_headers(_remove_nones(data)), destination)
dumper(
_remove_default_headers(_remove_nones(data), keep_headers=keep_headers),
destination,
)


class Recorder(RequestsMock):
Expand All @@ -109,15 +126,20 @@ 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, bytes, os.PathLike[Any]]" = "response.yaml",
keep_headers: "Optional[Iterable[str]]" = None,
) -> "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)
self.dump_to_file(
file_path=file_path, registered=self.get_registry().registered
file_path=file_path,
registered=self.get_registry().registered,
keep_headers=keep_headers,
)

return ret
Expand All @@ -131,12 +153,19 @@ def dump_to_file(
file_path: "Union[str, bytes, os.PathLike[Any]]",
*,
registered: "Optional[List[BaseResponse]]" = None,
keep_headers: "Optional[Iterable[str]]" = None,
) -> None:
"""Dump the recorded responses to a file."""
"""Dump the recorded responses to a file.

Header names listed in ``keep_headers`` are preserved in the output
even though they are normally stripped as verbose defaults (for
example ``Date`` when it is part of a signed response). Matching is
case-insensitive.
"""
if registered is None:
registered = self.get_registry().registered
with open(file_path, "w") as file:
_dump(registered, file, yaml.dump)
_dump(registered, file, yaml.dump, keep_headers=keep_headers)

def _on_request(
self,
Expand Down
106 changes: 104 additions & 2 deletions responses/tests/test_recorder.py
Original file line number Diff line number Diff line change
Expand Up @@ -106,6 +106,63 @@ def test_remove_default_headers_is_case_insensitive():
assert "headers" not in result["responses"][1]["response"]


def test_remove_default_headers_keeps_requested_headers():
"""Headers named in ``keep_headers`` survive the default-header stripping.

Some consumers need a normally-stripped header in the recorded file, for
example a ``Date`` value that is part of a signed response they later
verify. Matching is case-insensitive so a lowercase HTTP/2 name is kept
too, while the other default headers are still removed.
"""
data = {
"responses": [
{
"response": {
"headers": {
"date": "Mon, 01 Jan 2024 00:00:00 GMT",
"Server": "nginx",
"Content-Length": "12",
"x-custom": "keep-me",
}
}
},
]
}

result = _remove_default_headers(data, keep_headers=["Date"])

assert result["responses"][0]["response"]["headers"] == {
"date": "Mon, 01 Jan 2024 00:00:00 GMT",
"x-custom": "keep-me",
}


def test_remove_default_headers_accepts_a_single_string():
"""A bare string is treated as one header name, not a set of characters."""
data = {
"responses": [
{"response": {"headers": {"Date": "d", "Server": "nginx"}}},
]
}

result = _remove_default_headers(data, keep_headers="Date")

assert result["responses"][0]["response"]["headers"] == {"Date": "d"}


def test_remove_default_headers_keep_non_default_is_noop():
"""Naming a non-default header does not change which headers are stripped."""
data = {
"responses": [
{"response": {"headers": {"Date": "d", "x-custom": "keep-me"}}},
]
}

result = _remove_default_headers(data, keep_headers=["x-custom"])

assert result["responses"][0]["response"]["headers"] == {"x-custom": "keep-me"}


class TestRecord:
def setup_method(self):
self.out_file = Path("response_record")
Expand Down Expand Up @@ -133,12 +190,57 @@ def run():
data = yaml.safe_load(file)
assert data == get_data(httpserver.host, httpserver.port)

def test_recorder_keep_headers_preserves_date(self, httpserver):
"""``keep_headers`` keeps a normally-stripped header in the record.

Reproduces the reported need to record the ``Date`` header so a signed
response can be verified against the recording (issue #763). Without
``keep_headers`` the same header is stripped as a verbose default.
"""
httpserver.expect_request("/signed").respond_with_data(
"ok",
status=200,
content_type="text/plain",
headers={"Date": "Mon, 01 Jan 2024 00:00:00 GMT"},
)
url = httpserver.url_for("/signed")

@_recorder.record(file_path=self.out_file, keep_headers=["Date"])
def run_kept():
requests.get(url)

run_kept()
with open(self.out_file) as file:
kept = yaml.safe_load(file)
kept_headers = kept["responses"][0]["response"]["headers"]
# The server owns the exact Date value, so assert the header survives
# rather than pinning its contents.
assert "Date" in kept_headers
assert "2024" in kept_headers["Date"]

self.out_file.unlink()

@_recorder.record(file_path=self.out_file)
def run_default():
requests.get(url)

run_default()
with open(self.out_file) as file:
default = yaml.safe_load(file)
default_headers = default["responses"][0]["response"].get("headers", {})
assert "Date" not in default_headers

def test_recorder_toml(self, httpserver):
custom_recorder = _recorder.Recorder()

def dump_to_file(file_path, registered):
def dump_to_file(file_path, registered, *, keep_headers=None):
with open(file_path, "wb") as file:
_dump(registered, file, tomli_w.dump) # type: ignore[arg-type]
_dump(
registered,
file,
tomli_w.dump, # type: ignore[arg-type]
keep_headers=keep_headers,
)

custom_recorder.dump_to_file = dump_to_file # type: ignore[assignment]

Expand Down