PlantAI-dev/tests/unit/test_rnaseq_fastp_defaults.py

import asyncio
import json
import sys
import types
from dataclasses import dataclass
from pathlib import Path

import pytest
import yaml

ROOT = Path(__file__).resolve().parents[2]
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))

from apps.web import pipeline_backend  # noqa: E402
from apps.cli.run_rnaseq_qc import validate_input_files  # noqa: E402
from apps.web.pipeline_backend import JobInfo  # noqa: E402
from gfa_core.mas.tools.stage_tools import StageTool, StageToolExecutor  # noqa: E402


FASTP_BIN = "tools/fastp"


def _write_minimal_rnaseq_inputs(input_dir: Path) -> None:
    input_dir.mkdir(parents=True, exist_ok=True)
    (input_dir / "left.library.chunk.fq.gz").write_bytes(b"placeholder-r1")
    (input_dir / "right.library.chunk.fq.gz").write_bytes(b"placeholder-r2")
    sample_config = {
        "samples": {
            "S1": {
                "r1": "left.library.chunk.fq.gz",
                "r2": "right.library.chunk.fq.gz",
                "condition": "CK",
                "replicate": 1,
            }
        }
    }
    (input_dir / "samples.yaml").write_text(yaml.safe_dump(sample_config), encoding="utf-8")


def test_rnaseq_qc_backend_defaults_to_fastp_command(monkeypatch, tmp_path):
    captured = {}
    input_dir = tmp_path / "input" / "job_fastp"
    workspace_dir = tmp_path / "workspace" / "job_fastp"
    logs_dir = workspace_dir / "logs"
    final_dir = workspace_dir / "final"
    _write_minimal_rnaseq_inputs(input_dir)
    logs_dir.mkdir(parents=True, exist_ok=True)
    final_dir.mkdir(parents=True, exist_ok=True)

    def fake_run_cmd_logged(cmd, log_file, cwd=None):
        captured["cmd"] = list(cmd)
        captured["log_file"] = str(log_file)
        for name in (
            "Test.rnaseq_qc.S1_1.clean.fastq.gz",
            "Test.rnaseq_qc.S1_2.clean.fastq.gz",
            "Test.rnaseq_qc.S1_fastp.html",
            "Test.rnaseq_qc.S1_fastp.json",
            "Test.rnaseq_qc.S1_fastp.log",
        ):
            (final_dir / name).write_text("ok", encoding="utf-8")
        (final_dir / "Test.rnaseq_qc.manifest.json").write_text(
            json.dumps(
                {
                    "stage": "rnaseq_qc",
                    "status": "success",
                    "prefix": "Test",
                    "metrics": {"total_samples": 1, "samples": {"S1": {}}},
                    "parameters": {"qc_tool": "fastp", "fastp_bin": FASTP_BIN},
                    "artifacts": {},
                }
            ),
            encoding="utf-8",
        )
        return 0

    monkeypatch.setattr(pipeline_backend, "can_start_new_task", lambda: (True, "ok"))
    monkeypatch.setattr(pipeline_backend, "run_cmd_logged", fake_run_cmd_logged)

    message, manifest = pipeline_backend.run_stage(
        stage="rnaseq_qc",
        job=JobInfo(
            job_id="job_fastp",
            input_dir=str(input_dir),
            workspace_dir=str(workspace_dir),
            logs_dir=str(logs_dir),
            created_at="2026-06-29 00:00:00",
        ),
        prefix="Test",
        cpu=8,
    )

    assert "RNA-seq QC completed" in message
    assert manifest == final_dir / "Test.rnaseq_qc.manifest.json"
    assert "--qc-tool" in captured["cmd"]
    assert captured["cmd"][captured["cmd"].index("--qc-tool") + 1] == "fastp"
    assert "--fastp-bin" in captured["cmd"]
    assert captured["cmd"][captured["cmd"].index("--fastp-bin") + 1] == FASTP_BIN
    assert "--threads-per-sample" in captured["cmd"]
    assert captured["cmd"][captured["cmd"].index("-c") + 1] == str(input_dir / "samples.yaml")


def test_rnaseq_qc_uses_sample_table_roles_not_filename_patterns(tmp_path):
    input_dir = tmp_path / "input"
    input_dir.mkdir()
    left = input_dir / "left.library.chunk.fq.gz"
    right = input_dir / "right.library.chunk.fq.gz"
    left.write_bytes(b"left-read")
    right.write_bytes(b"right-read")
    (input_dir / "decoy_R1.fastq.gz").write_bytes(b"decoy-r1")
    (input_dir / "decoy_R2.fastq.gz").write_bytes(b"decoy-r2")

    validated = validate_input_files(
        input_dir,
        {"samples": {"S1": {"r1": left.name, "r2": right.name}}},
    )

    assert validated["S1"]["r1"] == left
    assert validated["S1"]["r2"] == right


def test_stage_tool_executor_defaults_rnaseq_qc_to_fastp(monkeypatch, tmp_path):
    captured = {}

    @dataclass
    class FakeJobInfo:
        job_id: str
        input_dir: str
        workspace_dir: str
        logs_dir: str
        created_at: str

    def fake_run_stage(**kwargs):
        captured.update(kwargs)
        final_dir = Path(kwargs["job"].workspace_dir) / "final"
        final_dir.mkdir(parents=True, exist_ok=True)
        manifest = final_dir / f"{kwargs['prefix']}.{kwargs['stage']}.manifest.json"
        manifest.write_text(json.dumps({"status": "success", "stage": kwargs["stage"]}), encoding="utf-8")
        return "fake rnaseq qc completed", manifest

    fake_module = types.ModuleType("apps.web.pipeline_backend")
    fake_module.JobInfo = FakeJobInfo
    fake_module.run_stage = fake_run_stage
    monkeypatch.setitem(sys.modules, "apps.web.pipeline_backend", fake_module)

    executor = StageToolExecutor(project_root=tmp_path)
    stage_tool = StageTool(
        id="rnaseq_qc",
        name="RNA-seq QC",
        script="apps/cli/run_rnaseq_qc.py",
        category="rnaseq",
        description="fake",
    )

    result = asyncio.run(
        executor.execute(
            stage_tool=stage_tool,
            job_id="job_fastp_tool",
            prefix="Test",
            extra_params={"cpu": 8},
        )
    )

    assert result["status"] == "success"
    assert captured["rnaseq_qc_tool"] == "fastp"
    assert captured["rnaseq_fastp_bin"] == FASTP_BIN