PayrollManagementSystem/backend/tests/integration/test_log_query.py

"""日志查询接口集成测试（需要 MySQL 与 Redis）。

守的是：两类日志的筛选、分页、权限，以及**写入与查询的口径一致** ——
写操作产生的审计记录必须能被查询接口按同样口径检索到（否则"有写入
没查询"或"查询条件对不上"，日志功能形同虚设）。

登录日志造数直接 ORM 插入，不经过登录接口：绕开验证码依赖，且能精确
控制 success/fail_reason/ip 组合。

运行前置：``docker compose -f docker/docker-compose.dev.yml up -d``
"""

import uuid
from datetime import datetime, timedelta

import pytest
from fastapi.testclient import TestClient
from sqlalchemy import delete

from app.core.db import get_session_factory
from app.models import LoginLog

pytestmark = pytest.mark.integration

_OPERATION_LOGS = "/api/v1/system/operation-logs"
_LOGIN_LOGS = "/api/v1/system/login-logs"
_USERS = "/api/v1/system/users"


def _make_audit(client: TestClient, admin: dict, action: str = "删除") -> None:
    """触发一条审计记录（失败操作最省事、不留业务数据）。"""
    resp = client.delete(f"{_USERS}/999999999", headers=admin["headers"])
    assert resp.status_code == 404
    assert action == "删除"


class TestOperationLogQuery:
    """操作日志查询。"""

    def test_写操作后可按模块与操作筛选(self, client: TestClient, admin: dict, make_user) -> None:
        target = make_user()
        resp = client.delete(f"{_USERS}/{target['id']}", headers=admin["headers"])
        assert resp.status_code == 200

        resp = client.get(
            _OPERATION_LOGS,
            params={"username": admin["username"], "module": "用户管理", "action": "删除"},
            headers=admin["headers"],
        )
        data = resp.json()["data"]
        assert data["total"] >= 1
        row = data["list"][0]
        assert row["success"] == 1
        assert target["username"] in row["summary"]
        assert row["module"] == "用户管理"

    def test_按结果筛选出失败记录(self, client: TestClient, admin: dict) -> None:
        _make_audit(client, admin)
        resp = client.get(
            _OPERATION_LOGS,
            params={"username": admin["username"], "success": 0},
            headers=admin["headers"],
        )
        data = resp.json()["data"]
        assert data["total"] >= 1
        assert data["list"][0]["error_message"]

    def test_时间范围筛选(self, client: TestClient, admin: dict) -> None:
        _make_audit(client, admin)
        now = datetime.now()
        fmt = "%Y-%m-%d %H:%M:%S"

        resp = client.get(
            _OPERATION_LOGS,
            params={
                "username": admin["username"],
                "start_time": (now - timedelta(minutes=5)).strftime(fmt),
                "end_time": (now + timedelta(minutes=5)).strftime(fmt),
            },
            headers=admin["headers"],
        )
        assert resp.json()["data"]["total"] >= 1

        # 已过去的时间窗（排除测试刚产生的记录）
        resp = client.get(
            _OPERATION_LOGS,
            params={
                "username": admin["username"],
                "start_time": "2020-01-01 00:00:00",
                "end_time": "2020-01-02 00:00:00",
            },
            headers=admin["headers"],
        )
        assert resp.json()["data"]["total"] == 0

    def test_起始晚于结束返回400(self, client: TestClient, admin: dict) -> None:
        resp = client.get(
            _OPERATION_LOGS,
            params={"start_time": "2026-10-02 00:00:00", "end_time": "2026-10-01 00:00:00"},
            headers=admin["headers"],
        )
        assert resp.status_code == 400

    def test_列表响应不含请求参数字段(self, client: TestClient, admin: dict) -> None:
        """列表为性能排除 request_data（TEXT 大字段），防止翻页负担随日志表增长。"""
        resp = client.get(_OPERATION_LOGS, headers=admin["headers"])
        rows = resp.json()["data"]["list"]
        if rows:
            assert "request_data" not in rows[0]

    def test_详情端点返回已脱敏的请求参数(self, client: TestClient, admin: dict, make_user) -> None:
        """列表 → 详情 的两段式：详情含 request_data 且已脱敏。"""
        secret = "S3cret@2026!"
        target = make_user()
        resp = client.put(
            f"{_USERS}/{target['id']}/password",
            json={"new_password": secret},
            headers=admin["headers"],
        )
        assert resp.status_code == 200

        resp = client.get(
            _OPERATION_LOGS,
            params={"username": admin["username"], "action": "重置密码"},
            headers=admin["headers"],
        )
        row = resp.json()["data"]["list"][0]
        assert "request_data" not in row

        detail = client.get(f"{_OPERATION_LOGS}/{row['id']}", headers=admin["headers"])
        assert detail.status_code == 200
        request_data = detail.json()["data"]["request_data"] or ""
        assert secret not in request_data
        assert "***" in request_data

    def test_详情端点不存在返回404(self, client: TestClient, admin: dict) -> None:
        resp = client.get(f"{_OPERATION_LOGS}/999999999", headers=admin["headers"])
        assert resp.status_code == 404

    def test_详情端点普通用户被拒(self, client: TestClient, normal: dict) -> None:
        resp = client.get(f"{_OPERATION_LOGS}/1", headers=normal["headers"])
        assert resp.status_code == 403

    def test_普通用户被拒(self, client: TestClient, normal: dict) -> None:
        assert client.get(_OPERATION_LOGS, headers=normal["headers"]).status_code == 403


class TestLoginLogQuery:
    """登录日志查询。"""

    @pytest.fixture
    def seeded_logs(self):
        """直插两条登录日志（一成功一失败），用后清理。"""
        username = f"t_login_{uuid.uuid4().hex[:6]}"
        factory = get_session_factory()
        with factory() as session:
            session.add(
                LoginLog(
                    username=username,
                    ip="10.1.2.3",
                    user_agent="pytest",
                    success=0,
                    fail_reason="用户名或密码错误",
                )
            )
            session.add(LoginLog(username=username, ip="10.1.2.4", user_agent="pytest", success=1))
            session.commit()
        yield username
        with factory() as session:
            session.execute(delete(LoginLog).where(LoginLog.username == username))
            session.commit()

    def test_按用户名查询并返回两端结果(
        self, client: TestClient, admin: dict, seeded_logs: str
    ) -> None:
        resp = client.get(_LOGIN_LOGS, params={"username": seeded_logs}, headers=admin["headers"])
        data = resp.json()["data"]
        assert data["total"] == 2
        assert {row["success"] for row in data["list"]} == {0, 1}

    def test_按IP精确筛选(self, client: TestClient, admin: dict, seeded_logs: str) -> None:
        resp = client.get(_LOGIN_LOGS, params={"ip": "10.1.2.3"}, headers=admin["headers"])
        data = resp.json()["data"]
        assert data["total"] == 1
        assert data["list"][0]["fail_reason"] == "用户名或密码错误"

    def test_按结果筛选(self, client: TestClient, admin: dict, seeded_logs: str) -> None:
        resp = client.get(
            _LOGIN_LOGS,
            params={"username": seeded_logs, "success": 0},
            headers=admin["headers"],
        )
        data = resp.json()["data"]
        assert data["total"] == 1
        assert data["list"][0]["success"] == 0

    def test_普通用户被拒(self, client: TestClient, normal: dict) -> None:
        assert client.get(_LOGIN_LOGS, headers=normal["headers"]).status_code == 403