DataOps-platform/tests/test_work_center_contract.py

from pathlib import Path

from app.core.system.permissions import (
    WORK_CENTER_MANAGE,
    WORK_CENTER_OPERATE,
    WORK_CENTER_READ,
    permission_for_request,
    permissions_for_roles,
)

ROOT = Path(__file__).resolve().parents[1]


def test_work_center_migration_is_versioned_audited_and_non_destructive():
    migration = (
        ROOT / "migrations/versions/20260802_430_unified_work_center.py"
    ).read_text(encoding="utf-8")
    assert 'revision = "20260802_430"' in migration
    assert 'down_revision = "20260801_420"' in migration
    for table in (
        "governance_workflows",
        "governance_workflow_versions",
        "governance_tasks",
        "governance_task_participants",
        "governance_task_reviews",
        "governance_task_comments",
        "governance_task_attachments",
        "governance_task_events",
        "governance_notification_templates",
        "governance_notification_preferences",
        "governance_notifications",
        "governance_notification_attempts",
    ):
        assert f"CREATE TABLE public.{table}" in migration
    assert "uq_governance_task_active_source" in migration
    assert "source_state_unchanged" in migration
    assert "raise RuntimeError" in migration
    assert "DROP TABLE" not in migration.upper()


def test_work_center_permissions_separate_read_operate_and_manage():
    assert WORK_CENTER_READ in permissions_for_roles(["viewer"])
    assert WORK_CENTER_OPERATE not in permissions_for_roles(["viewer"])
    assert WORK_CENTER_OPERATE in permissions_for_roles(["editor"])
    assert WORK_CENTER_MANAGE not in permissions_for_roles(["editor"])
    assert WORK_CENTER_MANAGE in permissions_for_roles(["admin"])
    assert permission_for_request("/api/system/work-center/tasks", "GET") == (
        WORK_CENTER_READ,
    )
    assert permission_for_request("/api/system/work-center/tasks/x/review", "POST") == (
        WORK_CENTER_OPERATE,
    )
    assert permission_for_request("/api/system/work-center/workflows", "POST") == (
        WORK_CENTER_MANAGE,
    )


def test_work_center_api_and_frontend_expose_the_complete_operator_surface():
    api = (ROOT / "app/api/system/work_center.py").read_text(encoding="utf-8")
    client = (ROOT / "frontend/src/api/workCenter.js").read_text(encoding="utf-8")
    view = (
        ROOT / "frontend/src/views/dataReview/workCenter.vue"
    ).read_text(encoding="utf-8")
    routes = (ROOT / "frontend/src/router/routes.js").read_text(encoding="utf-8")
    home = (ROOT / "frontend/src/views/home/index.vue").read_text(encoding="utf-8")

    for path in (
        "/work-center/workflows",
        "/work-center/tasks",
        "/comments",
        "/attachments",
        "/work-center/notifications",
        "/work-center/templates",
        "/work-center/preferences",
        "/work-center/dashboard",
    ):
        assert path in api
    for operation in (
        "listWorkCenterTasks",
        "getWorkCenterTask",
        "reviewWorkCenterTask",
        "transferWorkCenterTask",
        "closeWorkCenterTask",
        "reopenWorkCenterTask",
        "addWorkCenterComment",
        "addWorkCenterAttachment",
        "listWorkCenterNotifications",
        "getWorkCenterDashboard",
    ):
        assert operation in client
    assert "统一工作中心" in view
    assert "质量问题" in view
    assert "术语标准" in view
    assert "数据产品" in view
    assert "Agent 审批" in view
    assert "逾期待办" in view
    assert "/dataReview/work-center" in routes
    assert "getWorkCenterDashboard" in home