845 lines
34 KiB
Python
845 lines
34 KiB
Python
from __future__ import annotations
|
||
|
||
import json
|
||
from pathlib import Path
|
||
import re
|
||
|
||
import pytest
|
||
|
||
import gfa_core.gene_family_report as report_module
|
||
from gfa_core.gene_family_report import (
|
||
NoSuccessfulGeneFamilyEvidence,
|
||
generate_gene_family_report,
|
||
)
|
||
|
||
|
||
def _write_manifest(
|
||
workspace: Path,
|
||
*,
|
||
prefix: str,
|
||
stage: str,
|
||
status: str = "success",
|
||
scope_id: str = "scope-a",
|
||
analysis_route: str = "post_subfamily_confirm",
|
||
member_set_id: str = "members-a",
|
||
metric_value: int = 1,
|
||
) -> Path:
|
||
final_dir = workspace / "final"
|
||
final_dir.mkdir(parents=True, exist_ok=True)
|
||
token = f"{scope_id}.{analysis_route}.{member_set_id}"
|
||
table_path = final_dir / f"{prefix}.{token}.{stage}.tsv"
|
||
table_path.write_text(f"gene\tvalue\n{scope_id}_gene\t{metric_value}\n", encoding="utf-8")
|
||
manifest_path = final_dir / f"{prefix}.{token}.{stage}.manifest.json"
|
||
manifest = {
|
||
"manifest_version": "1.0",
|
||
"stage": stage,
|
||
"status": status,
|
||
"prefix": prefix,
|
||
"scope_id": scope_id,
|
||
"analysis_route": analysis_route,
|
||
"member_set_id": member_set_id,
|
||
"parameters": {
|
||
"scope_id": scope_id,
|
||
"analysis_route": analysis_route,
|
||
"member_set_id": member_set_id,
|
||
},
|
||
"metrics": {"member_count": metric_value},
|
||
"artifacts": {
|
||
"result_table": {
|
||
"path": f"final/{table_path.name}",
|
||
"full_path": str(table_path),
|
||
"type": "table",
|
||
}
|
||
},
|
||
}
|
||
manifest_path.write_text(json.dumps(manifest, ensure_ascii=False), encoding="utf-8")
|
||
return manifest_path
|
||
|
||
|
||
def _generate(
|
||
workspace: Path,
|
||
*,
|
||
expected_stages: list[str],
|
||
stage_statuses: dict[str, str],
|
||
use_model: bool = False,
|
||
action: str = "generate",
|
||
):
|
||
return generate_gene_family_report(
|
||
workspace,
|
||
"Ft.SPL",
|
||
expected_stages=expected_stages,
|
||
stage_statuses=stage_statuses,
|
||
scope_id="scope-a",
|
||
analysis_route="post_subfamily_confirm",
|
||
member_set_id="members-a",
|
||
use_model=use_model,
|
||
action=action,
|
||
job_id="job-report",
|
||
)
|
||
|
||
|
||
def _stage_section(markdown: str, stage: str) -> str:
|
||
heading = re.search(rf"(?m)^## (?!#).*?\(`{re.escape(stage)}`\)", markdown)
|
||
if heading is None:
|
||
raise AssertionError(f"missing report chapter for {stage}")
|
||
next_heading = re.search(r"(?m)^## (?!#)", markdown[heading.end():])
|
||
end = heading.end() + next_heading.start() if next_heading else len(markdown)
|
||
return markdown[heading.start():end]
|
||
|
||
|
||
def _subsection(markdown: str, number: str) -> str:
|
||
heading = re.search(rf"(?m)^### {re.escape(number)}(?:\s|$)", markdown)
|
||
if heading is None:
|
||
raise AssertionError(f"missing report subsection {number}")
|
||
next_heading = re.search(r"(?m)^#{2,3} ", markdown[heading.end():])
|
||
end = heading.end() + next_heading.start() if next_heading else len(markdown)
|
||
return markdown[heading.start():end]
|
||
|
||
|
||
def _valid_api_arrangement(payload: dict, *, reverse: bool = False) -> dict:
|
||
sections = []
|
||
for section in payload["sections"]:
|
||
sentence_ids = [sentence["sentence_id"] for sentence in section["sentences"]]
|
||
if reverse:
|
||
sentence_ids.reverse()
|
||
sections.append(
|
||
{
|
||
"section_id": section["section_id"],
|
||
"paragraphs": [
|
||
{
|
||
"sentence_ids": sentence_ids,
|
||
"connector_ids": ["also"] * max(0, len(sentence_ids) - 1),
|
||
}
|
||
],
|
||
}
|
||
)
|
||
return {"sections": sections}
|
||
|
||
|
||
def test_report_uses_only_planned_stages_and_current_identity(tmp_path: Path) -> None:
|
||
workspace = tmp_path / "job"
|
||
_write_manifest(workspace, prefix="Ft.SPL", stage="identify", metric_value=3)
|
||
other = _write_manifest(
|
||
workspace,
|
||
prefix="Ft.SPL",
|
||
stage="identify",
|
||
scope_id="scope-b",
|
||
analysis_route="target_only",
|
||
member_set_id="members-b",
|
||
metric_value=99,
|
||
)
|
||
|
||
result = _generate(
|
||
workspace,
|
||
expected_stages=["identify"],
|
||
stage_statuses={"identify": "success"},
|
||
)
|
||
|
||
report_text = result.report_path.read_text(encoding="utf-8")
|
||
assert "最终成员数：3" in report_text
|
||
assert other.name not in report_text
|
||
generated = {path.name for path in (workspace / "final").glob("Ft.SPL.gene_family_report.*")}
|
||
assert generated == {
|
||
"Ft.SPL.gene_family_report.md",
|
||
"Ft.SPL.gene_family_report.manifest.json",
|
||
}
|
||
|
||
|
||
def test_report_renders_template_subsections_for_only_planned_stages(tmp_path: Path) -> None:
|
||
workspace = tmp_path / "job"
|
||
for stage in ("identify", "motif", "pep"):
|
||
_write_manifest(workspace, prefix="Ft.SPL", stage=stage)
|
||
|
||
result = _generate(
|
||
workspace,
|
||
expected_stages=["identify", "motif", "pep"],
|
||
stage_statuses={"identify": "success", "motif": "success", "pep": "success"},
|
||
)
|
||
|
||
markdown = result.report_path.read_text(encoding="utf-8")
|
||
expected_subsections = {
|
||
"identify": "### 4.3 最终成员数量与命名",
|
||
"motif": "### 7.2 Motif 数量与出现频率",
|
||
"pep": "### 16.1 蛋白理化性质分布图",
|
||
}
|
||
for stage, subsection in expected_subsections.items():
|
||
assert subsection in _stage_section(markdown, stage)
|
||
assert "(`synteny`)" not in markdown
|
||
assert "【填写】" not in markdown
|
||
|
||
|
||
def test_report_renders_nested_motif_metrics_without_placeholder_values(tmp_path: Path) -> None:
|
||
workspace = tmp_path / "job"
|
||
manifest_path = _write_manifest(workspace, prefix="Ft.SPL", stage="motif")
|
||
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
|
||
manifest["metrics"] = {
|
||
"motif_count": 10,
|
||
"total_sites": 105,
|
||
"motifs": [
|
||
{"width": width}
|
||
for width in (78, 76, 80, 113, 57, 29, 52, 138, 27, 29)
|
||
],
|
||
}
|
||
manifest_path.write_text(json.dumps(manifest, ensure_ascii=False), encoding="utf-8")
|
||
|
||
result = _generate(
|
||
workspace,
|
||
expected_stages=["motif"],
|
||
stage_statuses={"motif": "success"},
|
||
)
|
||
|
||
motif_section = _stage_section(result.report_path.read_text(encoding="utf-8"), "motif")
|
||
assert re.search(r"识别\s*10\s*个\s*motif", motif_section)
|
||
assert re.search(r"总位点数为\s*105", motif_section)
|
||
assert re.search(r"宽度范围为\s*27–138", motif_section)
|
||
assert "未解析" not in motif_section
|
||
assert "N/A" not in motif_section
|
||
assert "None" not in motif_section
|
||
|
||
|
||
def test_identify_uses_real_manifest_member_count_aliases(tmp_path: Path) -> None:
|
||
workspace = tmp_path / "job"
|
||
manifest_path = _write_manifest(workspace, prefix="Ft.SPL", stage="identify")
|
||
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
|
||
manifest["metrics"] = {"identified_members_count": 24, "final_members": 24}
|
||
manifest_path.write_text(json.dumps(manifest, ensure_ascii=False), encoding="utf-8")
|
||
|
||
result = _generate(
|
||
workspace,
|
||
expected_stages=["identify"],
|
||
stage_statuses={"identify": "success"},
|
||
)
|
||
|
||
section = _stage_section(result.report_path.read_text(encoding="utf-8"), "identify")
|
||
assert "最终成员数：24" in section
|
||
assert re.search(r"形成\s*24\s*个\s*SPL家族", section)
|
||
assert "源结果未提供此指标" not in section
|
||
|
||
|
||
def test_phylogeny_renders_real_manifest_schema_with_explicit_labels(tmp_path: Path) -> None:
|
||
workspace = tmp_path / "job"
|
||
manifest_path = _write_manifest(workspace, prefix="Ft.SPL", stage="phylogeny")
|
||
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
|
||
manifest["metrics"] = {"sequences_count": 40, "msa_length": 1449}
|
||
manifest_path.write_text(json.dumps(manifest, ensure_ascii=False), encoding="utf-8")
|
||
|
||
result = _generate(
|
||
workspace,
|
||
expected_stages=["phylogeny"],
|
||
stage_statuses={"phylogeny": "success"},
|
||
)
|
||
|
||
section = _stage_section(result.report_path.read_text(encoding="utf-8"), "phylogeny")
|
||
assert "序列数：40" in section
|
||
assert "多序列比对长度：1449" in section
|
||
assert re.search(r"使用\s*40\s*条序列.*比对长度为\s*1449", section)
|
||
|
||
|
||
def test_interspecies_synteny_reads_only_named_real_schema_artifacts(tmp_path: Path) -> None:
|
||
workspace = tmp_path / "job"
|
||
manifest_path = _write_manifest(workspace, prefix="Ft.SPL", stage="interspecies_synteny")
|
||
final_dir = workspace / "final"
|
||
gene_to_species = final_dir / "Ft.SPL.gene_to_species.tsv"
|
||
gene_to_species.write_text(
|
||
"gene_id\tspecies\tgroup\n"
|
||
"FtGene1.01.T01\tFt\ttarget\n"
|
||
"FtGene2.01.T01\tFt\ttarget\n"
|
||
"AT1G00010\tAT\tref\n"
|
||
"AT1G00020\tAT\tref\n",
|
||
encoding="utf-8",
|
||
)
|
||
collinearity = final_dir / "Ft.At.collinearity"
|
||
lines = []
|
||
for index in range(521):
|
||
lines.append(f"## Alignment {index}: score=1 e_value=1e-10 N=1 Ft&At plus")
|
||
if index == 0:
|
||
lines.append(" 0- 0:\tAT1G00010.1\tFtGene1.01.T01\t1e-20")
|
||
elif index == 1:
|
||
lines.append(" 1- 0:\tAT1G00020.1\tFtGene2.01.T01\t1e-20")
|
||
collinearity.write_text("\n".join(lines) + "\n", encoding="utf-8")
|
||
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
|
||
manifest["metrics"] = {"pair_success_count": 1, "pair_total_count": 1}
|
||
manifest["artifacts"] = {
|
||
"gene_to_species": {"path": f"final/{gene_to_species.name}", "type": "tsv"},
|
||
"Ft.At_collinearity": {"path": f"final/{collinearity.name}", "type": "data"},
|
||
}
|
||
manifest_path.write_text(json.dumps(manifest, ensure_ascii=False), encoding="utf-8")
|
||
|
||
result = _generate(
|
||
workspace,
|
||
expected_stages=["interspecies_synteny"],
|
||
stage_statuses={"interspecies_synteny": "success"},
|
||
)
|
||
|
||
section = _stage_section(result.report_path.read_text(encoding="utf-8"), "interspecies_synteny")
|
||
assert "共线性区块数：521" in section
|
||
assert "家族成员目标—参考配对数：2" in section
|
||
assert re.search(r"全基因组\s*block\s*521\s*个.*目标-参考家族.*?2\s*对", section)
|
||
|
||
|
||
def test_pep_renders_real_manifest_task_summary_with_explicit_labels(tmp_path: Path) -> None:
|
||
workspace = tmp_path / "job"
|
||
manifest_path = _write_manifest(workspace, prefix="Ft.SPL", stage="pep")
|
||
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
|
||
manifest["metrics"] = {"protein_count": 40, "tasks_success": 3, "tasks_total": 3}
|
||
manifest_path.write_text(json.dumps(manifest, ensure_ascii=False), encoding="utf-8")
|
||
|
||
result = _generate(
|
||
workspace,
|
||
expected_stages=["pep"],
|
||
stage_statuses={"pep": "success"},
|
||
)
|
||
|
||
section = _stage_section(result.report_path.read_text(encoding="utf-8"), "pep")
|
||
assert "蛋白数：40" in section
|
||
assert "子任务成功数：3/3" in section
|
||
assert re.search(r"覆盖\s*40\s*个蛋白，子任务成功数为\s*3/3", section)
|
||
|
||
|
||
def test_report_has_no_empty_third_level_subsections_or_repeated_generic_interpretation(
|
||
tmp_path: Path,
|
||
) -> None:
|
||
workspace = tmp_path / "job"
|
||
for stage in ("identify", "phylogeny", "motif"):
|
||
manifest_path = _write_manifest(workspace, prefix="Ft.SPL", stage=stage)
|
||
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
|
||
if stage == "phylogeny":
|
||
manifest["metrics"] = {"sequences_count": 40, "msa_length": 1449}
|
||
elif stage == "motif":
|
||
manifest["metrics"] = {
|
||
"sequence_count": 40,
|
||
"motif_count": 10,
|
||
"total_sites": 105,
|
||
"motifs": [{"width": 27}, {"width": 138}],
|
||
}
|
||
manifest_path.write_text(json.dumps(manifest, ensure_ascii=False), encoding="utf-8")
|
||
|
||
result = _generate(
|
||
workspace,
|
||
expected_stages=["identify", "phylogeny", "motif"],
|
||
stage_statuses={stage: "success" for stage in ("identify", "phylogeny", "motif")},
|
||
)
|
||
|
||
markdown = result.report_path.read_text(encoding="utf-8")
|
||
for stage in ("identify", "phylogeny", "motif"):
|
||
section = _stage_section(markdown, stage)
|
||
chunks = re.split(r"(?m)(?=^### )", section)[1:]
|
||
assert chunks
|
||
for chunk in chunks:
|
||
body = "\n".join(chunk.splitlines()[1:]).strip()
|
||
assert body, f"empty subsection: {chunk.splitlines()[0]}"
|
||
assert markdown.count("本节解释仅限上述可追溯结果，不据此断言真实功能或因果关系。") <= 1
|
||
assert markdown.count("结果范围受当前输入、参数和源文件完整性限制。") <= 1
|
||
|
||
|
||
def test_front_summary_contains_project_numbers_in_six_to_ten_sentences(tmp_path: Path) -> None:
|
||
workspace = tmp_path / "job"
|
||
identify_path = _write_manifest(workspace, prefix="Ft.SPL", stage="identify")
|
||
identify = json.loads(identify_path.read_text(encoding="utf-8"))
|
||
identify["metrics"] = {"identified_members_count": 24, "final_members": 24}
|
||
identify_path.write_text(json.dumps(identify, ensure_ascii=False), encoding="utf-8")
|
||
phylogeny_path = _write_manifest(workspace, prefix="Ft.SPL", stage="phylogeny")
|
||
phylogeny = json.loads(phylogeny_path.read_text(encoding="utf-8"))
|
||
phylogeny["metrics"] = {"sequences_count": 40, "msa_length": 1449}
|
||
phylogeny_path.write_text(json.dumps(phylogeny, ensure_ascii=False), encoding="utf-8")
|
||
|
||
result = _generate(
|
||
workspace,
|
||
expected_stages=["identify", "phylogeny"],
|
||
stage_statuses={"identify": "success", "phylogeny": "success"},
|
||
)
|
||
|
||
summary = _subsection(result.report_path.read_text(encoding="utf-8"), "1.3")
|
||
assert re.search(r"目标物种.*24.*家族成员", summary)
|
||
assert re.search(r"目标.*参考.*40.*1449", summary)
|
||
assert 6 <= summary.count("。") <= 10
|
||
|
||
|
||
def test_stage_specific_key_facts_land_in_their_semantic_subsections(tmp_path: Path) -> None:
|
||
workspace = tmp_path / "job"
|
||
final_dir = workspace / "final"
|
||
stages = [
|
||
"reference_classification", "domain", "structure", "combo",
|
||
"promoter", "tfbs", "synteny",
|
||
]
|
||
manifests = {stage: _write_manifest(workspace, prefix="Ft.SPL", stage=stage) for stage in stages}
|
||
|
||
assignment = final_dir / "Ft.SPL.reference_classification.assignment.tsv"
|
||
assignment.write_text(
|
||
"target_gene\tassigned_subfamily\tconfidence\n"
|
||
+ "\n".join(
|
||
f"FtGene{i}\t{'I' if i < 2 else 'II'}\t{confidence}"
|
||
for i, confidence in enumerate(("high", "high", "medium", "low"), start=1)
|
||
)
|
||
+ "\n",
|
||
encoding="utf-8",
|
||
)
|
||
reference = json.loads(manifests["reference_classification"].read_text(encoding="utf-8"))
|
||
reference["metrics"] = {"target_gene_count": 4, "subfamily_rows": 6}
|
||
reference["artifacts"] = {"assignment_table": {"path": f"final/{assignment.name}", "type": "tsv"}}
|
||
manifests["reference_classification"].write_text(json.dumps(reference), encoding="utf-8")
|
||
|
||
domain = json.loads(manifests["domain"].read_text(encoding="utf-8"))
|
||
domain["metrics"] = {
|
||
"proteins_total": 40, "proteins_with_domains": 40, "domain_coverage_rate": 1.0,
|
||
"top_domains": [["SBP", 40], ["Ig_SPL7", 12]],
|
||
}
|
||
manifests["domain"].write_text(json.dumps(domain), encoding="utf-8")
|
||
|
||
structure = json.loads(manifests["structure"].read_text(encoding="utf-8"))
|
||
structure["metrics"] = {
|
||
"genes_total": 40, "genes_with_structure": 24, "structure_coverage_rate": 0.6,
|
||
"long_gene_outlier_count": 1,
|
||
"long_gene_outliers": [{"gene_id": "FtLong1", "length_bp": 12454, "cds_count": 10}],
|
||
}
|
||
manifests["structure"].write_text(json.dumps(structure), encoding="utf-8")
|
||
|
||
combo = json.loads(manifests["combo"].read_text(encoding="utf-8"))
|
||
combo["metrics"] = None
|
||
combo["parameters"] = {"panels": ["motif", "domain", "structure"]}
|
||
combo["artifacts"] = {
|
||
"combo_plot_png": {"path": "final/Ft.SPL.combo.png", "type": "image"},
|
||
"combo_plot_pdf": {"path": "final/Ft.SPL.combo.pdf", "type": "image"},
|
||
}
|
||
manifests["combo"].write_text(json.dumps(combo), encoding="utf-8")
|
||
|
||
element_counts = final_dir / "Ft.SPL.element_counts.txt"
|
||
element_counts.write_text("1460\tTATA-box\n852\tCAAT-box\n72\tMYB\n", encoding="utf-8")
|
||
category_summary = final_dir / "Ft.SPL.promoter_category_summary.tsv"
|
||
category_summary.write_text("big7\telement_instances\nCore promoter\t2600\nLight response\t188\n", encoding="utf-8")
|
||
enrichment = final_dir / "Ft.SPL.enrichment_analysis.txt"
|
||
enrichment.write_text(
|
||
"Subfamily\tElement\tCount\tSubfamilyTotal\tGlobalCount\tEnrichmentRatio\nI\tAACA_motif\t1\t222\t1\t16.66\n",
|
||
encoding="utf-8",
|
||
)
|
||
cooccurrence = final_dir / "Ft.SPL.cooccurrence_analysis.txt"
|
||
cooccurrence.write_text(
|
||
"Element1\tElement2\tCo-occurrenceCount\tGenes\nCAAT-box\tMYB\t23\tFtGene1,FtGene2\n",
|
||
encoding="utf-8",
|
||
)
|
||
complexity = final_dir / "Ft.SPL.complexity_analysis.txt"
|
||
complexity.write_text(
|
||
"Gene\tTotalElements\tUniqueElements\tComplexityScore\tComplexityClass\nFtComplex1\t278\t33\t361.44\tVery High\n",
|
||
encoding="utf-8",
|
||
)
|
||
promoter = json.loads(manifests["promoter"].read_text(encoding="utf-8"))
|
||
promoter["metrics"] = {"total_elements": 3699, "unique_elements": 87, "genes_with_elements": 23}
|
||
promoter["artifacts"] = {
|
||
"element_counts": {"path": f"final/{element_counts.name}"},
|
||
"viz_promoter_position_distribution_summary": {"path": f"final/{category_summary.name}"},
|
||
"enrichment_analysis": {"path": f"final/{enrichment.name}"},
|
||
"cooccurrence_analysis": {"path": f"final/{cooccurrence.name}"},
|
||
"complexity_analysis": {"path": f"final/{complexity.name}"},
|
||
}
|
||
manifests["promoter"].write_text(json.dumps(promoter), encoding="utf-8")
|
||
|
||
motif_summary = final_dir / "Ft.SPL.tfbs.motif_summary.tsv"
|
||
motif_summary.write_text(
|
||
"motif_id\tmotif_alt_id\tfamily_sites\tfamily_genes_with_site\tbackground_sites\tbackground_genes_with_site\n"
|
||
"MOTIF_A\t\t153\t18\t0\t0\n",
|
||
encoding="utf-8",
|
||
)
|
||
tfbs = json.loads(manifests["tfbs"].read_text(encoding="utf-8"))
|
||
tfbs["metrics"] = {
|
||
"family_promoter_count": 24, "background_promoter_count": 0,
|
||
"family_site_count": 3734, "family_motif_count": 230, "motif_database": "PlantTFDB",
|
||
}
|
||
tfbs["artifacts"] = {"motif_summary": {"path": f"final/{motif_summary.name}"}}
|
||
manifests["tfbs"].write_text(json.dumps(tfbs), encoding="utf-8")
|
||
|
||
synteny = json.loads(manifests["synteny"].read_text(encoding="utf-8"))
|
||
synteny["metrics"] = {"gene_count": 31679, "synteny_blocks": 316, "gene_pairs": 3267, "circos_chromosomes_plotted": 8}
|
||
manifests["synteny"].write_text(json.dumps(synteny), encoding="utf-8")
|
||
|
||
result = _generate(
|
||
workspace,
|
||
expected_stages=stages,
|
||
stage_statuses={stage: "success" for stage in stages},
|
||
)
|
||
markdown = result.report_path.read_text(encoding="utf-8")
|
||
assert re.search(r"4\s*个目标成员.*2\s*个小组.*高信心\s*2", _subsection(markdown, "6.2"))
|
||
assert re.search(r"40/40.*SBP.*40.*Ig_SPL7.*12", _subsection(markdown, "8.1"))
|
||
assert re.search(r"24/40.*FtLong1.*12454", _subsection(markdown, "9.1"))
|
||
assert "Motif、Domain、Structure" in _subsection(markdown, "10.1")
|
||
promoter_text = _subsection(markdown, "11.1")
|
||
assert re.search(r"23\s*个目标基因.*3699.*TATA-box.*1460", promoter_text)
|
||
assert "FtComplex1" in promoter_text and "CAAT-box—MYB" in promoter_text
|
||
assert re.search(r"24\s*条目标启动子.*3734.*MOTIF_A.*153", _subsection(markdown, "12.1"))
|
||
assert re.search(r"背景启动子为\s*0.*不能.*富集", _subsection(markdown, "12.1"))
|
||
assert re.search(r"全基因组.*316.*3267.*8\s*条染色体", _subsection(markdown, "13.1"))
|
||
assert "源结果未提供家族专属复制基因对" in _subsection(markdown, "13.1")
|
||
|
||
|
||
def test_all_planned_success_is_final(tmp_path: Path) -> None:
|
||
workspace = tmp_path / "job"
|
||
_write_manifest(workspace, prefix="Ft.SPL", stage="identify")
|
||
_write_manifest(workspace, prefix="Ft.SPL", stage="motif")
|
||
|
||
result = _generate(
|
||
workspace,
|
||
expected_stages=["identify", "motif"],
|
||
stage_statuses={"identify": "success", "motif": "success"},
|
||
)
|
||
|
||
manifest = json.loads(result.manifest_path.read_text(encoding="utf-8"))
|
||
assert result.report_status == "final"
|
||
assert manifest["metrics"]["report_status"] == "final"
|
||
assert manifest["metrics"]["included_stages"] == ["identify", "motif"]
|
||
|
||
|
||
def test_unsupported_successful_stage_is_visible_but_not_included_or_final(tmp_path: Path) -> None:
|
||
workspace = tmp_path / "job"
|
||
_write_manifest(workspace, prefix="Ft.SPL", stage="identify")
|
||
_write_manifest(workspace, prefix="Ft.SPL", stage="ppi")
|
||
|
||
result = _generate(
|
||
workspace,
|
||
expected_stages=["identify", "ppi"],
|
||
stage_statuses={"identify": "success", "ppi": "success"},
|
||
)
|
||
|
||
markdown = result.report_path.read_text(encoding="utf-8")
|
||
manifest = json.loads(result.manifest_path.read_text(encoding="utf-8"))
|
||
assert result.report_status == "partial"
|
||
assert result.included_stages == ("identify",)
|
||
assert "ppi" in markdown
|
||
assert "不支持" in markdown
|
||
assert "(`ppi`)" not in markdown
|
||
assert manifest["metrics"]["included_stages"] == ["identify"]
|
||
assert manifest["metrics"]["unsupported_stages"] == ["ppi"]
|
||
|
||
|
||
@pytest.mark.parametrize("expected_stages", [None, []])
|
||
def test_generate_rejects_missing_plan_instead_of_inferring_historical_manifests(
|
||
tmp_path: Path, expected_stages: list[str] | None
|
||
) -> None:
|
||
workspace = tmp_path / "job"
|
||
_write_manifest(workspace, prefix="Ft.SPL", stage="identify")
|
||
|
||
with pytest.raises(ValueError, match="expected_stages"):
|
||
generate_gene_family_report(
|
||
workspace,
|
||
"Ft.SPL",
|
||
expected_stages=expected_stages,
|
||
stage_statuses={"identify": "success"},
|
||
scope_id="scope-a",
|
||
analysis_route="post_subfamily_confirm",
|
||
member_set_id="members-a",
|
||
use_model=False,
|
||
action="generate",
|
||
)
|
||
|
||
assert not list((workspace / "final").glob("*.gene_family_report.*"))
|
||
|
||
|
||
def test_report_selects_key_facts_without_dumping_manifest_or_stale_paths(tmp_path: Path) -> None:
|
||
workspace = tmp_path / "job"
|
||
manifest_path = _write_manifest(workspace, prefix="Ft.SPL", stage="identify", metric_value=3)
|
||
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
|
||
manifest["metrics"].update({"total_proteins": 31679, "blast_enabled": True})
|
||
manifest["artifacts"]["result_table"]["path"] = "/home/xzq/old/job/final/Ft.SPL.identify.tsv"
|
||
manifest["artifacts"]["result_table"]["full_path"] = "/home/xzq/old/job/final/Ft.SPL.identify.tsv"
|
||
manifest_path.write_text(json.dumps(manifest, ensure_ascii=False), encoding="utf-8")
|
||
|
||
result = _generate(
|
||
workspace,
|
||
expected_stages=["identify"],
|
||
stage_statuses={"identify": "success"},
|
||
)
|
||
|
||
markdown = result.report_path.read_text(encoding="utf-8")
|
||
assert "最终成员数：3" in _subsection(markdown, "4.3")
|
||
assert not re.search(r"(?m)^- `(?:member_count|total_proteins|blast_enabled)\s*`:", markdown)
|
||
assert "/home/xzq/" not in markdown
|
||
|
||
|
||
def test_fallback_uses_nested_karyotype_metrics(tmp_path: Path) -> None:
|
||
workspace = tmp_path / "job"
|
||
manifest_path = _write_manifest(workspace, prefix="Ft.SPL", stage="karyotype")
|
||
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
|
||
manifest["metrics"] = {
|
||
"species": {
|
||
"Ft": {
|
||
"gene_count": 24,
|
||
"chromosome_count": 8,
|
||
"chromosomes": [f"Ft{i}" for i in range(1, 9)],
|
||
"tandem_count": 0,
|
||
}
|
||
}
|
||
}
|
||
manifest_path.write_text(json.dumps(manifest, ensure_ascii=False), encoding="utf-8")
|
||
|
||
result = _generate(
|
||
workspace,
|
||
expected_stages=["karyotype"],
|
||
stage_statuses={"karyotype": "success"},
|
||
)
|
||
|
||
markdown = result.report_path.read_text(encoding="utf-8")
|
||
assert "24个家族成员定位到8条染色体" in markdown
|
||
assert "核型分析将N/A" not in markdown
|
||
|
||
|
||
def test_artifact_path_remaps_remote_job_to_current_mount(tmp_path: Path) -> None:
|
||
from gfa_core.mas.agents.interpretation_evidence import _artifact_path
|
||
|
||
final_dir = tmp_path / "job" / "final"
|
||
final_dir.mkdir(parents=True)
|
||
manifest_path = final_dir / "Ft.SPL.stage.manifest.json"
|
||
local_artifact = final_dir / "Ft.SPL.result.tsv"
|
||
local_artifact.write_text("gene\tvalue\nFtSPL1\t1\n", encoding="utf-8")
|
||
manifest = {
|
||
"_manifest_path": str(manifest_path),
|
||
"artifacts": {
|
||
"result": {
|
||
"full_path": "/home/xzq/old/PlantAI-dev/workspace/job-old/final/Ft.SPL.result.tsv",
|
||
"path": "/home/xzq/old/PlantAI-dev/workspace/job-old/final/Ft.SPL.result.tsv",
|
||
}
|
||
},
|
||
}
|
||
|
||
assert _artifact_path(manifest, "result") == local_artifact
|
||
|
||
|
||
def test_failed_or_missing_planned_stage_is_partial(tmp_path: Path) -> None:
|
||
workspace = tmp_path / "job"
|
||
_write_manifest(workspace, prefix="Ft.SPL", stage="identify")
|
||
_write_manifest(workspace, prefix="Ft.SPL", stage="motif", status="failed")
|
||
|
||
result = _generate(
|
||
workspace,
|
||
expected_stages=["identify", "motif"],
|
||
stage_statuses={"identify": "success", "motif": "failed"},
|
||
)
|
||
|
||
markdown = result.report_path.read_text(encoding="utf-8")
|
||
manifest = json.loads(result.manifest_path.read_text(encoding="utf-8"))
|
||
assert result.report_status == "partial"
|
||
assert "motif" in markdown.lower()
|
||
assert "failed" in markdown.lower()
|
||
assert manifest["metrics"]["failed_or_missing_stages"] == ["motif"]
|
||
|
||
|
||
def test_model_failure_creates_labeled_fallback_report(
|
||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
workspace = tmp_path / "job"
|
||
_write_manifest(workspace, prefix="Ft.SPL", stage="identify")
|
||
|
||
def fail_model(_evidence: dict, _stages: list[str]):
|
||
raise TimeoutError("provider timeout")
|
||
|
||
monkeypatch.setattr(report_module, "_generate_api_narrative", fail_model)
|
||
result = _generate(
|
||
workspace,
|
||
expected_stages=["identify"],
|
||
stage_statuses={"identify": "success"},
|
||
use_model=True,
|
||
)
|
||
|
||
markdown = result.report_path.read_text(encoding="utf-8")
|
||
manifest = json.loads(result.manifest_path.read_text(encoding="utf-8"))
|
||
assert result.generation_mode == "fallback_script"
|
||
assert "生成模式：候补脚本" in markdown
|
||
assert manifest["metrics"]["generation_mode"] == "fallback_script"
|
||
assert manifest["model"]["reason"] == "api_unavailable_or_boundary_violation"
|
||
assert "{'action':" not in markdown
|
||
|
||
|
||
def test_valid_model_sections_create_api_report(
|
||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
workspace = tmp_path / "job"
|
||
_write_manifest(workspace, prefix="Ft.SPL", stage="identify")
|
||
|
||
seen_payload: dict = {}
|
||
|
||
def valid_model(payload: dict, _stages: list[str]):
|
||
seen_payload.update(payload)
|
||
return (
|
||
_valid_api_arrangement(payload, reverse=True),
|
||
{"status": "generated", "provider": "test", "model": "test-model"},
|
||
)
|
||
|
||
monkeypatch.setattr(report_module, "_generate_api_narrative", valid_model)
|
||
result = _generate(
|
||
workspace,
|
||
expected_stages=["identify"],
|
||
stage_statuses={"identify": "success"},
|
||
use_model=True,
|
||
)
|
||
|
||
markdown = result.report_path.read_text(encoding="utf-8")
|
||
manifest = json.loads(result.manifest_path.read_text(encoding="utf-8"))
|
||
assert result.generation_mode == "api"
|
||
assert set(seen_payload) == {"sections", "allowed_connectors"}
|
||
assert all(set(section) == {"section_id", "sentences"} for section in seen_payload["sections"])
|
||
assert all(
|
||
set(sentence) == {"sentence_id", "text"}
|
||
for section in seen_payload["sections"]
|
||
for sentence in section["sentences"]
|
||
)
|
||
assert "最终成员数：1" in markdown
|
||
assert manifest["metrics"]["generation_mode"] == "api"
|
||
assert "【填写】" not in markdown
|
||
|
||
|
||
def test_unknown_model_sentence_id_falls_back_for_the_whole_report(
|
||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
workspace = tmp_path / "job"
|
||
_write_manifest(workspace, prefix="Ft.SPL", stage="identify", metric_value=3)
|
||
|
||
def invalid_model(payload: dict, _stages: list[str]):
|
||
response = _valid_api_arrangement(payload)
|
||
response["sections"][0]["paragraphs"][0]["sentence_ids"][0] = "unknown.sentence"
|
||
return response, {"status": "generated"}
|
||
|
||
monkeypatch.setattr(report_module, "_generate_api_narrative", invalid_model)
|
||
result = _generate(
|
||
workspace,
|
||
expected_stages=["identify"],
|
||
stage_statuses={"identify": "success"},
|
||
use_model=True,
|
||
)
|
||
|
||
markdown = result.report_path.read_text(encoding="utf-8")
|
||
manifest = json.loads(result.manifest_path.read_text(encoding="utf-8"))
|
||
assert result.generation_mode == "fallback_script"
|
||
assert "最终成员数：3" in markdown
|
||
assert manifest["model"]["reason"] == "api_unavailable_or_boundary_violation"
|
||
|
||
|
||
def test_missing_and_duplicate_model_sentence_ids_fall_back_for_the_whole_report(
|
||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
workspace = tmp_path / "job"
|
||
_write_manifest(workspace, prefix="Ft.SPL", stage="identify", metric_value=3)
|
||
|
||
def invalid_model(payload: dict, _stages: list[str]):
|
||
response = _valid_api_arrangement(payload)
|
||
paragraph = next(
|
||
item["paragraphs"][0]
|
||
for item in response["sections"]
|
||
if len(item["paragraphs"][0]["sentence_ids"]) >= 2
|
||
)
|
||
paragraph["sentence_ids"][-1] = paragraph["sentence_ids"][0]
|
||
return response, {"status": "generated"}
|
||
|
||
monkeypatch.setattr(report_module, "_generate_api_narrative", invalid_model)
|
||
result = _generate(
|
||
workspace,
|
||
expected_stages=["identify"],
|
||
stage_statuses={"identify": "success"},
|
||
use_model=True,
|
||
)
|
||
|
||
markdown = result.report_path.read_text(encoding="utf-8")
|
||
manifest = json.loads(result.manifest_path.read_text(encoding="utf-8"))
|
||
assert result.generation_mode == "fallback_script"
|
||
assert "最终成员数：3" in markdown
|
||
assert manifest["model"]["reason"] == "api_unavailable_or_boundary_violation"
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"violation",
|
||
["unknown_connector", "extra_field", "free_text_field", "changed_section"],
|
||
)
|
||
def test_model_schema_or_connector_boundary_violation_falls_back_for_the_whole_report(
|
||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch, violation: str
|
||
) -> None:
|
||
workspace = tmp_path / "job"
|
||
_write_manifest(workspace, prefix="Ft.SPL", stage="identify", metric_value=3)
|
||
|
||
def invalid_model(payload: dict, _stages: list[str]):
|
||
response = _valid_api_arrangement(payload)
|
||
if violation == "unknown_connector":
|
||
paragraph = next(
|
||
item["paragraphs"][0]
|
||
for item in response["sections"]
|
||
if item["paragraphs"][0]["connector_ids"]
|
||
)
|
||
paragraph["connector_ids"][0] = "invented"
|
||
elif violation == "extra_field":
|
||
response["unexpected"] = True
|
||
elif violation == "free_text_field":
|
||
response["sections"][0]["paragraphs"][0]["text"] = "API 自由文本"
|
||
else:
|
||
response["sections"][0]["section_id"] = "changed.section"
|
||
return response, {"status": "generated"}
|
||
|
||
monkeypatch.setattr(report_module, "_generate_api_narrative", invalid_model)
|
||
result = _generate(
|
||
workspace,
|
||
expected_stages=["identify"],
|
||
stage_statuses={"identify": "success"},
|
||
use_model=True,
|
||
)
|
||
|
||
markdown = result.report_path.read_text(encoding="utf-8")
|
||
manifest = json.loads(result.manifest_path.read_text(encoding="utf-8"))
|
||
assert result.generation_mode == "fallback_script"
|
||
assert "最终成员数：3" in markdown
|
||
assert "API 自由文本" not in markdown
|
||
assert manifest["model"]["reason"] == "api_unavailable_or_boundary_violation"
|
||
|
||
|
||
def test_open_reuses_existing_report(
|
||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
workspace = tmp_path / "job"
|
||
manifest_path = _write_manifest(workspace, prefix="Ft.SPL", stage="identify")
|
||
calls = {"count": 0}
|
||
|
||
def valid_model(payload: dict, _stages: list[str]):
|
||
calls["count"] += 1
|
||
return (
|
||
_valid_api_arrangement(payload),
|
||
{"status": "generated", "provider": "test", "model": "test-model"},
|
||
)
|
||
|
||
monkeypatch.setattr(report_module, "_generate_api_narrative", valid_model)
|
||
first = _generate(
|
||
workspace,
|
||
expected_stages=["identify"],
|
||
stage_statuses={"identify": "success"},
|
||
use_model=True,
|
||
)
|
||
reopened = _generate(
|
||
workspace,
|
||
expected_stages=["identify"],
|
||
stage_statuses={"identify": "success"},
|
||
use_model=True,
|
||
action="open",
|
||
)
|
||
assert first.reused is False
|
||
assert reopened.reused is True
|
||
assert calls["count"] == 1
|
||
|
||
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
|
||
manifest["metrics"]["member_count"] = 2
|
||
manifest_path.write_text(json.dumps(manifest, ensure_ascii=False), encoding="utf-8")
|
||
reopened_after_source_change = _generate(
|
||
workspace,
|
||
expected_stages=["identify"],
|
||
stage_statuses={"identify": "success"},
|
||
use_model=True,
|
||
action="open",
|
||
)
|
||
assert reopened_after_source_change.reused is True
|
||
assert calls["count"] == 1
|
||
|
||
|
||
def test_no_successful_evidence_creates_no_empty_report(tmp_path: Path) -> None:
|
||
workspace = tmp_path / "job"
|
||
_write_manifest(workspace, prefix="Ft.SPL", stage="identify", status="failed")
|
||
|
||
with pytest.raises(NoSuccessfulGeneFamilyEvidence, match="暂无可生成"):
|
||
_generate(
|
||
workspace,
|
||
expected_stages=["identify"],
|
||
stage_statuses={"identify": "failed"},
|
||
)
|
||
|
||
assert not list((workspace / "final").glob("*.gene_family_report.*"))
|