114 lines
3.8 KiB
Python
114 lines
3.8 KiB
Python
import os
|
|
|
|
import pytest
|
|
|
|
from tht.evidence.adapters import FilesystemEvidenceSource
|
|
from tht.evidence.contracts import EvidenceSourceError
|
|
|
|
|
|
def test_filesystem_discovery_is_stable_and_acquisition_is_bounded(tmp_path):
|
|
(tmp_path / "z.md").write_text("z")
|
|
(tmp_path / "nested").mkdir()
|
|
(tmp_path / "nested" / "a.md").write_text("alpha")
|
|
source = FilesystemEvidenceSource(tmp_path, max_bytes=5)
|
|
|
|
first = list(source.discover())
|
|
assert [item.uri for item in first] == sorted(item.uri for item in first)
|
|
assert all(item.source_id.startswith("filesystem:") for item in first)
|
|
assert all(item.fingerprint.startswith("sha256:") for item in first)
|
|
assert source.acquire(first[0]).content in {b"alpha", b"z"}
|
|
|
|
(tmp_path / "large.md").write_bytes(b"123456")
|
|
with pytest.raises(EvidenceSourceError) as caught:
|
|
list(source.discover())
|
|
assert not caught.value.retryable
|
|
assert "large.md" not in str(caught.value)
|
|
|
|
|
|
def test_filesystem_rejects_symlink_escape(tmp_path):
|
|
root = tmp_path / "root"
|
|
root.mkdir()
|
|
outside = tmp_path / "secret.md"
|
|
outside.write_text("secret")
|
|
(root / "escape.md").symlink_to(outside)
|
|
|
|
with pytest.raises(EvidenceSourceError) as caught:
|
|
list(FilesystemEvidenceSource(root).discover())
|
|
assert not caught.value.retryable
|
|
assert str(outside) not in str(caught.value)
|
|
|
|
|
|
def test_filesystem_acquire_rejects_object_from_another_source(tmp_path):
|
|
left = tmp_path / "left"
|
|
right = tmp_path / "right"
|
|
left.mkdir()
|
|
right.mkdir()
|
|
(left / "doc.md").write_text("left")
|
|
(right / "doc.md").write_text("right")
|
|
item = next(iter(FilesystemEvidenceSource(left).discover()))
|
|
|
|
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())
|