fix: preserve exception notes when copying and pickling (#729)

Co-authored-by: Bernát Gábor <gaborjbernat@gmail.com>

Author: jackwalkerlabs

docs/changelog/729.bugfix.rst

@@ -0,0 +1 @@
+Preserve exception notes and custom attributes when copying or pickling ``Timeout`` and ``SoftFileLockProtocolError``.

src/filelock/_error.py

@@ -8,9 +8,9 @@ def __init__(self, lock_file: str) -> None:
         super().__init__()
         self._lock_file = lock_file
 
-    def __reduce__(self) -> tuple[type[Timeout], tuple[str]]:
+    def __reduce__(self) -> tuple[type[Timeout], tuple[str], dict[str, object]]:
         # __init__ needs lock_file, so pickle must restore it as a constructor arg
-        return self.__class__, (self._lock_file,)
+        return self.__class__, (self._lock_file,), self.__dict__
 
     def __str__(self) -> str:  # pragma: needs hard-link
         return f"The file lock '{self._lock_file}' could not be acquired."
@@ -43,8 +43,10 @@ def __init__(self, lock_file: str, claim_name: str | None, reason: str) -> None:
 
     def __reduce__(
         self,
-    ) -> tuple[type[SoftFileLockProtocolError], tuple[str, str | None, str]]:  # pragma: needs hard-link
-        return self.__class__, (self._lock_file, self._claim_name, self._reason)
+    ) -> tuple[
+        type[SoftFileLockProtocolError], tuple[str, str | None, str], dict[str, object]
+    ]:  # pragma: needs hard-link
+        return self.__class__, (self._lock_file, self._claim_name, self._reason), self.__dict__
 
     def __str__(self) -> str:
         location = self._lock_file if self._claim_name is None else f"{self._lock_file}: claim {self._claim_name!r}"

tests/test_error.py

@@ -1,11 +1,13 @@
 from __future__ import annotations
 
-import pickle  # ruff:ignore[suspicious-pickle-import]  # round-trips Timeout to assert it pickles
-from typing import TYPE_CHECKING
+import copy
+import pickle  # ruff:ignore[suspicious-pickle-import]  # round-trips exceptions created in these tests
+from functools import partial
+from typing import TYPE_CHECKING, Final
 
 import pytest
 
-from filelock import Timeout
+from filelock import SoftFileLockProtocolError, Timeout
 
 if TYPE_CHECKING:
     from collections.abc import Callable
@@ -17,23 +19,108 @@
         pytest.param(str, "The file lock '/path/to/lock' could not be acquired.", id="str"),
         pytest.param(repr, "Timeout('/path/to/lock')", id="repr"),
         pytest.param(lambda t: t.lock_file, "/path/to/lock", id="lock_file"),
-        pytest.param(lambda t: t.__reduce__(), (Timeout, ("/path/to/lock",)), id="reduce"),
     ],
 )
-def test_timeout_attribute(timeout: Timeout, extract: Callable[[Timeout], object], expected: object) -> None:
-    assert extract(timeout) == expected
+def test_timeout_attribute(extract: Callable[[Timeout], str], expected: str) -> None:
+    assert extract(Timeout("/path/to/lock")) == expected
 
 
-def test_timeout_pickle(timeout: Timeout) -> None:
-    reloaded = pickle.loads(pickle.dumps(timeout))  # ruff:ignore[suspicious-pickle-usage]  # input is the Timeout built in this test
-    assert (type(reloaded), str(reloaded), repr(reloaded), reloaded.lock_file) == (
-        type(timeout),
-        str(timeout),
-        repr(timeout),
-        timeout.lock_file,
+def test_exception_serialization_preserves_diagnostics(
+    error: Timeout | SoftFileLockProtocolError,
+    clone: Callable[[Timeout | SoftFileLockProtocolError], Timeout | SoftFileLockProtocolError],
+) -> None:
+    # Python 3.10 supports the notes attribute but has no add_note().
+    error.__notes__ = ["while acquiring the build cache"]
+    vars(error)["request_id"] = "request-1"
+    restored: Final = clone(error)
+    assert (
+        type(restored),
+        str(restored),
+        repr(restored),
+        restored.args,
+        restored.lock_file,
+        restored.__notes__,
+        vars(restored)["request_id"],
+    ) == (
+        type(error),
+        str(error),
+        repr(error),
+        error.args,
+        error.lock_file,
+        ["while acquiring the build cache"],
+        "request-1",
     )
 
 
-@pytest.fixture
-def timeout() -> Timeout:
-    return Timeout("/path/to/lock")
+def test_exception_serialization_creates_distinct_instance(
+    error: Timeout | SoftFileLockProtocolError,
+    clone: Callable[[Timeout | SoftFileLockProtocolError], Timeout | SoftFileLockProtocolError],
+) -> None:
+    assert clone(error) is not error
+
+
+def test_exception_serialization_preserves_note_alias(
+    error: Timeout | SoftFileLockProtocolError,
+    clone: Callable[[Timeout | SoftFileLockProtocolError], Timeout | SoftFileLockProtocolError],
+) -> None:
+    error.__notes__ = ["original"]
+    vars(error)["diagnostics"] = error.__notes__
+    restored: Final = clone(error)
+    assert vars(restored)["diagnostics"] is restored.__notes__
+
+
+def test_exception_copy_shares_notes(error: Timeout | SoftFileLockProtocolError) -> None:
+    error.__notes__ = ["original"]
+    copy.copy(error).__notes__.append("copy")
+    assert error.__notes__ == ["original", "copy"]
+
+
+def test_exception_deepcopy_isolates_notes(error: Timeout | SoftFileLockProtocolError) -> None:
+    error.__notes__ = ["original"]
+    copy.deepcopy(error).__notes__.append("copy")
+    assert error.__notes__ == ["original"]
+
+
+def test_exception_serialization_preserves_cycle(
+    error: Timeout | SoftFileLockProtocolError,
+    clone: Callable[[Timeout | SoftFileLockProtocolError], Timeout | SoftFileLockProtocolError],
+) -> None:
+    vars(error)["related"] = error
+    restored: Final = clone(error)
+    assert vars(restored)["related"] is (error if clone is copy.copy else restored)
+
+
+def test_protocol_error_serialization_preserves_claim(
+    clone: Callable[[Timeout | SoftFileLockProtocolError], Timeout | SoftFileLockProtocolError],
+) -> None:
+    restored: Final = clone(SoftFileLockProtocolError("/lock", "claim", "invalid marker"))
+    assert isinstance(restored, SoftFileLockProtocolError)
+    assert (restored.claim_name, restored.reason) == ("claim", "invalid marker")
+
+
+@pytest.fixture(
+    params=[pytest.param("timeout", id="timeout"), pytest.param("claim", id="claim"), pytest.param(None, id="no-claim")]
+)
+def error(request: pytest.FixtureRequest) -> Timeout | SoftFileLockProtocolError:
+    if request.param == "timeout":
+        return Timeout("/path/to/lock")
+    return SoftFileLockProtocolError("/path/to/lock", request.param, "invalid marker")
+
+
+def _pickle_round_trip(
+    error: Timeout | SoftFileLockProtocolError, *, protocol: int
+) -> Timeout | SoftFileLockProtocolError:
+    return pickle.loads(pickle.dumps(error, protocol=protocol))  # ruff:ignore[suspicious-pickle-usage]  # input is created by these tests
+
+
+@pytest.fixture(
+    params=[pytest.param(copy.copy, id="copy"), pytest.param(copy.deepcopy, id="deepcopy")]
+    + [
+        pytest.param(partial(_pickle_round_trip, protocol=p), id=f"pickle-{p}")
+        for p in range(pickle.HIGHEST_PROTOCOL + 1)
+    ],
+)
+def clone(
+    request: pytest.FixtureRequest,
+) -> Callable[[Timeout | SoftFileLockProtocolError], Timeout | SoftFileLockProtocolError]:
+    return request.param