50 lines
1.7 KiB
Python
50 lines
1.7 KiB
Python
import pytest
|
|
|
|
from tht.adapters.evidence import FilesystemEvidenceSource
|
|
from tht.ports.evidence 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)
|