generated from kgod/ai-review-template
145 lines
6.1 KiB
Python
145 lines
6.1 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.profile_summary import ProfileSummaryGenerator
|
|
from app.services import RuleBasedExperienceExtractor, RuleBasedResumeRewriter
|
|
from app.main import create_app
|
|
from app.settings import Settings
|
|
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",
|
|
}
|
|
|
|
|
|
class CountingSummaryGenerator(ProfileSummaryGenerator):
|
|
def __init__(self) -> None:
|
|
self.calls = 0
|
|
|
|
def generate(self, content: dict[str, Any]) -> str:
|
|
self.calls += 1
|
|
return f"? {self.calls} ??????????????????????????????"
|
|
|
|
|
|
def summary_client(tmp_path: Any) -> tuple[TestClient, CountingSummaryGenerator]:
|
|
generator = CountingSummaryGenerator()
|
|
app = create_app(
|
|
database_path=tmp_path / "summary.db",
|
|
extractor=RuleBasedExperienceExtractor(),
|
|
rewriter=RuleBasedResumeRewriter(),
|
|
settings=Settings(llm_provider="rule", offerpai_auth_required=False),
|
|
profile_summary_generator=generator,
|
|
)
|
|
return TestClient(app), generator
|
|
|
|
|
|
def create_resume(client: TestClient) -> tuple[str, dict[str, Any]]:
|
|
session_id, body = start_manual_profile(client, job_type="campus")
|
|
response = event(client, session_id, body, "submit", {**ANCHOR, "description": "?????????????"})
|
|
response = event(client, session_id, response.json(), "confirm", {"confirmed": True})
|
|
response = client.post(f"{BASE}/sessions/{session_id}/create", json={})
|
|
assert response.status_code == 200
|
|
return session_id, response.json()
|
|
|
|
|
|
def finish(client: TestClient, session_id: str, body: dict[str, Any]) -> dict[str, Any]:
|
|
card = next(
|
|
block for block in body["turn"]["blocks"]
|
|
if block["data"].get("component_name") == "ContentReadyCard"
|
|
)
|
|
response = client.post(
|
|
f"{BASE}/sessions/{session_id}/component-events",
|
|
json={"component_id": card["id"], "event": "finish_enrichment", "payload": {}},
|
|
)
|
|
assert response.status_code == 200, response.text
|
|
return response.json()
|
|
|
|
|
|
def test_finish_generates_profile_summary_once_and_marks_it_stale_on_resume_change(tmp_path: Any) -> None:
|
|
with summary_client(tmp_path)[0] as client:
|
|
session_id, body = create_resume(client)
|
|
finished = finish(client, session_id, body)
|
|
summary = finished["resume"]["content"]["profile_summary"]
|
|
assert summary["source"] == "ai_generated"
|
|
assert summary["stale"] is False
|
|
assert summary["content"]
|
|
patched = client.patch(
|
|
f"{BASE}/sessions/{session_id}/resume",
|
|
json={
|
|
"expected_revision": finished["resume"]["revision"],
|
|
"operation": {"type": "update_basics", "fields": {"name": "??"}},
|
|
},
|
|
)
|
|
assert patched.status_code == 200, patched.text
|
|
changed = patched.json()["resume"]["content"]["profile_summary"]
|
|
assert changed["content"] == summary["content"]
|
|
assert changed["stale"] is True
|
|
|
|
|
|
def test_summary_edit_regenerate_confirm_and_reject_keep_user_control(tmp_path: Any) -> None:
|
|
test_client, generator = summary_client(tmp_path)
|
|
with test_client as client:
|
|
session_id, body = create_resume(client)
|
|
finished = finish(client, session_id, body)
|
|
revision = finished["resume"]["revision"]
|
|
edit = client.patch(
|
|
f"{BASE}/sessions/{session_id}/resume",
|
|
json={
|
|
"expected_revision": revision,
|
|
"operation": {
|
|
"type": "update_profile_summary",
|
|
"fields": {"content": "?????????????????????????????"},
|
|
},
|
|
},
|
|
)
|
|
assert edit.status_code == 200, edit.text
|
|
edited = edit.json()["resume"]["content"]["profile_summary"]
|
|
assert edited["source"] == "user_edited"
|
|
assert edited["stale"] is False
|
|
|
|
generated = client.post(f"{BASE}/sessions/{session_id}/resume/profile-summary/generate")
|
|
assert generated.status_code == 200, generated.text
|
|
proposal = generated.json()["resume"]["content"]["profile_summary"]
|
|
assert proposal["content"] == edited["content"]
|
|
assert proposal["pending_proposal"]["content"].startswith("? 2 ?")
|
|
|
|
rejected = client.post(f"{BASE}/sessions/{session_id}/resume/profile-summary/reject")
|
|
assert rejected.status_code == 200, rejected.text
|
|
assert rejected.json()["resume"]["content"]["profile_summary"]["content"] == edited["content"]
|
|
assert "pending_proposal" not in rejected.json()["resume"]["content"]["profile_summary"]
|
|
|
|
generated = client.post(f"{BASE}/sessions/{session_id}/resume/profile-summary/generate")
|
|
assert generated.status_code == 200, generated.text
|
|
confirmed = client.post(f"{BASE}/sessions/{session_id}/resume/profile-summary/confirm")
|
|
assert confirmed.status_code == 200, confirmed.text
|
|
summary = confirmed.json()["resume"]["content"]["profile_summary"]
|
|
assert summary["content"].startswith("? 3 ?")
|
|
assert summary["source"] == "ai_generated"
|
|
assert summary["stale"] is False
|
|
assert "pending_proposal" not in summary
|
|
assert generator.calls == 3
|
|
|
|
|
|
def test_repeat_generation_keeps_current_summary_until_user_confirms(tmp_path: Any) -> None:
|
|
test_client, generator = summary_client(tmp_path)
|
|
with test_client as client:
|
|
session_id, body = create_resume(client)
|
|
finished = finish(client, session_id, body)
|
|
current = finished["resume"]["content"]["profile_summary"]["content"]
|
|
first = client.post(f"{BASE}/sessions/{session_id}/resume/profile-summary/generate")
|
|
assert first.status_code == 200, first.text
|
|
second = client.post(f"{BASE}/sessions/{session_id}/resume/profile-summary/generate")
|
|
assert second.status_code == 200, second.text
|
|
summary = second.json()["resume"]["content"]["profile_summary"]
|
|
assert summary["content"] == current
|
|
assert summary["pending_proposal"]["content"].startswith("? 3 ?")
|
|
assert generator.calls == 3
|