114 lines
3.9 KiB
Python
114 lines
3.9 KiB
Python
from __future__ import annotations
|
|
|
|
from threading import Thread
|
|
|
|
import pytest
|
|
|
|
import dpg_map as dpgm
|
|
from dpg_map.commands import CommandKind
|
|
from dpg_map.exceptions import CoordinateError
|
|
from dpg_map.overlays import TrajectoryOverlay
|
|
from dpg_map.state import DirtyFlags, create_map_state, get_map_state
|
|
|
|
|
|
def test_overlay_update_does_not_alter_center_or_zoom() -> None:
|
|
create_map_state(tag="overlay-isolation", center=(47.0, 2.0), zoom=8)
|
|
dpgm.add_marker("vehicle", lat=47.1, lon=2.1, map_tag="overlay-isolation")
|
|
|
|
before_center = dpgm.get_center(map_tag="overlay-isolation")
|
|
before_zoom = dpgm.get_zoom(map_tag="overlay-isolation")
|
|
|
|
dpgm.update_marker("vehicle", lat=47.2, lon=2.2, map_tag="overlay-isolation")
|
|
|
|
assert dpgm.get_center(map_tag="overlay-isolation") == before_center
|
|
assert dpgm.get_zoom(map_tag="overlay-isolation") == before_zoom
|
|
state = get_map_state("overlay-isolation")
|
|
assert state.dirty & DirtyFlags.OVERLAYS
|
|
|
|
|
|
def test_trajectory_inputs_are_copied() -> None:
|
|
create_map_state(tag="trajectory-copy")
|
|
lats = [1.0, 2.0]
|
|
lons = [3.0, 4.0]
|
|
|
|
dpgm.add_trajectory("track", lats=lats, lons=lons, map_tag="trajectory-copy")
|
|
lats[0] = 99.0
|
|
lons[0] = 99.0
|
|
|
|
state = get_map_state("trajectory-copy")
|
|
overlay = state.overlays["track"]
|
|
|
|
assert isinstance(overlay, TrajectoryOverlay)
|
|
assert overlay.points == ((1.0, 3.0), (2.0, 4.0))
|
|
|
|
|
|
def test_mismatched_lat_lon_lengths_raise() -> None:
|
|
create_map_state(tag="bad-coordinates")
|
|
|
|
with pytest.raises(CoordinateError):
|
|
dpgm.add_trajectory("track", lats=[1.0], lons=[2.0, 3.0], map_tag="bad-coordinates")
|
|
|
|
|
|
def test_layer_state_tracks_visibility_and_overlay_membership() -> None:
|
|
create_map_state(tag="layers")
|
|
|
|
dpgm.add_layer("fleet", map_tag="layers")
|
|
dpgm.add_marker("vehicle", lat=1.0, lon=2.0, layer="fleet", map_tag="layers")
|
|
dpgm.hide_layer("fleet", map_tag="layers")
|
|
|
|
state = get_map_state("layers")
|
|
assert state.layers["fleet"].show is False
|
|
assert state.layers["fleet"].overlay_tags == {"vehicle"}
|
|
|
|
dpgm.clear_layer("fleet", map_tag="layers")
|
|
|
|
assert state.layers["fleet"].overlay_tags == set()
|
|
assert "vehicle" not in state.overlays
|
|
|
|
|
|
def test_add_layer_can_update_visibility_and_z_index() -> None:
|
|
create_map_state(tag="layer-order")
|
|
|
|
dpgm.add_layer("fleet", z_index=25, show=False, map_tag="layer-order")
|
|
dpgm.add_layer("fleet", z_index=30, show=True, map_tag="layer-order")
|
|
|
|
state = get_map_state("layer-order")
|
|
assert state.layers["fleet"].show is True
|
|
assert state.layers["fleet"].z_index == 30
|
|
|
|
|
|
def test_threaded_marker_updates_coalesce_without_touching_view_or_drag_state() -> None:
|
|
create_map_state(tag="threaded-marker", center=(47.0, 2.0), zoom=9)
|
|
dpgm.add_marker("vehicle", lat=47.0, lon=2.0, map_tag="threaded-marker")
|
|
state = get_map_state("threaded-marker")
|
|
state.command_queue.drain()
|
|
state.dirty = DirtyFlags.NONE
|
|
state.interaction.active_drag = True
|
|
state.interaction.last_mouse_position = (100.0, 100.0)
|
|
before_center = state.center
|
|
before_zoom = state.zoom
|
|
|
|
def update_worker(offset: float) -> None:
|
|
for index in range(100):
|
|
dpgm.update_marker(
|
|
"vehicle",
|
|
lat=47.0 + offset,
|
|
lon=2.0 + index * 0.00001,
|
|
map_tag="threaded-marker",
|
|
)
|
|
|
|
threads = [Thread(target=update_worker, args=(worker * 0.0001,)) for worker in range(4)]
|
|
for thread in threads:
|
|
thread.start()
|
|
for thread in threads:
|
|
thread.join()
|
|
|
|
commands = state.command_queue.drain()
|
|
|
|
assert state.center == before_center
|
|
assert state.zoom == before_zoom
|
|
assert state.interaction.active_drag is True
|
|
assert state.interaction.last_mouse_position == (100.0, 100.0)
|
|
assert state.dirty == DirtyFlags.OVERLAYS
|
|
assert [command.kind for command in commands] == [CommandKind.UPDATE_OVERLAY]
|