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
3 changes: 3 additions & 0 deletions CHANGES
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,9 @@
* Fixed `fragment_identifier_matcher` treating opaque fragments (those without
``=``, e.g. ``/users/5``) as always equal, so a required fragment matched a
different one or none at all. See #806
* Fixed the recorder decorator not writing captured responses when the decorated
function raises. Dump failures during exception handling are logged without
replacing the original exception. See #705

0.26.2
------
Expand Down
28 changes: 22 additions & 6 deletions responses/_recorder.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
import logging
from functools import wraps
from typing import TYPE_CHECKING

Expand Down Expand Up @@ -29,6 +30,8 @@
from responses import _real_send
from responses.registries import OrderedRegistry

logger = logging.getLogger("responses")


def _remove_nones(d: "Any") -> "Any":
if isinstance(d, dict):
Expand Down Expand Up @@ -115,12 +118,25 @@ 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
)

return ret
function_raised = False
try:
return function(*args, **kwargs)
except BaseException:

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why use BaseException? Should dump files be written to if there are syntax errors?

function_raised = True
raise
finally:
try:
self.dump_to_file(
file_path=file_path,
registered=self.get_registry().registered,
)
except BaseException:
if not function_raised:
raise
logger.exception(
"Failed to dump recorded responses while handling "
"an exception from the decorated function"
)

return wrapper

Expand Down
57 changes: 57 additions & 0 deletions responses/tests/test_recorder.py
Original file line number Diff line number Diff line change
Expand Up @@ -133,6 +133,63 @@ def run():
data = yaml.safe_load(file)
assert data == get_data(httpserver.host, httpserver.port)

def test_recorder_dumps_when_decorated_function_raises(self, httpserver):
_, _, url404, _ = self.prepare_server(httpserver)
original_error = RuntimeError("original failure")

@_recorder.record(file_path=self.out_file)
def run():
requests.get(url404)
raise original_error

with pytest.raises(RuntimeError) as exc_info:
run()

assert exc_info.value is original_error
with open(self.out_file) as file:
data = yaml.safe_load(file)
assert (
data["responses"]
== get_data(httpserver.host, httpserver.port)["responses"][:1]
)

def test_dump_failure_does_not_mask_function_exception(self, httpserver, caplog):
custom_recorder = _recorder.Recorder()
_, _, url404, _ = self.prepare_server(httpserver)
original_error = RuntimeError("original failure")

def dump_to_file(*args, **kwargs):
raise OSError("dump failure")

custom_recorder.dump_to_file = dump_to_file # type: ignore[method-assign]

@custom_recorder.record(file_path=self.out_file)
def run():
requests.get(url404)
raise original_error

with caplog.at_level("ERROR", logger="responses"):
with pytest.raises(RuntimeError) as exc_info:
run()

assert exc_info.value is original_error
assert "Failed to dump recorded responses" in caplog.text

def test_dump_failure_is_raised_when_function_succeeds(self):
custom_recorder = _recorder.Recorder()

def dump_to_file(*args, **kwargs):
raise OSError("dump failure")

custom_recorder.dump_to_file = dump_to_file # type: ignore[method-assign]

@custom_recorder.record(file_path=self.out_file)
def run():
return None
Comment on lines +186 to +188

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Shouldn't success include a captured request?


with pytest.raises(OSError, match="dump failure"):
run()

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

Expand Down