242 lines
9.1 KiB
Python
242 lines
9.1 KiB
Python
"""Focused persistence and review-normalisation tests for the public-document library."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import pytest
|
||
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
||
|
||
|
||
async def _make_store():
|
||
from deerflow.persistence.base import Base
|
||
from deerflow.persistence.gongwen_documents.model import ( # noqa: F401
|
||
GongwenDocumentRow,
|
||
GongwenDocumentVersionRow,
|
||
)
|
||
from deerflow.persistence.gongwen_documents.store import SqlGongwenDocumentStore
|
||
|
||
engine = create_async_engine("sqlite+aiosqlite:///:memory:")
|
||
async with engine.begin() as connection:
|
||
await connection.run_sync(Base.metadata.create_all)
|
||
return SqlGongwenDocumentStore(async_sessionmaker(engine, expire_on_commit=False)), engine
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_document_versions_are_immutable_and_owner_scoped():
|
||
store, engine = await _make_store()
|
||
try:
|
||
created = await store.create_document(
|
||
{
|
||
"id": "gw_document_12345678",
|
||
"user_id": "u1",
|
||
"title": "测试通知",
|
||
"document_type": "notice",
|
||
"source": "upload",
|
||
}
|
||
)
|
||
assert created["status"] == "checking"
|
||
|
||
document, original = await store.append_version(
|
||
"gw_document_12345678",
|
||
"u1",
|
||
{
|
||
"id": "gwv_original",
|
||
"kind": "original",
|
||
"file_name": "source.docx",
|
||
"storage_key": "original.docx",
|
||
"content_type": "application/docx",
|
||
"size_bytes": 12,
|
||
"sha256": "abc",
|
||
"review_snapshot": {"issues": [{"severity": "P1"}]},
|
||
},
|
||
{"status": "review_required", "issue_counts": {"P0": 0, "P1": 1, "P2": 0}},
|
||
)
|
||
assert document["original_version_id"] == original["id"]
|
||
assert original["version_no"] == 1
|
||
|
||
repaired_document, repaired = await store.append_version(
|
||
"gw_document_12345678",
|
||
"u1",
|
||
{
|
||
"id": "gwv_repaired",
|
||
"kind": "repaired",
|
||
"file_name": "source-fixed.docx",
|
||
"storage_key": "repaired.docx",
|
||
"content_type": "application/docx",
|
||
"size_bytes": 14,
|
||
"sha256": "def",
|
||
"review_snapshot": {"issues": []},
|
||
},
|
||
{"status": "repaired", "issue_counts": {"P0": 0, "P1": 0, "P2": 0}},
|
||
)
|
||
assert repaired_document["original_version_id"] == original["id"]
|
||
assert repaired_document["latest_version_id"] == repaired["id"]
|
||
assert repaired["version_no"] == 2
|
||
|
||
versions = await store.list_versions("gw_document_12345678", "u1")
|
||
assert [row["id"] for row in versions] == ["gwv_repaired", "gwv_original"]
|
||
assert await store.get_document("gw_document_12345678", "u2") is None
|
||
assert await store.get_version("gw_document_12345678", "gwv_original", "u2") is None
|
||
finally:
|
||
await engine.dispose()
|
||
|
||
|
||
def test_review_snapshot_keeps_rule_location_and_severity():
|
||
from app.gateway.routers.gongwen_documents import _review_snapshot
|
||
|
||
review, counts = _review_snapshot(
|
||
{
|
||
"issues": [
|
||
{
|
||
"rule_id": "CHK-T001",
|
||
"severity": "P1",
|
||
"check_type": "format",
|
||
"name": "标题字体",
|
||
"location": "第 3 段",
|
||
"message": "标题应使用方正小标宋简体",
|
||
"standard_ref": "GB/T 9704-2012",
|
||
}
|
||
]
|
||
},
|
||
document_type="notice",
|
||
)
|
||
assert counts == {"P0": 0, "P1": 1, "P2": 0}
|
||
assert review["issues"][0]["paragraphIndex"] == 2
|
||
assert review["issues"][0]["standardRef"] == "GB/T 9704-2012"
|
||
|
||
|
||
def test_draft_planning_uses_a_document_type_specific_structure():
|
||
from app.gateway.routers.gongwen_documents import (
|
||
_recommend_template,
|
||
_template_profile,
|
||
)
|
||
|
||
selected, _reason = _recommend_template("请示上级批准项目实施计划")
|
||
|
||
assert selected == "request"
|
||
assert _template_profile("request")["outline"] == [
|
||
"基本情况",
|
||
"请示事项",
|
||
"建议方案",
|
||
]
|
||
assert _template_profile("work_plan")["outline"] != _template_profile("notice")[
|
||
"outline"
|
||
]
|
||
|
||
|
||
def test_personal_templates_are_owner_scoped_and_keep_their_outline(tmp_path, monkeypatch):
|
||
from app.gateway.routers import gongwen_documents as router
|
||
|
||
monkeypatch.setattr(router, "_documents_root", lambda: tmp_path / "gongwen-documents")
|
||
saved = router._save_custom_template_sync(
|
||
"user-a",
|
||
{
|
||
"key": "custom_abcdef12",
|
||
"name": "项目请示模板",
|
||
"description": "用于项目立项请示",
|
||
"category": "我的模板",
|
||
"outline": ["基本情况", "请示事项", "建议方案"],
|
||
"questions": [],
|
||
"engine_type": "request",
|
||
"created_at": "2026-09-15T00:00:00+00:00",
|
||
},
|
||
)
|
||
|
||
assert saved["scope"] == "personal"
|
||
assert saved["engineType"] == "request"
|
||
assert saved["outline"] == ["基本情况", "请示事项", "建议方案"]
|
||
assert router._get_custom_template_sync("user-a", "custom_abcdef12") is not None
|
||
assert router._get_custom_template_sync("user-b", "custom_abcdef12") is None
|
||
|
||
|
||
def test_element_injection_steps_follow_the_documented_workflow():
|
||
from pathlib import Path
|
||
|
||
from fastapi import HTTPException
|
||
|
||
from app.gateway.routers.gongwen_documents import (
|
||
DocumentElementsRequest,
|
||
_element_injection_steps,
|
||
)
|
||
|
||
target = Path("out") / "generated.docx"
|
||
|
||
# Element-less generation must stay byte-for-byte unchanged.
|
||
assert _element_injection_steps(None, target) == []
|
||
assert _element_injection_steps(DocumentElementsRequest(), target) == []
|
||
|
||
full = DocumentElementsRequest(
|
||
org_name="XXX市人民政府",
|
||
doc_number="X政发〔2026〕12号",
|
||
signer="张三",
|
||
cc="各区县人民政府",
|
||
printer="XXX市人民政府办公室",
|
||
print_date="2026年9月18日",
|
||
page_number=True,
|
||
)
|
||
steps = _element_injection_steps(full, target)
|
||
assert [step[0] for step in steps] == ["header", "footer", "pagenum"]
|
||
header, footer, pagenum = steps
|
||
assert header[:6] == ["header", str(target), "-o", str(target), "--org-name", "XXX市人民政府"]
|
||
assert header[6:] == ["--doc-number", "X政发〔2026〕12号", "--signer", "张三"]
|
||
assert footer[:4] == ["footer", str(target), "-o", str(target)]
|
||
assert footer[4:] == [
|
||
"--cc", "各区县人民政府",
|
||
"--printer", "XXX市人民政府办公室",
|
||
"--print-date", "2026年9月18日",
|
||
]
|
||
assert pagenum == ["pagenum", str(target), "-o", str(target)]
|
||
|
||
# Partial input injects only what the user confirmed.
|
||
partial = _element_injection_steps(
|
||
DocumentElementsRequest(org_name=" XXX局 ", page_number=True), target
|
||
)
|
||
assert [step[0] for step in partial] == ["header", "pagenum"]
|
||
assert partial[0][-1] == "XXX局"
|
||
|
||
# A header without the organisation mark is rejected up front with an
|
||
# actionable message instead of an engine failure.
|
||
with pytest.raises(HTTPException) as excinfo:
|
||
_element_injection_steps(DocumentElementsRequest(doc_number="X政发〔2026〕1号"), target)
|
||
assert excinfo.value.status_code == 422
|
||
|
||
|
||
def test_thread_artifact_import_paths_are_safe_and_idempotent(tmp_path, monkeypatch):
|
||
from app.gateway.routers import gongwen_documents as router
|
||
|
||
monkeypatch.setattr(router, "_documents_root", lambda: tmp_path / "gongwen-documents")
|
||
assert router._normalize_thread_artifact_path("finished.docx") == "mnt/user-data/outputs/finished.docx"
|
||
assert router._normalize_thread_artifact_path("/mnt/user-data/uploads/original.docx") == "mnt/user-data/uploads/original.docx"
|
||
with pytest.raises(Exception):
|
||
router._normalize_thread_artifact_path("mnt/user-data/outputs/../secret.docx")
|
||
with pytest.raises(Exception):
|
||
router._normalize_thread_artifact_path("mnt/user-data/outputs/report.pdf")
|
||
|
||
document_id = "gw_imported_12345678"
|
||
router._save_artifact_import_ref_sync(
|
||
"user-a",
|
||
document_id,
|
||
thread_id="thread-123",
|
||
artifact_path="mnt/user-data/outputs/finished.docx",
|
||
)
|
||
assert router._find_imported_artifact_sync(
|
||
"user-a",
|
||
thread_id="thread-123",
|
||
artifact_path="mnt/user-data/outputs/finished.docx",
|
||
) == document_id
|
||
router._save_artifact_import_ref_sync(
|
||
"user-a",
|
||
document_id,
|
||
thread_id="thread-123",
|
||
artifact_path="mnt/user-data/outputs/finished-fixed.docx",
|
||
)
|
||
assert router._find_imported_artifact_sync(
|
||
"user-a",
|
||
thread_id="thread-123",
|
||
artifact_path="mnt/user-data/outputs/finished-fixed.docx",
|
||
) == document_id
|
||
assert router._find_imported_artifact_sync(
|
||
"user-b",
|
||
thread_id="thread-123",
|
||
artifact_path="mnt/user-data/outputs/finished.docx",
|
||
) is None
|