Jean-Pierre MartineauandClaude Opus 5
4e7bcacc37
Profiling on the Lyra found vector-tile decode dominating everything else in the app by about two orders of magnitude: a median cached tile took 1236 ms of pure CPU, the largest 3.4 s, against 16 ms of SD read for a whole viewport. Decoded tiles are memoised for the session, so this is a first-pan cost -- but it runs on a GIL-bound worker thread, so it starves the render loop for those seconds rather than running beside it. Roughly 40% of that work was thrown away. A planet tile ships building, housenumber, poi and mountain_peak; _draw_tile looks up ten layer names and none of them is those. So decode_tile now takes the set of layers the caller actually wants, reads each layer's identity with a cheap header walk, and decodes the features only for the ones asked for. Unwanted layers are still returned, named and empty -- which matters, because the tile source decides whether a cache entry is real by whether the bytes parsed into layers, and a tile whose every layer we skip must not read as blank and get its perfectly good cache entry pruned and re-downloaded every session. That test moves off "has features" and onto "parsed into layers", which is sound because the other two ways a tile isn't real -- empty and corrupt -- decode to nothing and raise respectively. Only the decode narrows; whole tiles are still cached, so drawing more later costs a re-decode and never a re-download. Measured on the device across the 46 real cached tiles: 183 s -> 51 s, 3.6x, with the drawn layers byte-identical on every one. The median tile is 1259 -> 353 ms, the largest 3545 -> 759 ms. _Reader.skip stopped slicing out the bytes it steps over while it was here, which is most of what makes the header walk cheap. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
1038 lines
43 KiB
Python
1038 lines
43 KiB
Python
"""Tests for the node map: MVT decoding, projection, the braille canvas, and the tool.
|
|
|
|
The vector-tile decoder and geometry are exercised against a real cached tile fixture
|
|
(``tests/fixtures/tile_14_4843_5861.mvt``, central Montréal) so the parsing is verified end
|
|
to end without the network; the projection, canvas, and compositor are pure; the tool runs
|
|
against the :class:`MockDevice` simulator with the basemap disabled (no network in tests).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import io
|
|
import re
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from rich.console import Console
|
|
|
|
from meshterm.core.geo import BBox, Viewport, haversine_km, lonlat_to_world, world_to_lonlat
|
|
from meshterm.core.models import (
|
|
NODE_TYPE_REPEATER,
|
|
Observation,
|
|
utcnow,
|
|
)
|
|
from meshterm.core.mvt import GEOM_LINE, GEOM_POLYGON, decode_tile
|
|
from meshterm.tools.map import MapTool
|
|
from meshterm.ui.mapcanvas import MapCanvas, parse_hex
|
|
|
|
_FIXTURE = Path(__file__).parent / "fixtures" / "tile_14_4843_5861.mvt"
|
|
from tests.conftest import plain as _plain # THE strip-and-join screen reader
|
|
|
|
|
|
# -- MVT decoding -------------------------------------------------------------
|
|
|
|
|
|
def test_decode_tile_reads_expected_layers() -> None:
|
|
"""The real fixture decodes into the OpenMapTiles layers the map relies on."""
|
|
layers = {layer.name: layer for layer in decode_tile(_FIXTURE.read_bytes())}
|
|
for expected in ("transportation", "transportation_name", "water", "waterway", "place"):
|
|
assert expected in layers, expected
|
|
assert layers["transportation"].extent == 4096
|
|
assert layers["transportation"].features # roads were decoded
|
|
|
|
|
|
def test_decode_tile_resolves_names_and_geometry() -> None:
|
|
"""Street and place features carry resolved names and non-trivial geometry."""
|
|
layers = {layer.name: layer for layer in decode_tile(_FIXTURE.read_bytes())}
|
|
streets = [f.name for f in layers["transportation_name"].features if f.name]
|
|
assert any("Rue" in s or "Avenue" in s or "Av." in s for s in streets)
|
|
|
|
places = {f.name: f.get("class") for f in layers["place"].features if f.name}
|
|
assert "Montréal" in places and places["Montréal"] == "city"
|
|
|
|
# A road is a line with at least two points; water is a polygon.
|
|
assert any(
|
|
f.geom_type == GEOM_LINE and len(f.rings[0]) >= 2
|
|
for f in layers["transportation"].features
|
|
)
|
|
assert any(f.geom_type == GEOM_POLYGON for f in layers["water"].features)
|
|
|
|
|
|
def test_decode_tile_handles_empty_input() -> None:
|
|
"""An empty byte string decodes to no layers rather than raising."""
|
|
assert decode_tile(b"") == []
|
|
|
|
|
|
def test_decode_tile_filter_keeps_the_whole_tile_visible() -> None:
|
|
"""A narrowed decode still names every layer — it only skips their features.
|
|
|
|
The tile source decides whether a cache entry is real by whether it parsed into
|
|
layers, so narrowing must never make a tile look like it holds nothing.
|
|
"""
|
|
raw = _FIXTURE.read_bytes()
|
|
full = decode_tile(raw)
|
|
narrow = decode_tile(raw, layers={"water"})
|
|
|
|
assert [layer.name for layer in narrow] == [layer.name for layer in full]
|
|
assert all(layer.extent == 4096 for layer in narrow)
|
|
picked = {layer.name: layer for layer in narrow}
|
|
assert picked["water"].features # the one we asked for was decoded
|
|
assert not picked["building"].features # the rest were skipped, not dropped
|
|
|
|
|
|
def test_decode_tile_filter_is_invisible_to_the_renderer() -> None:
|
|
"""Narrowing to ``DRAWN_LAYERS`` draws exactly the map a full decode draws.
|
|
|
|
This is the anti-drift guard: if ``_draw_tile`` grows a lookup for a layer that
|
|
isn't in ``DRAWN_LAYERS``, that layer arrives featureless and the rendered output
|
|
diverges here.
|
|
"""
|
|
from meshterm.ui.map_render import DRAWN_LAYERS, MapMarker, render_map
|
|
|
|
raw = _FIXTURE.read_bytes()
|
|
vp = Viewport(45.5019, -73.5674, 14, 180, 120)
|
|
markers = [MapMarker("Yagi", 45.5019, -73.5674, is_repeater=True)]
|
|
|
|
def drawn(layers):
|
|
return render_map(vp, {(14, 4843, 5861): layers}, markers)
|
|
|
|
assert drawn(decode_tile(raw, layers=DRAWN_LAYERS)) == drawn(decode_tile(raw))
|
|
|
|
|
|
def test_decode_tile_filter_skips_the_layers_the_map_never_draws() -> None:
|
|
"""The layers left undecoded are the ones that made a tile expensive."""
|
|
from meshterm.ui.map_render import DRAWN_LAYERS
|
|
|
|
narrow = {layer.name: layer for layer in decode_tile(_FIXTURE.read_bytes(), layers=DRAWN_LAYERS)}
|
|
for skipped in ("building", "housenumber", "poi"):
|
|
assert skipped in narrow, f"{skipped} should still be named"
|
|
assert not narrow[skipped].features, f"{skipped} should not have been decoded"
|
|
|
|
|
|
# -- projection / viewport ----------------------------------------------------
|
|
|
|
|
|
def test_world_projection_round_trips() -> None:
|
|
"""lon/lat -> world pixels -> lon/lat recovers the original coordinate."""
|
|
for lat, lon, z in [(45.5, -73.6, 14), (0.0, 0.0, 3), (-33.9, 151.2, 12)]:
|
|
wx, wy = lonlat_to_world(lat, lon, z)
|
|
rlat, rlon = world_to_lonlat(wx, wy, z)
|
|
assert rlat == pytest.approx(lat, abs=1e-6)
|
|
assert rlon == pytest.approx(lon, abs=1e-6)
|
|
|
|
|
|
def test_haversine_known_distance() -> None:
|
|
"""One degree of longitude at the equator is ~111 km; identical points are 0."""
|
|
assert haversine_km(0.0, 0.0, 0.0, 1.0) == pytest.approx(111.19, abs=0.5)
|
|
assert haversine_km(45.0, -73.0, 45.0, -73.0) == 0.0
|
|
|
|
|
|
def test_viewport_center_projects_to_the_middle() -> None:
|
|
"""The viewport's centre lands at the middle dot; north is up."""
|
|
vp = Viewport(45.5, -73.6, 13, 200, 120)
|
|
cx, cy = vp.lonlat_to_dot(45.5, -73.6)
|
|
assert cx == pytest.approx(100, abs=0.5) and cy == pytest.approx(60, abs=0.5)
|
|
# A point due north projects above the centre (smaller y).
|
|
_, ny = vp.lonlat_to_dot(45.6, -73.6)
|
|
assert ny < cy
|
|
|
|
|
|
def test_viewport_fit_frames_all_points() -> None:
|
|
"""A fitted viewport is centred on the points and zoomed so they share a tile scale."""
|
|
pts = [(45.5019, -73.5674), (45.4768, -73.5990)]
|
|
vp = Viewport.fit(pts, 200, 120, max_zoom=14)
|
|
assert 8 <= vp.zoom <= 14
|
|
for lat, lon in pts:
|
|
x, y = vp.lonlat_to_dot(lat, lon)
|
|
assert 0 <= x < vp.dot_w and 0 <= y < vp.dot_h # every point is on-canvas
|
|
|
|
|
|
def test_viewport_fit_fraction_ignores_outliers() -> None:
|
|
"""Framing half the nodes zooms to the dense core, letting far outliers fall off-canvas."""
|
|
# A tight downtown cluster of five nodes (~50 m across) plus one distant outlier.
|
|
core = [
|
|
(45.5000, -73.5600), (45.5003, -73.5602), (45.4998, -73.5598),
|
|
(45.5001, -73.5599), (45.4999, -73.5601),
|
|
]
|
|
outlier = (46.80, -71.20)
|
|
pts = core + [outlier]
|
|
|
|
full = Viewport.fit(pts, 200, 120, max_zoom=16)
|
|
half = Viewport.fit(pts, 200, 120, max_zoom=16, fraction=0.5)
|
|
|
|
# The core-only frame is zoomed in tighter than the frame that must hold the outlier.
|
|
assert half.zoom > full.zoom
|
|
# Every clustered node stays on-canvas; the outlier is pushed off.
|
|
for lat, lon in core:
|
|
x, y = half.lonlat_to_dot(lat, lon)
|
|
assert 0 <= x < half.dot_w and 0 <= y < half.dot_h
|
|
ox, oy = half.lonlat_to_dot(*outlier)
|
|
assert not (0 <= ox < half.dot_w and 0 <= oy < half.dot_h)
|
|
|
|
|
|
def test_viewport_fit_fraction_keeps_small_sets_whole() -> None:
|
|
"""With only two nodes there is no core to isolate — both are still framed."""
|
|
pts = [(45.5019, -73.5674), (45.4768, -73.5990)]
|
|
half = Viewport.fit(pts, 200, 120, max_zoom=14, fraction=0.5)
|
|
full = Viewport.fit(pts, 200, 120, max_zoom=14)
|
|
assert (half.center_lat, half.center_lon, half.zoom) == (
|
|
full.center_lat, full.center_lon, full.zoom
|
|
)
|
|
|
|
|
|
def test_viewport_zoom_and_pan() -> None:
|
|
"""Zooming and panning return new viewports; zoom clamps to its bounds."""
|
|
vp = Viewport(45.5, -73.6, 10, 200, 120)
|
|
assert vp.zoomed(2).zoom == 12
|
|
assert vp.zoomed(50, max_zoom=16).zoom == 16 # clamped
|
|
assert vp.zoomed(-50, min_zoom=3).zoom == 3
|
|
east = vp.panned(0.5, 0)
|
|
assert east.center_lon > vp.center_lon # panning east increases longitude
|
|
north = vp.panned(0, -0.5)
|
|
assert north.center_lat > vp.center_lat # panning north increases latitude
|
|
|
|
|
|
def test_viewport_tiles_cover_the_view() -> None:
|
|
"""The viewport reports the handful of tiles overlapping it at the tile zoom."""
|
|
vp = Viewport(45.5019, -73.5674, 14, 200, 120)
|
|
tiles = vp.tiles(14)
|
|
assert tiles and all(z == 14 for z, _, _ in tiles)
|
|
# The fixture tile for this centre must be among them.
|
|
assert (14, 4843, 5861) in tiles
|
|
|
|
|
|
# -- braille canvas -----------------------------------------------------------
|
|
|
|
|
|
def test_canvas_plots_dots_as_braille() -> None:
|
|
"""A single plotted dot renders as a braille glyph in the top-left cell."""
|
|
canvas = MapCanvas(4, 2)
|
|
canvas.plot(0, 0, (255, 0, 0), 1)
|
|
out = _plain(canvas.to_ansi_lines())
|
|
assert out.splitlines()[0][0] == "⠁" # braille dot-1
|
|
|
|
|
|
def test_canvas_priority_decides_cell_colour() -> None:
|
|
"""When dots from two features share a cell, the higher priority sets its colour."""
|
|
canvas = MapCanvas(1, 1)
|
|
canvas.plot(0, 0, (10, 20, 30), 1) # low priority
|
|
canvas.plot(1, 0, (200, 100, 50), 9) # high priority, same cell
|
|
ansi = "".join(canvas.to_ansi_lines())
|
|
assert "38;2;200;100;50" in ansi # the high-priority colour won
|
|
assert "38;2;10;20;30" not in ansi
|
|
|
|
|
|
def test_canvas_marker_and_label_do_not_overprint() -> None:
|
|
"""A marker keeps its glyph and places its label in a neighbouring cell."""
|
|
canvas = MapCanvas(20, 3)
|
|
canvas.marker(4, 4, "★", (255, 255, 255))
|
|
assert canvas.marker_label(4, 4, "me", (255, 255, 255))
|
|
out = _plain(canvas.to_ansi_lines())
|
|
assert "★" in out and "me" in out
|
|
row = out.splitlines()[1]
|
|
assert "★me" not in row # a gap sits between the marker and its label
|
|
|
|
|
|
def test_canvas_label_collision_is_avoided() -> None:
|
|
"""A checked label is skipped when it would overlap an already-placed one."""
|
|
canvas = MapCanvas(20, 1)
|
|
assert canvas.place_label(10, 0, "First", (200, 200, 200))
|
|
assert not canvas.place_label(10, 0, "Second", (200, 200, 200)) # overlaps → skipped
|
|
|
|
|
|
def test_canvas_labels_keep_a_vertical_gap() -> None:
|
|
"""A checked label is skipped when it would sit flush above/below existing text."""
|
|
canvas = MapCanvas(20, 3)
|
|
assert canvas.place_label(20, 4, "Row1", (200, 200, 200)) # dot y=4 → cell row 1
|
|
# Directly below (cell row 2) with overlapping columns: rejected for lack of a gap.
|
|
assert not canvas.place_label(20, 8, "Row2", (200, 200, 200))
|
|
# Same row but clear of the horizontal span still fits.
|
|
assert canvas.place_label(2, 4, "Far", (200, 200, 200))
|
|
|
|
|
|
def test_parse_hex() -> None:
|
|
"""Hex colours parse to RGB triples, with or without the leading hash."""
|
|
assert parse_hex("#38bdf8") == (0x38, 0xBD, 0xF8)
|
|
assert parse_hex("ffffff") == (255, 255, 255)
|
|
|
|
|
|
# -- compositor over a real tile ----------------------------------------------
|
|
|
|
|
|
def test_render_map_draws_basemap_streets_and_labels() -> None:
|
|
"""Rendering the real fixture yields braille geometry and a place label."""
|
|
from meshterm.ui.map_render import MapMarker, render_map
|
|
|
|
vp = Viewport(45.5019, -73.5674, 14, 180, 120)
|
|
layers = decode_tile(_FIXTURE.read_bytes())
|
|
tiles = {(14, 4843, 5861): layers}
|
|
markers = [MapMarker("Yagi", 45.5019, -73.5674, is_repeater=True)]
|
|
out = _plain(render_map(vp, tiles, markers))
|
|
assert any(0x2800 <= ord(ch) <= 0x28FF for ch in out) # braille was drawn
|
|
assert "▲" in out and "Yagi" in out # the repeater marker + label
|
|
assert "Montréal" in out # a place label from the tile
|
|
|
|
|
|
def test_render_map_prioritises_repeater_glyph() -> None:
|
|
"""A repeater and a node at the same spot resolve to the repeater's marker on top."""
|
|
from meshterm.ui.map_render import MapMarker, render_map
|
|
|
|
vp = Viewport(45.50, -73.57, 14, 60, 40)
|
|
markers = [
|
|
MapMarker("node", 45.50, -73.57),
|
|
MapMarker("rptr", 45.50, -73.57, is_repeater=True),
|
|
]
|
|
out = _plain(render_map(vp, {}, markers))
|
|
assert "▲" in out and "●" not in out
|
|
|
|
|
|
def test_render_map_drops_crowded_labels_favouring_repeaters() -> None:
|
|
"""When labels can't all fit, the repeater's wins and a crowded node's is dropped."""
|
|
from meshterm.ui.map_render import MapMarker, render_map
|
|
|
|
# Three nodes stacked on one spot: only the two sides (left/right) can hold a label,
|
|
# so one of the three must show as a bare marker — and the repeater must not be it.
|
|
vp = Viewport(45.50, -73.57, 14, 40, 8)
|
|
markers = [
|
|
MapMarker("NODEONE", 45.50, -73.57),
|
|
MapMarker("NODETWO", 45.50, -73.57),
|
|
MapMarker("REPEATER", 45.50, -73.57, is_repeater=True),
|
|
]
|
|
out = _plain(render_map(vp, {}, markers))
|
|
assert "▲" in out # the repeater's glyph is on top
|
|
assert "REPEATER" in out # and it keeps its label (placed first)
|
|
# Only one of the two leaf nodes could fit a label; the other is a bare marker.
|
|
assert ("NODEONE" in out) != ("NODETWO" in out)
|
|
|
|
|
|
def _glyph_color(lines: list[str], glyph: str) -> tuple[int, int, int]:
|
|
"""Extract the truecolour ``(r, g, b)`` the given glyph was rendered with."""
|
|
m = re.search(
|
|
r"38;2;(\d+);(\d+);(\d+)m(?:\x1b\[1m)?" + re.escape(glyph), "".join(lines)
|
|
)
|
|
assert m, f"glyph {glyph!r} not found with a colour"
|
|
return int(m.group(1)), int(m.group(2)), int(m.group(3))
|
|
|
|
|
|
def test_render_map_brightens_piled_markers() -> None:
|
|
"""A cell many nodes share renders its glyph brighter than a lone node's."""
|
|
from meshterm.ui.map_render import _NODE, MapMarker, render_map
|
|
|
|
base = parse_hex(_NODE[1])
|
|
vp = Viewport(45.50, -73.57, 14, 60, 40)
|
|
|
|
lone = render_map(vp, {}, [MapMarker("solo", 45.50, -73.57)])
|
|
assert _glyph_color(lone, "●") == base # a single node keeps its base colour
|
|
|
|
crowd = [MapMarker(f"n{i}", 45.50, -73.57) for i in range(8)]
|
|
piled = _glyph_color(render_map(vp, {}, crowd), "●")
|
|
# Washed toward white: every channel is brighter than the base cyan.
|
|
assert all(p > b for p, b in zip(piled, base))
|
|
assert piled != base
|
|
|
|
|
|
def test_render_map_works_without_basemap() -> None:
|
|
"""With no tiles the nodes still render on a blank grid (the offline fallback)."""
|
|
from meshterm.ui.map_render import MapMarker, render_map
|
|
|
|
vp = Viewport.fit([(45.5, -73.6), (45.4, -73.5)], 120, 80, max_zoom=14)
|
|
markers = [
|
|
MapMarker("A", 45.5, -73.6, is_self=True),
|
|
MapMarker("B", 45.4, -73.5, is_repeater=True),
|
|
]
|
|
out = _plain(render_map(vp, {}, markers))
|
|
assert "★" in out and "▲" in out
|
|
|
|
|
|
# -- basemap source (offline behaviour) ---------------------------------------
|
|
|
|
|
|
def test_basemap_source_offline_is_graceful(tmp_path: Path) -> None:
|
|
"""An unreachable source reports unavailable and returns no tiles, never raising."""
|
|
from meshterm.services.basemap import BasemapSource
|
|
|
|
src = BasemapSource(tmp_path / "cache", tilejson_url="http://127.0.0.1:1/none", timeout=0.2)
|
|
assert src.available is False
|
|
assert src.max_zoom == 14 # sensible default when the TileJSON can't be fetched
|
|
assert src.load_tile(14, 4843, 5861) is None
|
|
|
|
|
|
def test_basemap_source_reads_cached_tile(tmp_path: Path) -> None:
|
|
"""A tile already on disk is decoded without any network access."""
|
|
from meshterm.services.basemap import BasemapSource
|
|
|
|
cache = tmp_path / "cache"
|
|
dest = cache / "tiles" / "14" / "4843"
|
|
dest.mkdir(parents=True)
|
|
(dest / "5861.pbf").write_bytes(_FIXTURE.read_bytes())
|
|
src = BasemapSource(cache, tilejson_url="http://127.0.0.1:1/none", timeout=0.2)
|
|
layers = src.load_tile(14, 4843, 5861)
|
|
assert layers is not None
|
|
assert any(layer.name == "transportation" for layer in layers)
|
|
|
|
|
|
def _offline_source(cache: Path):
|
|
"""A source that can't reach anything — every fetch is silence, not an empty tile."""
|
|
from meshterm.services.basemap import BasemapSource
|
|
|
|
return BasemapSource(cache, tilejson_url="http://127.0.0.1:1/none", timeout=0.2)
|
|
|
|
|
|
def test_basemap_source_never_caches_an_unanswered_fetch(tmp_path: Path) -> None:
|
|
"""A timeout writes nothing: a blank square now must not become a blank square for ever."""
|
|
from meshterm.services import basemap as basemap_mod
|
|
|
|
cache = tmp_path / "cache"
|
|
assert _offline_source(cache).load_tile(14, 4843, 5861) is None
|
|
assert list(cache.rglob("*.pbf")) == []
|
|
|
|
# The same, with the source resolved and the *fetch* the thing that fails — the flaky
|
|
# link the PicoCalc lives on. The tile stays unknown, so the next pan asks again.
|
|
src = _offline_source(cache)
|
|
src._template = "http://tiles.invalid/{z}/{x}/{y}.pbf"
|
|
src._resolved = True
|
|
calls: list[str] = []
|
|
|
|
def _get(url: str):
|
|
calls.append(url)
|
|
return basemap_mod._Response(False, b"")
|
|
|
|
src._http_get = _get # type: ignore[method-assign]
|
|
assert src.load_tile(14, 4843, 5861) is None
|
|
assert src.load_tile(14, 4843, 5861) is None
|
|
assert len(calls) == 2 # retried, not written off as empty
|
|
assert list(cache.rglob("*.pbf")) == []
|
|
|
|
|
|
def test_basemap_source_prunes_a_blank_cached_tile(tmp_path: Path) -> None:
|
|
"""A zero-byte entry (an older build's failure marker) is dropped rather than drawn."""
|
|
cache = tmp_path / "cache"
|
|
tile = cache / "tiles" / "14" / "4843" / "5861.pbf"
|
|
tile.parent.mkdir(parents=True)
|
|
tile.write_bytes(b"")
|
|
assert _offline_source(cache).load_tile(14, 4843, 5861) is None
|
|
assert not tile.exists() # pruned, so a session with network re-fetches it
|
|
|
|
|
|
def test_basemap_source_prunes_a_corrupt_cached_tile(tmp_path: Path) -> None:
|
|
"""Bytes cut short mid-write don't decode, so they're pruned instead of kept blank."""
|
|
cache = tmp_path / "cache"
|
|
tile = cache / "tiles" / "14" / "4843" / "5861.pbf"
|
|
tile.parent.mkdir(parents=True)
|
|
tile.write_bytes(_FIXTURE.read_bytes()[:200])
|
|
assert _offline_source(cache).load_tile(14, 4843, 5861) is None
|
|
assert not tile.exists()
|
|
|
|
|
|
def test_basemap_source_caches_only_tiles_with_content(tmp_path: Path) -> None:
|
|
"""The disk cache holds decodable geometry — never an empty or unparseable answer."""
|
|
from meshterm.services import basemap as basemap_mod
|
|
|
|
cache = tmp_path / "cache"
|
|
src = _offline_source(cache)
|
|
src._template = "http://tiles.invalid/{z}/{x}/{y}.pbf" # skip TileJSON resolution
|
|
src._resolved = True
|
|
|
|
bodies = iter([b"", b"not a vector tile at all", _FIXTURE.read_bytes()])
|
|
src._http_get = lambda url: basemap_mod._Response(True, next(bodies)) # type: ignore[method-assign]
|
|
|
|
assert src.load_tile(14, 1, 1) is None # answered "empty"
|
|
assert src.load_tile(14, 2, 2) is None # answered with junk
|
|
assert list(cache.rglob("*.pbf")) == []
|
|
layers = src.load_tile(14, 4843, 5861) # answered with a real tile
|
|
assert layers is not None
|
|
assert (cache / "tiles" / "14" / "4843" / "5861.pbf").exists()
|
|
|
|
|
|
def test_basemap_source_keeps_a_tile_whose_layers_are_all_undrawn(tmp_path: Path) -> None:
|
|
"""Narrowing the decode must not make a good cached tile look blank and get pruned.
|
|
|
|
A source told to decode a layer the tile doesn't carry sees no features at all. If
|
|
that were read as "nothing here", the entry would be deleted and re-downloaded on
|
|
every session — the cache paying for a rendering decision.
|
|
"""
|
|
from meshterm.services.basemap import BasemapSource
|
|
|
|
cache = tmp_path / "cache"
|
|
tile = cache / "tiles" / "14" / "4843" / "5861.pbf"
|
|
tile.parent.mkdir(parents=True)
|
|
tile.write_bytes(_FIXTURE.read_bytes())
|
|
|
|
src = BasemapSource(
|
|
cache,
|
|
tilejson_url="http://127.0.0.1:1/none",
|
|
timeout=0.2,
|
|
layers=frozenset({"a_layer_this_tile_does_not_have"}),
|
|
)
|
|
layers = src.load_tile(14, 4843, 5861)
|
|
|
|
assert layers, "a real tile must still read as real"
|
|
assert not any(layer.features for layer in layers) # nothing was decoded
|
|
assert tile.exists(), "the cache entry must survive"
|
|
|
|
|
|
def test_basemap_source_remembers_a_blank_tile_for_the_session(tmp_path: Path) -> None:
|
|
"""An answered-empty tile isn't re-requested this session (but isn't written down)."""
|
|
from meshterm.services import basemap as basemap_mod
|
|
|
|
src = _offline_source(tmp_path / "cache")
|
|
src._template = "http://tiles.invalid/{z}/{x}/{y}.pbf"
|
|
src._resolved = True
|
|
calls: list[str] = []
|
|
|
|
def _get(url: str):
|
|
calls.append(url)
|
|
return basemap_mod._Response(True, b"")
|
|
|
|
src._http_get = _get # type: ignore[method-assign]
|
|
assert src.load_tile(14, 1, 1) is None
|
|
assert src.load_tile(14, 1, 1) is None
|
|
assert len(calls) == 1
|
|
|
|
|
|
def test_basemap_http_get_separates_an_answer_from_silence(tmp_path: Path) -> None:
|
|
"""404 is an answer; a 500, and a body short of Content-Length, are not."""
|
|
import urllib.error
|
|
|
|
from meshterm.services.basemap import BasemapSource
|
|
|
|
src = BasemapSource(tmp_path / "cache")
|
|
|
|
class _Resp:
|
|
"""The slice of an ``http.client.HTTPResponse`` the fetcher touches."""
|
|
|
|
def __init__(self, body: bytes, declared: str | None) -> None:
|
|
self._body, self.headers = body, {"Content-Length": declared}
|
|
|
|
def read(self) -> bytes:
|
|
return self._body
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *_exc) -> bool:
|
|
return False
|
|
|
|
def _fake(result):
|
|
def _urlopen(_req, timeout=None): # noqa: ANN001
|
|
if isinstance(result, Exception):
|
|
raise result
|
|
return result
|
|
|
|
return _urlopen
|
|
|
|
def _get(result) -> tuple[bool, bytes]:
|
|
import urllib.request
|
|
|
|
saved = urllib.request.urlopen
|
|
urllib.request.urlopen = _fake(result) # type: ignore[assignment]
|
|
try:
|
|
return tuple(src._http_get("http://tiles.invalid/t"))
|
|
finally:
|
|
urllib.request.urlopen = saved # type: ignore[assignment]
|
|
|
|
err = lambda code: urllib.error.HTTPError("u", code, "", {}, None) # noqa: E731
|
|
assert _get(_Resp(b"tile-bytes", "10")) == (True, b"tile-bytes")
|
|
assert _get(_Resp(b"tile-b", "10")) == (False, b"") # truncated read
|
|
assert _get(_Resp(b"anything", None)) == (True, b"anything") # no declared length
|
|
assert _get(err(404)) == (True, b"") # "no tile here" — an answer
|
|
assert _get(err(500)) == (False, b"") # server fault — no answer
|
|
assert _get(err(429)) == (False, b"") # throttled — no answer
|
|
assert _get(TimeoutError("timed out")) == (False, b"")
|
|
|
|
|
|
# -- the tool -----------------------------------------------------------------
|
|
|
|
|
|
@pytest.fixture()
|
|
def ctx(tmp_path: Path):
|
|
"""A mock-backed application context with the plain (console) UI surface."""
|
|
from meshterm.context import AppContext
|
|
from meshterm.core.admin_store import AdminStore
|
|
from meshterm.core.config import Settings
|
|
from meshterm.core.device_store import DeviceStore
|
|
from meshterm.persistence.repository import Repository
|
|
from meshterm.ui.theme import MESH_THEME
|
|
|
|
settings = Settings(config_dir=tmp_path, db_path=tmp_path / "map.db")
|
|
context = AppContext(
|
|
console=Console(file=io.StringIO(), theme=MESH_THEME),
|
|
settings=settings,
|
|
repo=Repository(settings.db_path),
|
|
device_store=DeviceStore(tmp_path / "devices.json"),
|
|
admin_store=AdminStore(tmp_path / "admin.json"),
|
|
mock=True,
|
|
)
|
|
yield context
|
|
context.repo.close()
|
|
|
|
|
|
async def test_map_tool_static_render_plots_contacts(ctx) -> None:
|
|
"""The CLI path plots the device's located contacts and reports the breakdown.
|
|
|
|
The mock companion's contacts carry two located repeaters and one located leaf node
|
|
(a fourth, Alice, has no fix), so the contact list — not the observation history — is
|
|
what fills the map.
|
|
"""
|
|
# Seed an observation whose key matches a located contact, so its reception detail is
|
|
# merged onto that contact's marker.
|
|
run_id = ctx.repo.start_run("monitor", {})
|
|
ctx.repo.record_observation(
|
|
run_id,
|
|
Observation(node="a1b2c3d4", name="Yagi-Repeater", node_type=NODE_TYPE_REPEATER,
|
|
snr=6.0, observed_at=utcnow()),
|
|
)
|
|
result = await MapTool().run(ctx, {"static": True, "basemap": False})
|
|
assert result.summary["located"] == 3
|
|
assert result.summary["repeaters"] == 2
|
|
assert result.summary["nodes"] == 1
|
|
assert result.summary["self_located"] is False
|
|
out = ctx.console.file.getvalue()
|
|
assert out != ""
|
|
assert "1 pkts" in out # the seeded observation's detail merged onto the contact
|
|
|
|
|
|
def test_usable_fix_rejects_out_of_range_and_null_island() -> None:
|
|
"""A fix is usable only in valid lat/lon range and away from the 0/0 no-GPS sentinel."""
|
|
from meshterm.core.geo import usable_fix
|
|
|
|
assert usable_fix(45.5, -73.6) is True
|
|
assert usable_fix(90.0, 180.0) is True and usable_fix(-90.0, -180.0) is True # extremes
|
|
assert usable_fix(0.0, 0.0) is False # null island (a no-GPS companion reports this)
|
|
assert usable_fix(-97.0, -1041.97) is False # out of range (a real advert seen in the wild)
|
|
assert usable_fix(45.0, 181.0) is False and usable_fix(-91.0, -73.0) is False
|
|
|
|
|
|
async def test_gather_markers_drops_out_of_range_fix(ctx) -> None:
|
|
"""A node advertising nonsense coordinates is never plotted, only the real one is.
|
|
|
|
Some firmware reports an out-of-range fix (a companion has advertised ``lat -97,
|
|
lon -1042``); projecting it flings the view off the world, leaving the map an all-black
|
|
off-world frame. It is treated as no fix at all, the same guard the 0/0 null island gets.
|
|
"""
|
|
from meshterm.tools.map import gather_markers
|
|
|
|
run_id = ctx.repo.start_run("monitor", {})
|
|
ctx.repo.record_observation(run_id, Observation(
|
|
node="beefbeefbeef", name="BadFix", node_type=NODE_TYPE_REPEATER,
|
|
lat=-97.0, lon=-1041.97, observed_at=utcnow()))
|
|
ctx.repo.record_observation(run_id, Observation(
|
|
node="cafecafecafe", name="GoodFix", node_type=NODE_TYPE_REPEATER,
|
|
lat=45.5, lon=-73.6, observed_at=utcnow()))
|
|
labels = {m.label for m in await gather_markers(ctx)}
|
|
assert "GoodFix" in labels
|
|
assert "BadFix" not in labels
|
|
|
|
|
|
def test_repository_round_trips_map_view(ctx) -> None:
|
|
"""The saved map viewport persists and reads back; absent by default."""
|
|
assert ctx.repo.get_map_view() is None
|
|
ctx.repo.set_map_view(45.51, -73.57, 13)
|
|
lat, lon, zoom = ctx.repo.get_map_view()
|
|
assert (lat, lon) == pytest.approx((45.51, -73.57))
|
|
assert zoom == 13
|
|
# A later save overwrites the single stored view.
|
|
ctx.repo.set_map_view(40.0, -74.0, 9)
|
|
assert ctx.repo.get_map_view() == pytest.approx((40.0, -74.0, 9))
|
|
|
|
|
|
async def test_map_tool_reports_nothing_to_plot(ctx, monkeypatch) -> None:
|
|
"""With no located contacts and no located history the tool returns cleanly."""
|
|
async def _no_contacts(_ctx):
|
|
return []
|
|
|
|
monkeypatch.setattr("meshterm.tools.map._contacts", _no_contacts)
|
|
result = await MapTool().run(ctx, {"static": True, "basemap": False})
|
|
assert result.summary == {"located": 0}
|
|
|
|
|
|
async def test_gather_markers_drops_null_island_fixes(ctx, monkeypatch) -> None:
|
|
"""A node advertising a 0/0 fix (no GPS lock) is not plotted at null island.
|
|
|
|
Framing such a marker (filter to it, press Enter) would drop the map onto the empty
|
|
mid-Atlantic — a screen of solid water fill — so a both-near-zero fix is treated as no
|
|
fix at all, exactly as our own node's marker already is.
|
|
"""
|
|
from meshterm.core.models import Contact
|
|
from meshterm.tools.map import gather_markers
|
|
|
|
async def _contacts(_ctx):
|
|
return [
|
|
Contact(name="Real", public_key="aa" * 32, lat=45.50, lon=-73.57),
|
|
Contact(name="NoFix", public_key="bb" * 32, lat=0.0, lon=0.0),
|
|
]
|
|
|
|
monkeypatch.setattr("meshterm.tools.map._contacts", _contacts)
|
|
labels = [m.label for m in await gather_markers(ctx)]
|
|
assert "Real" in labels
|
|
assert "NoFix" not in labels # the null-island node is dropped
|
|
|
|
|
|
class _StubSession:
|
|
"""Minimal stand-in for :class:`TuiSession` for driving :class:`MapScreen`."""
|
|
|
|
def __init__(self, cols: int, rows: int) -> None:
|
|
self._cols, self._rows = cols, rows
|
|
self.invalidated = 0
|
|
|
|
def base_body_size(self) -> tuple[int, int]:
|
|
return self._cols, self._rows
|
|
|
|
def invalidate(self) -> None:
|
|
self.invalidated += 1
|
|
|
|
|
|
class _StubSource:
|
|
"""An always-offline tile source, so the screen renders nodes only (no network)."""
|
|
|
|
available = False
|
|
max_zoom = 14
|
|
|
|
def load_tile(self, z: int, x: int, y: int):
|
|
return None
|
|
|
|
|
|
class _CapturingSession(_StubSession):
|
|
"""Drives :func:`open_map` headless: captures the pushed screen, renders it once, then
|
|
replays a canned key sequence — so a test can assert on where the map opened and what
|
|
it did (or didn't) persist as the view moves."""
|
|
|
|
def __init__(self, cols: int, rows: int, keys: tuple[str, ...] = ()) -> None:
|
|
super().__init__(cols, rows)
|
|
self.keys = keys
|
|
self.screen = None
|
|
self.repainted = False
|
|
|
|
async def run_screen(self, screen): # noqa: ANN001, ANN201
|
|
self.screen = screen
|
|
screen.render_body(self._cols)
|
|
for key in self.keys:
|
|
screen.handle(key)
|
|
return None
|
|
|
|
def request_full_repaint(self) -> None:
|
|
self.repainted = True
|
|
|
|
|
|
def test_map_screen_renders_pans_zooms_and_resets() -> None:
|
|
"""The interactive screen fills its body, and wasd/zoom/reset move the viewport."""
|
|
from meshterm.core.geo import Viewport
|
|
from meshterm.ui.map_render import MapMarker
|
|
from meshterm.ui.map_screen import MapScreen
|
|
|
|
session = _StubSession(80, 24)
|
|
markers = [
|
|
MapMarker("A", 45.50, -73.60, is_repeater=True),
|
|
MapMarker("B", 45.40, -73.50),
|
|
]
|
|
screen = MapScreen(session, markers, _StubSource(), 14)
|
|
|
|
lines = screen.render_body(80)
|
|
assert len(lines) == 24 # fills the body height reported by the session
|
|
start = screen._viewport
|
|
|
|
screen.handle("right") # pan east → longitude increases
|
|
assert screen._viewport.center_lon > start.center_lon
|
|
screen.handle("up") # pan north → latitude increases
|
|
assert screen._viewport.center_lat > start.center_lat
|
|
|
|
z = screen._viewport.zoom
|
|
screen.handle("pageup")
|
|
assert screen._viewport.zoom == z + 1
|
|
screen.handle("pagedown")
|
|
assert screen._viewport.zoom == z
|
|
|
|
screen.handle("home") # reset refits to the nodes
|
|
refit = Viewport.fit([(m.lat, m.lon) for m in markers], start.dot_w, start.dot_h, max_zoom=14)
|
|
assert screen._viewport.zoom == refit.zoom
|
|
assert screen._viewport.center_lat == pytest.approx(refit.center_lat)
|
|
|
|
# Letters never pan or zoom — they feed the find filter instead.
|
|
view = screen._viewport
|
|
screen.handle("text", "d")
|
|
assert screen._viewport is view and screen._filter == "d"
|
|
screen.handle("backspace")
|
|
|
|
# Offline source: nodes still render and both markers are present.
|
|
assert "▲" in _plain(screen.render_body(80)) and "●" in _plain(screen.render_body(80))
|
|
assert "offline" in screen.footer_hint
|
|
|
|
|
|
def test_map_screen_shift_pans_by_a_single_cell() -> None:
|
|
"""Holding Shift with an arrow pans finely, by one character cell."""
|
|
from meshterm.core.geo import lonlat_to_world, world_to_lonlat
|
|
from meshterm.ui.map_render import MapMarker
|
|
from meshterm.ui.map_screen import MapScreen
|
|
|
|
markers = [MapMarker("A", 45.50, -73.60), MapMarker("B", 45.40, -73.50)]
|
|
screen = MapScreen(_StubSession(80, 24), markers, _StubSource(), 14)
|
|
screen.render_body(80)
|
|
start = screen._viewport
|
|
|
|
# A coarse east pan (30% of the view) moves much further than a fine one.
|
|
screen.handle("right")
|
|
coarse = screen._viewport.center_lon - start.center_lon
|
|
|
|
screen._viewport = start
|
|
screen.handle("shift_right")
|
|
fine = screen._viewport.center_lon - start.center_lon
|
|
assert 0 < fine < coarse
|
|
|
|
# The fine step is exactly one cell (2 dots) east at the current zoom.
|
|
cx, cy = lonlat_to_world(start.center_lat, start.center_lon, start.zoom)
|
|
_, expected_lon = world_to_lonlat(cx + 2, cy, start.zoom)
|
|
assert screen._viewport.center_lon == pytest.approx(expected_lon)
|
|
|
|
|
|
def test_map_fkey_lane_names_zoom_and_drops_the_dead_slot() -> None:
|
|
"""The map repurposes the nav keys, so its lane labels them — and offers no dead key."""
|
|
from meshterm.ui.map_render import MapMarker
|
|
from meshterm.ui.map_screen import MapScreen
|
|
from meshterm.ui.tui.fkeys import action_for
|
|
|
|
screen = MapScreen(_StubSession(80, 24), [MapMarker("A", 45.5, -73.6)], _StubSource(), 14)
|
|
lane = screen.fkey_lane
|
|
|
|
# PgUp/PgDn zoom here and Home refits — the shared "Top/End/PgUp/PgDn" labels would
|
|
# all be lies, and "end" is bound to nothing at all.
|
|
assert [pair.label if pair else None for pair in lane] == [
|
|
"Reset", None, None, "Zoom +", "Zoom -",
|
|
]
|
|
assert action_for(lane, 2) is None # F2 offers nothing rather than a dead End
|
|
assert action_for(lane, 1) == "home" and action_for(lane, 4) == "pageup"
|
|
assert all(pair.enabled for pair in lane if pair) # a map can always zoom or refit
|
|
|
|
|
|
def test_map_screen_find_filters_frames_and_clears() -> None:
|
|
"""Typing builds the find query; Enter frames matches; Esc peels filter then map."""
|
|
import asyncio
|
|
|
|
from meshterm.ui.map_render import MapMarker
|
|
from meshterm.ui.map_screen import MapScreen
|
|
|
|
markers = [
|
|
MapMarker("YUL-Cartierville", 45.53, -73.71, is_repeater=True),
|
|
MapMarker("Alice", 45.40, -73.50),
|
|
MapMarker("Yagi-North", 45.60, -73.65),
|
|
]
|
|
screen = MapScreen(_StubSession(80, 24), markers, _StubSource(), 14)
|
|
screen.render_body(80)
|
|
|
|
for ch in "yu":
|
|
screen.handle("text", ch)
|
|
assert screen._filter == "yu"
|
|
assert [m.label for m in screen._matches()] == ["YUL-Cartierville"]
|
|
screen.render_body(80)
|
|
assert "1 of 3 match" in screen.title
|
|
assert "find: yu" in screen.footer_hint
|
|
|
|
# Only the match keeps a label; the others dim to bare context glyphs.
|
|
body = _plain(screen.render_body(80))
|
|
assert "YUL-Cartierville" in body
|
|
assert "Alice" not in body and "Yagi-North" not in body
|
|
|
|
# Enter frames the matches: the view centres on YUL.
|
|
screen.handle("enter")
|
|
assert screen._viewport.center_lat == pytest.approx(45.53, abs=0.05)
|
|
assert screen._viewport.center_lon == pytest.approx(-73.71, abs=0.05)
|
|
|
|
# Backspace edits; Esc clears the filter first and only then dismisses.
|
|
screen.handle("backspace")
|
|
assert screen._filter == "y"
|
|
|
|
async def drive() -> object:
|
|
screen.future = asyncio.get_running_loop().create_future()
|
|
screen.handle("escape")
|
|
assert screen._filter == "" and not screen.future.done()
|
|
screen.handle("escape")
|
|
return await screen.future
|
|
|
|
assert asyncio.run(drive()) is None
|
|
|
|
|
|
def test_map_screen_enter_zooms_in_on_a_single_match() -> None:
|
|
"""Enter on a lone filtered node homes in close on it, not the fit's neutral default."""
|
|
from meshterm.ui.map_render import MapMarker
|
|
from meshterm.ui.map_screen import _FIND_ZOOM, MapScreen
|
|
|
|
markers = [
|
|
MapMarker("YUL-Cartierville", 45.53, -73.71, is_repeater=True),
|
|
MapMarker("Alice", 45.40, -73.50),
|
|
]
|
|
screen = MapScreen(_StubSession(80, 24), markers, _StubSource(), 19) # source max 19
|
|
screen.render_body(80)
|
|
for ch in "yul":
|
|
screen.handle("text", ch)
|
|
assert [m.label for m in screen._matches()] == ["YUL-Cartierville"]
|
|
screen.handle("enter")
|
|
assert screen._viewport.center_lat == pytest.approx(45.53, abs=0.01)
|
|
assert screen._viewport.center_lon == pytest.approx(-73.71, abs=0.01)
|
|
assert screen._viewport.zoom == _FIND_ZOOM # zoomed in, not the fit's default 14
|
|
|
|
|
|
def test_map_screen_restores_and_persists_view() -> None:
|
|
"""The screen reopens on a saved view and reports every centre/zoom change."""
|
|
from meshterm.ui.map_render import MapMarker
|
|
from meshterm.ui.map_screen import MapScreen
|
|
|
|
saved: list[tuple[float, float, int]] = []
|
|
markers = [MapMarker("A", 45.50, -73.60), MapMarker("B", 45.40, -73.50)]
|
|
screen = MapScreen(
|
|
_StubSession(80, 24), markers, _StubSource(), 14,
|
|
saved_view=(46.80, -71.20, 12),
|
|
on_view_change=lambda vp: saved.append(
|
|
(vp.center_lat, vp.center_lon, vp.zoom)
|
|
),
|
|
)
|
|
|
|
screen.render_body(80)
|
|
# Restored to the saved centre/zoom, not a fit of the markers.
|
|
assert screen._viewport.zoom == 12
|
|
assert screen._viewport.center_lat == pytest.approx(46.80)
|
|
assert screen._viewport.center_lon == pytest.approx(-71.20)
|
|
assert saved == [] # reopening unchanged doesn't rewrite the stored view
|
|
|
|
screen.handle("right") # pan east persists the moved view
|
|
assert saved and saved[-1][2] == 12 # zoom unchanged
|
|
assert saved[-1][1] > -71.20 # centre moved east
|
|
|
|
screen.handle("pageup") # zooming in persists too
|
|
assert saved[-1][2] == 13
|
|
|
|
|
|
async def test_open_map_focus_centres_on_the_node_without_clobbering_the_saved_view(
|
|
ctx, monkeypatch
|
|
) -> None:
|
|
"""Opening the full map focused on a node (from its detail page) centres there, at the
|
|
inline preview's zoom — and, being a transient peek, never overwrites the persisted
|
|
'where you left the map' global view, even as the peek is panned and zoomed."""
|
|
from meshterm.core.geo import clamp_lat
|
|
from meshterm.ui.map_render import MapMarker
|
|
from meshterm.ui.map_screen import open_map
|
|
from meshterm.ui.surface import TuiUi
|
|
|
|
monkeypatch.setattr("meshterm.ui.map_screen.basemap_source", lambda _c: _StubSource())
|
|
ctx.repo.set_map_view(10.0, 20.0, 8) # where the user last left the global map
|
|
|
|
session = _CapturingSession(80, 24, keys=("right", "pageup")) # pan + zoom the peek
|
|
ctx.ui = TuiUi(session)
|
|
await open_map(
|
|
ctx,
|
|
[MapMarker("Hub", 45.5, -73.6, is_repeater=True)],
|
|
focus=(45.5, -73.6),
|
|
find="Hub",
|
|
)
|
|
|
|
# Opened on the node, at the detail preview's zoom (min(13, max_zoom)).
|
|
assert session.screen._saved_view == (clamp_lat(45.5), -73.6, 13)
|
|
assert session.screen._on_view_change is None # a focused peek wires no persistence
|
|
assert session.screen._filter == "Hub" # the find opens seeded to the focused node
|
|
# Panning/zooming the peek left the global 'where you left the map' view untouched.
|
|
assert ctx.repo.get_map_view() == pytest.approx((10.0, 20.0, 8))
|
|
assert session.repainted # cleaned the terminal on the way out, like the plain open
|
|
|
|
|
|
async def test_open_map_without_focus_restores_and_persists_the_global_view(
|
|
ctx, monkeypatch
|
|
) -> None:
|
|
"""With no focus the full map is unchanged: it reopens where the user left it and pans
|
|
persist straight back to that global view."""
|
|
from meshterm.ui.map_render import MapMarker
|
|
from meshterm.ui.map_screen import open_map
|
|
from meshterm.ui.surface import TuiUi
|
|
|
|
monkeypatch.setattr("meshterm.ui.map_screen.basemap_source", lambda _c: _StubSource())
|
|
ctx.repo.set_map_view(46.80, -71.20, 12) # the saved view to reopen on
|
|
|
|
session = _CapturingSession(80, 24, keys=("right",)) # pan east
|
|
ctx.ui = TuiUi(session)
|
|
await open_map(ctx, [MapMarker("Hub", 46.8, -71.2)])
|
|
|
|
assert session.screen._viewport.zoom == 12 # reopened on the saved view, not a node fit
|
|
lat, lon, zoom = ctx.repo.get_map_view()
|
|
assert zoom == 12 and lon > -71.20 # the eastward pan persisted back to the global view
|
|
|
|
|
|
def test_map_screen_seeded_find_behaves_like_a_typed_one() -> None:
|
|
"""A seeded find matches, titles, and clears exactly as if the user had typed it."""
|
|
from meshterm.ui.map_render import MapMarker
|
|
from meshterm.ui.map_screen import MapScreen
|
|
|
|
markers = [MapMarker("Hub", 45.5, -73.6), MapMarker("Far", 45.6, -73.5)]
|
|
screen = MapScreen(_StubSession(80, 24), markers, _StubSource(), 14, find="Hub")
|
|
assert [m.label for m in screen._matches()] == ["Hub"] # only the seed's node lights
|
|
assert "find: Hub" in screen.footer_hint # the query shows, editable, as ever
|
|
screen.handle("escape") # first Esc clears the find (not the map)
|
|
assert screen._filter == "" and len(screen._matches()) == 2
|
|
|
|
|
|
def test_map_screen_scrubs_right_edge_after_move() -> None:
|
|
"""A move asks the session to scrub the right edge once, to clear any braille smear."""
|
|
from meshterm.ui.map_render import MapMarker
|
|
from meshterm.ui.map_screen import MapScreen
|
|
|
|
markers = [MapMarker("A", 45.5, -73.6)]
|
|
braille = MapScreen(_StubSession(80, 24), markers, _StubSource(), 14)
|
|
braille.render_body(80)
|
|
braille.handle("right") # a move flags the edge for a scrub
|
|
assert braille.consume_edge_scrub() == 2 # right padding + border columns
|
|
assert braille.consume_edge_scrub() == 0 # consumed — not repeated without another move
|
|
|
|
|
|
def test_map_screen_escape_dismisses() -> None:
|
|
"""Esc resolves the screen's future with None (backs out to the menu)."""
|
|
import asyncio
|
|
|
|
from meshterm.ui.map_render import MapMarker
|
|
from meshterm.ui.map_screen import MapScreen
|
|
|
|
screen = MapScreen(_StubSession(80, 24), [MapMarker("A", 45.5, -73.6)], _StubSource(), 14)
|
|
|
|
async def drive() -> object:
|
|
screen.future = asyncio.get_running_loop().create_future()
|
|
screen.handle("escape")
|
|
return await screen.future
|
|
|
|
assert asyncio.run(drive()) is None
|
|
|
|
|
|
def test_map_observation_round_trips_node_type(ctx) -> None:
|
|
"""A stored observation's advert type survives persistence and aggregation."""
|
|
run_id = ctx.repo.start_run("monitor", {})
|
|
ctx.repo.record_observation(
|
|
run_id,
|
|
Observation(node="a1", name="Yagi", node_type=NODE_TYPE_REPEATER, snr=6.0,
|
|
lat=45.5, lon=-73.5),
|
|
)
|
|
node = next(n for n in ctx.repo.heard_nodes() if n.node == "a1")
|
|
assert node.is_repeater and node.node_type == NODE_TYPE_REPEATER
|
|
|
|
|
|
def test_render_map_labels_take_the_name_hue_ours_white() -> None:
|
|
"""Node labels carry their key-derived hue; our own label is white while the
|
|
★ glyph stays yellow; a keyless label lands on the muted grey."""
|
|
from meshterm.ui.map_render import _SELF, MapMarker, render_map
|
|
from meshterm.ui.widgets import name_rgb
|
|
|
|
vp = Viewport.fit([(45.5, -73.6), (45.4, -73.5)], 120, 80, max_zoom=14)
|
|
markers = [
|
|
MapMarker("US", 45.5, -73.6, is_self=True, key="cc" * 32),
|
|
MapMarker("KEYED", 45.4, -73.5, key="d4" * 32),
|
|
MapMarker("BARE", 45.45, -73.55),
|
|
]
|
|
lines = render_map(vp, {}, markers)
|
|
assert _glyph_color(lines, "★") == parse_hex(_SELF[1]) # the glyph keeps its yellow
|
|
assert _glyph_color(lines, "US") == (255, 255, 255) # ...the label goes you-white
|
|
assert _glyph_color(lines, "KEYED") == name_rgb("KEYED", "d4" * 32)
|
|
assert _glyph_color(lines, "BARE") == (148, 163, 184) # no key, the muted grey
|
|
|
|
|
|
def test_render_map_find_matches_still_label_white() -> None:
|
|
"""An active find filter keeps forcing matching labels to full white."""
|
|
from meshterm.ui.map_render import MapMarker, render_map
|
|
|
|
vp = Viewport.fit([(45.5, -73.6), (45.4, -73.5)], 120, 80, max_zoom=14)
|
|
markers = [MapMarker("TARGET", 45.5, -73.6, key="d4" * 32)]
|
|
lines = render_map(vp, {}, markers, find="targ")
|
|
assert _glyph_color(lines, "TARGET") == (255, 255, 255)
|