generated from kgod/ai-review-template
172 lines
8.1 KiB
Python
172 lines
8.1 KiB
Python
"""Builder follow-up flows: editing, continuation, selection, streaming, revision."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
from builder_flow_helpers import (
|
|
BASE,
|
|
active_component,
|
|
confirm_card,
|
|
create_builder_session,
|
|
event,
|
|
finish_education,
|
|
send_message,
|
|
start_education,
|
|
)
|
|
|
|
|
|
def test_editing_existing_entry_merges_facts_without_refilling_identity(client: TestClient) -> None:
|
|
session_id, body = create_builder_session(client)
|
|
start_education(client, session_id, body)
|
|
created = finish_education(client, session_id, body, "GPA 3.7/4.0，完成数据库课程项目。")
|
|
confirm_card(client, session_id, created)
|
|
|
|
edit_start = send_message(client, session_id, "修改 Example University 教育经历")
|
|
assert "无需重填" in edit_start["turn"]["content"]
|
|
assert not any(block["type"] == "component" for block in edit_start["turn"]["blocks"])
|
|
|
|
revised = send_message(client, session_id, "补充获得两次校级奖学金。")
|
|
card = active_component(revised)
|
|
assert card["data"]["component"] == "experience_confirm_card"
|
|
assert "GPA 3.7/4.0" in card["data"]["value"]["description"]
|
|
assert "两次校级奖学金" in card["data"]["value"]["description"]
|
|
|
|
|
|
def test_recent_confirmed_entry_accepts_natural_follow_up_without_refilling_identity(client: TestClient) -> None:
|
|
session_id, body = create_builder_session(client)
|
|
start_education(client, session_id, body)
|
|
proposal = finish_education(client, session_id, body, "GPA 3.7/4.0，完成数据库课程项目。")
|
|
confirm_card(client, session_id, proposal, use_optimized=True)
|
|
|
|
revised = send_message(client, session_id, "对了，我还获得过校级一等奖学金。")
|
|
assert "已收到这条补充信息" in revised["turn"]["content"]
|
|
card = active_component(revised)
|
|
assert card["data"]["component"] == "experience_confirm_card"
|
|
assert "GPA 3.7/4.0" in card["data"]["value"]["description"]
|
|
assert "校级一等奖学金" in card["data"]["value"]["description"]
|
|
assert not any(
|
|
block["data"].get("component") == "record_fields"
|
|
for block in revised["turn"]["blocks"]
|
|
if block["type"] == "component"
|
|
)
|
|
assert not any(
|
|
block["data"].get("module") == "builder_next_section"
|
|
for block in revised["turn"]["blocks"]
|
|
if block["type"] == "component"
|
|
)
|
|
|
|
|
|
def test_multiple_same_type_entries_offer_a_selection_card(client: TestClient) -> None:
|
|
session_id, body = create_builder_session(client)
|
|
start_education(client, session_id, body, school="First University")
|
|
first = finish_education(client, session_id, body, "GPA 3.6/4.0，完成课程项目。")
|
|
after_first = confirm_card(client, session_id, first)
|
|
start_education(client, session_id, after_first, school="Second University")
|
|
second = finish_education(client, session_id, after_first, "获得学业奖学金，参与实验室实践。")
|
|
confirm_card(client, session_id, second)
|
|
|
|
response = send_message(client, session_id, "修改教育经历")
|
|
card = active_component(response)
|
|
assert card["data"]["component"] == "choice_chips"
|
|
assert card["data"]["module"] == "builder_entry_select"
|
|
assert len(card["data"]["options"]) == 2
|
|
|
|
|
|
def test_builder_message_stream_emits_gap_and_rewrite_statuses(client: TestClient) -> None:
|
|
session_id, body = create_builder_session(client)
|
|
start_education(client, session_id, body)
|
|
with client.stream(
|
|
"POST",
|
|
f"{BASE}/sessions/{session_id}/messages/stream",
|
|
json={"content": "GPA 3.8/4.0，专业前 10%，完成数据库课程项目。"},
|
|
) as response:
|
|
assert response.status_code == 200
|
|
raw = b"".join(response.iter_bytes()).decode("utf-8")
|
|
|
|
events = [line.removeprefix("event: ") for line in raw.splitlines() if line.startswith("event: ")]
|
|
frames = [line.removeprefix("data: ") for line in raw.splitlines() if line.startswith("data: ")]
|
|
statuses = [json.loads(frame)["phase"] for event_name, frame in zip(events, frames) if event_name == "status"]
|
|
assert statuses == ["structuring", "checking_gaps", "rewriting"]
|
|
assert "delta" in events
|
|
assert events[-1] == "complete"
|
|
assert json.loads(frames[-1])["stage"] == "BUILDER_CONVERSATION"
|
|
|
|
|
|
def test_recent_entry_stream_emits_structuring_and_rewriting_only(client: TestClient) -> None:
|
|
session_id, body = create_builder_session(client)
|
|
start_education(client, session_id, body)
|
|
proposal = finish_education(client, session_id, body, "GPA 3.7/4.0，完成数据库课程项目。")
|
|
confirm_card(client, session_id, proposal)
|
|
|
|
with client.stream(
|
|
"POST",
|
|
f"{BASE}/sessions/{session_id}/messages/stream",
|
|
json={"content": "对了，我还获得过校级一等奖学金。"},
|
|
) as response:
|
|
assert response.status_code == 200
|
|
raw = b"".join(response.iter_bytes()).decode("utf-8")
|
|
|
|
events = [line.removeprefix("event: ") for line in raw.splitlines() if line.startswith("event: ")]
|
|
frames = [line.removeprefix("data: ") for line in raw.splitlines() if line.startswith("data: ")]
|
|
statuses = [json.loads(frame)["phase"] for event_name, frame in zip(events, frames) if event_name == "status"]
|
|
assert statuses == ["structuring", "rewriting"]
|
|
assert events[-1] == "complete"
|
|
|
|
|
|
def test_existing_education_supplement_routes_to_the_only_saved_entry(client: TestClient) -> None:
|
|
session_id, body = create_builder_session(client)
|
|
start_education(client, session_id, body)
|
|
proposal = finish_education(client, session_id, body, "GPA 3.7/4.0，专业前 20%，完成数据库课程项目。")
|
|
confirm_card(client, session_id, proposal)
|
|
|
|
response = send_message(client, session_id, "我要补充已经填写过的教育经历")
|
|
assert "无需重填" in response["turn"]["content"]
|
|
assert not any(
|
|
block["data"].get("component") == "record_fields"
|
|
for block in response["turn"]["blocks"]
|
|
if block["type"] == "component"
|
|
)
|
|
|
|
|
|
def test_existing_education_supplement_offers_a_choice_for_multiple_entries(client: TestClient) -> None:
|
|
session_id, body = create_builder_session(client)
|
|
start_education(client, session_id, body, school="First University")
|
|
first = finish_education(client, session_id, body, "GPA 3.6/4.0，完成课程项目。")
|
|
after_first = confirm_card(client, session_id, first)
|
|
start_education(client, session_id, after_first, school="Second University")
|
|
second = finish_education(client, session_id, after_first, "GPA 3.8/4.0，获得学业奖学金并完成机器学习课程项目。")
|
|
confirm_card(client, session_id, second)
|
|
|
|
response = send_message(client, session_id, "补充已经填写过的教育经历")
|
|
card = active_component(response)
|
|
assert card["data"]["component"] == "choice_chips"
|
|
assert card["data"]["module"] == "builder_entry_select"
|
|
assert len(card["data"]["options"]) == 2
|
|
|
|
|
|
def test_explicit_new_education_entry_still_shows_an_identity_card(client: TestClient) -> None:
|
|
session_id, body = create_builder_session(client)
|
|
start_education(client, session_id, body)
|
|
proposal = finish_education(client, session_id, body, "GPA 3.7/4.0，完成数据库课程项目。")
|
|
after_confirm = confirm_card(client, session_id, proposal)
|
|
|
|
response = send_message(client, session_id, "新增一段教育经历")
|
|
card = active_component(response)
|
|
assert card["data"]["component"] == "record_fields"
|
|
assert card["data"]["record_type"] == "education"
|
|
|
|
|
|
def test_revision_correction_is_not_treated_as_a_no_information_answer(client: TestClient) -> None:
|
|
session_id, body = create_builder_session(client)
|
|
start_education(client, session_id, body)
|
|
proposal = finish_education(client, session_id, body, "GPA 3.7/4.0，完成数据库课程项目。")
|
|
revision = event(client, session_id, proposal, "edit", {})
|
|
assert revision.status_code == 200, revision.text
|
|
|
|
response = send_message(client, session_id, "我没有说要跳过这个经历")
|
|
assert "已理解你的调整说明" in response["turn"]["content"]
|
|
assert active_component(response)["data"]["component"] == "experience_confirm_card"
|