5dd1bc8535
期望始终派生隐式判定并与显式规则叠加执行：rule_logic 只组合显式规则， 期望是叠加其上的硬约束，任一不满足即用例不通过。隐式判定以 EvalResult 同构落库，reason 前缀 [期望] 标明来源。连通用例（无规则无期望）行为不变。
440 lines
16 KiB
Python
440 lines
16 KiB
Python
"""Unit tests for the async EvalEngine.
|
||
|
||
Focus on the new async behavior: cooperative cancellation, configurable
|
||
timeouts, and concurrent case execution via the semaphore.
|
||
"""
|
||
|
||
import asyncio
|
||
from typing import Any
|
||
|
||
import pytest
|
||
|
||
from agenteval.channels.base import SendResult
|
||
from agenteval.evaluation.engine import CancelledError, EvalEngine, TimeoutConfig
|
||
from agenteval.models import (
|
||
Case, CaseType, ChannelType, EvalTarget, Expectation, PlatformType,
|
||
RunStatus, Scenario, TargetStatus,
|
||
)
|
||
from agenteval.storage.repository import RunRepository
|
||
from tests.unit.mock_channel import MockChannel
|
||
|
||
|
||
def _make_target() -> EvalTarget:
|
||
return EvalTarget(
|
||
id="t-1",
|
||
name="mock-target",
|
||
platform=PlatformType.AI_DIGITAL_EMPLOYEE,
|
||
channel_type=ChannelType.TUTU_API,
|
||
channel_config={
|
||
"base_url": "http://mock",
|
||
"token": "x",
|
||
"tenant": "t",
|
||
"chat_channel_id": "c",
|
||
"chat_contact_id": "u",
|
||
},
|
||
status=TargetStatus.ACTIVE,
|
||
)
|
||
|
||
|
||
def _build_engine(
|
||
scenario: Scenario,
|
||
channel: MockChannel,
|
||
*,
|
||
cancel_token: asyncio.Event | None = None,
|
||
timeout_config: TimeoutConfig | None = None,
|
||
max_concurrent_cases: int = 1,
|
||
session=None,
|
||
) -> EvalEngine:
|
||
"""Construct an EvalEngine with a MockChannel injected.
|
||
|
||
The factory still runs (needs a valid channel_config on the target) but
|
||
we immediately replace the channel with the mock.
|
||
"""
|
||
engine = EvalEngine(
|
||
target=_make_target(),
|
||
scenario=scenario,
|
||
session=session,
|
||
cancel_token=cancel_token or asyncio.Event(),
|
||
timeout_config=timeout_config,
|
||
max_concurrent_cases=max_concurrent_cases,
|
||
)
|
||
engine.channel = channel
|
||
return engine
|
||
|
||
|
||
# ── basic happy path ─────────────────────────────────────────────────────
|
||
|
||
async def test_run_single_case_completes(db_session):
|
||
scenario = Scenario(
|
||
id="s-1", name="single",
|
||
cases=[Case(id="c1", type=CaseType.SINGLE, messages=["hello"])],
|
||
)
|
||
channel = MockChannel()
|
||
engine = _build_engine(scenario, channel, session=db_session)
|
||
|
||
run = await engine.run()
|
||
|
||
assert run.status == RunStatus.COMPLETED
|
||
assert run.summary["total_cases"] == 1
|
||
assert run.summary["passed_cases"] == 1
|
||
assert channel.send_calls == 1
|
||
assert channel.poll_calls == 1
|
||
|
||
|
||
async def test_run_multi_turn_collects_all_turns(db_session):
|
||
scenario = Scenario(
|
||
id="s-1", name="multi",
|
||
cases=[Case(id="c1", type=CaseType.MULTI_TURN, messages=["a", "b", "c"])],
|
||
)
|
||
channel = MockChannel()
|
||
engine = _build_engine(scenario, channel, session=db_session)
|
||
|
||
run = await engine.run()
|
||
|
||
assert run.status == RunStatus.COMPLETED
|
||
assert channel.sent == ["a", "b", "c"]
|
||
assert channel.send_calls == 3
|
||
# Each turn polls once.
|
||
assert channel.poll_calls == 3
|
||
|
||
|
||
async def test_run_multiple_cases(db_session):
|
||
scenario = Scenario(
|
||
id="s-1", name="multi-case",
|
||
cases=[
|
||
Case(id="c1", type=CaseType.SINGLE, messages=["one"]),
|
||
Case(id="c2", type=CaseType.SINGLE, messages=["two"]),
|
||
Case(id="c3", type=CaseType.SINGLE, messages=["three"]),
|
||
],
|
||
)
|
||
channel = MockChannel()
|
||
engine = _build_engine(scenario, channel, session=db_session)
|
||
|
||
run = await engine.run()
|
||
|
||
assert run.status == RunStatus.COMPLETED
|
||
assert run.summary["total_cases"] == 3
|
||
assert channel.sent == ["one", "two", "three"]
|
||
|
||
|
||
# ── progress callback ────────────────────────────────────────────────────
|
||
|
||
async def test_progress_callback_receives_events(db_session):
|
||
scenario = Scenario(
|
||
id="s-1", name="single",
|
||
cases=[Case(id="c1", type=CaseType.SINGLE, messages=["hi"])],
|
||
)
|
||
events: list[tuple[str, dict]] = []
|
||
|
||
async def cb(event: str, data: dict[str, Any]) -> None:
|
||
events.append((event, data))
|
||
|
||
channel = MockChannel()
|
||
engine = _build_engine(scenario, channel, session=db_session)
|
||
await engine.run(progress_callback=cb)
|
||
|
||
event_names = [e[0] for e in events]
|
||
assert "case_start" in event_names
|
||
assert "turn_start" in event_names
|
||
assert "turn_end" in event_names
|
||
assert "case_end" in event_names
|
||
assert "run_completed" in event_names
|
||
|
||
|
||
async def test_sync_progress_callback_also_works(db_session):
|
||
"""Engine must accept sync callbacks (CLI uses them)."""
|
||
scenario = Scenario(
|
||
id="s-1", name="single",
|
||
cases=[Case(id="c1", type=CaseType.SINGLE, messages=["hi"])],
|
||
)
|
||
seen: list[str] = []
|
||
|
||
def sync_cb(event: str, data: dict) -> None:
|
||
seen.append(event)
|
||
|
||
channel = MockChannel()
|
||
engine = _build_engine(scenario, channel, session=db_session)
|
||
await engine.run(progress_callback=sync_cb)
|
||
|
||
assert "run_completed" in seen
|
||
|
||
|
||
# ── cancellation ─────────────────────────────────────────────────────────
|
||
|
||
async def test_cancel_before_run_marks_failed(db_session):
|
||
scenario = Scenario(
|
||
id="s-1", name="single",
|
||
cases=[Case(id="c1", type=CaseType.SINGLE, messages=["hi"])],
|
||
)
|
||
cancel_token = asyncio.Event()
|
||
cancel_token.set() # already cancelled
|
||
|
||
channel = MockChannel()
|
||
engine = _build_engine(scenario, channel, cancel_token=cancel_token, session=db_session)
|
||
run = await engine.run()
|
||
|
||
assert run.status == RunStatus.FAILED
|
||
assert run.summary["error"]["code"] == "cancelled_by_user"
|
||
# Engine must NOT have called the channel.
|
||
assert channel.send_calls == 0
|
||
|
||
|
||
async def test_cancel_mid_run_stops_after_current_case(db_session):
|
||
"""Cancelling between cases should stop further cases from running."""
|
||
scenario = Scenario(
|
||
id="s-1", name="multi",
|
||
cases=[
|
||
Case(id="c1", type=CaseType.SINGLE, messages=["a"]),
|
||
Case(id="c2", type=CaseType.SINGLE, messages=["b"]),
|
||
Case(id="c3", type=CaseType.SINGLE, messages=["c"]),
|
||
],
|
||
)
|
||
cancel_token = asyncio.Event()
|
||
|
||
# Slow channel so we have time to set the cancel token from another task.
|
||
channel = MockChannel(reply_delay=0.05)
|
||
|
||
async def cancel_soon() -> None:
|
||
await asyncio.sleep(0.08)
|
||
cancel_token.set()
|
||
|
||
engine = _build_engine(scenario, channel, cancel_token=cancel_token, session=db_session)
|
||
|
||
run, _ = await asyncio.gather(
|
||
engine.run(),
|
||
cancel_soon(),
|
||
)
|
||
|
||
assert run.status == RunStatus.FAILED
|
||
assert run.summary["error"]["code"] == "cancelled_by_user"
|
||
# At least one case ran, but not all three.
|
||
assert 1 <= channel.send_calls < 3
|
||
|
||
|
||
# ── timeouts ─────────────────────────────────────────────────────────────
|
||
|
||
async def test_poll_timeout_records_missing_reply(db_session):
|
||
"""When the channel returns no reply within the timeout, the turn is saved
|
||
with reply=None but the engine keeps going (doesn't crash)."""
|
||
scenario = Scenario(
|
||
id="s-1", name="single",
|
||
cases=[Case(id="c1", type=CaseType.SINGLE, messages=["hi"])],
|
||
)
|
||
channel = MockChannel(missing_reply=True)
|
||
engine = _build_engine(
|
||
scenario, channel,
|
||
timeout_config=TimeoutConfig(poll_reply=0.1),
|
||
session=db_session,
|
||
)
|
||
|
||
run = await engine.run()
|
||
|
||
# Engine completes (doesn't hang), turn has no reply.
|
||
assert run.status == RunStatus.COMPLETED
|
||
turns = RunRepository(db_session).get_turns(run.id)
|
||
assert len(turns) == 1
|
||
assert turns[0].reply is None
|
||
|
||
|
||
# ── concurrency ──────────────────────────────────────────────────────────
|
||
|
||
async def test_concurrent_cases_respect_semaphore(db_session):
|
||
"""With max_concurrent_cases=2, at most 2 cases should be in-flight."""
|
||
scenario = Scenario(
|
||
id="s-1", name="concurrent",
|
||
cases=[
|
||
Case(id=f"c{i}", type=CaseType.SINGLE, messages=[f"m{i}"])
|
||
for i in range(5)
|
||
],
|
||
)
|
||
in_flight = {"n": 0, "max": 0}
|
||
lock = asyncio.Lock()
|
||
|
||
class TrackedChannel(MockChannel):
|
||
async def send(self, content: str, **kwargs):
|
||
async with lock:
|
||
in_flight["n"] += 1
|
||
in_flight["max"] = max(in_flight["max"], in_flight["n"])
|
||
await asyncio.sleep(0.05)
|
||
result = await super().send(content, **kwargs)
|
||
async with lock:
|
||
in_flight["n"] -= 1
|
||
return result
|
||
|
||
channel = TrackedChannel()
|
||
engine = _build_engine(
|
||
scenario, channel,
|
||
max_concurrent_cases=2,
|
||
session=db_session,
|
||
)
|
||
|
||
run = await engine.run()
|
||
|
||
assert run.status == RunStatus.COMPLETED
|
||
assert channel.send_calls == 5
|
||
assert in_flight["max"] <= 2
|
||
|
||
|
||
# ── send failure ─────────────────────────────────────────────────────────
|
||
|
||
async def test_send_failure_aborts_case(db_session):
|
||
"""When channel.send fails, the case is marked failed but the engine
|
||
continues with the next case."""
|
||
scenario = Scenario(
|
||
id="s-1", name="mixed",
|
||
cases=[
|
||
Case(id="c1", type=CaseType.SINGLE, messages=["a"]),
|
||
Case(id="c2", type=CaseType.SINGLE, messages=["b"]),
|
||
],
|
||
)
|
||
call_count = {"n": 0}
|
||
|
||
class FlakeyChannel(MockChannel):
|
||
async def send(self, content, **kwargs):
|
||
call_count["n"] += 1
|
||
if call_count["n"] == 1:
|
||
return SendResult(ok=False, error="boom")
|
||
return await super().send(content, **kwargs)
|
||
|
||
channel = FlakeyChannel()
|
||
engine = _build_engine(scenario, channel, session=db_session)
|
||
|
||
run = await engine.run()
|
||
|
||
assert run.status == RunStatus.COMPLETED
|
||
assert run.summary["failed_cases"] == 1
|
||
assert run.summary["passed_cases"] == 1
|
||
|
||
|
||
# ── dynamic case generation failure ──────────────────────────────────────
|
||
|
||
async def test_dynamic_generation_failure_records_case_error(db_session):
|
||
"""A dynamic case whose message generation fails must persist the reason
|
||
into run.summary.case_errors — not just emit it transiently. Otherwise a
|
||
run shows 0 rules / failed with no discoverable cause."""
|
||
scenario = Scenario(
|
||
id="s-1", name="dynamic",
|
||
cases=[Case(id="dyn-1", type=CaseType.DYNAMIC, prompt="生成问题", turns=3)],
|
||
llm_config=None, # 缺 llm_config → 生成消息立即失败
|
||
)
|
||
channel = MockChannel()
|
||
engine = _build_engine(scenario, channel, session=db_session)
|
||
|
||
run = await engine.run()
|
||
|
||
assert run.status == RunStatus.COMPLETED
|
||
assert run.summary["failed_cases"] == 1
|
||
assert run.summary["total_rules"] == 0
|
||
# 关键：失败原因被持久化到 summary，可在报告 / DB 查看
|
||
assert "case_errors" in run.summary
|
||
assert run.summary["case_errors"][0]["case_id"] == "dyn-1"
|
||
assert "llm_config" in run.summary["case_errors"][0]["error"]
|
||
# 被测通道不应被调用（生成阶段就失败了）
|
||
assert channel.send_calls == 0
|
||
|
||
|
||
# ── expectation + explicit rules are additive (ticket 01) ────────────────
|
||
# MockChannel replies "echo: q-1", so keyword "echo" passes, "__NOPE__" fails.
|
||
|
||
from agenteval.models import EvalRuleConfig, RuleLogic # noqa: E402
|
||
|
||
|
||
def _case_with(
|
||
*,
|
||
rules: list[EvalRuleConfig] | None = None,
|
||
expectations: Expectation | None = None,
|
||
rule_logic: RuleLogic = RuleLogic.ALL,
|
||
rule_pass_threshold: float = 0.6,
|
||
) -> Case:
|
||
return Case(
|
||
id="c1", type=CaseType.SINGLE, messages=["hi"],
|
||
eval_rules=rules or [],
|
||
expectations=expectations or Expectation(),
|
||
rule_logic=rule_logic,
|
||
rule_pass_threshold=rule_pass_threshold,
|
||
)
|
||
|
||
|
||
async def test_expectation_fails_case_even_when_rules_pass(db_session):
|
||
"""期望不满足 → 用例不通过，即使显式规则全部通过。"""
|
||
scenario = Scenario(id="s1", name="s", cases=[_case_with(
|
||
rules=[EvalRuleConfig(type="keyword_match", params={"keywords": ["echo"]})],
|
||
expectations=Expectation(keywords_include=["__NOPE__"]),
|
||
)])
|
||
channel = MockChannel()
|
||
engine = _build_engine(scenario, channel, session=db_session)
|
||
|
||
run = await engine.run()
|
||
|
||
assert run.status == RunStatus.COMPLETED
|
||
assert run.summary["failed_cases"] == 1
|
||
|
||
|
||
async def test_expectation_and_rules_both_pass(db_session):
|
||
"""期望与规则都满足 → 通过，且期望派生判定同构落库、reason 可辨识来源。"""
|
||
scenario = Scenario(id="s1", name="s", cases=[_case_with(
|
||
rules=[EvalRuleConfig(type="keyword_match", params={"keywords": ["echo"]})],
|
||
expectations=Expectation(keywords_include=["echo"], response_time_max_ms=99999),
|
||
)])
|
||
channel = MockChannel()
|
||
engine = _build_engine(scenario, channel, session=db_session)
|
||
|
||
run = await engine.run()
|
||
|
||
assert run.summary["passed_cases"] == 1
|
||
results = RunRepository(db_session).get_results(run.id)
|
||
# 1 显式规则 + 2 期望派生（keyword + response_time）
|
||
assert len(results) == 3
|
||
implicit = [r for r in results if "期望" in r.reason]
|
||
assert len(implicit) == 2
|
||
assert all(r.passed for r in results)
|
||
|
||
|
||
async def test_implicit_expectation_not_in_any_combination(db_session):
|
||
"""rule_logic=ANY 只组合显式规则：期望通过不能救活全败的显式规则组。"""
|
||
scenario = Scenario(id="s1", name="s", cases=[_case_with(
|
||
rules=[EvalRuleConfig(type="keyword_match", params={"keywords": ["__NOPE__"]})],
|
||
expectations=Expectation(keywords_include=["echo"]), # 通过
|
||
rule_logic=RuleLogic.ANY,
|
||
)])
|
||
channel = MockChannel()
|
||
engine = _build_engine(scenario, channel, session=db_session)
|
||
|
||
run = await engine.run()
|
||
|
||
assert run.summary["failed_cases"] == 1
|
||
|
||
|
||
async def test_implicit_expectation_is_hard_constraint_over_weighted(db_session):
|
||
"""rule_logic=WEIGHTED 达标但期望不满足 → 仍不通过（期望是硬约束）。"""
|
||
scenario = Scenario(id="s1", name="s", cases=[_case_with(
|
||
rules=[EvalRuleConfig(type="keyword_match", params={"keywords": ["echo"]}, weight=1.0)],
|
||
expectations=Expectation(keywords_include=["__NOPE__"]),
|
||
rule_logic=RuleLogic.WEIGHTED,
|
||
rule_pass_threshold=0.5, # 显式加权得分 1.0 ≥ 0.5
|
||
)])
|
||
channel = MockChannel()
|
||
engine = _build_engine(scenario, channel, session=db_session)
|
||
|
||
run = await engine.run()
|
||
|
||
assert run.summary["failed_cases"] == 1
|
||
|
||
|
||
async def test_pure_expectation_case_behavior_unchanged(db_session):
|
||
"""纯期望用例（无显式规则）：满足通过、不满足失败，与升级前一致。"""
|
||
scenario = Scenario(id="s1", name="s", cases=[
|
||
Case(id="ok", type=CaseType.SINGLE, messages=["hi"],
|
||
expectations=Expectation(keywords_include=["echo"])),
|
||
Case(id="bad", type=CaseType.SINGLE, messages=["hi"],
|
||
expectations=Expectation(keywords_include=["__NOPE__"])),
|
||
])
|
||
channel = MockChannel()
|
||
engine = _build_engine(scenario, channel, session=db_session)
|
||
|
||
run = await engine.run()
|
||
|
||
assert run.summary["passed_cases"] == 1
|
||
assert run.summary["failed_cases"] == 1
|
||
|