document-management/backend/tests/dms/q3b_http_verify.py

"""Q3-B 8755短生命周期真实HTTP闭环验证工具。"""

from __future__ import annotations

import io
import json
import os
import struct
import sys
import threading
from datetime import datetime, timedelta, timezone
from pathlib import Path
from urllib.parse import urlparse

import fitz
import requests
from docx import Document as WordDocument
from sqlalchemy import func, select, text
from werkzeug.serving import make_server

BACKEND_ROOT = Path(__file__).resolve().parents[2]
if str(BACKEND_ROOT) not in sys.path:
    sys.path.insert(0, str(BACKEND_ROOT))


def _assert_test_database(database_url: str) -> None:
    if urlparse(database_url).path.lstrip("/") != "dms_test":
        raise RuntimeError("Q3-B真实HTTP验证只允许连接dms_test")


def _docx_bytes(term: str) -> bytes:
    stream = io.BytesIO()
    document = WordDocument()
    document.add_paragraph(term)
    table = document.add_table(rows=1, cols=1)
    table.cell(0, 0).text = "Q3HttpDocxTableTerm"
    document.save(stream)
    return stream.getvalue()


def _pdf_bytes(term: str) -> bytes:
    document = fitz.open()
    page = document.new_page()
    page.insert_text((72, 72), term)
    content = document.tobytes()
    document.close()
    return content


def _doc_bytes() -> bytes:
    def entry(name: str, kind: int, child: int = 0xFFFFFFFF) -> bytes:
        value = bytearray(128)
        encoded = (name + "\0").encode("utf-16le")
        value[: len(encoded)] = encoded
        struct.pack_into("<H", value, 64, len(encoded))
        value[66] = kind
        value[67] = 1
        struct.pack_into("<III", value, 68, 0xFFFFFFFF, 0xFFFFFFFF, child)
        struct.pack_into("<I", value, 116, 0xFFFFFFFE)
        return bytes(value)

    header = bytearray(512)
    header[:8] = bytes.fromhex("D0CF11E0A1B11AE1")
    struct.pack_into("<HH", header, 24, 0x003E, 3)
    struct.pack_into("<H", header, 28, 0xFFFE)
    struct.pack_into("<HH", header, 30, 9, 6)
    struct.pack_into("<I", header, 44, 1)
    struct.pack_into("<I", header, 48, 0)
    struct.pack_into("<I", header, 56, 4096)
    struct.pack_into("<I", header, 60, 0xFFFFFFFE)
    struct.pack_into("<I", header, 68, 0xFFFFFFFE)
    struct.pack_into("<I", header, 76, 1)
    for index in range(1, 109):
        struct.pack_into("<I", header, 76 + 4 * index, 0xFFFFFFFF)
    directory = bytearray(512)
    directory[:128] = entry("Root Entry", 5, 1)
    directory[128:256] = entry("WordDocument", 2)
    fat = bytearray(b"\xff" * 512)
    struct.pack_into("<I", fat, 0, 0xFFFFFFFE)
    struct.pack_into("<I", fat, 4, 0xFFFFFFFD)
    return bytes(header + directory + fat)


def _token(base_url: str, username: str, password: str) -> str:
    response = requests.post(
        f"{base_url}/auth/login",
        json={
            "username": username,
            "password": password,
            "keepSignedIn": False,
        },
        timeout=10,
    )
    response.raise_for_status()
    return response.json()["data"]["accessToken"]


def _headers(token: str) -> dict[str, str]:
    return {"Authorization": f"Bearer {token}"}


def _tree_count(nodes: list[dict], category_id: str) -> int:
    pending = list(nodes)
    while pending:
        node = pending.pop()
        if node["id"] == category_id:
            return node["documentCount"]
        pending.extend(node["children"])
    raise AssertionError(f"category {category_id} not found")


def _upload(
    base_url: str,
    token: str,
    path: str,
    filename: str,
    content: bytes,
    metadata: dict,
) -> dict:
    response = requests.post(
        f"{base_url}{path}",
        headers=_headers(token),
        files={"file": (filename, content)},
        data={"metadata": json.dumps(metadata, ensure_ascii=False)},
        timeout=20,
    )
    response.raise_for_status()
    return response.json()["data"]


def main() -> None:
    database_url = os.environ.get("DMS_DATABASE_URL", "")
    _assert_test_database(database_url)
    admin_password = os.environ["DMS_HTTP_ADMIN_PASSWORD"]
    user_password = os.environ["DMS_HTTP_USER_PASSWORD"]

    from app import app
    from dms.extensions import db
    from dms.models import AuditLog, Category, Document, User

    with app.app_context():
        assert db.session.execute(text("SELECT DATABASE()" )).scalar() == "dms_test"
        leaf = db.session.scalar(
            select(Category).where(Category.category_code == "STYLE_A1")
        )
        parent = db.session.get(Category, leaf.parent_id)
        user = db.session.scalar(select(User).where(User.username == "user"))
        leaf_id, parent_id, user_id = str(leaf.id), str(parent.id), str(user.id)

    server = make_server("127.0.0.1", 8755, app, threaded=True)
    thread = threading.Thread(target=server.serve_forever, daemon=True)
    thread.start()
    base_url = "http://127.0.0.1:8755/api/v1"
    created: list[tuple[str, str, int]] = []
    main_version = 0
    result: dict[str, object] = {"port": 8755, "database": "dms_test"}

    try:
        admin_token = _token(base_url, "admin", admin_password)
        user_token = _token(base_url, "user", user_password)
        for path in ("attachments", "documents"):
            stale = requests.get(
                f"{base_url}/{path}",
                params={"keyword": "Q3_HTTP_", "pageSize": 100},
                headers=_headers(admin_token),
                timeout=10,
            )
            stale.raise_for_status()
            for item in stale.json()["data"]["items"]:
                if item["documentName"].startswith("Q3_HTTP_"):
                    removed = requests.delete(
                        f"{base_url}/{path}/{item['id']}",
                        params={"rowVersion": item["rowVersion"]},
                        headers=_headers(admin_token),
                        timeout=10,
                    )
                    removed.raise_for_status()
        auditor = requests.post(
            f"{base_url}/auth/login",
            json={
                "username": "auditor",
                "password": admin_password,
                "keepSignedIn": False,
            },
            timeout=10,
        )
        assert auditor.status_code == 401

        me = requests.get(
            f"{base_url}/users/me", headers=_headers(admin_token), timeout=10
        )
        me.raise_for_status()
        assert me.json()["data"]["allowedModules"] == [
            "DOCUMENT_BROWSER",
            "BACKEND_MANAGEMENT",
            "AUDIT_LOG",
        ]
        audit_api = requests.get(
            f"{base_url}/audit/logs", headers=_headers(admin_token), timeout=10
        )
        audit_api.raise_for_status()
        user_library = requests.get(
            f"{base_url}/attachments", headers=_headers(user_token), timeout=10
        )
        assert user_library.status_code == 403
        baseline_all = requests.get(
            f"{base_url}/documents", headers=_headers(admin_token), timeout=10
        )
        baseline_all.raise_for_status()
        baseline_recursive = requests.get(
            f"{base_url}/documents",
            params={"categoryId": parent_id, "includeDescendants": "true"},
            headers=_headers(admin_token),
            timeout=10,
        )
        baseline_recursive.raise_for_status()
        baseline_tree_counts = {}
        for username, token in (("admin", admin_token), ("user", user_token)):
            response = requests.get(
                f"{base_url}/categories/tree", headers=_headers(token), timeout=10
            )
            response.raise_for_status()
            baseline_tree_counts[username] = _tree_count(
                response.json()["data"], parent_id
            )

        docx_content = _docx_bytes("Q3HttpDocxBodyOnlyTerm")
        main = _upload(
            base_url,
            admin_token,
            "/documents",
            "q3-http.docx",
            docx_content,
            {
                "documentName": "Q3_HTTP_DOCX",
                "documentType": "MAIN",
                "categoryId": leaf_id,
                "securityLevel": "INTERNAL",
                "summary": "真实HTTP验证",
                "tags": ["Q3HTTP"],
            },
        )
        created.append(("document", main["id"], main["rowVersion"]))
        assert main["status"] == "PUBLISHED"
        assert "visibilityType" not in main

        sub = _upload(
            base_url,
            admin_token,
            "/documents",
            "q3-http-sub.docx",
            _docx_bytes("Q3HttpSubBodyTerm"),
            {
                "documentName": "Q3_HTTP_SUB",
                "documentType": "SUB_PLAN",
                "parentDocumentId": main["id"],
                "securityLevel": "INTERNAL",
                "summary": "真实HTTP子方案",
                "tags": ["Q3HTTP"],
            },
        )
        created.append(("document", sub["id"], sub["rowVersion"]))
        assert sub["categoryId"] == leaf_id

        pdf_content = _pdf_bytes("Q3HttpPdfBodyOnlyTerm")
        pdf = _upload(
            base_url,
            admin_token,
            "/documents",
            "q3-http.pdf",
            pdf_content,
            {
                "documentName": "Q3_HTTP_PDF",
                "documentType": "MAIN",
                "categoryId": leaf_id,
                "securityLevel": "INTERNAL",
                "summary": "真实HTTP PDF",
                "tags": ["Q3HTTP"],
            },
        )
        created.append(("document", pdf["id"], pdf["rowVersion"]))

        attachment = _upload(
            base_url,
            admin_token,
            "/attachments",
            "q3-http-attachment.docx",
            _docx_bytes("Q3HttpAttachmentTerm"),
            {
                "documentName": "Q3_HTTP_ATTACHMENT",
                "attachmentType": "OTHER",
                "summary": "真实HTTP附件",
                "tags": ["Q3HTTP"],
            },
        )
        created.append(("attachment", attachment["id"], attachment["rowVersion"]))

        legacy = _upload(
            base_url,
            admin_token,
            "/attachments",
            "q3-http.doc",
            _doc_bytes(),
            {
                "documentName": "Q3_HTTP_DOC",
                "attachmentType": "OTHER",
                "summary": "旧版DOC",
                "tags": ["Q3HTTP"],
            },
        )
        created.append(("attachment", legacy["id"], legacy["rowVersion"]))

        current_main = requests.get(
            f"{base_url}/documents/{main['id']}",
            headers=_headers(admin_token),
            timeout=10,
        )
        current_main.raise_for_status()
        main_version = current_main.json()["data"]["rowVersion"]
        permission = requests.put(
            f"{base_url}/documents/{main['id']}/permissions",
            headers=_headers(admin_token),
            json={
                "documentRowVersion": main_version,
                "entries": [
                    {
                        "subjectId": user_id,
                        "canView": True,
                        "canDownload": True,
                        "expiresAt": (
                            datetime.now(timezone.utc) + timedelta(days=7)
                        ).strftime("%Y-%m-%dT%H:%M:%S.000Z"),
                    }
                ],
            },
            timeout=10,
        )
        permission.raise_for_status()
        main_version = permission.json()["data"]["documentRowVersion"]
        binding = requests.post(
            f"{base_url}/main-plans/{main['id']}/attachments/bind",
            headers=_headers(admin_token),
            json={
                "attachmentIds": [attachment["id"]],
                "mainPlanRowVersion": main_version,
            },
            timeout=10,
        )
        binding.raise_for_status()
        main_version = binding.json()["data"]["mainPlanRowVersion"]
        mounted = requests.get(
            f"{base_url}/attachments/{attachment['id']}",
            headers=_headers(user_token),
            timeout=10,
        )
        mounted.raise_for_status()

        with app.app_context():
            audit_before_queries = db.session.scalar(
                select(func.count()).select_from(AuditLog)
            )
        recursive = requests.get(
            f"{base_url}/documents",
            params={"categoryId": parent_id, "includeDescendants": "true"},
            headers=_headers(admin_token),
            timeout=10,
        )
        recursive.raise_for_status()
        assert recursive.json()["data"]["total"] == (
            baseline_recursive.json()["data"]["total"] + 2
        )
        all_documents = requests.get(
            f"{base_url}/documents", headers=_headers(admin_token), timeout=10
        )
        all_documents.raise_for_status()
        assert all_documents.json()["data"]["total"] == (
            baseline_all.json()["data"]["total"] + 2
        )
        current_tree_counts = {}
        for username, token in (("admin", admin_token), ("user", user_token)):
            tree_response = requests.get(
                f"{base_url}/categories/tree", headers=_headers(token), timeout=10
            )
            tree_response.raise_for_status()
            current_tree_counts[username] = _tree_count(
                tree_response.json()["data"], parent_id
            )
        assert current_tree_counts["admin"] == baseline_tree_counts["admin"] + 2
        assert current_tree_counts["user"] == baseline_tree_counts["user"] + 1
        for term, expected_id in (
            ("Q3HttpDocxBodyOnlyTerm", main["id"]),
            ("Q3HttpPdfBodyOnlyTerm", pdf["id"]),
        ):
            search = requests.get(
                f"{base_url}/documents",
                params={"contentKeyword": term},
                headers=_headers(admin_token),
                timeout=10,
            )
            search.raise_for_status()
            assert [item["id"] for item in search.json()["data"]["items"]] == [
                expected_id
            ]
        with app.app_context():
            audit_after_queries = db.session.scalar(
                select(func.count()).select_from(AuditLog)
            )
            download_before = {
                row.id: row.download_count
                for row in db.session.scalars(
                    select(Document).where(
                        Document.id.in_([int(main["id"]), int(pdf["id"])])
                    )
                )
            }
        assert audit_after_queries == audit_before_queries

        docx_preview = requests.get(
            f"{base_url}/documents/{main['id']}/preview",
            headers=_headers(admin_token),
            timeout=10,
        )
        docx_preview.raise_for_status()
        assert docx_preview.content == docx_content
        assert docx_preview.headers["Content-Type"].startswith(
            "application/vnd.openxmlformats-officedocument.wordprocessingml.document"
        )
        assert docx_preview.headers["Content-Disposition"].startswith("inline")
        assert "filename*=" in docx_preview.headers["Content-Disposition"]
        pdf_preview = requests.get(
            f"{base_url}/documents/{pdf['id']}/preview",
            headers=_headers(admin_token),
            timeout=10,
        )
        pdf_preview.raise_for_status()
        assert pdf_preview.content == pdf_content
        assert pdf_preview.headers["Content-Type"].startswith("application/pdf")
        legacy_preview = requests.get(
            f"{base_url}/documents/{legacy['id']}/preview",
            headers=_headers(admin_token),
            timeout=10,
        )
        assert legacy_preview.status_code == 415
        assert legacy_preview.json()["code"] == "PREVIEW_UNAVAILABLE"
        with app.app_context():
            download_after = {
                row.id: row.download_count
                for row in db.session.scalars(
                    select(Document).where(
                        Document.id.in_([int(main["id"]), int(pdf["id"])])
                    )
                )
            }
        assert download_after == download_before

        result.update(
            {
                "adminLogin": 200,
                "userLogin": 200,
                "auditorLogin": auditor.status_code,
                "adminModules": me.json()["data"]["allowedModules"],
                "adminAudit": audit_api.status_code,
                "userAttachmentLibrary": user_library.status_code,
                "userMountedAttachment": mounted.status_code,
                "recursiveDocumentDelta": 2,
                "allDocumentDelta": 2,
                "adminTreeCountDelta": 2,
                "userTreeCountDelta": 1,
                "docxPreview": docx_preview.status_code,
                "pdfPreview": pdf_preview.status_code,
                "docPreview": legacy_preview.status_code,
                "bodySearch": True,
                "queryAuditUnchanged": True,
                "previewDownloadCountUnchanged": True,
            }
        )
    finally:
        if "admin_token" in locals():
            if "main" in locals() and "attachment" in locals() and main_version:
                response = requests.delete(
                    f"{base_url}/main-plans/{main['id']}/attachments/"
                    f"{attachment['id']}",
                    params={"mainPlanRowVersion": main_version},
                    headers=_headers(admin_token),
                    timeout=10,
                )
                if response.ok:
                    main_version = response.json()["data"]["mainPlanRowVersion"]
            for kind, document_id, version in reversed(created):
                if "main" in locals() and document_id == main["id"]:
                    current = requests.get(
                        f"{base_url}/documents/{document_id}",
                        headers=_headers(admin_token),
                        timeout=10,
                    )
                    if current.ok:
                        version = current.json()["data"]["rowVersion"]
                path = "attachments" if kind == "attachment" else "documents"
                requests.delete(
                    f"{base_url}/{path}/{document_id}",
                    params={"rowVersion": version},
                    headers=_headers(admin_token),
                    timeout=10,
                )
        server.shutdown()
        server.server_close()
        thread.join(timeout=5)

    with app.app_context():
        active_left = db.session.scalar(
            select(func.count()).select_from(Document).where(
                Document.document_name.like("Q3_HTTP_%"),
                Document.is_deleted.is_(False),
            )
        )
    assert active_left == 0
    result["activeTemporaryDocumentsAfterCleanup"] = active_left
    print(json.dumps(result, ensure_ascii=False, sort_keys=True))


if __name__ == "__main__":
    main()