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"], ]