104 lines
3.0 KiB
Python
104 lines
3.0 KiB
Python
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
|
|
GWAS_IDENTITIES = {
|
|
"gwas_emplot": {
|
|
"genotype_scope_id": "genotype:test",
|
|
"phenotype_scope_id": "phenotype:test",
|
|
"source_model_scope_id": "gwas-model:test",
|
|
"analysis_scope_id": "gwas-analysis:plot",
|
|
},
|
|
"gwas_glm": {
|
|
"genotype_scope_id": "genotype:test",
|
|
"phenotype_scope_id": "phenotype:test",
|
|
"covariate_scope_id": "covariate:test",
|
|
"model_scope_id": "gwas-model:glm",
|
|
"analysis_scope_id": "gwas-analysis:glm",
|
|
},
|
|
}
|
|
|
|
|
|
def _status(completed: bool):
|
|
return SimpleNamespace(
|
|
completed=completed,
|
|
errors=[] if completed else ["incomplete scientific scope identity"],
|
|
missing_artifacts=[],
|
|
to_dict=lambda: {"completed": completed},
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(("stage", "required_identity"), GWAS_IDENTITIES.items())
|
|
def test_celery_postflight_preserves_complete_gwas_manifest_identity(
|
|
monkeypatch, stage, required_identity
|
|
) -> None:
|
|
import tasks
|
|
from gfa_core.mas.tools import stage_identity
|
|
|
|
captured = {}
|
|
|
|
def fake_get_stage_status_for_identity(*, identity, strict_identity, **_kwargs):
|
|
captured["identity"] = identity
|
|
captured["strict_identity"] = strict_identity
|
|
complete = all(getattr(identity, key) == value for key, value in required_identity.items())
|
|
return _status(complete)
|
|
|
|
monkeypatch.setattr(
|
|
stage_identity,
|
|
"get_stage_status_for_identity",
|
|
fake_get_stage_status_for_identity,
|
|
)
|
|
manifest = {"stage": stage, "prefix": "Os", **required_identity}
|
|
|
|
result = tasks.validate_single_stage_artifacts(
|
|
"job",
|
|
"Os",
|
|
stage,
|
|
{"final_output": {"manifest": manifest}},
|
|
)
|
|
|
|
assert result is None
|
|
assert captured["strict_identity"] is True
|
|
for key, value in required_identity.items():
|
|
assert getattr(captured["identity"], key) == value
|
|
|
|
|
|
def test_celery_postflight_keeps_partial_gwas_manifest_invalid(monkeypatch) -> None:
|
|
import tasks
|
|
from gfa_core.mas.tools import stage_identity
|
|
|
|
def fake_get_stage_status_for_identity(*, identity, **_kwargs):
|
|
complete = all(
|
|
(
|
|
identity.genotype_scope_id,
|
|
identity.phenotype_scope_id,
|
|
identity.covariate_scope_id,
|
|
identity.model_scope_id,
|
|
identity.analysis_scope_id,
|
|
)
|
|
)
|
|
return _status(complete)
|
|
|
|
monkeypatch.setattr(
|
|
stage_identity,
|
|
"get_stage_status_for_identity",
|
|
fake_get_stage_status_for_identity,
|
|
)
|
|
partial_manifest = {
|
|
"stage": "gwas_glm",
|
|
"prefix": "Os",
|
|
"genotype_scope_id": "genotype:test",
|
|
"analysis_scope_id": "gwas-analysis:glm",
|
|
}
|
|
|
|
result = tasks.validate_single_stage_artifacts(
|
|
"job",
|
|
"Os",
|
|
"gwas_glm",
|
|
{"final_output": {"manifest": partial_manifest}},
|
|
)
|
|
|
|
assert result is not None
|
|
assert result["status"] == "failed"
|