Files
RAG/backend/tests/unit/test_document_review_persistence.py
YoVinchen ecdb10c37a
Some checks failed
verify / verify (push) Has been cancelled
Make the governed RAG evidence path executable end to end
Separate local parsing from model indexing, bind review decisions to immutable manifests, persist vectors behind active profiles, and expose retrieval, chat, evaluation, and document workflows through the React workbench.

Constraint: Live Bailian authentication currently fails for all three configured capabilities

Rejected: Direct upload-to-embedding flow | bypasses local review and manifest binding

Confidence: high

Scope-risk: broad

Directive: Keep private-data deployment blocked until authentication, RBAC, and separate database roles land

Tested: make verify; fresh and replay Docker document smoke; worker recovery smoke; frozen synthetic evaluation; migration 0003-0004 roundtrip

Not-tested: Successful live Bailian calls, OCR, real multi-user authorization
2026-07-13 05:58:11 +08:00

81 lines
2.7 KiB
Python

from __future__ import annotations
import uuid
from pathlib import Path
import pytest
from app.core.config import Settings
from app.persistence.document_review import (
_APPROVE_CHUNKS,
_APPROVE_VERSION,
_ENQUEUE_EMBED_JOB,
_LOCK_REVIEW,
_REJECT_VERSION,
PostgresDocumentReviewRepository,
)
from app.persistence.documents import DocumentActor
MANIFEST = "a" * 64
ACTOR = DocumentActor(
subject="synthetic-demo-maintainer",
knowledge_base_id=uuid.UUID("10000000-0000-0000-0000-000000000001"),
access_scope_id=uuid.UUID("20000000-0000-0000-0000-000000000002"),
)
def _normalized(statement: str) -> str:
return " ".join(statement.lower().split())
def test_review_lock_and_mutations_are_scope_manifest_and_revision_bound() -> None:
lock = _normalized(_LOCK_REVIEW)
approve = _normalized(_APPROVE_VERSION)
chunks = _normalized(_APPROVE_CHUNKS)
reject = _normalized(_REJECT_VERSION)
assert "document.knowledge_base_id = %s" in lock
assert "document.access_scope_id = %s" in lock
assert "for update of document, version" in lock
assert "review_revision = %s" in approve
assert "outbound_manifest_sha256 = %s" in approve
assert "review_state = 'local_parsed_pending_cloud_review'" in approve
assert "approval_status = 'local_parsed_pending_cloud_review'" in chunks
assert "embedding_model = %s" in chunks
assert "embedding_dimension = 1024" in chunks
assert "review_revision = %s" in reject
def test_embedding_job_payload_parameter_has_an_explicit_postgres_type() -> None:
normalized = _normalized(_ENQUEUE_EMBED_JOB)
assert "jsonb_build_object('document_version_id', %s::text)" in normalized
def test_decision_validation_fails_closed_before_database_access(tmp_path: Path) -> None:
password = tmp_path / "password"
password.write_text("synthetic-password", encoding="utf-8")
repository = PostgresDocumentReviewRepository(Settings(postgres_password_file=password))
with pytest.raises(ValueError, match="approval requires"):
repository.apply_decision(
actor=ACTOR,
document_id=uuid.uuid4(),
decision="APPROVE",
reason_code="SYNTHETIC_REVIEW_APPROVED",
expected_revision=0,
outbound_manifest_sha256=None,
trace_id=uuid.uuid4(),
)
with pytest.raises(ValueError, match="rejection requires"):
repository.apply_decision(
actor=ACTOR,
document_id=uuid.uuid4(),
decision="REJECT",
reason_code="SYNTHETIC_REVIEW_APPROVED",
expected_revision=0,
outbound_manifest_sha256=MANIFEST,
trace_id=uuid.uuid4(),
)