1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
|
from __future__ import annotations
import pytest
from pydantic import ValidationError
from gap_pipeline.models import (
JudgeVerdict,
KernelCandidate,
KernelPlan,
ProofPlanDAG,
)
def test_valid_plan_and_candidate(plan_dict: dict, candidate_dict: dict) -> None:
plan = KernelPlan.model_validate(plan_dict)
candidate = KernelCandidate.model_validate(candidate_dict)
assert len(plan.core_steps) == 2
assert candidate.question.startswith("Let x")
def test_plan_requires_slots(plan_dict: dict) -> None:
plan_dict["mutable_slots"] = {}
with pytest.raises(ValidationError, match="mutable slot"):
KernelPlan.model_validate(plan_dict)
def test_plan_rejects_more_than_five_core_steps(plan_dict: dict) -> None:
plan_dict["core_steps"] = [f"step {index}" for index in range(6)]
with pytest.raises(ValidationError):
KernelPlan.model_validate(plan_dict)
def test_candidate_requires_question_and_solution(candidate_dict: dict) -> None:
candidate_dict["solution"] = ""
with pytest.raises(ValidationError, match="non-empty"):
KernelCandidate.model_validate(candidate_dict)
def test_accept_verdict_cannot_report_blocking_issue() -> None:
with pytest.raises(ValidationError, match="cannot contain"):
JudgeVerdict(
verdict="accept",
step_by_step_check="checked",
blocking_issues="an error",
)
def test_path_dag_is_derived_from_ordered_plan(plan_dict: dict) -> None:
dag = ProofPlanDAG.from_plan(KernelPlan.model_validate(plan_dict))
assert [node.node_id for node in dag.nodes] == ["n1", "n2"]
assert dag.nodes[0].dependencies == []
assert dag.nodes[1].dependencies == ["n1"]
assert dag.terminal_node_id == "n2"
assert dag.method_labels_payload()[0].startswith("n1: ")
def test_dag_rejects_cycle() -> None:
with pytest.raises(ValidationError, match="cycle"):
ProofPlanDAG.model_validate(
{
"nodes": [
{
"node_id": "n1",
"method_label": "first",
"dependencies": ["n2"],
},
{
"node_id": "n2",
"method_label": "second",
"dependencies": ["n1"],
},
],
"terminal_node_id": "n2",
}
)
def test_path_dag_rejects_skipped_dependency() -> None:
with pytest.raises(ValidationError, match="lead to the terminal"):
ProofPlanDAG.model_validate(
{
"nodes": [
{
"node_id": "n1",
"method_label": "first",
"dependencies": [],
},
{
"node_id": "n2",
"method_label": "second",
"dependencies": [],
},
],
"terminal_node_id": "n2",
}
)
def test_judge_verdict_requires_every_dag_node(plan_dict: dict) -> None:
dag = ProofPlanDAG.from_plan(KernelPlan.model_validate(plan_dict))
verdict = JudgeVerdict(
verdict="accept",
step_by_step_check="n1 is valid; n2 is valid",
)
assert verdict.validate_coverage(dag) is verdict
with pytest.raises(ValueError, match="does not cover DAG nodes.*n2"):
JudgeVerdict(
verdict="accept",
step_by_step_check="n1 is valid",
).validate_coverage(dag)
def test_dag_labels_must_match_prompt_a_plan(plan_dict: dict) -> None:
plan = KernelPlan.model_validate(plan_dict)
dag = ProofPlanDAG.from_plan(plan)
changed_plan = plan.model_copy(
update={"core_steps": ["a different method", plan.core_steps[1]]}
)
with pytest.raises(ValueError, match="must equal Prompt-A core_steps"):
dag.validate_plan(changed_plan)
|
