File size: 3,498 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 | """Preprocess materialize stage — counts (raw flat_skills), inline experience_band, clean signals, text."""
from __future__ import annotations
import copy
from pathlib import Path
import orjson
from preprocess import DATASET_AS_OF, load_candidates, materialize
def _record() -> dict:
"""A populated candidate (keeps every nested column struct-typed)."""
return {
"candidate_id": "CAND_0000001",
"profile": {"years_of_experience": 7.0, "current_title": "AI Engineer",
"current_company": "Acme", "current_industry": "Tech"},
"career_history": [
{"title": "AI Engineer", "company": "Acme", "industry": "Tech", "company_size": "51-200",
"duration_months": 40, "is_current": True, "start_date": "2021-01-01",
"end_date": None, "description": "ranking systems"}
],
"education": [{"institution": "IIT", "degree": "BTech", "field_of_study": "CS",
"start_year": 2014, "end_year": 2018, "tier": "tier_1"}],
"skills": [
{"name": "Python", "proficiency": "expert", "endorsements": 9, "duration_months": 60},
{"name": "SQL", "proficiency": "beginner", "endorsements": 1, "duration_months": 12},
],
"certifications": [{"name": "AWS SAA", "issuer": "AWS", "year": 2022}],
"redrob_signals": {
"github_activity_score": 70, "offer_acceptance_rate": -1, "notice_period_days": 15,
"expected_salary_range_inr_lpa": {"min": 20, "max": 40},
"skill_assessment_scores": {"Python": 88.0},
},
}
def _materialize(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 materialize(load_candidates(path), DATASET_AS_OF)
def test_experience_band_is_computed_inline(tmp_path: Path) -> None:
bands = []
for yoe in (2.0, 5.0, 8.0, 12.0, 20.0):
record = _record()
record["profile"]["years_of_experience"] = yoe
bands.append(_materialize(tmp_path, [record]).row(0, named=True)["experience_band"])
assert bands == ["junior", "mid", "senior", "lead", "principal"]
def test_counts_use_raw_flat_skills(tmp_path: Path) -> None:
row = _materialize(tmp_path, [_record()]).row(0, named=True)
assert row["num_skills"] == 2 # from flat_skills, no canonicalization
assert row["num_advanced_skills"] == 1 and row["num_expert_skills"] == 1
assert row["num_roles"] == 1 and row["num_companies"] == 1
assert row["num_assessments_taken"] == 1 and row["max_skill_assessment"] == 88.0
def test_sentinel_signals_are_cleaned(tmp_path: Path) -> None:
row = _materialize(tmp_path, [_record()]).row(0, named=True)
assert row["has_github"] is True and row["github_score_clean"] == 70.0
assert row["has_offer_history"] is False and row["offer_acceptance_rate_clean"] is None
assert row["salary_mid"] == 30.0 and row["is_available_soon"] is True
def test_embedded_text_is_built(tmp_path: Path) -> None:
row = _materialize(tmp_path, [_record()]).row(0, named=True)
assert row["text"] == row["summary_text"]
assert "AI Engineer" in row["text"] and "senior-band" in row["text"]
assert len(row["embed_source_hash"]) == 16
def test_materialize_never_drops(tmp_path: Path) -> None:
second = copy.deepcopy(_record())
second["candidate_id"] = "CAND_0000002"
assert _materialize(tmp_path, [_record(), second]).height == 2
|