"""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