109 lines
3.1 KiB
Python
109 lines
3.1 KiB
Python
from __future__ import annotations
|
|
|
|
from io import StringIO
|
|
|
|
import pytest
|
|
from django.core.management import call_command
|
|
from django.core.management.base import CommandError
|
|
|
|
from apps.jobs.models import ScoreRun
|
|
from apps.jobs.services.scoring import score_and_save
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_import_geodata_validate_only_and_import_command(tmp_path):
|
|
csv_path = tmp_path / "geodata.csv"
|
|
csv_path.write_text(
|
|
"postal_code,municipality,region,latitude,longitude\n"
|
|
"9000,Gent,Vlaams-Brabant,51.05,3.73\n"
|
|
"1040,Brussel,Brussels Hoofdstedelijk Gewest,50.85,4.35\n"
|
|
)
|
|
out = StringIO()
|
|
call_command(
|
|
"import_geodata",
|
|
str(csv_path),
|
|
"--source-name",
|
|
"local-test",
|
|
"--dataset-version",
|
|
"2026-01-01",
|
|
"--validate-only",
|
|
"--license-name",
|
|
"Test Dataset",
|
|
"--license-url",
|
|
"https://example.org/license",
|
|
stdout=out,
|
|
)
|
|
assert "Validatie geslaagd (2 rijen)." in out.getvalue()
|
|
|
|
out = StringIO()
|
|
call_command(
|
|
"import_geodata",
|
|
str(csv_path),
|
|
"--source-name",
|
|
"local-test",
|
|
"--dataset-version",
|
|
"2026-01-01",
|
|
"--license-name",
|
|
"Test Dataset",
|
|
"--license-url",
|
|
"https://example.org/license",
|
|
stdout=out,
|
|
)
|
|
assert "Geocodebron geimporteerd (4 lookuprijen)" in out.getvalue()
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_import_geodata_rejects_half_license_metadata(tmp_path):
|
|
csv_path = tmp_path / "geodata.csv"
|
|
csv_path.write_text(
|
|
"postal_code,municipality,region,latitude,longitude\n9000,Gent,Vlaams-Brabant,51.05,3.73\n"
|
|
)
|
|
out = StringIO()
|
|
with pytest.raises(CommandError):
|
|
call_command(
|
|
"import_geodata",
|
|
str(csv_path),
|
|
"--source-name",
|
|
"local-test",
|
|
"--dataset-version",
|
|
"2026-01-01",
|
|
"--license-name",
|
|
"Test Dataset",
|
|
stdout=out,
|
|
)
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_import_geodata_triggers_targeted_rescore(tmp_path, profile, job):
|
|
job.postal_code = "9000"
|
|
job.municipality = "Gent"
|
|
job.save(update_fields=["postal_code", "municipality"])
|
|
|
|
score_and_save(job, profile)
|
|
initial_scores = ScoreRun.objects.filter(profile=profile, job=job).count()
|
|
|
|
csv_path = tmp_path / "geodata.csv"
|
|
csv_path.write_text(
|
|
"postal_code,municipality,region,latitude,longitude\n9000,Gent,Vlaams-Brabant,51.05,3.73\n"
|
|
)
|
|
out = StringIO()
|
|
call_command(
|
|
"import_geodata",
|
|
str(csv_path),
|
|
"--source-name",
|
|
"local-test",
|
|
"--dataset-version",
|
|
"2026-01-01",
|
|
"--license-name",
|
|
"Test Dataset",
|
|
"--license-url",
|
|
"https://example.org/license",
|
|
stdout=out,
|
|
)
|
|
assert "scoreregels herberekend voor aangepaste geodata" in out.getvalue()
|
|
assert ScoreRun.objects.filter(profile=profile, job=job).count() == initial_scores + 1
|
|
|
|
latest = ScoreRun.objects.filter(profile=profile, job=job).order_by("-created_at").first()
|
|
assert latest is not None
|
|
assert latest.evidence["distance_km"] is not None
|