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