136 lines
4.3 KiB
Python
136 lines
4.3 KiB
Python
import shutil
|
||
import uuid
|
||
from collections.abc import Iterator
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
|
||
from app.infrastructure.document_storage import DocumentStorage, LocalDocumentStorage
|
||
|
||
WORKROOT = Path(__file__).resolve().parents[2] / ".workdir"
|
||
|
||
|
||
@pytest.fixture
|
||
def tmp_path() -> Iterator[Path]:
|
||
"""工作区内的临时目录。
|
||
|
||
本机沙箱禁止枚举 `%TEMP%\\pytest-of-*`，pytest 内建 `tmp_path` 会 PermissionError，
|
||
故自管一个工作区目录；语义与 `tmp_path` 一致（每个用例唯一、用完即删）。
|
||
"""
|
||
path = WORKROOT / uuid.uuid4().hex[:12]
|
||
path.mkdir(parents=True)
|
||
try:
|
||
yield path
|
||
finally:
|
||
shutil.rmtree(path, ignore_errors=True)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_save_read_delete_round_trip(tmp_path: Path) -> None:
|
||
storage = LocalDocumentStorage(tmp_path)
|
||
key = "kb/2026/faq.txt"
|
||
|
||
path = await storage.save(key=key, content=b"hello", content_type="text/plain")
|
||
assert path == key
|
||
assert await storage.read(key=key) == b"hello"
|
||
|
||
await storage.delete(key=key)
|
||
with pytest.raises(FileNotFoundError):
|
||
await storage.read(key=key)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_archive_moves_without_deleting(tmp_path: Path) -> None:
|
||
storage = LocalDocumentStorage(tmp_path)
|
||
await storage.save(key="a.txt", content=b"x", content_type="text/plain")
|
||
|
||
await storage.archive(key="a.txt")
|
||
|
||
assert (tmp_path / "archive" / "a.txt").read_bytes() == b"x"
|
||
assert not (tmp_path / "a.txt").exists()
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_save_creates_nested_directories(tmp_path: Path) -> None:
|
||
storage = LocalDocumentStorage(tmp_path / "not-yet-created")
|
||
|
||
await storage.save(
|
||
key="kb/2026/09/faq.md", content="标题".encode(), content_type="text/markdown"
|
||
)
|
||
|
||
assert (tmp_path / "not-yet-created" / "kb" / "2026" / "09" / "faq.md").read_bytes() == (
|
||
"标题".encode()
|
||
)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_archive_preserves_sub_directory_layout(tmp_path: Path) -> None:
|
||
storage = LocalDocumentStorage(tmp_path)
|
||
await storage.save(key="kb/faq.txt", content=b"body", content_type="text/plain")
|
||
|
||
await storage.archive(key="kb/faq.txt")
|
||
|
||
assert (tmp_path / "archive" / "kb" / "faq.txt").read_bytes() == b"body"
|
||
assert not (tmp_path / "kb" / "faq.txt").exists()
|
||
assert (tmp_path / "kb").is_dir()
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_archive_is_idempotent_for_missing_key(tmp_path: Path) -> None:
|
||
storage = LocalDocumentStorage(tmp_path)
|
||
|
||
await storage.archive(key="missing.txt")
|
||
|
||
assert not (tmp_path / "archive").exists()
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_archive_rejects_already_archived_key(tmp_path: Path) -> None:
|
||
storage = LocalDocumentStorage(tmp_path)
|
||
(tmp_path / "archive").mkdir()
|
||
(tmp_path / "archive" / "a.txt").write_bytes(b"x")
|
||
|
||
with pytest.raises(ValueError):
|
||
await storage.save(key="archive/a.txt", content=b"x", content_type="text/plain")
|
||
with pytest.raises(ValueError):
|
||
await storage.archive(key="archive/a.txt")
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_delete_is_idempotent(tmp_path: Path) -> None:
|
||
storage = LocalDocumentStorage(tmp_path)
|
||
|
||
await storage.delete(key="never-existed.txt")
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_save_overwrites_existing_key(tmp_path: Path) -> None:
|
||
storage = LocalDocumentStorage(tmp_path)
|
||
await storage.save(key="a.txt", content=b"old", content_type="text/plain")
|
||
|
||
await storage.save(key="a.txt", content=b"new", content_type="text/plain")
|
||
|
||
assert await storage.read(key="a.txt") == b"new"
|
||
|
||
|
||
@pytest.mark.parametrize("key", ["../escape.txt", "kb/../../escape.txt", "..\\escape.txt"])
|
||
@pytest.mark.asyncio
|
||
async def test_parent_directory_reference_is_rejected(tmp_path: Path, key: str) -> None:
|
||
storage = LocalDocumentStorage(tmp_path)
|
||
|
||
with pytest.raises(ValueError):
|
||
await storage.save(key=key, content=b"x", content_type="text/plain")
|
||
|
||
|
||
@pytest.mark.parametrize("key", ["", "/etc/passwd", "C:/Windows/win.ini"])
|
||
@pytest.mark.asyncio
|
||
async def test_absolute_and_empty_keys_are_rejected(tmp_path: Path, key: str) -> None:
|
||
storage = LocalDocumentStorage(tmp_path)
|
||
|
||
with pytest.raises(ValueError):
|
||
await storage.read(key=key)
|
||
|
||
|
||
def test_local_storage_implements_protocol(tmp_path: Path) -> None:
|
||
assert isinstance(LocalDocumentStorage(tmp_path), DocumentStorage)
|