110 lines
3.3 KiB
Python
110 lines
3.3 KiB
Python
import unittest
|
||
from types import SimpleNamespace
|
||
|
||
from fastapi import HTTPException
|
||
|
||
from app.services.project_service import (
|
||
normalize_project_role,
|
||
require_project_read_access,
|
||
require_project_roles,
|
||
require_project_write_access,
|
||
)
|
||
|
||
|
||
class _ScalarResult:
|
||
def __init__(self, value):
|
||
self.value = value
|
||
|
||
def scalar_one_or_none(self):
|
||
return self.value
|
||
|
||
|
||
class _RecordingDB:
|
||
def __init__(self, values):
|
||
self.values = iter(values)
|
||
|
||
async def execute(self, statement):
|
||
return _ScalarResult(next(self.values))
|
||
|
||
|
||
class ProjectPermissionsTest(unittest.IsolatedAsyncioTestCase):
|
||
def setUp(self):
|
||
self.project = SimpleNamespace(id=25, owner_id=1)
|
||
self.current_user = SimpleNamespace(id=2)
|
||
|
||
def test_normalize_project_role_handles_legacy_case(self):
|
||
self.assertEqual(normalize_project_role(" ADMIN "), "admin")
|
||
self.assertEqual(normalize_project_role("Editor"), "editor")
|
||
self.assertIsNone(normalize_project_role(None))
|
||
|
||
async def test_write_access_accepts_legacy_uppercase_admin(self):
|
||
db = _RecordingDB([
|
||
self.project,
|
||
SimpleNamespace(role="ADMIN"),
|
||
])
|
||
|
||
project, role = await require_project_write_access(db, 25, self.current_user)
|
||
|
||
self.assertEqual(project, self.project)
|
||
self.assertEqual(role, "admin")
|
||
|
||
async def test_write_access_accepts_legacy_uppercase_editor(self):
|
||
db = _RecordingDB([
|
||
self.project,
|
||
SimpleNamespace(role="EDITOR"),
|
||
])
|
||
|
||
project, role = await require_project_write_access(db, 25, self.current_user)
|
||
|
||
self.assertEqual(project, self.project)
|
||
self.assertEqual(role, "editor")
|
||
|
||
async def test_write_access_rejects_viewer(self):
|
||
db = _RecordingDB([
|
||
self.project,
|
||
SimpleNamespace(role="viewer"),
|
||
])
|
||
|
||
with self.assertRaises(HTTPException) as context:
|
||
await require_project_write_access(db, 25, self.current_user)
|
||
|
||
self.assertEqual(context.exception.status_code, 403)
|
||
|
||
async def test_write_access_rejects_legacy_uppercase_viewer(self):
|
||
"""历史数据里 role 可能是大写 VIEWER，写权限校验必须先归一化再判断。"""
|
||
db = _RecordingDB([
|
||
self.project,
|
||
SimpleNamespace(role="VIEWER"),
|
||
])
|
||
|
||
with self.assertRaises(HTTPException) as context:
|
||
await require_project_write_access(db, 25, self.current_user)
|
||
|
||
self.assertEqual(context.exception.status_code, 403)
|
||
|
||
async def test_read_access_returns_normalized_role(self):
|
||
db = _RecordingDB([
|
||
self.project,
|
||
SimpleNamespace(role="VIEWER"),
|
||
])
|
||
|
||
project, role = await require_project_read_access(db, 25, self.current_user)
|
||
|
||
self.assertEqual(role, "viewer")
|
||
|
||
async def test_admin_only_permission_rejects_editor(self):
|
||
db = _RecordingDB([
|
||
self.project,
|
||
SimpleNamespace(role="EDITOR"),
|
||
])
|
||
|
||
with self.assertRaises(HTTPException) as context:
|
||
await require_project_roles(
|
||
db,
|
||
25,
|
||
self.current_user,
|
||
allowed_roles=["admin"],
|
||
)
|
||
|
||
self.assertEqual(context.exception.status_code, 403)
|