234 lines
7.3 KiB
Python
234 lines
7.3 KiB
Python
"""Shared fixtures.
|
|
|
|
No live Anthropic calls anywhere in the default suite: every test either injects
|
|
``FakeScorer`` (in place of the whole adapter) or a fake messages resource (to exercise
|
|
the adapter itself).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import io
|
|
import os
|
|
import re
|
|
import time
|
|
from collections.abc import Callable, Iterator, Sequence
|
|
from typing import Any
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from pypdf import PdfReader, PdfWriter
|
|
|
|
from app.core.config import Settings
|
|
from app.main import create_app
|
|
from app.models.scoring import ATSScore
|
|
|
|
# --- Minimal PDF construction ------------------------------------------------
|
|
#
|
|
# Built by hand rather than with a writer library so tests can produce documents
|
|
# with exactly-known text, including deliberately broken ones.
|
|
|
|
_SCORE_MARKER = re.compile(r"SCORE\s+(\d+)")
|
|
_NAME_MARKER = re.compile(r"Candidate (\S+)")
|
|
|
|
|
|
def _content_stream(lines: Sequence[str]) -> bytes:
|
|
if not lines:
|
|
return b""
|
|
parts = ["BT", "/F1 12 Tf", "14 TL", "72 720 Td"]
|
|
for index, line in enumerate(lines):
|
|
escaped = line.replace("\\", r"\\").replace("(", r"\(").replace(")", r"\)")
|
|
if index:
|
|
parts.append("T*")
|
|
parts.append(f"({escaped}) Tj")
|
|
parts.append("ET")
|
|
return "\n".join(parts).encode("latin-1")
|
|
|
|
|
|
def make_pdf(pages: Sequence[Sequence[str]]) -> bytes:
|
|
"""A structurally valid PDF (real xref table) containing the given text lines."""
|
|
page_count = len(pages)
|
|
font_id = 3 + 2 * page_count
|
|
objects: dict[int, bytes] = {}
|
|
|
|
kids = " ".join(f"{3 + 2 * i} 0 R" for i in range(page_count))
|
|
objects[1] = b"<< /Type /Catalog /Pages 2 0 R >>"
|
|
objects[2] = f"<< /Type /Pages /Kids [{kids}] /Count {page_count} >>".encode()
|
|
|
|
for index, lines in enumerate(pages):
|
|
page_id = 3 + 2 * index
|
|
content_id = 4 + 2 * index
|
|
objects[page_id] = (
|
|
f"<< /Type /Page /Parent 2 0 R /MediaBox [0 0 612 792] "
|
|
f"/Resources << /Font << /F1 {font_id} 0 R >> >> "
|
|
f"/Contents {content_id} 0 R >>"
|
|
).encode()
|
|
stream = _content_stream(lines)
|
|
objects[content_id] = (
|
|
b"<< /Length " + str(len(stream)).encode() + b" >>\nstream\n" + stream + b"\nendstream"
|
|
)
|
|
|
|
objects[font_id] = b"<< /Type /Font /Subtype /Type1 /BaseFont /Helvetica >>"
|
|
|
|
out = bytearray(b"%PDF-1.4\n")
|
|
offsets: dict[int, int] = {}
|
|
for number in sorted(objects):
|
|
offsets[number] = len(out)
|
|
out += f"{number} 0 obj\n".encode() + objects[number] + b"\nendobj\n"
|
|
|
|
xref_offset = len(out)
|
|
size = max(objects) + 1
|
|
out += f"xref\n0 {size}\n".encode()
|
|
out += b"0000000000 65535 f \n"
|
|
for number in range(1, size):
|
|
out += f"{offsets[number]:010d} 00000 n \n".encode()
|
|
out += (f"trailer\n<< /Size {size} /Root 1 0 R >>\nstartxref\n{xref_offset}\n%%EOF\n").encode()
|
|
return bytes(out)
|
|
|
|
|
|
def make_encrypted_pdf(password: str = "secret") -> bytes:
|
|
reader = PdfReader(io.BytesIO(make_pdf([["Confidential resume content here."]])))
|
|
writer = PdfWriter()
|
|
for page in reader.pages:
|
|
writer.add_page(page)
|
|
writer.encrypt(password, algorithm="RC4-128")
|
|
buffer = io.BytesIO()
|
|
writer.write(buffer)
|
|
return buffer.getvalue()
|
|
|
|
|
|
def resume_pdf(name: str, score: int | None = None, *, extra: str = "") -> bytes:
|
|
"""A readable resume PDF. ``score`` is embedded so fakes can score deterministically."""
|
|
lines = [
|
|
f"Candidate {name}",
|
|
"Experience: Python, FastAPI, Docker, REST APIs.",
|
|
"Built and operated backend services for four years.",
|
|
]
|
|
if score is not None:
|
|
lines.append(f"SCORE {score}")
|
|
if extra:
|
|
lines.append(extra)
|
|
return make_pdf([lines])
|
|
|
|
|
|
# --- Fake scorer -------------------------------------------------------------
|
|
|
|
|
|
def default_score(resume_text: str) -> ATSScore:
|
|
match = _SCORE_MARKER.search(resume_text)
|
|
value = int(match.group(1)) if match else 50
|
|
name = _NAME_MARKER.search(resume_text)
|
|
return ATSScore(
|
|
candidate_name=name.group(1) if name else None,
|
|
job_title="Backend Engineer",
|
|
current_company="Acme",
|
|
years_experience=4,
|
|
match_score=value,
|
|
matched_keywords=["Python", "FastAPI"],
|
|
missing_keywords=["Kubernetes"],
|
|
summary_critique="Solid backend experience with a gap in orchestration.",
|
|
)
|
|
|
|
|
|
class FakeScorer:
|
|
"""Records call timing and concurrency so orchestration behaviour is assertable."""
|
|
|
|
def __init__(
|
|
self,
|
|
handler: Callable[[str], ATSScore] | None = None,
|
|
*,
|
|
delay: float = 0.0,
|
|
) -> None:
|
|
self._handler = handler or default_score
|
|
self._delay = delay
|
|
self.calls: list[str] = []
|
|
self.events: list[tuple[str, str, float]] = []
|
|
self._active = 0
|
|
self.max_concurrent = 0
|
|
|
|
async def score(self, job_description: str, resume_text: str) -> ATSScore:
|
|
self.calls.append(resume_text)
|
|
self._active += 1
|
|
self.max_concurrent = max(self.max_concurrent, self._active)
|
|
self.events.append(("start", resume_text, time.perf_counter()))
|
|
try:
|
|
if self._delay:
|
|
await asyncio.sleep(self._delay)
|
|
return self._handler(resume_text)
|
|
finally:
|
|
self._active -= 1
|
|
self.events.append(("end", resume_text, time.perf_counter()))
|
|
|
|
|
|
# --- App / settings fixtures -------------------------------------------------
|
|
|
|
|
|
def build_settings(**overrides: Any) -> Settings:
|
|
base: dict[str, Any] = {
|
|
"openai_api_key": "test-key",
|
|
"openai_model": "gpt-5.4-mini",
|
|
"openai_max_output_tokens": 4000,
|
|
"openai_effort": "low",
|
|
"scoring_concurrency": 3,
|
|
"max_resumes_per_request": 5,
|
|
"max_pdf_size_mb": 10,
|
|
"max_jd_chars": 5_000,
|
|
"max_resume_chars": 60_000,
|
|
"log_format": "text",
|
|
}
|
|
base.update(overrides)
|
|
return Settings(_env_file=None, **base)
|
|
|
|
|
|
@pytest.fixture
|
|
def settings() -> Settings:
|
|
return build_settings()
|
|
|
|
|
|
@pytest.fixture
|
|
def fake_scorer() -> FakeScorer:
|
|
return FakeScorer()
|
|
|
|
|
|
@pytest.fixture
|
|
def make_client() -> Callable[..., TestClient]:
|
|
"""Factory so a test can vary settings or scorer without a new fixture."""
|
|
clients: list[TestClient] = []
|
|
|
|
def _factory(
|
|
scorer: Any | None = None,
|
|
**setting_overrides: Any,
|
|
) -> TestClient:
|
|
app = create_app(
|
|
settings=build_settings(**setting_overrides),
|
|
scorer=scorer or FakeScorer(),
|
|
)
|
|
client = TestClient(app)
|
|
client.__enter__()
|
|
clients.append(client)
|
|
return client
|
|
|
|
yield _factory
|
|
|
|
for client in clients:
|
|
client.__exit__(None, None, None)
|
|
|
|
|
|
@pytest.fixture
|
|
def client(make_client: Callable[..., TestClient], fake_scorer: FakeScorer) -> TestClient:
|
|
return make_client(scorer=fake_scorer)
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _hermetic_env(monkeypatch: pytest.MonkeyPatch) -> Iterator[None]:
|
|
"""Keep the suite hermetic.
|
|
|
|
A real key must never leak in from the environment, and a developer's local
|
|
OPENAI_MODEL / limits must not change what the tests assert.
|
|
"""
|
|
for name in list(os.environ):
|
|
upper = name.upper()
|
|
if upper.startswith(("OPENAI_", "ANTHROPIC_", "SCORING_", "MAX_", "LOG_")):
|
|
monkeypatch.delenv(name, raising=False)
|
|
yield
|