306 lines
13 KiB
Python
306 lines
13 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
|
|
def _write_valid_manifest(
|
|
final_dir: Path, *, prefix: str = "cohort", scope: str = "genotype:scope",
|
|
) -> tuple[Path, dict]:
|
|
from gfa_core.genotype_context import sha256_file
|
|
|
|
files = {
|
|
"sorted_vcf": final_dir / f"{prefix}.gwas_fmt.sorted.vcf.gz",
|
|
"tassel_hapmap": final_dir / f"{prefix}.gwas_fmt.hapmap.hmp.txt",
|
|
"report": final_dir / f"{prefix}.gwas_fmt.report.md",
|
|
"environment": final_dir / f"{prefix}.gwas_fmt.environment.txt",
|
|
}
|
|
for role, path in files.items():
|
|
path.write_bytes(f"{role}\n".encode())
|
|
artifacts = {
|
|
role: {
|
|
"artifact": role,
|
|
"role": role,
|
|
"path": f"final/{path.name}",
|
|
"full_path": str(path),
|
|
"type": "application/octet-stream",
|
|
"size": path.stat().st_size,
|
|
"sha256": sha256_file(path),
|
|
"size_bytes": path.stat().st_size,
|
|
}
|
|
for role, path in files.items()
|
|
}
|
|
manifest = {
|
|
"manifest_version": "1.0",
|
|
"job_id": "job",
|
|
"stage": "gwas_fmt",
|
|
"run_id": "gwas_fmt_20260720_120000",
|
|
"prefix": prefix,
|
|
"timestamp": "2026-07-20T12:00:12+00:00",
|
|
"status": "success",
|
|
"start_time": "2026-07-20T12:00:00+00:00",
|
|
"end_time": "2026-07-20T12:00:12+00:00",
|
|
"genotype_scope_id": scope,
|
|
"elapsed_sec": 12.34,
|
|
"cmd": ["python", "-m", "apps.cli.run_gwas_tassel"],
|
|
"workspace_root": str(final_dir.parent),
|
|
"workspace": str(final_dir),
|
|
"input_dir": str(final_dir.parent / "input"),
|
|
"parameters": {"unsupported_marker_policy": "fail"},
|
|
"software": {"tassel": {"version": "5.2.96"}},
|
|
"metrics": {
|
|
"sample_conservation": True,
|
|
"variant_conservation": True,
|
|
"genotype_conservation": True,
|
|
},
|
|
"artifacts": artifacts,
|
|
}
|
|
manifest_path = final_dir / f"{prefix}.gwas_fmt.manifest.json"
|
|
manifest_path.write_text(json.dumps(manifest), encoding="utf-8")
|
|
return manifest_path, manifest
|
|
|
|
|
|
def test_result_helper_validates_every_dict_artifact_before_returning_public_records(tmp_path: Path) -> None:
|
|
from gfa_core.gwas_fmt_context import gwas_fmt_result_artifacts
|
|
|
|
final_dir = tmp_path / "final"
|
|
final_dir.mkdir()
|
|
_, manifest = _write_valid_manifest(final_dir)
|
|
|
|
records = gwas_fmt_result_artifacts(manifest, "genotype:scope", tmp_path)
|
|
assert {record["role"] for record in records} == {
|
|
"sorted_vcf", "tassel_hapmap", "report", "environment",
|
|
}
|
|
|
|
manifest["artifacts"]["environment"]["size_bytes"] += 1
|
|
assert gwas_fmt_result_artifacts(manifest, "genotype:scope", tmp_path) == []
|
|
|
|
manifest["artifacts"]["environment"]["size_bytes"] -= 1
|
|
manifest["genotype_scope_id"] = ""
|
|
assert gwas_fmt_result_artifacts(manifest, "", tmp_path) == []
|
|
|
|
manifest["genotype_scope_id"] = "genotype:scope"
|
|
manifest["parameters"]["unsupported_marker_policy"] = "mask"
|
|
manifest["metrics"].update({
|
|
"lossy_conversion": True,
|
|
"strict_genotype_conservation": True,
|
|
"strict_marker_count": "invalid",
|
|
"genotypes_compared": 2,
|
|
"unsupported_marker_count": 1,
|
|
})
|
|
assert gwas_fmt_result_artifacts(manifest, "genotype:scope", tmp_path) == []
|
|
|
|
|
|
def test_result_helper_rejects_list_and_malformed_dict_artifacts(tmp_path: Path) -> None:
|
|
from gfa_core.gwas_fmt_context import gwas_fmt_result_artifacts
|
|
|
|
final_dir = tmp_path / "final"
|
|
final_dir.mkdir()
|
|
_, manifest = _write_valid_manifest(final_dir)
|
|
valid_artifacts = manifest["artifacts"]
|
|
|
|
manifest["artifacts"] = list(valid_artifacts.values())
|
|
assert gwas_fmt_result_artifacts(manifest, "genotype:scope", tmp_path) == []
|
|
|
|
manifest["artifacts"] = {**valid_artifacts, "invalid": "not-an-artifact-record"}
|
|
assert gwas_fmt_result_artifacts(manifest, "genotype:scope", tmp_path) == []
|
|
|
|
manifest["artifacts"] = {
|
|
role: record for role, record in valid_artifacts.items() if role != "report"
|
|
}
|
|
assert gwas_fmt_result_artifacts(manifest, "genotype:scope", tmp_path) == []
|
|
|
|
|
|
def test_result_helper_requires_complete_core_manifest_v1_shape(tmp_path: Path) -> None:
|
|
from gfa_core.gwas_fmt_context import gwas_fmt_result_artifacts
|
|
|
|
final_dir = tmp_path / "final"
|
|
final_dir.mkdir()
|
|
_, manifest = _write_valid_manifest(final_dir)
|
|
|
|
for required_field in (
|
|
"manifest_version", "job_id", "stage", "run_id", "prefix", "timestamp",
|
|
"status", "start_time", "end_time", "elapsed_sec", "cmd", "workspace_root",
|
|
"workspace", "input_dir", "artifacts", "parameters", "metrics",
|
|
):
|
|
incomplete = dict(manifest)
|
|
incomplete.pop(required_field)
|
|
assert gwas_fmt_result_artifacts(incomplete, "genotype:scope", tmp_path) == [], required_field
|
|
|
|
|
|
def test_result_helper_rejects_noncanonical_core_manifest_v1_locations(tmp_path: Path) -> None:
|
|
from gfa_core.gwas_fmt_context import gwas_fmt_result_artifacts
|
|
|
|
final_dir = tmp_path / "final"
|
|
final_dir.mkdir()
|
|
_, manifest = _write_valid_manifest(final_dir)
|
|
wrong_version = {**manifest, "manifest_version": "2.0"}
|
|
wrong_workspace_root = {**manifest, "workspace_root": str(tmp_path / "other")}
|
|
wrong_workspace = {**manifest, "workspace": str(tmp_path / "other" / "final")}
|
|
absolute_artifact_path = {
|
|
**manifest,
|
|
"artifacts": {key: dict(value) for key, value in manifest["artifacts"].items()},
|
|
}
|
|
absolute_artifact_path["artifacts"]["sorted_vcf"]["path"] = str(
|
|
final_dir / "cohort.gwas_fmt.sorted.vcf.gz"
|
|
)
|
|
|
|
invalid_manifests = (wrong_version, wrong_workspace_root, wrong_workspace, absolute_artifact_path)
|
|
assert [
|
|
gwas_fmt_result_artifacts(invalid, "genotype:scope", tmp_path)
|
|
for invalid in invalid_manifests
|
|
] == [[], [], [], []]
|
|
|
|
|
|
def test_result_helper_rejects_extra_role_and_key_role_artifact_mismatch(tmp_path: Path) -> None:
|
|
from gfa_core.genotype_context import sha256_file
|
|
from gfa_core.gwas_fmt_context import gwas_fmt_result_artifacts
|
|
|
|
final_dir = tmp_path / "final"
|
|
final_dir.mkdir()
|
|
_, manifest = _write_valid_manifest(final_dir)
|
|
extra = final_dir / "cohort.gwas_fmt.extra.txt"
|
|
extra.write_text("extra\n", encoding="utf-8")
|
|
manifest["artifacts"]["extra"] = {
|
|
"artifact": "extra", "role": "extra", "path": "final/cohort.gwas_fmt.extra.txt",
|
|
"full_path": str(extra), "type": "text/plain", "size": extra.stat().st_size,
|
|
"size_bytes": extra.stat().st_size, "sha256": sha256_file(extra),
|
|
}
|
|
assert gwas_fmt_result_artifacts(manifest, "genotype:scope", tmp_path) == []
|
|
|
|
_, manifest = _write_valid_manifest(final_dir)
|
|
manifest["artifacts"]["sorted_vcf"]["role"] = "report"
|
|
assert gwas_fmt_result_artifacts(manifest, "genotype:scope", tmp_path) == []
|
|
manifest["artifacts"]["sorted_vcf"]["role"] = "sorted_vcf"
|
|
manifest["artifacts"]["sorted_vcf"]["artifact"] = "report"
|
|
assert gwas_fmt_result_artifacts(manifest, "genotype:scope", tmp_path) == []
|
|
|
|
def test_results_api_synthesizes_manifest_record_and_uses_manifest_elapsed(monkeypatch, tmp_path: Path) -> None:
|
|
from apps.web.vue3 import main as web
|
|
|
|
workspace_root = tmp_path / "workspace"
|
|
final_dir = workspace_root / "job" / "final"
|
|
final_dir.mkdir(parents=True)
|
|
manifest_path, manifest = _write_valid_manifest(final_dir)
|
|
monkeypatch.setattr(web, "WORKSPACE_ROOT", workspace_root)
|
|
|
|
payload = asyncio.run(web.get_results(
|
|
job_id="job",
|
|
stage="gwas_fmt",
|
|
rnaseq_scope_id="",
|
|
scope_id="",
|
|
genotype_scope_id="genotype:scope",
|
|
))
|
|
|
|
files = payload["gwas_fmt"]["files"]
|
|
by_role = {record["role"]: record for record in files}
|
|
assert {"manifest", "report", "sorted_vcf", "tassel_hapmap"} <= set(by_role)
|
|
assert by_role["manifest"]["name"] == manifest_path.name
|
|
assert by_role["manifest"]["size"] == manifest_path.stat().st_size
|
|
assert by_role["manifest"]["artifact"] == manifest_path.name
|
|
assert by_role["manifest"]["type"] == "application/json"
|
|
assert by_role["manifest"]["media_type"] == "application/json"
|
|
assert "manifest" not in manifest["artifacts"]
|
|
assert payload["gwas_fmt"]["elapsed_time"] == 12.3
|
|
|
|
|
|
def test_results_api_rehydrates_unique_gwas_fmt_scope_after_refresh(monkeypatch, tmp_path: Path) -> None:
|
|
from apps.web.vue3 import main as web
|
|
|
|
workspace_root = tmp_path / "workspace"
|
|
final_dir = workspace_root / "job" / "final"
|
|
final_dir.mkdir(parents=True)
|
|
_write_valid_manifest(final_dir, prefix="cohort", scope="genotype:scope")
|
|
monkeypatch.setattr(web, "WORKSPACE_ROOT", workspace_root)
|
|
|
|
payload = asyncio.run(web.get_results(
|
|
job_id="job", stage="gwas_fmt", rnaseq_scope_id="", scope_id="",
|
|
genotype_scope_id="",
|
|
))
|
|
|
|
assert payload["gwas_fmt"]["status"] == "success"
|
|
assert {record["role"] for record in payload["gwas_fmt"]["files"]} >= {
|
|
"manifest", "report", "sorted_vcf", "tassel_hapmap",
|
|
}
|
|
|
|
|
|
def test_results_api_requires_requested_gwas_scope_and_selects_only_exact_history(monkeypatch, tmp_path: Path) -> None:
|
|
from apps.web.vue3 import main as web
|
|
|
|
workspace_root = tmp_path / "workspace"
|
|
final_dir = workspace_root / "job" / "final"
|
|
final_dir.mkdir(parents=True)
|
|
_write_valid_manifest(final_dir, prefix="old", scope="genotype:old")
|
|
_write_valid_manifest(final_dir, prefix="current", scope="genotype:current")
|
|
monkeypatch.setattr(web, "WORKSPACE_ROOT", workspace_root)
|
|
|
|
without_scope = asyncio.run(web.get_results(
|
|
job_id="job", stage="gwas_fmt", rnaseq_scope_id="", scope_id="", genotype_scope_id="",
|
|
))
|
|
assert without_scope["gwas_fmt"]["files"] == []
|
|
|
|
inferred_by_prefix = asyncio.run(web.get_results(
|
|
job_id="job", stage="gwas_fmt", result_prefix="current",
|
|
rnaseq_scope_id="", scope_id="", genotype_scope_id="",
|
|
))
|
|
assert inferred_by_prefix["gwas_fmt"]["files"]
|
|
assert all(
|
|
"current.gwas_fmt" in record["name"]
|
|
for record in inferred_by_prefix["gwas_fmt"]["files"]
|
|
)
|
|
|
|
current = asyncio.run(web.get_results(
|
|
job_id="job", stage="gwas_fmt", rnaseq_scope_id="", scope_id="",
|
|
genotype_scope_id="genotype:current",
|
|
))
|
|
assert current["gwas_fmt"]["files"]
|
|
assert all("current.gwas_fmt" in record["name"] for record in current["gwas_fmt"]["files"])
|
|
|
|
|
|
def test_gwas_frontend_displays_core_roles_but_hides_environment(tmp_path: Path) -> None:
|
|
frontend = (Path(__file__).resolve().parents[2] / "apps/web/vue3/index.html").read_text(encoding="utf-8")
|
|
start = frontend.index("const isReportOrManifestFile =")
|
|
end = frontend.index("const parseSimpleTable =", start)
|
|
helper_source = frontend[start:end]
|
|
node_script = tmp_path / "gwas-result-files.cjs"
|
|
node_script.write_text(
|
|
"const isImageFile=()=>false; const isJumpFile=()=>false;\n"
|
|
+ helper_source
|
|
+ "\nconst files=['manifest','report','sorted_vcf','tassel_hapmap','environment'].map(role=>({role,name:role}));"
|
|
+ "\nprocess.stdout.write(JSON.stringify(resultDisplayFiles({name:'gwas_fmt',files}).map(x=>x.role)));",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
completed = subprocess.run(["node", str(node_script)], text=True, capture_output=True, check=False)
|
|
|
|
assert completed.returncode == 0, completed.stderr
|
|
assert json.loads(completed.stdout) == ["manifest", "report", "sorted_vcf", "tassel_hapmap"]
|
|
|
|
|
|
def test_gwas_frontend_click_opens_core_artifact_path(tmp_path: Path) -> None:
|
|
frontend = (Path(__file__).resolve().parents[2] / "apps/web/vue3/index.html").read_text(encoding="utf-8")
|
|
start = frontend.index("const selectFile =")
|
|
end = frontend.index("// 缩放图片", start)
|
|
helper_source = frontend[start:end]
|
|
node_script = tmp_path / "gwas-select-file.cjs"
|
|
node_script.write_text(
|
|
"const selectedResultStage={value:''},selectedFile={value:''},selectedImage={value:''},selectedImageUrl={value:''},selectedImageSize={value:''},zoomLevel={value:0};"
|
|
"const opened=[]; const window={open:(...args)=>opened.push(args)}; const fetch=()=>Promise.reject();"
|
|
"const isJumpFile=()=>false,isImageFile=()=>false,openTablePreview=()=>{},formatSize=()=>'',ElMessage={info(){},error(){}};\n"
|
|
+ helper_source
|
|
+ "\nselectFile('gwas_fmt',{name:'cohort.vcf.gz',role:'sorted_vcf',path:'/workspace/job/final/cohort.vcf.gz'});"
|
|
+ "selectFile('gwas_fmt',{name:'cohort.hmp.txt',role:'tassel_hapmap',path:'/workspace/job/final/cohort.hmp.txt'});"
|
|
+ "process.stdout.write(JSON.stringify(opened));",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
completed = subprocess.run(["node", str(node_script)], text=True, capture_output=True, check=False)
|
|
|
|
assert completed.returncode == 0, completed.stderr
|
|
assert json.loads(completed.stdout) == [
|
|
["/workspace/job/final/cohort.vcf.gz", "_blank"],
|
|
["/workspace/job/final/cohort.hmp.txt", "_blank"],
|
|
]
|