109 lines
4.9 KiB
Python
109 lines
4.9 KiB
Python
"""细粒度权限矩阵单元测试：分组标签、required_permission 路由、旧角色自动升级。"""
|
|
import unittest
|
|
|
|
from app.auth import (KNOWN_PERMISSIONS, PERMISSION_LABELS, ROLE_PERMISSIONS,
|
|
_expand_legacy, has_permission, permissions_for_user,
|
|
required_permission)
|
|
from app.models import AppUser, RoleDefinition
|
|
|
|
|
|
class FakeDB:
|
|
"""只实现 permissions_for_user 需要的最小接口。"""
|
|
|
|
def __init__(self, row):
|
|
self._row = row
|
|
|
|
def exec(self, *_a, **_k):
|
|
class R:
|
|
def __init__(self, row):
|
|
self._row = row
|
|
|
|
def first(self):
|
|
return self._row
|
|
return R(self._row)
|
|
|
|
|
|
class PermissionMatrixTests(unittest.TestCase):
|
|
|
|
def test_permission_groups_complete(self):
|
|
"""16 项细分权限全部有中文标签，且不含旧粗权限名。"""
|
|
expected = {"read", "tasks:write", "tasks:review", "tasks:deliver",
|
|
"kb:read", "kb:write", "agents:write",
|
|
"memories:write", "memories:review",
|
|
"logs:read", "ops:manage",
|
|
"providers:write", "billing:manage", "users:manage",
|
|
"roles:manage", "audit:manage"}
|
|
self.assertEqual(set(KNOWN_PERMISSIONS), expected)
|
|
for item in KNOWN_PERMISSIONS:
|
|
self.assertIn(item, PERMISSION_LABELS)
|
|
self.assertNotIn("content:write", item)
|
|
|
|
def test_builtin_roles_use_fine_permissions(self):
|
|
"""内置角色直接携带细分权限，审查角色不能建任务。"""
|
|
editor = ROLE_PERMISSIONS["editor"]
|
|
self.assertIn("tasks:write", editor)
|
|
self.assertIn("kb:write", editor)
|
|
reviewer = ROLE_PERMISSIONS["reviewer"]
|
|
self.assertNotIn("tasks:write", reviewer)
|
|
self.assertIn("tasks:review", reviewer)
|
|
|
|
def test_required_permission_routing(self):
|
|
"""核心域路由：任务/审查/交付/知识库/Provider/运营/角色各归其位。"""
|
|
cases = [
|
|
("POST", "/api/campaigns", "tasks:write"),
|
|
("POST", "/api/campaigns/1/parse", "tasks:write"),
|
|
("POST", "/api/segments/5/review", "tasks:review"),
|
|
("GET", "/api/campaigns/1/export", "tasks:deliver"),
|
|
("GET", "/api/kbs", "kb:read"),
|
|
("POST", "/api/kbs/1/import", "kb:write"),
|
|
("GET", "/api/providers", "read"),
|
|
("POST", "/api/providers", "providers:write"),
|
|
("GET", "/api/ops/dashboard", "logs:read"),
|
|
("POST", "/api/ops/backup", "ops:manage"),
|
|
("POST", "/api/ops/dead-letter/1/requeue", "ops:manage"),
|
|
("GET", "/api/auth/users", "users:manage"),
|
|
("PUT", "/api/auth/roles/custom_x", "roles:manage"),
|
|
("POST", "/api/admin/points/grant", "audit:manage"),
|
|
("GET", "/api/usage/summary", "logs:read"),
|
|
("POST", "/api/usage/quota", "billing:manage"),
|
|
("GET", "/api/traces", "logs:read"),
|
|
("POST", "/api/memories", "memories:write"),
|
|
("POST", "/api/pending_memories/1/approve", "memories:review"),
|
|
("POST", "/api/agents", "agents:write"),
|
|
("POST", "/api/skills", "agents:write"),
|
|
]
|
|
for method, path, expected in cases:
|
|
self.assertEqual(required_permission(method, path), expected, f"{method} {path}")
|
|
|
|
def test_legacy_role_auto_expands(self):
|
|
"""旧自定义角色（content:write/review/admin:manage）自动展开为新权限集。"""
|
|
self.assertEqual(_expand_legacy({"read"}), {"read"})
|
|
self.assertEqual(_expand_legacy({"review"}), {"tasks:review", "memories:review"})
|
|
expanded = _expand_legacy({"content:write"})
|
|
self.assertIn("tasks:write", expanded)
|
|
self.assertIn("kb:write", expanded)
|
|
self.assertIn("agents:write", expanded)
|
|
admin_side = _expand_legacy({"admin:manage"})
|
|
self.assertIn("roles:manage", admin_side)
|
|
self.assertIn("billing:manage", admin_side)
|
|
self.assertEqual(_expand_legacy({"*"}), {"*"})
|
|
|
|
def test_legacy_role_still_has_old_capabilities(self):
|
|
"""升级不回退：旧「编辑」自定义角色仍能建任务和导知识库。"""
|
|
row = RoleDefinition(role="custom_old_editor", label="旧编辑",
|
|
permissions=["read", "content:write", "review", "audit:read"],
|
|
organization_id=1)
|
|
user = AppUser(username="legacy", display_name="旧编辑",
|
|
password_hash="x", role="custom_old_editor")
|
|
perms = permissions_for_user(FakeDB(row), user)
|
|
self.assertIn("tasks:write", perms)
|
|
self.assertIn("tasks:review", perms)
|
|
self.assertIn("kb:write", perms)
|
|
self.assertIn("logs:read", perms)
|
|
self.assertTrue(has_permission(user, "tasks:write", FakeDB(row)))
|
|
self.assertTrue(has_permission(user, "tasks:review", FakeDB(row)))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|