137 lines
4.5 KiB
Python
137 lines
4.5 KiB
Python
from __future__ import annotations
|
|
|
|
import re
|
|
|
|
import pytest
|
|
|
|
from gfa_core.gwas_scientific_identity import (
|
|
GwasIdentityError,
|
|
build_covariate_scope_id,
|
|
build_identity_recipe,
|
|
build_phenotype_scope_id,
|
|
materialize_stage_identity,
|
|
new_run_id,
|
|
)
|
|
|
|
|
|
SHA_A = "a" * 64
|
|
SHA_B = "b" * 64
|
|
SHA_C = "c" * 64
|
|
|
|
|
|
def test_phenotype_scope_is_canonical_and_sensitive_to_selected_trait() -> None:
|
|
first = build_phenotype_scope_id(
|
|
phenotype_sha256=SHA_A,
|
|
sample_column="ID",
|
|
trait_column="GC_BLUP",
|
|
trait_name="heading_days",
|
|
retained_sample_ids=["S1", "S2"],
|
|
)
|
|
same = build_phenotype_scope_id(
|
|
trait_name="heading_days",
|
|
retained_sample_ids=["S1", "S2"],
|
|
trait_column="GC_BLUP",
|
|
sample_column="ID",
|
|
phenotype_sha256=SHA_A,
|
|
)
|
|
changed = build_phenotype_scope_id(
|
|
phenotype_sha256=SHA_A,
|
|
sample_column="ID",
|
|
trait_column="other_trait",
|
|
trait_name="heading_days",
|
|
retained_sample_ids=["S1", "S2"],
|
|
)
|
|
assert first == same
|
|
assert first.startswith("phenotype:")
|
|
assert changed != first
|
|
|
|
|
|
def test_covariate_scope_requires_verified_source_genotype() -> None:
|
|
with pytest.raises(GwasIdentityError, match="source genotype"):
|
|
build_covariate_scope_id(
|
|
q_sha256=SHA_A,
|
|
fam_sha256=SHA_B,
|
|
source_genotype_scope_id="",
|
|
k=2,
|
|
)
|
|
scope = build_covariate_scope_id(
|
|
q_sha256=SHA_A,
|
|
fam_sha256=SHA_B,
|
|
source_genotype_scope_id=f"genotype:{SHA_C}",
|
|
k=2,
|
|
)
|
|
assert scope.startswith("covariate:")
|
|
|
|
|
|
def test_recipe_does_not_fabricate_downstream_identity() -> None:
|
|
recipe = build_identity_recipe(
|
|
stage="gwas_mlm",
|
|
required_artifact_roles=["tassel_hapmap", "kinship_matrix", "trait_table", "covariate_table"],
|
|
parameters={"p3d": True},
|
|
method_version="tassel_mlm.v1",
|
|
contract_version="v1",
|
|
tool_versions={"tassel": "5.2.96"},
|
|
)
|
|
assert "analysis_scope_id" not in recipe
|
|
assert "model_scope_id" not in recipe
|
|
with pytest.raises(GwasIdentityError, match="missing authoritative artifact"):
|
|
materialize_stage_identity(
|
|
recipe,
|
|
upstream_artifacts={"tassel_hapmap": {"sha256": SHA_A}},
|
|
data_scopes={"genotype_scope_id": f"genotype:{SHA_A}"},
|
|
)
|
|
|
|
|
|
def test_materialized_model_identity_uses_every_authoritative_hash() -> None:
|
|
roles = ["tassel_hapmap", "kinship_matrix", "trait_table", "covariate_table"]
|
|
recipe = build_identity_recipe(
|
|
stage="gwas_mlm",
|
|
required_artifact_roles=roles,
|
|
parameters={"p3d": True},
|
|
method_version="tassel_mlm.v1",
|
|
contract_version="v1",
|
|
tool_versions={"tassel": "5.2.96"},
|
|
)
|
|
artifacts = {role: {"sha256": str(index) * 64} for index, role in enumerate(roles, start=1)}
|
|
scopes = {
|
|
"genotype_scope_id": f"genotype:{SHA_A}",
|
|
"phenotype_scope_id": f"phenotype:{SHA_B}",
|
|
"covariate_scope_id": f"covariate:{SHA_C}",
|
|
}
|
|
identity = materialize_stage_identity(recipe, artifacts, scopes)
|
|
changed = {role: dict(value) for role, value in artifacts.items()}
|
|
changed["kinship_matrix"]["sha256"] = "9" * 64
|
|
changed_identity = materialize_stage_identity(recipe, changed, scopes)
|
|
|
|
assert identity["model_scope_id"].startswith("gwas-model:")
|
|
assert identity["analysis_scope_id"].startswith("gwas-analysis:")
|
|
assert identity["model_scope_id"] != changed_identity["model_scope_id"]
|
|
assert identity["analysis_scope_id"] != changed_identity["analysis_scope_id"]
|
|
|
|
|
|
def test_plot_has_own_analysis_scope_and_preserves_source_model_scope() -> None:
|
|
recipe = build_identity_recipe(
|
|
stage="gwas_plot",
|
|
required_artifact_roles=["mlm_association"],
|
|
parameters={"format": "png"},
|
|
method_version="cmplot.v1",
|
|
contract_version="v1",
|
|
tool_versions={"CMplot": "4.5.1"},
|
|
)
|
|
source = f"gwas-model:{SHA_A}"
|
|
identity = materialize_stage_identity(
|
|
recipe,
|
|
{"mlm_association": {"sha256": SHA_B}},
|
|
{"source_model_scope_id": source},
|
|
)
|
|
assert identity["source_model_scope_id"] == source
|
|
assert identity["analysis_scope_id"].startswith("gwas-analysis:")
|
|
assert identity["analysis_scope_id"] != source
|
|
|
|
|
|
def test_run_id_is_stage_scoped_and_unique() -> None:
|
|
first = new_run_id("gwas_mlm")
|
|
second = new_run_id("gwas_mlm")
|
|
assert re.fullmatch(r"gwas_mlm_\d{8}T\d{6}Z_[0-9a-f]{8}", first)
|
|
assert first != second
|