58 lines
1.9 KiB
Python
58 lines
1.9 KiB
Python
from __future__ import annotations
|
|
|
|
from apps.web.vue3.main import ExecuteRequest
|
|
from gfa_core.mas.agents.planning_v2 import PlanningAgentV2
|
|
from gfa_core.mas.stage_capabilities import CAPABILITIES
|
|
from gfa_core.mas.stage_catalog import EXECUTABLE_STAGES, STAGE_DEPENDENCIES
|
|
from gfa_core.mas.tools.stage_tools import StageToolRegistry
|
|
|
|
|
|
def test_wgs_gvcf_combine_vertical_registration_is_exact() -> None:
|
|
assert STAGE_DEPENDENCIES["wgs_gvcf_combine"] == [
|
|
"wgs_haplotype_calling",
|
|
"wgs_index",
|
|
]
|
|
assert "wgs_gvcf_combine" in EXECUTABLE_STAGES
|
|
capability = CAPABILITIES["wgs_gvcf_combine"]
|
|
assert {item.role for item in capability.consumes} == {
|
|
"gvcf",
|
|
"gvcf_index",
|
|
"reference_fasta",
|
|
"reference_fai",
|
|
"reference_dict",
|
|
}
|
|
assert {item.role for item in capability.provides} == {
|
|
"combined_gvcf",
|
|
"combined_gvcf_index",
|
|
"combine_log",
|
|
}
|
|
definition = next(
|
|
item
|
|
for item in StageToolRegistry.STAGE_DEFINITIONS
|
|
if item["id"] == "wgs_gvcf_combine"
|
|
)
|
|
assert definition["script"] == "run_wgs_gvcf_combine.py"
|
|
assert definition["conda_env"] == "ASAW"
|
|
assert capability.env == "ASAW"
|
|
assert definition["required_inputs"] == [
|
|
"wgs_haplotype_manifest",
|
|
"wgs_index_manifest",
|
|
]
|
|
assert PlanningAgentV2.REQUIRED_INPUTS["wgs_gvcf_combine"] == []
|
|
|
|
|
|
def test_api_contract_carries_stage07_authorities_and_gatk() -> None:
|
|
request = ExecuteRequest(
|
|
stage="wgs_gvcf_combine",
|
|
wgsHaplotypeManifest=(
|
|
"final/Cohort.wgs_haplotype_calling.manifest.json"
|
|
),
|
|
wgsIndexManifest="final/Cohort.wgs_index.manifest.json",
|
|
wgsGatk="/opt/gatk",
|
|
)
|
|
assert request.wgsHaplotypeManifest.endswith(
|
|
".wgs_haplotype_calling.manifest.json"
|
|
)
|
|
assert request.wgsIndexManifest.endswith(".wgs_index.manifest.json")
|
|
assert request.wgsGatk == "/opt/gatk"
|