Jean-Pierre MartineauandClaude Fable 5
e9af03f227
On the PicoCalc the query overlays the canvas last row - directly above the F-key lane - so the ground never shifts when a find begins. You (Ctrl+L) now clears the query as it comes home; its Shift half, You +, also zooms in close, and Clear behind Frame drops the query outright. Shifted arrows fine-pan on the handheld too: the Shift watcher covers a console that strips the modifier off an arrow. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
1485 lines
62 KiB
Python
1485 lines
62 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 os
|
|
import re
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from rich.cells import cell_len
|
|
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, Layer, 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 ("housenumber", "poi", "mountain_peak"):
|
|
assert skipped in narrow, f"{skipped} should still be named"
|
|
assert not narrow[skipped].features, f"{skipped} should not have been decoded"
|
|
|
|
|
|
# -- buildings ----------------------------------------------------------------
|
|
|
|
|
|
def _building_dots(zoom: int) -> int:
|
|
"""Lit cells in a render of the fixture's buildings alone, at ``zoom``."""
|
|
from meshterm.ui.map_render import render_map
|
|
|
|
layers = decode_tile(_FIXTURE.read_bytes(), layers={"building"})
|
|
vp = Viewport(45.4995, -73.5690, zoom, 53 * 2, 26 * 4)
|
|
out = _plain(render_map(vp, {(14, 4843, 5861): layers}, []))
|
|
return sum(1 for ch in out if ch not in " \n")
|
|
|
|
|
|
def test_buildings_wait_for_the_zoom_that_can_hold_them() -> None:
|
|
"""Below the gate a footprint is about two dots across, so the layer is a haze."""
|
|
from meshterm.ui.map_render import _BUILDING_MIN_ZOOM
|
|
|
|
assert _building_dots(_BUILDING_MIN_ZOOM) > 100
|
|
assert _building_dots(_BUILDING_MIN_ZOOM - 1) == 0
|
|
|
|
|
|
def test_building_fill_is_stippled_so_the_streets_survive_it() -> None:
|
|
"""A braille dot is one bit: a *solid* fill would erase the street grid it covers.
|
|
|
|
The stipple is what keeps a building a shade rather than an eraser — every cell it
|
|
touches keeps free dots for a road to be drawn through.
|
|
"""
|
|
from meshterm.ui.map_render import _BUILDING_MIN_ZOOM, render_map
|
|
|
|
layers = decode_tile(_FIXTURE.read_bytes(), layers={"building"})
|
|
vp = Viewport(45.4995, -73.5690, _BUILDING_MIN_ZOOM, 53 * 2, 26 * 4)
|
|
out = _plain(render_map(vp, {(14, 4843, 5861): layers}, []))
|
|
|
|
assert "⣿" not in out, "a fully lit cell means the fill left a road nowhere to go"
|
|
|
|
|
|
def test_offscreen_buildings_do_not_change_the_frame() -> None:
|
|
"""The bbox reject is an optimisation, so it must be invisible in the output.
|
|
|
|
A downtown tile carries thousands of footprints and a zoomed-in view holds a few
|
|
dozen; each one is rejected on its own tile-local bounds before a single point of it
|
|
is projected. Adding a footprint the view cannot reach must render byte-identically.
|
|
"""
|
|
from meshterm.core.mvt import Feature
|
|
from meshterm.ui.map_render import _BUILDING_MIN_ZOOM, render_map
|
|
|
|
vp = Viewport(45.4995, -73.5690, _BUILDING_MIN_ZOOM, 53 * 2, 26 * 4)
|
|
cx, cy = _tile_local_of_view_centre(vp, 14, 4843, 5861)
|
|
here = [(cx - 40, cy - 40), (cx + 40, cy - 40), (cx + 40, cy + 40),
|
|
(cx - 40, cy + 40), (cx - 40, cy - 40)]
|
|
far = [(10, 10), (90, 10), (90, 90), (10, 90), (10, 10)]
|
|
|
|
def frame(rings):
|
|
layer = Layer(name="building", extent=4096,
|
|
features=[Feature(geom_type=GEOM_POLYGON, rings=rings, tags={})])
|
|
return render_map(vp, {(14, 4843, 5861): [layer]}, [])
|
|
|
|
assert _plain(frame([here])).strip(), "the in-view footprint should draw something"
|
|
assert frame([here]) == frame([here, far])
|
|
|
|
|
|
# -- 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 _street_labels_placed(cols: int, rows: int, zoom: int) -> int:
|
|
"""How many street names actually reach the canvas at this size and zoom."""
|
|
from meshterm.ui.map_render import DRAWN_LAYERS, _draw_tile, _Frame
|
|
from meshterm.ui.mapcanvas import MapCanvas
|
|
|
|
layers = decode_tile(_FIXTURE.read_bytes(), layers=DRAWN_LAYERS)
|
|
vp = Viewport(45.5019, -73.5674, zoom, cols * 2, (rows - 4) * 4)
|
|
canvas = MapCanvas(vp.dot_w // 2, vp.dot_h // 4)
|
|
frame = _Frame(canvas=canvas, viewport=vp)
|
|
_draw_tile(frame, layers, 14, 4843, 5861)
|
|
|
|
placed = 0
|
|
for label in sorted(frame.labels, key=lambda l: l.rank):
|
|
if vp.zoom < label.min_zoom:
|
|
continue
|
|
for ax, ay in ((label.x, label.y), *label.alts):
|
|
if canvas.place_label(ax, ay, label.text, label.color, bold=label.bold):
|
|
placed += label.min_zoom == 15 # the street-label gate identifies them
|
|
break
|
|
return placed
|
|
|
|
|
|
def test_street_labels_get_denser_as_you_zoom_in() -> None:
|
|
"""Zooming toward a street must not make its name less likely to appear.
|
|
|
|
Anchoring a label to the feature's own midpoint pinned it to a fixed geographic
|
|
point, so the tighter the view the less often that point was still on screen — the
|
|
fixture's 265 named streets yielded 2 labels at zoom 16 and 10 at zoom 15.
|
|
"""
|
|
assert _street_labels_placed(53, 26, 16) >= 5
|
|
assert _street_labels_placed(53, 26, 15) >= 10
|
|
# A wider terminal sees more of the same ground, so it may name more -- never fewer.
|
|
assert _street_labels_placed(72, 24, 16) >= _street_labels_placed(53, 26, 16)
|
|
|
|
|
|
def _tile_local_of_view_centre(vp: Viewport, tz: int, tx: int, ty: int, extent: int = 4096):
|
|
"""The tile-local point that lands in the middle of this viewport (inverts feature_to_dot)."""
|
|
from meshterm.core.geo import TILE_PX
|
|
|
|
ox, oy = vp.origin_world
|
|
scale = 2.0 ** (vp.zoom - tz)
|
|
lx = extent * ((vp.dot_w / 2 + ox) / (TILE_PX * scale) - tx)
|
|
ly = extent * ((vp.dot_h / 2 + oy) / (TILE_PX * scale) - ty)
|
|
return lx, ly
|
|
|
|
|
|
def test_street_name_is_drawn_only_once() -> None:
|
|
"""OSM splits a long street into several features; the map names it once."""
|
|
from meshterm.ui.map_render import DRAWN_LAYERS, MapMarker, render_map
|
|
|
|
layers = decode_tile(_FIXTURE.read_bytes(), layers=DRAWN_LAYERS)
|
|
vp = Viewport(45.5019, -73.5674, 16, 53 * 2, 22 * 4)
|
|
out = _plain(render_map(vp, {(14, 4843, 5861): layers},
|
|
[MapMarker("YUL", 45.5040, -73.5700, is_repeater=True)]))
|
|
|
|
# René-Lévesque arrives as several segments and used to be drawn twice on one screen.
|
|
assert out.count("René-Lévesque") <= 1, "a street was named more than once"
|
|
|
|
|
|
def test_line_label_anchors_on_the_visible_stretch() -> None:
|
|
"""A street crossing the view is named even with both its endpoints off screen."""
|
|
from meshterm.ui.map_render import _Frame, _add_line_label
|
|
from meshterm.ui.mapcanvas import MapCanvas
|
|
|
|
vp = Viewport(45.5019, -73.5674, 14, 120, 80)
|
|
frame = _Frame(canvas=MapCanvas(60, 20), viewport=vp)
|
|
_, cy = _tile_local_of_view_centre(vp, 14, 4843, 5861)
|
|
# A line spanning the whole tile at the view's latitude: its own midpoint is far away,
|
|
# but it crosses the canvas, so the clip must find it.
|
|
_add_line_label(frame, [[(0, cy), (4096, cy)]], 4096, 14, 4843, 5861, "Rue Long",
|
|
("#9aa0aa", False, 7))
|
|
|
|
assert len(frame.labels) == 1
|
|
label = frame.labels[0]
|
|
assert 0 <= label.x <= vp.dot_w and 0 <= label.y <= vp.dot_h
|
|
|
|
|
|
def test_line_label_off_screen_is_not_queued() -> None:
|
|
"""A feature with nothing on screen costs no label slot at all."""
|
|
from meshterm.ui.map_render import _Frame, _add_line_label
|
|
from meshterm.ui.mapcanvas import MapCanvas
|
|
|
|
vp = Viewport(45.5019, -73.5674, 16, 120, 80)
|
|
frame = _Frame(canvas=MapCanvas(60, 20), viewport=vp)
|
|
# A short line in the far corner of the tile, well outside a zoom-16 window.
|
|
_add_line_label(frame, [[(0, 0), (8, 8)]], 4096, 14, 4843, 5861, "Nowhere",
|
|
("#9aa0aa", False, 7))
|
|
|
|
assert frame.labels == []
|
|
|
|
|
|
def test_line_label_offers_alternates_along_the_visible_run() -> None:
|
|
"""A crowded first choice falls back further along the street rather than vanishing."""
|
|
from meshterm.ui.map_render import _Frame, _add_line_label
|
|
from meshterm.ui.mapcanvas import MapCanvas
|
|
|
|
vp = Viewport(45.5019, -73.5674, 14, 120, 80)
|
|
frame = _Frame(canvas=MapCanvas(60, 20), viewport=vp)
|
|
cx, cy = _tile_local_of_view_centre(vp, 14, 4843, 5861)
|
|
# A zig-zag across the view gives several visible pieces to choose between.
|
|
ring = [(cx - 120 + i * 40, cy - 40 + (i % 2) * 80) for i in range(8)]
|
|
_add_line_label(frame, [ring], 4096, 14, 4843, 5861, "Rue Zig", ("#9aa0aa", False, 7))
|
|
|
|
assert frame.labels and frame.labels[0].alts, "no fallback anchors offered"
|
|
assert all(a != (frame.labels[0].x, frame.labels[0].y) for a in frame.labels[0].alts)
|
|
|
|
|
|
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_decoded_layers_survive_a_round_trip() -> None:
|
|
"""The sidecar encoding restores exactly what the decoder produced."""
|
|
from meshterm.core.mvt import dumps_layers, loads_layers
|
|
from meshterm.ui.map_render import DRAWN_LAYERS
|
|
|
|
original = decode_tile(_FIXTURE.read_bytes(), layers=DRAWN_LAYERS)
|
|
restored = loads_layers(dumps_layers(original, stamp="x"), stamp="x")
|
|
|
|
assert restored is not None
|
|
assert [(l.name, l.extent) for l in restored] == [(l.name, l.extent) for l in original]
|
|
for before, after in zip(original, restored):
|
|
assert after.features == before.features
|
|
|
|
|
|
def test_decoded_layers_refuse_a_blob_from_another_layer_set() -> None:
|
|
"""A narrowed blob must never be served to a caller that wants more of the tile."""
|
|
from meshterm.core.mvt import dumps_layers, loads_layers
|
|
|
|
blob = dumps_layers(decode_tile(_FIXTURE.read_bytes(), layers={"water"}), stamp="water")
|
|
assert loads_layers(blob, stamp="water") is not None
|
|
assert loads_layers(blob, stamp="water,place") is None
|
|
|
|
|
|
def test_decoded_layers_treat_damage_as_a_miss() -> None:
|
|
"""A truncated or foreign sidecar is a cache miss, never an exception."""
|
|
from meshterm.core.mvt import dumps_layers, loads_layers
|
|
|
|
blob = dumps_layers(decode_tile(_FIXTURE.read_bytes(), layers={"water"}), stamp="s")
|
|
assert loads_layers(blob[: len(blob) // 2], stamp="s") is None
|
|
assert loads_layers(b"", stamp="s") is None
|
|
assert loads_layers(b"not marshal at all", stamp="s") is None
|
|
|
|
|
|
def test_basemap_source_writes_and_reuses_a_decoded_sidecar(tmp_path: Path) -> None:
|
|
"""The second load parses nothing: it comes back from the sidecar."""
|
|
cache = tmp_path / "cache"
|
|
tile = cache / "tiles" / "14" / "4843" / "5861.pbf"
|
|
tile.parent.mkdir(parents=True)
|
|
tile.write_bytes(_FIXTURE.read_bytes())
|
|
|
|
src = _offline_source(cache)
|
|
first = src.load_tile(14, 4843, 5861)
|
|
sidecar = cache / "decoded" / "14" / "4843" / "5861.bin"
|
|
assert first is not None
|
|
assert sidecar.exists(), "decoding should have been written down"
|
|
|
|
# Break the raw tile: a second load that still works can only have used the sidecar.
|
|
tile.write_bytes(b"junk that cannot decode")
|
|
second = src.load_tile(14, 4843, 5861)
|
|
assert second is not None
|
|
assert [(l.name, len(l.features)) for l in second] == [
|
|
(l.name, len(l.features)) for l in first
|
|
]
|
|
|
|
|
|
def test_basemap_source_ignores_a_sidecar_from_another_layer_set(tmp_path: Path) -> None:
|
|
"""Widening what the map draws re-decodes rather than serving the narrower blob."""
|
|
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())
|
|
|
|
def source(layers):
|
|
return BasemapSource(
|
|
cache, tilejson_url="http://127.0.0.1:1/none", timeout=0.2, layers=layers
|
|
)
|
|
|
|
narrow = source(frozenset({"water"}))
|
|
assert narrow.load_tile(14, 4843, 5861) is not None
|
|
|
|
wider = source(frozenset({"water", "place"}))
|
|
layers = {layer.name: layer for layer in wider.load_tile(14, 4843, 5861) or []}
|
|
assert layers["place"].features, "the wider set must have been decoded afresh"
|
|
|
|
|
|
def test_basemap_source_prunes_the_decoded_cache_to_its_budget(tmp_path: Path) -> None:
|
|
"""The sidecar half of the cache is bounded; the raw tiles it derives from are not."""
|
|
from meshterm.services.basemap import BasemapSource
|
|
|
|
cache = tmp_path / "cache"
|
|
decoded = cache / "decoded" / "14" / "1"
|
|
decoded.mkdir(parents=True)
|
|
for i in range(6):
|
|
blob = decoded / f"{i}.bin"
|
|
blob.write_bytes(b"x" * 1000)
|
|
os.utime(blob, (i, i)) # oldest first
|
|
|
|
# 6 KB present, 3 KB allowed: the three oldest go, the three newest stay.
|
|
src = BasemapSource(cache, tilejson_url="http://127.0.0.1:1/none", max_decoded_bytes=3000)
|
|
src._prune_decoded()
|
|
|
|
left = sorted(p.stem for p in decoded.glob("*.bin"))
|
|
assert left == ["3", "4", "5"], "the oldest sidecars go first"
|
|
|
|
|
|
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
|
|
|
|
|
|
def _loaded_tile() -> list[Layer]:
|
|
"""A stand-in for a decoded tile: truthy, which is all the budget cares about."""
|
|
return [Layer(name="water", extent=4096)]
|
|
|
|
|
|
def _map_screen_with_tiles(count: int) -> "MapScreen":
|
|
"""A rendered map screen carrying ``count`` tiles' worth of pan history."""
|
|
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)
|
|
screen.render_body(80)
|
|
for i in range(count):
|
|
screen._tiles[(14, 9000 + i, 9000 + i)] = _loaded_tile()
|
|
return screen
|
|
|
|
|
|
def _budget(screen) -> int: # noqa: ANN001 - the screen's own arithmetic, mirrored
|
|
from meshterm.ui.map_screen import _MIN_TILE_CACHE, _TILE_CACHE_SCREENS
|
|
|
|
in_view = len(screen._viewport.tiles(14))
|
|
return max(_MIN_TILE_CACHE, in_view * _TILE_CACHE_SCREENS)
|
|
|
|
|
|
def test_map_tile_cache_stays_bounded_while_panning() -> None:
|
|
"""Panning far doesn't accumulate decoded tiles without limit.
|
|
|
|
One decoded tile is ~0.9 MB on the PicoCalc, which has ~100 MB in total, so an
|
|
unbounded cache turns a long pan into an out-of-memory kill.
|
|
"""
|
|
screen = _map_screen_with_tiles(200)
|
|
for _ in range(12):
|
|
screen.handle("right")
|
|
screen.render_body(80)
|
|
|
|
held = sum(1 for layers in screen._tiles.values() if layers)
|
|
assert held <= _budget(screen), f"{held} decoded tiles held, over budget"
|
|
|
|
|
|
def test_map_tile_cache_never_drops_what_is_on_screen() -> None:
|
|
"""Whatever the view needs survives the trim — eviction starts at the far end."""
|
|
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)
|
|
screen.render_body(80)
|
|
visible = list(screen._viewport.tiles(14))
|
|
for t in visible:
|
|
screen._tiles[t] = _loaded_tile()
|
|
for i in range(200): # a long pan's worth of history arriving after them
|
|
screen._tiles[(14, 9000 + i, 9000 + i)] = _loaded_tile()
|
|
|
|
screen.render_body(80)
|
|
|
|
for t in visible:
|
|
assert screen._tiles.get(t), f"{t} is on screen and must not have been evicted"
|
|
|
|
|
|
def test_map_tile_cache_keeps_the_absent_tile_markers() -> None:
|
|
"""A ``None`` is the memory of having asked; it costs a slot, so it is never evicted.
|
|
|
|
Dropping one would buy nothing back and cost a re-request on the very next repaint.
|
|
"""
|
|
screen = _map_screen_with_tiles(200)
|
|
for i in range(30):
|
|
screen._tiles[(14, 100 + i, 100)] = None
|
|
|
|
screen.render_body(80)
|
|
|
|
absent = [k for k, v in screen._tiles.items() if v is None]
|
|
assert len(absent) == 30, "absent-tile markers were evicted along with the geometry"
|
|
|
|
|
|
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))
|
|
# The basemap's state is a status atom, so it rides the title's `·` chain (standing in
|
|
# for the scale) and leaves the footer a pure key line inside its 72-cell budget.
|
|
assert screen.title.endswith("· offline")
|
|
assert "offline" not in screen.footer_hint
|
|
assert cell_len(screen.footer_hint) <= 72
|
|
|
|
|
|
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_its_three_destinations_and_the_zoom() -> 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 reframes — the shared paging labels would all be lies,
|
|
# and "end" is bound to nothing at all, so no jump rides the zoom rocker's Shift bank.
|
|
# The zoom pair rises to the right, like every right-hand directional pair.
|
|
assert [pair.label if pair else None for pair in lane] == [
|
|
"Region", "You", "Frame", "Zoom -", "Zoom +",
|
|
]
|
|
assert action_for(lane, 1) == "home" and action_for(lane, 5) == "pageup"
|
|
assert action_for(lane, 2) == "locate" and action_for(lane, 3) == "frame"
|
|
# Two slots carry a Shift half along their own axis: You + homes in on us zoomed,
|
|
# Clear drops the find query. The other three Shift keys stay unbound.
|
|
assert action_for(lane, 7) == "locate_zoom" and action_for(lane, 8) == "clear_find"
|
|
assert all(action_for(lane, n) is None for n in (6, 9, 10))
|
|
# Region and the zoom always act; the other two only where they'd land somewhere —
|
|
# and each Shift half gates with its own axis (no self marker, no query typed).
|
|
assert [pair.enabled for pair in lane] == [True, False, False, True, True]
|
|
assert lane[1].opp_enabled is False and lane[2].opp_enabled is False
|
|
|
|
|
|
def test_map_locate_recenters_on_our_own_node_at_the_current_zoom() -> None:
|
|
"""``You`` moves the view to us and changes nothing else — not even the zoom."""
|
|
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("Homestead", 45.40, -73.50, is_self=True),
|
|
]
|
|
screen = MapScreen(_StubSession(80, 24), markers, _StubSource(), 14)
|
|
screen.render_body(80)
|
|
screen.handle("pageup") # zoom in one step, so the recentre has a zoom to preserve
|
|
zoom = screen._viewport.zoom
|
|
screen.handle("left") # and pan away from wherever we are
|
|
|
|
assert screen.fkey_lane[1].enabled is True # we're on the map, so You can act
|
|
screen.handle("locate")
|
|
assert screen._viewport.center_lat == pytest.approx(45.40)
|
|
assert screen._viewport.center_lon == pytest.approx(-73.50)
|
|
assert screen._viewport.zoom == zoom
|
|
|
|
# A mesh we aren't located in has nowhere to go: the chip dims and the key no-ops.
|
|
away = MapScreen(_StubSession(80, 24), markers[:1], _StubSource(), 14)
|
|
away.render_body(80)
|
|
before = away._viewport
|
|
away.handle("locate")
|
|
assert away.fkey_lane[1].enabled is False and away._viewport is before
|
|
|
|
|
|
def test_map_echoes_the_find_query_in_the_body_only_where_the_footer_is_gone() -> None:
|
|
"""The PicoCalc draws no hint line, so the query it carries moves into the body."""
|
|
from meshterm.platforms import PICOCALC, REGULAR, set_platform
|
|
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)]
|
|
|
|
try:
|
|
set_platform(REGULAR)
|
|
desktop = MapScreen(_StubSession(72, 20), markers, _StubSource(), 14)
|
|
assert len(desktop.render_body(72)) == 20
|
|
for ch in "yul":
|
|
desktop.handle("text", ch)
|
|
# Unchanged: the footer already shows it, so the canvas keeps the whole body.
|
|
assert "/yul" not in _plain(desktop.render_body(72))
|
|
assert len(desktop.render_body(72)) == 20
|
|
assert "find: yul" in desktop.footer_hint
|
|
|
|
set_platform(PICOCALC)
|
|
device = MapScreen(_StubSession(53, 23), markers, _StubSource(), 14)
|
|
assert len(device.render_body(53)) == 23
|
|
for ch in "yul":
|
|
device.handle("text", ch)
|
|
lines = device.render_body(53)
|
|
# The echo overlays the canvas's *last* row — directly above the F-key lane —
|
|
# so the ground never shifts: same body height, same viewport, top anchor held.
|
|
assert _plain(lines[-1]).strip() == "/yul"
|
|
assert "/yul" not in _plain(lines[0])
|
|
assert len(lines) == 23
|
|
# Clearing the find hands the row back to the canvas, still without a reflow.
|
|
for _ in "yul":
|
|
device.handle("backspace")
|
|
assert len(device.render_body(53)) == 23
|
|
assert "/" not in _plain(device.render_body(53))
|
|
finally:
|
|
set_platform(REGULAR)
|
|
|
|
|
|
def test_map_frame_chip_lights_only_with_matches_to_frame() -> None:
|
|
"""``Frame`` is the find's Enter under another name — dim, and inert, without a query."""
|
|
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),
|
|
]
|
|
screen = MapScreen(_StubSession(80, 24), markers, _StubSource(), 14)
|
|
screen.render_body(80)
|
|
before = screen._viewport
|
|
screen.handle("frame") # no query typed: nothing to frame, so nothing moves
|
|
assert screen.fkey_lane[2].enabled is False and screen._viewport is before
|
|
|
|
for ch in "ali":
|
|
screen.handle("text", ch)
|
|
assert screen.fkey_lane[2].enabled is True
|
|
screen.handle("frame")
|
|
assert screen._viewport.center_lat == pytest.approx(45.40)
|
|
|
|
# A query nothing matches has no extent either — the chip goes back to dim.
|
|
for ch in "zzz":
|
|
screen.handle("text", ch)
|
|
assert screen._matches() == [] and screen.fkey_lane[2].enabled is False
|
|
|
|
|
|
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)
|