"""Step 02 sub-phase 1 contract — load coverage, no-drop, sentinel preservation."""

from __future__ import annotations

from pathlib import Path

import orjson

from preprocess.load import UNKNOWN_SENTINEL, load_candidates

_SCHEMA_PATH = Path(__file__).resolve().parents[1] / "data" / "candidate_schema.json"

# The five document arrays are kept structured (not hoisted); every other schema leaf becomes a column.
_ARRAY_FIELDS = ["career_history", "education", "skills", "certifications", "languages"]


def _write_jsonl(path: Path, records: list[dict]) -> Path:
    path.write_bytes(b"\n".join(orjson.dumps(r) for r in records))
    return path


def test_flatten_covers_every_schema_field(tmp_path: Path) -> None:
    """Adding a field to candidate_schema.json must force a new column — this guards that contract."""
    schema = orjson.loads(_SCHEMA_PATH.read_bytes())
    frame = load_candidates(_write_jsonl(tmp_path / "c.jsonl", [{"candidate_id": "CAND_0000001"}]))
    columns = set(frame.columns)

    assert "candidate_id" in columns
    for array_field in _ARRAY_FIELDS:
        assert array_field in columns, f"array field {array_field} missing"
    for profile_field in schema["properties"]["profile"]["properties"]:
        assert profile_field in columns, f"profile.{profile_field} not covered"
    for signal_field in schema["properties"]["redrob_signals"]["properties"]:
        if signal_field == "expected_salary_range_inr_lpa":
            assert {"expected_salary_min", "expected_salary_max"} <= columns
        else:
            assert signal_field in columns, f"redrob_signals.{signal_field} not covered"


def test_blank_and_malformed_lines_are_skipped_not_dropped(tmp_path: Path) -> None:
    """Two valid records survive; a blank line and a malformed line are skipped, never abort the load."""
    valid = [{"candidate_id": "CAND_0000001"}, {"candidate_id": "CAND_0000002"}]
    path = tmp_path / "c.jsonl"
    lines = [orjson.dumps(valid[0]), b"", b"{not valid json", orjson.dumps(valid[1])]
    path.write_bytes(b"\n".join(lines))
    frame = load_candidates(path)
    assert frame.height == 2
    assert frame["candidate_id"].to_list() == ["CAND_0000001", "CAND_0000002"]


def test_unknown_sentinel_is_preserved(tmp_path: Path) -> None:
    """github_activity_score / offer_acceptance_rate keep their -1 'unknown' marker verbatim on load."""
    record = {
        "candidate_id": "CAND_0000001",
        "redrob_signals": {"github_activity_score": -1, "offer_acceptance_rate": -1},
    }
    frame = load_candidates(_write_jsonl(tmp_path / "c.jsonl", [record]))
    assert frame["github_activity_score"][0] == UNKNOWN_SENTINEL
    assert frame["offer_acceptance_rate"][0] == UNKNOWN_SENTINEL


def test_derived_flat_lists_mirror_the_nested_arrays(tmp_path: Path) -> None:
    """Every nested array gets a name-only flat_* projection; assessment map → verified_skills + {name,score}."""
    record = {
        "candidate_id": "CAND_0000001",
        "career_history": [
            {"title": "AI Engineer", "industry": "Tech", "company_size": "51-200"},
            {"title": "ML Engineer", "industry": "Fintech", "company_size": "11-50"},
        ],
        "education": [{"institution": "IIT", "degree": "BTech", "tier": "tier_1"}],
        "skills": [{"name": "Python"}, {"name": "PyTorch"}],
        "languages": [{"language": "English"}, {"language": "Hindi"}],
        "certifications": [{"name": "AWS SAA"}],
        "redrob_signals": {"skill_assessment_scores": {"Python": 90.0}},
    }
    row = load_candidates(_write_jsonl(tmp_path / "c.jsonl", [record])).row(0, named=True)
    assert row["flat_job_titles"] == ["AI Engineer", "ML Engineer"]
    assert row["flat_career_industries"] == ["Tech", "Fintech"]
    assert row["flat_company_sizes"] == ["51-200", "11-50"]
    assert row["flat_institutions"] == ["IIT"]
    assert row["flat_institution_tiers"] == ["tier_1"]
    assert row["flat_degrees"] == ["BTech"]
    assert row["flat_skills"] == ["Python", "PyTorch"]
    assert row["flat_languages"] == ["English", "Hindi"]
    assert row["flat_certifications"] == ["AWS SAA"]
    assert row["verified_skills"] == ["Python"]
    assert row["skill_assessment_scores"] == [{"name": "Python", "score": 90.0}]


def test_salary_range_is_unnested(tmp_path: Path) -> None:
    """expected_salary_range_inr_lpa.{min,max} flatten to two scalar columns."""
    record = {
        "candidate_id": "CAND_0000001",
        "redrob_signals": {"expected_salary_range_inr_lpa": {"min": 18.0, "max": 32.0}},
    }
    frame = load_candidates(_write_jsonl(tmp_path / "c.jsonl", [record]))
    assert frame["expected_salary_min"][0] == 18.0
    assert frame["expected_salary_max"][0] == 32.0


def test_load_uses_and_writes_cache_for_default_path(tmp_path: Path, monkeypatch) -> None:
    """The DEFAULT_CANDIDATES + limit=None path caches to Parquet; a later call reads the cache, not the JSONL —
    proven by counting how many times the JSONL is actually opened+parsed, not just comparing results."""
    import preprocess.load as load_module

    source = _write_jsonl(tmp_path / "candidates.jsonl", [{"candidate_id": "CAND_0000001"}])
    cache = tmp_path / "cache.parquet"
    monkeypatch.setattr(load_module, "DEFAULT_CANDIDATES", source)
    monkeypatch.setattr(load_module, "LOADED_CANDIDATES_CACHE", cache)

    real_open_text = load_module._open_text
    open_count = 0

    def _counting_open_text(path):
        nonlocal open_count
        open_count += 1
        return real_open_text(path)

    monkeypatch.setattr(load_module, "_open_text", _counting_open_text)

    assert not cache.exists()
    first = load_candidates(source)  # limit=None, path == DEFAULT_CANDIDATES → cache-eligible; a real parse
    assert cache.exists()
    assert first["candidate_id"].to_list() == ["CAND_0000001"]
    assert open_count == 1

    second = load_candidates(source)  # must be served from the Parquet cache — the JSONL is NOT reopened
    assert second["candidate_id"].to_list() == ["CAND_0000001"]
    assert open_count == 1  # unchanged — proves the second call never touched the source file at all


def test_load_cache_not_used_for_custom_path_or_limit(tmp_path: Path, monkeypatch) -> None:
    """A custom path or a limit'd read must NEVER use/pollute the cache — it isn't the real full dataset."""
    import preprocess.load as load_module

    default_source = tmp_path / "candidates.jsonl"
    cache = tmp_path / "cache.parquet"
    monkeypatch.setattr(load_module, "DEFAULT_CANDIDATES", default_source)
    monkeypatch.setattr(load_module, "LOADED_CANDIDATES_CACHE", cache)
    _write_jsonl(default_source, [{"candidate_id": "CAND_0000001"}])

    other_path = _write_jsonl(tmp_path / "custom.jsonl", [{"candidate_id": "CAND_OTHER"}])
    load_candidates(other_path)  # different path → cache untouched
    assert not cache.exists()

    load_candidates(default_source, limit=1)  # limited read → cache untouched
    assert not cache.exists()


def test_load_stale_cache_is_ignored(tmp_path: Path, monkeypatch) -> None:
    """A cache older than the source JSONL is treated as stale and a fresh parse runs instead."""
    import os

    import preprocess.load as load_module

    source = tmp_path / "candidates.jsonl"
    cache = tmp_path / "cache.parquet"
    monkeypatch.setattr(load_module, "DEFAULT_CANDIDATES", source)
    monkeypatch.setattr(load_module, "LOADED_CANDIDATES_CACHE", cache)

    _write_jsonl(source, [{"candidate_id": "CAND_OLD"}])
    load_candidates(source)  # writes the cache reflecting CAND_OLD
    assert cache.exists()

    _write_jsonl(source, [{"candidate_id": "CAND_NEW"}])  # source content changes
    newer = cache.stat().st_mtime + 5
    os.utime(source, (newer, newer))  # force the source's mtime strictly past the cache's

    refreshed = load_candidates(source)
    assert refreshed["candidate_id"].to_list() == ["CAND_NEW"]  # stale cache ignored, live reparse happened