Spaces:
Sleeping
Sleeping
File size: 4,222 Bytes
652df4e bbbfba8 652df4e bbbfba8 652df4e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 | """Tests for the file store."""
from __future__ import annotations
from typing import TYPE_CHECKING
from src.app.persistence.file_store import FileStore
if TYPE_CHECKING:
from pathlib import Path
class TestFileStore:
def test_ensure_dirs(self, tmp_storage: Path) -> None:
store = FileStore(tmp_storage)
store.ensure_dirs()
assert (tmp_storage / "jobs").exists()
assert (tmp_storage / "providers").exists()
def test_save_and_load_json(self, tmp_storage: Path) -> None:
store = FileStore(tmp_storage)
data = {"key": "value", "number": 42}
store.save_json("job_001", "test.json", data)
loaded = store.load_json("job_001", "test.json")
assert loaded == data
def test_save_and_load_bytes(self, tmp_storage: Path) -> None:
store = FileStore(tmp_storage)
data = b"<alto>test</alto>"
store.save_bytes("job_001", "alto.xml", data)
loaded = store.load_bytes("job_001", "alto.xml")
assert loaded == data
def test_load_nonexistent_returns_none(self, tmp_storage: Path) -> None:
store = FileStore(tmp_storage)
assert store.load_json("missing", "test.json") is None
assert store.load_bytes("missing", "test.xml") is None
def test_save_raw_payload(self, tmp_storage: Path) -> None:
store = FileStore(tmp_storage)
store.save_raw_payload("job_001", {"payload": [1, 2, 3]})
loaded = store.load_raw_payload("job_001")
assert loaded == {"payload": [1, 2, 3]}
def test_save_canonical(self, tmp_storage: Path) -> None:
store = FileStore(tmp_storage)
store.save_canonical("job_001", {"document_id": "doc1"})
loaded = store.load_canonical("job_001")
assert loaded["document_id"] == "doc1"
def test_save_alto(self, tmp_storage: Path) -> None:
store = FileStore(tmp_storage)
xml = b"<?xml version='1.0'?><alto/>"
store.save_alto("job_001", xml)
assert store.load_alto("job_001") == xml
def test_save_page_xml(self, tmp_storage: Path) -> None:
store = FileStore(tmp_storage)
xml = b"<?xml version='1.0'?><PcGts/>"
store.save_page_xml("job_001", xml)
assert store.load_page_xml("job_001") == xml
def test_save_events(self, tmp_storage: Path) -> None:
store = FileStore(tmp_storage)
events = [{"step": "normalize", "status": "completed"}]
store.save_events("job_001", events)
loaded = store.load_events("job_001")
assert loaded == events
def test_list_jobs(self, tmp_storage: Path) -> None:
store = FileStore(tmp_storage)
store.save_json("job_aaa", "test.json", {})
store.save_json("job_bbb", "test.json", {})
jobs = store.list_jobs()
assert "job_aaa" in jobs
assert "job_bbb" in jobs
def test_job_exists(self, tmp_storage: Path) -> None:
store = FileStore(tmp_storage)
assert not store.job_exists("job_001")
store.save_json("job_001", "test.json", {})
assert store.job_exists("job_001")
def test_job_has_artifact(self, tmp_storage: Path) -> None:
store = FileStore(tmp_storage)
store.save_json("job_001", "canonical.json", {})
assert store.job_has_artifact("job_001", "canonical.json")
assert not store.job_has_artifact("job_001", "alto.xml")
def test_provider_crud(self, tmp_storage: Path) -> None:
store = FileStore(tmp_storage)
store.ensure_dirs()
store.save_provider("paddle", {"provider_id": "paddle", "family": "word_box_json"})
assert store.load_provider("paddle") is not None
assert "paddle" in store.list_providers()
assert store.delete_provider("paddle")
assert store.load_provider("paddle") is None
def test_save_input_image(self, tmp_storage: Path, tmp_path: Path) -> None:
store = FileStore(tmp_storage)
img = tmp_path / "test.png"
img.write_bytes(b"\x89PNG\r\n\x1a\n")
result = store.save_input_image("job_001", img)
assert result.exists()
assert result.name == "input.png"
assert store.get_input_image_path("job_001") == result
|