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

from __future__ import annotations

from uuid import UUID, uuid4


def test_dms_blueprint_is_registered(flask_app):
    rules = {rule.rule for rule in flask_app.url_map.iter_rules()}
    assert "/api/v1/health" in rules


def test_only_q2b_dms_routes_are_registered(flask_app):
    dms_rules = {
        rule.rule for rule in flask_app.url_map.iter_rules()
        if rule.rule.startswith("/api/v1")
    }
    assert dms_rules == {
        "/api/v1/health",
        "/api/v1/auth/login",
        "/api/v1/auth/logout",
        "/api/v1/auth/reauth",
        "/api/v1/auth/portal/verify",
        "/api/v1/auth/portal/setup-password",
        "/api/v1/users/me",
        "/api/v1/users/me/ui-theme",
        "/api/v1/users",
        "/api/v1/users/<id>",
        "/api/v1/users/<id>/reset-password",
        "/api/v1/users/reorder",
        "/api/v1/organizations/tree",
        "/api/v1/organizations",
        "/api/v1/organizations/<id>",
        "/api/v1/organizations/<id>/ip-assignments",
        "/api/v1/organizations/reorder",
        "/api/v1/config/ui-dictionaries",
        "/api/v1/config/login-background",
        "/api/v1/categories/tree",
        "/api/v1/categories",
        "/api/v1/categories/<id>",
        "/api/v1/categories/reorder",
        "/api/v1/documents",
        "/api/v1/documents/<id>",
        "/api/v1/documents/batch-import",
        "/api/v1/documents/<id>/preview",
        "/api/v1/documents/<id>/re-extract",
        "/api/v1/documents/<id>/move",
        "/api/v1/documents/<id>/download",
        "/api/v1/documents/<id>/description",
        "/api/v1/documents/<id>/password",
        "/api/v1/documents/<id>/relations",
        "/api/v1/documents/relation-presets",
        "/api/v1/documents/reorder",
        "/api/v1/documents/tags",
        "/api/v1/main-plans/<id>/sub-plans",
        "/api/v1/main-plans/<id>/sub-plans/bind",
        "/api/v1/main-plans/<id>/sub-plans/<subPlanId>",
        "/api/v1/attachments",
        "/api/v1/attachments/<id>",
        "/api/v1/attachments/batch-import",
        "/api/v1/attachments/<id>/main-plans",
        "/api/v1/main-plans/<id>/attachments",
        "/api/v1/main-plans/<id>/attachments/bind",
        "/api/v1/main-plans/<id>/attachments/reorder",
        "/api/v1/main-plans/<id>/attachments/<attachmentId>",
        "/api/v1/main-plans/<id>/attachments/<bindingId>/move",
        "/api/v1/documents/<id>/permissions",
        "/api/v1/documents/<id>/permissions/entries/<entryId>/voucher",
        "/api/v1/roles",
        "/api/v1/roles/<id>",
        "/api/v1/roles/<id>/permissions",
        "/api/v1/roles/<id>/users",
        "/api/v1/roles/reorder",
        "/api/v1/ip-entries",
        "/api/v1/ip-entries/<id>",
        "/api/v1/backup/config",
        "/api/v1/backup/jobs",
        "/api/v1/backup/jobs/current",
        "/api/v1/backup/jobs/current/cancel",
        "/api/v1/backup/test-target",
        "/api/v1/system/metrics/current",
        "/api/v1/system/metrics/history",
        "/api/v1/system/process",
        "/api/v1/system/storage-usage",
        "/api/v1/system/converter",
        "/api/v1/system/health",
        "/api/v1/system/info",
        "/api/v1/system/runtime-config",
        "/api/v1/system/logs",
        "/api/v1/system/logs/download",
        "/api/v1/system/maintenance/preview-cache",
        "/api/v1/system/users/<user_id>/force-offline",
        "/api/v1/audit/logs",
        "/api/v1/statistics/documents",
        "/api/v1/audit/statistics/trend",
        "/api/v1/audit/statistics/actions",
        "/api/v1/audit/statistics/users",
        "/api/v1/recycle-bin/documents",
        "/api/v1/recycle-bin/documents/<id>/restore",
    }


def test_legacy_health_response_is_unchanged(client):
    response = client.get("/api/health")
    assert response.status_code == 200
    assert response.get_json() == {"status": "ok"}
    assert "requestId" not in response.get_json()


def test_dms_health_uses_contract_response(client):
    response = client.get("/api/v1/health")
    payload = response.get_json()

    assert response.status_code == 200
    assert payload == {
        "code": "OK",
        "message": "success",
        "data": {"service": "dms", "status": "UP"},
        "requestId": payload["requestId"],
    }
    UUID(payload["requestId"])
    assert response.headers["X-Request-Id"] == payload["requestId"]


def test_request_id_is_passed_through(client):
    request_id = str(uuid4())
    response = client.get(
        "/api/v1/health",
        headers={"X-Request-Id": request_id},
    )
    assert response.get_json()["requestId"] == request_id
    assert response.headers["X-Request-Id"] == request_id


def test_request_id_is_generated_when_absent(client):
    response = client.get("/api/v1/health")
    UUID(response.get_json()["requestId"])


def test_invalid_request_id_uses_uniform_error_response(client):
    response = client.get(
        "/api/v1/health",
        headers={"X-Request-Id": "not-a-uuid"},
    )
    payload = response.get_json()

    assert response.status_code == 400
    assert payload["code"] == "INVALID_ARGUMENT"
    assert payload["details"] is None
    assert set(payload) == {"code", "message", "details", "requestId"}
    assert response.headers["X-Request-Id"] == payload["requestId"]
    UUID(payload["requestId"])


def test_unknown_dms_route_uses_uniform_not_found(client):
    request_id = str(uuid4())
    response = client.get(
        "/api/v1/not-implemented",
        headers={"X-Request-Id": request_id},
    )
    assert response.status_code == 404
    assert response.get_json() == {
        "code": "RESOURCE_NOT_FOUND",
        "message": "请求的资源不存在",
        "details": None,
        "requestId": request_id,
    }
    assert response.headers["X-Request-Id"] == request_id