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

"""Q1-B 8755短生命周期真实HTTP复验工具。"""

from __future__ import annotations

import json
import os
import sys
import threading
from pathlib import Path
from urllib.parse import urlparse

import requests
from sqlalchemy import func, select
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() -> None:
    database_url = os.environ.get("DMS_DATABASE_URL", "")
    database_name = urlparse(database_url).path.lstrip("/")
    if not database_name.endswith("_test"):
        raise RuntimeError("Q1-B真实HTTP复验只允许连接以_test结尾的测试库")


def main() -> None:
    _assert_test_database()
    password = os.environ["DMS_TEST_USER_PASSWORD"]

    from app import app
    from dms.extensions import db
    from dms.models import AuditLog

    server = make_server("127.0.0.1", 8755, app, threaded=True)
    server_thread = threading.Thread(
        target=server.serve_forever,
        name="q1b-http-verify",
        daemon=True,
    )
    server_thread.start()
    base_url = "http://127.0.0.1:8755/api/v1"
    origin = "http://127.0.0.1:9346"

    try:
        tokens: dict[str, str] = {}
        for username in ("admin", "user"):
            response = requests.post(
                f"{base_url}/auth/login",
                json={
                    "username": username,
                    "password": password,
                    "keepSignedIn": False,
                },
                timeout=5,
            )
            response.raise_for_status()
            tokens[username] = response.json()["data"]["accessToken"]

        with app.app_context():
            audit_before = db.session.scalar(
                select(func.count()).select_from(AuditLog)
            )

        role_results = {}
        stable_data = None
        for username, token in tokens.items():
            response = requests.get(
                f"{base_url}/config/ui-dictionaries",
                headers={
                    "Authorization": f"Bearer {token}",
                    "Origin": origin,
                },
                timeout=5,
            )
            response.raise_for_status()
            body = response.json()
            assert body["requestId"] == response.headers["X-Request-Id"]
            assert response.headers["Access-Control-Allow-Origin"] == origin
            exposed = {
                value.strip().lower()
                for value in response.headers[
                    "Access-Control-Expose-Headers"
                ].split(",")
            }
            assert "x-request-id" in exposed
            assert set(body["data"]["dictionaries"]) == {
                "roleCodes",
                "allowedModules",
                "documentTypes",
                "documentStatuses",
                "securityLevels",
                "visibilityTypes",
                "attachmentTypes",
                "subjectTypes",
                "allowedActions",
                "categoryTypes",
                "enabledStatuses",
                "auditResults",
                "auditActions",
                "auditTargets",
            }
            if stable_data is None:
                stable_data = body["data"]
            else:
                assert body["data"] == stable_data
            role_results[username] = {
                "status": response.status_code,
                "requestIdMatches": True,
            }

        repeated = requests.get(
            f"{base_url}/config/ui-dictionaries",
            headers={
                "Authorization": f"Bearer {tokens['admin']}",
                "Origin": origin,
            },
            timeout=5,
        )
        repeated.raise_for_status()
        assert repeated.json()["data"] == stable_data

        unauthenticated = requests.get(
            f"{base_url}/config/ui-dictionaries",
            headers={"Origin": origin},
            timeout=5,
        )
        assert unauthenticated.status_code == 401
        unauthenticated_body = unauthenticated.json()
        assert (
            unauthenticated_body["requestId"]
            == unauthenticated.headers["X-Request-Id"]
        )

        labels = {
            item["code"]: item["label"]
            for item in stable_data["dictionaries"]["securityLevels"]
        }
        assert labels["INTERNAL"] == "内部"
        assert repeated.encoding.lower() == "utf-8"

        with app.app_context():
            audit_after = db.session.scalar(
                select(func.count()).select_from(AuditLog)
            )
        assert audit_after == audit_before

        print(
            json.dumps(
                {
                    "port": 8755,
                    "database": "dms_test",
                    "roles": role_results,
                    "unauthenticatedStatus": unauthenticated.status_code,
                    "dictionaryCount": len(stable_data["dictionaries"]),
                    "utf8Chinese": True,
                    "continuousResponsesStable": True,
                    "auditCountBefore": audit_before,
                    "auditCountAfter": audit_after,
                    "requestIdMatches": True,
                    "corsOrigin": origin,
                    "exposeHeaders": repeated.headers[
                        "Access-Control-Expose-Headers"
                    ],
                },
                ensure_ascii=False,
            )
        )
    finally:
        server.shutdown()
        server.server_close()
        server_thread.join(timeout=5)


if __name__ == "__main__":
    main()