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

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

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

from __future__ import annotations

import pytest

from dms.common.enums import RoleCode, SecurityLevel
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,
    plan_allowed_actions,
)


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_assigned_scope_via_role_departments(matrix_env):
    from dms.models import RoleDepartment

    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", "ASSIGNED", can_view=True)
        db.session.add(
            RoleDepartment(
                role_id=role.id,
                organization_id=comms.id,
            )
        )
        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)