88 lines
3.2 KiB
Python
88 lines
3.2 KiB
Python
"""Hermes 移植模块单元测试：轮次预算 / 抖动退避 / 工具停滞护栏。"""
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
|
|
from app.hermes_core import (IterationBudget, StallGuard, jittered_backoff,
|
|
parse_retry_after_seconds, retry_sleep,
|
|
tool_call_fingerprint)
|
|
|
|
|
|
def test_budget_consumes_and_exhausts():
|
|
budget = IterationBudget(3)
|
|
assert budget.consume() is True
|
|
assert budget.consume() is True
|
|
assert budget.consume() is True
|
|
assert budget.consume() is False
|
|
assert budget.used == 3
|
|
assert budget.remaining == 0
|
|
|
|
|
|
def test_budget_refund_returns_capacity():
|
|
budget = IterationBudget(1)
|
|
assert budget.consume() is True
|
|
assert budget.consume() is False
|
|
budget.refund()
|
|
assert budget.consume() is True
|
|
|
|
|
|
def test_budget_thread_safe_under_concurrency():
|
|
budget = IterationBudget(10)
|
|
with ThreadPoolExecutor(max_workers=8) as pool:
|
|
results = list(pool.map(lambda _: budget.consume(), range(50)))
|
|
assert sum(1 for ok in results if ok) == 10
|
|
assert budget.used == 10
|
|
|
|
|
|
def test_retry_after_numeric_and_headers():
|
|
assert parse_retry_after_seconds("2.5") == 2.5
|
|
assert parse_retry_after_seconds({"Retry-After": "3"}) == 3.0
|
|
assert parse_retry_after_seconds({"retry-after": 1}) == 1.0
|
|
assert parse_retry_after_seconds(None) is None
|
|
assert parse_retry_after_seconds("not-a-number") is None
|
|
|
|
|
|
def test_retry_after_http_date():
|
|
from datetime import datetime, timedelta, timezone
|
|
future = datetime.now(timezone.utc) + timedelta(seconds=7)
|
|
header = future.strftime("%a, %d %b %Y %H:%M:%S GMT")
|
|
value = parse_retry_after_seconds(header)
|
|
assert value is not None and 5 <= value <= 8
|
|
|
|
|
|
def test_jittered_backoff_stays_in_range():
|
|
for attempt in range(8):
|
|
for _ in range(30):
|
|
delay = jittered_backoff(attempt)
|
|
assert 0 <= delay <= min(24.0, 1.6 * (2 ** attempt)) + 1e-9
|
|
|
|
|
|
def test_retry_sleep_prefers_retry_after():
|
|
assert retry_sleep(3, retry_after="9") == 9.0
|
|
assert 0 <= retry_sleep(2, retry_after=None) <= 1.6 * 4
|
|
|
|
|
|
def test_fingerprint_distinguishes_arguments():
|
|
a = tool_call_fingerprint("knowledge_search", {"q": "催情"})
|
|
b = tool_call_fingerprint("knowledge_search", {"q": "催情"})
|
|
c = tool_call_fingerprint("knowledge_search", {"q": "缩阴"})
|
|
d = tool_call_fingerprint("web_search", {"q": "催情"})
|
|
assert a == b and a != c and a != d
|
|
|
|
|
|
def test_stall_guard_notice_then_halt():
|
|
guard = StallGuard()
|
|
fp = tool_call_fingerprint("read_agent_memory", {})
|
|
assert guard.observe(fp) == "ok"
|
|
assert guard.observe(fp) == "ok"
|
|
assert guard.observe(fp) == "notice"
|
|
assert guard.observe(fp) == "notice"
|
|
assert guard.observe(fp) == "halt"
|
|
|
|
|
|
def test_stall_guard_different_calls_do_not_accumulate():
|
|
guard = StallGuard()
|
|
assert guard.observe(tool_call_fingerprint("knowledge_search", {"q": "1"})) == "ok"
|
|
assert guard.observe(tool_call_fingerprint("knowledge_search", {"q": "2"})) == "ok"
|
|
assert guard.observe(tool_call_fingerprint("knowledge_search", {"q": "1"})) == "ok"
|
|
assert guard.observe(tool_call_fingerprint("knowledge_search", {"q": "2"})) == "ok"
|
|
assert all(count < 3 for count in guard._counts.values())
|