fix(evidence): harden source acquisition

This commit is contained in:
2026-07-12 03:32:18 +02:00
parent ffd683c587
commit 81ff1810d1
8 changed files with 523 additions and 183 deletions
@@ -1,3 +1,5 @@
import os
import pytest
from tht.adapters.evidence import FilesystemEvidenceSource
@@ -47,3 +49,65 @@ def test_filesystem_acquire_rejects_object_from_another_source(tmp_path):
with pytest.raises(EvidenceSourceError):
FilesystemEvidenceSource(right).acquire(item)
def test_filesystem_acquire_rejects_content_changed_since_discovery(tmp_path):
path = tmp_path / "doc.md"
path.write_text("first")
source = FilesystemEvidenceSource(tmp_path)
item = next(iter(source.discover()))
path.write_text("second")
with pytest.raises(EvidenceSourceError) as caught:
source.acquire(item)
assert not caught.value.retryable
def test_filesystem_open_is_safe_when_file_is_swapped_for_symlink(tmp_path, monkeypatch):
root = tmp_path / "root"
root.mkdir()
path = root / "doc.md"
path.write_text("safe")
outside = tmp_path / "outside.md"
outside.write_text("secret")
source = FilesystemEvidenceSource(root)
real_open = os.open
swapped = False
def racing_open(name, flags, *args, **kwargs):
nonlocal swapped
if name == "doc.md" and not swapped:
swapped = True
path.unlink()
path.symlink_to(outside)
return real_open(name, flags, *args, **kwargs)
monkeypatch.setattr(os, "open", racing_open)
with pytest.raises(EvidenceSourceError):
list(source.discover())
def test_filesystem_open_is_safe_when_ancestor_is_swapped_for_symlink(tmp_path, monkeypatch):
root = tmp_path / "root"
nested = root / "nested"
nested.mkdir(parents=True)
(nested / "doc.md").write_text("safe")
outside = tmp_path / "outside"
outside.mkdir()
(outside / "doc.md").write_text("secret")
source = FilesystemEvidenceSource(root)
real_open = os.open
swapped = False
def racing_open(name, flags, *args, **kwargs):
nonlocal swapped
if name == "nested" and not swapped and kwargs.get("dir_fd") is not None:
swapped = True
(nested / "doc.md").unlink()
nested.rmdir()
nested.symlink_to(outside, target_is_directory=True)
return real_open(name, flags, *args, **kwargs)
monkeypatch.setattr(os, "open", racing_open)
with pytest.raises(EvidenceSourceError):
list(source.discover())