149 lines
4.6 KiB
Python
149 lines
4.6 KiB
Python
import json
|
|
from pathlib import Path
|
|
|
|
import httpx
|
|
import pytest
|
|
from django.test import override_settings
|
|
|
|
from apps.jobs.models import AiAnalysisCache
|
|
from apps.jobs.services import ai as ai_service
|
|
from apps.jobs.services.ai import analyze_job_text
|
|
|
|
pytestmark = pytest.mark.django_db
|
|
|
|
|
|
def _evaluation_cases() -> list[dict[str, object]]:
|
|
path = Path(__file__).resolve().parents[2] / "fixtures" / "ai" / "evaluation_set.json"
|
|
return json.loads(path.read_text(encoding="utf-8-sig"))["cases"]
|
|
|
|
|
|
@pytest.mark.parametrize("case", _evaluation_cases(), ids=lambda case: case["id"])
|
|
def test_ai_evaluation_cases(monkeypatch, case):
|
|
if case["response_type"] == "disabled":
|
|
with override_settings(OLLAMA_ENABLED=False, OLLAMA_MODEL=""):
|
|
result = analyze_job_text(
|
|
title=case["title"],
|
|
description=case["description"],
|
|
content_hash=case["content_hash"],
|
|
)
|
|
assert result.status == ai_service.AiAnalysisCache.Status.DISABLED
|
|
assert result.error_category == "ollama_disabled"
|
|
assert result.cached is False
|
|
assert result.summary_nl == ""
|
|
return
|
|
|
|
if case["response_type"] == "timeout":
|
|
|
|
class FakeClient:
|
|
def __init__(self, *args, **kwargs):
|
|
pass
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc, tb):
|
|
return False
|
|
|
|
def post(self, *args, **kwargs):
|
|
raise httpx.TimeoutException("timeout")
|
|
|
|
monkeypatch.setattr(httpx, "Client", FakeClient)
|
|
else:
|
|
|
|
class FakeClient:
|
|
def __init__(self, *args, **kwargs):
|
|
pass
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc, tb):
|
|
return False
|
|
|
|
def post(self, *args, **kwargs):
|
|
return _fake_response(case["response"])
|
|
|
|
monkeypatch.setattr(httpx, "Client", FakeClient)
|
|
|
|
with override_settings(
|
|
OLLAMA_ENABLED=True,
|
|
OLLAMA_MODEL="local-model",
|
|
OLLAMA_BASE_URL="http://ollama:11434",
|
|
OLLAMA_TIMEOUT_SECONDS=2,
|
|
):
|
|
result = analyze_job_text(
|
|
title=case["title"],
|
|
description=case["description"],
|
|
content_hash=case["content_hash"],
|
|
)
|
|
|
|
assert result.status == case["expected_status"]
|
|
assert result.error_category == case["expected_error_category"]
|
|
assert result.schema_version == "1.0.0"
|
|
if result.status == ai_service.AiAnalysisCache.Status.OK:
|
|
assert result.summary_nl == case["expected_summary"]
|
|
assert round(result.features.get("support_ratio", 0), 3) == case["expected_support_ratio"]
|
|
assert result.features.get("evidence") == case["expected_evidence"]
|
|
if result.status != ai_service.AiAnalysisCache.Status.OK:
|
|
assert result.features == {}
|
|
|
|
|
|
def _fake_response(response_text: object):
|
|
class FakeResponse:
|
|
def raise_for_status(self):
|
|
return None
|
|
|
|
def json(self):
|
|
return {"response": response_text}
|
|
|
|
return FakeResponse()
|
|
|
|
|
|
def test_ai_cache_reuses_result(monkeypatch):
|
|
calls = 0
|
|
|
|
class FakeClient:
|
|
def __init__(self, *args, **kwargs):
|
|
pass
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc, tb):
|
|
return False
|
|
|
|
def post(self, *args, **kwargs):
|
|
nonlocal calls
|
|
calls += 1
|
|
return _fake_response(
|
|
'{"summary_nl":"Gevalideerde IT-ops advertentie.",'
|
|
'"features":{"support_ratio":0.03,"consultancy_ratio":0.01,'
|
|
'"travel_ratio":0.02,"seniority":"senior","evidence":["IT"]},'
|
|
'"warnings":[]}'
|
|
)
|
|
|
|
monkeypatch.setattr(httpx, "Client", FakeClient)
|
|
|
|
with override_settings(
|
|
OLLAMA_ENABLED=True,
|
|
OLLAMA_MODEL="local-model",
|
|
OLLAMA_BASE_URL="http://ollama:11434",
|
|
):
|
|
first = analyze_job_text(
|
|
"Cloud Engineer",
|
|
"Cloud Engineer vacature. IT-ops taken incl. Azure en Linux.",
|
|
content_hash="cache-hit-case",
|
|
)
|
|
second = analyze_job_text(
|
|
"Cloud Engineer",
|
|
"Cloud Engineer vacature. IT-ops taken incl. Azure en Linux.",
|
|
content_hash="cache-hit-case",
|
|
)
|
|
|
|
assert first.status == ai_service.AiAnalysisCache.Status.OK
|
|
assert second.status == ai_service.AiAnalysisCache.Status.OK
|
|
assert first.cached is False
|
|
assert second.cached is True
|
|
assert calls == 1
|
|
assert AiAnalysisCache.objects.count() == 1
|