PayrollManagementSystem/backend/tests/integration/test_role_management.py

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

守的是角色管理的**保护规则** —— 它们不是界面体验，而是系统自举与授权
安全的最低防线：

- `admin` 角色禁删禁停（cli 建管理员与 require_admin 判定都依赖该编码）
- 有用户绑定的角色禁删（user_roles 外键是 CASCADE，直接删会静默剥离权限）
- `role_code` 创建后不可改（授权判定的锚点）

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

import uuid

import pytest
from fastapi.testclient import TestClient

from app.core.constants import ROLE_USER

pytestmark = pytest.mark.integration

_ROLES = "/api/v1/system/roles"


def _new_code() -> str:
    return f"t_role_{uuid.uuid4().hex[:6]}"


def _admin_role_id(client: TestClient, admin: dict) -> int:
    """从管理列表取内置 admin 角色的 id。"""
    resp = client.get(_ROLES, params={"keyword": "admin"}, headers=admin["headers"])
    rows = resp.json()["data"]["list"]
    matches = [row for row in rows if row["role_code"] == "admin"]
    assert matches, "内置 admin 角色缺失"
    return matches[0]["id"]


class TestCrudFlow:
    """正向链路：建 → 查（列表 + 下拉）→ 改 → 停用 → 删。"""

    def test_新增角色并联动列表与下拉(self, client: TestClient, admin: dict) -> None:
        code = _new_code()
        resp = client.post(
            _ROLES,
            json={"role_code": code, "role_name": "对外协作岗", "sort_order": 50},
            headers=admin["headers"],
        )
        assert resp.status_code == 200, resp.text
        role_id = resp.json()["data"]["id"]
        assert resp.json()["data"]["user_count"] == 0

        # 管理列表可查（keyword 同时匹配编码与名称）
        resp = client.get(_ROLES, params={"keyword": code}, headers=admin["headers"])
        assert resp.json()["data"]["total"] == 1

        # 用户表单下拉包含（启用状态）
        resp = client.get("/api/v1/system/users/meta/roles", headers=admin["headers"])
        assert code in [row["role_code"] for row in resp.json()["data"]]

        assert client.delete(f"{_ROLES}/{role_id}", headers=admin["headers"]).status_code == 200

    def test_重复编码被拒(self, client: TestClient, admin: dict, make_role) -> None:
        role = make_role()
        resp = client.post(
            _ROLES,
            json={"role_code": role["role_code"], "role_name": "重复"},
            headers=admin["headers"],
        )
        assert resp.status_code == 400
        assert "已存在" in resp.json()["message"]

    def test_编码格式非法被拒(self, client: TestClient, admin: dict) -> None:
        resp = client.post(
            _ROLES, json={"role_code": "BadCode", "role_name": "大写"}, headers=admin["headers"]
        )
        assert resp.status_code == 400

    def test_编辑角色名称与排序(self, client: TestClient, admin: dict, make_role) -> None:
        role = make_role()
        resp = client.put(
            f"{_ROLES}/{role['id']}",
            json={"role_name": "改名后", "description": "新的说明", "sort_order": 7},
            headers=admin["headers"],
        )
        assert resp.status_code == 200
        data = resp.json()["data"]
        assert data["role_name"] == "改名后"
        assert data["description"] == "新的说明"
        assert data["sort_order"] == 7

    def test_编码不可被修改(self, client: TestClient, admin: dict, make_role) -> None:
        """请求体携带 role_code 会被忽略（编辑模型不接收该字段），编码保持不变。"""
        role = make_role()
        resp = client.put(
            f"{_ROLES}/{role['id']}",
            json={"role_name": "改名", "role_code": "hacked"},
            headers=admin["headers"],
        )
        assert resp.status_code == 200
        assert resp.json()["data"]["role_code"] == role["role_code"]

    def test_停用后不在下拉但仍在管理列表(self, client: TestClient, admin: dict, make_role) -> None:
        role = make_role()
        resp = client.put(
            f"{_ROLES}/{role['id']}/status", json={"status": 0}, headers=admin["headers"]
        )
        assert resp.status_code == 200
        assert resp.json()["data"]["status"] == 0

        resp = client.get("/api/v1/system/users/meta/roles", headers=admin["headers"])
        assert role["role_code"] not in [row["role_code"] for row in resp.json()["data"]]

        resp = client.get(_ROLES, params={"keyword": role["role_code"]}, headers=admin["headers"])
        assert resp.json()["data"]["total"] == 1


class TestInitialPermissions:
    def test_历史目录外权限不计入角色权限数(self, client, admin, make_role):
        from app.core.db import get_session_factory
        from app.models import Permission, RolePermission

        role = make_role()
        with get_session_factory()() as session:
            permission = Permission(perm_code=f"obsolete:{uuid.uuid4().hex}", perm_name="旧权限")
            session.add(permission)
            session.flush()
            permission_id = permission.id
            session.add(RolePermission(role_id=role["id"], permission_id=permission_id))
            session.commit()
        try:
            listing = client.get(_ROLES, headers=admin["headers"], params={
                "keyword": role["role_code"],
            }).json()["data"]["list"][0]
            detail = client.get(f"{_ROLES}/{role['id']}/permissions", headers=admin["headers"])
            assert detail.status_code == 200
            assert listing["permission_count"] == len(detail.json()["data"]["permission_codes"]) == 0
        finally:
            with get_session_factory()() as session:
                session.delete(session.get(Permission, permission_id))
                session.commit()

    def test_角色和权限一起创建且绑定后原令牌即时继承(self, client, admin, normal):
        import json

        from sqlalchemy import select

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

        codes = ["base:personnel:view", "base:personnel:regions"]
        code = _new_code()
        response = client.post(_ROLES, headers=admin["headers"], json={
            "role_code": code, "role_name": "区域维护", "permission_codes": codes,
        })
        assert response.status_code == 200, response.text
        role = response.json()["data"]
        assert role["permission_count"] == 2
        listed = client.get(_ROLES, headers=admin["headers"], params={"keyword": code}).json()["data"]
        assert listed["list"][0]["permission_count"] == 2
        updated = client.put(f"{_ROLES}/{role['id']}", headers=admin["headers"],
                             json={"role_name": "区域维护", "sort_order": 0})
        assert updated.status_code == 200 and updated.json()["data"]["permission_count"] == 2
        with get_session_factory()() as session:
            log = session.scalar(select(OperationLog).where(
                OperationLog.user_id == admin["id"], OperationLog.summary.contains(code),
                OperationLog.success == 1,
            ))
            assert log is not None
            request_data = json.loads(log.request_data)
            assert request_data["reason"] == "创建角色并配置初始权限"
            assert request_data["permission_codes"] == codes
        stored = client.get(f"{_ROLES}/{role['id']}/permissions", headers=admin["headers"]).json()["data"]
        assert set(stored["permission_codes"]) == set(codes)
        before = client.get(f"/api/v1/system/users/{normal['id']}/authorization", headers=admin["headers"]).json()["data"]
        response = client.put(f"/api/v1/system/users/{normal['id']}/authorization", headers=admin["headers"], json={
            "role_codes": [role["role_code"]], "expected_revision": before["revision"], "reason": "分配新角色",
        })
        assert response.status_code == 200, response.text
        info = client.get("/api/v1/auth/me", headers=normal["headers"]).json()["data"]
        assert set(info["permissions"]) == set(codes)

    @pytest.mark.parametrize("codes", [
        ["unknown:permission"], ["base:personnel:view", "base:personnel:view"],
        ["system:users:view"], ["base:personnel:regions"],
    ])
    def test_非法初始权限不创建角色(self, client, admin, codes):
        code = _new_code()
        response = client.post(_ROLES, headers=admin["headers"], json={
            "role_code": code, "role_name": "不得部分创建", "permission_codes": codes, "reason": "校验授权边界",
        })
        assert response.status_code == 400
        result = client.get(_ROLES, headers=admin["headers"], params={"keyword": code}).json()["data"]
        assert result["total"] == 0

    def test_权限落库失败时角色和版本一起回滚(self, client, admin):
        from sqlalchemy import event

        from app.models import RolePermission

        def fail_insert(*_):
            raise RuntimeError("隔离测试模拟初始授权落库失败")

        code = _new_code()
        before = client.get("/api/v1/auth/me", headers=admin["headers"]).json()["data"]["authorization_revision"]
        event.listen(RolePermission, "before_insert", fail_insert)
        try:
            with pytest.raises(RuntimeError, match="隔离测试模拟初始授权落库失败"):
                client.post(_ROLES, headers=admin["headers"], json={
                    "role_code": code, "role_name": "回滚验证", "permission_codes": ["base:personnel:view"], "reason": "验证事务",
                })
        finally:
            event.remove(RolePermission, "before_insert", fail_insert)
        result = client.get(_ROLES, headers=admin["headers"], params={"keyword": code}).json()["data"]
        assert result["total"] == 0
        after = client.get("/api/v1/auth/me", headers=admin["headers"]).json()["data"]["authorization_revision"]
        assert after == before


class TestProtection:
    """保护规则：内置角色与有绑定的角色。"""

    def test_admin角色不可删除(self, client: TestClient, admin: dict) -> None:
        role_id = _admin_role_id(client, admin)
        resp = client.delete(f"{_ROLES}/{role_id}", headers=admin["headers"])
        assert resp.status_code == 403
        assert "不可删除" in resp.json()["message"]

    def test_admin角色不可停用(self, client: TestClient, admin: dict) -> None:
        role_id = _admin_role_id(client, admin)
        resp = client.put(
            f"{_ROLES}/{role_id}/status", json={"status": 0}, headers=admin["headers"]
        )
        assert resp.status_code == 403
        assert "不可停用" in resp.json()["message"]

    def test_有绑定的角色不可删除且解绑后可删除(
        self, client: TestClient, admin: dict, make_role, make_user
    ) -> None:
        """`user_roles` 外键是 CASCADE —— 后端 400 拦截是防止静默剥离权限的
        唯一防线，前端按钮禁用只是体验。"""
        role = make_role()
        bound = make_user([role["role_code"]])

        resp = client.delete(f"{_ROLES}/{role['id']}", headers=admin["headers"])
        assert resp.status_code == 400
        assert "1 个用户" in resp.json()["message"]

        # 解绑（换回内置 user 角色）后可删除
        resp = client.put(
            f"/api/v1/system/users/{bound['id']}",
            json={"roles": [ROLE_USER]},
            headers=admin["headers"],
        )
        assert resp.status_code == 200
        assert client.delete(f"{_ROLES}/{role['id']}", headers=admin["headers"]).status_code == 200

    def test_列表返回绑定用户数(
        self, client: TestClient, admin: dict, make_role, make_user
    ) -> None:
        role = make_role()
        make_user([role["role_code"]])
        resp = client.get(_ROLES, params={"keyword": role["role_code"]}, headers=admin["headers"])
        assert resp.json()["data"]["list"][0]["user_count"] == 1

    def test_编辑不存在的角色404(self, client: TestClient, admin: dict) -> None:
        resp = client.put(
            f"{_ROLES}/999999999", json={"role_name": "不存在"}, headers=admin["headers"]
        )
        assert resp.status_code == 404

    def test_停用角色不可再分配给用户(
        self, client: TestClient, admin: dict, make_role, make_user
    ) -> None:
        """回归：停用只挡住了表单下拉，接口层此前仍可绑定（或使用停用前
        加载的旧选项提交）—— 与「停用后不可再分配」的语义矛盾。"""
        role = make_role()
        resp = client.put(
            f"{_ROLES}/{role['id']}/status", json={"status": 0}, headers=admin["headers"]
        )
        assert resp.status_code == 200

        username = f"t_{uuid.uuid4().hex[:8]}"
        resp = client.post(
            "/api/v1/system/users",
            json={"username": username, "password": "TestPwd@2026", "roles": [role["role_code"]]},
            headers=admin["headers"],
        )
        assert resp.status_code == 400
        assert "停用" in resp.json()["message"]

        # 已绑定该角色的用户（历史数据）改回启用角色后可正常保存
        bound = make_user([role["role_code"]])
        resp = client.put(
            f"/api/v1/system/users/{bound['id']}",
            json={"roles": [ROLE_USER]},
            headers=admin["headers"],
        )
        assert resp.status_code == 200


class TestAuthorization:
    """权限：角色管理为管理员专属。"""

    def test_普通用户访问被拒(self, client: TestClient, normal: dict) -> None:
        assert client.get(_ROLES, headers=normal["headers"]).status_code == 403
        resp = client.post(
            _ROLES,
            json={"role_code": _new_code(), "role_name": "越权"},
            headers=normal["headers"],
        )
        assert resp.status_code == 403