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

"""需求16 矩阵判定融合测试（authorization_service）。

覆盖：
- OWN/OTHER 部门两分类矩阵判定（OTHER 涵盖所属部门子树外全部部门）
- 矩阵与单文档 ACL 的 OR 叠加
- 密级前置与状态前置不可被矩阵绕过
- 无角色→默认角色→无默认则贡献为空
- 显式角色停用后矩阵贡献为空且不回退默认
- plan_allowed_actions 的矩阵操作位叠加（子方案过滤授权/挂载位）
"""

from __future__ import annotations

import pytest

from dms.common.enums import AllowedAction, RoleCode, SecurityLevel
from dms.common.errors import (
    DocumentActionForbiddenError,
    SecurityLevelForbiddenError,
)
from dms.extensions import db
from dms.models import Document, Organization, Role, RolePermission, User
from dms.security.auth_context import AuthContext
from dms.services.authorization_service import (
    evaluate_plan_access,
    has_any_manage_action,
    plan_allowed_actions,
    require_plan_action,
)


def _make_document(name: str, *, org: Organization, security: str = "SECRET",
                   visibility: str = "ORGANIZATION", status: str = "PUBLISHED",
                   document_type: str = "MAIN") -> Document:
    item = Document(
        document_name=name,
        summary=f"{name}摘要",
        document_type=document_type,
        document_status=status,
        security_level=security,
        visibility_type=visibility,
        organization_id=org.id,
        organization_name=org.org_name,
        tags=["矩阵测试"],
        original_file_name=f"{name}.docx",
        file_relative_path=f"original/MATRIX_{name}.docx",
        file_extension="docx",
        mime_type="application/vnd.openxmlformats-officedocument.wordprocessingml.document",
        file_size=128,
        file_hash="b" * 64,
        search_text=f"{name} 矩阵测试",
        created_by_name="系统管理员",
        updated_by_name="系统管理员",
    )
    db.session.add(item)
    db.session.flush()
    return item


def _make_role(name: str, *, is_default: bool = False,
               status: str = "ENABLED", actor_id: int | None = None) -> Role:
    role = Role(
        role_name=name,
        sort_no=10,
        status=status,
        is_default=is_default,
        created_by=actor_id,
        updated_by=actor_id,
    )
    db.session.add(role)
    db.session.flush()
    return role


def _matrix_row(role_id: int, level: str, scope: str, **flags) -> None:
    db.session.add(
        RolePermission(
            role_id=role_id,
            security_level=level,
            dept_scope=scope,
            **flags,
        )
    )


def _user_context(user: User) -> AuthContext:
    return AuthContext(
        user_id=user.id,
        username=user.username,
        real_name=user.real_name,
        organization_id=user.organization_id,
        organization_name=user.organization_name,
        role_code=RoleCode(user.role_code),
        security_level=SecurityLevel(user.security_level),
        auth_version=user.auth_version,
    )


@pytest.fixture()
def matrix_env(b2_app, role_cleanup):
    """在 b2_app 基础上构造矩阵判定数据：机构、用户、文档。"""
    with b2_app.app_context():
        ops = db.session.scalar(
            db.select(Organization).where(Organization.org_code == "ORG_OPS")
        )
        comms = db.session.scalar(
            db.select(Organization).where(Organization.org_code == "ORG_COMMS")
        )
        user = db.session.scalar(
            db.select(User).where(User.username == "user")
        )
        assert ops and comms and user
        db.session.commit()
        ids = {"ops": ops.id, "comms": comms.id, "user": user.id}
    return b2_app, ids


@pytest.mark.integration
def test_own_scope_matrix_grants_view_when_acl_denies(matrix_env):
    app, ids = matrix_env
    with app.app_context():
        ops = db.session.get(Organization, ids["ops"])
        user = db.session.get(User, ids["user"])
        role = _make_role("作战部角色", actor_id=user.id)
        _matrix_row(
            role.id, "SECRET", "OWN", can_view=True, can_download=True
        )
        # ORGANIZATION 可见性且无匹配 ACL 行：纯矩阵放行
        doc = _make_document("矩阵本部门主案", org=ops, visibility="ORGANIZATION")
        db.session.commit()

        context = _user_context(user)
        access = evaluate_plan_access(doc, context)
        assert access.allowed is True
        assert access.can_download is True
        assert "VIEW" in access.matrix_actions
        assert "DOWNLOAD" in access.matrix_actions


@pytest.mark.integration
def test_other_scope_without_matrix_denies(matrix_env):
    app, ids = matrix_env
    with app.app_context():
        comms = db.session.get(Organization, ids["comms"])
        user = db.session.get(User, ids["user"])
        # 文档挂通信部：对作战部用户是 OTHER，且未配置部门
        doc = _make_document("矩阵其他部门主案", org=comms, visibility="ORGANIZATION")
        db.session.commit()

        context = _user_context(user)
        access = evaluate_plan_access(doc, context)
        assert access.allowed is False
        assert access.reason == "ACL"


@pytest.mark.integration
def test_other_scope_matrix_grants_cross_department_view(matrix_env):
    app, ids = matrix_env
    with app.app_context():
        comms = db.session.get(Organization, ids["comms"])
        user = db.session.get(User, ids["user"])
        role = _make_role("跨部门角色", actor_id=user.id)
        _matrix_row(role.id, "SECRET", "OTHER", can_view=True)
        doc = _make_document("矩阵其他部门主案", org=comms, visibility="ORGANIZATION")
        db.session.commit()

        context = _user_context(user)
        access = evaluate_plan_access(doc, context)
        assert access.allowed is True
        assert "VIEW" in access.matrix_actions


@pytest.mark.integration
def test_security_and_status_precedence_over_matrix(matrix_env):
    app, ids = matrix_env
    with app.app_context():
        ops = db.session.get(Organization, ids["ops"])
        user = db.session.get(User, ids["user"])  # 密级 SECRET
        role = _make_role("越权角色", actor_id=user.id)
        _matrix_row(role.id, "TOP_SECRET", "OWN", can_view=True)
        top_secret_doc = _make_document(
            "矩阵绝密主案", org=ops, security="TOP_SECRET"
        )
        draft_doc = _make_document(
            "矩阵草稿主案", org=ops, security="SECRET", status="DRAFT"
        )
        db.session.commit()

        context = _user_context(user)
        assert evaluate_plan_access(top_secret_doc, context).reason == "SECURITY"
        assert evaluate_plan_access(draft_doc, context).reason == "STATUS"


@pytest.mark.integration
def test_default_role_applies_and_disabled_role_yields_nothing(matrix_env):
    app, ids = matrix_env
    with app.app_context():
        ops = db.session.get(Organization, ids["ops"])
        user = db.session.get(User, ids["user"])  # 未挂角色
        default_role = _make_role("默认角色", is_default=True, actor_id=user.id)
        _matrix_row(default_role.id, "SECRET", "OWN", can_view=True)
        doc = _make_document("默认角色主案", org=ops, visibility="ORGANIZATION")
        db.session.commit()

        context = _user_context(user)
        assert evaluate_plan_access(doc, context).allowed is True

        # 默认角色停用：无显式角色且无默认 → 矩阵贡献为空
        default_role.status = "DISABLED"
        db.session.commit()
        denied = evaluate_plan_access(doc, context)
        assert denied.allowed is False

        # 显式挂停用角色：不回退默认角色
        disabled_role = _make_role(
            "停用角色", status="DISABLED", actor_id=user.id
        )
        _matrix_row(disabled_role.id, "SECRET", "OWN", can_view=True)
        user.role_id = disabled_role.id
        db.session.commit()
        assert evaluate_plan_access(doc, context).allowed is False


@pytest.mark.integration
def test_matrix_actions_expand_allowed_actions(matrix_env):
    app, ids = matrix_env
    with app.app_context():
        ops = db.session.get(Organization, ids["ops"])
        user = db.session.get(User, ids["user"])
        role = _make_role("完整权限角色", actor_id=user.id)
        _matrix_row(
            role.id,
            "SECRET",
            "OWN",
            can_view=True,
            can_download=True,
            can_edit=True,
            can_delete=True,
            can_config_permission=True,
            can_bind_attachment=True,
            can_unbind_attachment=True,
        )
        main_doc = _make_document("矩阵操作位主案", org=ops)
        sub_doc = _make_document(
            "矩阵操作位子案",
            org=ops,
            document_type="SUB_PLAN",
        )
        sub_doc.root_document_id = main_doc.id
        sub_doc.parent_document_id = main_doc.id
        db.session.commit()

        context = _user_context(user)
        main_actions = plan_allowed_actions(
            main_doc, context, evaluate_plan_access(main_doc, context)
        )
        assert set(main_actions) >= {
            "VIEW",
            "DOWNLOAD",
            "EDIT",
            "DELETE",
            "CONFIG_PERMISSION",
        }
        sub_actions = plan_allowed_actions(
            sub_doc, context, evaluate_plan_access(sub_doc, context)
        )
        # 子方案过滤授权/挂载/解除挂载
        assert "CONFIG_PERMISSION" not in sub_actions
        assert "BIND_ATTACHMENT" not in sub_actions
        assert "UNBIND_ATTACHMENT" not in sub_actions
        assert {"VIEW", "DOWNLOAD", "EDIT", "DELETE"} <= set(sub_actions)


# ===== 动态管理菜单（allowedModules）判定 =====================================

from dms.services.user_service import allowed_modules_for_user


@pytest.mark.integration
def test_admin_role_always_has_manage_capability(matrix_env):
    app, ids = matrix_env
    with app.app_context():
        admin = db.session.scalar(
            db.select(User).where(User.username == "admin")
        )
        context = _user_context(admin)
        assert has_any_manage_action(context) is True
        assert allowed_modules_for_user(admin) == [
            "DOCUMENT_BROWSER",
            "BACKEND_MANAGEMENT",
            "AUDIT_LOG",
            "ORGANIZATION_USER_MANAGEMENT",
        ]


@pytest.mark.integration
def test_user_without_manage_bits_gets_browser_only(matrix_env):
    app, ids = matrix_env
    with app.app_context():
        ops = db.session.get(Organization, ids["ops"])
        user = db.session.get(User, ids["user"])
        role = _make_role("只读角色", is_default=True, actor_id=user.id)
        # 矩阵只有查看/下载位，无任何管理位
        for scope in ("OWN", "OTHER"):
            _matrix_row(
                role.id, "SECRET", scope, can_view=True, can_download=True
            )
        _make_document("只读可看主案", org=ops, visibility="ALL_AUTHENTICATED")
        db.session.commit()

        assert has_any_manage_action(_user_context(user)) is False
        assert allowed_modules_for_user(user) == ["DOCUMENT_BROWSER"]


@pytest.mark.integration
def test_user_with_manage_bit_and_matching_document_gets_admin_menu(matrix_env):
    app, ids = matrix_env
    with app.app_context():
        ops = db.session.get(Organization, ids["ops"])
        user = db.session.get(User, ids["user"])
        role = _make_role("编辑角色", is_default=True, actor_id=user.id)
        # OWN × SECRET 勾选编辑：作战部存在已发布 SECRET 主案即命中
        _matrix_row(
            role.id, "SECRET", "OWN", can_view=True, can_edit=True
        )
        _make_document("本部门可编辑主案", org=ops)
        db.session.commit()

        assert has_any_manage_action(_user_context(user)) is True
        assert allowed_modules_for_user(user) == [
            "DOCUMENT_BROWSER",
            "BACKEND_MANAGEMENT",
        ]


@pytest.mark.integration
def test_manage_bit_without_matching_document_or_view_denies(matrix_env):
    app, ids = matrix_env
    with app.app_context():
        ops = db.session.get(Organization, ids["ops"])
        user = db.session.get(User, ids["user"])
        role = _make_role("空转编辑角色", is_default=True, actor_id=user.id)
        # 只有 OTHER × TOP_SECRET 勾编辑，但文档在本部门且密级只有 SECRET：
        # 部门范围不命中 → 不出管理菜单
        _matrix_row(
            role.id, "TOP_SECRET", "OTHER", can_view=True, can_edit=True
        )
        _matrix_row(role.id, "SECRET", "OWN", can_view=True, can_edit=True)
        _make_document("本部门SECRET主案", org=ops)
        db.session.commit()

        # 修正：OWN×SECRET 也勾了编辑且文档命中，应当可见 —— 先验证 OTHER 行不误伤
        assert has_any_manage_action(_user_context(user)) is True

        # 去掉 OWN 行的编辑位后只剩 OTHER 行：不再命中
        row = db.session.scalar(
            db.select(RolePermission).where(
                RolePermission.role_id == role.id,
                RolePermission.dept_scope == "OWN",
            )
        )
        row.can_edit = False
        db.session.commit()
        assert has_any_manage_action(_user_context(user)) is False


@pytest.mark.integration
def test_manage_bit_without_view_source_denies(matrix_env):
    app, ids = matrix_env
    with app.app_context():
        ops = db.session.get(Organization, ids["ops"])
        user = db.session.get(User, ids["user"])
        role = _make_role("无查看编辑角色", is_default=True, actor_id=user.id)
        # 编辑位开启但查看位全关且文档非全员可见：access 不成立 → 管理位无效
        _matrix_row(role.id, "SECRET", "OWN", can_edit=True)
        _make_document("ORGANIZATION可见主案", org=ops, visibility="ORGANIZATION")
        db.session.commit()

        assert has_any_manage_action(_user_context(user)) is False


# ===== 写路径集中鉴权 require_plan_action ====================================

from dms.common.errors import (
    DocumentActionForbiddenError,
    SecurityLevelForbiddenError,
)


@pytest.mark.integration
def test_require_plan_action_allows_matrix_edit(matrix_env):
    app, ids = matrix_env
    with app.app_context():
        ops = db.session.get(Organization, ids["ops"])
        user = db.session.get(User, ids["user"])
        role = _make_role("写路径编辑角色", is_default=True, actor_id=user.id)
        _matrix_row(role.id, "SECRET", "OWN", can_view=True, can_edit=True)
        doc = _make_document("写路径可编辑主案", org=ops)
        db.session.commit()

        require_plan_action(doc, _user_context(user), AllowedAction.EDIT)


@pytest.mark.integration
def test_require_plan_action_denies_without_bit(matrix_env):
    app, ids = matrix_env
    with app.app_context():
        ops = db.session.get(Organization, ids["ops"])
        user = db.session.get(User, ids["user"])
        role = _make_role("写路径只读角色", is_default=True, actor_id=user.id)
        _matrix_row(role.id, "SECRET", "OWN", can_view=True, can_download=True)
        doc = _make_document("写路径只读主案", org=ops)
        db.session.commit()

        with pytest.raises(DocumentActionForbiddenError):
            require_plan_action(doc, _user_context(user), AllowedAction.EDIT)
        with pytest.raises(DocumentActionForbiddenError):
            require_plan_action(doc, _user_context(user), AllowedAction.DELETE)
        with pytest.raises(DocumentActionForbiddenError):
            require_plan_action(
                doc, _user_context(user), AllowedAction.CONFIG_PERMISSION
            )


@pytest.mark.integration
def test_require_plan_action_security_gate_precedes_bits(matrix_env):
    app, ids = matrix_env
    with app.app_context():
        ops = db.session.get(Organization, ids["ops"])
        user = db.session.get(User, ids["user"])
        role = _make_role("写路径高密角色", is_default=True, actor_id=user.id)
        _matrix_row(role.id, "TOP_SECRET", "OWN", can_view=True, can_edit=True)
        # 用户密级 SECRET：即使矩阵在 TOP_SECRET 行勾了编辑，密级前置直接拒绝
        doc = _make_document("绝密主案", org=ops, security="TOP_SECRET")
        db.session.commit()

        with pytest.raises(SecurityLevelForbiddenError):
            require_plan_action(doc, _user_context(user), AllowedAction.EDIT)


@pytest.mark.integration
def test_require_plan_action_admin_passes_all(matrix_env):
    app, ids = matrix_env
    with app.app_context():
        ops = db.session.get(Organization, ids["ops"])
        admin = db.session.scalar(
            db.select(User).where(User.username == "admin")
        )
        doc = _make_document("管理员主案", org=ops, security="SECRET")
        db.session.commit()

        for action in (
            AllowedAction.EDIT,
            AllowedAction.DELETE,
            AllowedAction.CONFIG_PERMISSION,
            AllowedAction.BIND_ATTACHMENT,
            AllowedAction.UNBIND_ATTACHMENT,
        ):
            require_plan_action(doc, _user_context(admin), action)


@pytest.mark.integration
def test_require_plan_action_attachment_admin_only(matrix_env):
    app, ids = matrix_env
    with app.app_context():
        ops = db.session.get(Organization, ids["ops"])
        user = db.session.get(User, ids["user"])
        admin = db.session.scalar(
            db.select(User).where(User.username == "admin")
        )
        role = _make_role("附件全权角色", is_default=True, actor_id=user.id)
        for scope in ("OWN", "OTHER"):
            _matrix_row(
                role.id, "PUBLIC", scope,
                can_view=True, can_download=True, can_edit=True, can_delete=True,
            )
        att = _make_document(
            "共享附件", org=ops, security="PUBLIC",
            visibility="ALL_AUTHENTICATED", document_type="ATTACHMENT",
        )
        db.session.commit()

        with pytest.raises(DocumentActionForbiddenError):
            require_plan_action(att, _user_context(user), AllowedAction.EDIT)
        require_plan_action(att, _user_context(admin), AllowedAction.DELETE)