jiang
be6317eb48
Add project-bound OAuth MCP queries and controlled analysis with scoped evidence, limits, timeouts and a nearby-model spatial query using EPSG:4326 inputs. Default nearby searches to 200 meters and five results. Integrate investigation messaging/presentation, health/readiness and typed history; remove duplicate web search and fix online simulation and compressed baseline writes. Include project migrations 023-025; keep local credentials and demo seeding excluded. Validation: 319 focused pytest checks; real PostGIS temporary-table regression; read-only Jinqian bridge radius comparison; cross-language MCP discovery/calls.
57 lines
2.8 KiB
Python
57 lines
2.8 KiB
Python
from datetime import datetime, timezone
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
from pydantic import ValidationError
|
|
|
|
from app.domain.domain_errors import DomainError
|
|
from app.domain.schemas.operational_data import TaskDataSnapshot
|
|
from app.domain.schemas.operational_task import TaskPresentationEventCommand
|
|
from app.services.operational_presentation import resolve_overlay
|
|
|
|
|
|
def _snapshot(data):
|
|
return TaskDataSnapshot(source="workbench_runtime", source_ref="stored-spatial-evidence",
|
|
queried_at=datetime(2026, 10, 10, tzinfo=timezone.utc),
|
|
origin="system_data", data=data, truncated=False)
|
|
|
|
|
|
def test_overlay_preserves_recorded_polygon_and_never_converts_it_to_heatmap_points():
|
|
geometry = {"type": "Polygon", "coordinates": [[[121, 31], [122, 31], [122, 32], [121, 31]]]}
|
|
collection = {"type": "FeatureCollection", "features": [{"type": "Feature", "geometry": geometry,
|
|
"properties": {"source": "actual-analysis"}}]}
|
|
snapshot = _snapshot({"latest_analysis": {"map_payload": collection}})
|
|
assert resolve_overlay(snapshot, "geometry")["features"][0]["geometry"] == geometry
|
|
with pytest.raises(DomainError) as error:
|
|
resolve_overlay(snapshot, "heatmap")
|
|
assert error.value.code == "OVERLAY_HEATMAP_POINTS_REQUIRED"
|
|
|
|
|
|
@pytest.mark.parametrize("data", [
|
|
{"items": [{"location": "某小区，无法仅凭地址确定坐标"}]},
|
|
{"items": [{"longitude": True, "latitude": 31}]},
|
|
{"items": [{"longitude": 121, "latitude": 91}]},
|
|
{"items": [{"longitude": 121, "latitude": 31, "weight": -1}]},
|
|
{"type": "FeatureCollection", "features": []},
|
|
])
|
|
def test_overlay_rejects_missing_or_invalid_recorded_spatial_data(data):
|
|
with pytest.raises(DomainError):
|
|
resolve_overlay(_snapshot(data), "heatmap")
|
|
|
|
|
|
@pytest.mark.parametrize("command", [
|
|
{"action": "run_script", "code": "arbitrary"},
|
|
{"action": "highlight_features", "layer": "arbitrary-layer", "feature_ids": ["X"]},
|
|
{"action": "set_overlay", "overlay_id": "X", "title": "X", "mode": "geometry", "evidence_id": "E",
|
|
"data": {"type": "FeatureCollection", "features": []}},
|
|
{"action": "open_chart", "chart_id": "X", "evidence_refs": ["E"], "chart": {
|
|
"title": "X", "chart_type": "line", "x_data": ["t1", "t2"], "series": [{"name": "s", "data": [1]}],
|
|
}},
|
|
{"action": "open_chart", "chart_id": "X", "evidence_refs": ["E"], "chart": {
|
|
"title": "X", "chart_type": "line", "x_data": ["t1"], "series": [{"name": "s", "data": [True]}],
|
|
}},
|
|
])
|
|
def test_presentation_input_rejects_free_form_or_invalid_commands(command):
|
|
with pytest.raises(ValidationError):
|
|
TaskPresentationEventCommand(request_id=uuid4(), run_id=uuid4(), source_session_id="session", command=command)
|