119 lines
6.2 KiB
Python
119 lines
6.2 KiB
Python
from datetime import datetime, timezone
|
|
|
|
import pytest
|
|
from fastapi.encoders import jsonable_encoder
|
|
from pydantic import ValidationError
|
|
from shapely.geometry import Point, shape
|
|
|
|
from app.domain.schemas.operational_data import TaskDataSnapshot
|
|
from app.mcp.analysis_models import NodeCoverageInput
|
|
from app.services.node_coverage import build_node_coverage
|
|
from app.services.operational_presentation import resolve_overlay
|
|
|
|
|
|
def position(node, longitude, latitude=32.75):
|
|
return {"id": node, "longitude": longitude, "latitude": latitude}
|
|
|
|
|
|
def group(ids, group_id="selected"):
|
|
return {"group_id": group_id, "title": "节点覆盖比较", "node_ids": ids}
|
|
|
|
|
|
def link(link_id, start, end, status=None):
|
|
return {"id": link_id, "start_node_id": start, "end_node_id": end, "status": status}
|
|
|
|
|
|
def test_coverage_can_compare_arbitrary_groups_without_assigning_isolation_semantics():
|
|
positions = [position("A", 117.24), position("B", 117.241), position("C", 117.26)]
|
|
coverage = build_node_coverage([group(["A", "B"], "monitoring"), group(["C"], "candidate")], positions, [])
|
|
assert coverage["status"] == "ready"
|
|
assert set(coverage["groups"]) == {"monitoring", "candidate"}
|
|
assert [f["properties"]["coverage_group"] for f in coverage["map_payload"]["features"]] == ["monitoring", "candidate"]
|
|
assert coverage["parameters"]["split_by_connectivity"] is False
|
|
|
|
|
|
def test_connectivity_option_does_not_bridge_disconnected_groups():
|
|
positions = [position("A", 117.24), position("B", 117.241),
|
|
position("C", 117.26), position("D", 117.261)]
|
|
coverage = build_node_coverage([group(["A", "B", "C", "D"])], positions,
|
|
[link("P1", "A", "B"), link("P2", "C", "D")],
|
|
split_by_connectivity=True)
|
|
feature = coverage["map_payload"]["features"][0]
|
|
geometry = shape(feature["geometry"])
|
|
assert geometry.geom_type == "MultiPolygon" and geometry.is_valid
|
|
assert feature["properties"]["component_count"] == 2
|
|
assert all(geometry.covers(Point(row["longitude"], row["latitude"])) for row in positions)
|
|
assert not geometry.covers(Point(117.25, 32.75))
|
|
|
|
|
|
@pytest.mark.parametrize("status,excluded", [("CLOSED", []), (None, ["V"])])
|
|
def test_connectivity_excludes_initially_closed_and_explicitly_excluded_links(status, excluded):
|
|
coverage = build_node_coverage([group(["A", "B"])], [position("A", 117.24), position("B", 117.26)],
|
|
[link("V", "A", "B", status)], split_by_connectivity=True,
|
|
excluded_link_ids=excluded)
|
|
assert coverage["groups"]["selected"]["component_count"] == 2
|
|
|
|
|
|
def test_partial_geometries_are_reported_and_single_point_buffer_uses_meters():
|
|
coverage = build_node_coverage([group(["A", "A", "missing", "invalid"])],
|
|
[position("A", 117.24), position("invalid", float("nan"))], [])
|
|
assert coverage["status"] == "partial"
|
|
metadata = coverage["groups"]["selected"]
|
|
assert metadata["mapped_node_count"] == 1 and metadata["missing_node_ids"] == ["invalid", "missing"]
|
|
feature = coverage["map_payload"]["features"][0]
|
|
geometry = shape(feature["geometry"])
|
|
assert geometry.covers(Point(117.24, 32.75))
|
|
assert 0.0004 < geometry.bounds[2] - geometry.bounds[0] < 0.001
|
|
assert feature["properties"]["display_only"] is True
|
|
|
|
|
|
def test_duplicate_and_collinear_positions_are_valid_and_parameters_are_preserved():
|
|
coverage = build_node_coverage([group(["A", "B", "C"])],
|
|
[position("A", 117.24), position("B", 117.24), position("C", 117.241)], [],
|
|
concave_ratio=0.5, buffer_m=40)
|
|
assert coverage["status"] == "ready"
|
|
assert coverage["parameters"]["concave_ratio"] == 0.5 and coverage["parameters"]["buffer_m"] == 40
|
|
assert shape(coverage["map_payload"]["features"][0]["geometry"]).is_valid
|
|
|
|
|
|
def test_missing_positions_do_not_fabricate_an_area():
|
|
coverage = build_node_coverage([group(["missing"])], [], [])
|
|
assert coverage["status"] == "partial" and coverage["map_payload"]["features"] == []
|
|
|
|
|
|
@pytest.mark.parametrize("mcp_envelope", [False, True])
|
|
def test_derived_payload_round_trips_through_saved_evidence_overlay(mcp_envelope):
|
|
coverage = build_node_coverage([group(["A"])], [position("A", 117.24)], [])
|
|
data = jsonable_encoder({"result": {"map_payload": coverage["map_payload"]}})
|
|
if mcp_envelope:
|
|
data = {"source": "generate_node_coverage", "source_ref": "mcp:coverage", "data": data}
|
|
snapshot = TaskDataSnapshot(source="generate_node_coverage", source_ref="mcp:coverage",
|
|
queried_at=datetime.now(timezone.utc), origin="system_data",
|
|
truncated=False, data=data)
|
|
overlay = resolve_overlay(snapshot, "geometry")
|
|
assert overlay["features"][0]["properties"]["coverage_group"] == "selected"
|
|
assert shape(overlay["features"][0]["geometry"]).covers(Point(117.24, 32.75))
|
|
|
|
|
|
def test_service_resolves_only_selected_project_nodes_and_reads_topology_only_when_requested(monkeypatch):
|
|
from app.services import node_coverage
|
|
def positions(network, ids):
|
|
assert network == "bound-project" and ids == ["A", "B"]
|
|
return [position("A", 117.24), position("B", 117.241)]
|
|
monkeypatch.setattr(node_coverage, "get_node_geographic_positions", positions)
|
|
monkeypatch.setattr(node_coverage, "get_node_coverage_links", lambda *args: pytest.fail("unnecessary topology query"))
|
|
result = node_coverage.generate_node_coverage("bound-project", [group(["A", "B"])])
|
|
assert result["coverage"]["status"] == "ready"
|
|
assert len(result["map_payload"]["features"]) == 1
|
|
|
|
|
|
@pytest.mark.parametrize("extra", [
|
|
{"groups": []}, {"groups": [group(["A"]), group(["B"])]},
|
|
{"buffer_m": 0}, {"buffer_m": 1001}, {"concave_ratio": float("nan")},
|
|
{"excluded_link_ids": ["V"]}, {"groups": [{**group(["A"]), "coordinates": [117, 32]}]},
|
|
{"groups": [group([str(i) for i in range(10000)]), group(["extra"], "other")]},
|
|
])
|
|
def test_tool_rejects_invalid_parameters_and_unbounded_or_arbitrary_coordinate_inputs(extra):
|
|
with pytest.raises(ValidationError):
|
|
NodeCoverageInput.model_validate({"context_handle": "x" * 43, "groups": [group(["A"])], **extra})
|