Cheng Zhou
109 lines
3.2 KiB
Python
109 lines
3.2 KiB
Python
from __future__ import annotations
|
|
|
|
import uuid
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
from sqlalchemy import select, text
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.db.base_class import Base
|
|
from app.models.document import Document, DocumentScopeType
|
|
from app.models.document_version import DocumentVersion # noqa: F401
|
|
from app.models.distribution import Distribution # noqa: F401
|
|
from app.models.acknowledgement import Acknowledgement # noqa: F401
|
|
from app.models.user import UserRole, UserStatus
|
|
|
|
|
|
class UserStub:
|
|
def __init__(self) -> None:
|
|
self.id = uuid.uuid4()
|
|
self.role = UserRole.ADMIN
|
|
self.status = UserStatus.ACTIVE
|
|
|
|
|
|
def test_etmf_node_table_is_registered() -> None:
|
|
import app.db.base # noqa: F401
|
|
|
|
assert "etmf_nodes" in Base.metadata.tables
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_document_create_persists_etmf_node_id(db_session: AsyncSession) -> None:
|
|
from app.models.etmf import EtmfNode
|
|
from app.schemas.document import DocumentCreate
|
|
from app.services import document_service
|
|
|
|
study_id = (
|
|
await db_session.execute(text("SELECT id FROM studies ORDER BY created_at DESC LIMIT 1"))
|
|
).scalar_one()
|
|
node = EtmfNode(
|
|
study_id=study_id,
|
|
code="01.01",
|
|
name="方案与版本",
|
|
scope_type=DocumentScopeType.GLOBAL,
|
|
required=True,
|
|
sort_order=1,
|
|
)
|
|
db_session.add(node)
|
|
await db_session.commit()
|
|
await db_session.refresh(node)
|
|
|
|
doc = await document_service.create_document(
|
|
db_session,
|
|
DocumentCreate(
|
|
trial_id=study_id,
|
|
doc_no="PROTOCOL-001",
|
|
doc_type="Protocol",
|
|
title="研究方案",
|
|
scope_type=DocumentScopeType.GLOBAL,
|
|
etmf_node_id=node.id,
|
|
),
|
|
UserStub(),
|
|
)
|
|
|
|
assert doc.etmf_node_id == node.id
|
|
stored = (
|
|
await db_session.execute(select(Document).where(Document.id == doc.id))
|
|
).scalar_one()
|
|
assert stored.etmf_node_id == node.id
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_site_scoped_etmf_node_requires_site_document(db_session: AsyncSession) -> None:
|
|
from app.models.etmf import EtmfNode
|
|
from app.schemas.document import DocumentCreate
|
|
from app.services import document_service
|
|
|
|
study_id = (
|
|
await db_session.execute(text("SELECT id FROM studies ORDER BY created_at DESC LIMIT 1"))
|
|
).scalar_one()
|
|
node = EtmfNode(
|
|
study_id=study_id,
|
|
code="03.01",
|
|
name="中心启动文件",
|
|
scope_type=DocumentScopeType.SITE,
|
|
required=True,
|
|
sort_order=1,
|
|
)
|
|
db_session.add(node)
|
|
await db_session.commit()
|
|
await db_session.refresh(node)
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await document_service.create_document(
|
|
db_session,
|
|
DocumentCreate(
|
|
trial_id=study_id,
|
|
doc_no="SITE-STARTUP-001",
|
|
doc_type="Startup",
|
|
title="中心启动文件",
|
|
scope_type=DocumentScopeType.GLOBAL,
|
|
etmf_node_id=node.id,
|
|
),
|
|
UserStub(),
|
|
)
|
|
|
|
assert exc_info.value.status_code == 422
|
|
assert "中心" in str(exc_info.value.detail)
|