generated from kgod/ai-review-template
241 lines
9.3 KiB
Python
241 lines
9.3 KiB
Python
from __future__ import annotations
|
|
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
import json
|
|
from typing import Any
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
|
|
BASE = "/ai-api/resume-agent"
|
|
|
|
|
|
def active_component(body: dict[str, Any]) -> dict[str, Any]:
|
|
turns = body.get("turns") or ([body["turn"]] if body.get("turn") else [])
|
|
for turn in reversed(turns):
|
|
for block in reversed(turn["blocks"]):
|
|
if block["type"] == "component" and block["lifecycle"] == "active":
|
|
return block
|
|
raise AssertionError("response has no active component")
|
|
|
|
|
|
def event(
|
|
client: TestClient,
|
|
session_id: str,
|
|
body: dict[str, Any],
|
|
event_name: str,
|
|
payload: dict[str, Any] | None = None,
|
|
headers: dict[str, str] | None = None,
|
|
):
|
|
block = active_component(body)
|
|
return client.post(
|
|
f"{BASE}/sessions/{session_id}/component-events",
|
|
json={"component_id": block["id"], "event": event_name, "payload": payload or {}},
|
|
headers=headers,
|
|
)
|
|
|
|
|
|
def start_manual_profile(
|
|
client: TestClient,
|
|
*,
|
|
job_type: str = "campus",
|
|
target_position: str | None = "Backend Engineer",
|
|
) -> tuple[str, dict[str, Any]]:
|
|
created = client.post(f"{BASE}/sessions", json={})
|
|
assert created.status_code == 201
|
|
body = created.json()
|
|
session_id = body["session_id"]
|
|
assert body["stage"] == "PRIVACY_CONSENT"
|
|
|
|
phone_selector = event(client, session_id, body, "accept", {"accepted": True})
|
|
assert phone_selector.status_code == 200
|
|
assert phone_selector.json()["stage"] == "PHONE_SELECTION"
|
|
|
|
phone_input = event(client, session_id, phone_selector.json(), "select", {"source": "other"})
|
|
assert phone_input.status_code == 200
|
|
assert phone_input.json()["stage"] == "MANUAL_PHONE_INPUT"
|
|
|
|
personal = event(client, session_id, phone_input.json(), "submit", {"phone": "13800138000"})
|
|
assert personal.status_code == 200
|
|
assert personal.json()["stage"] == "PERSONAL_INFO"
|
|
|
|
job_selector = event(
|
|
client,
|
|
session_id,
|
|
personal.json(),
|
|
"submit",
|
|
{"name": "Zhang San", "email": "zhangsan@example.com", "city": "Shanghai"},
|
|
)
|
|
assert job_selector.status_code == 200
|
|
assert job_selector.json()["stage"] == "JOB_TYPE_SELECT"
|
|
|
|
target = event(client, session_id, job_selector.json(), "select", {"job_type": job_type})
|
|
assert target.status_code == 200
|
|
assert target.json()["stage"] == "TARGET_POSITION"
|
|
|
|
if target_position is None:
|
|
ready = event(client, session_id, target.json(), "skip")
|
|
else:
|
|
ready = event(client, session_id, target.json(), "submit", {"target_position": target_position})
|
|
assert ready.status_code == 200, ready.text
|
|
assert ready.json()["stage"] == "MINIMUM_READY"
|
|
return session_id, ready.json()
|
|
|
|
|
|
ANCHOR_CARD_VALUES = {
|
|
"school": "Example University",
|
|
"major": "Computer Science",
|
|
"degree": "Bachelor",
|
|
"start_date": "2021-09",
|
|
"end_date_or_present": "2025-06",
|
|
}
|
|
|
|
|
|
def fill_anchor(
|
|
client: TestClient,
|
|
session_id: str,
|
|
body: dict[str, Any],
|
|
values: dict[str, str],
|
|
) -> dict[str, Any]:
|
|
response = event(client, session_id, body, "submit", values)
|
|
assert response.status_code == 200, response.text
|
|
return response.json()
|
|
|
|
|
|
def campus_ready(client: TestClient) -> tuple[str, dict[str, Any]]:
|
|
return start_manual_profile(client, job_type="campus")
|
|
|
|
|
|
def test_privacy_continues_directly_into_new_resume_flow(client: TestClient) -> None:
|
|
created = client.post(f"{BASE}/sessions", json={})
|
|
session_id = created.json()["session_id"]
|
|
phone_selector = event(client, session_id, created.json(), "accept", {"accepted": True})
|
|
assert phone_selector.status_code == 200
|
|
body = phone_selector.json()
|
|
assert body["stage"] == "PHONE_SELECTION"
|
|
assert active_component(body)["data"]["component"] == "resume_phone_selector"
|
|
assert "导入已有简历" not in json.dumps(body, ensure_ascii=False)
|
|
|
|
with client.app.state.database.transaction() as connection:
|
|
session = client.app.state.database.fetch_session(connection, session_id)
|
|
assert session is not None
|
|
assert session["profile"]["resume_source"] == "manual"
|
|
|
|
|
|
@pytest.mark.parametrize("legacy_stage", ["RESUME_SOURCE_SELECT", "RESUME_IMPORT_UPLOAD"])
|
|
def test_legacy_source_stages_advance_to_new_resume_flow(
|
|
client: TestClient, legacy_stage: str
|
|
) -> None:
|
|
created = client.post(f"{BASE}/sessions", json={}).json()
|
|
session_id = created["session_id"]
|
|
accepted = event(client, session_id, created, "accept", {"accepted": True})
|
|
assert accepted.status_code == 200
|
|
|
|
with client.app.state.database.transaction(immediate=True) as connection:
|
|
session = client.app.state.database.fetch_session(connection, session_id)
|
|
assert session is not None
|
|
profile = dict(session["profile"])
|
|
profile["resume_source"] = "import"
|
|
client.app.state.database.update_session(
|
|
connection,
|
|
session_id,
|
|
stage=legacy_stage,
|
|
profile=profile,
|
|
)
|
|
|
|
migrated = client.get(f"{BASE}/sessions/{session_id}/timeline")
|
|
assert migrated.status_code == 200
|
|
body = migrated.json()
|
|
assert body["stage"] == "PHONE_SELECTION"
|
|
assert active_component(body)["data"]["component"] == "resume_phone_selector"
|
|
turn_count = len(body["turns"])
|
|
|
|
repeated = client.get(f"{BASE}/sessions/{session_id}/timeline").json()
|
|
assert len(repeated["turns"]) == turn_count
|
|
with client.app.state.database.transaction() as connection:
|
|
session = client.app.state.database.fetch_session(connection, session_id)
|
|
assert session is not None
|
|
assert session["profile"]["resume_source"] == "manual"
|
|
|
|
|
|
def test_concurrent_legacy_stage_refresh_advances_only_once(client: TestClient) -> None:
|
|
created = client.post(f"{BASE}/sessions", json={}).json()
|
|
session_id = created["session_id"]
|
|
accepted = event(client, session_id, created, "accept", {"accepted": True})
|
|
assert accepted.status_code == 200
|
|
baseline_turn_count = len(
|
|
client.get(f"{BASE}/sessions/{session_id}/timeline").json()["turns"]
|
|
)
|
|
|
|
with client.app.state.database.transaction(immediate=True) as connection:
|
|
session = client.app.state.database.fetch_session(connection, session_id)
|
|
assert session is not None
|
|
profile = dict(session["profile"])
|
|
profile["resume_source"] = "import"
|
|
client.app.state.database.update_session(
|
|
connection,
|
|
session_id,
|
|
stage="RESUME_SOURCE_SELECT",
|
|
profile=profile,
|
|
)
|
|
|
|
agent = client.app.state.resume_agent
|
|
with ThreadPoolExecutor(max_workers=4) as executor:
|
|
responses = list(executor.map(lambda _: agent.timeline(session_id), range(4)))
|
|
|
|
assert all(response.stage == "PHONE_SELECTION" for response in responses)
|
|
timeline = client.get(f"{BASE}/sessions/{session_id}/timeline").json()
|
|
assert len(timeline["turns"]) == baseline_turn_count + 1
|
|
|
|
|
|
def test_manual_phone_is_strict_and_retryable(client: TestClient) -> None:
|
|
created = client.post(f"{BASE}/sessions", json={}).json()
|
|
session_id = created["session_id"]
|
|
phone_selector = event(client, session_id, created, "accept", {"accepted": True}).json()
|
|
phone_input = event(client, session_id, phone_selector, "select", {"source": "other"}).json()
|
|
|
|
invalid = event(client, session_id, phone_input, "submit", {"phone": "+8613800138000"})
|
|
assert invalid.status_code == 422
|
|
assert invalid.json()["error"]["code"] == "invalid_phone"
|
|
|
|
valid = event(client, session_id, phone_input, "submit", {"phone": "13900139000"})
|
|
assert valid.status_code == 200
|
|
assert valid.json()["stage"] == "PERSONAL_INFO"
|
|
|
|
|
|
def test_target_position_creates_a_basic_resume_without_core_experience(client: TestClient) -> None:
|
|
session_id, ready = start_manual_profile(client, job_type="social")
|
|
assert ready["gate"]["allowed"] is True
|
|
assert ready["missing_fields"] == []
|
|
|
|
created = client.post(f"{BASE}/sessions/{session_id}/create", json={"idempotency_key": "basic-resume"})
|
|
assert created.status_code == 200, created.text
|
|
result = created.json()
|
|
assert result["created"] is True
|
|
assert result["stage"] == "BUILDER_CONVERSATION"
|
|
assert result["resume"]["content"]["sections"] == []
|
|
|
|
|
|
def test_full_campus_creation_is_idempotent_and_masks_phone(client: TestClient) -> None:
|
|
session_id, ready = campus_ready(client)
|
|
first = client.post(f"{BASE}/sessions/{session_id}/create", json={"idempotency_key": "create-once"})
|
|
assert first.status_code == 200, first.text
|
|
result = first.json()
|
|
assert result["created"] is True
|
|
assert result["stage"] == "BUILDER_CONVERSATION"
|
|
assert result["resume"]["content"]["basics"]["masked_phone"] == "138****8000"
|
|
assert "13800138000" not in json.dumps(result, ensure_ascii=False)
|
|
|
|
second = client.post(f"{BASE}/sessions/{session_id}/create", json={"idempotency_key": "another-key"})
|
|
assert second.status_code == 200
|
|
assert second.json()["created"] is False
|
|
assert second.json()["resume_id"] == result["resume_id"]
|
|
|
|
|
|
def test_target_position_exploration_can_create_without_core_experience(client: TestClient) -> None:
|
|
session_id, ready = start_manual_profile(client, job_type="campus", target_position=None)
|
|
assert ready["stage"] == "MINIMUM_READY"
|
|
assert ready["gate"]["allowed"] is True
|
|
assert session_id
|