新增电子试验主文件目录与文件归档

backend/tests/test_etmf_api.py

@@ -0,0 +1,67 @@
+from __future__ import annotations
+
+import uuid
+from types import SimpleNamespace
+
+import pytest
+from sqlalchemy.ext.asyncio import AsyncSession
+
+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.etmf import EtmfNode
+from app.services.etmf_service import calculate_node_status, list_etmf_tree
+
+
+def test_required_node_without_documents_is_missing() -> None:
+    node = SimpleNamespace(is_active=True, required=True)
+
+    assert calculate_node_status(node, []) == "MISSING"
+
+
+def test_node_with_document_without_effective_version_is_uploaded() -> None:
+    node = SimpleNamespace(is_active=True, required=True)
+    document = SimpleNamespace(current_effective_version_id=None)
+
+    assert calculate_node_status(node, [document]) == "UPLOADED"
+
+
+def test_node_with_current_effective_version_is_effective() -> None:
+    node = SimpleNamespace(is_active=True, required=True)
+    document = SimpleNamespace(current_effective_version_id=uuid.uuid4())
+
+    assert calculate_node_status(node, [document]) == "EFFECTIVE"
+
+
+@pytest.mark.asyncio
+async def test_etmf_tree_returns_hierarchy_and_status(db_session: AsyncSession) -> None:
+    study_id = uuid.uuid4()
+    parent = EtmfNode(
+        id=uuid.uuid4(),
+        study_id=study_id,
+        code="01",
+        name="项目管理",
+        scope_type=DocumentScopeType.GLOBAL,
+        required=False,
+        sort_order=1,
+    )
+    child = EtmfNode(
+        id=uuid.uuid4(),
+        study_id=study_id,
+        parent_id=parent.id,
+        code="01.01",
+        name="方案与版本",
+        scope_type=DocumentScopeType.GLOBAL,
+        required=True,
+        sort_order=1,
+    )
+    db_session.add_all([parent, child])
+    await db_session.commit()
+
+    tree = await list_etmf_tree(db_session, study_id=study_id, current_user=None)
+
+    assert len(tree) == 1
+    assert tree[0].code == "01"
+    assert tree[0].children[0].code == "01.01"
+    assert tree[0].children[0].status == "MISSING"

backend/tests/test_etmf_nodes.py

@@ -0,0 +1,108 @@
+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)