Hemprasad Badgujar
Add bm25 prefilter stage; unify candidates cache; validate reasoning quality
cd5f9d4 | """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 | |