github/deer-flow
1
0
Fork 0
mirror of https://github.com/bytedance/deer-flow.git synced 2026-09-15 00:19:14 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: lease published upload generations

Browse Source
This commit is contained in:
hetaoBackend 2026-08-06 20:03:06 +08:00
parent 3aff322f23
commit 01c5c2099f
6 changed files with 515 additions and 50 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
backend/packages/harness/deerflow/uploads/__init__.py
View File

@ -1,20 +1,23 @@
from .conversion import convert_uploaded_file_to_markdown
from .errors import AtomicUploadPublishError, PathTraversalError, UnsafeUploadPathError
from .layout import (
UPLOAD_CONVERSIONS_DIRNAME,
UPLOAD_LOCKS_DIRNAME,
UnsafeConversionPathError,
artifact_url_for_virtual_path,
conversion_dir_for_uploads,
conversion_path_for_upload,
conversion_virtual_path,
ensure_conversion_dir,
ensure_upload_lock_dir,
existing_conversion_path_for_upload,
validate_conversion_dir,
)
from .lease import UploadIdentity, UploadNameLease
from .manager import (
UPLOAD_STAGING_PREFIX,
UPLOAD_STAGING_SUFFIX,
AtomicUploadPublishError,
PathTraversalError,
PublishedUpload,
StagedUpload,
abort_staged_upload,
claim_unique_filename,
@ -28,9 +31,13 @@ from .manager import (
list_files_in_dir,
normalize_filename,
publish_staged_upload,
publish_staged_upload_leased,
publish_upload_bytes,
publish_upload_bytes_leased,
publish_upload_copy,
publish_upload_copy_leased,
replace_system_owned_staged_file,
rollback_published_upload,
upload_artifact_url,
upload_virtual_path,
validate_path_traversal,
@ -40,6 +47,7 @@ from .manager import (
__all__ = [
"get_uploads_dir",
"UPLOAD_CONVERSIONS_DIRNAME",
"UPLOAD_LOCKS_DIRNAME",
"UnsafeConversionPathError",
"conversion_dir_for_uploads",
"conversion_path_for_upload",
@ -47,21 +55,30 @@ __all__ = [
"artifact_url_for_virtual_path",
"validate_conversion_dir",
"ensure_conversion_dir",
"ensure_upload_lock_dir",
"existing_conversion_path_for_upload",
"convert_uploaded_file_to_markdown",
"ensure_uploads_dir",
"normalize_filename",
"PathTraversalError",
"UnsafeUploadPathError",
"AtomicUploadPublishError",
"UploadIdentity",
"UploadNameLease",
"StagedUpload",
"PublishedUpload",
"UPLOAD_STAGING_PREFIX",
"UPLOAD_STAGING_SUFFIX",
"claim_unique_filename",
"create_upload_staging_file",
"abort_staged_upload",
"publish_staged_upload",
"publish_staged_upload_leased",
"publish_upload_bytes",
"publish_upload_bytes_leased",
"publish_upload_copy",
"publish_upload_copy_leased",
"rollback_published_upload",
"replace_system_owned_staged_file",
"cleanup_stale_upload_staging_files",
"is_upload_staging_file",

13
backend/packages/harness/deerflow/uploads/errors.py Normal file
View File

@ -0,0 +1,13 @@
"""Shared upload validation and publication errors."""
class PathTraversalError(ValueError):
"""Raised when a path escapes its allowed base directory."""
class UnsafeUploadPathError(ValueError):
"""Raised when an upload destination is not a safe regular file path."""
class AtomicUploadPublishError(UnsafeUploadPathError):
"""Raised when storage cannot honor atomic no-replace publication."""

23
backend/packages/harness/deerflow/uploads/layout.py
View File

@ -8,6 +8,7 @@ from urllib.parse import quote
from deerflow.config.paths import VIRTUAL_PATH_PREFIX
UPLOAD_CONVERSIONS_DIRNAME = ".upload-conversions"
UPLOAD_LOCKS_DIRNAME = ".locks"
class UnsafeConversionPathError(ValueError):
@ -49,6 +50,28 @@ def ensure_conversion_dir(uploads_dir: Path) -> Path:
return validated
def upload_lock_dir_for_uploads(uploads_dir: Path) -> Path:
"""Return the stable per-upload lock directory."""
return conversion_dir_for_uploads(uploads_dir) / UPLOAD_LOCKS_DIRNAME
def ensure_upload_lock_dir(uploads_dir: Path) -> Path:
"""Create and validate the system-owned upload lock directory."""
conversion_dir = ensure_conversion_dir(uploads_dir)
lock_dir = conversion_dir / UPLOAD_LOCKS_DIRNAME
try:
lock_dir.mkdir(mode=0o700)
except FileExistsError:
pass
try:
lock_stat = os.lstat(lock_dir)
except FileNotFoundError as exc:
raise UnsafeConversionPathError("Upload lock directory disappeared") from exc
if stat.S_ISLNK(lock_stat.st_mode) or not stat.S_ISDIR(lock_stat.st_mode):
raise UnsafeConversionPathError("Unsafe upload lock directory")
return lock_dir
def existing_conversion_path_for_upload(upload_path: Path) -> Path | None:
"""Return an existing safe generated file owned by ``upload_path``."""
if validate_conversion_dir(upload_path.parent) is None:

159
backend/packages/harness/deerflow/uploads/lease.py Normal file
View File

@ -0,0 +1,159 @@
"""Cross-process leases and inode identities for published uploads."""
import hashlib
import os
import stat
import threading
from dataclasses import dataclass, field
from pathlib import Path
from typing import BinaryIO
from deerflow.uploads.errors import UnsafeUploadPathError
from deerflow.uploads.layout import ensure_upload_lock_dir
try:
import fcntl
except ImportError: # pragma: no cover - Windows only
fcntl = None # type: ignore[assignment]
import msvcrt
_LOCK_STRIPES = tuple(threading.Lock() for _ in range(64))
@dataclass(frozen=True, slots=True)
class UploadIdentity:
"""Filesystem identity of one published upload generation."""
device: int
inode: int
@classmethod
def from_path(cls, path: Path) -> "UploadIdentity":
upload_stat = os.lstat(path)
if not stat.S_ISREG(upload_stat.st_mode):
raise UnsafeUploadPathError("Published upload is not a regular file")
return cls(device=upload_stat.st_dev, inode=upload_stat.st_ino)
def matches(self, path: Path) -> bool:
"""Return whether *path* still names this generation."""
try:
upload_stat = os.lstat(path)
except FileNotFoundError:
return False
return stat.S_ISREG(upload_stat.st_mode) and (upload_stat.st_dev, upload_stat.st_ino) == (
self.device,
self.inode,
)
def _open_lock_file(lock_path: Path) -> BinaryIO:
flags = os.O_RDWR | os.O_CREAT
if hasattr(os, "O_BINARY"):
flags |= os.O_BINARY
if hasattr(os, "O_NOFOLLOW"):
flags |= os.O_NOFOLLOW
fd = os.open(lock_path, flags, 0o600)
try:
descriptor_stat = os.fstat(fd)
path_stat = os.lstat(lock_path)
if not stat.S_ISREG(descriptor_stat.st_mode) or descriptor_stat.st_nlink != 1 or not stat.S_ISREG(path_stat.st_mode) or (descriptor_stat.st_dev, descriptor_stat.st_ino) != (path_stat.st_dev, path_stat.st_ino):
raise UnsafeUploadPathError("Unsafe upload lock file")
if descriptor_stat.st_size == 0:
os.write(fd, b"\0")
os.lseek(fd, 0, os.SEEK_SET)
return os.fdopen(fd, "r+b", buffering=0)
except BaseException:
os.close(fd)
raise
def _lock_file(lock_file: BinaryIO) -> None:
if fcntl is not None:
fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX)
return
lock_file.seek(0)
msvcrt.locking(lock_file.fileno(), msvcrt.LK_LOCK, 1)
def _unlock_file(lock_file: BinaryIO) -> None:
if fcntl is not None:
fcntl.flock(lock_file.fileno(), fcntl.LOCK_UN)
return
lock_file.seek(0)
msvcrt.locking(lock_file.fileno(), msvcrt.LK_UNLCK, 1)
@dataclass(slots=True)
class UploadNameLease:
"""Exclusive thread-and-process lease for one actual upload filename."""
uploads_dir: Path
filename: str
lock_path: Path
_lock_file: BinaryIO
_stripe: threading.Lock
_active: bool = True
_state_lock: threading.Lock = field(default_factory=threading.Lock)
@classmethod
def acquire(cls, uploads_dir: Path, filename: str) -> "UploadNameLease":
"""Acquire the stable name lease, blocking until it is available."""
if not filename or Path(filename).name != filename or "\\" in filename:
raise UnsafeUploadPathError(f"Unsafe upload lease filename: {filename!r}")
if len(filename.encode("utf-8")) > 255:
raise UnsafeUploadPathError("Upload lease filename is too long")
digest = hashlib.sha256(filename.encode("utf-8")).hexdigest()
stripe = _LOCK_STRIPES[int(digest[:2], 16) % len(_LOCK_STRIPES)]
stripe.acquire()
lock_file: BinaryIO | None = None
try:
lock_path = ensure_upload_lock_dir(Path(uploads_dir)) / f"{digest}.lock"
lock_file = _open_lock_file(lock_path)
_lock_file(lock_file)
return cls(
uploads_dir=Path(uploads_dir),
filename=filename,
lock_path=lock_path,
_lock_file=lock_file,
_stripe=stripe,
)
except BaseException:
if lock_file is not None:
lock_file.close()
stripe.release()
raise
@property
def is_active(self) -> bool:
"""Return whether this object still owns the name lease."""
with self._state_lock:
return self._active
def release(self) -> None:
"""Release the OS lock and process stripe; repeated calls are harmless."""
with self._state_lock:
if not self._active:
return
error: BaseException | None = None
try:
_unlock_file(self._lock_file)
except BaseException as exc: # pragma: no cover - exceptional OS failure
error = exc
try:
self._lock_file.close()
except BaseException as exc: # pragma: no cover - exceptional OS failure
if error is None:
error = exc
finally:
self._active = False
self._stripe.release()
if error is not None:
raise error
def __enter__(self) -> "UploadNameLease":
return self
def __exit__(self, exc_type, exc_value, traceback) -> None:
self.release()

208
backend/packages/harness/deerflow/uploads/manager.py
View File

@ -17,27 +17,16 @@ from typing import BinaryIO
from deerflow.config.paths import get_paths
from deerflow.runtime.user_context import get_effective_user_id
from deerflow.uploads.errors import AtomicUploadPublishError, PathTraversalError, UnsafeUploadPathError
from deerflow.uploads.layout import (
UPLOAD_CONVERSIONS_DIRNAME,
artifact_url_for_virtual_path,
existing_conversion_path_for_upload,
upload_virtual_path,
)
from deerflow.uploads.lease import UploadIdentity, UploadNameLease
from deerflow.utils.thread_id import validate_thread_id
class PathTraversalError(ValueError):
"""Raised when a path escapes its allowed base directory."""
class UnsafeUploadPathError(ValueError):
"""Raised when an upload destination is not a safe regular file path."""
class AtomicUploadPublishError(UnsafeUploadPathError):
"""Raised when storage cannot honor atomic no-replace publication."""
logger = logging.getLogger(__name__)
UPLOAD_STAGING_PREFIX = ".upload-"
@ -53,6 +42,23 @@ class StagedUpload:
handle: BinaryIO
@dataclass(slots=True)
class PublishedUpload:
"""A published upload whose actual filename remains exclusively leased."""
path: Path
identity: UploadIdentity
lease: UploadNameLease
@property
def is_active(self) -> bool:
return self.lease.is_active
def release(self) -> None:
"""Release the publication's name lease."""
self.lease.release()
def get_uploads_dir(thread_id: str, *, user_id: str | None = None) -> Path:
"""Return the uploads directory path for a thread (no side effects)."""
validate_thread_id(thread_id)
@ -91,6 +97,8 @@ def normalize_filename(filename: str) -> str:
raise ValueError(f"Filename contains backslash: {filename!r}")
if len(safe.encode("utf-8")) > 255:
raise ValueError(f"Filename too long: {len(safe)} chars")
if is_upload_staging_file(safe):
raise ValueError(f"Filename uses reserved upload staging pattern: {filename!r}")
return safe
@ -155,9 +163,20 @@ def create_upload_staging_file(base_dir: Path) -> StagedUpload:
def abort_staged_upload(staged: StagedUpload) -> None:
"""Close and remove a staging file, tolerating repeated cleanup."""
if not staged.handle.closed:
staged.handle.close()
staged.path.unlink(missing_ok=True)
close_error: BaseException | None = None
try:
if not staged.handle.closed:
staged.handle.close()
except BaseException as exc:
close_error = exc
try:
staged.path.unlink(missing_ok=True)
except BaseException as unlink_error:
if close_error is not None:
raise close_error from unlink_error
raise
if close_error is not None:
raise close_error
def _validate_staged_upload(staged: StagedUpload) -> None:
@ -197,57 +216,146 @@ def _filename_candidates(name: str) -> Iterator[str]:
counter += 1
def publish_staged_upload(staged: StagedUpload, preferred_filename: str) -> Path:
"""Atomically publish a complete staging file without replacing an entry."""
def _unlink_matching_upload(path: Path, identity: UploadIdentity) -> None:
"""Remove *path* only while it still names *identity*."""
if identity.matches(path):
path.unlink(missing_ok=True)
def _rollback_link_without_masking(path: Path, identity: UploadIdentity) -> None:
try:
_unlink_matching_upload(path, identity)
except BaseException:
logger.warning("Failed to roll back partially published upload: %s", path, exc_info=True)
def _release_lease_without_masking(lease: UploadNameLease) -> None:
try:
lease.release()
except BaseException:
logger.warning("Failed to release upload name lease: %s", lease.lock_path, exc_info=True)
def publish_staged_upload_leased(staged: StagedUpload, preferred_filename: str) -> PublishedUpload:
"""Atomically publish a staging file and retain its actual-name lease."""
safe_name = normalize_filename(preferred_filename)
if not staged.handle.closed:
staged.handle.close()
try:
_validate_staged_upload(staged)
for candidate_name in _filename_candidates(safe_name):
candidate = staged.base_dir / candidate_name
_validate_staged_upload(staged)
staged_identity = UploadIdentity.from_path(staged.path)
for candidate_name in _filename_candidates(safe_name):
lease = UploadNameLease.acquire(staged.base_dir, candidate_name)
candidate = staged.base_dir / candidate_name
linked = False
try:
try:
os.link(staged.path, candidate, follow_symlinks=False)
linked = True
except FileExistsError:
lease.release()
continue
except (NotImplementedError, TypeError) as exc:
raise AtomicUploadPublishError("Storage does not support atomic no-replace publication") from exc
except OSError as exc:
if exc.errno == errno.EEXIST:
lease.release()
continue
raise AtomicUploadPublishError(f"Storage does not support atomic no-replace publication: {exc}") from exc
if not staged_identity.matches(candidate):
raise AtomicUploadPublishError("Published upload identity changed during publication")
try:
staged.path.unlink()
except OSError:
logger.warning("Failed to remove published upload staging link: %s", staged.path, exc_info=True)
return candidate
except Exception:
staged.path.unlink(missing_ok=True)
except OSError as exc:
_rollback_link_without_masking(candidate, staged_identity)
raise AtomicUploadPublishError("Failed to remove upload staging link after publication") from exc
try:
candidate_stat = os.lstat(candidate)
except FileNotFoundError as exc:
raise AtomicUploadPublishError("Published upload disappeared") from exc
if not stat.S_ISREG(candidate_stat.st_mode) or candidate_stat.st_nlink != 1 or not staged_identity.matches(candidate):
_rollback_link_without_masking(candidate, staged_identity)
raise AtomicUploadPublishError("Published upload did not become an exclusive regular file")
return PublishedUpload(path=candidate, identity=staged_identity, lease=lease)
except BaseException:
if linked:
_rollback_link_without_masking(candidate, staged_identity)
_release_lease_without_masking(lease)
raise
def publish_staged_upload(staged: StagedUpload, preferred_filename: str) -> Path:
"""Atomically publish a complete staging file without replacing an entry."""
publication = publish_staged_upload_leased(staged, preferred_filename)
try:
return publication.path
finally:
publication.release()
def _abort_staged_upload_without_masking(staged: StagedUpload) -> None:
try:
abort_staged_upload(staged)
except BaseException:
logger.warning("Failed to clean up upload staging file: %s", staged.path, exc_info=True)
def publish_upload_bytes_leased(base_dir: Path, preferred_filename: str, data: bytes) -> PublishedUpload:
"""Stage bytes, publish them atomically, and retain the actual-name lease."""
safe_name = normalize_filename(preferred_filename)
staged = create_upload_staging_file(base_dir)
try:
staged.handle.write(data)
return publish_staged_upload_leased(staged, safe_name)
except BaseException:
_abort_staged_upload_without_masking(staged)
raise
def publish_upload_bytes(base_dir: Path, preferred_filename: str, data: bytes) -> Path:
"""Stage and atomically publish an in-memory upload payload."""
publication = publish_upload_bytes_leased(base_dir, preferred_filename, data)
try:
return publication.path
finally:
publication.release()
def publish_upload_copy_leased(base_dir: Path, preferred_filename: str, source_path: Path) -> PublishedUpload:
"""Copy a source into staging, publish it, and retain the actual-name lease."""
safe_name = normalize_filename(preferred_filename)
staged = create_upload_staging_file(base_dir)
try:
staged.handle.write(data)
return publish_staged_upload(staged, preferred_filename)
except Exception:
abort_staged_upload(staged)
with Path(source_path).open("rb") as source:
shutil.copyfileobj(source, staged.handle)
return publish_staged_upload_leased(staged, safe_name)
except BaseException:
_abort_staged_upload_without_masking(staged)
raise
def publish_upload_copy(base_dir: Path, preferred_filename: str, source_path: Path) -> Path:
"""Copy a local source into staging and atomically publish it."""
staged = create_upload_staging_file(base_dir)
publication = publish_upload_copy_leased(base_dir, preferred_filename, source_path)
try:
with Path(source_path).open("rb") as source:
shutil.copyfileobj(source, staged.handle)
return publish_staged_upload(staged, preferred_filename)
except Exception:
abort_staged_upload(staged)
raise
return publication.path
finally:
publication.release()
def rollback_published_upload(publication: PublishedUpload) -> None:
"""Remove only the still-leased upload generation represented by *publication*."""
if not publication.is_active:
raise RuntimeError("Cannot roll back a publication after releasing its lease")
if publication.lease.filename != publication.path.name:
raise UnsafeUploadPathError("Publication lease does not match its upload path")
if not publication.identity.matches(publication.path):
return
owned_conversion = existing_conversion_path_for_upload(publication.path)
publication.path.unlink()
if owned_conversion is not None:
owned_conversion.unlink(missing_ok=True)
def replace_system_owned_staged_file(staged: StagedUpload, filename: str) -> Path:
@ -393,18 +501,22 @@ def delete_file_safe(base_dir: Path, filename: str) -> dict:
if isinstance(exc.__cause__, FileNotFoundError):
raise FileNotFoundError(f"File not found: {filename}") from exc
raise
file_path = base_dir / safe_name
lease = UploadNameLease.acquire(base_dir, safe_name)
try:
file_stat = os.lstat(file_path)
except FileNotFoundError:
raise FileNotFoundError(f"File not found: {filename}")
if not stat.S_ISREG(file_stat.st_mode) or file_stat.st_nlink != 1:
raise UnsafeUploadPathError(f"Unsafe upload file: {safe_name}")
file_path = base_dir / safe_name
try:
file_stat = os.lstat(file_path)
except FileNotFoundError:
raise FileNotFoundError(f"File not found: {filename}") from None
if not stat.S_ISREG(file_stat.st_mode) or file_stat.st_nlink != 1:
raise UnsafeUploadPathError(f"Unsafe upload file: {safe_name}")
owned_conversion = existing_conversion_path_for_upload(file_path)
file_path.unlink()
if owned_conversion is not None:
owned_conversion.unlink(missing_ok=True)
owned_conversion = existing_conversion_path_for_upload(file_path)
file_path.unlink()
if owned_conversion is not None:
owned_conversion.unlink(missing_ok=True)
finally:
lease.release()
return {"success": True, "message": f"Deleted {filename}"}

141
backend/tests/test_uploads_manager.py
View File

@ -1,8 +1,13 @@
"""Tests for deerflow.uploads.manager — shared upload management logic."""
import errno
import multiprocessing
import os
import threading
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
from queue import Empty
from typing import Any
from unittest.mock import patch
import pytest
@ -16,17 +21,39 @@ from deerflow.uploads.manager import (
AtomicUploadPublishError,
PathTraversalError,
UnsafeUploadPathError,
abort_staged_upload,
claim_unique_filename,
cleanup_stale_upload_staging_files,
create_upload_staging_file,
delete_file_safe,
list_files_in_dir,
normalize_filename,
publish_staged_upload,
publish_upload_bytes,
publish_upload_bytes_leased,
publish_upload_copy,
rollback_published_upload,
validate_path_traversal,
write_upload_file_no_symlink,
)
def _delete_upload_in_process(
uploads_dir: str,
filename: str,
started: Any,
finished: Any,
errors: Any,
) -> None:
try:
started.set()
delete_file_safe(Path(uploads_dir), filename)
except BaseException as exc: # pragma: no cover - surfaced in the parent
errors.put(repr(exc))
finally:
finished.set()
# ---------------------------------------------------------------------------
# normalize_filename
# ---------------------------------------------------------------------------
@ -119,6 +146,120 @@ class TestValidatePathTraversal:
class TestUploadPublication:
def test_reserved_staging_name_is_rejected_before_stage_creation(self, tmp_path):
with patch("deerflow.uploads.manager.create_upload_staging_file") as create_stage:
with pytest.raises(ValueError, match="reserved"):
publish_upload_bytes(tmp_path, ".upload-user.part", b"payload")
create_stage.assert_not_called()
def test_leased_publication_blocks_delete_until_release(self, tmp_path):
publication = publish_upload_bytes_leased(tmp_path, "report.pdf", b"old")
started = threading.Event()
finished = threading.Event()
def delete():
started.set()
delete_file_safe(tmp_path, "report.pdf")
finished.set()
worker = threading.Thread(target=delete)
worker.start()
try:
assert started.wait(1)
assert not finished.wait(0.1)
finally:
publication.release()
worker.join(2)
assert finished.is_set()
def test_leased_publication_blocks_delete_across_processes(self, tmp_path):
publication = publish_upload_bytes_leased(tmp_path, "report.pdf", b"old")
context = multiprocessing.get_context("spawn")
started = context.Event()
finished = context.Event()
errors = context.Queue()
worker = context.Process(
target=_delete_upload_in_process,
args=(str(tmp_path), "report.pdf", started, finished, errors),
)
worker.start()
try:
assert started.wait(10)
assert not finished.wait(0.2)
finally:
publication.release()
worker.join(10)
if worker.is_alive():
worker.terminate()
worker.join(2)
assert worker.exitcode == 0
try:
child_error = errors.get_nowait()
except Empty:
child_error = None
assert child_error is None
assert finished.is_set()
def test_lease_for_one_filename_does_not_block_another(self, tmp_path):
publication = publish_upload_bytes_leased(tmp_path, "report.pdf", b"old")
try:
with ThreadPoolExecutor(max_workers=1) as pool:
other = pool.submit(publish_upload_bytes, tmp_path, "notes.txt", b"new")
assert other.result(timeout=1) == tmp_path / "notes.txt"
finally:
publication.release()
def test_rollback_does_not_remove_reused_path(self, tmp_path):
publication = publish_upload_bytes_leased(tmp_path, "report.pdf", b"old")
try:
publication.path.unlink()
publication.path.write_bytes(b"new")
rollback_published_upload(publication)
finally:
publication.release()
assert (tmp_path / "report.pdf").read_bytes() == b"new"
def test_staging_unlink_failure_is_not_reported_as_success(self, tmp_path):
staged = create_upload_staging_file(tmp_path)
staged.handle.write(b"payload")
real_unlink = Path.unlink
def fail_only_for_stage(path, *args, **kwargs):
if path == staged.path:
raise OSError("cannot unlink stage")
return real_unlink(path, *args, **kwargs)
with patch.object(Path, "unlink", autospec=True, side_effect=fail_only_for_stage):
with pytest.raises(AtomicUploadPublishError, match="staging"):
publish_staged_upload(staged, "report.pdf")
assert not (tmp_path / "report.pdf").exists()
def test_abort_unlinks_stage_when_close_raises(self, tmp_path):
staged = create_upload_staging_file(tmp_path)
class CloseFailingHandle:
def __init__(self, wrapped):
self._wrapped = wrapped
@property
def closed(self):
return self._wrapped.closed
def close(self):
self._wrapped.close()
raise OSError("close failed")
staged.handle = CloseFailingHandle(staged.handle)
with pytest.raises(OSError, match="close failed"):
abort_staged_upload(staged)
assert not staged.path.exists()
def test_compatibility_wrapper_writes_new_file(self, tmp_path):
dest = write_upload_file_no_symlink(tmp_path, "notes.txt", b"hello")