Jean-Pierre MartineauandClaude Opus 5
7b3fdff84e
The feed was the one screen in the app with no Back row. Esc worked, but the exit group is not a shortcut — it is the app's standard closing furniture, and a screen that lists rows and offers none reads as a screen you have to already know how to leave. So the cursor gains a stop past the oldest packet: a blank separator and the bare word Back, exactly the shape back_rows() gives a select list, drawn by hand because the feed is a rendered screen rather than a list of choices. It is pinned chrome, struck from the viewport before the feed windows into what is left — so it sits at the foot of the screen however far the packets have scrolled, and the body still never scrolls as a whole. Enter on it resolves what Esc resolves. That makes the cursor one index over len(feed) + 1 stops, which collapses three special cases into a clamp: the "nothing highlighted" state the page keys used to slide the bare window in is now simply "the cursor is on Back", and an empty feed opens with the only stop it has already highlighted. End lands on the list's last row, as it does on every other list — the oldest packet is one ↑ up from there. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
427 lines
18 KiB
Python
427 lines
18 KiB
Python
"""Live feed tests: the promoted feed screen's state, rows, and windowing.
|
||
|
||
The screen is driven headless against fake sessions and canned observations — the
|
||
same approach as the dashboard tests, whose feed panel this screen grew out of.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
from datetime import timedelta
|
||
|
||
from rich.cells import cell_len
|
||
|
||
from meshterm.core.events import MeshEvent
|
||
from meshterm.core.models import Ack, Message, Observation, utcnow
|
||
from meshterm.ui.livefeed_screen import _ICON_LANE, LiveFeedScreen
|
||
|
||
|
||
class _FakeSession:
|
||
def __init__(self) -> None:
|
||
self.repaints = 0
|
||
|
||
def invalidate(self) -> None:
|
||
self.repaints += 1
|
||
|
||
|
||
class _Fut:
|
||
"""A minimal future stand-in so the screen can resolve without an event loop."""
|
||
|
||
def __init__(self) -> None:
|
||
self.value = None
|
||
self._done = False
|
||
|
||
def done(self) -> bool:
|
||
return self._done
|
||
|
||
def set_result(self, value) -> None:
|
||
self.value = value
|
||
self._done = True
|
||
|
||
|
||
#: Nodes the feed's resolver can name, by the hash a frame addresses them with (one byte)
|
||
#: and by the wider prefixes an advert carries.
|
||
_KNOWN = {"a1b2": "Alice", "3d63": "YUL", "a1": "Alice", "3d": "YUL", "c0": "Us"}
|
||
|
||
|
||
def _screen(seed=None, active=True, **kwargs) -> LiveFeedScreen:
|
||
screen = LiveFeedScreen(
|
||
session=_FakeSession(),
|
||
resolve=lambda h: _KNOWN.get(h, ""),
|
||
seed=list(seed or []),
|
||
hub_active=lambda: active,
|
||
**kwargs,
|
||
)
|
||
screen.note_viewport(48) # the frame records this before every real paint
|
||
return screen
|
||
|
||
|
||
def _obs(node="a1b2", kind="advert", snr=5.0, rssi=-90.0, age_s=0, **extra) -> Observation:
|
||
return Observation(
|
||
node=node, name=node, kind=kind, snr=snr, rssi=rssi,
|
||
observed_at=utcnow() - timedelta(seconds=age_s), **extra,
|
||
)
|
||
|
||
|
||
def _plain(lines: list[str]) -> str:
|
||
return "\n".join(lines)
|
||
|
||
|
||
def _stripped(lines: list[str]) -> list[str]:
|
||
"""The rendered lines with ANSI escapes removed, for structural assertions."""
|
||
return [re.sub(r"\x1b\[[0-9;]*m", "", line) for line in lines]
|
||
|
||
|
||
def _rows(screen: LiveFeedScreen, width: int) -> list[str]:
|
||
"""Just the packet rows — the body past its chrome, above the pinned exit group."""
|
||
return _stripped(screen.render_body(width))[2:-2]
|
||
|
||
|
||
def _col(line: str, needle: str) -> int:
|
||
"""The display column (cells, not characters) ``needle`` starts at within ``line``."""
|
||
return cell_len(line[: line.index(needle)])
|
||
|
||
|
||
def test_livefeed_tool_registers_under_the_dashboard() -> None:
|
||
"""The livefeed tool lands in the Mesh section, right below the dashboard."""
|
||
from meshterm.tools import load_all_tools
|
||
from meshterm.tools.base import all_tools
|
||
|
||
load_all_tools()
|
||
tools = {t.name: t for t in all_tools()}
|
||
tool = tools["livefeed"]
|
||
assert tool.title == "Live feed" and tool.category == "Mesh"
|
||
assert tools["dashboard"].order < tool.order < tools["contacts"].order
|
||
|
||
|
||
def test_livefeed_live_events_land_in_the_feed() -> None:
|
||
"""Observations, messages, and acks each fold into the feed as they arrive."""
|
||
screen = _screen()
|
||
screen.on_event(MeshEvent.observation_event(_obs(snr=7.5)))
|
||
screen.on_event(MeshEvent.message_event(Message(text="hi", sender="a1b2")))
|
||
screen.on_event(MeshEvent.ack_event(Ack(code="01c3")))
|
||
body = _plain(screen.render_body(100))
|
||
assert "+7.5 dB" in body
|
||
assert "message" in body and "ack" in body
|
||
# Newest first: the ack row sits above the message row, which sits above the advert.
|
||
assert body.index("ack") < body.index("message") < body.index("advert")
|
||
|
||
|
||
def test_livefeed_rows_leave_the_relay_path_to_the_viewer() -> None:
|
||
"""A row says what arrived and how well it was heard; the route is Enter's job.
|
||
|
||
Whatever the fixed lanes left a route was never enough to draw one in — it arrived
|
||
elided to a stub — so the feed carries none, and the whole row fits a 72-column
|
||
screen with its class label intact.
|
||
"""
|
||
screen = _screen()
|
||
screen.on_event(
|
||
MeshEvent.observation_event(_obs(kind="packet", path="3d63,a1b2", snr=1.0))
|
||
)
|
||
row = _rows(screen, 72)[0]
|
||
assert "via" not in row and "YUL" not in row # no route, not even a stub of one
|
||
assert "📦 packet" in row # the class still reads in words at 72
|
||
assert "+1.0 dB" in row and "-90 dBm" in row # …as does the reception it was heard at
|
||
assert len(row.rstrip()) <= 72
|
||
|
||
|
||
def test_livefeed_class_lane_names_the_payload_class_once() -> None:
|
||
"""A raw frame is filed under what it *is*, and the subject lane doesn't repeat it.
|
||
|
||
``packet`` names only the event family the frame arrived in, so the class lane
|
||
reads its parsed payload class — the very label the viewer's card headlines. The
|
||
subject lane is then free to be about the packet's own subject, and a class that
|
||
carries nothing to be about says so with a dash rather than standing the class in a
|
||
second time.
|
||
"""
|
||
screen = _screen()
|
||
raw = {"payload_typename": "MULTIPART", "route_typename": "FLOOD"}
|
||
screen.on_event(
|
||
MeshEvent.observation_event(
|
||
_obs(node="", kind="packet", snr=1.0, path="3d63,a1b2", raw=raw)
|
||
)
|
||
)
|
||
row = _rows(screen, 100)[0]
|
||
assert "🧩 multipart" in row # the class lane, under the payload class's own icon
|
||
assert row.count("multipart") == 1 # said once, not once per lane
|
||
assert "packet" not in row # …and never as the generic event family
|
||
assert "—" in row # the subject lane: the class is about nothing
|
||
assert "?" not in row # …but never the useless placeholder
|
||
|
||
|
||
def test_livefeed_names_a_channel_message_by_its_sender() -> None:
|
||
"""A channel message's ``Name:`` prefix names the subject lane, not the bare channel."""
|
||
screen = _screen()
|
||
screen.on_event(
|
||
MeshEvent.message_event(Message(text="Alice: hi all", channel=3, is_channel=True))
|
||
)
|
||
body = _plain(screen.render_body(100))
|
||
assert "Alice" in body # the parsed sender leads the row
|
||
assert "ch 3" in body # …with the channel kept as the trailing context
|
||
|
||
|
||
def test_livefeed_unsigned_channel_post_is_filed_under_its_channel() -> None:
|
||
"""With nobody signing it, a channel message is about the channel — named, once."""
|
||
screen = _screen(channel_names={3: "Alerts"})
|
||
screen.on_event(
|
||
MeshEvent.message_event(Message(text="beep boop", channel=3, is_channel=True))
|
||
)
|
||
row = _rows(screen, 100)[0]
|
||
assert "Alerts" in row # the channel name, not its slot number
|
||
assert row.count("Alerts") == 1 # …and the trailing note doesn't say it again
|
||
assert "ch 3" not in row
|
||
|
||
|
||
def test_livefeed_channel_frame_is_about_its_channel() -> None:
|
||
"""An overheard channel frame is filed under the channel whose key confirms its MAC."""
|
||
from Crypto.Hash import HMAC, SHA256
|
||
|
||
from meshterm.core.channels import channel_hash, derive_secret
|
||
|
||
secret = derive_secret("Public")
|
||
crypted = bytes(range(16))
|
||
mac = HMAC.new(secret, digestmod=SHA256)
|
||
mac.update(crypted)
|
||
raw = {
|
||
"payload_typename": "GRP_TXT", "chan_hash": channel_hash(secret),
|
||
"cipher_mac": mac.digest()[:2].hex(), "crypted": crypted.hex(),
|
||
}
|
||
screen = _screen(channels=[("Public", secret)])
|
||
screen.on_event(MeshEvent.observation_event(_obs(node="", kind="packet", raw=raw)))
|
||
assert "Public" in _rows(screen, 100)[0]
|
||
|
||
# A channel we hold no key for can only be named by the fingerprint it advertised —
|
||
# never by a name, since a fingerprint alone is not proof of which channel it is.
|
||
other = _screen(channels=[("Public", secret)])
|
||
other.on_event(
|
||
MeshEvent.observation_event(_obs(node="", kind="packet", raw={**raw, "chan_hash": "a3"}))
|
||
)
|
||
row = _rows(other, 100)[0]
|
||
assert "hash a3" in row and "Public" not in row
|
||
|
||
|
||
def test_livefeed_addressed_frame_is_about_its_two_ends() -> None:
|
||
"""A frame that names a recipient reads as sender → recipient, resolved to names."""
|
||
screen = _screen(self_name="Us")
|
||
raw = {"payload_typename": "TEXT_MSG", "dest_hash": "c0", "src_hash": "a1"}
|
||
screen.on_event(MeshEvent.observation_event(_obs(node="", kind="packet", raw=raw)))
|
||
row = _rows(screen, 100)[0]
|
||
assert "📩 direct message" in row
|
||
assert "Alice → Us" in row # …including us, when a frame is addressed to us
|
||
|
||
# A pair too wide for the lane spends its cells on the addressee: the sender drops to
|
||
# the hash it was named by rather than the recipient being the half cut off.
|
||
wide = _screen()
|
||
wide._resolve = lambda h: {"a1": "Alice-With-A-Long-Name", "3d": "YUL-Cartierville"}.get(h, "")
|
||
wide.on_event(
|
||
MeshEvent.observation_event(
|
||
_obs(node="", kind="packet",
|
||
raw={"payload_typename": "REQ", "dest_hash": "3d", "src_hash": "a1"})
|
||
)
|
||
)
|
||
assert "a1 → YUL-Cartierv" in _rows(wide, 100)[0]
|
||
|
||
|
||
def test_livefeed_tokened_classes_are_about_their_token() -> None:
|
||
"""An ack names the message it answers; a trace names its tag. Neither says "packet"."""
|
||
screen = _screen()
|
||
screen.on_event(
|
||
MeshEvent.observation_event(
|
||
_obs(node="", kind="packet", age_s=1,
|
||
raw={"payload_typename": "TRACE", "trace_tag": "5f3c2a10"})
|
||
)
|
||
)
|
||
screen.on_event(
|
||
MeshEvent.observation_event(
|
||
_obs(node="", kind="packet",
|
||
raw={"payload_typename": "ACK", "ack_crc": "9b71e004"})
|
||
)
|
||
)
|
||
acked, traced = _rows(screen, 100)[:2]
|
||
assert "for 9b71e004" in acked
|
||
assert "tag 5f3c2a10" in traced
|
||
|
||
|
||
def test_livefeed_column_header_sits_over_the_lanes_it_names() -> None:
|
||
"""The header names each lane, and every label lands on the values beneath it."""
|
||
screen = _screen(seed=[_obs(node="3d63", kind="telemetry", snr=12.8, rssi=-105.0)])
|
||
header, row = _stripped(screen.render_body(72))[1:3]
|
||
assert header.split() == ["TIME", "CLASS", "SUBJECT", "SNR", "RSSI"]
|
||
# Measured in display cells, not characters — the class icon is one character wide
|
||
# but two cells, so a character index would report every later lane one column early.
|
||
# The clock is read off the row itself: a literal needle would only match during the
|
||
# minute it was written in.
|
||
stamp = re.search(r"\d\d:\d\d:\d\d", row)
|
||
assert stamp is not None
|
||
assert _col(header, "TIME") == _col(row, stamp.group(0))
|
||
assert _col(header, "CLASS") == _col(row, "telemetry") - _ICON_LANE
|
||
assert _col(header, "SUBJECT") == _col(row, "YUL")
|
||
# The two readings right-align their number, so their labels end where the digits do.
|
||
assert _col(header, "SNR") + 3 == _col(row, "+12.8") + 5
|
||
assert _col(header, "RSSI") + 4 == _col(row, "-105") + 4
|
||
|
||
|
||
def test_livefeed_column_header_is_pinned_and_carries_no_sort_cue() -> None:
|
||
"""The header never scrolls away, and it advertises no sort — the feed has one order."""
|
||
screen = _screen(seed=[_obs(node=f"n{i}", age_s=i) for i in range(40)])
|
||
screen.note_viewport(12)
|
||
lines = _stripped(screen.render_body(100))
|
||
assert "TIME" in lines[1]
|
||
assert not any(mark in lines[1] for mark in ("▲", "▼")) # nothing here sorts
|
||
screen.handle("end") # the list's last row — Back
|
||
screen.handle("up") # …and one up from it, the oldest packet: the window scrolls
|
||
after = _stripped(screen.render_body(100))
|
||
assert after[1] == lines[1] # …and the header is exactly where it was
|
||
assert "↑" in after[2] and "more" in after[2] # the rows really did travel
|
||
|
||
|
||
def test_livefeed_highlight_uses_the_app_wide_cursor() -> None:
|
||
"""The feed marks its row like every other list: a ``❯`` pointer over a brand row."""
|
||
screen = _screen(seed=[_obs(node="n0", age_s=1), _obs(node="n1", age_s=0)])
|
||
rows = _rows(screen, 100)
|
||
assert rows[0].startswith("❯ ") and rows[1].startswith(" ")
|
||
assert "▸" not in "\n".join(rows) # the old odd-one-out mark is gone
|
||
raw = screen.render_body(100)[2]
|
||
assert raw.startswith("\x1b[") # the pointer carries the brand style, not bare text
|
||
|
||
|
||
def test_livefeed_page_keys_move_the_feed_selection() -> None:
|
||
"""PgUp/PgDn walk the cursor a windowful at a time, clamped to the list's two ends."""
|
||
seed = [_obs(node=f"n{i}", age_s=i) for i in range(20)]
|
||
screen = _screen(seed=seed)
|
||
screen._feed_window.page = 5 # as if the last paint settled a five-row window
|
||
assert screen._selected == 0 # the newest packet is highlighted from the start
|
||
screen.handle("pagedown")
|
||
assert screen._selected == 5 # the selection travelled down a page, not just the view
|
||
screen.handle("pageup")
|
||
assert screen._selected == 0
|
||
screen.handle("pageup")
|
||
assert screen._selected == 0 # …and stops at the newest rather than wrapping
|
||
|
||
for _ in range(5): # paged past the oldest packet, the cursor lands on Back and stays
|
||
screen.handle("pagedown")
|
||
assert screen._selected is None
|
||
|
||
|
||
def test_livefeed_ends_in_the_apps_exit_group() -> None:
|
||
"""The screen closes with the standard exit group, pinned to the foot of the body."""
|
||
seed = [_obs(node=f"n{i}", age_s=i) for i in range(40)]
|
||
screen = _screen(seed=seed)
|
||
screen.note_viewport(24)
|
||
lines = _stripped(screen.render_body(100))
|
||
assert len(lines) <= 24 # the group is chrome: it fits, it doesn't grow
|
||
assert lines[-2].strip() == "" # exactly one blank separator above it…
|
||
assert lines[-1].strip() == "Back" # …then the bare word, no arrow and no icon
|
||
assert "↓" in lines[-3] and "more" in lines[-3] # the feed windows above it
|
||
|
||
# It stays put as the feed scrolls: the packets travel under it, it does not.
|
||
screen.handle("end")
|
||
screen.handle("up") # onto the oldest packet, so the window really has scrolled
|
||
after = _stripped(screen.render_body(100))
|
||
assert "↑" in after[2] and "more" in after[2]
|
||
assert after[-1].strip() == "Back" and after[-2].strip() == ""
|
||
|
||
|
||
def test_livefeed_back_row_is_the_cursors_last_stop() -> None:
|
||
"""↓ off the oldest packet lands on Back, End jumps to it, and Enter there leaves."""
|
||
screen = _screen(seed=[_obs(node="n0", age_s=1), _obs(node="n1", age_s=0)])
|
||
screen.future = _Fut()
|
||
|
||
screen.handle("down")
|
||
screen.handle("down")
|
||
assert screen._selected is None # past the oldest packet: Back
|
||
assert _stripped(screen.render_body(100))[-1].startswith("❯ Back")
|
||
screen.handle("down")
|
||
assert screen._selected is None # …and the cursor stops there
|
||
screen.handle("up")
|
||
assert screen._selected == 1 # back onto the oldest packet
|
||
screen.handle("end")
|
||
assert screen._selected is None # End is the list's last row
|
||
|
||
screen.handle("enter") # Enter on Back leaves…
|
||
assert screen.future.done() and screen.future.value is None # …exactly as Esc does
|
||
|
||
|
||
def test_livefeed_empty_feed_still_offers_a_way_out() -> None:
|
||
"""With nothing heard, Back is the only stop there is — and it is already the cursor's."""
|
||
screen = _screen()
|
||
lines = _stripped(screen.render_body(100))
|
||
assert "nothing heard yet" in lines[1]
|
||
assert lines[-1].startswith("❯ Back")
|
||
screen.handle("up") # nowhere else to go
|
||
assert screen._selected is None and _stripped(screen.render_body(100))[-1].startswith("❯ ")
|
||
|
||
|
||
def test_livefeed_windows_inside_the_fixed_screen() -> None:
|
||
"""The heading stays pinned: the body fits the viewport and the feed rows window."""
|
||
seed = [_obs(node=f"n{i}", age_s=i) for i in range(40)]
|
||
screen = _screen(seed=seed)
|
||
screen.note_viewport(24)
|
||
lines = screen.render_body(100)
|
||
assert len(lines) <= 24 # heading + feed window == the viewport, never more
|
||
body = _plain(_stripped(lines))
|
||
assert "newest first" in body # the pinned status line is still there
|
||
assert "↓" in body and "more" in body # hidden feed rows are counted below
|
||
|
||
|
||
#: A terminal too narrow to hold the feed's fixed lanes — the only thing that makes a
|
||
#: row overflow now that no route rides in it (see ``_FEED_LABEL_MIN_WIDTH``).
|
||
_CRAMPED = 44
|
||
|
||
|
||
def test_livefeed_the_whole_row_fits_a_72_column_screen() -> None:
|
||
"""No row overflows at 72 — which is why ←→ has nothing to advertise there."""
|
||
screen = _screen(seed=[_obs(node="n0", kind="telemetry")])
|
||
for line in _stripped(screen.render_body(72))[1:]:
|
||
assert len(line.rstrip()) <= 72
|
||
assert screen._hmax == 0
|
||
assert "←→" not in screen.footer_hint
|
||
|
||
|
||
def test_livefeed_highlighted_row_scrolls_sideways_to_its_tail() -> None:
|
||
"""On a terminal too narrow for the lanes, ←→ slide the highlighted row to its end."""
|
||
screen = _screen(seed=[_obs()])
|
||
opening = _rows(screen, _CRAMPED)[0]
|
||
assert screen._hmax > 0 # the row does run past the right edge
|
||
assert not opening.rstrip().endswith("dBm") # …so its reception tail is off-screen
|
||
|
||
for _ in range(40): # → saturates at the row's own end, never past it
|
||
screen.handle("right")
|
||
scrolled = _rows(screen, _CRAMPED)[0]
|
||
assert screen._hshift == screen._hmax
|
||
assert scrolled.startswith("❯ ") # the pointer lane stays pinned while the row slides
|
||
assert scrolled.rstrip().endswith("-90 dBm") # the tail is now readable
|
||
assert opening[2:10] not in scrolled # …at the cost of the time lane, slid off left
|
||
|
||
for _ in range(40):
|
||
screen.handle("left")
|
||
assert screen._hshift == 0
|
||
assert _rows(screen, _CRAMPED)[0] == opening # back where it started
|
||
|
||
|
||
def test_livefeed_moving_the_selection_abandons_the_rows_scroll() -> None:
|
||
"""Each row scrolls on its own: landing on another packet starts it at its beginning."""
|
||
screen = _screen(seed=[_obs(node="n0", age_s=1), _obs(node="n1", age_s=0)])
|
||
screen.render_body(_CRAMPED)
|
||
screen.handle("right")
|
||
assert screen._hshift > 0
|
||
screen.handle("down")
|
||
assert screen._hshift == 0
|
||
|
||
|
||
def test_livefeed_advertises_line_scroll_only_where_it_acts() -> None:
|
||
"""←→ earns its footer atom exactly while the highlighted row overflows."""
|
||
screen = _screen(seed=[_obs()])
|
||
screen.render_body(100)
|
||
assert "←→" not in screen.footer_hint # a row with room to spare scrolls nowhere
|
||
|
||
screen.render_body(_CRAMPED)
|
||
assert "←→ scroll line" in screen.footer_hint
|
||
assert len(screen.footer_hint) <= 72 # the screens-at-72 rule, fullest state
|
||
|
||
|
||
def test_livefeed_without_a_device_reads_as_waiting() -> None:
|
||
"""With the hub idle the heading says so instead of pretending to be live."""
|
||
screen = _screen(active=False)
|
||
assert "○ waiting for a device" in _plain(screen.render_body(100))
|