Cheng Zhou
20ce6bccef
后端： - 删除 StudyRolePermission 模型和 project_permissions API 文件 - 重写 project_permissions.py core，移除所有模块级权限函数 - 重写 permission_cache.py，移除模块级权限缓存逻辑 - 新增权限模板功能（PermissionTemplate 模型、API、服务层） - 新增 permission_templates 数据库迁移 - 迁移 8 个模块（attachments、audit_logs、dashboard、faqs、 faq_categories、fees_attachments、knowledge_notes、 material_equipments、overview、subject_histories、subject_pds） 至接口级权限检查 - 删除所有模块级权限相关测试文件，新增权限模板测试 前端： - 删除 ProjectPermissions.vue 和 ProjectPermissionsModule.vue - 重写 projectRoutePermissions.ts，改为基于接口级权限格式 - 更新 store/study.ts、router/index.ts、AuditLogs.vue、Projects.vue 中的权限 API 调用，从 fetchProjectRolePermissions 改为 fetchApiEndpointPermissions - 清理 types/api.ts 中的旧模块级权限类型定义 - 新增 PermissionTemplateSelector.vue 组件 - 更新权限管理页面，移除模块级权限 tab Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
132 lines
3.9 KiB
Python
132 lines
3.9 KiB
Python
from __future__ import annotations
|
|
|
|
"""Pytest configuration and fixtures for tests."""
|
|
|
|
import asyncio
|
|
import uuid
|
|
from typing import AsyncGenerator
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
from sqlalchemy import text, event, String, TypeDecorator
|
|
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
|
|
from sqlalchemy.dialects.postgresql import UUID as PG_UUID
|
|
|
|
|
|
class GUID(TypeDecorator):
|
|
"""Platform-independent GUID type that uses CHAR(36) on SQLite."""
|
|
impl = String
|
|
cache_ok = True
|
|
|
|
def load_dialect_impl(self, dialect):
|
|
if dialect.name == 'sqlite':
|
|
return dialect.type_descriptor(String(36))
|
|
return dialect.type_descriptor(PG_UUID(as_uuid=True))
|
|
|
|
def process_bind_param(self, value, dialect):
|
|
if value is None:
|
|
return None
|
|
if dialect.name == 'sqlite':
|
|
if isinstance(value, uuid.UUID):
|
|
return str(value)
|
|
return value
|
|
return value
|
|
|
|
def process_result_value(self, value, dialect):
|
|
if value is None:
|
|
return None
|
|
if dialect.name == 'sqlite':
|
|
if isinstance(value, str):
|
|
return uuid.UUID(value)
|
|
return value
|
|
return value
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def event_loop():
|
|
"""Create an instance of the default event loop for the test session."""
|
|
loop = asyncio.new_event_loop()
|
|
yield loop
|
|
loop.close()
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def test_engine(event_loop):
|
|
"""Create a test database engine."""
|
|
# Use SQLite in-memory database for testing
|
|
engine = event_loop.run_until_complete(
|
|
_create_test_engine()
|
|
)
|
|
yield engine
|
|
event_loop.run_until_complete(engine.dispose())
|
|
|
|
|
|
async def _create_test_engine():
|
|
"""Helper to create and initialize test engine."""
|
|
engine = create_async_engine(
|
|
"sqlite+aiosqlite:///:memory:",
|
|
future=True,
|
|
echo=False,
|
|
connect_args={"check_same_thread": False},
|
|
)
|
|
|
|
# Import all models to register them with Base metadata
|
|
from app.db.base_class import Base
|
|
import app.models.study
|
|
import app.models.api_endpoint_permission
|
|
import app.models.api_endpoint_registry
|
|
import app.models.user
|
|
import app.models.study_member
|
|
import app.models.subject
|
|
import app.models.visit
|
|
import app.models.ae
|
|
import app.models.monitoring_visit_issue
|
|
import app.models.site
|
|
import app.models.permission_template
|
|
|
|
# Replace PostgreSQL UUID type with custom GUID type for SQLite
|
|
for table in Base.metadata.tables.values():
|
|
for column in table.columns:
|
|
if isinstance(column.type, PG_UUID):
|
|
column.type = GUID()
|
|
|
|
# Create all tables
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
|
|
return engine
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def db_session(test_engine) -> AsyncGenerator[AsyncSession, None]:
|
|
"""Create a test database session."""
|
|
async_session = async_sessionmaker(
|
|
bind=test_engine,
|
|
class_=AsyncSession,
|
|
expire_on_commit=False,
|
|
)
|
|
|
|
async with async_session() as session:
|
|
# Create a test study for foreign key references with unique code
|
|
study_code = f"TEST-STUDY-{uuid.uuid4().hex[:8]}"
|
|
await session.execute(
|
|
text("""
|
|
INSERT INTO studies (id, code, name, status, is_locked, visit_schedule)
|
|
VALUES (:id, :code, :name, :status, :is_locked, :visit_schedule)
|
|
"""),
|
|
{
|
|
"id": str(uuid.uuid4()),
|
|
"code": study_code,
|
|
"name": "Test Study",
|
|
"status": "ACTIVE",
|
|
"is_locked": False,
|
|
"visit_schedule": "[]",
|
|
}
|
|
)
|
|
await session.commit()
|
|
|
|
yield session
|
|
|
|
# Clean up after test
|
|
await session.rollback()
|