test_budget_gate.py
8.44 KB
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
#!/usr/bin/env python3
"""Budget gate boundary tests — TokenHub 企业级三原则落地（2026-08-06 吸收）。
TokenHub 知识核心：「企业级不是功能列表更长，而是关键行为可验证、故障可复现、
修改不容易引入回归。边界问题（长上下文/流式中断/配额耗尽）必须固化为自动化测试。」
本文件把 Hermes 预算模块（app.py _check_budget/_record_budget_spend）的
「配额耗尽」边界固化为自动化测试，防止未来改动（如 L2 加 depth_tiers /
success_weight / 缓存感知定价）引入回归。
覆盖边界：
- 配额耗尽（BUDGET_EXHAUSTED）→ 精确拦截，不产生超额调用
- 预警阈值（BUDGET_WARNING）→ 放行但打 trace
- 正常路径 → 放行
- 未知 job_type → 不受限
- 月度自动重置 → 跨月隔离
- 记账累加 → cost_per_call 单调累计
"""
import json
from datetime import datetime as _dt
import pytest
import app
def _cur_month() -> str:
"""预算模块用当前月份作 state key——测试必须用同一 key。"""
return _dt.now().strftime("%Y-%m")
@pytest.fixture(autouse=True)
def _isolated_budget_file(tmp_path, monkeypatch):
"""每个测试用独立预算文件，互不污染；测试后还原真实文件路径。"""
monkeypatch.setattr(app, "BUDGET_FILE", tmp_path / "hermes_budget.json")
yield
# =============================================================================
# 配额耗尽边界（TokenHub 三类边界问题之③）
# =============================================================================
class TestBudgetExhausted:
"""配额耗尽时精确拦截——不产生超额调用。"""
def test_exhausted_blocks_with_reason_code(self, monkeypatch):
"""已达上限 → (False, BUDGET_EXHAUSTED)，reason_code 可归因。"""
monkeypatch.setattr(
app, "_load_budget_state",
lambda: {_cur_month(): {"learn_absorb": 20.0}}, # >= monthly_limit
)
allowed, rc = app._check_budget("learn_absorb")
assert allowed is False
assert rc == app.VerifyReason.BUDGET_EXHAUSTED
def test_exhausted_returns_gate_action_block(self, monkeypatch):
"""BUDGET_EXHAUSTED 必须映射到 BLOCK 动作（spawn 层拒绝启动）。"""
monkeypatch.setattr(
app, "_load_budget_state",
lambda: {_cur_month(): {"learn_absorb": 99.0}},
)
allowed, rc = app._check_budget("learn_absorb")
assert rc == app.VerifyReason.BUDGET_EXHAUSTED
assert app.GateAction.BLOCK is not None # BLOCK 动作常量存在
def test_exactly_at_limit_blocks(self, monkeypatch):
"""恰好等于上限也拦截（>= 语义，不产生超额调用）。"""
monkeypatch.setattr(
app, "_load_budget_state",
lambda: {_cur_month(): {"learn_absorb": 20.0}},
)
allowed, _ = app._check_budget("learn_absorb")
assert allowed is False
def test_below_limit_allowed(self, monkeypatch):
"""未达上限 → 放行。"""
monkeypatch.setattr(
app, "_load_budget_state",
lambda: {_cur_month(): {"learn_absorb": 10.0}},
)
allowed, rc = app._check_budget("learn_absorb")
assert allowed is True
assert rc is None
# =============================================================================
# 预警阈值边界
# =============================================================================
class TestBudgetWarning:
"""达到预警阈值（默认 80%）→ 放行但返回 BUDGET_WARNING reason_code。"""
def test_at_warn_threshold_warns(self, monkeypatch):
monkeypatch.setattr(
app, "_load_budget_state",
lambda: {_cur_month(): {"learn_absorb": 16.0}}, # 80% of 20.0
)
allowed, rc = app._check_budget("learn_absorb")
assert allowed is True
assert rc == app.VerifyReason.BUDGET_WARNING
def test_below_warn_threshold_no_reason(self, monkeypatch):
monkeypatch.setattr(
app, "_load_budget_state",
lambda: {_cur_month(): {"learn_absorb": 15.9}},
)
allowed, rc = app._check_budget("learn_absorb")
assert allowed is True
assert rc is None
# =============================================================================
# 正常路径 + 未知 job_type
# =============================================================================
class TestBudgetNormalPath:
def test_no_state_file_allowed(self):
"""预算文件不存在（首次运行）→ 放行，不崩溃。"""
allowed, rc = app._check_budget("learn_absorb")
assert allowed is True
assert rc is None
def test_unknown_job_type_unrestricted(self):
"""无配置的 job_type 不受预算限制。"""
allowed, rc = app._check_budget("no_such_job")
assert allowed is True
assert rc is None
def test_zero_cost_job_never_exhausts(self, monkeypatch):
"""cost_per_call=0 的任务即使有记录也不触发耗尽（除以 0 保护）。"""
monkeypatch.setattr(
app, "_load_budget_state",
lambda: {"2099-01": {"learn_absorb": 0.0}},
)
allowed, _ = app._check_budget("learn_absorb")
assert allowed is True
# =============================================================================
# 月度自动重置（跨月隔离）
# =============================================================================
class TestBudgetMonthlyReset:
def test_previous_month_spend_ignored(self, monkeypatch):
"""上月已耗尽 → 本月重新计数，放行。"""
monkeypatch.setattr(
app, "_load_budget_state",
lambda: {"2000-01": {"learn_absorb": 999.0}}, # 远古月份
)
allowed, rc = app._check_budget("learn_absorb")
assert allowed is True
assert rc is None
def test_record_spend_writes_current_month(self, tmp_path):
"""记账写入当前月份 key（YYYY-MM），不写入旧月份。"""
app._record_budget_spend("learn_absorb")
state = json.loads((tmp_path / "hermes_budget.json").read_text())
keys = list(state.keys())
assert len(keys) == 1
import datetime as _dt
assert keys[0] == _dt.datetime.now().strftime("%Y-%m")
# =============================================================================
# 记账累加
# =============================================================================
class TestBudgetRecordSpend:
def test_spend_accumulates(self, tmp_path):
"""两次调用 → 2× cost_per_call，单调累计。"""
app._record_budget_spend("learn_absorb")
app._record_budget_spend("learn_absorb")
state = json.loads((tmp_path / "hermes_budget.json").read_text())
month = list(state.keys())[0]
assert state[month]["learn_absorb"] == pytest.approx(2 * 0.50)
def test_spend_isolated_per_job_type(self, tmp_path):
"""不同 job_type 独立记账。"""
app._record_budget_spend("chat_reply")
app._record_budget_spend("learn_absorb")
state = json.loads((tmp_path / "hermes_budget.json").read_text())
month = list(state.keys())[0]
assert state[month]["chat_reply"] == pytest.approx(0.05)
assert state[month]["learn_absorb"] == pytest.approx(0.50)
def test_unknown_job_type_no_record(self, tmp_path):
"""未知 job_type 不写预算文件。"""
app._record_budget_spend("no_such_job")
assert not (tmp_path / "hermes_budget.json").exists()
# =============================================================================
# End-to-end：耗尽 → BLOCK（spawn gate 四元组 trace）
# =============================================================================
class TestBudgetGateTrace:
def test_exhausted_writes_block_trace(self, monkeypatch, tmp_path):
"""模拟 spawn gate：耗尽时 trace 含 node/attempt/gate/action 四元组。"""
monkeypatch.setattr(
app, "_load_budget_state",
lambda: {"2099-01": {"learn_absorb": 20.0}},
)
allowed, rc = app._check_budget("learn_absorb")
if not allowed:
trace = (f"[node=spawn,attempt=1,gate={rc},"
f"action={app.GateAction.BLOCK}]")
assert "node=spawn" in trace
assert "attempt=1" in trace
assert f"gate={app.VerifyReason.BUDGET_EXHAUSTED}" in trace
assert "action=BLOCK" in trace