matchmaking-v2/tests/test_stack_contracts.py

"""Step-0 DEEP contract tests for the new signal-weighted stack.

These lock the load-bearing contracts BEFORE the actors land, so nothing is missed:
  1. Stack alignment — BOARDS_ENABLED == the target stack; Indeed off-but-registered; every enabled
     board has a builder + normalizer (an enabled board with no normalizer = silent drop).
  2. Builder contract — every enabled board's input builder emits a NON-EMPTY query (no actor 400s).
  3. Normalizer contract (on REAL cached data) — every board's normalizer turns real raw items into
     valid ScoutJobs with the required fields AND DISTINCT dedup keys (the WorkIndia 20→1 collapse class).
  4. Signal contract — each board's normalizer actually yields the signals we claim it does (skills /
     salary / seniority), so a regression in the actor or mapping fails loudly.
  5. Pool contract — floor ≤ reorder, and the gate's three outcomes are well-defined.

Data contracts read .apify_cache (real captured sweeps). A target board with no cache FAILS (so we can't
'forget' to validate a new actor) — except where explicitly allowed pre-build.
"""
from __future__ import annotations

import glob
import json
import os

import pytest

from app.engine import search, normalize as N
from app.engine.search import pool_decision
from app.config import get_settings

# ── the TARGET stack (what we're building toward) ────────────────────────────
TARGET_STACK = ["naukri", "foundit", "linkedin", "timesjobs", "workindia"]
OFF_BUT_REGISTERED = "indeed"            # dropped from enabled, kept registered

# board → cache-file actor prefix (real raw data to validate the normalizer against)
CACHE_PREFIX = {
    "naukri": "blackfalcondata_naukri-jobs-feed",
    "foundit": "shahidirfan_Foundit-Jobs-Scraper",
    "linkedin": "curious_coder_linkedin-jobs-scraper",
    "timesjobs": "shahidirfan_Timesjobs-Scraper",
    "workindia": "shahidirfan_workindia-jobs-scraper",
}
# signals each board MUST confirm (guards actor / mapping regressions). 'offsite' = a real apply link.
SIGNAL_MAP = {
    "naukri": {"skills", "offsite"},
    "foundit": {"skills", "offsite"},
    "linkedin": {"industry"},            # curious_coder structured industries (100%); seniority often N/A in IN
    "timesjobs": {"skills", "salary"},
    "workindia": {"skills"},
}

SAMPLE_PREFS = {
    "title": "Sales Manager", "location": ["Mumbai · India"], "role": ["Sales"],
    "industry": ["Fintech"], "experience": ["Senior"],
    "user_context": {"skills": ["B2B Sales", "Account Management", "CRM"]},
}
REQUIRED_FIELDS = {"id", "title", "organization", "location_country", "required_skills", "details"}
# Stable, COMMITTED per-board raw samples (curated from real sweeps) — so the contract is reproducible
# and not at the mercy of whatever is currently in the volatile .apify_cache.
_SAMPLES_PATH = os.path.join(os.path.dirname(__file__), "fixtures", "board_samples.json")
_SAMPLES = json.load(open(_SAMPLES_PATH)) if os.path.exists(_SAMPLES_PATH) else {}


def _board_sample(board: str):
    return _SAMPLES.get(board)


def _dedup_key(j: dict) -> str:
    return (j.get("apply_url") or "").strip() or f"{j['organization']}|{j['title']}|{j.get('location_city')}".lower()


def _has_signal(sj: dict, signal: str) -> bool:
    d = sj.get("details") or {}
    if signal == "skills":
        return bool(d.get("skills") or sj.get("required_skills"))
    if signal == "salary":
        return sj.get("salary_lpa") is not None or bool(sj.get("payLabel") and "disclos" not in (sj["payLabel"] or "").lower())
    if signal == "seniority":
        return bool(sj.get("seniority_level"))
    if signal == "industry":
        return bool(sj.get("industry") or (sj.get("details") or {}).get("industry"))
    if signal == "offsite":
        return bool(sj.get("offsite_apply"))
    return False


# ── 1. Stack alignment ───────────────────────────────────────────────────────
def test_enabled_equals_target_stack():
    enabled = [b.strip() for b in get_settings().BOARDS_ENABLED.split(",") if b.strip()]
    assert set(enabled) == set(TARGET_STACK), f"BOARDS_ENABLED {enabled} != target {TARGET_STACK}"


def test_indeed_off_but_registered():
    enabled = [b.strip() for b in get_settings().BOARDS_ENABLED.split(",")]
    assert OFF_BUT_REGISTERED not in enabled, "indeed must be DROPPED from enabled"
    assert OFF_BUT_REGISTERED in search.BOARDS, "indeed must stay REGISTERED (off, not deleted)"


def test_every_target_board_has_builder_and_normalizer():
    for b in TARGET_STACK:
        assert b in search.BOARDS, f"target board '{b}' not registered in BOARDS"
        assert b in N.NORMALIZERS, f"target board '{b}' has no normalizer (silent drop)"


# ── 2. Builder contract ──────────────────────────────────────────────────────
def test_every_enabled_builder_emits_nonempty_query():
    enabled = [b for b in TARGET_STACK if b in search.BOARDS]
    for b in enabled:
        inp = search.BOARDS[b][1](dict(SAMPLE_PREFS))
        assert isinstance(inp, dict) and inp, f"{b} builder returned empty input"
        q = inp.get("keyword") or inp.get("title") or (inp.get("jobTitles") or [None])[0] or inp.get("includeKeyword") or inp.get("urls")
        assert q, f"{b} builder emitted no query/keyword/title → actor would 400"


# ── 3 + 4. Normalizer + signal contracts on REAL cached data ─────────────────
@pytest.mark.parametrize("board", TARGET_STACK)
def test_normalizer_contract_on_real_data(board):
    items = _board_sample(board)
    if items is None:
        pytest.fail(f"no committed sample for '{board}' in {_SAMPLES_PATH} — capture one")
    fn = N.NORMALIZERS[board]
    jobs = [sj for it in items[:25] if (sj := fn(it))]
    assert len(jobs) >= 2, f"{board}: normalizer produced < 2 jobs from {len(items)} raw items"
    # required fields present
    for j in jobs:
        missing = REQUIRED_FIELDS - set(j)
        assert not missing, f"{board} ScoutJob missing {missing}"
        assert isinstance(j["required_skills"], list) and isinstance(j["details"], dict)
    # ANTI-COLLAPSE: distinct dedup keys (the WorkIndia 20→1 bug class)
    keys = [_dedup_key(j) for j in jobs]
    assert len(set(keys)) >= max(2, int(0.8 * len(jobs))), \
        f"{board}: dedup keys collapse ({len(set(keys))} unique of {len(jobs)}) — apply_url not unique"
    # SIGNAL contract: the board confirms what we claim, on a majority of jobs
    for sig in SIGNAL_MAP[board]:
        hits = sum(_has_signal(j, sig) for j in jobs)
        assert hits >= 0.5 * len(jobs), f"{board}: signal '{sig}' present on only {hits}/{len(jobs)} (claimed)"


# ── 4b. Legacy-board city aliasing (TimesJobs/Foundit need "Delhi" not "New Delhi") ──
@pytest.mark.parametrize("builder", ["build_timesjobs_input", "build_foundit_input"])
def test_legacy_boards_alias_modern_city_names(builder):
    from app.engine.board_adapters import adapters as A
    fn = getattr(A, builder)
    for modern, legacy in [("New Delhi · India", "Delhi"), ("Bengaluru · India", "Bangalore"),
                           ("Gurugram · India", "Gurgaon")]:
        loc = fn({"title": "Data Architect", "location": [modern], "role": ["Data"]}).get("location")
        assert loc == legacy, f"{builder}: {modern!r} must alias to {legacy!r} (board returns 0 otherwise), got {loc!r}"


# ── 5. Pool contract ─────────────────────────────────────────────────────────
def test_pool_floor_le_reorder():
    s = get_settings()
    assert s.POOL_SAFETY_FLOOR <= s.POOL_REORDER_AT, "floor must be ≤ reorder (reorder early, floor backstop)"


def test_pool_gate_outcomes_well_defined():
    f, r = 90, 100
    assert pool_decision(200, False, f, r) == "pool"
    assert pool_decision(95, False, f, r) == "pool+refill"
    assert pool_decision(50, False, f, r) == "sweep"
    assert pool_decision(9999, True, f, r) == "sweep"      # forced fresh always sweeps