2132 lines
81 KiB
Python
2132 lines
81 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import importlib.util
|
|
import re
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
import gfa_core.rnaseq_report as rnaseq_report_module
|
|
from gfa_core.rnaseq_report import NoSuccessfulRNASeqEvidence, generate_rnaseq_report
|
|
from gfa_core.mas.utils import intent_parser as intent_parser_module
|
|
from gfa_core.mas.utils.intent_parser import parse_intent_from_query
|
|
|
|
|
|
def _fast_intent_recognition(query: str) -> dict | None:
|
|
module_path = Path(__file__).resolve().parents[2] / "gfa_core" / "mas" / "core" / "fast_intent.py"
|
|
spec = importlib.util.spec_from_file_location("rnaseq_report_fast_intent", module_path)
|
|
assert spec and spec.loader
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module.fast_intent_recognition(query)
|
|
|
|
|
|
def _write_manifest(
|
|
workspace: Path,
|
|
prefix: str,
|
|
stage: str,
|
|
status: str,
|
|
scope_id: str,
|
|
*,
|
|
suffix: str = "",
|
|
) -> Path:
|
|
final_dir = workspace / "final"
|
|
final_dir.mkdir(parents=True, exist_ok=True)
|
|
data_path = final_dir / f"{prefix}.{stage}{suffix}.table.tsv"
|
|
data_path.write_text("metric\tvalue\nreads\t100\n", encoding="utf-8")
|
|
manifest_path = final_dir / f"{prefix}.{stage}{suffix}.manifest.json"
|
|
manifest = {
|
|
"manifest_version": "1.0",
|
|
"stage": stage,
|
|
"status": status,
|
|
"prefix": prefix,
|
|
"rnaseq_scope_id": scope_id,
|
|
"metrics": {"sample_count": 2, "total_reads": 100},
|
|
"parameters": {"rnaseq_scope_id": scope_id},
|
|
"artifacts": {
|
|
"result_table": {
|
|
"artifact": "result_table",
|
|
"path": f"final/{data_path.name}",
|
|
"full_path": str(data_path),
|
|
"type": "table",
|
|
}
|
|
},
|
|
}
|
|
manifest_path.write_text(json.dumps(manifest, ensure_ascii=False), encoding="utf-8")
|
|
return manifest_path
|
|
|
|
|
|
def _write_artifact_manifest(
|
|
workspace: Path,
|
|
prefix: str,
|
|
stage: str,
|
|
scope_id: str,
|
|
artifacts: dict[str, tuple[str, str]],
|
|
) -> Path:
|
|
final_dir = workspace / "final"
|
|
final_dir.mkdir(parents=True, exist_ok=True)
|
|
manifest_artifacts = {}
|
|
for key, (filename, content) in artifacts.items():
|
|
artifact_path = final_dir / filename
|
|
artifact_path.write_text(content, encoding="utf-8")
|
|
manifest_artifacts[key] = {
|
|
"artifact": key,
|
|
"path": f"final/{filename}",
|
|
"full_path": str(artifact_path),
|
|
"type": artifact_path.suffix.lstrip("."),
|
|
}
|
|
manifest_path = final_dir / f"{prefix}.{stage}.manifest.json"
|
|
manifest_path.write_text(
|
|
json.dumps(
|
|
{
|
|
"manifest_version": "1.0",
|
|
"stage": stage,
|
|
"status": "success",
|
|
"prefix": prefix,
|
|
"rnaseq_scope_id": scope_id,
|
|
"parameters": {"rnaseq_scope_id": scope_id},
|
|
"metrics": {"comparison_count": 2},
|
|
"artifacts": manifest_artifacts,
|
|
},
|
|
ensure_ascii=False,
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
return manifest_path
|
|
|
|
|
|
def _figure_section(markdown: str, filename: str) -> str:
|
|
figure_area = markdown.index("## 4. 图表与产物说明")
|
|
image_at = markdown.index(filename, figure_area)
|
|
section_start = markdown.rfind("### 图 ", 0, image_at)
|
|
assert section_start >= 0, f"missing figure heading for {filename}"
|
|
next_figure = markdown.find("### 图 ", image_at)
|
|
section_end = next_figure if next_figure >= 0 else markdown.find("\n## 5.", image_at)
|
|
return markdown[section_start : section_end if section_end >= 0 else None]
|
|
|
|
|
|
def _embedded_image_names(markdown: str) -> list[str]:
|
|
return [Path(link).name for link in re.findall(r"!\[[^]]*]\(([^)]+)\)", markdown)]
|
|
|
|
|
|
def _write_enrichment_manifests(workspace: Path) -> None:
|
|
prefix = "Os"
|
|
scope = "scope-enrichment"
|
|
go_a = f"{prefix}.rnaseq_go_enrich.HH11__24_vs_HH11__0.GO_enrich"
|
|
go_b = f"{prefix}.rnaseq_go_enrich.IR29__6_vs_IR29__0.GO_enrich"
|
|
go_table_header = "ID\tDescription\tOntology\tGeneRatio\tBgRatio\tpvalue\tp.adjust\tCount\tgeneID\n"
|
|
_write_artifact_manifest(
|
|
workspace,
|
|
prefix,
|
|
"rnaseq_go_enrich",
|
|
scope,
|
|
{
|
|
"go_a_table": (f"{go_a}.txt", go_table_header + "GO:0009651\tsalt stress response\tBP\t4/20\t20/1000\t0.0001\t0.001\t4\tOs01/Os02/Os03/Os04\n"),
|
|
"go_b_table": (f"{go_b}.txt", go_table_header + "GO:0015979\tphotosynthesis\tBP\t3/20\t18/1000\t0.01\t0.04\t3\tOs05/Os06/Os07\n"),
|
|
"go_a_bar_png": (f"{go_a}.barplot.png", "png"),
|
|
"go_a_bar_pdf": (f"{go_a}.barplot.pdf", "pdf"),
|
|
"go_a_bar_svg": (f"{go_a}.barplot.svg", "<svg/>"),
|
|
"go_b_bar_png": (f"{go_b}.barplot.png", "png"),
|
|
"go_a_dot": (f"{go_a}.dotplot.png", "png"),
|
|
"go_b_dot": (f"{go_b}.dotplot.png", "png"),
|
|
"go_a_ontology": (f"{go_a}.ontology_barplot.png", "png"),
|
|
"go_b_ontology": (f"{go_b}.ontology_barplot.png", "png"),
|
|
},
|
|
)
|
|
|
|
kegg_a = f"{prefix}.rnaseq_kegg_enrich.IR29__6_vs_IR29__0"
|
|
kegg_b = f"{prefix}.rnaseq_kegg_enrich.HH11__48_vs_HH11__0"
|
|
kegg_header = "ID\tDescription\tGeneRatio\tBgRatio\tpvalue\tp.adjust\tCount\tgeneID\n"
|
|
_write_artifact_manifest(
|
|
workspace,
|
|
prefix,
|
|
"rnaseq_kegg_enrich",
|
|
scope,
|
|
{
|
|
"kegg_a_table": (
|
|
f"{kegg_a}.KEGG_enrich.txt",
|
|
kegg_header
|
|
+ "ko00073\tCutin, suberine and wax biosynthesis\t2/20\t10/1000\t0.0001\t0.001\t2\tOsWax1/OsWax2\n"
|
|
+ "ko00196\tPhotosynthesis - antenna proteins\t2/20\t12/1000\t0.0002\t0.002\t2\tOsLight1/OsLight2\n",
|
|
),
|
|
"kegg_b_table": (
|
|
f"{kegg_b}.KEGG_enrich.txt",
|
|
kegg_header
|
|
+ "ko01200\tCarbon metabolism\t3/20\t30/1000\t0.01\t0.04\t3\tOsC1/OsC2/OsC3\n"
|
|
+ "ko00073\tCutin, suberine and wax biosynthesis\t1/20\t10/1000\t0.02\t0.05\t1\tOsWaxH\n",
|
|
),
|
|
"kegg_a_hits": (
|
|
f"{kegg_a}.KEGG_pathway_gene_hits.txt",
|
|
"PathwayID\tPathwayName\tKO\tKOName\tGeneCount\tGenes\n"
|
|
"ko00073\tCutin, suberine and wax biosynthesis\tK00073\twax enzyme\t2\tOsWax1/OsWax2\n"
|
|
"ko00196\tPhotosynthesis - antenna proteins\tK00196\tantenna protein\t2\tOsLight1/OsLight2\n",
|
|
),
|
|
"kegg_a_fc": (
|
|
f"{kegg_a}.KEGG_pathway_map.ko_log2fc.tsv",
|
|
"KO\tlog2FoldChange\nK00073\t2.5\nK00196\t-1.5\n",
|
|
),
|
|
"kegg_a_bar_png": (f"{kegg_a}_KEGG_enrich.png", "png"),
|
|
"kegg_a_bar_pdf": (f"{kegg_a}_KEGG_enrich.pdf", "pdf"),
|
|
"kegg_b_bar": (f"{kegg_b}_KEGG_enrich.png", "png"),
|
|
"kegg_a_dot": (f"{kegg_a}_KEGG_dotplot.png", "png"),
|
|
"kegg_b_dot": (f"{kegg_b}_KEGG_dotplot.png", "png"),
|
|
"pathway_a_00073": (f"{kegg_a}.KEGG_pathway_map.ko00073.png", "png"),
|
|
"pathway_a_00196": (f"{kegg_a}.KEGG_pathway_map.ko00196.png", "png"),
|
|
"pathway_b_00073": (f"{kegg_b}.KEGG_pathway_map.ko00073.png", "png"),
|
|
},
|
|
)
|
|
|
|
fgsea_a = f"{prefix}.rnaseq_fgsea.IR29__6_vs_IR29__0"
|
|
fgsea_b = f"{prefix}.rnaseq_fgsea.HH11__48_vs_HH11__0"
|
|
fgsea_header = "pathway\tpathway_name\tpval\tpadj\tES\tNES\tsize\tleadingEdge\n"
|
|
_write_artifact_manifest(
|
|
workspace,
|
|
prefix,
|
|
"rnaseq_fgsea",
|
|
scope,
|
|
{
|
|
"fgsea_a_table": (
|
|
f"{fgsea_a}.fgsea_kegg.tsv",
|
|
fgsea_header
|
|
+ "ko00196\tPhotosynthesis - antenna proteins\t0.0001\t0.001\t0.7\t2.1\t18\tOsLight1,OsLight2\n",
|
|
),
|
|
"fgsea_b_table": (
|
|
f"{fgsea_b}.fgsea_kegg.tsv",
|
|
fgsea_header
|
|
+ "ko01200\tCarbon metabolism\t0.01\t0.04\t-0.4\t-1.2\t30\tOsC1,OsC2\n",
|
|
),
|
|
"fgsea_a_dot": (f"{fgsea_a}.fgsea_kegg_dotplot.png", "png"),
|
|
"fgsea_b_dot": (f"{fgsea_b}.fgsea_kegg_dotplot.png", "png"),
|
|
"fgsea_a_curve": (f"{fgsea_a}.fgsea_kegg_top_enrichment.png", "png"),
|
|
"fgsea_b_curve": (f"{fgsea_b}.fgsea_kegg_top_enrichment.png", "png"),
|
|
},
|
|
)
|
|
|
|
|
|
def _write_analysis_ready_family_manifest(
|
|
workspace: Path,
|
|
*,
|
|
prefix: str,
|
|
scope_id: str,
|
|
rows: list[tuple[str, str, str, str]],
|
|
) -> Path:
|
|
"""Write the real analysis-ready TSV shape used by rnaseq_family_score."""
|
|
header = "rank\tfamily_id\tfamily_name\tbest_comparison\tbest_support_labels\n"
|
|
body = "".join(
|
|
f"{rank}\t{family_id}\t{family_name}\t{comparison}\t{support_labels}\n"
|
|
for rank, (family_id, family_name, comparison, support_labels) in enumerate(rows, 1)
|
|
)
|
|
return _write_artifact_manifest(
|
|
workspace,
|
|
prefix,
|
|
"rnaseq_family_score",
|
|
scope_id,
|
|
{
|
|
"analysis_ready_family": (
|
|
f"{prefix}.rnaseq_family_score.analysis_ready_family.tsv",
|
|
header + body,
|
|
)
|
|
},
|
|
)
|
|
|
|
|
|
def _write_candidate_guided_enrichment_manifests(
|
|
workspace: Path,
|
|
*,
|
|
combined: bool = False,
|
|
include_fgsea_support: bool = True,
|
|
) -> dict[str, str]:
|
|
"""Create arbitrary identifiers so selection tests cannot rely on the sample job."""
|
|
prefix = "Zm"
|
|
scope = "scope-random-candidate-guided"
|
|
candidate_comparison = "LineX__17_vs_LineX__0"
|
|
unrelated_comparison = "LineY__29_vs_LineY__0"
|
|
supported_pathway_a = "ko54321"
|
|
supported_pathway_b = "ko67890"
|
|
unrelated_pathway = "ko13579"
|
|
go_id = "GO:9876543"
|
|
enrich_stage = "rnaseq_enrich" if combined else ""
|
|
go_stage = enrich_stage or "rnaseq_go_enrich"
|
|
kegg_stage = enrich_stage or "rnaseq_kegg_enrich"
|
|
|
|
go_candidate = f"{prefix}.{go_stage}.{candidate_comparison}.GO_enrich"
|
|
go_unrelated = f"{prefix}.{go_stage}.{unrelated_comparison}.GO_enrich"
|
|
go_header = "ID\tDescription\tOntology\tGeneRatio\tp.adjust\tCount\tgeneID\n"
|
|
go_artifacts = {
|
|
"candidate_go_table": (
|
|
f"{go_candidate}.txt",
|
|
go_header
|
|
+ f"{go_id}\trandom candidate function\tBP\t3/20\t0.02\t3\tGeneA/GeneB/GeneC\n",
|
|
),
|
|
"unrelated_go_table": (
|
|
f"{go_unrelated}.txt",
|
|
go_header
|
|
+ "GO:1111111\tunrelated stronger function\tBP\t8/20\t0.000001\t8\tOther1/Other2\n",
|
|
),
|
|
"candidate_go_bar": (f"{go_candidate}.barplot.png", "png"),
|
|
"candidate_go_dot": (f"{go_candidate}.dotplot.png", "png"),
|
|
"candidate_go_ontology": (f"{go_candidate}.ontology_barplot.png", "png"),
|
|
"unrelated_go_bar": (f"{go_unrelated}.barplot.png", "png"),
|
|
"unrelated_go_dot": (f"{go_unrelated}.dotplot.png", "png"),
|
|
"unrelated_go_ontology": (f"{go_unrelated}.ontology_barplot.png", "png"),
|
|
}
|
|
|
|
kegg_candidate = f"{prefix}.{kegg_stage}.{candidate_comparison}"
|
|
kegg_unrelated = f"{prefix}.{kegg_stage}.{unrelated_comparison}"
|
|
kegg_header = "ID\tDescription\tGeneRatio\tp.adjust\tCount\tgeneID\n"
|
|
kegg_artifacts = {
|
|
"candidate_kegg_table": (
|
|
f"{kegg_candidate}.KEGG_enrich.txt",
|
|
kegg_header
|
|
+ f"{supported_pathway_a}\trandom pathway alpha\t2/20\t0.02\t2\tGeneA/GeneB\n"
|
|
+ f"{supported_pathway_b}\trandom pathway beta\t2/20\t0.03\t2\tGeneB/GeneC\n",
|
|
),
|
|
"unrelated_kegg_table": (
|
|
f"{kegg_unrelated}.KEGG_enrich.txt",
|
|
kegg_header
|
|
+ f"{supported_pathway_a}\trandom pathway alpha\t7/20\t0.000001\t7\tOther1/Other2\n"
|
|
+ f"{unrelated_pathway}\tunrelated pathway\t9/20\t0.0000001\t9\tOther3/Other4\n",
|
|
),
|
|
"candidate_kegg_bar": (f"{kegg_candidate}_KEGG_enrich.png", "png"),
|
|
"candidate_kegg_dot": (f"{kegg_candidate}_KEGG_dotplot.png", "png"),
|
|
"unrelated_kegg_bar": (f"{kegg_unrelated}_KEGG_enrich.png", "png"),
|
|
"unrelated_kegg_dot": (f"{kegg_unrelated}_KEGG_dotplot.png", "png"),
|
|
"candidate_pathway_a": (
|
|
f"{kegg_candidate}.KEGG_pathway_map.{supported_pathway_a}.png",
|
|
"png",
|
|
),
|
|
"candidate_pathway_b": (
|
|
f"{kegg_candidate}.KEGG_pathway_map.{supported_pathway_b}.png",
|
|
"png",
|
|
),
|
|
"other_time_supported_pathway": (
|
|
f"{kegg_unrelated}.KEGG_pathway_map.{supported_pathway_a}.png",
|
|
"png",
|
|
),
|
|
"unrelated_pathway": (
|
|
f"{kegg_unrelated}.KEGG_pathway_map.{unrelated_pathway}.png",
|
|
"png",
|
|
),
|
|
}
|
|
|
|
if combined:
|
|
_write_artifact_manifest(
|
|
workspace,
|
|
prefix,
|
|
"rnaseq_enrich",
|
|
scope,
|
|
{**go_artifacts, **kegg_artifacts},
|
|
)
|
|
else:
|
|
_write_artifact_manifest(workspace, prefix, go_stage, scope, go_artifacts)
|
|
_write_artifact_manifest(workspace, prefix, kegg_stage, scope, kegg_artifacts)
|
|
|
|
fgsea_candidate = f"{prefix}.rnaseq_fgsea.{candidate_comparison}"
|
|
fgsea_unrelated = f"{prefix}.rnaseq_fgsea.{unrelated_comparison}"
|
|
fgsea_header = "pathway\tpathway_name\tpadj\tNES\tsize\tleadingEdge\n"
|
|
_write_artifact_manifest(
|
|
workspace,
|
|
prefix,
|
|
"rnaseq_fgsea",
|
|
scope,
|
|
{
|
|
"candidate_fgsea_table": (
|
|
f"{fgsea_candidate}.fgsea_kegg.tsv",
|
|
fgsea_header
|
|
+ f"{supported_pathway_b}\trandom pathway beta\t0.04\t2.1\t12\tGeneB,GeneC\n",
|
|
),
|
|
"unrelated_fgsea_table": (
|
|
f"{fgsea_unrelated}.fgsea_kegg.tsv",
|
|
fgsea_header
|
|
+ f"{unrelated_pathway}\tunrelated pathway\t0.0001\t3.0\t20\tOther1,Other2\n",
|
|
),
|
|
"candidate_fgsea_dot": (f"{fgsea_candidate}.fgsea_kegg_dotplot.png", "png"),
|
|
"candidate_fgsea_curve": (
|
|
f"{fgsea_candidate}.fgsea_kegg_top_enrichment.png",
|
|
"png",
|
|
),
|
|
"unrelated_fgsea_dot": (f"{fgsea_unrelated}.fgsea_kegg_dotplot.png", "png"),
|
|
"unrelated_fgsea_curve": (
|
|
f"{fgsea_unrelated}.fgsea_kegg_top_enrichment.png",
|
|
"png",
|
|
),
|
|
},
|
|
)
|
|
|
|
support = (
|
|
f"GO:{go_id}:random candidate function;"
|
|
f"KEGG:{supported_pathway_a}:random pathway alpha;"
|
|
f"KEGG:{supported_pathway_b}:random pathway beta"
|
|
)
|
|
if include_fgsea_support:
|
|
support += f";fgsea:{supported_pathway_b}:random pathway beta"
|
|
_write_analysis_ready_family_manifest(
|
|
workspace,
|
|
prefix=prefix,
|
|
scope_id=scope,
|
|
rows=[
|
|
("FAM_RANDOM_ALPHA", "random alpha family", candidate_comparison, support),
|
|
(
|
|
"FAM_RANDOM_BETA",
|
|
"random beta family",
|
|
candidate_comparison,
|
|
f"KEGG:{supported_pathway_a}:random pathway alpha",
|
|
),
|
|
],
|
|
)
|
|
return {
|
|
"prefix": prefix,
|
|
"scope": scope,
|
|
"candidate_comparison": candidate_comparison,
|
|
"unrelated_comparison": unrelated_comparison,
|
|
"supported_pathway_a": supported_pathway_a,
|
|
"supported_pathway_b": supported_pathway_b,
|
|
"unrelated_pathway": unrelated_pathway,
|
|
"go_stage": go_stage,
|
|
"kegg_stage": kegg_stage,
|
|
}
|
|
|
|
def test_report_is_deterministic_scope_bound_and_reusable(tmp_path: Path) -> None:
|
|
workspace = tmp_path / "job_scope"
|
|
_write_manifest(workspace, "Os", "rnaseq_qc", "success", "scope-alpha")
|
|
_write_manifest(workspace, "Os", "rnaseq_de", "success", "scope-beta")
|
|
_write_manifest(workspace, "Os", "rnaseq_report", "success", "scope-alpha", suffix=".old")
|
|
|
|
result = generate_rnaseq_report(
|
|
workspace,
|
|
"Os",
|
|
rnaseq_scope_id="scope-alpha",
|
|
expected_stages=["rnaseq_qc"],
|
|
use_model=False,
|
|
)
|
|
|
|
assert result.coverage_status == "complete"
|
|
assert result.included_stages == ("rnaseq_qc",)
|
|
assert result.rnaseq_scope_id == "scope-alpha"
|
|
markdown = result.report_path.read_text(encoding="utf-8")
|
|
assert "## 0. 报告摘要" in markdown
|
|
assert "## 7. 证据追溯" in markdown
|
|
assert "rnaseq_de" not in markdown
|
|
evidence = json.loads(result.evidence_path.read_text(encoding="utf-8"))
|
|
assert evidence["formal_report"]["source_fingerprint"]
|
|
assert {item["stage"] for item in evidence["formal_report"]["source_manifests"]} == {"rnaseq_qc"}
|
|
manifest = json.loads(result.manifest_path.read_text(encoding="utf-8"))
|
|
assert manifest["status"] == "success"
|
|
assert manifest["metrics"]["coverage_status"] == "complete"
|
|
assert manifest["metrics"]["model_status"] == "disabled"
|
|
|
|
reopened = generate_rnaseq_report(
|
|
workspace,
|
|
"Os",
|
|
rnaseq_scope_id="scope-alpha",
|
|
expected_stages=["rnaseq_qc"],
|
|
use_model=False,
|
|
action="open",
|
|
)
|
|
assert reopened.reused is True
|
|
|
|
|
|
def test_successful_report_can_have_partial_coverage(tmp_path: Path) -> None:
|
|
workspace = tmp_path / "job_partial"
|
|
_write_manifest(workspace, "Ft", "rnaseq_qc", "success", "scope-1")
|
|
_write_manifest(workspace, "Ft", "rnaseq_de", "failed", "scope-1")
|
|
|
|
result = generate_rnaseq_report(
|
|
workspace,
|
|
"Ft",
|
|
rnaseq_scope_id="scope-1",
|
|
expected_stages=["rnaseq_qc", "rnaseq_de"],
|
|
use_model=False,
|
|
)
|
|
|
|
manifest = json.loads(result.manifest_path.read_text(encoding="utf-8"))
|
|
assert manifest["status"] == "success"
|
|
assert manifest["metrics"]["coverage_status"] == "partial"
|
|
assert manifest["metrics"]["stage_statuses"]["rnaseq_de"] == "failed"
|
|
assert result.included_stages == ("rnaseq_qc",)
|
|
|
|
|
|
def test_report_relocates_stale_absolute_artifact_paths(tmp_path: Path) -> None:
|
|
workspace = tmp_path / "relocated_job"
|
|
manifest_path = _write_manifest(workspace, "Os", "rnaseq_qc", "success", "scope-moved")
|
|
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
|
|
manifest["artifacts"]["result_table"]["full_path"] = "/old/server/workspace/final/result.tsv"
|
|
manifest_path.write_text(json.dumps(manifest, ensure_ascii=False), encoding="utf-8")
|
|
|
|
result = generate_rnaseq_report(
|
|
workspace,
|
|
"Os",
|
|
rnaseq_scope_id="scope-moved",
|
|
expected_stages=["rnaseq_qc"],
|
|
use_model=False,
|
|
)
|
|
|
|
markdown = result.report_path.read_text(encoding="utf-8")
|
|
assert "指标 sample_count" in markdown
|
|
assert "manifest 登记但文件不存在" not in markdown
|
|
evidence = json.loads(result.evidence_path.read_text(encoding="utf-8"))
|
|
result_card = next(card for card in evidence["evidence_cards"] if card["claim"] == "产物 result_table")
|
|
assert "exists=True" in result_card["evidence"]
|
|
|
|
|
|
def test_no_successful_rnaseq_stage_does_not_create_empty_report(tmp_path: Path) -> None:
|
|
workspace = tmp_path / "job_empty"
|
|
_write_manifest(workspace, "At", "rnaseq_de", "failed", "scope-empty")
|
|
|
|
with pytest.raises(NoSuccessfulRNASeqEvidence, match="暂无可生成"):
|
|
generate_rnaseq_report(
|
|
workspace,
|
|
"At",
|
|
rnaseq_scope_id="scope-empty",
|
|
expected_stages=["rnaseq_de"],
|
|
use_model=False,
|
|
)
|
|
|
|
assert not (workspace / "final" / "At.rnaseq_report.md").exists()
|
|
|
|
|
|
def test_fallback_selects_one_of_each_enrichment_plot_and_deduplicates_pathway_ids(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
workspace = tmp_path / "job_enrichment_figures"
|
|
_write_enrichment_manifests(workspace)
|
|
|
|
result = generate_rnaseq_report(
|
|
workspace,
|
|
"Os",
|
|
rnaseq_scope_id="scope-enrichment",
|
|
expected_stages=["rnaseq_go_enrich", "rnaseq_kegg_enrich", "rnaseq_fgsea"],
|
|
use_model=False,
|
|
)
|
|
markdown = result.report_path.read_text(encoding="utf-8")
|
|
embedded = _embedded_image_names(markdown)
|
|
inventory = json.loads(result.evidence_path.read_text(encoding="utf-8"))["formal_report"][
|
|
"figure_inventory"
|
|
]
|
|
|
|
assert sum(name.endswith(".GO_enrich.barplot.png") for name in embedded) == 1
|
|
assert sum(name.endswith(".GO_enrich.dotplot.png") for name in embedded) == 1
|
|
assert sum(name.endswith(".GO_enrich.ontology_barplot.png") for name in embedded) == 1
|
|
assert sum(name.endswith("_KEGG_enrich.png") for name in embedded) == 1
|
|
assert sum(name.endswith("_KEGG_dotplot.png") for name in embedded) == 1
|
|
assert sum(name.endswith(".fgsea_kegg_dotplot.png") for name in embedded) == 1
|
|
assert sum(name.endswith(".fgsea_kegg_top_enrichment.png") for name in embedded) == 1
|
|
assert not any(name.endswith("GO_enrich.barplot.pdf") for name in embedded)
|
|
assert not any(name.endswith("GO_enrich.barplot.svg") for name in embedded)
|
|
assert not any(name.endswith("KEGG_enrich.pdf") for name in embedded)
|
|
|
|
required_pathway_maps = [
|
|
"Os.rnaseq_kegg_enrich.IR29__6_vs_IR29__0.KEGG_pathway_map.ko00073.png",
|
|
"Os.rnaseq_kegg_enrich.IR29__6_vs_IR29__0.KEGG_pathway_map.ko00196.png",
|
|
]
|
|
for filename in required_pathway_maps:
|
|
assert embedded.count(filename) == 1
|
|
assert not any(
|
|
name == "Os.rnaseq_kegg_enrich.HH11__48_vs_HH11__0.KEGG_pathway_map.ko00073.png"
|
|
for name in embedded
|
|
)
|
|
|
|
|
|
def test_each_selected_enrichment_figure_has_four_part_table_bound_interpretation(tmp_path: Path) -> None:
|
|
workspace = tmp_path / "job_enrichment_explanations"
|
|
_write_enrichment_manifests(workspace)
|
|
|
|
result = generate_rnaseq_report(
|
|
workspace,
|
|
"Os",
|
|
rnaseq_scope_id="scope-enrichment",
|
|
expected_stages=["rnaseq_go_enrich", "rnaseq_kegg_enrich", "rnaseq_fgsea"],
|
|
use_model=False,
|
|
)
|
|
markdown = result.report_path.read_text(encoding="utf-8")
|
|
expected_evidence = {
|
|
"Os.rnaseq_go_enrich.HH11__24_vs_HH11__0.GO_enrich.barplot.png": "salt stress response",
|
|
"Os.rnaseq_go_enrich.HH11__24_vs_HH11__0.GO_enrich.dotplot.png": "GO:0009651",
|
|
"Os.rnaseq_go_enrich.HH11__24_vs_HH11__0.GO_enrich.ontology_barplot.png": "BP",
|
|
"Os.rnaseq_kegg_enrich.IR29__6_vs_IR29__0_KEGG_enrich.png": "Cutin, suberine and wax biosynthesis",
|
|
"Os.rnaseq_kegg_enrich.IR29__6_vs_IR29__0_KEGG_dotplot.png": "ko00073",
|
|
"Os.rnaseq_kegg_enrich.IR29__6_vs_IR29__0.KEGG_pathway_map.ko00073.png": "OsWax1",
|
|
"Os.rnaseq_kegg_enrich.IR29__6_vs_IR29__0.KEGG_pathway_map.ko00196.png": "OsLight1",
|
|
"Os.rnaseq_fgsea.IR29__6_vs_IR29__0.fgsea_kegg_dotplot.png": "NES",
|
|
"Os.rnaseq_fgsea.IR29__6_vs_IR29__0.fgsea_kegg_top_enrichment.png": "leadingEdge",
|
|
}
|
|
for filename, evidence_value in expected_evidence.items():
|
|
section = _figure_section(markdown, filename)
|
|
assert "图形含义：" in section
|
|
assert "阅读方法：" in section
|
|
assert "本次结果：" in section
|
|
assert "解释边界：" in section
|
|
assert evidence_value in section
|
|
|
|
|
|
def test_combined_rnaseq_enrich_stage_keeps_all_plot_forms_and_deduplicates_pathway_ids(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
workspace = tmp_path / "job_combined_enrichment"
|
|
go_a = "Os.rnaseq_enrich.HH11__24_vs_HH11__0.GO_enrich"
|
|
go_b = "Os.rnaseq_enrich.IR29__6_vs_IR29__0.GO_enrich"
|
|
kegg_a = "Os.rnaseq_enrich.IR29__6_vs_IR29__0"
|
|
kegg_b = "Os.rnaseq_enrich.HH11__48_vs_HH11__0"
|
|
go_header = "ID\tDescription\tOntology\tGeneRatio\tp.adjust\tCount\tgeneID\n"
|
|
kegg_header = "ID\tDescription\tGeneRatio\tp.adjust\tCount\tgeneID\n"
|
|
_write_artifact_manifest(
|
|
workspace,
|
|
"Os",
|
|
"rnaseq_enrich",
|
|
"scope-combined-enrichment",
|
|
{
|
|
"go_a_table": (
|
|
f"{go_a}.txt",
|
|
go_header
|
|
+ "GO:0009651\tsalt stress response\tBP\t4/20\t0.001\t4\tOsSalt1/OsSalt2\n"
|
|
+ "GO:0009507\tchloroplast\tCC\t3/20\t0.002\t3\tOsChl1/OsChl2\n"
|
|
+ "GO:0016491\toxidoreductase activity\tMF\t2/20\t0.003\t2\tOsOx1/OsOx2\n",
|
|
),
|
|
"go_b_table": (
|
|
f"{go_b}.txt",
|
|
go_header + "GO:0015979\tphotosynthesis\tBP\t3/20\t0.04\t3\tOsLight1/OsLight2\n",
|
|
),
|
|
"go_a_bar_png": (f"{go_a}.barplot.png", "png"),
|
|
"go_a_bar_pdf": (f"{go_a}.barplot.pdf", "pdf"),
|
|
"go_b_bar": (f"{go_b}.barplot.png", "png"),
|
|
"go_a_dot": (f"{go_a}.dotplot.png", "png"),
|
|
"go_b_dot": (f"{go_b}.dotplot.png", "png"),
|
|
"go_a_ontology": (f"{go_a}.ontology_barplot.png", "png"),
|
|
"go_b_ontology": (f"{go_b}.ontology_barplot.png", "png"),
|
|
"kegg_a_table": (
|
|
f"{kegg_a}.KEGG_enrich.txt",
|
|
kegg_header
|
|
+ "ko00073\tCutin, suberine and wax biosynthesis\t2/20\t0.001\t2\tOsWax1/OsWax2\n"
|
|
+ "ko00196\tPhotosynthesis - antenna proteins\t2/20\t0.002\t2\tOsLight1/OsLight2\n",
|
|
),
|
|
"kegg_b_table": (
|
|
f"{kegg_b}.KEGG_enrich.txt",
|
|
kegg_header + "ko00073\tCutin, suberine and wax biosynthesis\t1/20\t0.04\t1\tOsWaxH\n",
|
|
),
|
|
"kegg_a_hits": (
|
|
f"{kegg_a}.KEGG_pathway_gene_hits.txt",
|
|
"PathwayID\tPathwayName\tKO\tGenes\n"
|
|
"ko00073\tCutin, suberine and wax biosynthesis\tK00073\tOsWax1/OsWax2\n"
|
|
"ko00196\tPhotosynthesis - antenna proteins\tK00196\tOsLight1/OsLight2\n",
|
|
),
|
|
"kegg_b_hits": (
|
|
f"{kegg_b}.KEGG_pathway_gene_hits.txt",
|
|
"PathwayID\tPathwayName\tKO\tGenes\n"
|
|
"ko00073\tCutin, suberine and wax biosynthesis\tK10073\tOsWaxH\n",
|
|
),
|
|
"kegg_a_bar_png": (f"{kegg_a}_KEGG_enrich.png", "png"),
|
|
"kegg_a_bar_pdf": (f"{kegg_a}_KEGG_enrich.pdf", "pdf"),
|
|
"kegg_b_bar": (f"{kegg_b}_KEGG_enrich.png", "png"),
|
|
"kegg_a_dot": (f"{kegg_a}_KEGG_dotplot.png", "png"),
|
|
"kegg_b_dot": (f"{kegg_b}_KEGG_dotplot.png", "png"),
|
|
"pathway_a_00073_png": (f"{kegg_a}.KEGG_pathway_map.ko00073.png", "png"),
|
|
"pathway_a_00073_pdf": (f"{kegg_a}.KEGG_pathway_map.ko00073.pdf", "pdf"),
|
|
"pathway_a_00196": (f"{kegg_a}.KEGG_pathway_map.ko00196.png", "png"),
|
|
"pathway_b_00073": (f"{kegg_b}.KEGG_pathway_map.ko00073.png", "png"),
|
|
},
|
|
)
|
|
|
|
result = generate_rnaseq_report(
|
|
workspace,
|
|
"Os",
|
|
rnaseq_scope_id="scope-combined-enrichment",
|
|
expected_stages=["rnaseq_enrich"],
|
|
use_model=False,
|
|
)
|
|
markdown = result.report_path.read_text(encoding="utf-8")
|
|
embedded = _embedded_image_names(markdown)
|
|
inventory = json.loads(result.evidence_path.read_text(encoding="utf-8"))["formal_report"][
|
|
"figure_inventory"
|
|
]
|
|
|
|
assert sum(name.endswith(".GO_enrich.barplot.png") for name in embedded) == 1
|
|
assert sum(name.endswith(".GO_enrich.dotplot.png") for name in embedded) == 1
|
|
assert sum(name.endswith(".GO_enrich.ontology_barplot.png") for name in embedded) == 1
|
|
assert sum(name.endswith("_KEGG_enrich.png") for name in embedded) == 1
|
|
assert sum(name.endswith("_KEGG_dotplot.png") for name in embedded) == 1
|
|
assert not any(name.endswith((".pdf", ".svg")) for name in embedded)
|
|
assert {
|
|
category: sum(item["category"] == category for item in inventory)
|
|
for category in ("go_bar", "go_dot", "go_ontology", "kegg_bar", "kegg_dot", "kegg_pathway")
|
|
} == {
|
|
"go_bar": 1,
|
|
"go_dot": 1,
|
|
"go_ontology": 1,
|
|
"kegg_bar": 1,
|
|
"kegg_dot": 1,
|
|
"kegg_pathway": 2,
|
|
}
|
|
assert not any(item["filename"].endswith(".pdf") for item in inventory)
|
|
|
|
pathway_names = [name for name in embedded if ".KEGG_pathway_map.ko" in name]
|
|
assert pathway_names == sorted(
|
|
[
|
|
f"{kegg_a}.KEGG_pathway_map.ko00073.png",
|
|
f"{kegg_a}.KEGG_pathway_map.ko00196.png",
|
|
]
|
|
)
|
|
expected_evidence = {
|
|
f"{go_a}.barplot.png": "salt stress response",
|
|
f"{go_a}.dotplot.png": "GO:0009651",
|
|
f"{go_a}.ontology_barplot.png": "MF=oxidoreductase activity",
|
|
f"{kegg_a}_KEGG_enrich.png": "Cutin, suberine and wax biosynthesis",
|
|
f"{kegg_a}_KEGG_dotplot.png": "ko00073",
|
|
f"{kegg_a}.KEGG_pathway_map.ko00073.png": "OsWax1",
|
|
f"{kegg_a}.KEGG_pathway_map.ko00196.png": "OsLight1",
|
|
}
|
|
for filename, evidence_value in expected_evidence.items():
|
|
section = _figure_section(markdown, filename)
|
|
assert all(
|
|
label in section
|
|
for label in ("图形含义：", "阅读方法：", "本次结果：", "解释边界：")
|
|
)
|
|
assert evidence_value in section
|
|
|
|
|
|
def test_enrichment_selection_does_not_remove_de_or_other_stage_figures(tmp_path: Path) -> None:
|
|
workspace = tmp_path / "job_non_enrichment_figures"
|
|
_write_artifact_manifest(
|
|
workspace,
|
|
"Os",
|
|
"rnaseq_de",
|
|
"scope-all-figures",
|
|
{
|
|
"de_volcano_a": ("Os.rnaseq_de.HH11__24_vs_HH11__0.volcano.png", "png"),
|
|
"de_volcano_b": ("Os.rnaseq_de.IR29__6_vs_IR29__0.volcano.png", "png"),
|
|
},
|
|
)
|
|
_write_artifact_manifest(
|
|
workspace,
|
|
"Os",
|
|
"rnaseq_qc",
|
|
"scope-all-figures",
|
|
{
|
|
"qc_before": ("Os.rnaseq_qc.before.png", "png"),
|
|
"qc_after": ("Os.rnaseq_qc.after.png", "png"),
|
|
},
|
|
)
|
|
|
|
result = generate_rnaseq_report(
|
|
workspace,
|
|
"Os",
|
|
rnaseq_scope_id="scope-all-figures",
|
|
expected_stages=["rnaseq_qc", "rnaseq_de"],
|
|
use_model=False,
|
|
)
|
|
markdown = result.report_path.read_text(encoding="utf-8")
|
|
embedded = _embedded_image_names(markdown)
|
|
|
|
for filename in (
|
|
"Os.rnaseq_de.HH11__24_vs_HH11__0.volcano.png",
|
|
"Os.rnaseq_de.IR29__6_vs_IR29__0.volcano.png",
|
|
"Os.rnaseq_qc.before.png",
|
|
"Os.rnaseq_qc.after.png",
|
|
):
|
|
assert embedded.count(filename) == 1
|
|
|
|
|
|
def test_open_regenerates_v3_report_with_renderer_v4(tmp_path: Path) -> None:
|
|
workspace = tmp_path / "job_legacy_report"
|
|
_write_manifest(workspace, "Os", "rnaseq_qc", "success", "scope-renderer")
|
|
generated = generate_rnaseq_report(
|
|
workspace,
|
|
"Os",
|
|
rnaseq_scope_id="scope-renderer",
|
|
expected_stages=["rnaseq_qc"],
|
|
use_model=False,
|
|
)
|
|
generated.report_path.write_text("STALE LEGACY REPORT\n", encoding="utf-8")
|
|
old_manifest = json.loads(generated.manifest_path.read_text(encoding="utf-8"))
|
|
old_manifest["metrics"]["report_renderer_version"] = "3"
|
|
generated.manifest_path.write_text(json.dumps(old_manifest, ensure_ascii=False), encoding="utf-8")
|
|
|
|
reopened = generate_rnaseq_report(
|
|
workspace,
|
|
"Os",
|
|
rnaseq_scope_id="scope-renderer",
|
|
expected_stages=["rnaseq_qc"],
|
|
use_model=False,
|
|
action="open",
|
|
)
|
|
|
|
assert reopened.reused is False
|
|
assert "STALE LEGACY REPORT" not in reopened.report_path.read_text(encoding="utf-8")
|
|
refreshed_manifest = json.loads(reopened.manifest_path.read_text(encoding="utf-8"))
|
|
assert refreshed_manifest["metrics"]["report_renderer_version"] == "4"
|
|
|
|
|
|
def test_pathway_without_matching_kegg_row_reports_insufficient_evidence(tmp_path: Path) -> None:
|
|
workspace = tmp_path / "job_pathway_mismatch"
|
|
base = "Os.rnaseq_kegg_enrich.IR29__6_vs_IR29__0"
|
|
_write_artifact_manifest(
|
|
workspace,
|
|
"Os",
|
|
"rnaseq_kegg_enrich",
|
|
"scope-pathway-mismatch",
|
|
{
|
|
"kegg_table": (
|
|
f"{base}.KEGG_enrich.txt",
|
|
"ID\tDescription\tGeneRatio\tp.adjust\tCount\tgeneID\n"
|
|
"ko00073\tCutin, suberine and wax biosynthesis\t2/20\t0.001\t2\tOsWax1/OsWax2\n",
|
|
),
|
|
"unmatched_pathway": (f"{base}.KEGG_pathway_map.ko99999.png", "png"),
|
|
},
|
|
)
|
|
|
|
result = generate_rnaseq_report(
|
|
workspace,
|
|
"Os",
|
|
rnaseq_scope_id="scope-pathway-mismatch",
|
|
expected_stages=["rnaseq_kegg_enrich"],
|
|
use_model=False,
|
|
)
|
|
section = _figure_section(result.report_path.read_text(encoding="utf-8"), "KEGG_pathway_map.ko99999.png")
|
|
|
|
assert "证据不足" in section
|
|
assert "Cutin, suberine and wax biosynthesis" not in section
|
|
|
|
|
|
def test_pathway_without_ko_mapping_does_not_summarize_all_ko_fold_changes(tmp_path: Path) -> None:
|
|
workspace = tmp_path / "job_pathway_without_ko_mapping"
|
|
base = "Os.rnaseq_kegg_enrich.IR29__6_vs_IR29__0"
|
|
_write_artifact_manifest(
|
|
workspace,
|
|
"Os",
|
|
"rnaseq_kegg_enrich",
|
|
"scope-no-ko-map",
|
|
{
|
|
"kegg_table": (
|
|
f"{base}.KEGG_enrich.txt",
|
|
"ID\tDescription\tGeneRatio\tp.adjust\tCount\tgeneID\n"
|
|
"ko00073\tCutin, suberine and wax biosynthesis\t2/20\t0.001\t2\tOsWax1/OsWax2\n",
|
|
),
|
|
"pathway_hits_without_ko": (
|
|
f"{base}.KEGG_pathway_gene_hits.txt",
|
|
"PathwayID\tPathwayName\tGenes\n"
|
|
"ko00073\tCutin, suberine and wax biosynthesis\tOsWax1/OsWax2\n",
|
|
),
|
|
"unscoped_ko_fc": (
|
|
f"{base}.KEGG_pathway_map.ko_log2fc.tsv",
|
|
"KO\tlog2FoldChange\nK11111\t2.5\nK22222\t-1.5\n",
|
|
),
|
|
"pathway": (f"{base}.KEGG_pathway_map.ko00073.png", "png"),
|
|
},
|
|
)
|
|
|
|
result = generate_rnaseq_report(
|
|
workspace,
|
|
"Os",
|
|
rnaseq_scope_id="scope-no-ko-map",
|
|
expected_stages=["rnaseq_kegg_enrich"],
|
|
use_model=False,
|
|
)
|
|
section = _figure_section(result.report_path.read_text(encoding="utf-8"), "KEGG_pathway_map.ko00073.png")
|
|
|
|
assert "KO对应关系证据不足" in section
|
|
assert "KO变化中上调1个、下调1个" not in section
|
|
|
|
|
|
def test_kegg_representative_selection_ignores_raw_enrichment_tables(tmp_path: Path) -> None:
|
|
workspace = tmp_path / "job_kegg_raw_selection"
|
|
header = "ID\tDescription\tGeneRatio\tp.adjust\tCount\tgeneID\n"
|
|
a = "Os.rnaseq_kegg_enrich.AA__1_vs_AA__0"
|
|
b = "Os.rnaseq_kegg_enrich.BB__1_vs_BB__0"
|
|
_write_artifact_manifest(
|
|
workspace,
|
|
"Os",
|
|
"rnaseq_kegg_enrich",
|
|
"scope-kegg-raw-selection",
|
|
{
|
|
"a_filtered": (f"{a}.KEGG_enrich.txt", header + "ko00001\tFiltered A\t1/20\t0.05\t1\tOsA\n"),
|
|
"a_raw": (f"{a}.KEGG_enrich.raw.txt", header + "ko00001\tRaw A\t1/20\t0.000000001\t1\tOsA\n"),
|
|
"a_bar": (f"{a}_KEGG_enrich.png", "png"),
|
|
"b_filtered": (f"{b}.KEGG_enrich.txt", header + "ko00002\tFiltered B\t2/20\t0.01\t2\tOsB1/OsB2\n"),
|
|
"b_raw": (f"{b}.KEGG_enrich.raw.txt", header + "ko00002\tRaw B\t2/20\t0.9\t2\tOsB1/OsB2\n"),
|
|
"b_bar": (f"{b}_KEGG_enrich.png", "png"),
|
|
},
|
|
)
|
|
|
|
result = generate_rnaseq_report(
|
|
workspace,
|
|
"Os",
|
|
rnaseq_scope_id="scope-kegg-raw-selection",
|
|
expected_stages=["rnaseq_kegg_enrich"],
|
|
use_model=False,
|
|
)
|
|
embedded = _embedded_image_names(result.report_path.read_text(encoding="utf-8"))
|
|
|
|
assert f"{b}_KEGG_enrich.png" in embedded
|
|
assert f"{a}_KEGG_enrich.png" not in embedded
|
|
|
|
|
|
def test_kegg_figure_interpretation_uses_filtered_table_not_raw_table(tmp_path: Path) -> None:
|
|
workspace = tmp_path / "job_kegg_raw_interpretation"
|
|
base = "Os.rnaseq_kegg_enrich.IR29__6_vs_IR29__0"
|
|
header = "ID\tDescription\tGeneRatio\tp.adjust\tCount\tgeneID\n"
|
|
_write_artifact_manifest(
|
|
workspace,
|
|
"Os",
|
|
"rnaseq_kegg_enrich",
|
|
"scope-kegg-raw-interpretation",
|
|
{
|
|
"raw_first": (f"{base}.KEGG_enrich.raw.txt", header + "ko99998\tRaw-only pathway\t9/20\t0.0001\t9\tRawGene\n"),
|
|
"filtered": (f"{base}.KEGG_enrich.txt", header + "ko00073\tFiltered pathway\t2/20\t0.01\t2\tOsWax1/OsWax2\n"),
|
|
"bar": (f"{base}_KEGG_enrich.png", "png"),
|
|
},
|
|
)
|
|
|
|
result = generate_rnaseq_report(
|
|
workspace,
|
|
"Os",
|
|
rnaseq_scope_id="scope-kegg-raw-interpretation",
|
|
expected_stages=["rnaseq_kegg_enrich"],
|
|
use_model=False,
|
|
)
|
|
section = _figure_section(result.report_path.read_text(encoding="utf-8"), f"{base}_KEGG_enrich.png")
|
|
|
|
assert "Filtered pathway" in section
|
|
assert "Raw-only pathway" not in section
|
|
|
|
|
|
def test_go_ontology_interpretation_covers_bp_cc_and_mf_rows(tmp_path: Path) -> None:
|
|
workspace = tmp_path / "job_go_three_ontologies"
|
|
base = "Os.rnaseq_go_enrich.MixedCase__6_vs_MixedCase__0.GO_enrich"
|
|
header = "ID\tDescription\tOntology\tGeneRatio\tp.adjust\tCount\tgeneID\n"
|
|
_write_artifact_manifest(
|
|
workspace,
|
|
"Os",
|
|
"rnaseq_go_enrich",
|
|
"scope-go-ontologies",
|
|
{
|
|
"table": (
|
|
f"{base}.txt",
|
|
header
|
|
+ "GO:0000001\tbiological process theme\tBP\t3/20\t0.001\t3\tA/B/C\n"
|
|
+ "GO:0000002\tcellular component theme\tCC\t2/20\t0.002\t2\tD/E\n"
|
|
+ "GO:0000003\tmolecular function theme\tMF\t1/20\t0.003\t1\tF\n",
|
|
),
|
|
"ontology": (f"{base}.ontology_barplot.png", "png"),
|
|
},
|
|
)
|
|
|
|
result = generate_rnaseq_report(
|
|
workspace,
|
|
"Os",
|
|
rnaseq_scope_id="scope-go-ontologies",
|
|
expected_stages=["rnaseq_go_enrich"],
|
|
use_model=False,
|
|
)
|
|
section = _figure_section(result.report_path.read_text(encoding="utf-8"), "ontology_barplot.png")
|
|
|
|
assert "biological process theme" in section
|
|
assert "cellular component theme" in section
|
|
assert "molecular function theme" in section
|
|
|
|
|
|
def test_svg_fallbacks_are_classified_and_selected_once_per_plot_form(tmp_path: Path) -> None:
|
|
workspace = tmp_path / "job_svg_fallbacks"
|
|
go_header = "ID\tDescription\tOntology\tGeneRatio\tp.adjust\tCount\tgeneID\n"
|
|
kegg_header = "ID\tDescription\tGeneRatio\tp.adjust\tCount\tgeneID\n"
|
|
_write_artifact_manifest(
|
|
workspace,
|
|
"Os",
|
|
"rnaseq_go_enrich",
|
|
"scope-svg-fallback",
|
|
{
|
|
"go_a_table": ("Os.rnaseq_go_enrich.AA.GO_enrich.txt", go_header + "GO:1\tA\tBP\t1/2\t0.01\t1\tA\n"),
|
|
"go_b_table": ("Os.rnaseq_go_enrich.BB.GO_enrich.txt", go_header + "GO:2\tB\tBP\t1/2\t0.02\t1\tB\n"),
|
|
"go_a_svg": ("Os.rnaseq_go_enrich.AA.GO_enrich.svg", "<svg/>"),
|
|
"go_b_svg": ("Os.rnaseq_go_enrich.BB.GO_enrich.svg", "<svg/>"),
|
|
},
|
|
)
|
|
_write_artifact_manifest(
|
|
workspace,
|
|
"Os",
|
|
"rnaseq_kegg_enrich",
|
|
"scope-svg-fallback",
|
|
{
|
|
"kegg_a_table": ("Os.rnaseq_kegg_enrich.AA.KEGG_enrich.txt", kegg_header + "ko1\tA\t1/2\t0.01\t1\tA\n"),
|
|
"kegg_b_table": ("Os.rnaseq_kegg_enrich.BB.KEGG_enrich.txt", kegg_header + "ko2\tB\t1/2\t0.02\t1\tB\n"),
|
|
"kegg_a_svg": ("Os.rnaseq_kegg_enrich.AA.KEGG_enrich.svg", "<svg/>"),
|
|
"kegg_b_svg": ("Os.rnaseq_kegg_enrich.BB.KEGG_enrich.svg", "<svg/>"),
|
|
},
|
|
)
|
|
|
|
result = generate_rnaseq_report(
|
|
workspace,
|
|
"Os",
|
|
rnaseq_scope_id="scope-svg-fallback",
|
|
expected_stages=["rnaseq_go_enrich", "rnaseq_kegg_enrich"],
|
|
use_model=False,
|
|
)
|
|
evidence = json.loads(result.evidence_path.read_text(encoding="utf-8"))
|
|
categories = [item["category"] for item in evidence["formal_report"]["figure_inventory"]]
|
|
|
|
assert categories.count("go_bar") == 1
|
|
assert categories.count("kegg_bar") == 1
|
|
|
|
|
|
def test_comparison_case_is_preserved_in_figure_interpretation(tmp_path: Path) -> None:
|
|
workspace = tmp_path / "job_comparison_case"
|
|
base = "Os.rnaseq_kegg_enrich.IR29__6_vs_IR29__0"
|
|
_write_artifact_manifest(
|
|
workspace,
|
|
"Os",
|
|
"rnaseq_kegg_enrich",
|
|
"scope-case",
|
|
{
|
|
"table": (
|
|
f"{base}.KEGG_enrich.txt",
|
|
"ID\tDescription\tGeneRatio\tp.adjust\tCount\tgeneID\nko00073\tWax\t2/20\t0.01\t2\tA/B\n",
|
|
),
|
|
"bar": (f"{base}_KEGG_enrich.png", "png"),
|
|
},
|
|
)
|
|
|
|
result = generate_rnaseq_report(
|
|
workspace,
|
|
"Os",
|
|
rnaseq_scope_id="scope-case",
|
|
expected_stages=["rnaseq_kegg_enrich"],
|
|
use_model=False,
|
|
)
|
|
section = _figure_section(result.report_path.read_text(encoding="utf-8"), f"{base}_KEGG_enrich.png")
|
|
|
|
assert "本次结果：比较组 `IR29__6_vs_IR29__0`" in section
|
|
|
|
|
|
def test_invalid_adjusted_p_values_do_not_participate_in_representative_selection(tmp_path: Path) -> None:
|
|
workspace = tmp_path / "job_invalid_padj"
|
|
header = "ID\tDescription\tGeneRatio\tp.adjust\tCount\tgeneID\n"
|
|
artifacts: dict[str, tuple[str, str]] = {}
|
|
for label, padj in (("NEG", "-0.1"), ("NAN", "NaN"), ("INF", "Inf"), ("VALID", "0.02")):
|
|
base = f"Os.rnaseq_kegg_enrich.{label}__1_vs_{label}__0"
|
|
artifacts[f"{label}_table"] = (f"{base}.KEGG_enrich.txt", header + f"ko00001\t{label}\t1/20\t{padj}\t1\tGene\n")
|
|
artifacts[f"{label}_bar"] = (f"{base}_KEGG_enrich.png", "png")
|
|
_write_artifact_manifest(
|
|
workspace,
|
|
"Os",
|
|
"rnaseq_kegg_enrich",
|
|
"scope-invalid-padj",
|
|
artifacts,
|
|
)
|
|
|
|
result = generate_rnaseq_report(
|
|
workspace,
|
|
"Os",
|
|
rnaseq_scope_id="scope-invalid-padj",
|
|
expected_stages=["rnaseq_kegg_enrich"],
|
|
use_model=False,
|
|
)
|
|
embedded = _embedded_image_names(result.report_path.read_text(encoding="utf-8"))
|
|
|
|
assert "Os.rnaseq_kegg_enrich.VALID__1_vs_VALID__0_KEGG_enrich.png" in embedded
|
|
|
|
|
|
def test_fgsea_curve_tied_padj_uses_largest_absolute_nes_and_matching_pathway(tmp_path: Path) -> None:
|
|
workspace = tmp_path / "job_fgsea_tie"
|
|
header = "pathway\tpathway_name\tpval\tpadj\tES\tNES\tsize\tleadingEdge\n"
|
|
aa = "Os.rnaseq_fgsea.AA__1_vs_AA__0"
|
|
zz = "Os.rnaseq_fgsea.ZZ__1_vs_ZZ__0"
|
|
_write_artifact_manifest(
|
|
workspace,
|
|
"Os",
|
|
"rnaseq_fgsea",
|
|
"scope-fgsea-tie",
|
|
{
|
|
"aa_table": (f"{aa}.fgsea_kegg.tsv", header + "ko00001\tAA pathway\t0.001\t0.01\t0.4\t2.0\t10\tA,B\n"),
|
|
"aa_curve": (f"{aa}.fgsea_kegg_top_enrichment.png", "png"),
|
|
"zz_table": (
|
|
f"{zz}.fgsea_kegg.tsv",
|
|
header
|
|
+ "ko00002\tZZ weaker tied pathway\t0.001\t0.01\t0.2\t1.0\t10\tC,D\n"
|
|
+ "ko00003\tZZ strongest tied pathway\t0.001\t0.01\t-0.7\t-3.0\t12\tE,F\n",
|
|
),
|
|
"zz_curve": (f"{zz}.fgsea_kegg_top_enrichment.png", "png"),
|
|
},
|
|
)
|
|
|
|
result = generate_rnaseq_report(
|
|
workspace,
|
|
"Os",
|
|
rnaseq_scope_id="scope-fgsea-tie",
|
|
expected_stages=["rnaseq_fgsea"],
|
|
use_model=False,
|
|
)
|
|
markdown = result.report_path.read_text(encoding="utf-8")
|
|
embedded = _embedded_image_names(markdown)
|
|
|
|
assert f"{zz}.fgsea_kegg_top_enrichment.png" in embedded
|
|
section = _figure_section(markdown, f"{zz}.fgsea_kegg_top_enrichment.png")
|
|
assert "ko00003" in section
|
|
assert "NES=-3.0" in section
|
|
|
|
|
|
def test_pdf_only_pathway_is_linked_or_recorded_in_figure_inventory(tmp_path: Path) -> None:
|
|
workspace = tmp_path / "job_pdf_pathway"
|
|
base = "Os.rnaseq_kegg_enrich.IR29__6_vs_IR29__0"
|
|
pdf_name = f"{base}.KEGG_pathway_map.ko00073.pdf"
|
|
_write_artifact_manifest(
|
|
workspace,
|
|
"Os",
|
|
"rnaseq_kegg_enrich",
|
|
"scope-pdf-pathway",
|
|
{
|
|
"table": (
|
|
f"{base}.KEGG_enrich.txt",
|
|
"ID\tDescription\tGeneRatio\tp.adjust\tCount\tgeneID\nko00073\tWax\t2/20\t0.01\t2\tA/B\n",
|
|
),
|
|
"pdf_pathway": (pdf_name, "pdf"),
|
|
},
|
|
)
|
|
|
|
result = generate_rnaseq_report(
|
|
workspace,
|
|
"Os",
|
|
rnaseq_scope_id="scope-pdf-pathway",
|
|
expected_stages=["rnaseq_kegg_enrich"],
|
|
use_model=False,
|
|
)
|
|
markdown = result.report_path.read_text(encoding="utf-8")
|
|
evidence = json.loads(result.evidence_path.read_text(encoding="utf-8"))
|
|
inventory_names = {item["filename"] for item in evidence["formal_report"]["figure_inventory"]}
|
|
linked = re.search(rf"\[[^]]*\]\([^)]*{re.escape(pdf_name)}[^)]*\)", markdown) is not None
|
|
|
|
assert linked or pdf_name in inventory_names
|
|
|
|
|
|
def test_renderer_and_figure_inventory_are_consistent_in_evidence_and_manifest(tmp_path: Path) -> None:
|
|
workspace = tmp_path / "job_report_metadata"
|
|
_write_enrichment_manifests(workspace)
|
|
|
|
result = generate_rnaseq_report(
|
|
workspace,
|
|
"Os",
|
|
rnaseq_scope_id="scope-enrichment",
|
|
expected_stages=["rnaseq_go_enrich", "rnaseq_kegg_enrich", "rnaseq_fgsea"],
|
|
use_model=False,
|
|
)
|
|
evidence = json.loads(result.evidence_path.read_text(encoding="utf-8"))
|
|
manifest = json.loads(result.manifest_path.read_text(encoding="utf-8"))
|
|
formal = evidence["formal_report"]
|
|
metrics = manifest["metrics"]
|
|
|
|
assert formal["report_renderer_version"] == metrics["report_renderer_version"]
|
|
assert formal["figure_inventory"]
|
|
assert metrics.get("figure_inventory") == formal["figure_inventory"]
|
|
|
|
|
|
def test_go_ontology_interpretation_accepts_real_long_ontology_names(tmp_path: Path) -> None:
|
|
workspace = tmp_path / "job_go_real_ontology_names"
|
|
base = "Os.rnaseq_go_enrich.A__1_vs_A__0.GO_enrich"
|
|
_write_artifact_manifest(
|
|
workspace,
|
|
"Os",
|
|
"rnaseq_go_enrich",
|
|
"scope-go-real-ontology",
|
|
{
|
|
"table": (
|
|
f"{base}.txt",
|
|
"ID\tDescription\tOntology\tGeneRatio\tp.adjust\tCount\tgeneID\n"
|
|
"GO:0000001\tlong-name BP theme\tBiological process\t3/20\t0.001\t3\tA/B/C\n"
|
|
"GO:0000002\tlong-name CC theme\tCellular component\t2/20\t0.002\t2\tD/E\n"
|
|
"GO:0000003\tlong-name MF theme\tMolecular function\t1/20\t0.003\t1\tF\n",
|
|
),
|
|
"ontology": (f"{base}.ontology_barplot.png", "png"),
|
|
},
|
|
)
|
|
|
|
result = generate_rnaseq_report(
|
|
workspace,
|
|
"Os",
|
|
rnaseq_scope_id="scope-go-real-ontology",
|
|
expected_stages=["rnaseq_go_enrich"],
|
|
use_model=False,
|
|
)
|
|
section = _figure_section(result.report_path.read_text(encoding="utf-8"), "ontology_barplot.png")
|
|
|
|
assert "BP=long-name BP theme" in section
|
|
assert "CC=long-name CC theme" in section
|
|
assert "MF=long-name MF theme" in section
|
|
|
|
|
|
def test_pathway_interpretation_merges_genes_from_all_matching_ko_rows(tmp_path: Path) -> None:
|
|
workspace = tmp_path / "job_pathway_multi_ko_hits"
|
|
base = "Os.rnaseq_kegg_enrich.A__1_vs_A__0"
|
|
_write_artifact_manifest(
|
|
workspace,
|
|
"Os",
|
|
"rnaseq_kegg_enrich",
|
|
"scope-pathway-multi-ko",
|
|
{
|
|
"enrichment": (
|
|
f"{base}.KEGG_enrich.txt",
|
|
"ID\tDescription\tGeneRatio\tp.adjust\tCount\tgeneID\n"
|
|
"ko00073\tWax pathway\t3/20\t0.001\t3\tGeneA/GeneB/GeneC\n",
|
|
),
|
|
"hits": (
|
|
f"{base}.KEGG_pathway_gene_hits.txt",
|
|
"PathwayID\tPathwayName\tKO\tGenes\n"
|
|
"ko00073\tWax pathway\tK00001\tGeneA/GeneB\n"
|
|
"ko00073\tWax pathway\tK00002\tGeneC\n",
|
|
),
|
|
"pathway": (f"{base}.KEGG_pathway_map.ko00073.png", "png"),
|
|
},
|
|
)
|
|
|
|
result = generate_rnaseq_report(
|
|
workspace,
|
|
"Os",
|
|
rnaseq_scope_id="scope-pathway-multi-ko",
|
|
expected_stages=["rnaseq_kegg_enrich"],
|
|
use_model=False,
|
|
)
|
|
section = _figure_section(result.report_path.read_text(encoding="utf-8"), "KEGG_pathway_map.ko00073.png")
|
|
|
|
assert "GeneA" in section
|
|
assert "GeneB" in section
|
|
assert "GeneC" in section
|
|
|
|
|
|
def test_figure_uses_exact_comparison_table_when_names_contain_each_other(tmp_path: Path) -> None:
|
|
workspace = tmp_path / "job_exact_comparison"
|
|
base_a = "Os.rnaseq_go_enrich.A.GO_enrich"
|
|
base_aa = "Os.rnaseq_go_enrich.AA.GO_enrich"
|
|
_write_artifact_manifest(
|
|
workspace,
|
|
"Os",
|
|
"rnaseq_go_enrich",
|
|
"scope-exact-comparison",
|
|
{
|
|
"aa_table": (
|
|
f"{base_aa}.txt",
|
|
"ID\tDescription\tOntology\tGeneRatio\tp.adjust\tCount\tgeneID\n"
|
|
"GO:0000002\tAA-only evidence\tBP\t2/20\t0.0001\t2\tAA1/AA2\n",
|
|
),
|
|
"a_table": (
|
|
f"{base_a}.txt",
|
|
"ID\tDescription\tOntology\tGeneRatio\tp.adjust\tCount\tgeneID\n"
|
|
"GO:0000001\tA exact evidence\tBP\t1/20\t0.01\t1\tA1\n",
|
|
),
|
|
"a_bar": (f"{base_a}.barplot.png", "png"),
|
|
},
|
|
)
|
|
|
|
result = generate_rnaseq_report(
|
|
workspace,
|
|
"Os",
|
|
rnaseq_scope_id="scope-exact-comparison",
|
|
expected_stages=["rnaseq_go_enrich"],
|
|
use_model=False,
|
|
)
|
|
section = _figure_section(result.report_path.read_text(encoding="utf-8"), f"{base_a}.barplot.png")
|
|
|
|
assert "A exact evidence" in section
|
|
assert "AA-only evidence" not in section
|
|
|
|
|
|
@pytest.mark.parametrize("invalid_padj", ["NaN", "Inf", "-0.1"])
|
|
def test_mandatory_pathway_does_not_render_invalid_padj_as_significance(
|
|
tmp_path: Path, invalid_padj: str
|
|
) -> None:
|
|
workspace = tmp_path / f"job_invalid_pathway_padj_{invalid_padj.replace('-', 'neg')}"
|
|
base = "Os.rnaseq_kegg_enrich.A__1_vs_A__0"
|
|
_write_artifact_manifest(
|
|
workspace,
|
|
"Os",
|
|
"rnaseq_kegg_enrich",
|
|
"scope-invalid-pathway-padj",
|
|
{
|
|
"table": (
|
|
f"{base}.KEGG_enrich.txt",
|
|
"ID\tDescription\tGeneRatio\tp.adjust\tCount\tgeneID\n"
|
|
f"ko00073\tWax pathway\t2/20\t{invalid_padj}\t2\tA/B\n",
|
|
),
|
|
"pathway": (f"{base}.KEGG_pathway_map.ko00073.png", "png"),
|
|
},
|
|
)
|
|
|
|
result = generate_rnaseq_report(
|
|
workspace,
|
|
"Os",
|
|
rnaseq_scope_id="scope-invalid-pathway-padj",
|
|
expected_stages=["rnaseq_kegg_enrich"],
|
|
use_model=False,
|
|
)
|
|
section = _figure_section(result.report_path.read_text(encoding="utf-8"), "KEGG_pathway_map.ko00073.png")
|
|
|
|
assert "p.adjust=未登记" in section
|
|
assert f"p.adjust={invalid_padj}" not in section
|
|
|
|
|
|
def test_pdf_only_non_pathway_enrichment_forms_are_linked_and_in_inventory(tmp_path: Path) -> None:
|
|
workspace = tmp_path / "job_pdf_only_enrichment_forms"
|
|
go_base = "Os.rnaseq_go_enrich.A__1_vs_A__0.GO_enrich"
|
|
kegg_base = "Os.rnaseq_kegg_enrich.A__1_vs_A__0"
|
|
fgsea_base = "Os.rnaseq_fgsea.A__1_vs_A__0"
|
|
_write_artifact_manifest(
|
|
workspace,
|
|
"Os",
|
|
"rnaseq_go_enrich",
|
|
"scope-pdf-forms",
|
|
{
|
|
"table": (
|
|
f"{go_base}.txt",
|
|
"ID\tDescription\tOntology\tGeneRatio\tp.adjust\tCount\tgeneID\n"
|
|
"GO:1\tGO evidence\tBP\t1/20\t0.01\t1\tA\n",
|
|
),
|
|
"bar": (f"{go_base}.barplot.pdf", "pdf"),
|
|
"dot": (f"{go_base}.dotplot.pdf", "pdf"),
|
|
"ontology": (f"{go_base}.ontology_barplot.pdf", "pdf"),
|
|
},
|
|
)
|
|
_write_artifact_manifest(
|
|
workspace,
|
|
"Os",
|
|
"rnaseq_kegg_enrich",
|
|
"scope-pdf-forms",
|
|
{
|
|
"table": (
|
|
f"{kegg_base}.KEGG_enrich.txt",
|
|
"ID\tDescription\tGeneRatio\tp.adjust\tCount\tgeneID\n"
|
|
"ko00001\tKEGG evidence\t1/20\t0.01\t1\tA\n",
|
|
),
|
|
"bar": (f"{kegg_base}_KEGG_enrich.pdf", "pdf"),
|
|
"dot": (f"{kegg_base}_KEGG_dotplot.pdf", "pdf"),
|
|
},
|
|
)
|
|
_write_artifact_manifest(
|
|
workspace,
|
|
"Os",
|
|
"rnaseq_fgsea",
|
|
"scope-pdf-forms",
|
|
{
|
|
"table": (
|
|
f"{fgsea_base}.fgsea_kegg.tsv",
|
|
"pathway\tpathway_name\tpadj\tNES\tleadingEdge\n"
|
|
"ko00001\tFGSEA evidence\t0.01\t2.0\tA,B\n",
|
|
),
|
|
"dot": (f"{fgsea_base}.fgsea_kegg_dotplot.pdf", "pdf"),
|
|
"curve": (f"{fgsea_base}.fgsea_kegg_top_enrichment.pdf", "pdf"),
|
|
},
|
|
)
|
|
|
|
result = generate_rnaseq_report(
|
|
workspace,
|
|
"Os",
|
|
rnaseq_scope_id="scope-pdf-forms",
|
|
expected_stages=["rnaseq_go_enrich", "rnaseq_kegg_enrich", "rnaseq_fgsea"],
|
|
use_model=False,
|
|
)
|
|
markdown = result.report_path.read_text(encoding="utf-8")
|
|
evidence = json.loads(result.evidence_path.read_text(encoding="utf-8"))
|
|
expected = {
|
|
f"{go_base}.barplot.pdf",
|
|
f"{go_base}.dotplot.pdf",
|
|
f"{go_base}.ontology_barplot.pdf",
|
|
f"{kegg_base}_KEGG_enrich.pdf",
|
|
f"{kegg_base}_KEGG_dotplot.pdf",
|
|
f"{fgsea_base}.fgsea_kegg_dotplot.pdf",
|
|
f"{fgsea_base}.fgsea_kegg_top_enrichment.pdf",
|
|
}
|
|
inventory_names = {item["filename"] for item in evidence["formal_report"]["figure_inventory"]}
|
|
|
|
assert expected <= inventory_names
|
|
for filename in expected:
|
|
assert re.search(rf"\[[^]]*\]\([^)]*{re.escape(filename)}[^)]*\)", markdown)
|
|
|
|
|
|
def test_representative_selection_requires_legal_nonempty_evidence_before_format(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
workspace = tmp_path / "job_representative_evidence_priority"
|
|
header = "ID\tDescription\tGeneRatio\tp.adjust\tCount\tgeneID\n"
|
|
artifacts: dict[str, tuple[str, str]] = {}
|
|
cases = (
|
|
("EMPTY", "", "png"),
|
|
("INVALID", header + "ko1\tinvalid\t1/20\t-0.1\t1\tA\n", "png"),
|
|
("VALID_SVG", header + "ko2\tvalid svg\t1/20\t0.04\t1\tB\n", "svg"),
|
|
("VALID_PNG", header + "ko3\tvalid png\t1/20\t0.03\t1\tC\n", "png"),
|
|
)
|
|
for label, table_content, extension in cases:
|
|
base = f"Os.rnaseq_kegg_enrich.{label}"
|
|
artifacts[f"{label}_table"] = (f"{base}.KEGG_enrich.txt", table_content)
|
|
artifacts[f"{label}_figure"] = (f"{base}_KEGG_enrich.{extension}", "figure")
|
|
_write_artifact_manifest(
|
|
workspace,
|
|
"Os",
|
|
"rnaseq_kegg_enrich",
|
|
"scope-representative-priority",
|
|
artifacts,
|
|
)
|
|
|
|
result = generate_rnaseq_report(
|
|
workspace,
|
|
"Os",
|
|
rnaseq_scope_id="scope-representative-priority",
|
|
expected_stages=["rnaseq_kegg_enrich"],
|
|
use_model=False,
|
|
)
|
|
inventory = json.loads(result.evidence_path.read_text(encoding="utf-8"))["formal_report"]["figure_inventory"]
|
|
selected = [item["filename"] for item in inventory if item["category"] == "kegg_bar"]
|
|
|
|
assert selected == ["Os.rnaseq_kegg_enrich.VALID_PNG_KEGG_enrich.png"]
|
|
|
|
|
|
def test_model_enhancement_rejects_numbers_and_activation_claims_absent_from_evidence(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
from gfa_core.mas import llm as llm_module
|
|
from gfa_core.mas.config import config
|
|
|
|
class FakeManager:
|
|
def __init__(self, _config: object) -> None:
|
|
pass
|
|
|
|
def call(self, *_args: object, **_kwargs: object) -> object:
|
|
return type(
|
|
"Response",
|
|
(),
|
|
{
|
|
"content": (
|
|
"模型声称42个基因显著变化，因此该通路被激活并直接导致抗逆性增强；"
|
|
"同时还声称处理后产量提高了37%，这些数字与因果结论均不在输入证据中。"
|
|
),
|
|
"provider": "fake",
|
|
"model": "fake-model",
|
|
},
|
|
)()
|
|
|
|
monkeypatch.setattr(config, "load_llm_config", lambda: {})
|
|
monkeypatch.setattr(llm_module, "LLMManager", FakeManager)
|
|
evidence = {
|
|
"metadata": {"included_stages": ["rnaseq_kegg_enrich"]},
|
|
"biological_story": "检测到KEGG富集结果，具体证据见结果表。",
|
|
"key_findings": [],
|
|
"major_claims": [],
|
|
"risk_flags": [],
|
|
}
|
|
|
|
summary, metadata = rnaseq_report_module._model_enhancement(evidence, enabled=True)
|
|
|
|
assert summary == ""
|
|
assert metadata["status"] == "fallback"
|
|
assert metadata["error"]
|
|
|
|
|
|
def _install_fake_model_response(monkeypatch: pytest.MonkeyPatch, content: str) -> None:
|
|
from gfa_core.mas import llm as llm_module
|
|
from gfa_core.mas.config import config
|
|
|
|
class FakeManager:
|
|
def __init__(self, _config: object) -> None:
|
|
pass
|
|
|
|
def call(self, *_args: object, **_kwargs: object) -> object:
|
|
return type(
|
|
"Response",
|
|
(),
|
|
{"content": content, "provider": "fake", "model": "fake-model"},
|
|
)()
|
|
|
|
monkeypatch.setattr(config, "load_llm_config", lambda: {})
|
|
monkeypatch.setattr(llm_module, "LLMManager", FakeManager)
|
|
|
|
|
|
def test_free_form_model_output_is_rejected_and_never_embedded_in_markdown(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
workspace = tmp_path / "job_model_not_embedded"
|
|
_write_manifest(workspace, "Os", "rnaseq_qc", "success", "scope-model-not-embedded")
|
|
model_text = (
|
|
"现有证据仅支持对已完成阶段进行描述性总结，各项结果仍需结合实验设计和原始结果表审慎解释，"
|
|
"不应扩展为未被数据支持的生物学结论。"
|
|
)
|
|
_install_fake_model_response(monkeypatch, model_text)
|
|
|
|
result = generate_rnaseq_report(
|
|
workspace,
|
|
"Os",
|
|
rnaseq_scope_id="scope-model-not-embedded",
|
|
expected_stages=["rnaseq_qc"],
|
|
use_model=True,
|
|
)
|
|
|
|
markdown = result.report_path.read_text(encoding="utf-8")
|
|
model_record = json.loads(result.model_path.read_text(encoding="utf-8"))
|
|
assert model_text not in markdown
|
|
assert "### 模型辅助语言润色（不作为新增证据）" in markdown
|
|
assert "API 仅参与已有证据语句的排序和组织" in markdown
|
|
assert model_record["content"] == model_text
|
|
assert model_record["status"] == "fallback"
|
|
assert model_record["policy"] == "evidence_claim_ordering_only"
|
|
assert model_record["error"]
|
|
|
|
|
|
def test_noncausal_free_form_model_text_is_still_rejected(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
model_text = (
|
|
"这些富集结果只能说明统计关联，不能证明因果关系，也无法认定相关通路被激活，"
|
|
"仍需结合独立实验进行验证。"
|
|
)
|
|
_install_fake_model_response(monkeypatch, model_text)
|
|
evidence = {
|
|
"metadata": {"included_stages": ["rnaseq_kegg_enrich"]},
|
|
"biological_story": "检测到KEGG富集结果，具体证据见结果表。",
|
|
"key_findings": [],
|
|
"major_claims": [],
|
|
"risk_flags": [],
|
|
}
|
|
|
|
summary, metadata = rnaseq_report_module._model_enhancement(evidence, enabled=True)
|
|
|
|
assert summary == ""
|
|
assert metadata["content"] == model_text
|
|
assert metadata["status"] == "fallback"
|
|
assert metadata["policy"] == "evidence_claim_ordering_only"
|
|
assert metadata["error"]
|
|
|
|
|
|
def test_model_cannot_embed_mechanistic_claims_even_when_all_numbers_exist_in_evidence(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
model_text = (
|
|
"已有结果中的12个基因显著上调并促进耐盐，揭示了该通路控制抗逆性的调控机制，"
|
|
"因此可作为后续育种的确定依据。"
|
|
)
|
|
_install_fake_model_response(monkeypatch, model_text)
|
|
evidence = {
|
|
"metadata": {"included_stages": ["rnaseq_kegg_enrich"]},
|
|
"biological_story": "结果表登记12个命中基因。",
|
|
"key_findings": [],
|
|
"major_claims": [],
|
|
"risk_flags": [],
|
|
}
|
|
|
|
summary, metadata = rnaseq_report_module._model_enhancement(evidence, enabled=True)
|
|
|
|
assert summary == ""
|
|
assert metadata["content"] == model_text
|
|
assert metadata["status"] == "fallback"
|
|
assert metadata["policy"] == "evidence_claim_ordering_only"
|
|
|
|
|
|
def test_model_can_only_order_and_group_approved_evidence_claims(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
evidence = {
|
|
"metadata": {"included_stages": ["rnaseq_qc"]},
|
|
"biological_story": "质控阶段已经完成，具体质量指标见结果表。",
|
|
"key_findings": [
|
|
{"stage": "rnaseq_qc", "description": "成功阶段提供了可追溯的质控结果。"}
|
|
],
|
|
"major_claims": [],
|
|
"risk_flags": [],
|
|
}
|
|
organization = {
|
|
"paragraph_groups": [["finding_1", "story_1"]],
|
|
}
|
|
_install_fake_model_response(monkeypatch, json.dumps(organization, ensure_ascii=False))
|
|
|
|
summary, metadata = rnaseq_report_module._model_enhancement(evidence, enabled=True)
|
|
|
|
assert summary == "成功阶段提供了可追溯的质控结果。质控阶段已经完成，具体质量指标见结果表。"
|
|
assert metadata["status"] == "embedded_evidence_only"
|
|
assert metadata["policy"] == "evidence_claim_ordering_only"
|
|
assert metadata["organization_plan"] == organization
|
|
assert metadata["selected_claim_ids"] == ["finding_1", "story_1"]
|
|
assert {row["claim_id"] for row in metadata["claim_catalog"]} == {"story_1", "finding_1"}
|
|
|
|
|
|
def test_unknown_model_claim_id_forces_deterministic_fallback(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
evidence = {
|
|
"metadata": {"included_stages": ["rnaseq_qc"]},
|
|
"biological_story": "质控阶段已经完成。",
|
|
"key_findings": [],
|
|
"major_claims": [],
|
|
"risk_flags": [],
|
|
}
|
|
_install_fake_model_response(
|
|
monkeypatch,
|
|
json.dumps({"paragraph_groups": [["story_1", "invented_claim"]]}, ensure_ascii=False),
|
|
)
|
|
|
|
summary, metadata = rnaseq_report_module._model_enhancement(evidence, enabled=True)
|
|
|
|
assert summary == ""
|
|
assert metadata["status"] == "fallback"
|
|
assert "invented_claim" in metadata["error"]
|
|
|
|
|
|
def test_report_always_states_the_api_evidence_boundary(tmp_path: Path) -> None:
|
|
workspace = tmp_path / "job_model_boundary"
|
|
_write_manifest(workspace, "Os", "rnaseq_qc", "success", "scope-model-boundary")
|
|
|
|
result = generate_rnaseq_report(
|
|
workspace,
|
|
"Os",
|
|
rnaseq_scope_id="scope-model-boundary",
|
|
expected_stages=["rnaseq_qc"],
|
|
use_model=False,
|
|
)
|
|
markdown = result.report_path.read_text(encoding="utf-8")
|
|
|
|
assert "### 模型辅助语言润色（不作为新增证据）" in markdown
|
|
assert "API 仅参与已有证据语句的排序和组织" in markdown
|
|
assert "不参与结果分析或推断" in markdown
|
|
assert "未采用模型组织结果" in markdown
|
|
|
|
|
|
def test_pathway_interpretation_previews_only_twelve_genes_and_kos_and_links_full_tables(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
workspace = tmp_path / "job_pathway_long_lists"
|
|
base = "Os.rnaseq_kegg_enrich.IR29__6_vs_IR29__0"
|
|
genes = [f"Gene{index:03d}" for index in range(1, 456)]
|
|
kos = [f"K{index:05d}" for index in range(1, 456)]
|
|
hit_rows = "".join(
|
|
f"ko00073\tWax pathway\t{ko}\t{gene}\n" for ko, gene in zip(kos, genes)
|
|
)
|
|
ko_rows = "".join(f"{ko}\t1.0\n" for ko in kos)
|
|
_write_artifact_manifest(
|
|
workspace,
|
|
"Os",
|
|
"rnaseq_kegg_enrich",
|
|
"scope-pathway-long-lists",
|
|
{
|
|
"filtered": (
|
|
f"{base}.KEGG_enrich.txt",
|
|
"ID\tDescription\tGeneRatio\tp.adjust\tCount\tgeneID\n"
|
|
f"ko00073\tWax pathway\t455/1000\t0.001\t455\t{'/'.join(genes)}\n",
|
|
),
|
|
"hits": (
|
|
f"{base}.KEGG_pathway_gene_hits.txt",
|
|
"PathwayID\tPathwayName\tKO\tGenes\n" + hit_rows,
|
|
),
|
|
"ko_log2fc": (
|
|
f"{base}.KEGG_pathway_map.ko_log2fc.tsv",
|
|
"KO\tlog2FoldChange\n" + ko_rows,
|
|
),
|
|
"pathway": (f"{base}.KEGG_pathway_map.ko00073.png", "png"),
|
|
},
|
|
)
|
|
|
|
result = generate_rnaseq_report(
|
|
workspace,
|
|
"Os",
|
|
rnaseq_scope_id="scope-pathway-long-lists",
|
|
expected_stages=["rnaseq_kegg_enrich"],
|
|
use_model=False,
|
|
)
|
|
section = _figure_section(result.report_path.read_text(encoding="utf-8"), "KEGG_pathway_map.ko00073.png")
|
|
|
|
assert "命中基因总数=455" in section
|
|
assert "KO总数=455" in section
|
|
assert all(gene in section for gene in genes[:12])
|
|
assert all(ko in section for ko in kos[:12])
|
|
assert genes[12] not in section
|
|
assert kos[12] not in section
|
|
assert f"[{base}.KEGG_pathway_gene_hits.txt]" in section
|
|
assert f"[{base}.KEGG_pathway_map.ko_log2fc.tsv]" in section
|
|
|
|
|
|
def test_fgsea_leading_edge_previews_twelve_genes_and_links_full_result_table(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
workspace = tmp_path / "job_fgsea_long_leading_edge"
|
|
base = "Os.rnaseq_fgsea.IR29__6_vs_IR29__0"
|
|
leading_edge = [f"LeadGene{index:03d}" for index in range(1, 31)]
|
|
table_name = f"{base}.fgsea_kegg.tsv"
|
|
_write_artifact_manifest(
|
|
workspace,
|
|
"Os",
|
|
"rnaseq_fgsea",
|
|
"scope-fgsea-long-leading-edge",
|
|
{
|
|
"fgsea_table": (
|
|
table_name,
|
|
"pathway\tpathway_name\tpadj\tNES\tsize\tleadingEdge\n"
|
|
f"ko00196\tPhotosynthesis - antenna proteins\t0.001\t2.1\t30\t{','.join(leading_edge)}\n",
|
|
),
|
|
"fgsea_curve": (f"{base}.fgsea_kegg_top_enrichment.png", "png"),
|
|
},
|
|
)
|
|
|
|
result = generate_rnaseq_report(
|
|
workspace,
|
|
"Os",
|
|
rnaseq_scope_id="scope-fgsea-long-leading-edge",
|
|
expected_stages=["rnaseq_fgsea"],
|
|
use_model=False,
|
|
)
|
|
markdown = result.report_path.read_text(encoding="utf-8")
|
|
section = _figure_section(markdown, f"{base}.fgsea_kegg_top_enrichment.png")
|
|
|
|
assert "leadingEdge总数=30" in section
|
|
assert "前12项=" in section
|
|
assert all(gene in section for gene in leading_edge[:12])
|
|
assert all(gene not in section for gene in leading_edge[12:])
|
|
assert f"[{table_name}](" in section
|
|
assert (
|
|
"> 报告生成原则：结果判断只来自脚本、Manifest 和结果表；API 仅参与已有证据语句的排序和组织，"
|
|
"不参与结果分析或推断。" in markdown
|
|
)
|
|
|
|
|
|
def test_candidate_guided_pathways_keep_all_supported_contents_and_omit_unrelated(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
workspace = tmp_path / "job_candidate_guided_pathways"
|
|
ids = _write_candidate_guided_enrichment_manifests(workspace)
|
|
|
|
result = generate_rnaseq_report(
|
|
workspace,
|
|
ids["prefix"],
|
|
rnaseq_scope_id=ids["scope"],
|
|
expected_stages=[
|
|
"rnaseq_go_enrich",
|
|
"rnaseq_kegg_enrich",
|
|
"rnaseq_fgsea",
|
|
"rnaseq_family_score",
|
|
],
|
|
use_model=False,
|
|
)
|
|
embedded = _embedded_image_names(result.report_path.read_text(encoding="utf-8"))
|
|
evidence = json.loads(result.evidence_path.read_text(encoding="utf-8"))
|
|
manifest = json.loads(result.manifest_path.read_text(encoding="utf-8"))
|
|
formal = evidence["formal_report"]
|
|
pathway_rows = [
|
|
row for row in formal["figure_inventory"] if row["category"] == "kegg_pathway"
|
|
]
|
|
|
|
assert formal["selection_mode"] == "candidate_guided"
|
|
assert {row["pathway_id"] for row in pathway_rows} == {
|
|
ids["supported_pathway_a"],
|
|
ids["supported_pathway_b"],
|
|
}
|
|
assert all(row["comparison"] == ids["candidate_comparison"] for row in pathway_rows)
|
|
assert not any(ids["unrelated_pathway"] in name for name in embedded)
|
|
assert not any(
|
|
ids["unrelated_comparison"] in name and ids["supported_pathway_a"] in name
|
|
for name in embedded
|
|
)
|
|
|
|
shared = next(
|
|
row for row in pathway_rows if row["pathway_id"] == ids["supported_pathway_a"]
|
|
)
|
|
assert shared["candidate_families"] == ["FAM_RANDOM_ALPHA", "FAM_RANDOM_BETA"]
|
|
assert any(
|
|
label.startswith(f"KEGG:{ids['supported_pathway_a']}:")
|
|
for label in shared["support_labels"]
|
|
)
|
|
assert shared["selection_reason"] == "candidate_family_support"
|
|
assert sum(ids["supported_pathway_a"] in name for name in embedded) == 1
|
|
|
|
omitted = formal["omitted_figure_inventory"]
|
|
assert any(row["pathway_id"] == ids["unrelated_pathway"] for row in omitted)
|
|
assert formal["omitted_figure_count"] == len(omitted)
|
|
assert manifest["metrics"]["selection_mode"] == formal["selection_mode"]
|
|
assert manifest["metrics"]["omitted_figure_count"] == formal["omitted_figure_count"]
|
|
|
|
|
|
def test_candidate_guided_representatives_use_candidate_comparison_and_supported_fgsea(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
workspace = tmp_path / "job_candidate_guided_forms"
|
|
ids = _write_candidate_guided_enrichment_manifests(workspace)
|
|
|
|
result = generate_rnaseq_report(
|
|
workspace,
|
|
ids["prefix"],
|
|
rnaseq_scope_id=ids["scope"],
|
|
expected_stages=[
|
|
"rnaseq_go_enrich",
|
|
"rnaseq_kegg_enrich",
|
|
"rnaseq_fgsea",
|
|
"rnaseq_family_score",
|
|
],
|
|
use_model=False,
|
|
)
|
|
inventory = json.loads(result.evidence_path.read_text(encoding="utf-8"))["formal_report"][
|
|
"figure_inventory"
|
|
]
|
|
selected_categories = {
|
|
category: [row for row in inventory if row["category"] == category]
|
|
for category in (
|
|
"go_bar",
|
|
"go_dot",
|
|
"go_ontology",
|
|
"kegg_bar",
|
|
"kegg_dot",
|
|
"fgsea_dot",
|
|
"fgsea_curve",
|
|
)
|
|
}
|
|
|
|
assert all(len(rows) == 1 for rows in selected_categories.values())
|
|
assert all(
|
|
rows[0]["comparison"] == ids["candidate_comparison"]
|
|
for rows in selected_categories.values()
|
|
)
|
|
assert all(rows[0]["candidate_families"] for rows in selected_categories.values())
|
|
assert all(
|
|
rows[0]["selection_reason"] == "candidate_family_support"
|
|
for rows in selected_categories.values()
|
|
)
|
|
fgsea_curve = selected_categories["fgsea_curve"][0]
|
|
assert fgsea_curve["candidate_families"] == ["FAM_RANDOM_ALPHA"]
|
|
assert fgsea_curve["support_labels"] == [
|
|
f"fgsea:{ids['supported_pathway_b']}:random pathway beta"
|
|
]
|
|
|
|
|
|
def test_fgsea_curve_requires_exact_candidate_support_for_its_actual_top_pathway(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
workspace = tmp_path / "job_fgsea_curve_exact_identity"
|
|
ids = _write_candidate_guided_enrichment_manifests(workspace)
|
|
fgsea_table = (
|
|
workspace
|
|
/ "final"
|
|
/ f"{ids['prefix']}.rnaseq_fgsea.{ids['candidate_comparison']}.fgsea_kegg.tsv"
|
|
)
|
|
fgsea_table.write_text(
|
|
"pathway\tpathway_name\tpadj\tNES\tsize\tleadingEdge\n"
|
|
"ko99999\tnon-candidate top pathway\t0.000001\t3.5\t20\tOtherA,OtherB\n"
|
|
f"{ids['supported_pathway_b']}\trandom pathway beta\t0.04\t2.1\t12\tGeneB,GeneC\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
result = generate_rnaseq_report(
|
|
workspace,
|
|
ids["prefix"],
|
|
rnaseq_scope_id=ids["scope"],
|
|
expected_stages=[
|
|
"rnaseq_go_enrich",
|
|
"rnaseq_kegg_enrich",
|
|
"rnaseq_fgsea",
|
|
"rnaseq_family_score",
|
|
],
|
|
use_model=False,
|
|
)
|
|
formal = json.loads(result.evidence_path.read_text(encoding="utf-8"))["formal_report"]
|
|
selected = formal["figure_inventory"]
|
|
omitted = formal["omitted_figure_inventory"]
|
|
fgsea_dots = [row for row in selected if row["category"] == "fgsea_dot"]
|
|
fgsea_curves = [row for row in selected if row["category"] == "fgsea_curve"]
|
|
|
|
assert len(fgsea_dots) == 1
|
|
assert fgsea_dots[0]["comparison"] == ids["candidate_comparison"]
|
|
assert fgsea_dots[0]["candidate_families"] == ["FAM_RANDOM_ALPHA"]
|
|
assert fgsea_curves == []
|
|
assert any(row["category"] == "fgsea_curve" for row in omitted)
|
|
|
|
|
|
def test_fgsea_is_excluded_from_candidate_mainline_without_fgsea_support(tmp_path: Path) -> None:
|
|
workspace = tmp_path / "job_candidate_without_fgsea"
|
|
ids = _write_candidate_guided_enrichment_manifests(
|
|
workspace,
|
|
include_fgsea_support=False,
|
|
)
|
|
|
|
result = generate_rnaseq_report(
|
|
workspace,
|
|
ids["prefix"],
|
|
rnaseq_scope_id=ids["scope"],
|
|
expected_stages=[
|
|
"rnaseq_go_enrich",
|
|
"rnaseq_kegg_enrich",
|
|
"rnaseq_fgsea",
|
|
"rnaseq_family_score",
|
|
],
|
|
use_model=False,
|
|
)
|
|
formal = json.loads(result.evidence_path.read_text(encoding="utf-8"))["formal_report"]
|
|
categories = {row["category"] for row in formal["figure_inventory"]}
|
|
omitted_categories = {row["category"] for row in formal["omitted_figure_inventory"]}
|
|
|
|
assert "fgsea_dot" not in categories
|
|
assert "fgsea_curve" not in categories
|
|
assert {"fgsea_dot", "fgsea_curve"}.issubset(omitted_categories)
|
|
|
|
|
|
@pytest.mark.parametrize("analysis_ready_state", ["missing", "empty"])
|
|
def test_missing_or_empty_analysis_ready_table_uses_non_candidate_fallback(
|
|
tmp_path: Path,
|
|
analysis_ready_state: str,
|
|
) -> None:
|
|
workspace = tmp_path / f"job_fallback_{analysis_ready_state}"
|
|
_write_enrichment_manifests(workspace)
|
|
expected_stages = ["rnaseq_go_enrich", "rnaseq_kegg_enrich", "rnaseq_fgsea"]
|
|
if analysis_ready_state == "empty":
|
|
_write_analysis_ready_family_manifest(
|
|
workspace,
|
|
prefix="Os",
|
|
scope_id="scope-enrichment",
|
|
rows=[],
|
|
)
|
|
expected_stages.append("rnaseq_family_score")
|
|
|
|
result = generate_rnaseq_report(
|
|
workspace,
|
|
"Os",
|
|
rnaseq_scope_id="scope-enrichment",
|
|
expected_stages=expected_stages,
|
|
use_model=False,
|
|
)
|
|
formal = json.loads(result.evidence_path.read_text(encoding="utf-8"))["formal_report"]
|
|
inventory = formal["figure_inventory"]
|
|
categories = [row["category"] for row in inventory]
|
|
pathway_rows = [row for row in inventory if row["category"] == "kegg_pathway"]
|
|
|
|
assert formal["selection_mode"] == "representative_fallback"
|
|
assert {
|
|
category: categories.count(category)
|
|
for category in ("go_bar", "go_dot", "go_ontology", "kegg_bar", "kegg_dot")
|
|
} == {
|
|
"go_bar": 1,
|
|
"go_dot": 1,
|
|
"go_ontology": 1,
|
|
"kegg_bar": 1,
|
|
"kegg_dot": 1,
|
|
}
|
|
assert {(row["comparison"], row["pathway_id"]) for row in pathway_rows} == {
|
|
("IR29__6_vs_IR29__0", "ko00073"),
|
|
("IR29__6_vs_IR29__0", "ko00196"),
|
|
}
|
|
assert all(row["candidate_families"] == [] for row in inventory)
|
|
assert all(row["support_labels"] == [] for row in inventory)
|
|
assert all(
|
|
row["selection_reason"] == "representative_fallback"
|
|
for row in inventory
|
|
if row["category"]
|
|
)
|
|
assert "候选家族驱动筛选不可用" in result.report_path.read_text(encoding="utf-8")
|
|
|
|
|
|
def test_candidate_guided_selection_supports_combined_rnaseq_enrich_stage(tmp_path: Path) -> None:
|
|
workspace = tmp_path / "job_candidate_combined"
|
|
ids = _write_candidate_guided_enrichment_manifests(workspace, combined=True)
|
|
|
|
result = generate_rnaseq_report(
|
|
workspace,
|
|
ids["prefix"],
|
|
rnaseq_scope_id=ids["scope"],
|
|
expected_stages=["rnaseq_enrich", "rnaseq_fgsea", "rnaseq_family_score"],
|
|
use_model=False,
|
|
)
|
|
formal = json.loads(result.evidence_path.read_text(encoding="utf-8"))["formal_report"]
|
|
enrichment_rows = [row for row in formal["figure_inventory"] if row["category"]]
|
|
|
|
assert formal["selection_mode"] == "candidate_guided"
|
|
assert {row["category"] for row in enrichment_rows} == {
|
|
"go_bar",
|
|
"go_dot",
|
|
"go_ontology",
|
|
"kegg_bar",
|
|
"kegg_dot",
|
|
"kegg_pathway",
|
|
"fgsea_dot",
|
|
"fgsea_curve",
|
|
}
|
|
assert all(row["comparison"] == ids["candidate_comparison"] for row in enrichment_rows)
|
|
assert {
|
|
row["pathway_id"]
|
|
for row in enrichment_rows
|
|
if row["category"] == "kegg_pathway"
|
|
} == {ids["supported_pathway_a"], ids["supported_pathway_b"]}
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"support_labels",
|
|
["", "KEGG:ko99998:pathway absent from all selected comparison outputs"],
|
|
ids=["empty-support-labels", "unmatched-support-labels"],
|
|
)
|
|
def test_analysis_ready_rows_without_matchable_support_use_representative_fallback(
|
|
tmp_path: Path,
|
|
support_labels: str,
|
|
) -> None:
|
|
workspace = tmp_path / "job_analysis_ready_without_matchable_support"
|
|
ids = _write_candidate_guided_enrichment_manifests(workspace)
|
|
_write_analysis_ready_family_manifest(
|
|
workspace,
|
|
prefix=ids["prefix"],
|
|
scope_id=ids["scope"],
|
|
rows=[
|
|
(
|
|
"FAM_WITHOUT_MATCHABLE_EVIDENCE",
|
|
"family without usable evidence",
|
|
ids["candidate_comparison"],
|
|
support_labels,
|
|
)
|
|
],
|
|
)
|
|
|
|
result = generate_rnaseq_report(
|
|
workspace,
|
|
ids["prefix"],
|
|
rnaseq_scope_id=ids["scope"],
|
|
expected_stages=[
|
|
"rnaseq_go_enrich",
|
|
"rnaseq_kegg_enrich",
|
|
"rnaseq_fgsea",
|
|
"rnaseq_family_score",
|
|
],
|
|
use_model=False,
|
|
)
|
|
formal = json.loads(result.evidence_path.read_text(encoding="utf-8"))["formal_report"]
|
|
|
|
assert formal["selection_mode"] == "representative_fallback"
|
|
assert all(row["candidate_families"] == [] for row in formal["figure_inventory"])
|
|
markdown = result.report_path.read_text(encoding="utf-8")
|
|
assert "候选家族驱动筛选不可用" in markdown
|
|
assert "candidate_guided" not in markdown
|
|
|
|
|
|
def test_unmatched_fgsea_only_support_uses_fallback_and_does_not_mark_dot_candidate(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
workspace = tmp_path / "job_unmatched_fgsea_only"
|
|
ids = _write_candidate_guided_enrichment_manifests(workspace)
|
|
_write_analysis_ready_family_manifest(
|
|
workspace,
|
|
prefix=ids["prefix"],
|
|
scope_id=ids["scope"],
|
|
rows=[
|
|
(
|
|
"FAM_FGSEA_PATHWAY_ABSENT",
|
|
"family with absent fgsea pathway",
|
|
ids["candidate_comparison"],
|
|
"fgsea:ko99998:pathway absent from comparison fgsea table",
|
|
)
|
|
],
|
|
)
|
|
|
|
result = generate_rnaseq_report(
|
|
workspace,
|
|
ids["prefix"],
|
|
rnaseq_scope_id=ids["scope"],
|
|
expected_stages=[
|
|
"rnaseq_go_enrich",
|
|
"rnaseq_kegg_enrich",
|
|
"rnaseq_fgsea",
|
|
"rnaseq_family_score",
|
|
],
|
|
use_model=False,
|
|
)
|
|
formal = json.loads(result.evidence_path.read_text(encoding="utf-8"))["formal_report"]
|
|
fgsea_dots = [
|
|
row for row in formal["figure_inventory"] if row["category"] == "fgsea_dot"
|
|
]
|
|
|
|
assert formal["selection_mode"] == "representative_fallback"
|
|
assert len(fgsea_dots) == 1
|
|
assert fgsea_dots[0]["candidate_families"] == []
|
|
assert fgsea_dots[0]["support_labels"] == []
|
|
assert fgsea_dots[0]["selection_reason"] == "representative_fallback"
|
|
|
|
|
|
def test_candidate_mode_omits_fgsea_dot_when_only_fgsea_label_is_unmatched(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
workspace = tmp_path / "job_go_candidate_with_unmatched_fgsea"
|
|
ids = _write_candidate_guided_enrichment_manifests(workspace)
|
|
_write_analysis_ready_family_manifest(
|
|
workspace,
|
|
prefix=ids["prefix"],
|
|
scope_id=ids["scope"],
|
|
rows=[
|
|
(
|
|
"FAM_GO_ONLY_USABLE",
|
|
"family with usable go only",
|
|
ids["candidate_comparison"],
|
|
"GO:GO:9876543:random candidate function;"
|
|
"fgsea:ko99998:pathway absent from comparison fgsea table",
|
|
)
|
|
],
|
|
)
|
|
|
|
result = generate_rnaseq_report(
|
|
workspace,
|
|
ids["prefix"],
|
|
rnaseq_scope_id=ids["scope"],
|
|
expected_stages=[
|
|
"rnaseq_go_enrich",
|
|
"rnaseq_kegg_enrich",
|
|
"rnaseq_fgsea",
|
|
"rnaseq_family_score",
|
|
],
|
|
use_model=False,
|
|
)
|
|
formal = json.loads(result.evidence_path.read_text(encoding="utf-8"))["formal_report"]
|
|
selected = formal["figure_inventory"]
|
|
omitted = formal["omitted_figure_inventory"]
|
|
|
|
assert formal["selection_mode"] == "candidate_guided"
|
|
assert any(
|
|
row["category"] == "go_bar"
|
|
and row["candidate_families"] == ["FAM_GO_ONLY_USABLE"]
|
|
for row in selected
|
|
)
|
|
assert not any(row["category"] == "fgsea_dot" for row in selected)
|
|
assert any(
|
|
row["category"] == "fgsea_dot"
|
|
and row["comparison"] == ids["candidate_comparison"]
|
|
for row in omitted
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("query", "action"),
|
|
[
|
|
("生成rnaseq结果报告", "generate"),
|
|
("生成 RNA-seq 结果报告", "generate"),
|
|
("打开rnaseq结果报告", "open"),
|
|
("生成转录组结果报告", "generate"),
|
|
],
|
|
)
|
|
def test_explicit_rnaseq_report_intent_is_provider_independent(query: str, action: str) -> None:
|
|
parsed = parse_intent_from_query(query)
|
|
fast = _fast_intent_recognition(query)
|
|
|
|
assert parsed["stages"] == ["rnaseq_report"]
|
|
assert parsed["params"]["rnaseqReportAction"] == action
|
|
assert fast is not None
|
|
assert fast["stages"] == ["rnaseq_report"]
|
|
assert fast["params"]["rnaseqReportAction"] == action
|
|
|
|
|
|
def test_generic_report_without_rnaseq_context_is_not_stolen() -> None:
|
|
parsed = parse_intent_from_query("生成结果报告")
|
|
assert parsed["stages"] != ["rnaseq_report"]
|
|
|
|
|
|
def test_explicit_report_chat_intent_does_not_require_router_model() -> None:
|
|
assert hasattr(intent_parser_module, "deterministic_agent_chat_intent")
|
|
parser = intent_parser_module.deterministic_agent_chat_intent
|
|
|
|
opened = parser("打开rnaseq结果报告")
|
|
generated = parser("生成 RNA-seq 结果报告")
|
|
|
|
assert opened["stages"] == ["rnaseq_report"]
|
|
assert opened["params"]["rnaseqReportAction"] == "open"
|
|
assert generated["params"]["rnaseqReportAction"] == "generate"
|
|
assert parser("解释一下差异表达结果") is None
|
|
|
|
|
|
def test_generic_report_chat_intent_requires_rnaseq_job_context() -> None:
|
|
parser = intent_parser_module.deterministic_agent_chat_intent
|
|
|
|
assert parser("生成结果报告") is None
|
|
generated = parser("生成结果报告", rnaseq_context=True)
|
|
opened = parser("打开结果报告", rnaseq_context=True)
|
|
|
|
assert generated["stages"] == ["rnaseq_report"]
|
|
assert generated["params"]["rnaseqReportAction"] == "generate"
|
|
assert opened["params"]["rnaseqReportAction"] == "open"
|