"""需求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)