PayrollManagementSystem/backend/tests/integration/test_user_management.py

"""用户管理接口集成测试（需要 MySQL 与 Redis）。

守的是：用户 CRUD 的**串联正确性**与**审计留痕**。

审计是重点验收项：每个写操作（含失败的）都必须在 operation_logs 留下
可读记录 —— 模块、操作、摘要、操作人、请求元信息完整，且 request_data
中**绝无明文密码**；账号被删除后审计行通过「user_id 置空 + 冗余 username」
保留可追溯性。

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

import uuid

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

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

pytestmark = pytest.mark.integration

_PASSWORD = "TestPwd@2026"
_USERS = "/api/v1/system/users"


def _new_username() -> str:
    return f"t_{uuid.uuid4().hex[:8]}"


def _create_via_api(client: TestClient, admin: dict, username: str, **overrides: object) -> dict:
    """经接口创建用户并返回响应 data（账号归 autouse 清理兜底）。"""
    payload: dict = {
        "username": username,
        "password": _PASSWORD,
        "real_name": "接口新用户",
        "roles": ["user"],
    }
    payload.update(overrides)
    resp = client.post(_USERS, json=payload, headers=admin["headers"])
    assert resp.status_code == 200, resp.text
    return resp.json()["data"]


def _logs_of(username: str) -> list[OperationLog]:
    """取某操作人最近的审计行（按 id 倒序）。"""
    factory = get_session_factory()
    with factory() as session:
        rows = list(
            session.execute(
                select(OperationLog)
                .where(OperationLog.username == username)
                .order_by(OperationLog.id.desc())
            ).scalars()
        )
    return rows


class TestCrudFlow:
    """正向链路：建 → 查 → 改 → 启停 → 重置密码 → 删。"""

    def test_新增用户并可在列表查到(self, client: TestClient, admin: dict) -> None:
        username = _new_username()
        data = _create_via_api(client, admin, username)
        assert data["username"] == username
        assert data["roles"] == ["user"]
        assert data["status"] == 1

        resp = client.get(_USERS, params={"username": username}, headers=admin["headers"])
        body = resp.json()["data"]
        assert body["total"] == 1
        assert body["list"][0]["id"] == data["id"]

    def test_重名用户被拒(self, client: TestClient, admin: dict, make_user) -> None:
        target = make_user()
        resp = client.post(
            _USERS,
            json={"username": target["username"], "password": _PASSWORD, "roles": ["user"]},
            headers=admin["headers"],
        )
        assert resp.status_code == 400
        assert "已存在" in resp.json()["message"]

    def test_分页参数以驼峰命名生效(self, client: TestClient, admin: dict) -> None:
        """pageNum/pageSize 是前端契约（PageQuery 的 alias）。

        防回归：若被改成蛇形命名，FastAPI 会忽略这两个参数、恒返回
        第 1 页 10 条 —— 表现为「翻页没反应」，且没有任何报错。
        """
        resp = client.get(_USERS, params={"pageNum": 2, "pageSize": 5}, headers=admin["headers"])
        data = resp.json()["data"]
        assert {"list", "total", "pageNum", "pageSize"} <= set(data)
        assert data["pageNum"] == 2
        assert data["pageSize"] == 5
        assert len(data["list"]) <= 5

    def test_编辑用户信息与角色(self, client: TestClient, admin: dict, make_user) -> None:
        target = make_user()
        resp = client.put(
            f"{_USERS}/{target['id']}",
            json={"real_name": "改后姓名", "roles": ["admin"]},
            headers=admin["headers"],
        )
        assert resp.status_code == 200
        data = resp.json()["data"]
        assert data["real_name"] == "改后姓名"
        assert data["roles"] == ["admin"]

    def test_编辑用户保留既有角色不触发唯一键冲突(
        self, client: TestClient, admin: dict, make_user
    ) -> None:
        """回归：新旧角色集合有交集时（编辑时未改角色必然如此），
        `replace_roles` 若在 clear 后不先落库删除，SQLAlchemy 同一 flush
        内先 INSERT 后 DELETE 会撞 uk_user_roles_user_id_role_id（1062）。
        该缺陷最初由前端 e2e「编辑用户」步骤暴露 —— 集成测试当时只覆盖
        了「换掉全部角色」的场景。"""
        target = make_user(["user"])
        resp = client.put(
            f"{_USERS}/{target['id']}",
            json={"real_name": "保留角色", "roles": ["user", "admin"]},
            headers=admin["headers"],
        )
        assert resp.status_code == 200, resp.text
        assert sorted(resp.json()["data"]["roles"]) == ["admin", "user"]

    def test_禁用与重新启用(self, client: TestClient, admin: dict, make_user) -> None:
        target = make_user()
        resp = client.put(
            f"{_USERS}/{target['id']}/status", json={"status": 0}, headers=admin["headers"]
        )
        assert resp.status_code == 200
        assert resp.json()["data"]["status"] == 0

        resp = client.put(
            f"{_USERS}/{target['id']}/status", json={"status": 1}, headers=admin["headers"]
        )
        assert resp.json()["data"]["status"] == 1

    def test_禁用后旧令牌立即失效(
        self, client: TestClient, admin: dict, make_user, auth_headers
    ) -> None:
        """禁用会提 token_version —— 不依赖 Redis 的「离职即失效」。"""
        target = make_user()
        old_headers = auth_headers(target)
        assert client.get("/api/v1/auth/me", headers=old_headers).status_code == 200

        client.put(f"{_USERS}/{target['id']}/status", json={"status": 0}, headers=admin["headers"])
        assert client.get("/api/v1/auth/me", headers=old_headers).status_code == 401

    def test_重置密码后旧令牌立即失效(
        self, client: TestClient, admin: dict, make_user, auth_headers
    ) -> None:
        target = make_user()
        old_headers = auth_headers(target)
        assert client.get("/api/v1/auth/me", headers=old_headers).status_code == 200

        resp = client.put(
            f"{_USERS}/{target['id']}/password",
            json={"new_password": "NewPwd@2026"},
            headers=admin["headers"],
        )
        assert resp.status_code == 200
        assert client.get("/api/v1/auth/me", headers=old_headers).status_code == 401

    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(f"{_USERS}/{target['id']}", headers=admin["headers"])
        assert resp.status_code == 404


class TestSelfProtection:
    """自我保护：不允许对当前登录账号执行危险操作。"""

    def test_不能删除自己(self, client: TestClient, admin: dict) -> None:
        resp = client.delete(f"{_USERS}/{admin['id']}", headers=admin["headers"])
        assert resp.status_code == 403

    def test_不能禁用自己(self, client: TestClient, admin: dict) -> None:
        resp = client.put(
            f"{_USERS}/{admin['id']}/status", json={"status": 0}, headers=admin["headers"]
        )
        assert resp.status_code == 403

    def test_不能移除自己的管理员角色(self, client: TestClient, admin: dict) -> None:
        resp = client.put(
            f"{_USERS}/{admin['id']}", json={"roles": ["user"]}, headers=admin["headers"]
        )
        assert resp.status_code == 403


class TestValidation:
    """参数与业务校验。"""

    def test_无效角色编码被拒(self, client: TestClient, admin: dict) -> None:
        resp = client.post(
            _USERS,
            json={"username": _new_username(), "password": _PASSWORD, "roles": ["nonexist"]},
            headers=admin["headers"],
        )
        assert resp.status_code == 400
        assert "角色编码无效" in resp.json()["message"]

    def test_空角色列表被拒(self, client: TestClient, admin: dict) -> None:
        resp = client.post(
            _USERS,
            json={"username": _new_username(), "password": _PASSWORD, "roles": []},
            headers=admin["headers"],
        )
        assert resp.status_code == 400

    def test_密码25个汉字被拒(self, client: TestClient, admin: dict) -> None:
        """75 字节超 bcrypt 上限；按字符数校验拦不住（25 字符很短）。"""
        resp = client.post(
            _USERS,
            json={"username": _new_username(), "password": "中" * 25, "roles": ["user"]},
            headers=admin["headers"],
        )
        assert resp.status_code == 400

    def test_查询不存在的用户返回404(self, client: TestClient, admin: dict) -> None:
        resp = client.get(f"{_USERS}/999999999", headers=admin["headers"])
        assert resp.status_code == 404

    def test_编辑未修改字段不污染空值语义(self, client: TestClient, admin: dict, make_user) -> None:
        """回归：编辑表单把未修改字段以空串提交，此前会把库中 NULL 写成 ''——
        两类空值（未填 vs 空串）语义漂移且不可逆；'' 现归一为 NULL。"""
        target = make_user()  # 直建账号 phone/email/remark 均为 NULL
        resp = client.put(
            f"{_USERS}/{target['id']}",
            json={"real_name": "改名", "phone": "", "email": "", "remark": ""},
            headers=admin["headers"],
        )
        assert resp.status_code == 200
        data = resp.json()["data"]
        assert data["real_name"] == "改名"
        assert data["phone"] is None
        assert data["email"] is None
        assert data["remark"] is None


class TestFiltering:
    """筛选条件的口径。"""

    def test_含下划线的用户名不误配(self, client: TestClient, admin: dict, make_user) -> None:
        """回归：模糊筛选未转义 LIKE 通配符时 `_` 匹配任意单字符，
        搜 t_xx 会命中 txXx 之类无关账号 —— 用户名普遍含下划线，非理论场景。"""
        target = make_user()
        decoy_name = target["username"].replace("_", "x", 1)
        make_user(username=decoy_name)

        resp = client.get(_USERS, params={"username": target["username"]}, headers=admin["headers"])
        names = [row["username"] for row in resp.json()["data"]["list"]]
        assert target["username"] in names
        assert decoy_name not in names


class TestAuditTrail:
    """审计留痕 —— 本模块的核心验收点。"""

    def test_每个写操作都留下审计记录(self, client: TestClient, admin: dict, make_user) -> None:
        target = make_user()
        client.put(
            f"{_USERS}/{target['id']}",
            json={"real_name": "审计留痕"},
            headers=admin["headers"],
        )
        client.put(f"{_USERS}/{target['id']}/status", json={"status": 0}, headers=admin["headers"])
        client.put(
            f"{_USERS}/{target['id']}/password",
            json={"new_password": "AuditPwd@2026"},
            headers=admin["headers"],
        )
        client.delete(f"{_USERS}/{target['id']}", headers=admin["headers"])

        logs = _logs_of(admin["username"])
        actions = {(log.module, log.action) for log in logs}
        assert ("用户管理", "修改") in actions
        assert ("用户管理", "禁用") in actions
        assert ("用户管理", "重置密码") in actions
        assert ("用户管理", "删除") in actions

        # 摘要含目标用户名（供人阅读的追溯锚点）
        assert any(target["username"] in log.summary for log in logs)

        # 请求元信息与结果完整
        success_logs = [log for log in logs if log.success == 1]
        assert success_logs
        for log in success_logs:
            assert log.method and log.path
            assert log.duration_ms is not None and log.duration_ms >= 0

    def test_失败操作记为失败且带中文原因(self, client: TestClient, admin: dict) -> None:
        resp = client.delete(f"{_USERS}/999999999", headers=admin["headers"])
        assert resp.status_code == 404

        latest = _logs_of(admin["username"])[0]
        assert latest.success == 0
        assert latest.action == "删除"
        assert latest.error_message == "用户不存在"

    def test_审计请求参数绝不包含明文密码(self, client: TestClient, admin: dict, make_user) -> None:
        """密码是审计脱敏的红线：全表扫描该操作人的 request_data 不得命中。"""
        secret = "S3cret@2026!"
        _create_via_api(client, admin, _new_username(), password=secret)
        target = make_user()
        client.put(
            f"{_USERS}/{target['id']}/password",
            json={"new_password": secret},
            headers=admin["headers"],
        )

        factory = get_session_factory()
        with factory() as session:
            rows = list(
                session.execute(
                    select(OperationLog.request_data).where(
                        OperationLog.username == admin["username"]
                    )
                ).scalars()
            )
        joined = "\n".join(row for row in rows if row)
        assert secret not in joined
        assert "***" in joined  # 键名保留、值被抹除

    def test_账号删除后审计行保留用户名(
        self, client: TestClient, admin: dict, make_user, auth_headers
    ) -> None:
        """user_id 外键 SET NULL + username 冗余 —— 删除账号不得让审计断链。"""
        victim = make_user(["admin"])
        victim_headers = auth_headers(victim)
        # victim 作为操作人产生一条审计（失败操作最省事，不留业务数据）
        client.delete(f"{_USERS}/999999999", headers=victim_headers)

        assert (
            client.delete(f"{_USERS}/{victim['id']}", headers=admin["headers"]).status_code == 200
        )

        factory = get_session_factory()
        with factory() as session:
            logs = list(
                session.execute(
                    select(OperationLog).where(OperationLog.username == victim["username"])
                ).scalars()
            )
        assert logs, "审计行不应随账号删除而消失"
        assert all(log.user_id is None for log in logs)
        assert all(log.username == victim["username"] for log in logs)


class TestAuthorization:
    """权限：系统管理为管理员专属。"""

    def test_普通用户访问用户管理被拒(self, client: TestClient, normal: dict) -> None:
        assert client.get(_USERS, headers=normal["headers"]).status_code == 403
        resp = client.post(
            _USERS,
            json={"username": _new_username(), "password": _PASSWORD, "roles": ["user"]},
            headers=normal["headers"],
        )
        assert resp.status_code == 403