397 lines
13 KiB
Python
Executable File
397 lines
13 KiB
Python
Executable File
from pathlib import Path
|
|
import sys
|
|
from typing import Any
|
|
|
|
import pytest
|
|
import yaml
|
|
from fastapi import HTTPException
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
if str(ROOT) not in sys.path:
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
from apps.cli.run_samples_autofill import main as samples_autofill_main
|
|
from apps.cli.run_rnaseq_de import build_de_plan
|
|
from apps.web.vue3.main import _write_samples_draft_rows
|
|
|
|
|
|
def _seed_yaml(path: Path) -> None:
|
|
path.write_text(
|
|
yaml.safe_dump(
|
|
{
|
|
"samples": {},
|
|
"groups": {},
|
|
"compare": [],
|
|
"experiment": {},
|
|
"de": {"strategy": "auto"},
|
|
},
|
|
allow_unicode=True,
|
|
sort_keys=False,
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
|
|
def _row(sample: str, condition: str, time: Any, rep: int, role: str) -> dict[str, Any]:
|
|
return {
|
|
"run": sample,
|
|
"sample_id": sample,
|
|
"fastq_1": f"{sample}_R1.fq.gz",
|
|
"fastq_2": f"{sample}_R2.fq.gz",
|
|
"condition": condition,
|
|
"time": time,
|
|
"replicate": str(rep),
|
|
"library_strandness": "reverse",
|
|
"group_role": role,
|
|
"include": True,
|
|
}
|
|
|
|
|
|
def _write_case(tmp_path: Path, name: str, rows: list[dict[str, Any]], design: str) -> dict[str, Any]:
|
|
path = tmp_path / f"{name}.samples.yaml"
|
|
_seed_yaml(path)
|
|
_write_samples_draft_rows(path, rows, design=design)
|
|
return yaml.safe_load(path.read_text(encoding="utf-8"))
|
|
|
|
|
|
def test_time_series_requires_control_and_generates_baseline_comparisons(tmp_path):
|
|
rows = [
|
|
_row("T0_r1", "mock", "0h", 1, "control"),
|
|
_row("T0_r2", "mock", "0h", 2, "control"),
|
|
_row("T1_r1", "mock", "1h", 1, "compare"),
|
|
_row("T1_r2", "mock", "1h", 2, "compare"),
|
|
_row("T3_r1", "mock", "3h", 1, "compare"),
|
|
_row("T3_r2", "mock", "3h", 2, "compare"),
|
|
]
|
|
|
|
data = _write_case(tmp_path, "time_series", rows, "time_series")
|
|
|
|
assert data["experiment"]["control_group"] == "0h"
|
|
assert data["compare"] == [["1h", "0h"], ["3h", "0h"]]
|
|
assert data["de"]["pairwise_modes"] == ["baseline"]
|
|
|
|
|
|
def test_time_series_accepts_numeric_zero_time(tmp_path):
|
|
rows = [
|
|
_row("T0_r1", "mock", 0, 1, "control"),
|
|
_row("T0_r2", "mock", 0, 2, "control"),
|
|
_row("T1_r1", "mock", "1h", 1, "compare"),
|
|
_row("T1_r2", "mock", "1h", 2, "compare"),
|
|
]
|
|
|
|
data = _write_case(tmp_path, "time_series_numeric_zero", rows, "time_series")
|
|
|
|
assert data["experiment"]["design"] == "time_series"
|
|
assert data["experiment"]["control_group"] == "0"
|
|
assert data["groups"]["T0_r1"] == "0"
|
|
assert data["compare"] == [["1h", "0"]]
|
|
|
|
|
|
def test_manual_design_hint_is_ignored_for_time_series_table(tmp_path):
|
|
rows = [
|
|
_row("T0_r1", "mock", "0h", 1, "control"),
|
|
_row("T0_r2", "mock", "0h", 2, "control"),
|
|
_row("T1_r1", "mock", "1h", 1, "compare"),
|
|
_row("T1_r2", "mock", "1h", 2, "compare"),
|
|
_row("T3_r1", "mock", "3h", 1, "compare"),
|
|
_row("T3_r2", "mock", "3h", 2, "compare"),
|
|
]
|
|
|
|
data = _write_case(tmp_path, "manual_hint_time_series", rows, "manual")
|
|
|
|
assert data["experiment"]["design"] == "time_series"
|
|
assert data["compare"] == [["1h", "0h"], ["3h", "0h"]]
|
|
|
|
|
|
def test_two_group_without_control_is_saved_with_de_warning(tmp_path):
|
|
rows = [
|
|
_row("CK_r1", "CK", "", 1, "compare"),
|
|
_row("CK_r2", "CK", "", 2, "compare"),
|
|
_row("salt_r1", "salt", "", 1, "compare"),
|
|
_row("salt_r2", "salt", "", 2, "compare"),
|
|
]
|
|
|
|
data = _write_case(tmp_path, "two_group_no_control", rows, "two_group")
|
|
|
|
assert data["samples"]["CK_r1"]["group_role"] == "compare"
|
|
assert data["groups"] == {}
|
|
assert data["compare"] == []
|
|
assert data["de"] == {}
|
|
assert any("DE design pending user input" in item for item in data["_autofill"]["warnings"])
|
|
|
|
|
|
def test_samples_with_blank_group_role_are_saved_with_de_warning(tmp_path):
|
|
rows = [
|
|
_row("CK_r1", "CK", "", 1, ""),
|
|
_row("salt_r1", "salt", "", 1, ""),
|
|
]
|
|
|
|
data = _write_case(tmp_path, "blank_group_roles", rows, "manual")
|
|
|
|
assert data["samples"]["CK_r1"]["group_role"] == ""
|
|
assert data["samples"]["salt_r1"]["group_role"] == ""
|
|
assert data["groups"] == {}
|
|
assert data["compare"] == []
|
|
assert data["de"] == {}
|
|
assert any("DE design pending user input" in item for item in data["_autofill"]["warnings"])
|
|
|
|
|
|
def test_multi_group_without_control_is_saved_with_de_warning(tmp_path):
|
|
rows = [
|
|
_row("root_r1", "root", "", 1, "compare"),
|
|
_row("root_r2", "root", "", 2, "compare"),
|
|
_row("stem_r1", "stem", "", 1, "compare"),
|
|
_row("stem_r2", "stem", "", 2, "compare"),
|
|
_row("leaf_r1", "leaf", "", 1, "compare"),
|
|
_row("leaf_r2", "leaf", "", 2, "compare"),
|
|
]
|
|
|
|
data = _write_case(tmp_path, "multi_group_no_control", rows, "multi_group")
|
|
|
|
assert data["samples"]["leaf_r1"]["group_role"] == "compare"
|
|
assert data["groups"] == {}
|
|
assert data["compare"] == []
|
|
assert data["de"] == {}
|
|
assert any("DE design pending user input" in item for item in data["_autofill"]["warnings"])
|
|
|
|
|
|
def test_multi_group_with_control_generates_each_group_vs_control(tmp_path):
|
|
rows = [
|
|
_row("root_r1", "root", "", 1, "control"),
|
|
_row("root_r2", "root", "", 2, "control"),
|
|
_row("stem_r1", "stem", "", 1, "compare"),
|
|
_row("stem_r2", "stem", "", 2, "compare"),
|
|
_row("leaf_r1", "leaf", "", 1, "compare"),
|
|
_row("leaf_r2", "leaf", "", 2, "compare"),
|
|
]
|
|
|
|
data = _write_case(tmp_path, "multi_group_control", rows, "multi_group")
|
|
|
|
assert data["compare"] == [["leaf", "root"], ["stem", "root"]]
|
|
assert data["de"]["pairwise_modes"] == ["control"]
|
|
|
|
|
|
def test_manual_design_hint_is_ignored_for_group_role_table(tmp_path):
|
|
rows = [
|
|
_row("root_r1", "root", "", 1, "control"),
|
|
_row("stem_r1", "stem", "", 1, "compare"),
|
|
]
|
|
|
|
data = _write_case(tmp_path, "manual_hint", rows, "manual")
|
|
|
|
assert data["experiment"]["design"] == "two_group"
|
|
assert data["compare"] == [["stem", "root"]]
|
|
|
|
|
|
def test_confirmed_samples_draft_rows_drop_resolved_group_role_warning(tmp_path):
|
|
path = tmp_path / "confirmed_warning_cleanup.samples.yaml"
|
|
_seed_yaml(path)
|
|
data = yaml.safe_load(path.read_text(encoding="utf-8"))
|
|
data["_autofill"] = {
|
|
"warnings": [
|
|
"DE design pending user input: samples.yaml 缺少 group_role；请在样本表中为这些样本标记 control/compare/ignore: root_r1",
|
|
"Sample info table contains 1 run ID not found in current FASTQ path",
|
|
]
|
|
}
|
|
path.write_text(yaml.safe_dump(data, allow_unicode=True, sort_keys=False), encoding="utf-8")
|
|
rows = [
|
|
_row("root_r1", "root", "", 1, "control"),
|
|
_row("stem_r1", "stem", "", 1, "compare"),
|
|
]
|
|
|
|
_write_samples_draft_rows(path, rows, design="manual")
|
|
updated = yaml.safe_load(path.read_text(encoding="utf-8"))
|
|
|
|
assert updated["compare"] == [["stem", "root"]]
|
|
assert updated["_autofill"]["warnings"] == [
|
|
"Sample info table contains 1 run ID not found in current FASTQ path"
|
|
]
|
|
|
|
|
|
def test_samples_autofill_confirm_cleans_resolved_warning_but_keeps_metadata_warning(tmp_path, monkeypatch):
|
|
input_dir = tmp_path / "input"
|
|
workspace = tmp_path / "workspace"
|
|
final_dir = workspace / "final"
|
|
final_dir.mkdir(parents=True)
|
|
input_dir.mkdir()
|
|
draft_path = final_dir / "samples.autofill.draft.yaml"
|
|
rows = [
|
|
_row("root_r1", "root", "", 1, "control"),
|
|
_row("stem_r1", "stem", "", 1, "compare"),
|
|
]
|
|
payload = {
|
|
"samples": {
|
|
row["sample_id"]: {
|
|
"run": row["run"],
|
|
"r1": row["fastq_1"],
|
|
"r2": row["fastq_2"],
|
|
"condition": row["condition"],
|
|
"time": row["time"],
|
|
"replicate": row["replicate"],
|
|
"library_strandness": row["library_strandness"],
|
|
"group_role": row["group_role"],
|
|
"include": True,
|
|
}
|
|
for row in rows
|
|
},
|
|
"groups": {},
|
|
"experiment": {},
|
|
"compare": [],
|
|
"de": {},
|
|
"_autofill": {
|
|
"rows": rows,
|
|
"warnings": [
|
|
"DE design pending user input: samples.yaml 缺少 group_role；请在样本表中为这些样本标记 control/compare/ignore: root_r1",
|
|
"Sample info table contains 1 run ID not found in current FASTQ path",
|
|
],
|
|
},
|
|
}
|
|
draft_path.write_text(yaml.safe_dump(payload, allow_unicode=True, sort_keys=False), encoding="utf-8")
|
|
monkeypatch.setattr(
|
|
sys,
|
|
"argv",
|
|
[
|
|
"run_samples_autofill.py",
|
|
"--input-dir",
|
|
str(input_dir),
|
|
"--workspace",
|
|
str(workspace),
|
|
"--confirm",
|
|
"--draft",
|
|
str(draft_path),
|
|
],
|
|
)
|
|
|
|
assert samples_autofill_main() == 0
|
|
confirmed = yaml.safe_load((input_dir / "samples.yaml").read_text(encoding="utf-8"))
|
|
|
|
assert confirmed["compare"] == [["stem", "root"]]
|
|
assert confirmed["_autofill"]["warnings"] == [
|
|
"Sample info table contains 1 run ID not found in current FASTQ path"
|
|
]
|
|
|
|
|
|
def test_samples_autofill_confirm_allows_pending_de_roles(tmp_path, monkeypatch):
|
|
input_dir = tmp_path / "input"
|
|
workspace = tmp_path / "workspace"
|
|
final_dir = workspace / "final"
|
|
final_dir.mkdir(parents=True)
|
|
input_dir.mkdir()
|
|
draft_path = final_dir / "samples.autofill.draft.yaml"
|
|
rows = [
|
|
_row("CK_r1", "CK", "", 1, ""),
|
|
_row("salt_r1", "salt", "", 1, ""),
|
|
]
|
|
payload = {
|
|
"samples": {
|
|
row["sample_id"]: {
|
|
"run": row["run"],
|
|
"r1": row["fastq_1"],
|
|
"r2": row["fastq_2"],
|
|
"condition": row["condition"],
|
|
"time": row["time"],
|
|
"replicate": row["replicate"],
|
|
"library_strandness": row["library_strandness"],
|
|
"group_role": row["group_role"],
|
|
"include": True,
|
|
}
|
|
for row in rows
|
|
},
|
|
"groups": {},
|
|
"experiment": {},
|
|
"compare": [],
|
|
"de": {},
|
|
"_autofill": {"rows": rows, "warnings": []},
|
|
}
|
|
draft_path.write_text(yaml.safe_dump(payload, allow_unicode=True, sort_keys=False), encoding="utf-8")
|
|
monkeypatch.setattr(
|
|
sys,
|
|
"argv",
|
|
[
|
|
"run_samples_autofill.py",
|
|
"--input-dir",
|
|
str(input_dir),
|
|
"--workspace",
|
|
str(workspace),
|
|
"--confirm",
|
|
"--draft",
|
|
str(draft_path),
|
|
],
|
|
)
|
|
|
|
assert samples_autofill_main() == 0
|
|
confirmed = yaml.safe_load((input_dir / "samples.yaml").read_text(encoding="utf-8"))
|
|
|
|
assert confirmed["samples"]["CK_r1"]["group_role"] == ""
|
|
assert confirmed["groups"] == {}
|
|
assert confirmed["compare"] == []
|
|
assert confirmed["de"] == {}
|
|
assert any("DE design pending user input" in item for item in confirmed["_autofill"]["warnings"])
|
|
|
|
|
|
def test_build_de_plan_honors_configured_compare_and_auto_modes():
|
|
matrix_samples = ["root_r1", "root_r2", "stem_r1", "stem_r2", "leaf_r1", "leaf_r2"]
|
|
groups = {
|
|
"root_r1": "root",
|
|
"root_r2": "root",
|
|
"stem_r1": "stem",
|
|
"stem_r2": "stem",
|
|
"leaf_r1": "leaf",
|
|
"leaf_r2": "leaf",
|
|
}
|
|
samples = {
|
|
"root_r1": {"condition": "root", "group_role": "control"},
|
|
"root_r2": {"condition": "root", "group_role": "control"},
|
|
"stem_r1": {"condition": "stem", "group_role": "compare"},
|
|
"stem_r2": {"condition": "stem", "group_role": "compare"},
|
|
"leaf_r1": {"condition": "leaf", "group_role": "compare"},
|
|
"leaf_r2": {"condition": "leaf", "group_role": "compare"},
|
|
}
|
|
|
|
configured = build_de_plan(
|
|
{
|
|
"samples": samples,
|
|
"compare": [["leaf", "root"]],
|
|
"de": {"strategy": "auto", "design": "multi_group"},
|
|
"experiment": {"design": "multi_group"},
|
|
},
|
|
matrix_samples,
|
|
groups,
|
|
)
|
|
assert configured["comparisons"] == [["leaf", "root"]]
|
|
assert configured["comparison_mode"] == "configured"
|
|
|
|
no_control_samples = {
|
|
sample: {**meta, "group_role": "compare"}
|
|
for sample, meta in samples.items()
|
|
}
|
|
with pytest.raises(ValueError, match="唯一对照组"):
|
|
build_de_plan(
|
|
{"samples": no_control_samples, "de": {"strategy": "auto", "design": "multi_group"}},
|
|
matrix_samples,
|
|
groups,
|
|
)
|
|
|
|
control = build_de_plan(
|
|
{
|
|
"samples": samples,
|
|
"de": {"strategy": "auto", "design": "multi_group", "control_group": "root"},
|
|
"experiment": {"design": "multi_group", "control_group": "root"},
|
|
},
|
|
matrix_samples,
|
|
groups,
|
|
)
|
|
assert control["comparisons"] == [["leaf", "root"], ["stem", "root"]]
|
|
assert control["comparison_mode"] == "group_role_control"
|
|
|
|
missing_role_samples = {
|
|
sample: {"condition": meta["condition"]}
|
|
for sample, meta in samples.items()
|
|
}
|
|
with pytest.raises(ValueError, match="缺少 group_role"):
|
|
build_de_plan(
|
|
{"samples": missing_role_samples, "de": {"strategy": "auto"}},
|
|
matrix_samples,
|
|
groups,
|
|
)
|