tests/test_api_parity.py

"""
Parity test: verifies that the API code path (generate_resume_for_api) and the
batch pipeline code path (_run_provider_chain direct call) produce IDENTICAL
deterministic scores for the same (resume_pdf, jd_text) inputs when using
StubProvider (bypasses LLM entirely).

This test is the gatekeeper for R-EXT2: "Score must be identical to the web app."

Uses StubProvider so no API keys or network are needed.  Output is deterministic.

Skips gracefully if data/resume/resume.pdf is absent (CI without resume artifact).
"""
import os
import sys
import tempfile

import pytest

# Ensure project root is on the path regardless of how pytest is invoked.
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

# Resolve paths relative to the project root (one level above tests/).
_PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
FIXTURE_PDF = os.path.join(_PROJECT_ROOT, "data", "resume", "resume.pdf")
FIXTURE_JD = os.path.join(_PROJECT_ROOT, "tests", "fixtures", "sample_jd.txt")


def test_fixture_jd_exists():
    """Verify the sample JD fixture exists and has meaningful PM content."""
    assert os.path.exists(FIXTURE_JD), f"Fixture JD not found: {FIXTURE_JD}"
    with open(FIXTURE_JD) as f:
        content = f.read()
    assert len(content) > 200, (
        f"Fixture JD is too short ({len(content)} chars); expected > 200"
    )
    assert "Product Manager" in content or "product manager" in content.lower(), (
        "Fixture JD should be a PM job description"
    )
    print(f"Fixture JD OK: {len(content)} chars")


@pytest.mark.skipif(
    not os.path.exists(FIXTURE_PDF),
    reason="data/resume/resume.pdf not found — skip on CI without resume artifact",
)
def test_api_parity_deterministic():
    """
    PATH A — generate_resume_for_api() (the API helper in api_server.py)
      Uses build_provider_chain() monkey-patched to return [StubProvider()].

    PATH B — ResumeCustomizer._run_provider_chain() called directly
      with [StubProvider()] as the explicit provider chain.

    Both paths use the SAME base resume (data/resume/resume.pdf) and the same
    JD text (tests/fixtures/sample_jd.txt).

    Asserts: ats_readability, independent_jd_match, and download_allowed are
    IDENTICAL between the two code paths.
    """
    import api_server
    import src.providers as providers_mod
    from src.providers import StubProvider
    from src.resume_parser_v2 import parse_resume_pdf
    from src.resume_customizer import ResumeCustomizer
    from src.llm_client import LLMClient

    with open(FIXTURE_PDF, "rb") as f:
        pdf_bytes = f.read()
    with open(FIXTURE_JD) as f:
        jd_text = f.read()

    job_title = "Product Manager"
    company = "TestCo"

    # ── PATH A: via API helper ──────────────────────────────────────────────────
    # Monkey-patch build_provider_chain (used inside generate_resume_for_api via a
    # local import from src.providers) so it returns only StubProvider.
    # We patch the function on the module object; local imports inside the function
    # resolve against the module at call time.
    original_build = providers_mod.build_provider_chain
    providers_mod.build_provider_chain = lambda llm=None: [StubProvider()]
    try:
        docx_path_a, report_a = api_server.generate_resume_for_api(
            pdf_bytes, jd_text, job_title, company
        )
    finally:
        providers_mod.build_provider_chain = original_build

    assert docx_path_a is not None, (
        "generate_resume_for_api returned None — pipeline produced no output"
    )
    assert isinstance(report_a, dict), (
        f"generate_resume_for_api returned non-dict report: {type(report_a)}"
    )

    # ── PATH B: via batch pipeline directly ────────────────────────────────────
    base_resume = parse_resume_pdf(FIXTURE_PDF)

    with tempfile.TemporaryDirectory() as tmp:
        # Build a ResumeCustomizer with a dummy LLM (not used by StubProvider).
        llm = LLMClient.__new__(LLMClient)
        customizer = ResumeCustomizer(llm, base_resume.to_flat_text(), tmp)

        job_b: dict = {
            "title": job_title,
            "company": company,
            "description": jd_text[:16000],  # match the API's 16000-char truncation
            "ats_keywords": "",
            "_raw_assessment": {},
        }
        filepath_b = os.path.join(tmp, "test_parity.docx")

        result_b = customizer._run_provider_chain(
            job_b,
            filepath_b,
            [StubProvider()],
            base_resume_override=base_resume,
        )
        report_b = job_b.get("_v2_report", {}) or {}

    assert result_b is not None, (
        "_run_provider_chain returned None — pipeline produced no output for Path B"
    )

    # ── PARITY ASSERTIONS ───────────────────────────────────────────────────────
    scores_a = report_a.get("estimated_scores", {}) or {}
    scores_b = report_b.get("estimated_scores", {}) or {}

    ats_a = scores_a.get("ats_readability")
    ats_b = scores_b.get("ats_readability")
    assert ats_a == ats_b, (
        f"ATS readability mismatch: API path={ats_a!r} vs batch path={ats_b!r}\n"
        f"  report_a keys: {list(report_a.keys())}\n"
        f"  report_b keys: {list(report_b.keys())}"
    )

    ind_a = report_a.get("independent_jd_match")
    ind_b = report_b.get("independent_jd_match")
    assert ind_a == ind_b, (
        f"independent_jd_match mismatch: API path={ind_a!r} vs batch path={ind_b!r}"
    )

    dl_a = bool(report_a.get("download_allowed"))
    dl_b = bool(report_b.get("download_allowed"))
    assert dl_a == dl_b, (
        f"download_allowed mismatch: API path={dl_a!r} vs batch path={dl_b!r}"
    )

    print(
        f"PARITY OK — "
        f"ATS readability: {ats_a}, "
        f"independent_jd_match: {ind_a}, "
        f"download_allowed: {dl_a}"
    )