Spaces:
Runtime error
Runtime error
File size: 1,190 Bytes
aad7814 | 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 | from __future__ import annotations
import time
from backend.core.document_library import (
document_created_at_iso,
is_reingest_running,
recover_stale_processing_documents,
schedule_reingest_all_documents,
)
from backend.core.report_session import UploadedDocument, list_documents, save_document
def test_document_created_at_iso_from_epoch():
ts = time.time()
doc = UploadedDocument(
document_id="abc",
filename="test.pdf",
created_at=ts,
)
iso = document_created_at_iso(doc)
assert "T" in iso
assert iso.endswith("+00:00") or iso.endswith("Z") or "+" in iso
def test_recover_stale_processing_when_no_worker(tmp_path, monkeypatch):
monkeypatch.setenv("DATA_DIR", str(tmp_path))
tenant = "t-recover"
doc = UploadedDocument(
document_id="d1",
filename="a.pdf",
status="processing",
storage_path=str(tmp_path / "a.pdf"),
)
(tmp_path / "a.pdf").write_bytes(b"x")
save_document(tenant, doc)
assert is_reingest_running(tenant) is False
n = recover_stale_processing_documents(tenant)
assert n == 1
assert list_documents(tenant)["d1"].status == "complete"
|