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