fix: make burst impact coverage envelopes exclusive
Generic Container CI/CD / test-build-publish (push) Successful in 4m21s
Server CI/CD v2 / build-test-publish-and-deploy (push) Successful in 4m21s

This commit is contained in:
2026-11-15 16:06:41 +08:00
parent 4b47ae00d1
commit 105c1ca8ab
5 changed files with 100 additions and 8 deletions
+41 -2
View File
@@ -3,11 +3,12 @@ from datetime import datetime, timezone
import pytest
from fastapi.encoders import jsonable_encoder
from pydantic import ValidationError
from shapely.geometry import Point, shape
from shapely.geometry import Point, box, mapping, shape
from shapely.ops import unary_union
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.node_coverage import build_node_coverage, exclusive_coverage_features
from app.services.operational_presentation import resolve_overlay
@@ -32,6 +33,44 @@ def test_coverage_can_compare_arbitrary_groups_without_assigning_isolation_seman
assert coverage["parameters"]["split_by_connectivity"] is False
def test_exclusive_envelopes_keep_union_and_counts_with_higher_groups_taking_precedence():
originals = [box(0, 0, 4, 4), box(2, 0, 6, 4), box(3, 1, 5, 3)]
features = [{"type": "Feature", "id": str(i), "properties": {
"coverage_group": str(i), "node_count": 10 + i, "mapped_node_count": 10 + i,
}, "geometry": mapping(geometry)} for i, geometry in enumerate(originals)]
result = exclusive_coverage_features(features)
visible = [shape(feature["geometry"]) for feature in result]
assert len(result) == 3
assert all(a.intersection(b).area == 0 for i, a in enumerate(visible) for b in visible[i + 1:])
assert unary_union(visible).equals(unary_union(originals))
assert visible[-1].equals(originals[-1])
assert [feature["properties"]["node_count"] for feature in result] == [10, 11, 12]
assert [feature["properties"]["mapped_node_count"] for feature in result] == [10, 11, 12]
assert all(feature["properties"]["overlap_policy"] == "higher_priority_group_first" for feature in result)
assert shape(features[0]["geometry"]).equals(originals[0]) # Never rewrite saved inputs.
def test_covered_grade_keeps_node_metadata_without_invalid_empty_geometry():
positions = [position("A", 117.24), position("B", 117.24)]
groups = [group(["A"], "low"), group(["B"], "high")]
ordinary = build_node_coverage(groups, positions, [])
assert len(ordinary["map_payload"]["features"]) == 2
exclusive = build_node_coverage(groups, positions, [], exclusive_groups=True)
assert [feature["id"] for feature in exclusive["map_payload"]["features"]] == ["high"]
assert exclusive["groups"]["low"]["node_ids"] == ["A"]
assert exclusive["groups"]["low"]["mapped_node_count"] == 1
assert exclusive["groups"]["low"]["display_area_available"] is False
assert exclusive["groups"]["high"]["display_area_available"] is True
summaries = exclusive["map_payload"]["features"][0]["properties"]["coverage_group_summary"]
assert summaries["low"]["node_count"] == 1
assert summaries["low"]["mapped_node_count"] == 1
assert summaries["low"]["display_area_available"] is False
# A second application must not drop the summary for the clipped group.
repeated = exclusive_coverage_features(exclusive["map_payload"]["features"])
assert repeated[0]["properties"]["coverage_group_summary"] == summaries
assert exclusive["status"] == "ready"
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)]