wehub-resource-sync
133 lines
4.5 KiB
Python
133 lines
4.5 KiB
Python
"""Tests for atomic media write in transcript persistence.
|
|
|
|
Verifies the tmp + flush + fsync + os.replace flow, that the exception
|
|
path unlinks the tmp file, that os.replace is invoked, that a failed
|
|
write leaves no residue, and that concurrent writes to the same target
|
|
produce an uncorrupted file.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from concurrent.futures import ThreadPoolExecutor, as_completed
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
from opensquilla.gateway.transcripts import _atomic_write_bytes
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# write_uses_replace
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_write_uses_replace(tmp_path: Path) -> None:
|
|
"""os.replace must be called when writing atomically."""
|
|
target = tmp_path / "target.bin"
|
|
data = b"hello atomic"
|
|
|
|
with patch("opensquilla.gateway.transcripts.os.replace", wraps=os.replace) as mock_replace:
|
|
_atomic_write_bytes(target, data)
|
|
mock_replace.assert_called_once()
|
|
# First arg of the call must differ from target (it's the tmp path)
|
|
called_src = Path(mock_replace.call_args[0][0])
|
|
called_dst = Path(mock_replace.call_args[0][1])
|
|
assert called_src != called_dst
|
|
assert called_dst == target
|
|
|
|
assert target.read_bytes() == data
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# failed_write_no_residue
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_failed_write_no_residue(tmp_path: Path) -> None:
|
|
"""If the write raises mid-way, target must not exist and tmp must be cleaned up."""
|
|
target = tmp_path / "target.bin"
|
|
data = b"should never land"
|
|
|
|
original_open = open
|
|
|
|
call_count = {"n": 0}
|
|
|
|
def exploding_open(path, mode="r", **kw):
|
|
if "wb" in mode or mode == "wb":
|
|
call_count["n"] += 1
|
|
# Return a context manager that raises on __exit__ after write
|
|
class _ExplodingFile:
|
|
def __init__(self):
|
|
self._f = original_open(path, mode, **kw)
|
|
self._path = path
|
|
|
|
def write(self, b):
|
|
return self._f.write(b)
|
|
|
|
def flush(self):
|
|
raise OSError("simulated disk failure")
|
|
|
|
def fileno(self):
|
|
return self._f.fileno()
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *args):
|
|
try:
|
|
self._f.close()
|
|
except Exception:
|
|
pass
|
|
return False # re-raise
|
|
|
|
return _ExplodingFile()
|
|
return original_open(path, mode, **kw)
|
|
|
|
with patch("builtins.open", side_effect=exploding_open):
|
|
with pytest.raises(OSError, match="simulated disk failure"):
|
|
_atomic_write_bytes(target, data)
|
|
|
|
# Target must not exist
|
|
assert not target.exists(), "target file must not exist after failed write"
|
|
|
|
# No .tmp.* residue in the directory
|
|
residue = list(tmp_path.glob("*.tmp.*"))
|
|
assert residue == [], f"tmp residue found: {residue}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# concurrent_write_same_target
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_concurrent_write_same_target(tmp_path: Path) -> None:
|
|
"""Two concurrent writes to the same target must not corrupt the file.
|
|
|
|
The final content must be exactly one of the two payloads (no mixing).
|
|
"""
|
|
target = tmp_path / "shared.bin"
|
|
payload_a = b"AAAA" * 4096 # 16 KiB, distinct pattern
|
|
payload_b = b"BBBB" * 4096
|
|
|
|
errors: list[Exception] = []
|
|
|
|
def write_it(data: bytes) -> None:
|
|
try:
|
|
_atomic_write_bytes(target, data)
|
|
except Exception as exc:
|
|
errors.append(exc)
|
|
|
|
with ThreadPoolExecutor(max_workers=2) as pool:
|
|
futures = [
|
|
pool.submit(write_it, payload_a),
|
|
pool.submit(write_it, payload_b),
|
|
]
|
|
for f in as_completed(futures):
|
|
f.result() # propagate any unexpected exception
|
|
|
|
assert not errors, f"unexpected errors during concurrent write: {errors}"
|
|
assert target.exists(), "target file must exist after concurrent writes"
|
|
|
|
final = target.read_bytes()
|
|
assert final in (payload_a, payload_b), (
|
|
f"file content is corrupted (length={len(final)}, "
|
|
f"expected {len(payload_a)} bytes of either A or B pattern)"
|
|
)
|