Jens
faeb58ef6d
GeoIntel release gates / Compile, test, contracts and builds (push) Successful in 1m49s
GeoIntel release gates / Python and npm vulnerability policy (push) Successful in 21s
GeoIntel release gates / Production AI image, SBOM and container scan (push) Successful in 5m39s
GeoIntel release gates / Deploy exact gated revision to Unraid (push) Failing after 58m43s
424 lines
16 KiB
Python
424 lines
16 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.core.config import Settings
|
|
from app.core.errors import AppError
|
|
from app.main import app
|
|
from app.schemas.assistant import AssistantContextMetric, AssistantModelRead, AssistantQueryRequest, AssistantStatus, AssistantTemporalSeries
|
|
from app.services.geo_assistant_service import GeoAssistantService
|
|
from app.services.temporal_analysis_service import TemporalAnalysisService
|
|
from tests.frontend_contract import read_feature
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
|
|
|
|
def ollama_settings() -> Settings:
|
|
return Settings(
|
|
_env_file=None,
|
|
ollama_enabled=True,
|
|
ollama_base_url="http://ollama.internal:11434/",
|
|
ollama_default_model="qwen3.5:9b",
|
|
)
|
|
|
|
|
|
def test_ollama_model_catalog_reports_only_installed_models(monkeypatch) -> None:
|
|
service = GeoAssistantService(ollama_settings())
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_request_json",
|
|
lambda path, payload=None: {
|
|
"models": [
|
|
{
|
|
"name": "qwen3.5:9b",
|
|
"size": 123,
|
|
"details": {"parameter_size": "9.7B", "quantization_level": "Q4_K_M"},
|
|
"capabilities": ["completion", "tools"],
|
|
}
|
|
]
|
|
},
|
|
)
|
|
|
|
models = service.list_models()
|
|
|
|
assert [model.name for model in models] == ["qwen3.5:9b"]
|
|
assert models[0].parameter_size == "9.7B"
|
|
assert service.settings.ollama_base_url == "http://ollama.internal:11434"
|
|
|
|
|
|
def test_assistant_status_endpoint_uses_canonical_envelope(monkeypatch) -> None:
|
|
monkeypatch.setattr(
|
|
GeoAssistantService,
|
|
"status",
|
|
lambda self: AssistantStatus(
|
|
enabled=True,
|
|
reachable=True,
|
|
status="configured",
|
|
base_url="http://ollama.internal:11434",
|
|
default_model="qwen3.5:9b",
|
|
model_count=3,
|
|
limitation_message="Local only",
|
|
),
|
|
)
|
|
|
|
response = TestClient(app).get("/api/v1/assistant/status")
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["data"]["status"] == "configured"
|
|
assert response.json()["data"]["model_count"] == 3
|
|
|
|
|
|
def test_geo_assistant_rejects_model_that_is_not_installed(monkeypatch) -> None:
|
|
service = GeoAssistantService(ollama_settings())
|
|
monkeypatch.setattr(service, "list_models", lambda: [AssistantModelRead(name="qwen3.5:9b")])
|
|
|
|
with pytest.raises(AppError) as exc_info:
|
|
service.query(
|
|
object(),
|
|
project_id=uuid4(),
|
|
payload=AssistantQueryRequest(question="Hoeveel bos is er?", model="missing:latest"),
|
|
)
|
|
|
|
assert exc_info.value.code == "OLLAMA_MODEL_UNAVAILABLE"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"question",
|
|
[
|
|
"Hoe evolueerden bevolking en bosoppervlakte?",
|
|
"Toon de historische ontwikkeling van water.",
|
|
"Welke trend zien we sinds 2013?",
|
|
],
|
|
)
|
|
def test_geo_assistant_recognizes_dutch_historical_questions(question: str) -> None:
|
|
assert GeoAssistantService.history_requested(question) is True
|
|
|
|
|
|
def test_geo_assistant_limits_explicit_cross_domain_question_to_requested_themes() -> None:
|
|
themes = GeoAssistantService.requested_themes(
|
|
"Geef een profiel met ruimtebeslag, open ruimte, bevolking, bereikbaarheid, voorzieningen en bodem."
|
|
)
|
|
|
|
assert themes == {"space_occupation", "open_space", "population", "accessibility", "services", "soil"}
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("question", "expected"),
|
|
[
|
|
("Hoe evolueerden bevolking en bosoppervlakte?", {"population", "forest"}),
|
|
("Toon wegen, waterlopen en overstromingen.", {"roads", "water", "flood_hazard"}),
|
|
("Welke bodemtypes en landbouwteelten komen voor?", {"soil", "agriculture"}),
|
|
("Geef bodemdetails en perceeloppervlaktes voor Mol.", {"soil", "parcels"}),
|
|
("Vergelijk bevolkingsontwikkeling en voorzieningenniveau.", {"population", "services"}),
|
|
("Vat de belangrijkste gebiedsmetingen samen.", None),
|
|
("Welke officiële bronnen zijn beschikbaar?", None),
|
|
],
|
|
)
|
|
def test_geo_assistant_theme_selection_preserves_general_overviews(question: str, expected: set[str] | None) -> None:
|
|
assert GeoAssistantService.requested_themes(question) == expected
|
|
|
|
|
|
def test_geo_assistant_discloses_estimated_population_values() -> None:
|
|
metrics = [
|
|
AssistantContextMetric(
|
|
theme="population",
|
|
label="Geschatte bevolking",
|
|
value=74_254,
|
|
unit="personen",
|
|
source="Statbel",
|
|
dataset_id=uuid4(),
|
|
is_estimate=True,
|
|
)
|
|
]
|
|
|
|
answer = GeoAssistantService.ensure_estimate_disclosure(
|
|
"De bevolking bedraagt 74.254 personen.",
|
|
metrics,
|
|
)
|
|
|
|
assert answer.startswith(
|
|
"Datakwaliteit: bevolkingswaarden in dit antwoord zijn schattingen volgens de bronmetadata, "
|
|
"geen exacte tellingen."
|
|
)
|
|
|
|
|
|
def test_geo_assistant_never_labels_area_weighted_population_as_official_count() -> None:
|
|
metrics = [
|
|
AssistantContextMetric(
|
|
theme="population",
|
|
label="Geraamd aantal inwoners",
|
|
value=38_675,
|
|
unit="inwoners",
|
|
source="Statbel",
|
|
dataset_id=uuid4(),
|
|
is_estimate=True,
|
|
)
|
|
]
|
|
|
|
answer = GeoAssistantService.ensure_estimate_disclosure(
|
|
"De officiële telling uit januari 2025 bedraagt 38.675 inwoners. Deze waarde is een schatting.",
|
|
metrics,
|
|
)
|
|
|
|
assert "officiële telling" not in answer.casefold()
|
|
assert answer.startswith("De uit de officiële bron afgeleide schatting")
|
|
|
|
|
|
def test_geo_assistant_does_not_add_irrelevant_estimate_disclosure() -> None:
|
|
metrics = [
|
|
AssistantContextMetric(
|
|
theme="population",
|
|
label="Geschatte bevolking",
|
|
value=74_254,
|
|
unit="personen",
|
|
source="Statbel",
|
|
dataset_id=uuid4(),
|
|
is_estimate=True,
|
|
)
|
|
]
|
|
|
|
answer = GeoAssistantService.ensure_estimate_disclosure(
|
|
"De bosoppervlakte bedraagt 3.626,56 hectare.",
|
|
metrics,
|
|
)
|
|
|
|
assert answer == "De bosoppervlakte bedraagt 3.626,56 hectare."
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("value", "unit", "expected"),
|
|
[
|
|
(36_782.6497, "inwoners", 36_783),
|
|
(3_638.4167, "ha", 3_638.42),
|
|
(31.76431, "%", 31.76),
|
|
(0.680612, "score", 0.6806),
|
|
],
|
|
)
|
|
def test_geo_assistant_rounds_prompt_values_by_semantic_unit(value: float, unit: str, expected: int | float) -> None:
|
|
assert GeoAssistantService.rounded_context_value(value, unit) == expected
|
|
|
|
|
|
def test_geo_assistant_omits_supporting_object_count_from_richer_model_context() -> None:
|
|
metrics = [
|
|
{"metric_label": "Bodemkaartoppervlakte", "metric_value": 11_448.35, "metric_unit": "ha"},
|
|
{"metric_label": "Bodemkaartvlakken", "metric_value": 1_159, "metric_unit": "objecten"},
|
|
]
|
|
|
|
assert GeoAssistantService.model_context_metrics(metrics) == metrics[:1]
|
|
|
|
|
|
def test_geo_assistant_keeps_object_count_when_it_is_the_only_metric() -> None:
|
|
metrics = [{"metric_label": "Objecten", "metric_value": 12, "metric_unit": "objecten"}]
|
|
|
|
assert GeoAssistantService.model_context_metrics(metrics) == metrics
|
|
|
|
|
|
def test_geo_assistant_normalizes_model_markdown_for_plain_text_renderer() -> None:
|
|
answer = GeoAssistantService.normalize_plain_text("**Bevolking**\n* 36.783 inwoners\n`Bron: Statbel`")
|
|
|
|
assert answer == "Bevolking\n- 36.783 inwoners\nBron: Statbel"
|
|
|
|
|
|
def test_geo_assistant_sends_grounded_context_without_thinking_trace(monkeypatch) -> None:
|
|
service = GeoAssistantService(ollama_settings())
|
|
project_id = uuid4()
|
|
dataset_id = uuid4()
|
|
captured: dict = {}
|
|
monkeypatch.setattr(service, "list_models", lambda: [AssistantModelRead(name="qwen3.5:9b")])
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_build_context",
|
|
lambda *args, **kwargs: (
|
|
{
|
|
"scope": {"label": "Gemeente Mol"},
|
|
"current_measurements": [{"label": "Bosoppervlakte", "value": 3626.56, "unit": "ha"}],
|
|
"rules": {"water_volume_available": False},
|
|
},
|
|
[
|
|
AssistantContextMetric(
|
|
theme="forest",
|
|
label="Bosoppervlakte",
|
|
value=3626.56,
|
|
unit="ha",
|
|
source="Departement Omgeving",
|
|
dataset_id=dataset_id,
|
|
)
|
|
],
|
|
[
|
|
AssistantTemporalSeries(
|
|
temporal_series_key="forest:mol",
|
|
label="Bos 2013-2025",
|
|
source="Departement Omgeving",
|
|
first_year=2013,
|
|
last_year=2025,
|
|
observation_count=5,
|
|
)
|
|
],
|
|
[dataset_id],
|
|
[],
|
|
"Gemeente Mol",
|
|
),
|
|
)
|
|
|
|
def fake_request(path, payload=None):
|
|
captured.update({"path": path, "payload": payload})
|
|
return {"message": {"role": "assistant", "content": "Mol telt 3.626,56 ha bos volgens Departement Omgeving."}}
|
|
|
|
monkeypatch.setattr(service, "_request_json", fake_request)
|
|
result = service.query(
|
|
object(),
|
|
project_id=project_id,
|
|
payload=AssistantQueryRequest(question="Hoeveel bos is er in Mol?"),
|
|
)
|
|
|
|
assert result.model == "qwen3.5:9b"
|
|
assert result.context_metrics[0].value == 3626.56
|
|
assert captured["path"] == "/api/chat"
|
|
assert captured["payload"]["stream"] is False
|
|
assert captured["payload"]["think"] is False
|
|
assert captured["payload"]["options"]["temperature"] == 0.0
|
|
assert captured["payload"]["options"]["num_ctx"] == 16_384
|
|
assert captured["payload"]["options"]["num_predict"] == 1_200
|
|
assert "Gebruik uitsluitend feiten en cijfers uit CONTEXT_JSON" in captured["payload"]["messages"][0]["content"]
|
|
assert "scope.label is het exact geanalyseerde gebied" in captured["payload"]["messages"][0]["content"]
|
|
assert "noem de waarde verplicht een schatting" in captured["payload"]["messages"][0]["content"]
|
|
assert "noem een schatting nooit officieel geteld" in captured["payload"]["messages"][0]["content"]
|
|
assert "contextwaarden zijn al bronveilig afgerond" in captured["payload"]["messages"][0]["content"]
|
|
assert "uitsluitend het jaar, de bron en de meetkwaliteit van dezelfde dataset" in captured["payload"]["messages"][0]["content"]
|
|
assert "voeg bron, jaar of kwaliteit nooit samen" in captured["payload"]["messages"][0]["content"]
|
|
assert "verzin geen oorzaak, voorspelling, verzadiging" in captured["payload"]["messages"][0]["content"]
|
|
assert "bereken zelf geen gemiddelde, tempo, oorzaak of afgeleide trend" in captured["payload"]["messages"][0]["content"]
|
|
assert "zonder Markdown-symbolen" in captured["payload"]["messages"][0]["content"]
|
|
assert "behandel elk gevraagd thema en voeg geen ongevraagd thema toe" in captured["payload"]["messages"][0]["content"]
|
|
assert "Houd het antwoord beknopt" in captured["payload"]["messages"][0]["content"]
|
|
assert "water_volume_available" in captured["payload"]["messages"][0]["content"]
|
|
|
|
|
|
def test_geo_assistant_rejects_truncated_ollama_answer(monkeypatch) -> None:
|
|
service = GeoAssistantService(ollama_settings())
|
|
monkeypatch.setattr(service, "list_models", lambda: [AssistantModelRead(name="qwen3.5:9b")])
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_build_context",
|
|
lambda *args, **kwargs: (
|
|
{"scope": {"label": "Gemeente Mol"}},
|
|
[],
|
|
[],
|
|
[],
|
|
[],
|
|
"Gemeente Mol",
|
|
),
|
|
)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_request_json",
|
|
lambda path, payload=None: {
|
|
"done": True,
|
|
"done_reason": "length",
|
|
"message": {"role": "assistant", "content": "Een onvolledige zin"},
|
|
},
|
|
)
|
|
|
|
with pytest.raises(AppError) as exc_info:
|
|
service.query(
|
|
object(),
|
|
project_id=uuid4(),
|
|
payload=AssistantQueryRequest(question="Hoe evolueerde Mol?"),
|
|
)
|
|
|
|
assert exc_info.value.code == "OLLAMA_RESPONSE_TRUNCATED"
|
|
|
|
|
|
def test_unraid_ollama_context_window_is_configurable() -> None:
|
|
compose = (ROOT / "docker-compose.unraid.yml").read_text(encoding="utf-8")
|
|
template = (ROOT / "deploy/unraid/geointel-unraid-template.xml").read_text(encoding="utf-8")
|
|
env_example = (ROOT / "deploy/unraid/geointel.env.example").read_text(encoding="utf-8")
|
|
|
|
assert "OLLAMA_CONTEXT_TOKENS: ${OLLAMA_CONTEXT_TOKENS:-16384}" in compose
|
|
assert 'Target="OLLAMA_CONTEXT_TOKENS"' in template
|
|
assert "OLLAMA_CONTEXT_TOKENS=16384" in env_example
|
|
assert "OLLAMA_MAX_OUTPUT_TOKENS: ${OLLAMA_MAX_OUTPUT_TOKENS:-1200}" in compose
|
|
assert 'Target="OLLAMA_MAX_OUTPUT_TOKENS"' in template
|
|
assert "OLLAMA_MAX_OUTPUT_TOKENS=1200" in env_example
|
|
|
|
|
|
def test_temporal_comparison_preserves_all_compatible_semantic_metrics() -> None:
|
|
earlier = {
|
|
"metrics": [
|
|
{
|
|
"metric_key": "water_area_ha",
|
|
"metric_label": "Wateroppervlakte",
|
|
"metric_value": 110.0,
|
|
"metric_unit": "ha",
|
|
"aggregation_method": "clipped_area_ha",
|
|
"is_estimate": False,
|
|
},
|
|
{
|
|
"metric_key": "water_length_km",
|
|
"metric_label": "Lengte waterlopen",
|
|
"metric_value": 42.5,
|
|
"metric_unit": "km",
|
|
"aggregation_method": "clipped_length_km",
|
|
"is_estimate": False,
|
|
},
|
|
]
|
|
}
|
|
later = {
|
|
"metrics": [
|
|
{
|
|
"metric_key": "water_area_ha",
|
|
"metric_label": "Wateroppervlakte",
|
|
"metric_value": 121.0,
|
|
"metric_unit": "ha",
|
|
"aggregation_method": "clipped_area_ha",
|
|
"is_estimate": False,
|
|
},
|
|
{
|
|
"metric_key": "water_length_km",
|
|
"metric_label": "Lengte waterlopen",
|
|
"metric_value": 40.0,
|
|
"metric_unit": "km",
|
|
"aggregation_method": "clipped_length_km",
|
|
"is_estimate": False,
|
|
},
|
|
]
|
|
}
|
|
|
|
result = TemporalAnalysisService._compare_summary_metrics(earlier, later)
|
|
|
|
assert [metric.metric_key for metric in result] == ["water_area_ha", "water_length_km"]
|
|
assert result[0].absolute_change == 11.0
|
|
assert result[0].percent_change == 10.0
|
|
assert result[1].absolute_change == -2.5
|
|
|
|
|
|
def test_landuse_operator_exposes_more_honest_historical_themes() -> None:
|
|
operator = (ROOT / "scripts/provision_official_landuse_timeseries.py").read_text(encoding="utf-8")
|
|
regional = (ROOT / "scripts/provision_regional_timeseries.py").read_text(encoding="utf-8")
|
|
|
|
assert 'ThemeDefinition("water", "Water", (17,)' in operator
|
|
assert '"Bebouwde functies"' in operator
|
|
assert '"Transportinfrastructuur"' in operator
|
|
assert '"forest,water,built,transport"' in regional
|
|
assert "legacy_forest_raster" in operator
|
|
|
|
|
|
def test_frontend_exposes_source_inventory_timeline_and_ai_window() -> None:
|
|
app = read_feature("shell")
|
|
workspace = read_feature("map_workspace")
|
|
catalog = read_feature("datasets")
|
|
assistant_hook = (ROOT / "frontend/src/hooks/useGeoAssistant.ts").read_text(encoding="utf-8")
|
|
|
|
assert "SourceCatalogPanel" in app
|
|
assert "TemporalTrendChart" in workspace
|
|
assert "nextAssistantMessageId" in assistant_hook
|
|
assert "crypto.randomUUID" not in assistant_hook
|
|
assert "Officiële bronnen die hierna kunnen worden ingeladen" in catalog
|
|
assert "vergelijkbare meetmomenten" in catalog
|
|
assert "andere bronmethode" in catalog
|