resume-agent/backend/tests/test_resume_patch_api.py

from __future__ import annotations

from typing import Any

from fastapi.testclient import TestClient

from test_api import event, start_manual_profile

BASE = "/ai-api/resume-agent"
ANCHOR = {
    "school": "示例大学",
    "major": "计算机科学",
    "degree": "本科",
    "start_date": "2021-09",
    "end_date_or_present": "2025-06",
}


def create_resume_with_anchor(client: TestClient) -> tuple[str, dict[str, Any]]:
    session_id, body = start_manual_profile(client, job_type="campus")
    assert body["stage"] == "ANCHOR_COLLECTING"
    response = event(
        client,
        session_id,
        body,
        "submit",
        {**ANCHOR, "description": "做过课程设计"},
    )
    assert response.status_code == 200
    assert response.json()["stage"] == "ANCHOR_CONFIRM"
    response = event(client, session_id, response.json(), "confirm", {"confirmed": True})
    assert response.status_code == 200
    assert response.json()["stage"] == "MINIMUM_READY"
    response = client.post(f"{BASE}/sessions/{session_id}/create", json={})
    assert response.status_code == 200
    body = response.json()
    assert body["created"] is True
    return session_id, body


def first_entry(body: dict[str, Any]) -> dict[str, Any]:
    return body["resume"]["content"]["sections"][0]["items"][0]


def test_response_carries_resume_with_ids(client: TestClient) -> None:
    _, body = create_resume_with_anchor(client)
    content = body["resume"]["content"]
    assert content["schema_version"] == 3
    assert content["skill_groups"] == []
    assert content["sections"][0]["id"].startswith("sec_")
    assert first_entry(body)["id"].startswith("entry_")


def test_patch_update_basics_and_entry(client: TestClient) -> None:
    session_id, body = create_resume_with_anchor(client)
    entry_id = first_entry(body)["id"]
    response = client.patch(
        f"{BASE}/sessions/{session_id}/resume",
        json={
            "expected_revision": body["resume"]["revision"],
            "operation": {"type": "update_basics", "fields": {"name": "李四"}},
        },
    )
    assert response.status_code == 200
    body = response.json()
    assert body["resume"]["content"]["basics"]["name"] == "李四"
    response = client.patch(
        f"{BASE}/sessions/{session_id}/resume",
        json={
            "expected_revision": body["resume"]["revision"],
            "operation": {
                "type": "update_entry",
                "entry_id": entry_id,
                "fields": {"major": "软件工程"},
            },
        },
    )
    assert response.status_code == 200
    entry = first_entry(response.json())
    assert entry["major"] == "软件工程"
    assert entry["provenance"] == "user_edited"
    assert entry["id"] == entry_id


def test_patch_revision_conflict(client: TestClient) -> None:
    session_id, _ = create_resume_with_anchor(client)
    response = client.patch(
        f"{BASE}/sessions/{session_id}/resume",
        json={
            "expected_revision": 999,
            "operation": {"type": "update_basics", "fields": {"name": "x"}},
        },
    )
    assert response.status_code == 409
    assert response.json()["error"]["code"] == "revision_conflict"


def test_patch_delete_entry(client: TestClient) -> None:
    session_id, body = create_resume_with_anchor(client)
    response = client.patch(
        f"{BASE}/sessions/{session_id}/resume",
        json={
            "expected_revision": body["resume"]["revision"],
            "operation": {"type": "delete_entry", "entry_id": first_entry(body)["id"]},
        },
    )
    assert response.status_code == 200
    assert response.json()["resume"]["content"]["sections"] == []


def test_patch_before_create_returns_409(client: TestClient) -> None:
    session_id, _ = start_manual_profile(client, job_type="campus")
    response = client.patch(
        f"{BASE}/sessions/{session_id}/resume",
        json={
            "expected_revision": 1,
            "operation": {"type": "update_basics", "fields": {"name": "x"}},
        },
    )
    assert response.status_code == 409
    assert response.json()["error"]["code"] == "resume_not_created"


def test_patch_skill_groups_and_recommendations_do_not_auto_apply(client: TestClient) -> None:
    session_id, body = create_resume_with_anchor(client)
    revision = body["resume"]["revision"]
    response = client.patch(
        f"{BASE}/sessions/{session_id}/resume",
        json={
            "expected_revision": revision,
            "operation": {
                "type": "update_skill_groups",
                "skills": ["Python", "FastAPI", "Vue", "Docker"],
            },
        },
    )

    assert response.status_code == 200, response.text
    updated = response.json()["resume"]
    groups = updated["content"]["skill_groups"]
    assert {skill for group in groups for skill in group["skills"]} == {
        "Python", "FastAPI", "Vue", "Docker"
    }
    assert all(group["category"] != "其他技能" for group in groups)

    recommendation = client.post(
        f"{BASE}/sessions/{session_id}/resume/skills/recommend",
        json={"question": "还有哪些与后端工程师岗位匹配的技术栈？"},
    )

    assert recommendation.status_code == 200, recommendation.text
    payload = recommendation.json()
    assert payload["candidates"]
    assert all("skill" in item and "category" in item for item in payload["candidates"])
    timeline = client.get(f"{BASE}/sessions/{session_id}/timeline").json()
    assert timeline["resume"]["content"]["skill_groups"] == groups


def test_patch_phone_masks_resume_content_and_updates_session_profile(client: TestClient) -> None:
    session_id, body = create_resume_with_anchor(client)
    response = client.patch(
        f"{BASE}/sessions/{session_id}/resume",
        json={
            "expected_revision": body["resume"]["revision"],
            "operation": {"type": "update_basics", "fields": {"phone": "13900139000"}},
        },
    )

    assert response.status_code == 200, response.text
    basics = response.json()["resume"]["content"]["basics"]
    assert basics["masked_phone"] == "139****9000"
    assert "phone" not in basics

    session = client.app.state.database.get_session(session_id)
    assert session is not None
    assert session["profile"]["phone"] == "13900139000"
    assert session["profile"]["phone_source"] == "resume_edit"