213 lines
8.1 KiB
Python
213 lines
8.1 KiB
Python
from __future__ import annotations
|
|
|
|
from decimal import Decimal
|
|
|
|
import pytest
|
|
|
|
from apps.jobs.models import AiAnalysisCache, Employer, JobPosting, ScoreRun
|
|
from apps.jobs.services import scoring
|
|
from apps.jobs.services.ai import AiAnalysis
|
|
from apps.jobs.services.scoring import calculate_score
|
|
|
|
|
|
@pytest.fixture
|
|
def matching_job(db):
|
|
employer = Employer.objects.create(
|
|
name="Example Public IT",
|
|
normalized_name="example public it",
|
|
domain="jobs.example.org",
|
|
is_direct_employer=True,
|
|
)
|
|
return JobPosting.objects.create(
|
|
employer=employer,
|
|
original_title="Infrastructure Engineer",
|
|
normalized_title="infrastructure engineer",
|
|
canonical_url="https://jobs.example.org/jobs/1",
|
|
canonical_key="a" * 64,
|
|
content_hash="b" * 64,
|
|
description_text=(
|
|
"Beheer Microsoft 365 en VMware. Hybride werk en beperkte tweedelijnssupport."
|
|
),
|
|
raw_location="Hasselt, Limburg",
|
|
region="Limburg",
|
|
municipality="Hasselt",
|
|
postal_code="3500",
|
|
latitude=Decimal("50.930700"),
|
|
longitude=Decimal("5.332500"),
|
|
workplace_type=JobPosting.Workplace.HYBRID,
|
|
employment_types=["full_time", "permanent"],
|
|
skills_required=["Microsoft 365", "VMware"],
|
|
analysis_features={"support_ratio": 0.1, "public_sector_signal": 0.8},
|
|
direct_employer=True,
|
|
recruiter=False,
|
|
extraction_confidence=Decimal("0.95"),
|
|
status=JobPosting.Status.ACTIVE,
|
|
)
|
|
|
|
|
|
def test_matching_job_gets_recommendation(matching_job, profile):
|
|
result = calculate_score(matching_job, profile)
|
|
assert result.score >= profile.recommendation_threshold
|
|
assert result.recommendation in {
|
|
ScoreRun.Recommendation.STRONG,
|
|
ScoreRun.Recommendation.POSSIBLE,
|
|
}
|
|
assert not result.hard_exclusions
|
|
|
|
|
|
def test_hard_title_exclusion_wins(matching_job, profile):
|
|
matching_job.original_title = "IT Sales Infrastructure Engineer"
|
|
matching_job.save(update_fields=["original_title"])
|
|
result = calculate_score(matching_job, profile)
|
|
assert result.recommendation == ScoreRun.Recommendation.HIDDEN
|
|
assert any("sales" in reason.lower() for reason in result.hard_exclusions)
|
|
|
|
|
|
def test_non_it_vacancy_is_hidden_for_it_profile(matching_job, profile):
|
|
matching_job.original_title = "Spontane sollicitatie - logopedist"
|
|
matching_job.normalized_title = "spontane sollicitatie logopedist"
|
|
matching_job.description_text = "Werk met het digitale patiëntendossier in ons ziekenhuis."
|
|
|
|
result = calculate_score(matching_job, profile)
|
|
|
|
assert result.recommendation == ScoreRun.Recommendation.HIDDEN
|
|
assert any("Geen aantoonbaar IT-signaal" in reason for reason in result.hard_exclusions)
|
|
assert result.evidence["it_relevance"]["relevant"] is False
|
|
|
|
|
|
def test_distance_boundary_inclusief(matching_job, profile, monkeypatch):
|
|
profile.max_distance_km = 50
|
|
profile.save(update_fields=["max_distance_km"])
|
|
monkeypatch.setattr(scoring, "haversine_km", lambda *_args, **_kwargs: 50.0)
|
|
matching_job.refresh_from_db()
|
|
matching_job.save(update_fields=["latitude", "longitude"])
|
|
result = calculate_score(matching_job, profile)
|
|
assert result.evidence["distance_km"] == 50.0
|
|
assert result.recommendation != ScoreRun.Recommendation.HIDDEN
|
|
assert not any("boven maximum" in issue for issue in result.hard_exclusions)
|
|
|
|
|
|
def test_remote_job_heeft_geen_afstandsexclusie(matching_job, profile):
|
|
matching_job.workplace_type = JobPosting.Workplace.REMOTE
|
|
matching_job.postal_code = ""
|
|
matching_job.municipality = ""
|
|
matching_job.raw_location = ""
|
|
matching_job.save(
|
|
update_fields=["workplace_type", "postal_code", "municipality", "raw_location"]
|
|
)
|
|
matching_job.latitude = None
|
|
matching_job.longitude = None
|
|
result = calculate_score(matching_job, profile)
|
|
assert result.recommendation != ScoreRun.Recommendation.HIDDEN
|
|
assert all("boven maximum" not in issue for issue in result.hard_exclusions)
|
|
|
|
|
|
def test_onbekende_locatie_leidt_niet_tot_automatische_uitsluiting(matching_job, profile):
|
|
matching_job.postal_code = ""
|
|
matching_job.municipality = ""
|
|
matching_job.raw_location = "onbekend"
|
|
matching_job.latitude = None
|
|
matching_job.longitude = None
|
|
matching_job.save(
|
|
update_fields=["postal_code", "municipality", "raw_location", "latitude", "longitude"]
|
|
)
|
|
result = calculate_score(matching_job, profile)
|
|
assert result.evidence["distance_km"] is None
|
|
assert result.recommendation != ScoreRun.Recommendation.HIDDEN
|
|
|
|
|
|
def test_score_is_tijdzoneonafhankelijk(profile, user, matching_job):
|
|
utc_profile = profile
|
|
utc_profile.timezone = "UTC"
|
|
utc_profile.save(update_fields=["timezone"])
|
|
result_utc = calculate_score(matching_job, utc_profile)
|
|
|
|
ny_profile = profile.__class__.objects.create(
|
|
user=user,
|
|
name="UTC vergelijken",
|
|
is_active=False,
|
|
home_municipality=utc_profile.home_municipality,
|
|
home_latitude=utc_profile.home_latitude,
|
|
home_longitude=utc_profile.home_longitude,
|
|
max_distance_km=utc_profile.max_distance_km,
|
|
desired_titles=utc_profile.desired_titles,
|
|
excluded_titles=utc_profile.excluded_titles,
|
|
desired_skills=utc_profile.desired_skills,
|
|
excluded_skills=utc_profile.excluded_skills,
|
|
allowed_employment_types=utc_profile.allowed_employment_types,
|
|
preferred_workplace=utc_profile.preferred_workplace,
|
|
preferred_regions=utc_profile.preferred_regions,
|
|
excluded_regions=utc_profile.excluded_regions,
|
|
recommendation_threshold=utc_profile.recommendation_threshold,
|
|
top_match_threshold=utc_profile.top_match_threshold,
|
|
digest_time=utc_profile.digest_time,
|
|
quiet_hours_start=utc_profile.quiet_hours_start,
|
|
quiet_hours_end=utc_profile.quiet_hours_end,
|
|
learning_enabled=utc_profile.learning_enabled,
|
|
weights=utc_profile.weights,
|
|
timezone="America/New_York",
|
|
)
|
|
result_ny = calculate_score(matching_job, ny_profile)
|
|
assert result_ny.score == result_utc.score
|
|
|
|
|
|
def test_ai_score_influence_is_opt_in_and_bounded(matching_job, profile, monkeypatch):
|
|
profile.ai_scoring_enabled = True
|
|
profile.weights = {**profile.weights, "ai": 999}
|
|
profile.save(update_fields=["ai_scoring_enabled", "weights"])
|
|
|
|
monkeypatch.setattr(
|
|
scoring,
|
|
"analyze_job_text",
|
|
lambda *args, **kwargs: AiAnalysis(
|
|
features={
|
|
"support_ratio": 0.1,
|
|
"consultancy_ratio": 0.0,
|
|
"travel_ratio": 0.0,
|
|
"seniority": "senior",
|
|
"evidence": ["microsoft 365"],
|
|
},
|
|
summary_nl="Sterke technische rol.",
|
|
warnings=[],
|
|
model="local-model",
|
|
status=AiAnalysisCache.Status.OK,
|
|
error_category="",
|
|
cached=False,
|
|
),
|
|
)
|
|
|
|
result = calculate_score(matching_job, profile)
|
|
assert result.recommendation in {
|
|
ScoreRun.Recommendation.STRONG,
|
|
ScoreRun.Recommendation.POSSIBLE,
|
|
}
|
|
assert result.model_version == "local-model"
|
|
assert result.evidence["ai"]["status"] == AiAnalysisCache.Status.OK
|
|
assert result.evidence["ai"]["weight_applied"] == 20.0
|
|
assert "ai" in result.evidence
|
|
|
|
|
|
def test_ai_failure_keeps_deterministic_scoring(profile, matching_job, monkeypatch):
|
|
profile.ai_scoring_enabled = True
|
|
profile.weights = {**profile.weights, "ai": 10}
|
|
profile.save(update_fields=["ai_scoring_enabled", "weights"])
|
|
|
|
monkeypatch.setattr(
|
|
scoring,
|
|
"analyze_job_text",
|
|
lambda *args, **kwargs: AiAnalysis(
|
|
features={},
|
|
summary_nl="",
|
|
warnings=["service-timeout"],
|
|
model="local-model",
|
|
status=AiAnalysisCache.Status.TIMEOUT,
|
|
error_category="timeout",
|
|
cached=False,
|
|
),
|
|
)
|
|
|
|
result = calculate_score(matching_job, profile)
|
|
assert result.evidence["ai"]["status"] == AiAnalysisCache.Status.TIMEOUT
|
|
assert "AI-analyse" in "".join(result.concerns)
|
|
assert "components" in result.__dict__
|