Files

873 lines
29 KiB
Python

from __future__ import annotations
import json
from collections.abc import Mapping
from datetime import date, datetime, timezone
from typing import Any
import pytest
from pydantic import BaseModel
from resume_harness.backend import LLMBackend
from resume_harness.models import (
CandidateProfile,
ContactInfo,
ContentPlan,
ConstraintKind,
DraftClaim,
DraftSection,
EvidenceCategory,
EvidenceItem,
EvidenceMap,
EvidenceMatch,
EvidenceMatchType,
EvidenceSource,
GenerationConfig,
JobAnalysis,
JobPosting,
JobRequirement,
PlannedSection,
PostingConstraint,
QualityCategory,
QualityFinding,
QualityReport,
QualitySeverity,
RequirementCategory,
RequirementKind,
ResumeDraft,
ResumeMode,
SectionType,
SensitiveDataCategory,
SensitiveDataConsent,
)
from resume_harness.pipeline import PipelineError, PipelineStatus, ResumePipeline
from resume_harness.records import (
CareerRecord,
EmploymentType,
RecordDate,
RecordPeriod,
ResumeRecords,
)
UTC = timezone.utc
NOW = datetime(2026, 7, 1, 12, tzinfo=UTC)
class FakeBackend:
"""Strict FIFO fake: an unexpected stage fails the integration test."""
def __init__(self, responses: list[tuple[str, BaseModel | Mapping[str, Any]]]):
self.responses = list(responses)
self.calls: list[dict[str, Any]] = []
def complete_json(
self,
*,
stage: str,
system_prompt: str,
task_prompt: str,
user_payload: Mapping[str, Any],
output_model: type[BaseModel],
) -> BaseModel | Mapping[str, Any]:
assert system_prompt
assert task_prompt
assert self.responses, f"unexpected backend call for {stage}"
expected_stage, response = self.responses.pop(0)
assert stage == expected_stage
assert isinstance(response, output_model) or isinstance(response, Mapping)
self.calls.append(
{
"stage": stage,
"payload": user_payload,
"output_model": output_model,
}
)
return response
def _inputs() -> tuple[CandidateProfile, JobPosting, GenerationConfig]:
consent = SensitiveDataConsent(
consent_id="consent-photo",
category=SensitiveDataCategory.PHOTO,
purpose="지정 양식 사진",
granted_at=datetime(2026, 1, 1, tzinfo=UTC),
expires_at=datetime(2027, 1, 1, tzinfo=UTC),
)
facts = [
EvidenceItem(
evidence_id="ev-api",
category=EvidenceCategory.PROJECT,
content="Python 결제 API 응답 시간을 40% 단축해 35ms로 개선했다.",
source=EvidenceSource.PORTFOLIO,
source_reference="https://private.example/internal/source",
metrics={"latency_reduction": "40%", "latency": "35ms"},
keywords=["Python", "API"],
),
EvidenceItem(
evidence_id="ev-photo",
category=EvidenceCategory.OTHER,
content="지원자 증명사진 secret-photo-token",
source=EvidenceSource.DOCUMENT,
sensitive_category=SensitiveDataCategory.PHOTO,
consent_id="consent-photo",
),
EvidenceItem(
evidence_id="ev-confidential",
category=EvidenceCategory.PROJECT,
content="secret-company-project 내부 수치",
source=EvidenceSource.USER_STATEMENT,
metrics={"latency": "35ms"},
keywords=["Python"],
confidential=True,
),
]
profile = CandidateProfile(
candidate_id="candidate-1",
name="김하네스",
name_en="Harness Kim",
contact=ContactInfo(
email="harness@example.com",
phone="010-1234-5678",
city="서울",
links=["https://portfolio.example/harness"],
),
facts=facts,
consents=[consent],
updated_at=NOW,
)
posting = JobPosting(
posting_id="posting-1",
company_name="합성테크",
title="백엔드 엔지니어",
raw_text=(
"필수 요건\nPython 기반 API 개발 경험을 갖춘 "
"백엔드 엔지니어를 채용합니다."
),
collected_at=NOW,
)
config = GenerationConfig(as_of_date=date(2026, 7, 1))
return profile, posting, config
def _analysis() -> JobAnalysis:
return JobAnalysis(
analysis_id="analysis-1",
posting_id="posting-1",
target_role="백엔드 엔지니어",
summary="Python API 개발 경험을 중시한다.",
requirements=[
JobRequirement(
requirement_id="req-python",
text="Python 기반 API 개발 경험",
kind=RequirementKind.REQUIRED,
category=RequirementCategory.SKILL,
priority=5,
source_quote="Python 기반 API 개발 경험",
classification_quote="필수 요건\nPython 기반 API 개발 경험",
keywords=["Python", "API"],
)
],
keywords=["Python", "API"],
analysed_at=NOW,
)
def _evidence_map() -> EvidenceMap:
return EvidenceMap(
map_id="map-1",
posting_id="posting-1",
analysis_id="analysis-1",
matches=[
EvidenceMatch(
requirement_id="req-python",
evidence_ids=["ev-api"],
match_type=EvidenceMatchType.DIRECT,
relevance_score=0.95,
rationale="Python API 개선 경험이 직접 연결된다.",
)
],
generated_at=NOW,
)
def _plan(
mode: ResumeMode = ResumeMode.PRIVATE_MODERN,
) -> ContentPlan:
return ContentPlan(
plan_id="plan-1",
candidate_id="candidate-1",
posting_id="posting-1",
mode=mode,
sections=[
PlannedSection(
section_id="section-projects",
section_type=SectionType.PROJECTS,
heading="주요 프로젝트",
evidence_ids=["ev-api"],
requirement_ids=["req-python"],
bullet_budget=2,
order=0,
)
],
created_at=NOW,
)
def _draft(
text: str = "Python 결제 API 응답 시간을 40% 단축",
*,
mode: ResumeMode = ResumeMode.PRIVATE_MODERN,
) -> ResumeDraft:
return ResumeDraft(
draft_id="draft-1",
candidate_id="candidate-1",
posting_id="posting-1",
title="백엔드 엔지니어 이력서",
mode=mode,
sections=[
DraftSection(
section_id="section-projects",
section_type=SectionType.PROJECTS,
heading="주요 프로젝트",
claims=[
DraftClaim(
claim_id="claim-api",
text=text,
evidence_ids=["ev-api"],
requirement_ids=["req-python"],
order=0,
)
],
order=0,
)
],
generated_at=NOW,
)
def _good_report(draft: ResumeDraft | None = None) -> QualityReport:
bound_draft = draft or _draft()
return QualityReport(
report_id="report-good",
draft_id="draft-1",
draft_fingerprint=bound_draft.fingerprint(),
overall_score=94,
evidence_coverage=1.0,
requirement_coverage=1.0,
category_scores={
QualityCategory.EVIDENCE: 100,
QualityCategory.JOB_ALIGNMENT: 92,
QualityCategory.COMPLETENESS: 92,
QualityCategory.KOREAN_LANGUAGE: 91,
QualityCategory.READABILITY: 92,
QualityCategory.FORMATTING: 95,
QualityCategory.CONSISTENCY: 95,
QualityCategory.PRIVACY: 100,
},
findings=[],
evaluated_at=NOW,
)
def _bad_report(
report_id: str, draft: ResumeDraft | None = None
) -> QualityReport:
bound_draft = draft or _draft()
return QualityReport(
report_id=report_id,
draft_id="draft-1",
draft_fingerprint=bound_draft.fingerprint(),
overall_score=84,
evidence_coverage=1.0,
requirement_coverage=1.0,
category_scores={
QualityCategory.EVIDENCE: 100,
QualityCategory.JOB_ALIGNMENT: 76,
QualityCategory.COMPLETENESS: 82,
QualityCategory.KOREAN_LANGUAGE: 78,
QualityCategory.READABILITY: 80,
QualityCategory.FORMATTING: 90,
QualityCategory.CONSISTENCY: 88,
QualityCategory.PRIVACY: 100,
},
findings=[
QualityFinding(
finding_id=f"finding-{report_id}",
code="STYLE.ABSTRACT_ACTION",
severity=QualitySeverity.WARNING,
category=QualityCategory.KOREAN_LANGUAGE,
message="행동과 결과의 연결을 더 분명히 해야 한다.",
claim_id="claim-api",
evidence_ids=["ev-api"],
suggestion="근거 범위에서 행동을 명확히 한다.",
)
],
evaluated_at=NOW,
)
def _base_responses(
final_report: QualityReport | Mapping[str, Any],
) -> list[tuple[str, BaseModel | Mapping[str, Any]]]:
return [
("analyze-job", _analysis()),
("map-evidence", _evidence_map()),
("plan-content", _plan()),
("draft-resume", _draft()),
("evaluate-resume", final_report),
]
def test_success_orders_stages_and_never_sends_candidate_pii() -> None:
profile, posting, config = _inputs()
# A provider adapter may return a mapping produced from a model. Computed
# read-only fields in that mapping are tolerated and the writable fields are
# still validated strictly.
backend = FakeBackend(_base_responses(_good_report().model_dump(mode="json")))
assert isinstance(backend, LLMBackend)
result = ResumePipeline(backend).run(profile, posting, config)
assert result.status is PipelineStatus.PASSED
assert result.repair_attempts == 0
assert [call["stage"] for call in backend.calls] == [
"analyze-job",
"map-evidence",
"plan-content",
"draft-resume",
"evaluate-resume",
]
assert backend.responses == []
transmitted = json.dumps(
[call["payload"] for call in backend.calls], ensure_ascii=False
)
for forbidden in (
"김하네스",
"Harness Kim",
"harness@example.com",
"010-1234-5678",
"01012345678",
"https://portfolio.example/harness",
"서울",
"secret-photo-token",
"secret-company-project",
):
assert forbidden not in transmitted
# Shared technical and numeric values from an excluded fact are not privacy
# tokens and must remain usable in an allowed fact.
map_payload = backend.calls[1]["payload"]
assert map_payload["candidate_facts"][0]["keywords"] == ["Python", "API"]
assert map_payload["candidate_facts"][0]["metrics"]["latency"] == "35ms"
assert "source_reference" not in map_payload["candidate_facts"][0]
assert "consent_id" not in map_payload["candidate_facts"][0]
assert "confidential" not in map_payload["candidate_facts"][0]
def test_public_blind_school_fact_is_removed_before_llm_boundary() -> None:
profile, posting, _ = _inputs()
profile = profile.model_copy(
update={
"facts": [
*profile.facts,
EvidenceItem(
evidence_id="ev-school",
category=EvidenceCategory.EDUCATION,
content="서울대에서 컴퓨터공학을 전공했다.",
source=EvidenceSource.DOCUMENT,
keywords=["서울대", "컴퓨터공학"],
),
EvidenceItem(
evidence_id="ev-school-snu",
category=EvidenceCategory.EDUCATION,
content="SNU 컴퓨터공학 과정을 졸업했다.",
source=EvidenceSource.DOCUMENT,
keywords=["SNU", "컴퓨터공학"],
),
]
}
)
config = GenerationConfig(
resume_mode=ResumeMode.PUBLIC_BLIND,
as_of_date=date(2026, 7, 1),
)
blind_draft = _draft(mode=ResumeMode.PUBLIC_BLIND)
backend = FakeBackend(
[
("analyze-job", _analysis()),
("map-evidence", _evidence_map()),
("plan-content", _plan(ResumeMode.PUBLIC_BLIND)),
("draft-resume", blind_draft),
("evaluate-resume", _good_report(blind_draft)),
]
)
result = ResumePipeline(backend).run(profile, posting, config)
assert result.status is PipelineStatus.PASSED
transmitted = json.dumps(
[call["payload"] for call in backend.calls], ensure_ascii=False
)
assert "ev-school" not in transmitted
assert "서울대" not in transmitted
assert "ev-school-snu" not in transmitted
assert "SNU" not in transmitted
def test_public_blind_origin_fact_is_removed_before_llm_boundary() -> None:
profile, posting, _ = _inputs()
profile = profile.model_copy(
update={
"facts": [
*profile.facts,
EvidenceItem(
evidence_id="ev-origin",
category=EvidenceCategory.PROJECT,
content="고향은 대전이며 Python API를 개발했다.",
source=EvidenceSource.USER_STATEMENT,
keywords=["Python", "API"],
),
EvidenceItem(
evidence_id="ev-hometown-compact",
category=EvidenceCategory.PROJECT,
content="고향 대전, Python API를 개발했다.",
source=EvidenceSource.USER_STATEMENT,
keywords=["Python", "API"],
),
EvidenceItem(
evidence_id="ev-grown",
category=EvidenceCategory.PROJECT,
content="대전에서 자랐고 Python API를 개발했다.",
source=EvidenceSource.USER_STATEMENT,
keywords=["Python", "API"],
),
EvidenceItem(
evidence_id="ev-region-work",
category=EvidenceCategory.PROJECT,
content="대전 지역 고객을 위한 Python API를 개발했다.",
source=EvidenceSource.PORTFOLIO,
keywords=["Python", "API", "대전 지역"],
),
]
}
)
config = GenerationConfig(
resume_mode=ResumeMode.PUBLIC_BLIND,
as_of_date=date(2026, 7, 1),
)
blind_draft = _draft(mode=ResumeMode.PUBLIC_BLIND)
backend = FakeBackend(
[
("analyze-job", _analysis()),
("map-evidence", _evidence_map()),
("plan-content", _plan(ResumeMode.PUBLIC_BLIND)),
("draft-resume", blind_draft),
("evaluate-resume", _good_report(blind_draft)),
]
)
result = ResumePipeline(backend).run(profile, posting, config)
assert result.status is PipelineStatus.PASSED
transmitted = json.dumps(
[call["payload"] for call in backend.calls], ensure_ascii=False
)
assert "ev-origin" not in transmitted
assert "고향은 대전" not in transmitted
assert "ev-hometown-compact" not in transmitted
assert "고향 대전" not in transmitted
assert "ev-grown" not in transmitted
assert "대전에서 자랐고" not in transmitted
assert "ev-region-work" in transmitted
assert "대전 지역 고객" in transmitted
def test_korean_name_with_postposition_is_redacted_before_backend() -> None:
profile, posting, config = _inputs()
posting = posting.model_copy(
update={
"raw_text": (
"김하네스는 외부 입력입니다.\n필수 요건\n"
"Python 기반 API 개발 경험을 "
"갖춘 지원자를 찾습니다."
)
}
)
backend = FakeBackend(_base_responses(_good_report()))
ResumePipeline(backend).run(profile, posting, config)
analyze_payload = json.dumps(backend.calls[0]["payload"], ensure_ascii=False)
assert "김하네스" not in analyze_payload
assert "[REDACTED]는" in analyze_payload
def test_short_korean_name_does_not_redact_an_ordinary_verb() -> None:
profile, posting, config = _inputs()
profile = profile.model_copy(update={"name": "이수"})
posting = posting.model_copy(
update={
"raw_text": (
"교육 과정을 이수했다면 우대합니다.\n"
"필수 요건\n"
"Python 기반 API 개발 경험을 확인합니다."
)
}
)
backend = FakeBackend(_base_responses(_good_report()))
ResumePipeline(backend).run(profile, posting, config)
analyze_payload = json.dumps(backend.calls[0]["payload"], ensure_ascii=False)
assert "이수했다" in analyze_payload
assert "[REDACTED]했다" not in analyze_payload
def test_no_generation_safe_evidence_returns_early_needs_user_input() -> None:
profile, posting, config = _inputs()
profile = profile.model_copy(
update={
"facts": [
fact.model_copy(update={"confidential": True})
for fact in profile.facts
]
}
)
backend = FakeBackend([("analyze-job", _analysis())])
result = ResumePipeline(backend).run(profile, posting, config)
assert result.status is PipelineStatus.NEEDS_USER_INPUT
assert result.gate_failures == ["no_generation_safe_evidence"]
assert result.evidence_map is None
assert result.draft is None
assert [call["stage"] for call in backend.calls] == ["analyze-job"]
def test_all_gap_evidence_map_returns_before_impossible_draft() -> None:
profile, posting, config = _inputs()
gap_map = EvidenceMap(
map_id="map-gap",
posting_id="posting-1",
analysis_id="analysis-1",
matches=[
EvidenceMatch(
requirement_id="req-python",
match_type=EvidenceMatchType.GAP,
relevance_score=0,
gap_reason="직접 근거가 없다.",
)
],
generated_at=NOW,
)
backend = FakeBackend(
[("analyze-job", _analysis()), ("map-evidence", gap_map)]
)
result = ResumePipeline(backend).run(profile, posting, config)
assert result.status is PipelineStatus.NEEDS_USER_INPUT
assert result.gate_failures == ["all_requirements_gap"]
assert result.evidence_map == gap_map
assert result.content_plan is None
assert result.questions
assert [call["stage"] for call in backend.calls] == [
"analyze-job",
"map-evidence",
]
def test_posting_specific_company_name_fact_is_withheld_before_mapping() -> None:
profile, posting, config = _inputs()
profile = profile.model_copy(
update={
"facts": [
fact.model_copy(
update={
"content": "가상페이에서 Python 기반 API 개발 경험을 쌓았다."
}
)
if fact.evidence_id == "ev-api"
else fact
for fact in profile.facts
]
}
)
posting = posting.model_copy(
update={"raw_text": posting.raw_text + "\n회사명 기재 금지"}
)
analysis = _analysis().model_copy(
update={
"constraints": [
PostingConstraint(
constraint_id="blind-company",
kind=ConstraintKind.BLIND_FIELD,
description="회사명을 본문에 기재하지 않는다.",
source_quote="회사명 기재 금지",
fields=["회사명"],
)
]
}
)
backend = FakeBackend([("analyze-job", analysis)])
result = ResumePipeline(backend).run(profile, posting, config)
assert result.status is PipelineStatus.NEEDS_USER_INPUT
assert result.gate_failures == ["no_generation_safe_evidence"]
assert [call["stage"] for call in backend.calls] == ["analyze-job"]
def test_unknown_company_fact_is_withheld_even_with_structured_career() -> None:
profile, posting, config = _inputs()
facts = [
fact.model_copy(
update={"content": "가상페이에서 Python API를 개발했다."}
)
if fact.evidence_id == "ev-api"
else fact
for fact in profile.facts
]
facts.append(
EvidenceItem(
evidence_id="ev-career",
category=EvidenceCategory.CAREER,
content="2024년부터 기록회사 백엔드 엔지니어로 근무했다.",
source=EvidenceSource.EMPLOYMENT_RECORD,
date_range={"start": {"year": 2024}, "ongoing": True},
)
)
profile = profile.model_copy(
update={
"facts": facts,
"records": ResumeRecords(
careers=[
CareerRecord(
record_id="career-1",
organization="기록회사",
role="백엔드 엔지니어",
period=RecordPeriod(
start=RecordDate(year=2024), ongoing=True
),
employment_type=EmploymentType.FULL_TIME,
evidence_ids=["ev-career"],
)
]
),
}
)
posting = posting.model_copy(
update={"raw_text": posting.raw_text + "\n회사명 기재 금지"}
)
analysis = _analysis().model_copy(
update={
"constraints": [
PostingConstraint(
constraint_id="blind-company-with-records",
kind=ConstraintKind.BLIND_FIELD,
description="회사명을 본문에 기재하지 않는다.",
source_quote="회사명 기재 금지",
fields=["회사명"],
)
]
}
)
backend = FakeBackend([("analyze-job", analysis)])
result = ResumePipeline(backend).run(profile, posting, config)
assert result.status is PipelineStatus.NEEDS_USER_INPUT
assert result.gate_failures == ["no_generation_safe_evidence"]
assert [call["stage"] for call in backend.calls] == ["analyze-job"]
def test_claim_finding_triggers_targeted_repair_and_re_evaluation() -> None:
profile, posting, config = _inputs()
repaired = _draft("Python 결제 API 응답 시간을 40% 단축")
backend = FakeBackend(
[
*_base_responses(_bad_report("report-before")),
("repair-resume", repaired),
("evaluate-resume", _good_report(repaired)),
]
)
result = ResumePipeline(backend).run(profile, posting, config)
assert result.status is PipelineStatus.PASSED
assert result.repair_attempts == 1
assert result.draft.sections[0].claims[0].text == repaired.sections[0].claims[0].text
assert [call["stage"] for call in backend.calls] == [
"analyze-job",
"map-evidence",
"plan-content",
"draft-resume",
"evaluate-resume",
"repair-resume",
"evaluate-resume",
]
repair_payload = backend.calls[5]["payload"]
assert [item["claim_id"] for item in repair_payload["approved_findings"]] == [
"claim-api"
]
def test_two_failed_repairs_return_needs_user_input_with_bounded_questions() -> None:
profile, posting, config = _inputs()
repaired_once = _draft("Python 결제 API 응답 시간을 40% 단축함")
repaired_twice = _draft("결제 API 응답 시간을 Python으로 40% 단축")
backend = FakeBackend(
[
*_base_responses(_bad_report("report-0")),
("repair-resume", repaired_once),
("evaluate-resume", _bad_report("report-1", repaired_once)),
("repair-resume", repaired_twice),
("evaluate-resume", _bad_report("report-2", repaired_twice)),
]
)
result = ResumePipeline(backend).run(profile, posting, config)
assert result.status is PipelineStatus.NEEDS_USER_INPUT
assert result.repair_attempts == 2
assert 1 <= len(result.questions) <= 3
assert result.gate_failures
assert [call["stage"] for call in backend.calls].count("repair-resume") == 2
assert [call["stage"] for call in backend.calls].count("evaluate-resume") == 3
assert backend.responses == []
def test_judge_cannot_inflate_deterministic_requirement_coverage() -> None:
profile, posting, config = _inputs()
uncovered = _draft()
uncovered.sections[0].claims[0].requirement_ids = []
backend = FakeBackend(
[
("analyze-job", _analysis()),
("map-evidence", _evidence_map()),
("plan-content", _plan()),
("draft-resume", uncovered),
("evaluate-resume", _good_report(uncovered)),
]
)
result = ResumePipeline(backend, max_repair_attempts=0).run(
profile, posting, config
)
assert result.status is PipelineStatus.NEEDS_USER_INPUT
assert result.quality_report is not None
assert result.quality_report.requirement_coverage == 0
assert any(
failure.startswith("requirement_coverage:0<")
for failure in result.gate_failures
)
def test_high_judge_score_cannot_release_unrelated_korean_claim() -> None:
profile, posting, config = _inputs()
hallucinated = _draft(
"고객 만족도를 혁신적으로 높이고 조직 문화를 획기적으로 개선했다"
)
backend = FakeBackend(
[
("analyze-job", _analysis()),
("map-evidence", _evidence_map()),
("plan-content", _plan()),
("draft-resume", hallucinated),
("evaluate-resume", _good_report(hallucinated)),
]
)
result = ResumePipeline(backend, max_repair_attempts=0).run(
profile, posting, config
)
assert result.status is PipelineStatus.NEEDS_USER_INPUT
assert "GROUNDING.LOW_LEXICAL_SUPPORT" in {
finding.code for finding in result.deterministic_findings
}
def test_release_pipeline_rejects_non_strict_evidence_mode() -> None:
profile, posting, config = _inputs()
config = config.model_copy(update={"strict_evidence": False})
backend = FakeBackend([])
with pytest.raises(PipelineError, match="strict_evidence=true"):
ResumePipeline(backend).run(profile, posting, config)
assert backend.calls == []
def test_job_analysis_constraints_participate_in_deterministic_gate() -> None:
profile, posting, config = _inputs()
posting = posting.model_copy(
update={"raw_text": posting.raw_text + " 학교명 기재 금지"}
)
analysis = _analysis().model_copy(
update={
"constraints": [
PostingConstraint(
constraint_id="constraint-school",
kind=ConstraintKind.BLIND_FIELD,
description="학교명을 본문에 기재하지 않는다.",
source_quote="학교명 기재 금지",
fields=["학교명"],
)
]
}
)
constrained_draft = _draft(
"학교명: 합성대학교에서 Python 결제 API 응답 시간을 40% 단축"
)
backend = FakeBackend(
[
("analyze-job", analysis),
("map-evidence", _evidence_map()),
("plan-content", _plan()),
("draft-resume", constrained_draft),
("evaluate-resume", _good_report(constrained_draft)),
]
)
result = ResumePipeline(backend, max_repair_attempts=0).run(
profile, posting, config
)
assert result.status is PipelineStatus.NEEDS_USER_INPUT
assert "PRIVACY.POSTING_FIELD_LEAK" in {
finding.code for finding in result.deterministic_findings
}
def test_consented_employer_form_sensitive_fact_stays_renderer_only() -> None:
profile, posting, _ = _inputs()
config = GenerationConfig(
resume_mode=ResumeMode.EMPLOYER_FORM,
as_of_date=date(2026, 7, 1),
include_photo=True,
allowed_sensitive_categories={SensitiveDataCategory.PHOTO},
employer_required_sensitive_categories={SensitiveDataCategory.PHOTO},
)
employer_draft = _draft(mode=ResumeMode.EMPLOYER_FORM)
backend = FakeBackend(
[
("analyze-job", _analysis()),
("map-evidence", _evidence_map()),
("plan-content", _plan(ResumeMode.EMPLOYER_FORM)),
("draft-resume", employer_draft),
("evaluate-resume", _good_report(employer_draft)),
]
)
result = ResumePipeline(backend).run(profile, posting, config)
assert result.status is PipelineStatus.PASSED
transmitted = json.dumps(
[call["payload"] for call in backend.calls], ensure_ascii=False
)
assert "ev-photo" not in transmitted
assert "secret-photo-token" not in transmitted