231 lines
8.0 KiB
Python
231 lines
8.0 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
|
|
|
|
|
|
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_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 "Gebruik uitsluitend feiten en cijfers uit CONTEXT_JSON" in captured["payload"]["messages"][0]["content"]
|
|
assert "water_volume_available" in captured["payload"]["messages"][0]["content"]
|
|
|
|
|
|
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 = (ROOT / "frontend/src/App.tsx").read_text(encoding="utf-8")
|
|
workspace = (ROOT / "frontend/src/components/map/MapWorkspace.tsx").read_text(encoding="utf-8")
|
|
catalog = (ROOT / "frontend/src/components/datasets/SourceCatalogPanel.tsx").read_text(encoding="utf-8")
|
|
|
|
assert "SourceCatalogPanel" in app
|
|
assert "TemporalTrendChart" in workspace
|
|
assert "Officiële bronnen die hierna kunnen worden ingeladen" in catalog
|