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", ""), "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", ""), "go_b_svg": ("Os.rnaseq_go_enrich.BB.GO_enrich.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", ""), "kegg_b_svg": ("Os.rnaseq_kegg_enrich.BB.KEGG_enrich.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"