DataOps-platform/tests/data_research/test_development_frontend_contract.py

from pathlib import Path


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


def test_data_development_frontend_api_covers_v60_job_contract():
    source = (ROOT / "frontend/src/api/dataDevelopment.js").read_text(
        encoding="utf-8"
    )

    assert "/development/v1/ingestion-jobs" in source
    assert "createIngestionJob" in source
    assert "getIngestionJobs" in source
    assert "getIngestionJob" in source
    assert "retryIngestionJob" in source
    assert "cancelIngestionJob" in source
    assert "executeIngestionJob" in source
    assert "getCatalogSnapshots" in source
    assert "getJobEvidence" in source
    assert "password" not in source.lower()
    assert "credentials" not in source.lower()


def test_database_ingestion_ui_uses_configured_sources_and_shows_diagnostics():
    ingestion = (
        ROOT / "frontend/src/views/dataGovernance/development/ingestion.vue"
    ).read_text(encoding="utf-8")
    tasks = (
        ROOT / "frontend/src/views/dataGovernance/development/tasks.vue"
    ).read_text(encoding="utf-8")

    assert "getDatasourceList" in ingestion
    assert "executeIngestionJob" in ingestion
    assert "force_rerun" in ingestion
    assert "选择已配置数据源" in ingestion
    assert "尝试次数" in tasks
    assert "失败阶段" in tasks
    assert "来源证据" in tasks
    assert "retryIngestionJob" in tasks
    assert "executeIngestionJob" in tasks