Eric Wang
7f52f1c4d0
- fields_registry: FORM_TO_BUSINESS_KEY bidirectional mapping between the graph's six form keys (has_injury...) and the acd business keys (ywrysw...); bjrjs/ywmtc have no form source (set_info only); business_patch_from_form maps graph diffs to whitelisted business keys and filters internal/unknown keys - app_service: chat injection now maps repository business keys back to form keys; both complete() and stream() write back field diffs (FormUpdate events captured in stream, closing the skeleton gap); form keys never leak into the business repository - docs/assignment6-compat.md: mapping table documented, stream write-back backlog item resolved - tests: mapping coverage, write-back mapping end-to-end, reverse mapping injection (bjrjs/ywmtc excluded), stream FormUpdate write-back Verified: 205 unit tests
223 lines
8.4 KiB
Python
223 lines
8.4 KiB
Python
"""作业 6：应用服务四接口组合测试。
|
||
|
||
验收演示对应（计划作业 6）：
|
||
创建会话 -> set_info 写车牌 -> get_info 读回 -> chat 使用车牌继续核实
|
||
-> 分组读值 -> 删除 -> 再读与再次聊天均不泄漏旧数据。
|
||
|
||
spy 断言：set_info / get_info / delete_session 全程零模型调用。
|
||
"""
|
||
|
||
import json
|
||
|
||
import pytest
|
||
|
||
from src.backends.chat import ChatResult, TextDelta, FormUpdate
|
||
from src.services.app_service import AccidentAppService, SET_INFO_INVALID_KEY_MSG
|
||
from src.services.repository import InMemorySessionRepository
|
||
|
||
|
||
class SpyBackend:
|
||
"""记录调用的 ChatBackend 替身；form_update 模拟图产出的字段差量。"""
|
||
|
||
def __init__(self, form_update=None):
|
||
self.calls = [] # complete 收到的 ChatInput
|
||
self.stream_calls = []
|
||
self._form_update = form_update or {}
|
||
# 用调用次数当“模型调用”spy：真实 backend 每次 complete 至少一次 LLM
|
||
self.llm_calls = 0
|
||
|
||
async def complete(self, chat_input):
|
||
self.calls.append(chat_input)
|
||
self.llm_calls += 1
|
||
return ChatResult(
|
||
content="<state>1002</state>收到",
|
||
status_code=None,
|
||
form_update=dict(self._form_update),
|
||
)
|
||
|
||
async def stream(self, chat_input):
|
||
self.stream_calls.append(chat_input)
|
||
self.llm_calls += 1
|
||
yield TextDelta("<state>1002</state>")
|
||
if self._form_update:
|
||
yield FormUpdate(dict(self._form_update))
|
||
yield TextDelta("收到")
|
||
|
||
|
||
def make_service(form_update=None):
|
||
backend = SpyBackend(form_update=form_update)
|
||
service = AccidentAppService(backend, InMemorySessionRepository())
|
||
return service, backend
|
||
|
||
|
||
# ---- set/get：白名单、分组、外部表示 ----
|
||
|
||
def test_set_then_get_single_key_roundtrip():
|
||
service, backend = make_service()
|
||
code, msg = service.set_info("s1", "hphm1", "苏B12345")
|
||
assert (code, msg) == ("200", "")
|
||
assert json.loads(service.get_info("s1", "hphm1")) == "苏B12345"
|
||
# set/get 零模型调用
|
||
assert backend.llm_calls == 0
|
||
|
||
|
||
def test_get_groups_match_existing_contract():
|
||
service, _ = make_service()
|
||
service.set_info("s1", "ywrysw", "0") # False
|
||
service.set_info("s1", "jdcsl", "0") # int 0
|
||
service.set_info("s1", "xm2", "张三")
|
||
|
||
acd = json.loads(service.get_info("s1", "acdinfo"))
|
||
assert acd["ywrysw"] == "0" # False -> "0"
|
||
assert acd["jdcsl"] == "0" # int 0 -> "0"（内部类型不同）
|
||
assert acd["sgyy"] == "" # 未知 -> 空串
|
||
|
||
human2 = json.loads(service.get_info("s1", "acdhuman2"))
|
||
assert human2["xm2"] == "张三"
|
||
assert human2["sfzmwh2"] == "" # 全字段齐全，未填为空
|
||
|
||
all_groups = json.loads(service.get_info("s1", "all"))
|
||
assert set(all_groups) == {"acdinfo", "acdhuman1", "acdhuman2"}
|
||
|
||
|
||
def test_get_unknown_key_returns_json_empty_string():
|
||
"""兼容现状：未知 key 返回 JSON 空串 '""'。"""
|
||
service, _ = make_service()
|
||
service.set_info("s1", "hphm1", "苏B12345")
|
||
assert service.get_info("s1", "no_such_key") == '""'
|
||
assert service.get_info("ghost-session", "hphm1") == '""'
|
||
|
||
|
||
def test_set_unknown_or_internal_key_rejected():
|
||
service, backend = make_service()
|
||
code, msg = service.set_info("s1", "no_such_key", "x")
|
||
assert code == "500" and msg == SET_INFO_INVALID_KEY_MSG
|
||
code, _ = service.set_info("s1", "stage_code", "1002")
|
||
assert code == "500" # 内部字段保护
|
||
assert service.get_info("s1", "hphm1") == '""' # 未被污染
|
||
assert backend.llm_calls == 0
|
||
|
||
|
||
def test_set_rejects_unparsable_value():
|
||
service, _ = make_service()
|
||
code, msg = service.set_info("s1", "ywrysw", "maybe")
|
||
assert code == "500" and "无法解析" in msg
|
||
|
||
|
||
# ---- chat：注入 + 写回 ----
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_chat_sees_set_info_values_and_writes_back_patch():
|
||
"""set_info 写入 -> chat 输入注入 -> 图差量写回 -> get_info 读到。"""
|
||
service, backend = make_service(form_update={"ywrysw": False})
|
||
service.set_info("s1", "hphm1", "苏B12345")
|
||
service.set_info("s1", "xm2", "李四")
|
||
|
||
result = await service.chat(
|
||
_input("s1", "我是刚才报警的驾驶员")
|
||
)
|
||
assert result.content == "<state>1002</state>收到"
|
||
|
||
# 图输入必须携带 repository 最新业务字段（车牌供核实使用）
|
||
injected = backend.calls[0].injected_fields
|
||
assert injected["person_fields"]["hphm1"] == "苏B12345"
|
||
assert injected["person_fields"]["xm2"] == "李四"
|
||
|
||
# 图产出的差量写回 repository，get_info 立即可读
|
||
assert json.loads(service.get_info("s1", "acdinfo"))["ywrysw"] == "0"
|
||
assert json.loads(service.get_info("s1", "hphm1")) == "苏B12345"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_chat_form_patch_mapped_to_business_keys():
|
||
"""图表单键差量经映射写回业务键（作业 6 映射的端到端）。"""
|
||
service, backend = make_service(form_update={
|
||
"has_injury": False, "vehicle_count": 2, "accident_cause": "追尾",
|
||
})
|
||
await service.chat(_input("s1", "两辆汽车追尾，没有人受伤"))
|
||
acd = json.loads(service.get_info("s1", "acdinfo"))
|
||
assert acd["ywrysw"] == "0"
|
||
assert acd["jdcsl"] == "2"
|
||
assert acd["sgyy"] == "追尾"
|
||
# 表单键本身不落 repository（未知业务键 -> JSON 空串）
|
||
assert json.loads(service.get_info("s1", "has_injury")) == ""
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_set_info_acd_value_mapped_into_form_injection():
|
||
"""set_info 写业务键 -> chat 注入为图表单键（反向映射）。"""
|
||
service, backend = make_service()
|
||
service.set_info("s1", "ywrysw", "0")
|
||
service.set_info("s1", "jdcsl", "2")
|
||
service.set_info("s1", "bjrjs", "1") # 无表单来源，不注入
|
||
await service.chat(_input("s1", "继续办理"))
|
||
injected = backend.calls[0].injected_fields
|
||
assert injected["accident_fields"] == {"has_injury": False, "vehicle_count": 2}
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_chat_without_business_state_passes_no_injection():
|
||
service, backend = make_service()
|
||
await service.chat(_input("fresh-session", "你好"))
|
||
assert backend.calls[0].injected_fields is None
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_stream_injects_fields_writes_back_form_update():
|
||
"""流式：注入同 chat；FormUpdate 事件差量捕获写回后透传。"""
|
||
service, backend = make_service(form_update={"has_injury": False})
|
||
service.set_info("s1", "xm1", "王五")
|
||
events = [
|
||
event async for event in service.stream(_input("s1", "继续办理"))
|
||
]
|
||
kinds = [type(event).__name__ for event in events]
|
||
assert kinds == ["TextDelta", "FormUpdate", "TextDelta"]
|
||
assert backend.stream_calls[0].injected_fields["person_fields"]["xm1"] == "王五"
|
||
# FormUpdate 差量已写回 repository（表单键 -> 业务键）
|
||
assert json.loads(service.get_info("s1", "acdinfo"))["ywrysw"] == "0"
|
||
|
||
|
||
# ---- delete：清理 + 幂等 ----
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_delete_clears_business_state_and_is_idempotent():
|
||
drops = []
|
||
|
||
async def drop_checkpoint(session_id):
|
||
drops.append(session_id)
|
||
|
||
service, backend = make_service()
|
||
service.set_info("s1", "hphm1", "苏B12345")
|
||
await service.chat(_input("s1", "你好"))
|
||
|
||
await service.delete_session("s1", drop_checkpoint=drop_checkpoint)
|
||
assert drops == ["s1"]
|
||
# 再读为空、再聊天不注入旧数据
|
||
assert service.get_info("s1", "hphm1") == '""'
|
||
await service.chat(_input("s1", "还在吗"))
|
||
assert backend.calls[-1].injected_fields is None
|
||
|
||
# 重复删除固定行为：成功返回；checkpoint 清理会再次触发，
|
||
# 但缺失时静默容忍（幂等语义由 drop_checkpoint 内部保证）
|
||
await service.delete_session("s1", drop_checkpoint=drop_checkpoint)
|
||
assert drops == ["s1", "s1"]
|
||
assert backend.llm_calls == 2 # 两次 chat 各一次；set/get/delete 零调用
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_delete_tolerates_missing_checkpoint():
|
||
async def missing(_session_id):
|
||
raise RuntimeError("no such thread")
|
||
|
||
service, _ = make_service()
|
||
service.set_info("s1", "xm1", "王五")
|
||
# checkpoint 不存在视为删除成功的一部分，不向上抛
|
||
await service.delete_session("s1", drop_checkpoint=missing)
|
||
assert service.get_info("s1", "xm1") == '""'
|
||
|
||
|
||
def _input(session_id, text):
|
||
from src.backends.chat import ChatInput
|
||
|
||
return ChatInput(session_id=session_id, text=text, client_mode="direct")
|