File size: 4,572 Bytes
700e896 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 | """Preprocess validate stage + pipeline — integrity/graded checks, flag-never-drop, contract columns."""
from __future__ import annotations
import copy
from pathlib import Path
import orjson
from preprocess import DATASET_AS_OF, FLAG_COLUMNS, load_candidates, run_preprocess, validate
def _clean_record() -> dict:
"""A coherent candidate that should trip no checks."""
return {
"candidate_id": "CAND_0000001",
"profile": {"years_of_experience": 5.0, "current_title": "AI Engineer"},
"career_history": [
{"start_date": "2018-01-01", "end_date": "2023-01-01", "duration_months": 60,
"is_current": False, "company": "X", "title": "Eng", "industry": "Tech",
"company_size": "51-200", "description": "built things"}
],
"skills": [{"name": "Python", "proficiency": "advanced", "endorsements": 5, "duration_months": 36}],
"redrob_signals": {
"signup_date": "2023-01-01", "last_active_date": "2024-01-01",
"expected_salary_range_inr_lpa": {"min": 10, "max": 20},
"verified_email": True, "verified_phone": True, "linkedin_connected": True,
},
}
def _validate_records(tmp_path: Path, records: list[dict]):
path = tmp_path / "c.jsonl"
path.write_bytes(b"\n".join(orjson.dumps(r) for r in records))
return validate(load_candidates(path), DATASET_AS_OF)
def test_clean_record_trips_no_checks(tmp_path: Path) -> None:
row = _validate_records(tmp_path, [_clean_record()]).row(0, named=True)
assert row["validation_failed"] is False
assert row["incoherence_score"] == 0.0
assert row["trust_score"] == 3
assert row["active_validated"] == 1 and row["experience_validated"] == 1
def test_expert_zero_is_integrity_failure(tmp_path: Path) -> None:
record = _clean_record()
record["skills"] = [{"name": "Rust", "proficiency": "expert", "endorsements": 1, "duration_months": 0}]
row = _validate_records(tmp_path, [record]).row(0, named=True)
assert row["chk_expert_zero"] is True
assert row["validation_failed"] is True # any integrity hit hard-fails
def test_career_dates_inverted_is_integrity_failure(tmp_path: Path) -> None:
record = _clean_record()
record["career_history"][0]["start_date"] = "2023-01-01"
record["career_history"][0]["end_date"] = "2020-01-01"
row = _validate_records(tmp_path, [record]).row(0, named=True)
assert row["chk_career_dates_invalid"] is True and row["validation_failed"] is True
def test_impossible_yoe_is_integrity_failure(tmp_path: Path) -> None:
record = _clean_record()
record["profile"]["years_of_experience"] = 50.0 # vs a ~8yr career span
row = _validate_records(tmp_path, [record]).row(0, named=True)
assert row["chk_yoe_impossible"] is True and row["validation_failed"] is True
def test_salary_incoherence_is_graded_not_hard_fail(tmp_path: Path) -> None:
record = _clean_record()
record["redrob_signals"]["expected_salary_range_inr_lpa"] = {"min": 30, "max": 10}
row = _validate_records(tmp_path, [record]).row(0, named=True)
assert row["chk_salary_incoherent"] is True
assert row["validation_failed"] is False # graded → penalty, not a hard fail
assert abs(row["incoherence_score"] - 0.10) < 1e-6
assert row["salary_range_ok"] == 0
def test_date_order_is_graded(tmp_path: Path) -> None:
record = _clean_record()
record["redrob_signals"]["signup_date"] = "2024-06-01"
record["redrob_signals"]["last_active_date"] = "2024-01-01"
row = _validate_records(tmp_path, [record]).row(0, named=True)
assert row["chk_date_order_invalid"] is True
assert abs(row["incoherence_score"] - 0.15) < 1e-6
assert row["date_order_ok"] == 0 and row["active_validated"] == 0
def test_validate_never_drops_and_emits_every_flag_column(tmp_path: Path) -> None:
validated = _validate_records(tmp_path, [_clean_record(), _clean_record()])
assert validated.height == 2 # flag, never drop
for flag_column in FLAG_COLUMNS:
assert flag_column in validated.columns, f"missing flag column {flag_column}"
def test_run_preprocess_returns_frame_and_reference_date(tmp_path: Path) -> None:
path = tmp_path / "c.jsonl"
records = [_clean_record(), copy.deepcopy(_clean_record())]
records[1]["candidate_id"] = "CAND_0000002"
path.write_bytes(b"\n".join(orjson.dumps(r) for r in records))
frame, as_of = run_preprocess(path)
assert as_of == DATASET_AS_OF
assert frame.height == 2
assert "validation_failed" in frame.columns
|