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

from __future__ import annotations

from datetime import UTC, datetime

import pytest


RANGE = "createdFrom=2026-01-01T00:00:00.000Z&createdTo=2026-02-01T00:00:00.000Z"


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


def _get(client, path: str, token: str):
    return client.get(path, headers=_headers(token))


@pytest.mark.parametrize(
    "path",
    [
        "/api/v1/audit/logs",
        "/api/v1/audit/statistics/trend?granularity=DAY",
        "/api/v1/audit/statistics/actions",
        "/api/v1/audit/statistics/users",
        "/api/v1/statistics/documents",
    ],
)
def test_all_b7_endpoints_require_authentication(b7_client, path):
    assert b7_client.get(path).status_code == 401


@pytest.mark.parametrize(
    "path",
    [
        "/api/v1/audit/logs",
        "/api/v1/audit/statistics/trend?granularity=DAY",
        "/api/v1/audit/statistics/actions",
        "/api/v1/audit/statistics/users",
    ],
)
def test_audit_endpoints_are_auditor_only(b7_client, token_for, path):
    assert _get(b7_client, path, token_for("auditor")).status_code == 200
    assert _get(b7_client, path, token_for("admin")).status_code == 403
    assert _get(b7_client, path, token_for("user")).status_code == 403


def test_document_statistics_is_admin_only(b7_client, token_for):
    assert _get(
        b7_client, "/api/v1/statistics/documents", token_for("admin")
    ).status_code == 200
    assert _get(
        b7_client, "/api/v1/statistics/documents", token_for("auditor")
    ).status_code == 403
    assert _get(
        b7_client, "/api/v1/statistics/documents", token_for("user")
    ).status_code == 403


def test_audit_list_dto_paging_and_stable_sort(b7_client, token_for):
    token = token_for("auditor")
    response = _get(
        b7_client,
        f"/api/v1/audit/logs?{RANGE}&page=1&pageSize=2&sortDirection=asc",
        token,
    )
    assert response.status_code == 200
    data = response.get_json()["data"]
    assert (data["page"], data["pageSize"], data["total"], data["totalPages"]) == (
        1,
        2,
        8,
        4,
    )
    expected_fields = {
        "id",
        "username",
        "realName",
        "organizationName",
        "actionType",
        "targetType",
        "targetId",
        "targetName",
        "operationResult",
        "failureReason",
        "clientIp",
        "requestId",
        "createdAt",
    }
    assert set(data["items"][0]) == expected_fields
    assert data["items"][0]["createdAt"] == "2026-01-05T00:00:00.000Z"
    assert isinstance(data["items"][0]["id"], str)
    assert isinstance(data["items"][0]["targetId"], str)
    assert "operationDetail" not in data["items"][0]
    assert "userAgent" not in data["items"][0]

    tied = _get(
        b7_client,
        "/api/v1/audit/logs?"
        "createdFrom=2026-01-07T01:00:00.000Z&"
        "createdTo=2026-01-07T01:00:00.001Z&sortDirection=asc",
        token,
    ).get_json()["data"]["items"]
    assert [int(item["id"]) for item in tied] == sorted(
        int(item["id"]) for item in tied
    )


@pytest.mark.parametrize(
    ("keyword", "expected_request"),
    [
        ("admin", "b7-request-alpha"),
        ("管理员旧名", "b7-request-alpha"),
        ("旧组织", "b7-request-alpha"),
        ("关键目标", "b7-request-alpha"),
        ("关键失败原因", "b7-request-beta"),
        ("b7-request-gamma", "b7-request-gamma"),
    ],
)
def test_keyword_searches_only_contract_fields(
    b7_client, token_for, keyword, expected_request
):
    response = _get(
        b7_client,
        f"/api/v1/audit/logs?{RANGE}&keyword={keyword}",
        token_for("auditor"),
    )
    request_ids = {
        item["requestId"] for item in response.get_json()["data"]["items"]
    }
    assert expected_request in request_ids


def test_keyword_escapes_sql_wildcards(b7_client, token_for):
    response = _get(
        b7_client,
        f"/api/v1/audit/logs?{RANGE}&keyword=%25",
        token_for("auditor"),
    )
    assert response.get_json()["data"]["total"] == 0


@pytest.mark.parametrize(
    ("query", "request_id"),
    [
        ("userId={admin}", "b7-request-alpha"),
        ("organizationId={admin_org}", "b7-request-alpha"),
        ("actionType=DOWNLOAD_DOCUMENT", "b7-request-beta"),
        ("operationResult=FAILURE", "b7-request-beta"),
        ("targetType=AUTH", "b7-request-gamma"),
        ("targetId=103", "b7-request-epsilon"),
    ],
)
def test_audit_list_contract_filters(
    b7_app, b7_client, token_for, query, request_id
):
    with b7_app.app_context():
        from dms.extensions import db
        from dms.models import User

        admin = db.session.scalar(
            db.select(User).where(User.username == "admin")
        )
        query = query.format(
            admin=admin.id, admin_org=admin.organization_id
        )
    response = _get(
        b7_client,
        f"/api/v1/audit/logs?{RANGE}&{query}",
        token_for("auditor"),
    )
    assert request_id in {
        item["requestId"] for item in response.get_json()["data"]["items"]
    }


def test_time_range_is_half_open_and_each_bound_is_optional(
    b7_client, token_for
):
    token = token_for("auditor")
    inclusive = _get(
        b7_client,
        "/api/v1/audit/logs?"
        "createdFrom=2026-01-05T00:00:00.000Z&"
        "createdTo=2026-01-05T01:00:00.000Z",
        token,
    ).get_json()["data"]["items"]
    assert [item["requestId"] for item in inclusive] == ["b7-request-alpha"]
    only_from = _get(
        b7_client,
        "/api/v1/audit/logs?createdFrom=2026-02-01T00:00:00.000Z",
        token,
    ).get_json()["data"]
    assert any(item["requestId"] == "b7-request-iota" for item in only_from["items"])
    only_to = _get(
        b7_client,
        "/api/v1/audit/logs?createdTo=2026-01-05T00:00:00.001Z",
        token,
    ).get_json()["data"]
    assert only_to["total"] == 1


@pytest.mark.parametrize(
    "query",
    [
        "createdFrom=2026-01-01T00:00:00",
        "createdFrom=2026-01-01T00:00:00%2B08:00",
        "createdFrom=not-a-date",
        "createdFrom=2026-01-02T00:00:00Z&createdTo=2026-01-01T00:00:00Z",
        "actionType=UNKNOWN",
        "operationResult=UNKNOWN",
        "targetType=UNKNOWN",
        "sortDirection=sideways",
        "page=0",
        "pageSize=101",
        "userId=abc",
        "unknown=value",
        "page=1&page=2",
    ],
)
def test_invalid_audit_list_parameters_return_400(
    b7_client, token_for, query
):
    response = _get(
        b7_client, f"/api/v1/audit/logs?{query}", token_for("auditor")
    )
    assert response.status_code == 400
    assert response.get_json()["code"] == "INVALID_ARGUMENT"


def test_document_statistics_and_business_timezone_day(
    b7_app, b7_client, token_for, monkeypatch
):
    from dms.extensions import db
    from dms.models import Document
    from dms.services import audit_query_service

    monkeypatch.setattr(
        audit_query_service,
        "_utc_now",
        lambda: datetime(2026, 1, 7, 12, 0, tzinfo=UTC),
    )
    response = _get(
        b7_client, "/api/v1/statistics/documents", token_for("admin")
    )
    data = response.get_json()["data"]
    with b7_app.app_context():
        counts = {
            kind: db.session.scalar(
                db.select(db.func.count(Document.id)).where(
                    Document.document_type == kind,
                    Document.is_deleted.is_(False),
                )
            )
            for kind in ("MAIN", "SUB_PLAN", "ATTACHMENT")
        }
    assert data["mainPlanCount"] == counts["MAIN"]
    assert data["subPlanCount"] == counts["SUB_PLAN"]
    assert data["attachmentCount"] == counts["ATTACHMENT"]
    assert data["planDocumentCount"] == counts["MAIN"] + counts["SUB_PLAN"]
    assert data["todayViewerCount"] == 2


@pytest.mark.parametrize(
    ("granularity", "expected_periods"),
    [
        ("DAY", ["2026-01-05", "2026-01-06", "2026-01-07", "2026-01-08"]),
        ("WEEK", ["2026-01-05"]),
        ("MONTH", ["2026-01"]),
    ],
)
def test_trend_database_buckets(
    b7_client, token_for, granularity, expected_periods
):
    response = _get(
        b7_client,
        f"/api/v1/audit/statistics/trend?{RANGE}&granularity={granularity}",
        token_for("auditor"),
    )
    assert response.status_code == 200
    items = response.get_json()["data"]["items"]
    assert [item["period"] for item in items] == expected_periods
    assert sum(item["operationCount"] for item in items) == 8


def test_trend_requires_valid_granularity_and_supports_empty_result(
    b7_client, token_for
):
    token = token_for("auditor")
    assert _get(
        b7_client, "/api/v1/audit/statistics/trend", token
    ).status_code == 400
    assert _get(
        b7_client,
        "/api/v1/audit/statistics/trend?granularity=YEAR",
        token,
    ).status_code == 400
    data = _get(
        b7_client,
        "/api/v1/audit/statistics/trend?"
        "granularity=DAY&createdFrom=2025-01-01T00:00:00Z&"
        "createdTo=2025-01-02T00:00:00Z",
        token,
    ).get_json()["data"]
    assert data == {"items": []}


def test_action_distribution_order_and_all_results(b7_client, token_for):
    items = _get(
        b7_client,
        f"/api/v1/audit/statistics/actions?{RANGE}",
        token_for("auditor"),
    ).get_json()["data"]["items"]
    assert items[0] == {"actionType": "VIEW_DOCUMENT", "count": 5}
    assert sum(item["count"] for item in items) == 8
    tied = [item["actionType"] for item in items if item["count"] == 1]
    assert tied == sorted(tied)


def test_active_users_latest_audit_snapshot_and_order(b7_client, token_for):
    items = _get(
        b7_client,
        f"/api/v1/audit/statistics/users?{RANGE}&limit=2",
        token_for("auditor"),
    ).get_json()["data"]["items"]
    assert [item["operationCount"] for item in items] == [4, 2]
    assert items[0]["realName"] == "管理员新名"
    assert items[0]["organizationName"] == "新组织"
    assert all(isinstance(item["userId"], str) for item in items)


@pytest.mark.parametrize("limit", ["0", "101", "abc"])
def test_active_user_limit_validation(b7_client, token_for, limit):
    assert _get(
        b7_client,
        f"/api/v1/audit/statistics/users?limit={limit}",
        token_for("auditor"),
    ).status_code == 400


def test_audit_queries_do_not_write_audit_log(b7_app, b7_client, token_for):
    from dms.extensions import db
    from dms.models import AuditLog

    token = token_for("auditor")
    with b7_app.app_context():
        before = db.session.scalar(db.select(db.func.count(AuditLog.id)))
    for path in (
        f"/api/v1/audit/logs?{RANGE}",
        f"/api/v1/audit/statistics/trend?{RANGE}&granularity=DAY",
        f"/api/v1/audit/statistics/actions?{RANGE}",
        f"/api/v1/audit/statistics/users?{RANGE}",
    ):
        assert _get(b7_client, path, token).status_code == 200
    with b7_app.app_context():
        after = db.session.scalar(db.select(db.func.count(AuditLog.id)))
    assert after == before


def test_b7_real_http_on_8755_uses_only_formal_apis_for_audit_writes(
    b4_app, b2_password
):
    """单进程内启停真实TCP服务，避免后台子进程遗留或占住测试通道。"""
    import json
    import threading
    from datetime import timedelta
    from urllib.error import HTTPError
    from urllib.parse import urlencode
    from urllib.request import Request, urlopen

    from werkzeug.serving import make_server
    import requests

    from dms.extensions import db
    from dms.models import User

    origin = "http://127.0.0.1:9346"
    base_url = "http://127.0.0.1:8755"

    def call(method, path, *, token=None, body=None):
        headers = {"Origin": origin}
        if token:
            headers["Authorization"] = f"Bearer {token}"
        data = None
        if body is not None:
            data = json.dumps(body, ensure_ascii=False).encode("utf-8")
            headers["Content-Type"] = "application/json"
        request = Request(
            f"{base_url}{path}",
            data=data,
            headers=headers,
            method=method,
        )
        try:
            response = urlopen(request, timeout=5)
        except HTTPError as error:
            response = error
        content = response.read()
        payload = (
            json.loads(content.decode("utf-8"))
            if response.headers.get_content_type() == "application/json"
            else content
        )
        return response.status, response.headers, payload

    def login(username, password=b2_password):
        status, _, payload = call(
            "POST",
            "/api/v1/auth/login",
            body={
                "username": username,
                "password": password,
                "keepSignedIn": False,
            },
        )
        assert status == 200
        return payload["data"]["accessToken"]

    with b4_app.app_context():
        ids = b4_app.config["B4_IDS"]
        admin = db.session.scalar(
            db.select(User).where(User.username == "admin")
        )
        admin_id = admin.id
        organization_id = admin.organization_id

    server = make_server("127.0.0.1", 8755, b4_app, threaded=True)
    thread = threading.Thread(target=server.serve_forever, daemon=True)
    thread.start()
    try:
        admin_token = login("admin")
        user_token = login("user")
        assert call(
            "POST",
            "/api/v1/auth/login",
            body={
                "username": "missing-user",
                "password": "wrong-password",
                "keepSignedIn": False,
            },
        )[0] == 401
        auditor_token = login("auditor")

        assert call(
            "GET", f"/api/v1/documents/{ids['main_all']}", token=admin_token
        )[0] == 200
        assert call(
            "GET", f"/api/v1/documents/{ids['main_all']}", token=user_token
        )[0] == 200
        upload_response = requests.post(
            f"{base_url}/api/v1/attachments",
            files={
                "file": (
                    "B7真实HTTP附件.pdf",
                    b"%PDF-1.4\nB7-REAL-HTTP-DOWNLOAD",
                    "application/pdf",
                )
            },
            data={
                "metadata": json.dumps(
                    {
                        "documentName": "B7真实HTTP附件",
                        "attachmentType": "POLICY",
                        "summary": "B7真实HTTP验证",
                        "tags": ["B7"],
                    },
                    ensure_ascii=False,
                )
            },
            headers={
                "Authorization": f"Bearer {admin_token}",
                "Origin": origin,
            },
            timeout=5,
        )
        assert upload_response.status_code == 201, upload_response.text
        attachment_id = upload_response.json()["data"]["id"]
        download_result = call(
            "GET",
            f"/api/v1/documents/{attachment_id}/download",
            token=admin_token,
        )
        assert download_result[0] == 200, download_result[2]
        status, _, _ = call(
            "POST",
            "/api/v1/categories",
            token=admin_token,
            body={
                "categoryCode": "B7_HTTP_CATEGORY",
                "categoryName": "B7真实HTTP分类",
                "categoryType": "OTHER",
                "parentId": None,
                "sortNo": 99,
            },
        )
        assert status == 201

        now = datetime.now(UTC)
        time_query = {
            "createdFrom": (now - timedelta(minutes=5))
            .isoformat(timespec="milliseconds")
            .replace("+00:00", "Z"),
            "createdTo": (now + timedelta(minutes=5))
            .isoformat(timespec="milliseconds")
            .replace("+00:00", "Z"),
        }
        base_query = urlencode(time_query)
        status, headers, payload = call(
            "GET",
            f"/api/v1/audit/logs?{base_query}&page=1&pageSize=100&"
            "sortDirection=asc",
            token=auditor_token,
        )
        assert status == 200
        assert headers["Access-Control-Allow-Origin"] == origin
        assert "X-Request-Id" in headers["Access-Control-Expose-Headers"]
        before_total = payload["data"]["total"]
        assert before_total >= 9

        checks = [
            f"/api/v1/audit/logs?{base_query}&"
            f"{urlencode({'keyword': 'B7真实HTTP分类'})}",
            f"/api/v1/audit/logs?{base_query}&userId={admin_id}",
            f"/api/v1/audit/logs?{base_query}&organizationId={organization_id}",
            f"/api/v1/audit/logs?{base_query}&actionType=VIEW_DOCUMENT",
            f"/api/v1/audit/logs?{base_query}&operationResult=FAILURE",
            f"/api/v1/audit/logs?{base_query}&targetType=DOCUMENT&"
            f"targetId={ids['main_all']}",
            f"/api/v1/audit/statistics/trend?{base_query}&granularity=DAY",
            f"/api/v1/audit/statistics/trend?{base_query}&granularity=WEEK",
            f"/api/v1/audit/statistics/trend?{base_query}&granularity=MONTH",
            f"/api/v1/audit/statistics/actions?{base_query}",
            f"/api/v1/audit/statistics/users?{base_query}&limit=10",
        ]
        for path in checks:
            assert call("GET", path, token=auditor_token)[0] == 200

        status, _, statistics = call(
            "GET", "/api/v1/statistics/documents", token=admin_token
        )
        assert status == 200
        assert statistics["data"]["todayViewerCount"] == 2
        assert call(
            "GET", "/api/v1/audit/logs", token=admin_token
        )[0] == 403
        assert call(
            "GET", "/api/v1/statistics/documents", token=auditor_token
        )[0] == 403
        assert call(
            "GET", "/api/v1/audit/export", token=auditor_token
        )[0] == 404

        after_total = call(
            "GET",
            f"/api/v1/audit/logs?{base_query}&pageSize=100",
            token=auditor_token,
        )[2]["data"]["total"]
        assert after_total == before_total
    finally:
        server.shutdown()
        server.server_close()
        thread.join(timeout=5)
        assert not thread.is_alive()