122 lines
5.0 KiB
Python
122 lines
5.0 KiB
Python
import json
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from evals.run import (
|
|
ChatResult,
|
|
DEFAULT_CASES_DIR,
|
|
check_expectations,
|
|
load_cases,
|
|
parse_sse,
|
|
run_case,
|
|
transcript_to_json,
|
|
)
|
|
|
|
|
|
def sse_response(*events):
|
|
body = "".join(
|
|
f"event: {name}\ndata: {json.dumps(payload, ensure_ascii=False)}\n\n"
|
|
for name, payload in events
|
|
)
|
|
return httpx.Response(200, text=body)
|
|
|
|
|
|
def test_parse_form_deltas_and_text():
|
|
result = parse_sse(sse_response(
|
|
("stage_code", {"nextStageCode": "1002"}),
|
|
("text_delta", {"text": "请问事故"}),
|
|
("formUpdate", {"has_non_motor_vehicle_hit": True}),
|
|
("text_delta", {"text": "是什么时候发生的？"}),
|
|
("formUpdate", {"vehicle_count": 0}),
|
|
))
|
|
assert result.stage_codes == ["1002"]
|
|
assert result.text == "请问事故是什么时候发生的？"
|
|
assert result.form_update == {"has_non_motor_vehicle_hit": True, "vehicle_count": 0}
|
|
|
|
|
|
@pytest.mark.parametrize("update,reply,error", [
|
|
({"has_non_motor_vehicle_hit": False}, "事故什么时候发生的？", "has_non_motor_vehicle_hit"),
|
|
({"vehicle_count": 1}, "事故什么时候发生的？", "vehicle_count"),
|
|
({"vehicle_count": False}, "事故什么时候发生的？", "vehicle_count"),
|
|
({}, "有没有涉及非机动车？", "replyMatches"),
|
|
({"is_on_scene": True}, "事故什么时候发生的？", "is_on_scene"),
|
|
])
|
|
def test_regression_assertions_reject_bad_extraction_and_repeated_questions(update, reply, error):
|
|
form = {"has_non_motor_vehicle_hit": True, "vehicle_count": 0, **update}
|
|
failures = check_expectations(ChatResult(["1002"], [], reply), {
|
|
"stageCode": "1002",
|
|
"form": {"has_non_motor_vehicle_hit": True, "vehicle_count": 0},
|
|
"formAbsent": ["is_on_scene"],
|
|
"replyMatches": "(什么时候|几点|事故.*时间)",
|
|
"replyNotMatches": "(有没有|是否).*非机动车",
|
|
}, form)
|
|
assert any(error in failure for failure in failures)
|
|
|
|
|
|
def test_description_assertion_and_sse_error():
|
|
expect = {"stageCode": "1002", "formMatches": {
|
|
"accident_cause": "^其他：(?=.*非机动车)(?=.*石墩).+",
|
|
}}
|
|
form = {"accident_cause": "其他：非机动车自行撞到路边石墩"}
|
|
assert not check_expectations(ChatResult(["1002"], []), expect, form)
|
|
assert check_expectations(ChatResult(["1002"], [{"message": "failed"}]), expect, form)
|
|
assert check_expectations(ChatResult(["1002"], []), expect, {"accident_cause": "其他"})
|
|
|
|
|
|
@pytest.mark.parametrize("last_update,passed", [
|
|
({}, True),
|
|
({"has_non_motor_vehicle_hit": False, "vehicle_count": 1}, False),
|
|
])
|
|
def test_run_case_checks_cumulative_form_and_requests_extraction(last_update, passed):
|
|
calls = []
|
|
responses = iter([
|
|
{"has_non_motor_vehicle_hit": True},
|
|
{"vehicle_count": 0},
|
|
last_update,
|
|
])
|
|
|
|
def handler(request):
|
|
payload = json.loads(request.content)
|
|
calls.append((request.method, payload))
|
|
if request.method == "DELETE":
|
|
return httpx.Response(200, json={"code": "200"})
|
|
assert payload["needFormUpdate"] is True
|
|
return sse_response(
|
|
("stage_code", {"nextStageCode": "1002"}),
|
|
("formUpdate", next(responses)),
|
|
("text_delta", {"text": "事故什么时候发生的？"}),
|
|
)
|
|
|
|
case = {"name": "cumulative", "_path": "mock.yaml", "clientMode": "direct", "steps": [
|
|
{"input": "非机动车撞石墩", "expect": {"stageCode": "1002", "form": {"has_non_motor_vehicle_hit": True}}},
|
|
{"input": "没有别的车", "expect": {"stageCode": "1002", "form": {"vehicle_count": 0}}},
|
|
{"input": "没有人员伤亡", "expect": {"stageCode": "1002", "form": {"has_non_motor_vehicle_hit": True, "vehicle_count": 0}}},
|
|
]}
|
|
with httpx.Client(base_url="http://eval.test", transport=httpx.MockTransport(handler)) as client:
|
|
transcript = run_case(client, case)
|
|
assert transcript.passed is passed
|
|
assert calls[-1][0] == "DELETE"
|
|
assert transcript.turns[0].form == {"has_non_motor_vehicle_hit": True}
|
|
saved = transcript_to_json(transcript)
|
|
assert saved["turns"][1]["form"]["vehicle_count"] == 0
|
|
assert saved["turns"][2]["formUpdate"] == last_update
|
|
|
|
|
|
def test_explicit_correction_can_replace_prior_fact():
|
|
form = {"has_non_motor_vehicle_hit": True}
|
|
form.update({"has_non_motor_vehicle_hit": False, "vehicle_count": 1})
|
|
assert not check_expectations(ChatResult(["1002"], []), {
|
|
"stageCode": "1002", "form": {"has_non_motor_vehicle_hit": False, "vehicle_count": 1},
|
|
}, form)
|
|
|
|
|
|
def test_current_cases_load_without_removed_photo_states():
|
|
cases = load_cases(DEFAULT_CASES_DIR, [])
|
|
assert len({case["name"] for case in cases}) == len(cases)
|
|
regressions = [case for case in cases if "non_motor_vehicle" in case["name"]]
|
|
assert {case["clientMode"] for case in regressions} == {"direct", "browser_addon"}
|
|
for case in cases:
|
|
for step in case["steps"]:
|
|
assert step["expect"]["stageCode"].split(".")[-1] not in {"2004", "2016"}
|