119 lines
4.7 KiB
Python
119 lines
4.7 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import hashlib
|
|
import json
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
|
|
def _sha256(path: Path) -> str:
|
|
return hashlib.sha256(path.read_bytes()).hexdigest()
|
|
|
|
|
|
def _write_kinship_manifest(
|
|
final_dir: Path,
|
|
*,
|
|
genotype_scope_id: str = "genotype:" + "a" * 64,
|
|
) -> Path:
|
|
stem = "cohort.gwas_kinship.123456789abc"
|
|
artifacts = {}
|
|
for role, suffix, media_type in (
|
|
("kinship_matrix", "kinship.txt", "text/tab-separated-values"),
|
|
("environment", "environment.txt", "text/plain"),
|
|
("report", "report.md", "text/markdown"),
|
|
):
|
|
path = final_dir / f"{stem}.{suffix}"
|
|
path.write_text(f"{role}\n", encoding="utf-8")
|
|
artifacts[role] = {
|
|
"role": role,
|
|
"path": f"final/{path.name}",
|
|
"full_path": str(path.resolve()),
|
|
"type": media_type,
|
|
"size": path.stat().st_size,
|
|
"sha256": _sha256(path),
|
|
}
|
|
manifest = {
|
|
"manifest_version": "1.0",
|
|
"job_id": "job",
|
|
"stage": "gwas_kinship",
|
|
"run_id": "gwas_kinship_20260721_120000_deadbeef",
|
|
"prefix": "cohort",
|
|
"status": "success",
|
|
"start_time": "2026-07-21T12:00:00+00:00",
|
|
"end_time": "2026-07-21T12:00:03+00:00",
|
|
"elapsed_sec": 3.9,
|
|
"genotype_scope_id": genotype_scope_id,
|
|
"analysis_scope_id": "gwas-analysis:" + "1" * 64,
|
|
"artifacts": artifacts,
|
|
}
|
|
manifest_path = final_dir / f"{stem}.manifest.json"
|
|
manifest_path.write_text(json.dumps(manifest), encoding="utf-8")
|
|
return manifest_path
|
|
|
|
|
|
def test_results_exposes_only_exact_scope_hash_valid_kinship_artifacts(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 = _write_kinship_manifest(final_dir)
|
|
monkeypatch.setattr(web, "WORKSPACE_ROOT", workspace_root)
|
|
|
|
payload = asyncio.run(web.get_results(
|
|
job_id="job", stage="gwas_kinship", rnaseq_scope_id="", scope_id="",
|
|
genotype_scope_id="genotype:" + "a" * 64,
|
|
))
|
|
by_role = {item["role"]: item for item in payload["gwas_kinship"]["files"]}
|
|
assert {"manifest", "kinship_matrix", "report", "environment"} == set(by_role)
|
|
assert by_role["manifest"]["name"] == manifest_path.name
|
|
assert payload["gwas_kinship"]["elapsed_time"] == 3.9
|
|
assert payload["gwas_kinship"]["status"] == "success"
|
|
|
|
wrong_scope = asyncio.run(web.get_results(
|
|
job_id="job", stage="gwas_kinship", rnaseq_scope_id="", scope_id="",
|
|
genotype_scope_id="genotype:" + "b" * 64,
|
|
))
|
|
assert wrong_scope["gwas_kinship"]["files"] == []
|
|
|
|
matrix = final_dir / "cohort.gwas_kinship.123456789abc.kinship.txt"
|
|
matrix.write_text("tampered\n", encoding="utf-8")
|
|
tampered = asyncio.run(web.get_results(
|
|
job_id="job", stage="gwas_kinship", rnaseq_scope_id="", scope_id="",
|
|
genotype_scope_id="genotype:" + "a" * 64,
|
|
))
|
|
assert tampered["gwas_kinship"]["files"] == []
|
|
|
|
|
|
def test_frontend_declares_complete_gwas_dag_and_stage19_result_roles(tmp_path: Path) -> None:
|
|
frontend = (Path(__file__).resolve().parents[2] / "apps/web/vue3/index.html").read_text(encoding="utf-8")
|
|
for stage in (
|
|
"gwas_fmt", "gwas_kinship", "gwas_trait", "gwas_cov", "gwas_mlm",
|
|
"gwas_plot", "gwas_emmax", "gwas_emplot", "gwas_glm",
|
|
):
|
|
assert f"name: '{stage}'" in frontend
|
|
assert f"'{stage}':" in frontend
|
|
|
|
assert "'gwas_kinship': ['gwas_fmt']" in frontend
|
|
assert "'gwas_mlm': ['gwas_fmt', 'gwas_kinship', 'gwas_trait', 'gwas_cov']" in frontend
|
|
assert "'gwas_plot': ['gwas_mlm']" in frontend
|
|
assert "'gwas_emmax': ['gwas_fmt', 'gwas_trait']" in frontend
|
|
assert "'gwas_emplot': ['gwas_emmax']" in frontend
|
|
assert "'gwas_glm': ['gwas_fmt', 'gwas_trait', 'gwas_cov']" in frontend
|
|
|
|
start = frontend.index("const isReportOrManifestFile =")
|
|
end = frontend.index("const parseSimpleTable =", start)
|
|
helper_source = frontend[start:end]
|
|
node_script = tmp_path / "gwas-kinship-result-files.cjs"
|
|
node_script.write_text(
|
|
"const isImageFile=()=>false; const isJumpFile=()=>false;\n"
|
|
+ helper_source
|
|
+ "\nconst files=['manifest','report','kinship_matrix','environment'].map(role=>({role,name:role}));"
|
|
+ "\nprocess.stdout.write(JSON.stringify(resultDisplayFiles({name:'gwas_kinship',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", "kinship_matrix", "environment"]
|