feat(services): assignment 6 full - form/business key mapping and stream write-back

- 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
This commit is contained in:
Eric Wang
2026-10-22 21:07:01 +08:00
parent 8717d9066c
commit 7f52f1c4d0
5 changed files with 144 additions and 17 deletions
+32 -2
View File
@@ -128,6 +128,33 @@ async def test_chat_sees_set_info_values_and_writes_back_patch():
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()
@@ -136,8 +163,9 @@ async def test_chat_without_business_state_passes_no_injection():
@pytest.mark.asyncio
async def test_stream_injects_fields_and_passthrough_events():
service, backend = make_service(form_update={"ywrysw": False})
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", "继续办理"))
@@ -145,6 +173,8 @@ async def test_stream_injects_fields_and_passthrough_events():
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:清理 + 幂等 ----
+32
View File
@@ -6,8 +6,11 @@ from src.services.fields_registry import (
FIELD_REGISTRY,
GROUP_KEYS,
INTERNAL_FIELD_KEYS,
FORM_TO_BUSINESS_KEY,
BUSINESS_TO_FORM_KEY,
FieldGroup,
SensitiveLevel,
business_patch_from_form,
parse_field_value,
to_external_string,
)
@@ -87,3 +90,32 @@ def test_internal_field_protection_listed():
"""图内部键登记在保护清单中,set_info 白名单不会收录它们。"""
for key in INTERNAL_FIELD_KEYS:
assert key not in FIELD_REGISTRY
def test_form_business_key_mapping_covers_form_fields():
"""图表单六键全部映射到 acd 业务键(作业 6 建立的映射)。"""
assert set(FORM_TO_BUSINESS_KEY) == {
"has_injury", "has_non_motor_vehicle_hit", "is_on_scene",
"vehicle_count", "accident_time", "accident_cause",
}
for business_key in FORM_TO_BUSINESS_KEY.values():
assert business_key in FIELD_REGISTRY
assert FIELD_REGISTRY[business_key].group is FieldGroup.ACD
def test_business_patch_from_form_maps_and_filters():
"""表单键映射、人员键原样、未知键过滤。"""
patch = business_patch_from_form({
"has_injury": False, # -> ywrysw
"vehicle_count": 2, # -> jdcsl
"xm2": "李四", # 人员键原样
"stage_code": "1002", # 内部键过滤
"no_such_form_key": "x", # 未知键过滤
})
assert patch == {"ywrysw": False, "jdcsl": 2, "xm2": "李四"}
def test_business_keys_without_form_source_not_injected():
"""bjrjs/ywmtc 无表单来源:不进入注入映射(只能 set_info 直写)。"""
assert "bjrjs" not in BUSINESS_TO_FORM_KEY
assert "ywmtc" not in BUSINESS_TO_FORM_KEY