Jean-Pierre MartineauandClaude Fable 5
2d6e20ba2e
The identity block was eating the stage's vertical budget on every tab. The pinned chrome shrinks to the one-line identity header + tab strip; the vitals rows (key, heard, packets, signal, where), the location preview, and the Open full map / Time machine actions all move into a new Info tab (replacing the Map tab — the preview now grows into whatever rows the tab spares). The Routes tab spends the freed lines on its graph and route list, and the strip costs a single line of air (the graph's own top padding provides it). Opening the full map from the page now also seeds the map's find filter with the node's name, so it opens centred on the node *and* lit among the rest — exactly as if the user had typed the find. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
808 lines
33 KiB
Python
808 lines
33 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"
|
|
_ANSI = re.compile(r"\x1b\[[0-9;]*m")
|
|
|
|
|
|
def _plain(lines: list[str]) -> str:
|
|
"""Strip ANSI colour from rendered lines for content assertions."""
|
|
return "\n".join(_ANSI.sub("", line) for line in lines)
|
|
|
|
|
|
# -- 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"") == []
|
|
|
|
|
|
# -- 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)
|
|
|
|
|
|
# -- 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_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)
|