105 lines
3.9 KiB
Python
105 lines
3.9 KiB
Python
"""tests/test_archive.py —— 存档打包/大小统计/报告定位/图清单。"""
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import json
|
||
|
||
import pytest
|
||
|
||
from app.services.bioflow.archive import (
|
||
archive_from_job,
|
||
normalize_archive_name,
|
||
)
|
||
from app.services.bioflow.reports import (
|
||
extract_figure_refs,
|
||
locate_report,
|
||
report_payload,
|
||
)
|
||
|
||
pytestmark = pytest.mark.bioflow_unit
|
||
|
||
|
||
def _run(coro):
|
||
return asyncio.new_event_loop().run_until_complete(coro)
|
||
|
||
|
||
def _make_job_dir(tmp_path, with_intermediate=True, with_report=True):
|
||
job_dir = tmp_path / "job" / "j1"
|
||
(job_dir / "workspace" / "final").mkdir(parents=True)
|
||
(job_dir / "workspace" / "identify" / "analysis").mkdir(parents=True)
|
||
(job_dir / "logs").mkdir(parents=True)
|
||
(job_dir / "input").mkdir(parents=True)
|
||
(job_dir / "workspace" / "final" / "Ft.SPL.final_members.id").write_text(
|
||
"gene1\ngene2\n", encoding="utf-8")
|
||
if with_intermediate:
|
||
(job_dir / "workspace" / "identify" / "analysis" / "domtblout").write_text(
|
||
"intermediate", encoding="utf-8")
|
||
(job_dir / "logs" / "identify.log").write_text("log line", encoding="utf-8")
|
||
(job_dir / "input" / "genome.pep.fa").write_text(">x\nACGT", encoding="utf-8")
|
||
(job_dir / "events.jsonl").write_text(json.dumps({"type": "job_created"}), encoding="utf-8")
|
||
if with_report:
|
||
html = job_dir / "workspace" / "final" / "Ft.SPL.gene_family_report.html"
|
||
html.write_text(
|
||
'<html><body><img src="fig://heatmap_1"/><img src="fig://tree_2"/></body></html>',
|
||
encoding="utf-8")
|
||
return job_dir
|
||
|
||
|
||
def test_archive_pack(tmp_path):
|
||
job_dir = _make_job_dir(tmp_path)
|
||
archives_root = tmp_path / "archives"
|
||
job = {"id": "j1", "session_id": "s1", "title": "t",
|
||
"stage_chain": ["identify"], "params": {}}
|
||
result = _run(archive_from_job(job, job_dir, archives_root, name="SPL 鉴定"))
|
||
assert result.name == "SPL 鉴定"
|
||
assert result.size_bytes > 0
|
||
dest = tmp_path / "archives" / result.id
|
||
assert (dest / "final" / "Ft.SPL.final_members.id").exists()
|
||
assert (dest / "logs" / "identify.log").exists()
|
||
assert (dest / "input" / "genome.pep.fa").exists()
|
||
assert (dest / "events.jsonl").exists()
|
||
# 不含中间文件（默认 include_intermediate=False）
|
||
assert not (dest / "intermediate").exists()
|
||
|
||
|
||
def test_archive_include_intermediate(tmp_path):
|
||
job_dir = _make_job_dir(tmp_path)
|
||
result = _run(archive_from_job(
|
||
{"id": "j1", "title": "t", "stage_chain": [], "params": {}},
|
||
job_dir, tmp_path / "archives", name="X",
|
||
include_intermediate=True))
|
||
assert (tmp_path / "archives" / result.id / "intermediate" / "identify" / "analysis" / "domtblout").exists()
|
||
|
||
|
||
def test_normalize_name():
|
||
assert normalize_archive_name(" SPL 鉴定 ") == "SPL 鉴定"
|
||
assert normalize_archive_name(" ") == "未命名存档"
|
||
|
||
|
||
def test_locate_report(tmp_path):
|
||
job_dir = _make_job_dir(tmp_path)
|
||
assert locate_report(job_dir / "workspace") is not None
|
||
assert locate_report(job_dir / "workspace").suffix == ".html"
|
||
|
||
|
||
def test_locate_report_md_fallback(tmp_path):
|
||
job_dir = _make_job_dir(tmp_path, with_report=False)
|
||
(job_dir / "workspace" / "final" / "Ft.SPL.identify_report.md").write_text("# report", encoding="utf-8")
|
||
assert locate_report(job_dir / "workspace").suffix == ".md"
|
||
|
||
|
||
def test_extract_figure_refs(tmp_path):
|
||
job_dir = _make_job_dir(tmp_path)
|
||
report = locate_report(job_dir / "workspace")
|
||
figs = extract_figure_refs(report)
|
||
assert figs == ["heatmap_1", "tree_2"]
|
||
|
||
|
||
def test_report_payload(tmp_path):
|
||
job_dir = _make_job_dir(tmp_path)
|
||
report = locate_report(job_dir / "workspace")
|
||
payload = report_payload({"id": "j1", "session_id": "s1", "title": "SPL 报告"}, report)
|
||
assert payload["job_id"] == "j1"
|
||
assert payload["figures"] == ["heatmap_1", "tree_2"]
|
||
assert payload["title"] == "SPL 报告"
|