marcopan
4a1272fa0a
- workflow.yaml: 8-phase definition, data-driven prerequisites, no hardcoded ladder - workflow.py: load_workflow() reader; max_phase=len(phases), decision_min_phase derived from prerequisites scan (no duplication) - 6 tests: phase count, decision_min_phase derivation, name lookup (incl. F8 presence -- the JS drift bug structurally impossible now), artifacts_out, advance strategy, schema_version. All 8 harness tests pass.
56 lines
2.1 KiB
Python
56 lines
2.1 KiB
Python
from nsp.workflow import load_workflow
|
|
|
|
|
|
def test_workflow_loads_8_phases():
|
|
wf = load_workflow()
|
|
assert len(wf.phases) == 8
|
|
assert wf.phases[0].id == "F1"
|
|
assert wf.max_phase == 8
|
|
assert wf.phase_by_num(1).name == "chiarimento"
|
|
|
|
|
|
def test_decision_min_phase_derived():
|
|
wf = load_workflow()
|
|
# question_rewritten is a prerequisite of F3 -> min phase 3
|
|
assert wf.decision_min_phase("question_rewritten") == 3
|
|
# sql_approved is a prerequisite of F7 -> min phase 7
|
|
assert wf.decision_min_phase("sql_approved") == 7
|
|
# datamart_requested / datamart_declined are prerequisites of F8 -> min phase 8
|
|
assert wf.decision_min_phase("datamart_requested") == 8
|
|
assert wf.decision_min_phase("datamart_declined") == 8
|
|
# phase_skipped referenced via decision_subject_exists in F6 -> min phase 6
|
|
assert wf.decision_min_phase("phase_skipped") == 6
|
|
# unknown type -> phase 1 (default)
|
|
assert wf.decision_min_phase("nonexistent_type") == 1
|
|
|
|
|
|
def test_phase_name_lookup():
|
|
wf = load_workflow()
|
|
assert wf.phase_name(1) == "chiarimento"
|
|
assert wf.phase_name(6) == "cte"
|
|
assert wf.phase_name(8) == "datamart" # the JS drift bug in ChironeWp3 -- F8 MUST be present
|
|
assert wf.phase_name(0) == "?" # out of range lower
|
|
assert wf.phase_name(9) == "?" # out of range upper
|
|
|
|
|
|
def test_artifacts_out_per_phase():
|
|
wf = load_workflow()
|
|
assert "schema_linking.json" in wf.phase_by_num(4).artifacts_out
|
|
assert "sql_final.sql" in wf.phase_by_num(7).artifacts_out
|
|
assert "cte_plan.json" in wf.phase_by_num(6).artifacts_out
|
|
assert "ctes/" in wf.phase_by_num(6).artifacts_out # dir artifact
|
|
assert wf.phase_by_num(1).artifacts_out == [] # F1 produces none
|
|
|
|
|
|
def test_advance_strategy_per_phase():
|
|
wf = load_workflow()
|
|
assert wf.phase_by_num(2).advance == "auto_if_empty"
|
|
assert wf.phase_by_num(6).advance == "auto_if_empty_or_skipped"
|
|
assert wf.phase_by_num(4).advance == "reviewer_decide"
|
|
assert wf.phase_by_num(5).advance == "kind:phase"
|
|
|
|
|
|
def test_schema_version():
|
|
wf = load_workflow()
|
|
assert wf.schema_version == 1
|