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 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 main["visibilityType"] == "CUSTOM"

        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={
                "visibilityType": "CUSTOM",
                "documentRowVersion": main_version,
                "entries": [
                    {
                        "subjectType": "USER",
                        "subjectId": user_id,
                        "canView": True,
                        "canDownload": True,
                        "canEdit": False,
                        "canManagePermission": False,
                        "canDelete": False,
                    }
                ],
            },
            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={"keyword": 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()