generated from kgod/ai-review-template
172 lines
7.4 KiB
Python
172 lines
7.4 KiB
Python
from __future__ import annotations
|
|
|
|
from io import BytesIO
|
|
|
|
from docx import Document
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.main import create_app
|
|
from app.resume_import_models import ParsedResumeDraft
|
|
from app.resume_import_service import ResumeImportService
|
|
from app.services import RuleBasedEntryExpander, RuleBasedExperienceExtractor, RuleBasedResumeRewriter
|
|
from app.settings import Settings
|
|
|
|
|
|
BASE = "/ai-api/resume-agent"
|
|
|
|
|
|
class FakeResumeImportParser:
|
|
def parse(self, *, text: str, source_name: str) -> ParsedResumeDraft:
|
|
return ParsedResumeDraft(
|
|
document={
|
|
"schema_version": 3,
|
|
"basics": {"name": "Imported Name", "phone": "13800138000", "email": "import@example.com"},
|
|
"target": {"job_type": "campus", "position": "Backend Engineer"},
|
|
"sections": [{"kind": "education", "heading": "Education", "items": [{"school": "Example University", "major": "Computer Science"}]}],
|
|
"skill_groups": [{"category": "Programming Languages", "skills": ["Python"]}],
|
|
},
|
|
field_reviews=[],
|
|
)
|
|
|
|
|
|
def docx_bytes(text: str) -> bytes:
|
|
document = Document()
|
|
document.add_paragraph(text)
|
|
buffer = BytesIO()
|
|
document.save(buffer)
|
|
return buffer.getvalue()
|
|
|
|
|
|
def client_for_import(tmp_path) -> TestClient:
|
|
application = create_app(
|
|
database_path=tmp_path / "test.db",
|
|
cors_origins=["http://localhost:5173"],
|
|
extractor=RuleBasedExperienceExtractor(),
|
|
rewriter=RuleBasedResumeRewriter(),
|
|
expander=RuleBasedEntryExpander(),
|
|
settings=Settings(llm_provider="rule", offerpai_auth_required=False),
|
|
resume_import_service=ResumeImportService(storage_root=tmp_path / "imports", parser=FakeResumeImportParser()),
|
|
)
|
|
return TestClient(application)
|
|
|
|
|
|
def _active_component(body: dict) -> dict:
|
|
turns = body.get("turns") or [body["turn"]]
|
|
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, name: str, payload: dict | None = None):
|
|
return client.post(
|
|
f"{BASE}/sessions/{session_id}/component-events",
|
|
json={"component_id": _active_component(body)["id"], "event": name, "payload": payload or {}},
|
|
)
|
|
|
|
|
|
def import_session(client: TestClient) -> str:
|
|
created = client.post(f"{BASE}/sessions", json={})
|
|
session_id = created.json()["session_id"]
|
|
source = _event(client, session_id, created.json(), "accept", {"accepted": True})
|
|
selected = _event(client, session_id, source.json(), "select", {"value": "import"})
|
|
assert selected.status_code == 200
|
|
assert selected.json()["stage"] == "RESUME_IMPORT_UPLOAD"
|
|
return session_id
|
|
|
|
|
|
def upload(client: TestClient, session_id: str, name: str = "resume.docx"):
|
|
return client.post(
|
|
f"{BASE}/sessions/{session_id}/resume-imports",
|
|
files={"file": (name, docx_bytes("Imported Name\nExample University"), "application/vnd.openxmlformats-officedocument.wordprocessingml.document")},
|
|
)
|
|
|
|
|
|
def test_import_requires_privacy_consent_and_import_selection(tmp_path) -> None:
|
|
with client_for_import(tmp_path) as client:
|
|
session_id = client.post(f"{BASE}/sessions", json={}).json()["session_id"]
|
|
before_consent = upload(client, session_id)
|
|
assert before_consent.status_code == 409
|
|
assert before_consent.json()["error"]["code"] == "privacy_consent_required"
|
|
|
|
timeline = client.get(f"{BASE}/sessions/{session_id}/timeline").json()
|
|
source = _event(client, session_id, timeline, "accept", {"accepted": True})
|
|
without_choice = upload(client, session_id)
|
|
assert without_choice.status_code == 409
|
|
assert without_choice.json()["error"]["code"] == "resume_import_not_selected"
|
|
|
|
manual = _event(client, session_id, source.json(), "select", {"value": "manual"})
|
|
assert manual.status_code == 200
|
|
after_manual_choice = upload(client, session_id)
|
|
assert after_manual_choice.status_code == 409
|
|
assert after_manual_choice.json()["error"]["code"] == "resume_import_not_selected"
|
|
|
|
|
|
def test_docx_import_is_reviewable_and_apply_updates_live_resume(tmp_path) -> None:
|
|
with client_for_import(tmp_path) as client:
|
|
session_id = import_session(client)
|
|
imported = upload(client, session_id)
|
|
assert imported.status_code == 201, imported.text
|
|
view = imported.json()
|
|
assert view["status"] == "awaiting_review"
|
|
|
|
applied = client.post(
|
|
f"{BASE}/sessions/{session_id}/resume-imports/{view['id']}/apply",
|
|
json={"expected_revision": 0},
|
|
)
|
|
assert applied.status_code == 200, applied.text
|
|
body = applied.json()
|
|
assert body["stage"] == "RESUME_ENRICHING"
|
|
content = body["resume"]["content"]
|
|
assert content["basics"]["name"] == "Imported Name"
|
|
assert content["basics"]["masked_phone"] == "138****8000"
|
|
assert "phone" not in content["basics"]
|
|
assert content["basics"]["email"] == "import@example.com"
|
|
assert content["sections"][0]["items"][0]["school"] == "Example University"
|
|
|
|
|
|
def test_import_is_blocked_after_an_imported_resume_is_applied(tmp_path) -> None:
|
|
with client_for_import(tmp_path) as client:
|
|
session_id = import_session(client)
|
|
first_upload = upload(client, session_id)
|
|
imported = first_upload.json()
|
|
applied = client.post(
|
|
f"{BASE}/sessions/{session_id}/resume-imports/{imported['id']}/apply",
|
|
json={"expected_revision": 0},
|
|
)
|
|
assert applied.status_code == 200
|
|
blocked = upload(client, session_id, "second.docx")
|
|
assert blocked.status_code == 409
|
|
assert blocked.json()["error"]["code"] == "resume_import_not_allowed"
|
|
|
|
|
|
def test_legacy_doc_and_scanned_pdf_return_stable_errors(tmp_path) -> None:
|
|
with client_for_import(tmp_path) as client:
|
|
session_id = import_session(client)
|
|
legacy = client.post(f"{BASE}/sessions/{session_id}/resume-imports", files={"file": ("resume.doc", b"not-a-docx", "application/msword")})
|
|
assert legacy.status_code == 422
|
|
assert legacy.json()["error"]["code"] == "legacy_doc_unsupported"
|
|
scanned = client.post(f"{BASE}/sessions/{session_id}/resume-imports", files={"file": ("scan.pdf", b"%PDF-1.7\n", "application/pdf")})
|
|
assert scanned.status_code == 422
|
|
assert scanned.json()["error"]["code"] == "ocr_required"
|
|
|
|
|
|
def test_imported_resume_continue_enriching_keeps_imported_content(tmp_path) -> None:
|
|
with client_for_import(tmp_path) as client:
|
|
session_id = import_session(client)
|
|
imported = upload(client, session_id)
|
|
applied = client.post(
|
|
f"{BASE}/sessions/{session_id}/resume-imports/{imported.json()['id']}/apply",
|
|
json={"expected_revision": 0},
|
|
)
|
|
assert applied.status_code == 200, applied.text
|
|
before = applied.json()["resume"]["content"]
|
|
|
|
continued = _event(client, session_id, applied.json(), "continue_enriching")
|
|
assert continued.status_code == 200, continued.text
|
|
body = continued.json()
|
|
assert body["stage"] == "RESUME_ENRICHING"
|
|
assert _active_component(body)["data"]["component"] == "custom_card_picker"
|
|
assert body["resume"]["content"] == before
|