PlantAI-dev/tests/unit/test_reference_bundle_download_state.py

from __future__ import annotations

from datetime import datetime, timedelta
import json
from pathlib import Path


def test_reference_bundle_running_task_remains_running_after_elapsed_time(tmp_path, monkeypatch):
    from apps.web.vue3 import main

    monkeypatch.setattr(main, "WORKSPACE_ROOT", tmp_path / "workspace")
    task = {
        "id": "task_ref_long_running",
        "task_id": "task_ref_long_running",
        "stage": "reference_bundle_download",
        "job_id": "job_ref_long_running",
        "status": "running",
        "message": "正在后台下载参考物种注释包",
        "created_at": (datetime.now() - timedelta(seconds=500)).isoformat(),
        "start_time": (datetime.now() - timedelta(seconds=500)).isoformat(),
        "timeout_sec": 300,
        "payload": {"job_id": "job_ref_long_running", "timeout_sec": 300},
    }

    main._apply_reference_bundle_stale_watchdog(task)

    assert task["status"] == "running"
    assert task["pending_seconds"] >= 500
    assert task["progress"]["phase"] == "running"
    assert task["progress"]["indeterminate"] is True
    assert "action_required" not in task
    assert "retryable" not in task
    assert "result" not in task

    status_path = (
        tmp_path
        / "workspace"
        / "job_ref_long_running"
        / "final"
        / "reference_bundle_tasks"
        / "task_ref_long_running.json"
    )
    assert status_path.exists()
    persisted = json.loads(status_path.read_text(encoding="utf-8"))
    assert persisted["status"] == "running"
    assert persisted["progress"]["phase"] == "running"


def test_reference_bundle_success_clears_stale_fallback_fields(tmp_path, monkeypatch):
    from apps.web.vue3 import main

    monkeypatch.setattr(main, "WORKSPACE_ROOT", tmp_path / "workspace")
    monkeypatch.setattr(main, "INPUT_ROOT", tmp_path / "input")
    task_id = "task_ref_success_cleanup"
    main.tasks[task_id] = {
        "id": task_id,
        "task_id": task_id,
        "stage": "reference_bundle_download",
        "job_id": "job_ref_success_cleanup",
        "status": "failed",
        "message": "old timeout",
        "action_required": True,
        "retryable": True,
        "recommended_action": "fallback_or_retry_reference_download",
        "result": {"success": False, "can_fallback_ath": True},
    }

    def fake_download_reference_annotation_bundle(**_kwargs):
        return {
            "status": "success",
            "species_code": "soe",
            "files": {"pep": "/tmp/soe.pep.fasta", "cds": "/tmp/soe.cds.fasta"},
            "missing_required": [],
        }

    monkeypatch.setattr(
        "apps.web.pipeline_backend.download_reference_annotation_bundle",
        fake_download_reference_annotation_bundle,
    )

    main.run_reference_bundle_download_task(
        task_id,
        {
            "job_id": "job_ref_success_cleanup",
            "species_code": "soe",
            "species_name": "Spinacia oleracea",
            "taxon_id": "3562",
            "requested_stages": ["ppi"],
            "include_recommended": False,
            "accession": "",
            "timeout_sec": 300,
        },
    )

    task = main.tasks[task_id]
    assert task["status"] == "success"
    assert task["result"]["success"] is True
    assert "action_required" not in task
    assert "retryable" not in task
    assert "recommended_action" not in task



def test_ppi_resolution_prefers_downloaded_near_reference_over_recent_at(tmp_path):
    from apps.web import pipeline_backend as backend

    input_dir = tmp_path / "input"
    final_dir = tmp_path / "workspace" / "final"
    logs_dir = tmp_path / "workspace" / "logs"
    soe_dir = input_dir / "soe"
    soe_dir.mkdir(parents=True)
    final_dir.mkdir(parents=True)
    logs_dir.mkdir(parents=True)
    (soe_dir / "soe.pep.fasta").write_text(">SOE1\nMAAA\n", encoding="utf-8")
    (soe_dir / "soe.reference_bundle_download.manifest.json").write_text(
        json.dumps({
            "stage": "reference_bundle_download",
            "status": "success",
            "species_code": "soe",
            "files": {"pep": str(soe_dir / "soe.pep.fasta")},
        }),
        encoding="utf-8",
    )
    (final_dir / "Ft.SPL.ortholog.manifest.json").write_text(
        json.dumps({
            "stage": "ortholog",
            "status": "success",
            "prefix": "Ft.SPL",
            "reference_species": ["At"],
            "stage_context": {"reference_species": ["At"]},
        }),
        encoding="utf-8",
    )

    job = backend.JobInfo(
        job_id="job_ref_prefer_downloaded",
        input_dir=str(input_dir),
        workspace_dir=str(tmp_path / "workspace"),
        logs_dir=str(logs_dir),
        created_at="",
    )

    resolution = backend.resolve_ppi_species_resolution(
        job=job,
        prefix="Ft.SPL",
        ppi_species="auto",
        target_species=["Ft"],
        reference_species=["At"],
        family="SPL",
        ignore_existing_ortholog=True,
        requested_stages=["ppi"],
    )

    assert resolution["reference_species"] == ""
    assert resolution["selected_string_species"] == ""
    assert resolution["status"] == "confirmation_required"
    assert resolution["candidates"][0]["selected_string_taxon_id"] == "3562"