File size: 7,972 Bytes
28c12aa 7d4eb85 28c12aa 7d4eb85 28c12aa 7d4eb85 28c12aa 7d4eb85 28c12aa 700e896 28c12aa cd5f9d4 | 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 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 | """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
|