File size: 8,270 Bytes
9a88af0 | 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 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 | """Tests for the InvoiceGuard environment."""
import sys
import os
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
from models import (
ActionType,
DecisionType,
ExceptionType,
InvoiceGuardAction,
TaskID,
)
from server.invoice_guard_environment import InvoiceGuardEnvironment
def _make_env():
return InvoiceGuardEnvironment()
class TestReset:
def test_reset_returns_observation(self):
env = _make_env()
obs = env.reset(task_id="task_1_clean_match")
assert obs.case_id != ""
assert obs.task_id == "task_1_clean_match"
assert obs.difficulty == "easy"
assert obs.done is False
def test_reset_provides_goal(self):
env = _make_env()
obs = env.reset(task_id="task_1_clean_match")
assert "accounts payable" in obs.goal.lower()
assert len(obs.goal) > 100
def test_reset_provides_suggested_actions(self):
env = _make_env()
obs = env.reset(task_id="task_1_clean_match")
assert len(obs.suggested_next_actions) > 0
assert "inspect_purchase_order" in obs.suggested_next_actions
def test_reset_has_available_actions(self):
env = _make_env()
obs = env.reset(task_id="task_1_clean_match")
assert len(obs.available_actions) == 12
assert "submit_final_resolution" in obs.available_actions
def test_reset_clean_state(self):
env = _make_env()
obs = env.reset(task_id="task_1_clean_match")
assert obs.revealed_documents == []
assert obs.findings == []
assert obs.remaining_steps > 0
def test_reset_sequential_tasks(self):
env = _make_env()
obs1 = env.reset()
obs2 = env.reset()
assert obs1.task_id != obs2.task_id or obs1.case_id != obs2.case_id
def test_all_canonical_tasks_load(self):
env = _make_env()
canonical = [
"task_1_clean_match",
"task_2_partial_receipt",
"task_3_price_variance",
"task_4_duplicate_invoice",
"task_5_mixed_discrepancy",
"task_6_false_positive_duplicate",
"task_7_retroactive_price",
"task_8_split_invoice_pattern",
"task_9_clean_from_risky_vendor",
"task_10_rounding_false_alarm",
"task_11_authorized_overship",
"task_12_corrected_resubmission",
]
for tid in canonical:
obs = env.reset(task_id=tid)
assert obs.task_id == tid
assert obs.done is False
class TestStep:
def test_investigation_action(self):
env = _make_env()
env.reset(task_id="task_1_clean_match")
action = InvoiceGuardAction(action_type=ActionType.inspect_purchase_order)
obs = env.step(action)
assert obs.done is False
assert "purchase_order" in obs.revealed_documents
assert obs.reward > 0
def test_repeated_action_penalty(self):
env = _make_env()
env.reset(task_id="task_1_clean_match")
action = InvoiceGuardAction(action_type=ActionType.inspect_purchase_order)
obs1 = env.step(action)
obs2 = env.step(action)
assert obs2.reward < obs1.reward
def test_step_decrements_remaining(self):
env = _make_env()
obs = env.reset(task_id="task_1_clean_match")
initial = obs.remaining_steps
action = InvoiceGuardAction(action_type=ActionType.inspect_purchase_order)
obs = env.step(action)
assert obs.remaining_steps == initial - 1
def test_suggested_actions_update_after_step(self):
env = _make_env()
env.reset(task_id="task_1_clean_match")
action = InvoiceGuardAction(action_type=ActionType.inspect_purchase_order)
obs = env.step(action)
assert "inspect_purchase_order" not in obs.suggested_next_actions
def test_findings_accumulate(self):
env = _make_env()
env.reset(task_id="task_1_clean_match")
a1 = InvoiceGuardAction(action_type=ActionType.inspect_purchase_order)
env.step(a1)
a2 = InvoiceGuardAction(action_type=ActionType.inspect_goods_receipt_note)
env.step(a2)
a3 = InvoiceGuardAction(action_type=ActionType.compare_quantity)
obs = env.step(a3)
assert len(obs.findings) > 0
class TestSubmission:
def test_correct_resolution(self):
env = _make_env()
env.reset(task_id="task_1_clean_match")
env.step(InvoiceGuardAction(action_type=ActionType.inspect_purchase_order))
env.step(InvoiceGuardAction(action_type=ActionType.inspect_goods_receipt_note))
env.step(InvoiceGuardAction(action_type=ActionType.compare_quantity))
action = InvoiceGuardAction(
action_type=ActionType.submit_final_resolution,
final_decision=DecisionType.approve_for_payment,
exception_type=ExceptionType.clean_match,
evidence_references=["inspect_purchase_order", "compare_quantity"],
explanation="All quantities and prices match within tolerance.",
)
obs = env.step(action)
assert obs.done is True
assert obs.reward > 0
assert obs.grader_result.get("score", 0) > 0.5
def test_wrong_decision_lower_score(self):
env = _make_env()
env.reset(task_id="task_1_clean_match")
env.step(InvoiceGuardAction(action_type=ActionType.inspect_purchase_order))
action = InvoiceGuardAction(
action_type=ActionType.submit_final_resolution,
final_decision=DecisionType.reject_invoice,
exception_type=ExceptionType.duplicate_invoice,
evidence_references=["inspect_purchase_order"],
explanation="Rejecting invoice.",
)
obs = env.step(action)
assert obs.done is True
assert obs.grader_result.get("score", 1) < 0.5
def test_missing_fields_returns_error(self):
env = _make_env()
env.reset(task_id="task_1_clean_match")
action = InvoiceGuardAction(
action_type=ActionType.submit_final_resolution,
)
obs = env.step(action)
assert obs.last_action_error is True or obs.last_action_result != ""
class TestState:
def test_state_accessible(self):
env = _make_env()
env.reset(task_id="task_1_clean_match")
state = env.state
assert state.task_id == "task_1_clean_match"
assert state.step_count == 0
def test_state_tracks_actions(self):
env = _make_env()
env.reset(task_id="task_1_clean_match")
env.step(InvoiceGuardAction(action_type=ActionType.inspect_purchase_order))
env.step(InvoiceGuardAction(action_type=ActionType.compare_quantity))
state = env.state
assert "inspect_purchase_order" in state.actions_taken
assert "compare_quantity" in state.actions_taken
assert state.step_count == 2
class TestTaskDiversity:
def test_scores_in_valid_range(self):
"""Verify all tasks produce grader scores in [0, 1]."""
env = _make_env()
for tid in [
"task_1_clean_match",
"task_2_partial_receipt",
"task_3_price_variance",
"task_4_duplicate_invoice",
"task_5_mixed_discrepancy",
"task_6_false_positive_duplicate",
]:
env.reset(task_id=tid)
env.step(InvoiceGuardAction(action_type=ActionType.inspect_purchase_order))
obs = env.step(
InvoiceGuardAction(
action_type=ActionType.submit_final_resolution,
final_decision=DecisionType.approve_for_payment,
exception_type=ExceptionType.clean_match,
evidence_references=["inspect_purchase_order"],
explanation="Approving.",
)
)
assert obs.done is True
score = obs.grader_result.get("score", -1)
assert 0.0 <= score <= 1.0
|