mirror of
https://github.com/calibrain/shelfmark.git
synced 2026-09-24 18:40:29 +01:00
@@ -101,6 +101,8 @@ def run_blocking_io[T](func: Callable[..., T], *args: Any, **kwargs: Any) -> T:
|
||||
|
||||
_VERIFY_IO_WAIT_SECONDS = 3.0
|
||||
_PUBLISH_VERIFY_RETRY_SECONDS = 0.25
|
||||
_TEMPFILE_PREFIX = ".shelfmark."
|
||||
_TEMPFILE_SUFFIX = ".tmp"
|
||||
|
||||
|
||||
def _verify_transfer_size(
|
||||
@@ -339,10 +341,16 @@ def _hardlink_not_supported(error: OSError) -> bool:
|
||||
|
||||
|
||||
def _create_temp_path(dest_path: Path) -> Path:
|
||||
"""Create a destination-adjacent temp file without inheriting the full basename.
|
||||
|
||||
Reusing the entire destination filename in the temp prefix can push otherwise
|
||||
valid long names over the filesystem component limit once `tempfile` adds its
|
||||
random suffix.
|
||||
"""
|
||||
fd, temp_path = run_blocking_io(
|
||||
tempfile.mkstemp,
|
||||
prefix=f".{dest_path.name}.",
|
||||
suffix=".tmp",
|
||||
prefix=_TEMPFILE_PREFIX,
|
||||
suffix=_TEMPFILE_SUFFIX,
|
||||
dir=str(dest_path.parent),
|
||||
)
|
||||
run_blocking_io(os.close, fd)
|
||||
|
||||
@@ -147,6 +147,20 @@ class TestAtomicCopy:
|
||||
|
||||
assert result == tmp_path / "dest_3.txt"
|
||||
|
||||
def test_long_destination_name(self, tmp_path):
|
||||
"""Copies long-but-valid destination names without overflowing temp filenames."""
|
||||
from shelfmark.download.fs import atomic_copy as _atomic_copy
|
||||
|
||||
source = tmp_path / "source.epub"
|
||||
source.write_bytes(b"epub content")
|
||||
dest = tmp_path / f"{'A' * 240}.epub"
|
||||
|
||||
result = _atomic_copy(source, dest)
|
||||
|
||||
assert result == dest
|
||||
assert result.exists()
|
||||
assert result.read_bytes() == b"epub content"
|
||||
|
||||
def test_preserves_extension(self, tmp_path):
|
||||
"""Keeps extension when adding counter suffix."""
|
||||
from shelfmark.download.fs import atomic_copy as _atomic_copy
|
||||
|
||||
@@ -350,6 +350,27 @@ class TestAtomicMove:
|
||||
assert not source.exists()
|
||||
assert result.read_text() == "content"
|
||||
|
||||
def test_cross_filesystem_fallback_handles_long_destination_name(self, tmp_path, monkeypatch):
|
||||
"""Cross-filesystem fallback handles long destination names safely."""
|
||||
import errno
|
||||
|
||||
from shelfmark.download.fs import atomic_move as _atomic_move
|
||||
|
||||
source = tmp_path / "source.epub"
|
||||
source.write_text("content")
|
||||
dest = tmp_path / f"{'A' * 240}.epub"
|
||||
|
||||
def _raise_exdev(*_args, **_kwargs):
|
||||
raise OSError(errno.EXDEV, "Cross-device link")
|
||||
|
||||
monkeypatch.setattr(os, "rename", _raise_exdev)
|
||||
|
||||
result = _atomic_move(source, dest)
|
||||
|
||||
assert result == dest
|
||||
assert not source.exists()
|
||||
assert result.read_text() == "content"
|
||||
|
||||
def test_cross_filesystem_permission_fallback(self, tmp_path, monkeypatch):
|
||||
"""Falls back to copy when cross-filesystem move hits permission error."""
|
||||
import errno
|
||||
|
||||
Reference in New Issue
Block a user