tests/test_atomic_scoring.py

"""tests/test_atomic_scoring.py — R32 atomic keyword scoring (Phase 13).

The scorer was judging resumes against multi-word JD run-grams requiring verbatim
in-order matches, deflating well-fit resumes (~26% gram vs ~82% atomic on Porter).
These tests lock: atomic decomposition, 2-word skill preservation, the Porter
gram-vs-atomic delta, the Experian calibration bound (stays tracking Jobalytics 58%),
the coverage-aware weave pass trigger/skip, and the feed-URL easy-apply guard.

Run: python -m pytest tests/test_atomic_scoring.py -q
"""
from __future__ import annotations

import json
from pathlib import Path

import pytest

FIX = Path(__file__).parent / "fixtures"

# Top-level import: fails (RED) until atomic_keywords exists in Plan 02.
from src.external_ats import (
    atomic_keywords,
    extract_external_keywords,
    external_coverage,
    filter_scraped_noise,
)


def _read(rel: str) -> str:
    return (FIX / rel).read_text(encoding="utf-8")


# ── 1. Atomic decomposition ──────────────────────────────────────────────────

def test_atomic_decomposition():
    out = [t.lower() for t in atomic_keywords(["saas cloud aws azure"])]
    for w in ("saas", "cloud", "aws", "azure"):
        assert w in out, f"{w} not decomposed from 4-word gram"
    # A genuine 2-word unit is kept whole, not split.
    assert "metric definition" in [t.lower() for t in atomic_keywords(["metric definition"])]
    # Known multi-word skills kept unchanged.
    keep = [t.lower() for t in atomic_keywords(["go-to-market", "product roadmap"])]
    assert "go-to-market" in keep and "product roadmap" in keep
    # Pure stopwords decompose to nothing.
    assert atomic_keywords(["the and for with"]) == []


# ── 2. 2-word skill preservation ─────────────────────────────────────────────

def test_2word_skills_preserved():
    skills = ["go-to-market", "machine learning", "product roadmap",
              "cross-functional", "a/b testing", "stakeholder management"]
    out = [t.lower() for t in atomic_keywords(skills)]
    for s in skills:
        assert s in out, f"2-word skill shredded: {s!r}"


# ── 3. Porter: gram-based deflation vs atomic truth ──────────────────────────

def test_porter_fixture_atomic_vs_gram():
    """Atomic decomposition must materially fix the multi-word-gram deflation.
    (Absolute ceiling is bounded by remaining denominator prose-noise — that's the
    next lever, denominator skill-filtering; here we lock the matching fix itself.)"""
    jd = _read("jds/porter_pm.txt")
    resume = _read("resumes/porter_resume.txt")
    grams = extract_external_keywords(jd)
    gram_pct = external_coverage(grams, resume)["pct"]
    atom_pct = external_coverage(atomic_keywords(grams), resume)["pct"]
    assert gram_pct <= 40, f"gram pct unexpectedly high ({gram_pct})"
    assert atom_pct >= gram_pct + 15, (
        f"atomic ({atom_pct}) must clearly beat gram ({gram_pct}) — the deflation fix"
    )
    assert atom_pct >= 50, f"atomic pct too low ({atom_pct})"


# ── 4. Calibration: atomic must still track a real checker (no re-inflation) ──

def test_calibration_experian():
    jd = _read("jds/experian_tpo.txt")
    resume = _read("resumes/experian_current.txt")
    exp = atomic_keywords(
        filter_scraped_noise(extract_external_keywords(jd), jd, "")
    )
    pct = external_coverage(exp, resume)["pct"]
    assert 48 <= pct <= 68, (
        f"calibration drifted to {pct}% (Jobalytics ~58% ±10 → 48–68). "
        "Atomic scoring must not re-inflate vs real checkers."
    )


# ── 5/6. Coverage-aware weave pass trigger / skip ────────────────────────────

def _fake_llm(counter: dict):
    def _call(cfg, system, user, max_tokens=2000):
        counter["n"] += 1
        return json.dumps({"psm": ["Owned backlog and go-to-market planning."]})
    return type("L", (), {"_call_with_cfg": staticmethod(_call)})()


def test_weave_pass_triggers(monkeypatch):
    from src import resume_v2_natural as v2
    counter = {"n": 0}
    monkeypatch.setattr(v2, "external_coverage",
                        lambda exp, txt: {"pct": 80, "missing": ["backlog"],
                                          "found": 4, "expected": 5, "present": []})
    monkeypatch.setattr(v2, "atomic_keywords", lambda terms: list(terms))
    monkeypatch.setattr(v2, "filter_scraped_noise", lambda terms, jd, co: list(terms))
    monkeypatch.setattr(v2, "_place_sentences_structured",
                        lambda src, sent, alloc: (src + "\nwoven", []))
    monkeypatch.setattr(v2, "_v2_honesty_check", lambda txt, base: (True, []))
    monkeypatch.setattr(v2, "_specialty_hit", lambda t: None)
    decision = {"expected_terms": ["backlog"]}
    out, did = v2._coverage_weave_pass(
        "BASE SRC", decision, "jd text", "Co", ["backlog"], {}, "base text",
        {"name": "Kimi-K2.6", "api_key": "x"}, _fake_llm(counter))
    assert counter["n"] >= 1, "weave pass should make at least one Kimi call (loop-aware)"
    assert did is True and "woven" in out


def test_weave_pass_skipped(monkeypatch):
    from src import resume_v2_natural as v2
    counter = {"n": 0}
    monkeypatch.setattr(v2, "external_coverage",
                        lambda exp, txt: {"pct": 95, "missing": [],
                                          "found": 5, "expected": 5, "present": []})
    monkeypatch.setattr(v2, "atomic_keywords", lambda terms: list(terms))
    monkeypatch.setattr(v2, "filter_scraped_noise", lambda terms, jd, co: list(terms))
    decision = {"expected_terms": ["backlog"]}
    out, did = v2._coverage_weave_pass(
        "BASE SRC", decision, "jd", "Co", ["backlog"], {}, "base",
        {"name": "Kimi-K2.6", "api_key": "x"}, _fake_llm(counter))
    assert counter["n"] == 0, "weave pass must be skipped when pct >= 90"
    assert did is False and out == "BASE SRC"


# ── 7. Single-job isolation: easy-apply feed guard ───────────────────────────

def test_isolation_easy_apply_guard():
    from src.resume_v2_natural import _sanitize_jd_v2
    big = "Real product management job description content here. " * 150  # >4000 chars
    feed = big + " Easy Apply  Easy Apply  Easy Apply " + big
    out = _sanitize_jd_v2(feed)
    assert len(out) <= 4000, f"feed page not truncated ({len(out)} chars)"
    # A clean single JD (no Easy Apply spam) is returned intact (failsafe).
    clean = ("We seek a Product Manager to own the roadmap, run discovery, write PRDs, "
             "drive go-to-market, and partner with engineering on agile delivery. ") * 3
    assert "roadmap" in _sanitize_jd_v2(clean).lower()