rich/MeshTerm
1
0 Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity
Files
8dff5bbb29ef20a5ca9f09f4930b2bf14860476a
MeshTerm/tests/test_chat.py
T
Jean-Pierre MartineauandClaude Haiku 5.5 8dff5bbb29 The selected chat message whitens its author chip and its unknown mentions 
The author chip of the selected message's group now turns its label white,
as the message's text does, and goes back when the selection leaves the
group. A mention of an unknown node turns white too: its grey fill already
says the node is unknown, and white reads better on it than black.

Co-Authored-By: Claude Haiku 5.5 <noreply@anthropic.com>
2026-10-19 23:59:45 -04:00

3498 lines
146 KiB
Python
Raw Blame History

This file contains invisible Unicode characters
This file contains invisible Unicode characters that are indistinguishable to humans but may be processed differently by a computer. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
# SPDX-License-Identifier: Apache-2.0
"""Tests for the chat feature: device send, persistence, and the chat service.
All the tests run against the :class:`MockDevice` simulator and a temporary database. They
do not use hardware.
"""
from __future__ import annotations
import asyncio
import logging
import re
from pathlib import Path
import pytest
from rich.cells import cell_len
from meshterm.core.channels import DEFAULT_PUBLIC_SECRET, derive_secret
from meshterm.core.config import Settings
from meshterm.core.connection import MockDevice
from meshterm.core.events import MeshEvent
from meshterm.core.models import (
ChatMessage,
Contact,
Conversation,
MapMarker,
Message,
conversation_key,
)
from meshterm.core.preferences import Preferences
from meshterm.persistence.repository import Repository
from meshterm.services.chat_service import ChatService
from meshterm.services.event_hub import EventHub
from meshterm.tools.chat import _lanes, _LiveLasts, _preview_text, _title
from meshterm.ui.chat import ChatScreen
from meshterm.ui.tui import fkeys
from meshterm.ui.tui.screen import CANCEL
from tests.conftest import plain as _strip_ansi # the only strip-and-join screen reader
class _StubSession:
"""A session stand-in that counts the requests for a paint."""
def __init__(self) -> None:
self.invalidations = 0
def invalidate(self) -> None:
self.invalidations += 1
def run_detached(self, work): # noqa: ANN001, ANN201
"""Start the flow of a key handler as a task, as ``TuiSession.run_detached`` does."""
return asyncio.ensure_future(work)
class _StubContext:
"""Minimal :class:`~meshterm.context.AppContext` stand-in for chat tests."""
def __init__(self, device: MockDevice, repo: Repository) -> None:
self._device = device
self.repo = repo
self.log = logging.getLogger("test.chat")
self.profile_name = None
self.settings = Settings()
self.preferences = Preferences()
self.events = EventHub(self)
async def device(self) -> MockDevice:
await self._device.connect()
return self._device
@pytest.fixture()
def repo(tmp_path: Path) -> Repository:
"""A Repository on a scratch database. It closes when the test ends."""
r = Repository(tmp_path / "chat.db")
yield r
r.close()
# -- domain models ------------------------------------------------------------
def test_conversation_key_distinguishes_channels_and_directs() -> None:
"""A channel key uses the identity of the channel. A direct key ignores the case of the peer."""
assert conversation_key(True, "deadbeef", None) == "chan:deadbeef"
assert conversation_key(False, None, "AABBCC") == "dm:aabbcc"
contact = Contact(name="Alice", public_key="d4e5" + "0" * 60, key_prefix="d4e5f6a7")
conv = Conversation(label="Alice", is_channel=False, contact=contact)
assert conv.key == "dm:d4e5f6a7"
assert conv.peer == "d4e5f6a7"
chan = Conversation(label="#public", is_channel=True, channel_idx=2, channel_id="deadbeef")
assert chan.key == "chan:deadbeef" # keyed by identity, not by the slot index
assert chan.peer is None
def test_chat_message_from_inbound_message() -> None:
"""An inbound direct Message becomes a received ChatMessage on the correct thread."""
message = Message(text="hi", sender="a1b2c3d4", is_channel=False, snr=5.5)
chat = ChatMessage.from_message(message, peer_name="Yagi")
assert chat.outbound is False
assert chat.is_channel is False
assert chat.peer == "a1b2c3d4"
assert chat.peer_name == "Yagi"
assert chat.snr == 5.5
assert chat.key == "dm:a1b2c3d4"
# -- device send --------------------------------------------------------------
async def test_mock_send_direct_returns_ack() -> None:
"""The simulator acknowledges direct messages, so they show as delivered."""
device = MockDevice()
await device.connect()
contact = (await device.get_contacts())[0]
delivery = await device.send_direct_message(contact, "hello")
assert delivery.acked and delivery.code
await device.send_channel_message(0, "hi channel") # no return value, and it must not raise
await device.disconnect()
async def test_sending_to_a_contact_the_device_forgot_is_refused_by_name() -> None:
"""A recipient that the radio has no entry for is refused in words that name the contact.
MeshTerm lists the contacts of the device and the contacts that it remembers for that
device. Thus a contact that the firmware removed still shows in the list, and only the
send finds the problem. The rejection must carry the contact, so that the chat can offer
to write it back.
"""
from meshterm.core.connection import ContactNotOnDeviceError
device = MockDevice()
await device.connect()
stranger = Contact(name="bob_caribou", public_key="a3" * 32, key_prefix="a3a3a3a3")
with pytest.raises(ContactNotOnDeviceError) as caught:
await device.send_direct_message(stranger, "hello")
assert caught.value.contact is stranger
assert "bob_caribou" in str(caught.value)
assert "error_code" not in str(caught.value) # not the payload from the wire
await device.disconnect()
async def test_adding_a_forgotten_contact_back_makes_it_messageable() -> None:
"""To write the contact to the device is the whole repair. Then the same send succeeds."""
device = MockDevice()
await device.connect()
stranger = Contact(name="bob_caribou", public_key="a3" * 32, key_prefix="a3a3a3a3")
await device.add_contact(stranger)
assert "bob_caribou" in {c.name for c in await device.get_contacts()}
assert (await device.send_direct_message(stranger, "hello")).acked
await device.disconnect()
async def test_restore_contact_writes_only_when_the_dialog_is_accepted() -> None:
"""The device write is offered, not silent. If the user declines, the table does not change."""
from meshterm.core.connection import ContactNotOnDeviceError
from meshterm.ui.chat import _restore_contact
class _Ui:
def __init__(self, answer: bool) -> None:
self.answer = answer
self.prompt = ""
async def dialog(self, prompt, buttons, **kwargs): # noqa: ANN001, ANN003
self.prompt = prompt
return self.answer
class _Devstate:
def __init__(self) -> None:
self.invalidated = 0
def invalidate_contacts(self) -> None:
self.invalidated += 1
class _Ctx:
def __init__(self, device: MockDevice, answer: bool) -> None:
self._device = device
self.ui = _Ui(answer)
self.devstate = _Devstate()
async def device(self) -> MockDevice:
return self._device
stranger = Contact(name="bob_caribou", public_key="a3" * 32, key_prefix="a3a3a3a3")
missing = ContactNotOnDeviceError(stranger)
device = MockDevice()
await device.connect()
before = len(await device.get_contacts())
declined = _Ctx(device, answer=False)
assert await _restore_contact(declined, missing) is False
assert len(await device.get_contacts()) == before
assert declined.devstate.invalidated == 0
assert "bob_caribou" in declined.ui.prompt
accepted = _Ctx(device, answer=True)
assert await _restore_contact(accepted, missing) is True
assert len(await device.get_contacts()) == before + 1
assert accepted.devstate.invalidated == 1
await device.disconnect()
async def test_a_declined_restore_lets_the_rejection_stand() -> None:
"""The send tries again if the contact is back. If not, the send reports the refusal."""
from meshterm.core.connection import ContactNotOnDeviceError
from meshterm.ui.chat import _with_restore
stranger = Contact(name="bob_caribou", public_key="a3" * 32)
tries = 0
async def attempt() -> str:
nonlocal tries
tries += 1
if tries == 1:
raise ContactNotOnDeviceError(stranger)
return "sent"
class _Ctx:
def __init__(self, restored: bool) -> None:
self.restored = restored
async def restore(ctx, missing) -> bool: # noqa: ANN001
return ctx.restored
import meshterm.ui.chat as chat_module
original = chat_module._restore_contact
chat_module._restore_contact = restore
try:
with pytest.raises(ContactNotOnDeviceError):
await _with_restore(_Ctx(restored=False), attempt)
assert tries == 1 # declined: no second transmission
tries = 0
assert await _with_restore(_Ctx(restored=True), attempt) == "sent"
assert tries == 2 # the same send runs again after the contact is written back
finally:
chat_module._restore_contact = original
def test_a_rejected_command_is_explained_in_words_not_wire_codes() -> None:
"""An error code becomes a sentence. Only an unknown rejection shows its payload."""
from meshterm.core.connection import reject_reason
class _Ev:
def __init__(self, payload: dict) -> None:
self.payload = payload
assert reject_reason(_Ev({"error_code": 2})) == "the device has no contact with that key"
assert reject_reason(_Ev({"error_code": 3})) == "the device's table is full"
assert reject_reason(None) == "the device didn't answer"
assert "97" in reject_reason(_Ev({"error_code": 97}))
async def test_add_contact_writes_a_flood_record_with_the_name_the_field_fits() -> None:
"""The record for the firmware has no learned route and a name that is cut to fit in bytes."""
from meshterm.core.connection import MeshCoreDevice
class _Ok:
payload: dict = {}
def is_error(self) -> bool:
return False
class _FakeContacts:
def __init__(self) -> None:
self.record = None
async def add_contact(self, record): # noqa: ANN001
self.record = record
return _Ok()
class _FakeMc:
def __init__(self) -> None:
self.commands = _FakeContacts()
device = MeshCoreDevice.__new__(MeshCoreDevice)
mc = _FakeMc()
device._require = lambda: mc # type: ignore[method-assign]
long_name = "é" * 40 # 80 bytes: twice the size of the name field of the firmware
await device.add_contact(Contact(name=long_name, public_key="a3" * 32, node_type=1))
record = mc.commands.record
assert record["out_path_len"] == -1 and record["out_path"] == ""
assert len(record["adv_name"].encode("utf-8")) == 32
assert record["public_key"] == "a3" * 32
async def test_message_pump_drains_until_empty() -> None:
"""The RX pump calls get_msg() until the queue is empty (the pull model).
MeshCore never pushes message bodies, so the client must pull them. Without this, a
send works but the client receives nothing. The test guards these parts: the immediate
drain, the loop until NO_MORE_MSGS, the MESSAGES_WAITING subscription, and the clean
teardown.
"""
from meshcore import EventType
from meshterm.core.connection import MeshCoreDevice
class _Ev:
def __init__(self, t) -> None: # noqa: ANN001
self.type = t
class _FakeCommands:
def __init__(self, script: list) -> None:
self.calls = 0
self._script = script
async def get_msg(self, timeout=None): # noqa: ANN001, ANN201
i = min(self.calls, len(self._script) - 1)
self.calls += 1
return _Ev(self._script[i])
class _FakeMC:
def __init__(self, script: list) -> None:
self.commands = _FakeCommands(script)
self.subs: list = []
def subscribe(self, etype, cb): # noqa: ANN001, ANN201
self.subs.append(etype)
return object()
# One real message, then the empty sentinel: the drain loop pulls two times.
mc = _FakeMC([EventType.CONTACT_MSG_RECV, EventType.NO_MORE_MSGS])
device = MeshCoreDevice(port="COM_TEST")
subs: list = []
stop = device._message_pump(mc, subs, mc.subscribe)
try:
await asyncio.sleep(0.01) # let the task for the immediate drain run
assert mc.commands.calls == 2 # CONTACT_MSG_RECV then NO_MORE_MSGS
assert EventType.MESSAGES_WAITING in mc.subs # the subscription for a push with low latency
finally:
stop()
# -- persistence --------------------------------------------------------------
def test_record_and_load_direct_conversation(repo: Repository) -> None:
"""Direct messages make a round trip and return in time order."""
repo.record_chat_message(ChatMessage(text="hi", outbound=True, peer="D4E5F6A7", acked=True))
repo.record_chat_message(ChatMessage(text="yo", outbound=False, peer="d4e5f6a7", snr=3.0))
got = repo.recent_chat_messages(is_channel=False, peer="d4e5f6a7")
assert [m.text for m in got] == ["hi", "yo"]
assert got[0].outbound is True and got[0].acked is True
assert got[1].outbound is False and got[1].snr == 3.0
def test_record_and_load_channel_conversation(repo: Repository) -> None:
"""Channel messages have the identity of the channel as key, apart from direct messages."""
repo.record_chat_message(ChatMessage(text="c0", is_channel=True, channel_id="aa00"))
repo.record_chat_message(ChatMessage(text="c1", is_channel=True, channel_id="bb11"))
assert [m.text for m in repo.recent_chat_messages(is_channel=True, channel_id="aa00")] == ["c0"]
assert [m.text for m in repo.recent_chat_messages(is_channel=True, channel_id="bb11")] == ["c1"]
def test_last_chat_messages_returns_latest_per_conversation(repo: Repository) -> None:
"""The preview in the picker shows the newest message of each conversation."""
repo.record_chat_message(ChatMessage(text="old", peer="aa"))
repo.record_chat_message(ChatMessage(text="new", peer="aa"))
repo.record_chat_message(ChatMessage(text="chan", is_channel=True, channel_id="aa00"))
lasts = repo.last_chat_messages()
assert lasts["dm:aa"].text == "new"
assert lasts["chan:aa00"].text == "chan"
# -- picker row rendering -----------------------------------------------------
class _FakeChat:
"""Stand-in for the chat service. It has only the unread lookup that a row title reads."""
def __init__(self, unread: dict[str, int]) -> None:
self._unread = unread
def unread(self, key: str) -> int:
return self._unread.get(key, 0)
class _RowCtx:
"""Minimal ctx. It has only what the picker-row helpers use (repo and chat)."""
def __init__(self, repo: Repository, unread: dict[str, int] | None = None) -> None:
self.repo = repo
self.chat = _FakeChat(unread or {})
def test_live_lasts_refreshes_preview_after_ttl(repo: Repository) -> None:
"""A new message is visible through _LiveLasts after the cache TTL ends."""
repo.record_chat_message(ChatMessage(text="first", is_channel=True, channel_id="c0"))
live = _LiveLasts(_RowCtx(repo), seed=repo.last_chat_messages(), ttl=0) # 0: always fresh
assert live.get("chan:c0").text == "first"
repo.record_chat_message(ChatMessage(text="second", is_channel=True, channel_id="c0"))
assert live.get("chan:c0").text == "second" # the live value, not the seed
def test_live_lasts_serves_seed_within_ttl(repo: Repository) -> None:
"""Within the TTL, the seeded snapshot is used. The repository is not queried again."""
live = _LiveLasts(_RowCtx(repo), seed={"chan:c0": ChatMessage(text="seed")}, ttl=999)
repo.record_chat_message(ChatMessage(text="later", is_channel=True, channel_id="c0"))
assert live.get("chan:c0").text == "seed"
#: A resolver that places no name. Each hue falls back to muted.
_NO_KEYS = lambda name: None # noqa: E731 - a one-line stub is easier to read inline
def _keys_of(mapping: dict[str, str]):
"""A fixed name→key resolver for ``mapping`` (case-folded, like the real resolver)."""
return lambda name: mapping.get(name.casefold())
#: The public key of our node in the tests of our chat chip.
_OUR_KEY = "5e" + "7f" * 31
def _ink(text, run: str) -> str:
"""The style that colours ``run`` in ``text``: its identity span, not the path line's stamp."""
from meshterm.ui.pathline import PATH_INK
return next(
str(s.style)
for s in text.spans
if text.plain[s.start : s.end] == run and str(s.style) != PATH_INK
)
def test_preview_prefixes_own_messages_only() -> None:
"""Only an outbound message gets a ``you:`` prefix. Inbound text shows exactly as it is.
The firmware puts the sender of a channel message in the text of the message. Thus the
preview does not make an author, because the author is already in the text. The author
of a direct message is the label of the row.
"""
chan_in = ChatMessage(text="Bob: hi", is_channel=True, channel_idx=0) # sender in text
assert _preview_text(chan_in, _NO_KEYS).plain == "Bob: hi"
dm_in = ChatMessage(text="hey", peer="aa", peer_name="Bob")
assert _preview_text(dm_in, _NO_KEYS).plain == "hey"
mine = ChatMessage(text="yo", outbound=True, is_channel=True, channel_idx=0)
assert _preview_text(mine, _NO_KEYS).plain == "you: yo" # our node with no name
# With our name, the preview shows our chat chip: a padded block in reverse video.
assert _preview_text(mine, _NO_KEYS, us=("Me", _OUR_KEY)).plain == " ★ Me yo"
def test_preview_colours_channel_sender_and_mentions() -> None:
"""A preview lights the names of senders and mentions that resolve, in the hue of the key.
A name that no known node has stays muted, because colour is only for keyed identities.
"""
from meshterm.ui.theme import node_style
key_of = _keys_of({"bob": "d4" + "0" * 62, "alice": "60" + "0" * 62})
msg = ChatMessage(text="Bob: hi @[Alice] @[Zed]", is_channel=True, channel_idx=0)
preview = _preview_text(msg, key_of)
assert preview.plain == "Bob: hi @Alice @Zed" # the brackets are removed for display
styles = {span.style for span in preview.spans}
assert node_style("d4") in styles and node_style("60") in styles
# The @Zed that does not resolve takes the unknown-node grey, as each name with no place.
zed = preview.plain.index("@Zed")
assert any(s.style == "node.unknown" and s.start <= zed < s.end for s in preview.spans)
def test_preview_draws_the_author_and_the_mentions_as_the_transcript_does(powerline) -> None:
"""With chips, the author closes on the point and a mention opens on the notch.
The point says what the colon said, so the chip has no colon, and a mention chip has
no ``@``. Our own author is ``★ Name``, as the sender chip of our messages in the
transcript. A mention of our name has no star. With no name for our node, the bare
``★`` stays.
"""
from meshterm.ui.pathline import POWERLINE_ROUND_CLOSE, POWERLINE_ROUND_OPEN, POWERLINE_SEP
powerline(True)
theirs = ChatMessage(text="Bob: hi @[Alice]", is_channel=True, channel_idx=0)
author = POWERLINE_ROUND_OPEN + " Bob " + POWERLINE_SEP
mention = POWERLINE_SEP + " Alice " + POWERLINE_ROUND_CLOSE
assert _preview_text(theirs, _NO_KEYS).plain == f"{author} hi {mention}"
mine = ChatMessage(text="yo", outbound=True, is_channel=True, channel_idx=0)
assert _preview_text(mine, _NO_KEYS).plain == f"{POWERLINE_ROUND_OPEN} ★ {POWERLINE_SEP} yo"
us = ("Me", _OUR_KEY)
ours = f"{POWERLINE_ROUND_OPEN} ★ Me {POWERLINE_SEP} yo"
assert _preview_text(mine, _NO_KEYS, us=us).plain == ours
named = ChatMessage(text="Bob: hi @[Me]", is_channel=True, channel_idx=0)
me = POWERLINE_SEP + " Me " + POWERLINE_ROUND_CLOSE
assert _preview_text(named, _NO_KEYS, us=us).plain == f"{author} hi {me}"
def test_preview_keeps_the_whole_message() -> None:
"""The preview is built whole. The cut of the row is the part that the ←→ scroll passes."""
long = ChatMessage(text="x" * 300, is_channel=True, channel_idx=0)
assert _preview_text(long, _NO_KEYS).plain == "x" * 300
def test_title_shows_badge_and_author_preview(repo: Repository) -> None:
"""A channel row shows its live unread badge and its preview with the author prefix."""
conv = Conversation(label="General", is_channel=True, channel_idx=0, channel_id="c0")
ctx = _RowCtx(repo, unread={"chan:c0": 3})
last = ChatMessage(text="Bob: hi there", is_channel=True, channel_id="c0") # sender in text
title = _title(ctx, conv, {"chan:c0": last}, _NO_KEYS, _lanes([conv]))
line = title.plain # a Text, because there are unread messages
assert line.startswith("🔒 General") # a private channel starts with its openness glyph
assert "● 3" in line
assert "Bob: hi there" in line
def test_title_reddens_only_the_unread_dot(repo: Repository) -> None:
"""If there are unread messages, only the ``●`` glyph of the row is red. If not, no dot."""
from rich.text import Text
conv = Conversation(label="General", is_channel=True, channel_idx=0, channel_id="c0")
unread = _title(_RowCtx(repo, unread={"chan:c0": 2}), conv, {}, _NO_KEYS, _lanes([conv]))
assert isinstance(unread, Text)
dot = unread.plain.index("●")
reddened = [
span for span in unread.spans if span.style == "err" and span.start <= dot < span.end
]
assert reddened and all(span.end - span.start == 1 for span in reddened) # only the glyph
read = _title(_RowCtx(repo, unread={}), conv, {}, _NO_KEYS, _lanes([conv]))
assert isinstance(read, Text) # always a Text, so its spans can have the colour of the row
assert "●" not in read.plain # nothing unread, thus no badge dot
assert not any(span.style == "err" for span in read.spans)
def test_title_preview_column_aligns_regardless_of_label_length(repo: Repository) -> None:
"""The preview starts at the same column when the label is short or long (and clipped)."""
ctx = _RowCtx(repo)
short = Conversation(label="A", is_channel=True, channel_idx=0, channel_id="c0")
long = Conversation(
label="A much longer channel name here", is_channel=True, channel_idx=1, channel_id="c1"
)
m0 = ChatMessage(text="X: hello", is_channel=True, channel_id="c0")
m1 = ChatMessage(text="Y: hello", is_channel=True, channel_id="c1")
lanes = _lanes([short, long]) # one measurement for the list, as the picker does it
l0 = _title(ctx, short, {"chan:c0": m0}, _NO_KEYS, lanes).plain
l1 = _title(ctx, long, {"chan:c1": m1}, _NO_KEYS, lanes).plain
assert l0.index("X: hello") == l1.index("Y: hello")
def test_title_leads_with_openness_glyph(repo: Repository) -> None:
"""A channel row starts with an openness glyph: ＃ from the name, 🌐 public, 🔒 private."""
ctx = _RowCtx(repo)
def head(conv):
return _title(ctx, conv, {}, _NO_KEYS, _lanes([conv])).plain.split(" ", 1)[0]
named = Conversation(
label="#general", is_channel=True, channel_id="c0", secret=derive_secret("#general")
)
public = Conversation(
label="Public", is_channel=True, channel_id="c1", secret=DEFAULT_PUBLIC_SECRET
)
private = Conversation(label="Ops", is_channel=True, channel_id="c2", secret=bytes(range(16)))
assert head(named) == "＃"
assert head(public) == "🌐"
assert head(private) == "🔒"
def test_title_contact_dot_reflects_conversation_history(repo: Repository) -> None:
"""The contact dot of the title shows if there is history with the contact.
The dot is filled (●) after the user and the contact talked, and hollow (○) before. It
has the colour of the companion mark in both cases. The name has the hue of the key of
the contact.
"""
from meshterm.tools.chat import _COMPANION_DOT_STYLE
from meshterm.ui.theme import node_style
ctx = _RowCtx(repo)
contact = Conversation(
label="Alice", is_channel=False, contact=Contact(name="Alice", public_key="d4" + "0" * 62)
)
def dot_has_mark_colour(row) -> bool:
return any(
span.style == _COMPANION_DOT_STYLE and span.start == 0 and span.end == 1
for span in row.spans
)
# No history yet: a hollow ring in the colour of the companion mark.
fresh = _title(ctx, contact, {}, _NO_KEYS, _lanes([contact]))
assert fresh.plain.startswith("○") and dot_has_mark_colour(fresh)
# The name lane has the hue of the key of Alice.
name_at = fresh.plain.index("Alice")
assert any(s.style == node_style("d4") and s.start <= name_at < s.end for s in fresh.spans)
# After the user and the contact exchange messages, the same dot is filled.
last = ChatMessage(text="hi", peer=contact.peer)
talked = _title(ctx, contact, {contact.key: last}, _NO_KEYS, _lanes([contact]))
assert talked.plain.startswith("●") and dot_has_mark_colour(talked)
# -- chat service -------------------------------------------------------------
async def test_service_records_inbound_and_tracks_unread(repo: Repository) -> None:
"""Inbound messages are stored and increase the unread count of the conversation."""
device = MockDevice()
ctx = _StubContext(device, repo)
chat = ChatService(ctx)
await chat.start()
try:
ctx.events.publish(
MeshEvent.message_event(Message(text="ping", sender="ffeeddcc", is_channel=False))
)
await chat._queue.join() # let the inbound worker resolve and store the message
assert chat.unread("dm:ffeeddcc") == 1
assert chat.unread_total() >= 1
stored = repo.recent_chat_messages(is_channel=False, peer="ffeeddcc")
assert [m.text for m in stored] == ["ping"]
finally:
await chat.stop()
await device.disconnect()
async def test_service_active_conversation_suppresses_unread(repo: Repository) -> None:
"""The open conversation has no unread messages and gets no more while it is active."""
device = MockDevice()
ctx = _StubContext(device, repo)
chat = ChatService(ctx)
await chat.start()
try:
chat.set_active("dm:ffeeddcc")
ctx.events.publish(
MeshEvent.message_event(Message(text="ping", sender="ffeeddcc", is_channel=False))
)
await chat._queue.join() # let the inbound worker resolve and store the message
assert chat.unread("dm:ffeeddcc") == 0 # an active thread gets no unread messages
assert repo.recent_chat_messages(is_channel=False, peer="ffeeddcc") # still stored
finally:
await chat.stop()
await device.disconnect()
async def test_service_files_channel_message_by_current_slot_occupant(repo: Repository) -> None:
"""A channel message is filed under the channel that is in its slot now.
The wire gives only a slot index, which the firmware assigns from the current position of
the channel. A change of the order of the slots must not send later messages to the wrong
channel. (The test does this change out of band, and does not tell the service.) The
resolution reads the slot again each time. Thus a message on slot 0 goes to the channel
that is in slot 0 at that time, and never to an old cached identity.
"""
from meshterm.core.channels import channel_identity
device = MockDevice()
ctx = _StubContext(device, repo)
await device.connect()
secret_a, secret_b = bytes(range(16)), bytes(range(16, 32))
await device.set_channel(0, "Alpha", secret_a)
await device.set_channel(1, "Bravo", secret_b)
id_a = channel_identity("Alpha", secret_a)
id_b = channel_identity("Bravo", secret_b)
chat = ChatService(ctx)
await chat.start() # primes slot 0 to Alpha and slot 1 to Bravo
try:
ctx.events.publish(
MeshEvent.message_event(Message(text="from alpha", is_channel=True, channel=0))
)
await chat._queue.join()
# Swap the channels in slots 0 and 1 out of band. The service is not told.
await device.set_channel(0, "Bravo", secret_b)
await device.set_channel(1, "Alpha", secret_a)
ctx.events.publish(
MeshEvent.message_event(Message(text="from bravo", is_channel=True, channel=0))
)
await chat._queue.join()
alpha = repo.recent_chat_messages(is_channel=True, channel_id=id_a)
bravo = repo.recent_chat_messages(is_channel=True, channel_id=id_b)
assert [m.text for m in alpha] == ["from alpha"] # the new order has no effect on it
assert [m.text for m in bravo] == ["from bravo"] # not filed under the identity of Alpha
finally:
await chat.stop()
await device.disconnect()
async def test_service_records_channel_messages_in_arrival_order(repo: Repository) -> None:
"""A burst of channel messages is stored in the order of arrival, also with async resolution.
Each channel message resolves its identity with a device read. The serial inbound worker
makes sure that the messages go into the transcript (ordered by insertion) in the order
that the app received them.
"""
from meshterm.core.channels import channel_identity
device = MockDevice()
ctx = _StubContext(device, repo)
await device.connect()
secret = bytes(range(16))
await device.set_channel(0, "Alpha", secret)
channel_id = channel_identity("Alpha", secret)
chat = ChatService(ctx)
await chat.start()
try:
for i in range(5):
ctx.events.publish(
MeshEvent.message_event(Message(text=f"m{i}", is_channel=True, channel=0))
)
await chat._queue.join()
stored = repo.recent_chat_messages(is_channel=True, channel_id=channel_id)
assert [m.text for m in stored] == ["m0", "m1", "m2", "m3", "m4"]
finally:
await chat.stop()
await device.disconnect()
# -- chat screen --------------------------------------------------------------
def _screen(session: _StubSession, send, *, messages=None, resend=None) -> ChatScreen:
"""Build a ChatScreen for a direct conversation, with a stub session and a send hook."""
conv = Conversation(
label="Alice",
is_channel=False,
contact=Contact(name="Alice", public_key="d4" + "0" * 62, key_prefix="d4e5f6a7"),
)
return ChatScreen(
conv,
messages or [],
send=send,
names={"d4e5f6a7": "Alice"},
session=session,
resend=resend,
)
def test_chat_screen_renders_transcript_and_input() -> None:
"""The body shows each message, and it always ends with the input line."""
session = _StubSession()
messages = [ChatMessage(text="hi there", outbound=True, peer="d4e5f6a7", acked=True)]
screen = _screen(session, send=None, messages=messages)
lines = screen.render_body(60)
joined = "\n".join(lines)
assert "hi there" in joined
assert "›" in joined # the input editor's prompt marker
class _PasteSession(_StubSession):
"""A stub session. Its paste confirm gives a scripted answer (Cancel or Paste)."""
def __init__(self, answer: bool) -> None:
super().__init__()
self.answer = answer
self.dialogs: list = []
async def button_dialog(self, prompt, buttons, **kwargs): # noqa: ANN001, ANN201
self.dialogs.append((prompt, buttons, kwargs))
return self.answer
async def _drain_paste(screen: ChatScreen) -> None:
"""Let the scheduled paste-confirm task run to its end."""
for _ in range(100):
await asyncio.sleep(0)
if not screen._paste_open:
return
async def test_chat_paste_confirms_amber_then_inserts() -> None:
"""Ctrl-V paste first asks in an amber Cancel/Paste dialog, then puts the text in compose."""
session = _PasteSession(answer=True)
screen = _screen(session, send=None)
screen.handle("paste", "hello\nworld")
await _drain_paste(screen)
# The newline changed to a space, and the text went into the compose line…
assert screen._editor.text == "hello world"
prompt, buttons, kwargs = session.dialogs[0]
assert "Paste 11 characters" in prompt.plain # the length of the folded and stripped text
assert kwargs.get("border_style") == "warn" # the amber (danger) tier, "yellow"
assert [label for label, _ in buttons] == ["Cancel", "Paste"] # the safe way out is on the left
async def test_chat_paste_declined_leaves_compose_untouched() -> None:
"""If the user selects Cancel on the paste confirm, the screen inserts nothing."""
session = _PasteSession(answer=False)
screen = _screen(session, send=None)
screen.handle("paste", "unwanted")
await _drain_paste(screen)
assert screen._editor.text == ""
async def test_chat_screen_enter_sends_and_appends() -> None:
"""Enter sends the line and appends the returned message to the transcript."""
session = _StubSession()
sent: list[str] = []
async def send(text: str) -> ChatMessage:
sent.append(text)
return ChatMessage(text=text, outbound=True, peer="d4e5f6a7", acked=True)
screen = _screen(session, send=send)
for ch in "hello":
screen.handle("text", ch)
screen.handle("enter")
await asyncio.sleep(0) # let the scheduled send task run
assert sent == ["hello"]
assert screen._messages[-1].text == "hello"
assert session.invalidations > 0
async def test_direct_send_spins_until_the_ack_resolves(monkeypatch) -> None:
"""A pending direct message spins while the send is in progress, then shows ✓."""
from meshterm.ui.tui.spinner import Spinner
monkeypatch.setattr("meshterm.ui.chat.spinner_interval", lambda: 0.005)
session = _StubSession()
release = asyncio.Event()
async def send(text: str) -> ChatMessage:
await release.wait() # keep the send open, so that the glyph spins
return ChatMessage(text=text, outbound=True, peer="d4e5f6a7", acked=True)
screen = _screen(session, send=send)
for ch in "hi":
screen.handle("text", ch)
screen.handle("enter")
# While the send is open, the last glyph is an animation step of the spinner, and it advances.
await asyncio.sleep(0.03)
first = screen._spinner.frame
joined = "\n".join(screen.render_body(60))
assert first in Spinner.BRAILLE and first in joined and "⏳" not in joined
await asyncio.sleep(0.03)
assert screen._spinner.frame != first # the animation runs
# When the ack arrives, the spinner is gone and the message shows its delivered glyph.
release.set()
for _ in range(100): # let _send_direct finish (ticker cancel and swap) before the assert
await asyncio.sleep(0.005)
if screen._messages[-1].acked is not None:
break
assert screen._messages[-1].acked is True
assert "✓" in "\n".join(screen.render_body(60))
def test_byte_counter_shows_used_over_limit_and_colors_only_used() -> None:
"""The inline budget shows ``used/limit``. Only the used count has a colour (max is muted)."""
from rich.text import Text
screen = _screen(_StubSession(), send=None)
for ch in "hello":
screen.handle("text", ch)
counter = screen._byte_counter(screen._byte_limit())
assert isinstance(counter, Text)
assert counter.plain.strip() == "5/150" # 5 bytes used of the 150-byte cap of a direct message
# The budget is at the right edge of the input line. Padding separates it from the
# compose text, and it does not follow the cursor. It shares the row of the input, so a
# full transcript cannot push it off the bottom of the viewport.
body = re.sub(r"\x1b\[[0-9;]*m", "", "\n".join(screen.render_body(80)))
input_row = next(line for line in body.splitlines() if "5/150" in line)
assert re.search(r"› hello +5/150$", input_row.rstrip()) # padding, then the counter
assert len(input_row.rstrip()) == 80 # the counter reaches the right edge of the width
# Only the "5" has a colour. The "/150" tail stays muted (it does not change).
used_at = counter.plain.index("5")
slash_at = counter.plain.index("/")
used_spans = [s for s in counter.spans if s.start <= used_at < s.end]
tail_spans = [s for s in counter.spans if s.start <= slash_at < s.end]
assert used_spans and used_spans[0].style == "ok" # green: much room remains
assert tail_spans and tail_spans[0].style == "muted"
def test_byte_counter_stays_bottom_right_when_the_compose_wraps() -> None:
"""The byte counter stays at the bottom right when the compose line wraps.
It stays at the right edge of the last line. It does not follow the cursor down to the
second row.
"""
screen = _screen(_StubSession(), send=None)
text = "this is a long compose line that certainly wraps onto several rows here ok"
for ch in text:
screen.handle("text", ch)
counter = f"{len(text)}/150"
body = re.sub(r"\x1b\[[0-9;]*m", "", "\n".join(screen.render_body(40)))
counter_rows = [line for line in body.splitlines() if counter in line]
assert len(counter_rows) == 1
row = counter_rows[0]
assert row.rstrip().endswith(counter) # at the right edge, not in the middle of the line
assert len(row.rstrip()) == 40 # reaches the right edge of the render width
assert not row.lstrip().startswith("›") # on a wrapped row, not on the first input row
def test_byte_style_escalates_as_budget_runs_out() -> None:
"""The colour of the used bytes goes from green to yellow, orange, and red as bytes run out."""
from meshterm.ui.tui.prompt import _BYTES_ORANGE, _BYTES_YELLOW, byte_style
assert byte_style(80) == "ok" # much room remains: green
assert byte_style(20) == _BYTES_YELLOW # in the tight band: yellow
assert byte_style(10) == _BYTES_ORANGE # in the low band: orange
assert byte_style(0) == "err" # the limit is reached: red
assert byte_style(-5) == "err" # over the limit: still red
def test_channel_byte_limit_is_lower_than_direct() -> None:
"""A channel broadcast has a smaller byte budget than a direct message."""
from meshterm.ui.tui.prompt import CHANNEL_BYTE_LIMIT, DM_BYTE_LIMIT
direct = _screen(_StubSession(), send=None)
channel = _channel_screen([])
assert direct._byte_limit() == DM_BYTE_LIMIT == 150
assert channel._byte_limit() == CHANNEL_BYTE_LIMIT == 130
def test_overflow_counts_multibyte_characters_by_byte() -> None:
"""An emoji (4 UTF-8 bytes) counts as 4 in the budget. The overflow mark is on a character."""
screen = _channel_screen([]) # 130-byte limit
# 32 emojis are 128 bytes (under the limit). The 33rd makes 132 (over) at that whole character.
for _ in range(33):
screen.handle("text", "😀")
assert screen._used_bytes() == 33 * 4
assert screen._overflow_at(screen._byte_limit()) == 32 # the 33rd emoji is the first over
async def test_over_limit_message_is_not_sent_and_buffer_is_kept() -> None:
"""Enter on a line over the budget reports the overage and sends nothing. The text stays."""
session = _StubSession()
sent: list[str] = []
async def send(text: str) -> ChatMessage:
sent.append(text)
return ChatMessage(text=text, outbound=True, peer="d4e5f6a7", acked=True)
screen = _screen(session, send=send)
for ch in "x" * 151: # one byte over the 150-byte cap of a direct message
screen.handle("text", ch)
screen.handle("enter")
await asyncio.sleep(0) # the screen must not schedule a task, but let the loop turn
assert sent == [] # the send was refused
assert screen._editor.text == "x" * 151 # the buffer is intact, so the user can trim it
assert "Too long by 1 byte" in screen._status
# If the user trims the text to the limit, the notice clears.
screen.handle("backspace")
assert screen._status == ""
def test_chat_screen_shows_delivery_glyphs() -> None:
"""An outbound direct message ends with a delivery mark: a spinner, then ✓ or ✗."""
from meshterm.ui.tui.spinner import Spinner
async def resend(message): # noqa: ANN001, ANN202 - this test does not await it
return message
messages = [
ChatMessage(text="delivered", outbound=True, peer="d4e5f6a7", acked=True),
ChatMessage(text="dropped", outbound=True, peer="d4e5f6a7", acked=False),
ChatMessage(text="inflight", outbound=True, peer="d4e5f6a7", acked=None),
]
screen = _screen(_StubSession(), send=None, messages=messages, resend=resend)
joined = "\n".join(screen.render_body(60))
# A message that waits for its ack spins (a Braille animation step). A resolved message
# shows the app-wide ✓ or ✗ status mark (not the ✅ or ❌ emoji, which are packet-class icons).
assert "✓" in joined and "✗" in joined
assert "✅" not in joined and "❌" not in joined
assert Spinner.BRAILLE[0] in joined and "⏳" not in joined
# For a failed message, the footer hint names the retry shortcut (in the app-wide
# compact ^ notation for Ctrl chords).
assert "^R" in screen.footer_hint
def test_wrapped_body_hangs_under_the_first_line() -> None:
"""A body that is too long for the width wraps with a hanging indent under its first line."""
from datetime import datetime, timezone
base = datetime(2026, 7, 5, 14, 24, tzinfo=timezone.utc)
# Many short words make several wrap points at a narrow width.
text = " ".join(["word"] * 20)
messages = [ChatMessage(text=text, peer="d4e5f6a7", created_at=base)]
screen = _screen(_StubSession(), send=None, messages=messages)
body_lines = [_strip_ansi(ln) for ln in screen._body_lines(text, messages[0], 30)]
assert len(body_lines) > 1 # the text wrapped
stamp = base.astimezone().strftime("%H:%M")
indent = body_lines[0].index(stamp) + len(stamp) + 2 # gutter: " HH:MM "
first_word = body_lines[0].index("word")
assert first_word == indent
# The text of each continuation line starts at the same column as the body of the first line.
for cont in body_lines[1:]:
assert cont.startswith(" " * indent)
assert cont[indent] != " " # the text continues under the body, not under the gutter
def test_at_mention_renders_as_name_in_sender_hue() -> None:
"""An ``@[Name]`` token renders as a bare ``@Name`` in the hue of the key of the node.
A mentioned name that no contact or stored advert has stays muted. This follows the
app-wide rule that colour marks a keyed identity.
"""
from datetime import datetime, timezone
from meshterm.ui.theme import node_style
base = datetime(2026, 7, 5, 14, 24, tzinfo=timezone.utc)
conv = Conversation(label="#public", is_channel=True, channel_idx=0)
message = ChatMessage(
text="Bob: @[Alice] and @[Zed] around?",
is_channel=True,
channel_idx=0,
created_at=base,
)
screen = ChatScreen(
conv,
[message],
send=None,
names={},
session=_StubSession(),
key_of=_keys_of({"alice": "60" + "0" * 62}),
)
_, body = screen._sender_and_body(message)
text = screen._render_mentions(body, selected=False)
assert "@Alice" in text.plain # the token with brackets became a bare mention
assert "@[Alice]" not in text.plain and "[Alice]" not in text.plain
# The "@Alice" text has the spectrum hue of the key (the same hue that the header uses).
assert _ink(text, "@Alice") == node_style("60")
# The unknown @Zed gets the unknown-node grey, because it has no key and no hue.
assert _ink(text, "@Zed") == "node.unknown"
def test_direct_chat_unknown_mention_stays_muted() -> None:
"""In a direct chat, an unresolved ``@mention`` stays muted, as it does in a channel.
The sender label of a direct thread falls back to the key of the peer. Thus it keeps its
hue when the display name does not resolve. But this fallback must not go to an
``@mention`` in the body, because a mention names any person, and an unknown person is
grey. This is a regression guard against the fallback to the peer key in mentions.
"""
from meshterm.ui.theme import node_style
# Direct chat with the peer key d4…: the sender label can use this hue, but mentions must not.
conv = Conversation(
label="Alice",
is_channel=False,
contact=Contact(name="Alice", public_key="d4" + "0" * 62, key_prefix="d4e5f6a7"),
)
screen = ChatScreen(
conv,
[],
send=None,
names={"d4e5f6a7": "Alice"},
session=_StubSession(),
key_of=_keys_of({"alice": "60" + "0" * 62}),
)
body = screen._render_mentions("@[Alice] and @[Zed]", selected=False)
# A mention that resolves still lights in the hue of its own key…
assert _ink(body, "@Alice") == node_style("60")
# …but an unknown mention is grey. It does not get the hue (d4…) of the peer.
assert _ink(body, "@Zed") == "node.unknown"
# The sender label keeps the fallback to the peer key (the behaviour is the same).
inbound = ChatMessage(text="hi", peer="d4e5f6a7")
assert _ink(screen._group_header("Zed", inbound), "Zed") == node_style("d4")
def test_direct_transcript_groups_under_sender_headers() -> None:
"""A direct chat has the same grouped layout as a channel: one header for each sender run."""
from datetime import datetime, timezone
base = datetime(2026, 7, 5, 14, 24, tzinfo=timezone.utc)
messages = [
ChatMessage(text="hi", peer="d4e5f6a7", created_at=base),
ChatMessage(text="you there?", peer="d4e5f6a7", created_at=base),
ChatMessage(text="yes!", outbound=True, peer="d4e5f6a7", acked=True, created_at=base),
]
screen = _screen(_StubSession(), send=None, messages=messages)
rendered = _strip_ansi("\n".join(screen._render_grouped(80)))
# The inbound sender resolves to the contact name (from the names map), not to the raw key.
assert rendered.count("Alice") == 1 # the two inbound messages share one header
assert "d4e5f6a7" not in rendered # if a name is known, the screen does not show the key
assert "you" in rendered # our reply has its own header
assert "hi" in rendered and "you there?" in rendered and "yes!" in rendered
assert "✓" in rendered # the outbound message keeps its delivery mark
async def test_chat_screen_retry_resends_failed_message() -> None:
"""Ctrl-R tries the newest unacknowledged message again and changes it in place."""
session = _StubSession()
failed = ChatMessage(text="oops", outbound=True, peer="d4e5f6a7", acked=False, row_id=7)
async def resend(message: ChatMessage) -> ChatMessage:
message.acked = True # this time, the retry succeeds
return message
conv = Conversation(
label="Alice",
is_channel=False,
contact=Contact(name="Alice", public_key="d4" + "0" * 62, key_prefix="d4e5f6a7"),
)
screen = ChatScreen(conv, [failed], send=None, names={}, session=session, resend=resend)
screen.handle("retry")
await asyncio.sleep(0) # let the scheduled resend task run
assert screen._messages[-1].acked is True # the same object, now acknowledged
assert "✓" in "\n".join(screen.render_body(60))
async def test_chat_fkey_lane_dims_retry_and_the_nav_slots() -> None:
"""The lane offers Retry only for a message to retry, and navigation only with a transcript."""
async def resend(message: ChatMessage) -> ChatMessage:
message.acked = True
return message
empty = _screen(_StubSession(), send=None, resend=resend)
assert not any(pair.enabled for pair in empty.picocalc_lyra_lane if pair) # nothing to move
failed = ChatMessage(text="oops", outbound=True, peer="d4e5f6a7", acked=False)
screen = _screen(_StubSession(), send=None, messages=[failed], resend=resend)
assert all(pair.enabled for pair in screen.picocalc_lyra_lane[3:]) # the pager moves the pick
assert screen.picocalc_lyra_lane[0].opp_enabled is True # F6 Retry: the message has no ack
screen.handle("retry")
await asyncio.sleep(0)
assert failed.acked is True
assert screen.picocalc_lyra_lane[0].opp_enabled is False # acknowledged: nothing to retry
assert (
screen.picocalc_lyra_lane[0].opp_label == "Retry"
) # dim, not removed: it is part of this screen
def test_chat_fkey_lane_speaks_the_transcript_and_drops_channel_retry() -> None:
"""A chip names what it does here, and a channel has no retry."""
messages = [ChatMessage(text="hi", peer="d4e5f6a7")]
direct = _screen(_StubSession(), send=None, messages=messages)
# The labels are not "Bottom" and "Top". In a conversation, the Shift bank goes back to
# the newest message and the compose line, or goes to the oldest message. The chips have
# new labels in place, so each jump stays on the pager that goes in that direction.
assert [pair.opp_label for pair in direct.picocalc_lyra_lane[3:]] == ["Latest", "Oldest"]
assert [pair.opp_action for pair in direct.picocalc_lyra_lane[3:]] == ["ctrl_end", "ctrl_home"]
assert [pair.label for pair in direct.picocalc_lyra_lane[3:]] == ["Page ↓", "Page ↑"]
assert direct.picocalc_lyra_lane[0].opp_label == "Retry"
channel = ChatScreen(
Conversation(label="#general", is_channel=True, channel_id="1"),
messages,
send=None,
names={},
session=_StubSession(),
)
# A channel message has no acknowledgement, so a retry is not possible here. The slot is
# empty (it is not dim), and its key resolves to nothing.
assert channel.picocalc_lyra_lane[0].opp_label == ""
assert fkeys.PICOCALC_LYRA_DECK.action_for(channel.picocalc_lyra_lane, 6) is None
def test_chat_screen_up_picks_and_ctrl_end_returns_to_compose() -> None:
"""↑ selects the newest message (and leaves the tail). ^End returns to compose."""
messages = [
ChatMessage(text="hi", peer="d4e5f6a7"),
ChatMessage(text="reply", outbound=True, peer="d4e5f6a7", acked=True),
]
screen = _screen(_StubSession(), send=None, messages=messages)
assert screen._stick is True and screen._selected is None
screen.handle("up")
assert screen._stick is False
assert screen._selected == len(screen._messages) - 1 # the newest message
screen.handle("ctrl_end")
assert screen._stick is True and screen._selected is None
def test_split_channel_sender_extracts_name_prefix() -> None:
"""A ``Name: message`` channel line splits into the sender and a body without the prefix."""
from meshterm.core.channels import split_channel_sender
assert split_channel_sender("Alice: hey there") == ("Alice", "hey there")
assert split_channel_sender("Yagi Repeater: online") == ("Yagi Repeater", "online")
# No possible prefix: the text does not change.
assert split_channel_sender("just a message") == (None, "just a message")
assert split_channel_sender("https://example.com") == (None, "https://example.com")
assert split_channel_sender("14:30 standup") == (None, "14:30 standup")
def test_channel_transcript_groups_by_sender() -> None:
"""Channel messages in a row from the same sender share one header. The body has no prefix."""
from datetime import datetime, timezone
conv = Conversation(label="#public", is_channel=True, channel_idx=0)
base = datetime(2026, 7, 5, 14, 24, tzinfo=timezone.utc)
def at(minutes: int) -> datetime:
return base.replace(minute=24 + minutes)
messages = [
ChatMessage(text="Alice: hi", is_channel=True, channel_idx=0, created_at=at(0)),
ChatMessage(text="Alice: again", is_channel=True, channel_idx=0, created_at=at(6)),
ChatMessage(text="Bob: yo", is_channel=True, channel_idx=0, created_at=at(8)),
ChatMessage(
text="hello all", outbound=True, is_channel=True, channel_idx=0, created_at=at(9)
),
]
screen = ChatScreen(
conv,
messages,
send=None,
names={},
session=_StubSession(),
self_name="Me",
self_key=_OUR_KEY,
)
rendered = _strip_ansi("\n".join(screen._render_grouped(80)))
assert rendered.count("Alice") == 1 # the two Alice messages share one header
# The header of our group is our name after the ★ (JP, 2026-10-13), not the bare ★.
assert "Bob" in rendered and "★ Me" in rendered
assert "hi" in rendered and "again" in rendered # the bodies are there, without the prefix
assert "Alice: hi" not in rendered # the raw name prefix moved into the header
# Each message keeps its own timestamp on its line, also when grouped under one sender.
stamps = [at(m).astimezone().strftime("%H:%M") for m in (0, 6, 8, 9)]
for stamp in stamps:
assert stamp in rendered
assert stamps[0] != stamps[1] # the grouped Alice messages show different times
def test_transcript_draws_the_sender_as_an_origin_and_a_mention_as_a_destination(
powerline,
) -> None:
"""The sender chip closes on the point, and a mention chip opens on the notch.
A message goes from its author toward the node that it names. Thus the two chips are
the two ends of one route that is not drawn (JP, 2026-10-10).
"""
from meshterm.ui.pathline import POWERLINE_SEP
from meshterm.ui.widgets import ORIGIN, mention_chip, name_chip
powerline(True) # this test is about the chip, so ask for a terminal that draws one
conv = Conversation(label="#public", is_channel=True, channel_idx=0)
messages = [
ChatMessage(text="Alice: @[Bob] you around?", is_channel=True, channel_idx=0),
]
screen = ChatScreen(conv, messages, send=None, names={}, session=_StubSession())
rendered = _strip_ansi("\n".join(screen._render_grouped(80)))
# The test builds the chips through the widget, not as literal text. The end caps are
# the caps of the path line, and the test asserts the framing, not the glyph.
sender = name_chip("Alice", role=ORIGIN).plain
mention = mention_chip("Bob", None).plain
assert "@" not in mention # the chip says that it is a mention
assert sender.endswith(POWERLINE_SEP) and sender in rendered # the header goes on
assert mention.startswith(POWERLINE_SEP) and mention in rendered # the mention is reached
assert f"{mention} you around?" in rendered # in its place in the sentence
def test_a_mention_chip_moves_to_the_next_line_whole(powerline) -> None:
"""The wrap breaks a body at a space, but never at a space inside a mention chip.
A chip has a pad at each side of its label, and a name can have spaces. Without the
hold, the notch stays at the end of one line and the name starts the next line.
"""
from meshterm.ui.widgets import mention_chip
powerline(True)
conv = Conversation(label="#public", is_channel=True, channel_idx=0)
text = "Alice: the relay on the hill looks bad @[Bob Smith] you around?"
messages = [ChatMessage(text=text, is_channel=True, channel_idx=0)]
screen = ChatScreen(conv, messages, send=None, names={}, session=_StubSession())
mention = mention_chip("Bob Smith", None).plain
for width in range(30, 60):
lines = [_strip_ansi(line).rstrip() for line in screen._render_grouped(width)]
assert sum(mention.strip() in line for line in lines) == 1, width
assert all(cell_len(line) <= width for line in lines), width
def test_unidentified_sender_takes_the_node_grey_not_the_chrome_grey() -> None:
"""``·`` is a sender that has no place. It is content, so it is ``node.unknown``, not ``muted``.
The two styles are one hex value apart on the desktop and a whole slot apart on the
console (light grey and dark grey). On the console, a sender in the grey of the chrome
does not look like a node.
"""
conv = Conversation(label="#public", is_channel=True, channel_idx=0)
screen = ChatScreen(conv, [], send=None, names={}, session=_StubSession())
inbound = ChatMessage(text="hi", is_channel=True, channel_idx=0)
for sender in ("·", "nobody-knows-me"): # no prefix, and a name with no key and no hue
assert _ink(screen._group_header(sender, inbound), sender) == "node.unknown"
def _two_day_messages():
"""Two days of direct messages. They are long enough to overflow a small viewport."""
from datetime import datetime, timedelta, timezone
base = datetime(2026, 7, 5, 9, 0, tzinfo=timezone.utc)
day1 = [
ChatMessage(text=f"day1-{i}", peer="d4e5f6a7", created_at=base + timedelta(minutes=i))
for i in range(4)
]
day2 = [
ChatMessage(
text=f"day2-{i}", peer="d4e5f6a7", created_at=base + timedelta(days=1, minutes=i)
)
for i in range(4)
]
return day1 + day2
def test_chat_sticky_block_pins_the_governing_day_divider() -> None:
"""The day divider above the top row pins there when it scrolls off (as in the picker).
A day has only its divider, so each block is that one row. The heading of a select list
can also have its description in the block.
"""
screen = _screen(_StubSession(), send=None, messages=_two_day_messages())
screen.render_body(60)
(idx0, day0), (idx1, day1) = screen._sticky_headers
assert len(day0) == 1 and len(day1) == 1 # one divider, and no preamble under it
assert screen.sticky_block(0) == [] # the first divider is the top row
assert screen.sticky_block(idx1 - 1) == day0 # still in day one, so its divider pins
assert screen.sticky_block(idx1) == [] # the divider of day two is now the top row
assert screen.sticky_block(idx1 + 1) == day1 # scrolled past it, so the divider of day two pins
def test_chat_frame_pins_a_day_divider_when_stuck_to_the_newest() -> None:
"""When the frame renders the tail, a day divider is in the pinned top row."""
from meshterm.ui.tui import frame
screen = _screen(_StubSession(), send=None, messages=_two_day_messages())
lines = screen.render_body(60) # stays at the newest message, so early days scroll off
visible, above, _below = frame._visible_slice(screen, lines, 6)
top = _strip_ansi(visible[0]).strip()
assert top.startswith("──") and "Jul" in top # a day divider is pinned to the top row
assert above is True # and the frame shows that there is more above the pin
def test_chat_top_of_transcript_clears_the_more_above_flag() -> None:
"""If the user selects the oldest message, the head of the transcript scrolls into the viewport.
The selection of the first message anchors at line 0, so the frame goes to offset 0. Its
day divider and sender chip are real rows again, not a pin over a hidden line. The clip
arrow of the title bar is dim, because nothing is above.
"""
from meshterm.ui.tui import frame
screen = _screen(_StubSession(), send=None, messages=_two_day_messages())
lines = screen.render_body(60)
screen.handle("ctrl_home") # go to the oldest message
lines = screen.render_body(60)
visible, above, below = frame._visible_slice(screen, lines, 6)
assert screen.scroll == 0
assert above is False and below is True
assert _strip_ansi(visible[0]).strip().startswith("──") # the head, drawn and not pinned
def test_chat_home_end_and_word_keys_move_the_compose_cursor() -> None:
"""In a chat, Home, End, and Ctrl+←/→ act on the compose line, not on the transcript scroll."""
screen = _screen(_StubSession(), send=None, messages=_two_day_messages())
for ch in "hello world":
screen.handle("text", ch)
assert screen._editor.cursor == 11
screen.handle("home")
assert screen._editor.cursor == 0 # the start of the line, not a scroll to the top
screen.handle("end")
assert screen._editor.cursor == 11
screen.handle("ctrl_left")
assert screen._editor.cursor == 6 # the start of "world"
async def test_chat_paths_key_needs_a_picked_message() -> None:
"""^P shows the paths of the selected message. It does nothing until a message is selected."""
import asyncio
shown: list[str] = []
async def paths(message) -> None: # noqa: ANN001 - ChatMessage
shown.append(message.text)
messages = [
ChatMessage(text="first", peer="d4e5f6a7"),
ChatMessage(text="second", peer="d4e5f6a7"),
]
conv = Conversation(
label="Alice",
is_channel=False,
contact=Contact(name="Alice", public_key="d4" + "0" * 62, key_prefix="d4e5f6a7"),
)
screen = ChatScreen(conv, messages, send=None, names={}, session=_StubSession(), paths=paths)
# A path is the route of one message. If the user selects no message, there is nothing
# to show. The key does nothing, and the footer and the F-key lane both show this. They
# do not guess a message at the tail.
screen.handle("paths")
await asyncio.sleep(0)
assert shown == []
assert "^P" not in screen.footer_hint
assert screen.picocalc_lyra_lane[0].enabled is False
screen.handle("up")
screen.handle("up") # select the older message
screen.handle("paths")
await asyncio.sleep(0)
assert shown == ["first"]
# With a selected message, the screen offers paths. Here the key is Enter, because a
# direct chat has no reply to prepare. (The hint of a channel keeps the ^P atom beside
# its Enter reply.)
assert "Enter paths" in screen.footer_hint
assert screen.picocalc_lyra_lane[0].enabled is True
async def test_chat_direct_enter_on_a_pick_opens_paths() -> None:
"""In a direct chat, Enter on a selected message opens its paths (in a channel, it replies)."""
import asyncio
shown: list[str] = []
async def paths(message) -> None: # noqa: ANN001 - ChatMessage
shown.append(message.text)
messages = [ChatMessage(text="only", peer="d4e5f6a7")]
conv = Conversation(
label="Alice",
is_channel=False,
contact=Contact(name="Alice", public_key="d4" + "0" * 62, key_prefix="d4e5f6a7"),
)
screen = ChatScreen(conv, messages, send=None, names={}, session=_StubSession(), paths=paths)
screen.handle("up")
assert "Enter paths" in screen.footer_hint
screen.handle("enter")
await asyncio.sleep(0)
assert shown == ["only"]
def test_chat_pick_walks_with_page_keys_and_ctrl_home() -> None:
"""PgUp moves the selection one screen back. ^Home moves it to the first message."""
screen = _screen(_StubSession(), send=None, messages=_two_day_messages())
lines = screen.render_body(60)
screen.note_metrics(total=len(lines), viewport=5)
screen.handle("pageup")
assert screen._stick is False and screen._selected is not None
screen.handle("ctrl_home")
assert screen._selected == 0 # the first message
screen.handle("ctrl_end")
assert screen._stick is True and screen._selected is None
def test_chat_ctrl_page_picks_across_day_dividers() -> None:
"""Ctrl+PageDown and Ctrl+PageUp move the selection between days (in both types of chat)."""
screen = _screen(_StubSession(), send=None, messages=_two_day_messages())
starts = screen._day_start_indices()
assert len(starts) == 2
screen.handle("ctrl_home")
assert screen._selected == 0
screen.handle("ctrl_pagedown")
assert screen._selected == starts[1] # the first message of the second day
screen.handle("ctrl_pageup")
assert screen._selected == starts[0] # back to the first day
def _two_day_channel_messages():
"""Channel messages from two local days (one sender), for the tests of section jumps."""
from datetime import datetime, timedelta, timezone
base = datetime(2026, 7, 5, 9, 0, tzinfo=timezone.utc)
day1 = [
ChatMessage(
text=f"Alice: d1-{i}",
is_channel=True,
channel_idx=0,
created_at=base + timedelta(minutes=i),
)
for i in range(3)
]
day2 = [
ChatMessage(
text=f"Alice: d2-{i}",
is_channel=True,
channel_idx=0,
created_at=base + timedelta(days=1, minutes=i),
)
for i in range(3)
]
return day1 + day2
def test_chat_channel_ctrl_page_selects_across_days() -> None:
"""In a channel, Ctrl+PageDown and Ctrl+PageUp move the reply selection between days."""
screen = _channel_screen(_two_day_channel_messages())
starts = screen._day_start_indices()
assert starts == [0, 3]
screen.handle("ctrl_home")
assert screen._selected == 0
screen.handle("ctrl_pagedown")
assert screen._selected == starts[1] # the first message of the second day
screen.handle("ctrl_pageup")
assert screen._selected == starts[0] # back to the first day
def test_chat_channel_home_moves_the_compose_cursor() -> None:
"""The Home key of a channel acts on the compose line. It does not move the selection."""
screen = _channel_screen(_two_day_channel_messages())
for ch in "reply":
screen.handle("text", ch)
assert screen._editor.cursor == 5
screen.handle("home")
assert screen._editor.cursor == 0
assert screen._selected is None # a change to the compose line clears the reply selection
def test_channel_self_style_keyed_on_concept_not_label() -> None:
"""Our own chip depends on the message being outbound, not on the 'you' label.
A remote sender with the name 'you' must not take the white style that is only for us.
It reads as a normal sender (the hue of its key if the key resolves, and muted if not).
"""
from meshterm.ui.theme import node_style
ours = ChatMessage(text="hi", is_channel=True, channel_idx=0, outbound=True)
theirs = ChatMessage(text="you: hi", is_channel=True, channel_idx=0)
mine = _channel_screen([], self_name="Me", self_key=_OUR_KEY)
white = _ink(mine._group_header("you", ours), " ★ Me ") # our name, in white
assert white.startswith("bold #ffffff on ")
screen = _channel_screen([])
assert _ink(screen._group_header("you", ours), "★") == "you" # no name: the white star
assert _ink(screen._group_header("you", theirs), "you") == "node.unknown" # no key: grey
keyed = _channel_screen([], key_of=_keys_of({"you": "d4" + "0" * 62}))
remote = _ink(keyed._group_header("you", theirs), "you")
assert remote == node_style("d4") # the spectrum hue of the key
def test_channel_own_messages_do_not_merge_with_remote_namesake() -> None:
"""A remote sender with the exact name 'you' is a separate group from our own messages."""
from datetime import datetime, timezone
base = datetime(2026, 7, 5, 14, 24, tzinfo=timezone.utc)
messages = [
ChatMessage(text="you: impostor", is_channel=True, channel_idx=0, created_at=base),
ChatMessage(text="mine", outbound=True, is_channel=True, channel_idx=0, created_at=base),
]
screen = _channel_screen(messages, self_name="Me", self_key=_OUR_KEY)
rendered = _strip_ansi("\n".join(screen._render_grouped(80)))
# Two headers, and they are not only apart: theirs is the name that they sent, ours is
# our name after the ★.
assert rendered.count("you") == 1 and rendered.count("★ Me") == 1
def test_a_blank_line_closes_each_author_where_the_rows_allow(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""By default, a terminal of 24 rows or more puts a blank line after each author.
The blank line also comes before the divider of a new day, and never before the first
line. A terminal of fewer rows (the 14 of the Cardputer Zero) has no row to spare for
it. A second paint reads the cached lines, and the divider of the second day still pins
at its own line, not at the blank line before it. The two ends of the
``section_spacing_min_rows`` preference ignore the rows, and a change applies at the
next paint.
"""
from datetime import datetime, timedelta, timezone
from meshterm.core import preferences
base = datetime(2026, 7, 5, 14, 24, tzinfo=timezone.utc)
def at(minutes: int) -> datetime:
return base + timedelta(minutes=minutes)
def transcript() -> list[ChatMessage]:
return [
ChatMessage(text="Alice: hi", is_channel=True, channel_idx=0, created_at=at(0)),
ChatMessage(text="Alice: again", is_channel=True, channel_idx=0, created_at=at(1)),
ChatMessage(text="Bob: yo", is_channel=True, channel_idx=0, created_at=at(2)),
ChatMessage(text="Bob: later", is_channel=True, channel_idx=0, created_at=at(1440)),
]
prefs = preferences.Preferences()
monkeypatch.setattr(preferences, "current", lambda: prefs)
screen = _channel_screen(transcript())
screen.note_terminal_rows(24)
lines = [_strip_ansi(line) for line in screen._render_grouped(53)]
assert lines[0].startswith("──") and lines[1] == "Alice"
assert lines[4] == "" and lines[5] == "Bob" # after the two messages of Alice
assert lines[7] == "" and lines[8].startswith("── Mon") # after Bob, before the new day
assert lines.count("") == 2
first = list(screen._sticky_headers)
screen._render_grouped(53) # the cached lines
assert screen._sticky_headers == first and [at for at, _ in first] == [0, 8]
short = _channel_screen(transcript())
short.note_terminal_rows(14)
assert "" not in [_strip_ansi(line) for line in short._render_grouped(53)]
prefs.set("section_spacing_min_rows", 0) # any height, also 14 rows
assert "" in [_strip_ansi(line) for line in short._render_grouped(53)]
prefs.set("section_spacing_min_rows", 10_000) # never, and the cached lines change too
assert "" not in [_strip_ansi(line) for line in screen._render_grouped(53)]
def test_a_mention_of_our_name_takes_our_chip_with_no_star(powerline) -> None:
"""The chip of a mention of us is white, as our sender chip, but it has no star.
The mark of a message that names us is at the left of its time (refer to the next
test). With chips, no mention has its ``@``.
"""
powerline(True)
screen = _channel_screen([], self_name="Me", self_key=_OUR_KEY)
text = screen._render_mentions("hi @[Me] and @[Bob]", selected=False)
assert "★" not in text.plain and "@" not in text.plain
me = next(s for s in text.spans if text.plain[s.start : s.end] == "Me")
assert str(me.style).startswith("bold #ffffff on ") # the white label
def test_a_message_that_names_us_has_the_mark_left_of_its_time() -> None:
"""The mark goes in the two cells before the time, and each time moves two cells right.
The mark is a ``★`` in our hue (refer to the widget tests for its colour).
"""
from datetime import datetime, timezone
at = datetime(2026, 7, 5, 14, 24, tzinfo=timezone.utc)
stamp = at.astimezone().strftime("%H:%M")
messages = [
ChatMessage(text="Bob: hi @[Me]", is_channel=True, channel_idx=0, created_at=at),
ChatMessage(text="Bob: hi @[Zed]", is_channel=True, channel_idx=0, created_at=at),
]
screen = _channel_screen(messages, self_name="Me", self_key=_OUR_KEY)
rows = _strip_ansi("\n".join(screen._render_grouped(72))).splitlines()
assert any(row.startswith(f" ★ {stamp} hi") for row in rows)
assert any(row.startswith(f" {stamp} hi") for row in rows) # the other message
screen.handle("up")
screen.handle("up") # the first message: the pointer, then the mark
rows = _strip_ansi("\n".join(screen._render_grouped(72))).splitlines()
assert any(row.startswith(f"❯ ★ {stamp} hi") for row in rows)
def _channel_screen(
messages, session=None, key_of=None, self_name=None, self_key=None
) -> ChatScreen:
"""Build a channel ChatScreen for ``messages``, for the selection and reply tests."""
conv = Conversation(label="#public", is_channel=True, channel_idx=0)
return ChatScreen(
conv,
messages,
send=None,
names={},
session=session or _StubSession(),
key_of=key_of,
self_name=self_name,
self_key=self_key,
)
def _channel_messages():
"""Three inbound channel messages (Alice, Alice, Bob) for the reply-selection tests."""
from datetime import datetime, timezone
base = datetime(2026, 7, 5, 14, 24, tzinfo=timezone.utc)
return [
ChatMessage(text="Alice: hi", is_channel=True, channel_idx=0, created_at=base),
ChatMessage(text="Alice: still here", is_channel=True, channel_idx=0, created_at=base),
ChatMessage(text="Bob: yo", is_channel=True, channel_idx=0, created_at=base),
]
def test_channel_up_enters_selection_from_newest() -> None:
"""No message is selected until ↑ selects the newest. Then ↑ moves toward older messages."""
screen = _channel_screen(_channel_messages())
assert screen._selected is None # compose has the focus: nothing is selected at the start
screen.handle("up")
assert screen._selected == 2 # the newest message
assert screen._stick is False
screen.handle("up")
assert screen._selected == 1 # moves to the previous message
def test_channel_down_past_newest_clears_selection() -> None:
"""↓ after the newest message returns the focus to compose (nothing is selected)."""
screen = _channel_screen(_channel_messages())
screen.handle("up") # select the newest (index 2)
assert screen._selected == 2
screen.handle("down") # after the newest: deselect, and stay at the tail again
assert screen._selected is None
assert screen._stick is True
def test_channel_selected_line_tracked_for_scroll() -> None:
"""The render records the body line of the selected message, so the frame keeps it visible."""
screen = _channel_screen(_channel_messages())
screen.handle("up") # select Bob (the newest)
screen.render_body(80)
assert screen._selected_line is not None
# No selection means no cursor line, so the transcript scrolls freely, as before.
screen.handle("end")
screen.render_body(80)
assert screen._selected_line is None
def test_channel_enter_on_selection_primes_at_mention() -> None:
"""Enter on a selected message puts the @mention of the sender in the compose line."""
screen = _channel_screen(_channel_messages())
screen.handle("up")
screen.handle("up") # select an Alice message (index 1)
screen.handle("enter")
assert screen._editor.text == "@[Alice] "
assert screen._selected is None # the focus returns to compose after the reply starts
def test_each_reply_puts_its_mention_at_the_cursor() -> None:
"""Two replies name two nodes in one message, each where the cursor was."""
screen = _channel_screen(_channel_messages())
for ch in "hi and":
screen.handle("text", ch)
screen.handle("ctrl_left") # before "and"
screen.handle("up") # Bob
screen.handle("enter")
assert screen._editor.text == "hi @[Bob] and"
screen.handle("end")
screen.handle("up")
screen.handle("up") # Alice
screen.handle("enter")
assert screen._editor.text == "hi @[Bob] and @[Alice] "
assert screen._editor.cursor == len(screen._editor.text)
screen.handle("backspace") # the space
screen.handle("backspace") # all of the mention
assert screen._editor.text == "hi @[Bob] and "
def test_the_compose_line_keeps_a_chip_whole_when_it_wraps(powerline) -> None:
"""A chip in the compose line moves to the next line as one word, never split."""
from meshterm.ui.widgets import mention_chip
powerline(True)
screen = _channel_screen([])
for ch in "the relay on the hill looks bad @[Bob Smith] ok":
screen.handle("text", ch)
chip = mention_chip("Bob Smith", None).plain
for width in range(30, 60):
lines = _strip_ansi(screen.render_body(width)).splitlines()
assert sum(chip in line for line in lines) == 1, width
def test_a_row_whose_last_message_names_us_has_the_mark_left_of_its_age(
repo: Repository,
) -> None:
"""The Messages list marks the row as the transcript marks the message."""
conv = Conversation(label="General", is_channel=True, channel_idx=0, channel_id="c0")
ctx = _RowCtx(repo)
us = ("Me", _OUR_KEY)
named = ChatMessage(text="Bob: hi @[Me]", is_channel=True, channel_id="c0")
other = ChatMessage(text="Bob: hi @[Zed]", is_channel=True, channel_id="c0")
lanes = _lanes([conv])
marked = _title(ctx, conv, {"chan:c0": named}, _NO_KEYS, lanes, us=us).plain
plain = _title(ctx, conv, {"chan:c0": other}, _NO_KEYS, lanes, us=us).plain
assert "★ now " in marked and "★" not in plain
assert marked.index("now") == plain.index("now") # the mark takes the cells of the gap
def test_channel_typing_clears_selection() -> None:
"""A change to the compose line clears the reply selection (compose has the focus)."""
screen = _channel_screen(_channel_messages())
screen.handle("up")
assert screen._selected is not None
screen.handle("text", "x")
assert screen._selected is None
assert screen._editor.text == "x"
def test_chat_screen_escape_cancels() -> None:
"""Esc resolves the future of the screen with CANCEL, so the caller pops it."""
screen = _screen(_StubSession(), send=None)
loop = asyncio.new_event_loop()
try:
screen.future = loop.create_future()
screen.handle("escape")
assert screen.future.result() is CANCEL
finally:
loop.close()
async def test_service_send_records_outbound(repo: Repository) -> None:
"""A send through the service stores the outbound message with its ack state."""
device = MockDevice()
ctx = _StubContext(device, repo)
chat = ChatService(ctx)
await chat.start()
try:
contact = next(c for c in await device.get_contacts() if c.name == "Alice")
sent = await chat.send_direct(contact, "hey")
assert sent.outbound is True and sent.acked is True
await chat.send_channel(0, "hello all", label="#public")
channel_id = await chat.channel_id_for(0)
stored = repo.recent_chat_messages(is_channel=True, channel_id=channel_id)
assert [m.text for m in stored] == ["hello all"]
assert stored[0].outbound is True
finally:
await chat.stop()
await device.disconnect()
async def test_service_resend_updates_ack_in_place(repo: Repository) -> None:
"""A resend of a failed message changes its stored ack. It does not add a duplicate row."""
device = MockDevice()
ctx = _StubContext(device, repo)
chat = ChatService(ctx)
await chat.start()
try:
contact = next(c for c in await device.get_contacts() if c.name == "Alice")
peer = contact.key_prefix or contact.public_key[:12]
failed = ChatMessage(
text="retry me", outbound=True, peer=peer, peer_name=contact.name, acked=False
)
failed.row_id = repo.record_chat_message(failed)
await chat.resend_direct(contact, failed)
assert failed.acked is True # the mock always acknowledges
stored = repo.recent_chat_messages(is_channel=False, peer=peer)
assert len(stored) == 1 # changed in place, not duplicated
assert stored[0].acked is True
finally:
await chat.stop()
await device.disconnect()
class _ScriptedDevice:
"""A device stub. The acks of its direct messages follow a fixed script.
Each :meth:`send_direct_message` takes the next value from ``acks``. The value is an
:class:`Ack` stand-in (any object that is not ``None`` counts as delivered), or ``None``
for a transmission with no ack. The method also appends the sent text to :attr:`sent`,
so that a test can assert how many soft retries happened. When the script has no more
values, each further send has no ack. Each send gets its own ack code, as the radio does.
"""
def __init__(self, acks: list) -> None:
self._acks = list(acks)
self.sent: list[str] = []
async def connect(self) -> None: # for _StubContext.device()
return None
async def send_direct_message(self, contact: Contact, text: str):
from meshterm.core.models import Delivery
self.sent.append(text)
ack = self._acks.pop(0) if self._acks else None
return Delivery(f"{len(self.sent):08x}", ack)
def _contact() -> Contact:
return Contact(name="Alice", public_key="d4e5" + "0" * 60, key_prefix="d4e5f6a7")
async def test_send_direct_soft_retries_until_acked(repo: Repository) -> None:
"""A DM with no ack on the first tries is sent again. It is stored as delivered after an ack."""
device = _ScriptedDevice([None, None, object()]) # an ack only on the third try
ctx = _StubContext(device, repo)
ctx.preferences.direct_message_soft_retries = 2 # 1 send and 2 retries make 3 tries
chat = ChatService(ctx)
sent = await chat.send_direct(_contact(), "hey")
assert sent.acked is True
assert device.sent == ["hey", "hey", "hey"] # three transmissions
async def test_send_direct_soft_retries_capped_by_setting(repo: Repository) -> None:
"""The retry budget stops the resends. A DM that never gets an ack is tried retries+1 times."""
device = _ScriptedDevice([]) # never acknowledges
ctx = _StubContext(device, repo)
ctx.preferences.direct_message_soft_retries = 2
chat = ChatService(ctx)
sent = await chat.send_direct(_contact(), "hey")
assert sent.acked is False
assert len(device.sent) == 3 # the cap: one send and two soft retries at the most
async def test_send_direct_zero_retries_is_one_shot(repo: Repository) -> None:
"""If soft retries are off, a DM is transmitted one time, with an ack or without it."""
device = _ScriptedDevice([]) # never acknowledges
ctx = _StubContext(device, repo)
ctx.preferences.direct_message_soft_retries = 0
chat = ChatService(ctx)
sent = await chat.send_direct(_contact(), "hey")
assert sent.acked is False
assert device.sent == ["hey"] # one transmission, no soft retry
# -- end-to-end through the real session --------------------------------------
async def test_open_chat_sends_through_real_session(tmp_path: Path) -> None:
"""open_chat, driven with piped keys, sends a message and stores it, end to end."""
from prompt_toolkit.input.defaults import create_pipe_input
from prompt_toolkit.output import DummyOutput
from rich.console import Console
from meshterm.context import AppContext
from meshterm.core.admin_store import AdminStore
from meshterm.core.device_store import DeviceStore
from meshterm.ui.chat import open_chat
from meshterm.ui.surface import TuiUi
from meshterm.ui.tui.session import TuiSession
settings = Settings(config_dir=tmp_path, db_path=tmp_path / "e2e.db")
ctx = AppContext(
console=Console(),
settings=settings,
repo=Repository(settings.db_path),
device_store=DeviceStore(tmp_path / "devices.json"),
admin_store=AdminStore(tmp_path / "admin.json"),
mock=True,
)
conv = Conversation(
label="Alice",
is_channel=False,
contact=Contact(name="Alice", public_key="d4e5f6a7" + "0" * 56, key_prefix="d4e5f6a7"),
)
try:
with create_pipe_input() as inp:
session = TuiSession(input=inp, output=DummyOutput())
ctx.ui = TuiUi(session)
def stored_texts() -> list[str]:
rows = ctx.repo.recent_chat_messages(is_channel=False, peer="d4e5f6a7")
return [m.text for m in rows]
async def drive() -> None:
# Send Esc only after the app stores the send. The send runs in a task that
# the screen starts. If the test pipes Esc next to Enter, a slow runner can
# leave (and cancel the task) before the app writes the row.
inp.send_text("hi\r") # type "hi", then Enter (send)
while not stored_texts():
await asyncio.sleep(0.01)
inp.send_text("\x1b") # Esc (leave)
async def main() -> None:
driver = asyncio.ensure_future(drive())
try:
await open_chat(ctx, conv)
finally:
driver.cancel()
await asyncio.wait_for(session.run(main()), timeout=5)
stored = ctx.repo.recent_chat_messages(is_channel=False, peer="d4e5f6a7")
assert [m.text for m in stored] == ["hi"]
assert stored[0].outbound is True
finally:
await ctx.aclose()
# -- the picker's companion filter and history delete -------------------------
def test_delete_chat_history_removes_only_that_peer(repo: Repository) -> None:
"""If the user deletes one direct thread, the other peers and the channel history stay."""
repo.record_chat_message(ChatMessage(text="a", peer="aa"))
repo.record_chat_message(ChatMessage(text="b", peer="bb"))
repo.record_chat_message(ChatMessage(text="c", is_channel=True, channel_id="c0"))
assert repo.delete_chat_history("aa") == 1
assert repo.recent_chat_messages(is_channel=False, peer="aa") == []
assert [m.text for m in repo.recent_chat_messages(is_channel=False, peer="bb")] == ["b"]
assert [m.text for m in repo.recent_chat_messages(is_channel=True, channel_id="c0")] == ["c"]
class _PickerDevstate:
def __init__(self, contacts) -> None:
self._contacts = contacts
async def channel_slots(self):
return []
async def contacts(self):
return list(self._contacts)
class _PickerChat(_FakeChat):
def __init__(self, unread=None) -> None:
super().__init__(unread or {})
self.cleared: list[str] = []
def clear_unread(self, key: str) -> None:
self.cleared.append(key)
async def test_picker_lists_companions_only(repo: Repository) -> None:
"""Only companion contacts (and contacts of an unknown type) appear in Direct, not repeaters."""
from types import SimpleNamespace
from meshterm.core.models import NODE_TYPE_REPEATER
from meshterm.tools.chat import ChatTool
from meshterm.ui.tui import Choice
contacts = [
Contact(name="Ally", public_key="d4" + "0" * 62, node_type=1),
Contact(name="Mystery", public_key="60" + "0" * 62), # the type was not in an advert
Contact(name="Tower", public_key="a1" + "0" * 62, node_type=NODE_TYPE_REPEATER),
]
ctx = SimpleNamespace(devstate=_PickerDevstate(contacts), repo=repo, chat=_PickerChat())
items = await ChatTool()._picker_items(ctx)
direct = [
it.value.label
for it in items
if isinstance(it, Choice) and isinstance(it.value, Conversation) and not it.value.is_channel
]
assert "Ally" in direct and "Mystery" in direct and "Tower" not in direct
async def test_picker_pins_its_column_header_over_the_group_heading(
repo: Repository,
) -> None:
"""On a page scrolled into Direct, the lane names stay at the top, over ``Direct``."""
from types import SimpleNamespace
from meshterm.tools.chat import ChatTool
from meshterm.ui.tui import SelectScreen, frame
ansi = re.compile(r"\x1b\[[0-9;]*m")
contacts = [
Contact(name=f"Peer{i:02d}", public_key=f"{i:02x}" + "0" * 62, node_type=1)
for i in range(12)
]
ctx = SimpleNamespace(devstate=_PickerDevstate(contacts), repo=repo, chat=_PickerChat())
items = await ChatTool()._picker_items(ctx)
header = items[0]
assert header.pinned # the lanes mean the same in both groups, so they pin for the list
screen = SelectScreen("Chat", items)
for _ in range(8): # down, after the Direct heading
screen.handle("down")
visible, above, _below = frame._visible_slice(screen, screen.render_body(72), 8)
top = [ansi.sub("", row).strip() for row in visible[:2]]
# Short words over a badge and an age, then the message.
assert top[0].split() == ["CONVERSATION", "NEW", "LAST", "MESSAGE"]
assert top[1] == "── Direct ──"
assert above is True
# If the width is too small for the whole line, the last label gets shorter and does
# not wrap. The lane has the size of these six-cell names, so this starts at a width
# much smaller than any real terminal. The test is for the fallback, not for the width
# where it starts.
assert header.text(36).endswith(" MSG")
assert "\n" not in header.text(30)
async def test_picker_lane_is_sized_to_the_names_it_holds(repo: Repository) -> None:
"""The conversation lane fits the longest name, between its minimum and its maximum.
Each cell of the lane after the longest name is padding that the app takes from the
message preview. The preview is the only content of the row that has no limit. Thus the
app measures the lane and does not use a fixed number.
"""
from meshterm.tools.chat import _LABEL_MAX, _LABEL_MIN, _lanes
def lane(*names: str) -> int:
return _lanes([Conversation(label=n, is_channel=True, channel_id=n) for n in names]).label
assert lane("Bob", "Ann") == _LABEL_MIN # short names cannot make it smaller than the header
assert lane("Bob", "YUL-Cartierville") == len("YUL-Cartierville")
assert lane("A Rather Long Contact Name Indeed") == _LABEL_MAX # and it has a maximum
async def test_picker_rows_scroll_the_message_under_pinned_lanes(repo: Repository) -> None:
"""In a row, ←→ scroll only the preview. The lanes before it stay in place.
Those lanes show the place of the user in the list, so the scroll starts where the
message starts. This is one column for each row, with a long name or a short name.
"""
from types import SimpleNamespace
from meshterm.tools.chat import ChatTool
from meshterm.ui.tui import Choice
contacts = [
Contact(name="Alice", public_key="d4" + "0" * 62, node_type=1),
Contact(name="A Rather Long Contact Name", public_key="60" + "0" * 62, node_type=1),
]
alice = Conversation(label="Alice", is_channel=False, contact=contacts[0])
repo.record_chat_message(ChatMessage(text="x" * 200, peer=alice.peer))
ctx = SimpleNamespace(devstate=_PickerDevstate(contacts), repo=repo, chat=_PickerChat())
rows = [it for it in await ChatTool()._picker_items(ctx) if isinstance(it, Choice)]
heads = {row.hscroll_from for row in rows}
assert len(heads) == 1 and heads.pop() > 0 # one head block for the whole list
talked = next(row for row in rows if row.value.label == "Alice")
line = talked.label.plain
# The test measures in cells, not in characters: a channel glyph is one character and two cells.
assert cell_len(line[: line.index("x")]) == talked.hscroll_from
# Also, the preview keeps the whole message. The row makes the cut, and ←→ move past it.
assert talked.label.plain.endswith("x" * 200)
def test_picker_marker_lane_lines_the_two_groups_up(repo: Repository) -> None:
"""A channel glyph and a contact dot use the same cells, on each platform.
That glyph is two cells wide on regular and one cell wide in the console font. Thus the
dot gets padding to the width that the glyph measures, not to a fixed width. A fixed
padding put each channel row one cell out of line with each direct row on the PicoCalc.
"""
from meshterm.platforms import PICOCALC_LYRA, REGULAR, set_platform
from meshterm.tools.chat import _lanes
channel = Conversation(label="Public", is_channel=True, channel_idx=0, channel_id="c0")
contact = Conversation(
label="Alice", is_channel=False, contact=Contact(name="Alice", public_key="d4" * 32)
)
ctx = _RowCtx(repo)
try:
for platform in (REGULAR, PICOCALC_LYRA):
set_platform(platform)
lanes = _lanes([channel, contact])
rows = [_title(ctx, c, {}, _NO_KEYS, lanes).plain for c in (channel, contact)]
names = ("Public", "Alice")
at = [cell_len(r[: r.index(n)]) for r, n in zip(rows, names, strict=True)]
assert at[0] == at[1] == lanes.marker
finally:
set_platform(REGULAR)
async def test_picker_footer_fits_with_every_atom_showing(repo: Repository, monkeypatch) -> None:
"""The scroll atom, the erase atom, and the base atoms together fit in the 72-cell footer.
The three groups of atoms show together on an ordinary row: a thread with a history that
is long enough to scroll. Thus the budget is a limit on the wording, not a rare case.
(This is why the atom is ``Del erase``: the row already names the conversation whose
history the key deletes.)
"""
import asyncio
from types import SimpleNamespace
from prompt_toolkit.input.defaults import create_pipe_input
from prompt_toolkit.output import DummyOutput
from meshterm.tools.chat import ChatTool
from meshterm.ui.surface import TuiUi
from meshterm.ui.tui.screen import CANCEL
from meshterm.ui.tui.session import TuiSession
ally = Contact(name="Ally", public_key="d4" + "0" * 62, node_type=1)
conv = Conversation(label="Ally", is_channel=False, contact=ally)
repo.record_chat_message(ChatMessage(text="y" * 200, peer=conv.peer))
ctx = SimpleNamespace(devstate=_PickerDevstate([ally]), repo=repo, chat=_PickerChat())
hints: list[str] = []
with create_pipe_input() as inp:
session = TuiSession(input=inp, output=DummyOutput())
ctx.ui = TuiUi(session)
async def main() -> None:
run = asyncio.ensure_future(ChatTool()._run_live(ctx))
loop = asyncio.get_running_loop()
deadline = loop.time() + 2
while not session._stack:
assert loop.time() < deadline, "the picker never opened"
await asyncio.sleep(0)
picker = session._stack[-1]
picker.render_body(72) # the overflow probe reads the last render width
picker.handle("down") # to the one Direct row: it can be deleted, and it overflows
picker.render_body(72)
hints.append(picker.footer_hint)
picker.resolve(CANCEL)
await asyncio.wait_for(run, timeout=2)
await asyncio.wait_for(session.run(main()), timeout=5)
hint = hints[0]
assert cell_len(hint) <= 72, f"{cell_len(hint)} cells: {hint!r}"
for atom in ("←→ scroll", "Enter open", "Del erase", "Esc back"):
assert atom in hint
async def test_picker_del_deletes_history_after_a_red_confirm(repo: Repository) -> None:
"""Del on a thread with history deletes the history, after a red confirm.
The confirm is the destructive type. Afterwards, the history is gone, the unread count
is cleared, and the rebuilt rows show the thread as hollow and not deletable.
"""
from types import SimpleNamespace
from meshterm.tools.chat import ChatTool
ally = Contact(name="Ally", public_key="d4" + "0" * 62, node_type=1)
conv = Conversation(label="Ally", is_channel=False, contact=ally)
repo.record_chat_message(ChatMessage(text="hi", peer=conv.peer))
repo.record_chat_message(ChatMessage(text="chan", is_channel=True, channel_id="c0"))
class _Ui:
def __init__(self) -> None:
self.dialogs: list[dict] = []
async def dialog(self, prompt, buttons, **kw):
self.dialogs.append({"prompt": prompt, "buttons": buttons, **kw})
return True # select Delete
chat = _PickerChat()
ctx = SimpleNamespace(devstate=_PickerDevstate([ally]), repo=repo, ui=_Ui(), chat=chat)
tool = ChatTool()
assert _direct_row(await tool._picker_items(ctx)).deletable # there is history to delete
await tool._delete_history(ctx, conv)
confirm = ctx.ui.dialogs[0]
assert confirm["destructive"] is True # the reserved red, for data loss
assert confirm["buttons"] == [("Cancel", False), ("Delete", True)]
assert repo.recent_chat_messages(is_channel=False, peer=conv.peer) == []
assert [m.text for m in repo.recent_chat_messages(is_channel=True, channel_id="c0")] == [
"chan"
] # the channel history stays
assert chat.cleared == [conv.key]
# The new rows show this: the history is gone, so the row changes to a contact with no history.
assert not _direct_row(await tool._picker_items(ctx)).deletable
async def test_picker_del_cancel_keeps_the_history(repo: Repository) -> None:
"""If the user cancels the confirm, the messages of the thread stay."""
from types import SimpleNamespace
from meshterm.tools.chat import ChatTool
ally = Contact(name="Ally", public_key="d4" + "0" * 62, node_type=1)
conv = Conversation(label="Ally", is_channel=False, contact=ally)
repo.record_chat_message(ChatMessage(text="hi", peer=conv.peer))
class _Ui:
async def dialog(self, prompt, buttons, **kw):
return False # Cancel is the way out
chat = _PickerChat()
ctx = SimpleNamespace(devstate=_PickerDevstate([ally]), repo=repo, ui=_Ui(), chat=chat)
await ChatTool()._delete_history(ctx, conv)
assert [m.text for m in repo.recent_chat_messages(is_channel=False, peer=conv.peer)] == ["hi"]
assert chat.cleared == []
def _direct_row(items: list):
"""The one Direct row in a built picker list (the tests above have a single contact)."""
from meshterm.ui.tui import Choice
return next(
it
for it in items
if isinstance(it, Choice) and isinstance(it.value, Conversation) and not it.value.is_channel
)
async def test_the_conversation_picker_stays_pushed_under_an_open_chat(
repo: Repository, monkeypatch
) -> None:
"""Esc in a chat is one pop, back to the row from which the chat opened.
The app once gathered the picker in ``prompt_params`` and rebuilt it in each round. A
``default=`` restore put the highlight back. Now the app keeps the one screen, so the
typed filter and the scroll stay with it. The app changes the rows in place, so a thread
that has new messages is where its recency puts it.
"""
import asyncio
from types import SimpleNamespace
from prompt_toolkit.input.defaults import create_pipe_input
from prompt_toolkit.output import DummyOutput
from meshterm.tools.chat import ChatTool
from meshterm.ui.surface import TuiUi
from meshterm.ui.tui.screen import CANCEL
from meshterm.ui.tui.session import TuiSession
ally = Contact(name="Ally", public_key="d4" + "0" * 62, node_type=1)
ctx = SimpleNamespace(devstate=_PickerDevstate([ally]), repo=repo, chat=_PickerChat())
depths: list[int] = []
with create_pipe_input() as inp:
session = TuiSession(input=inp, output=DummyOutput())
ctx.ui = TuiUi(session)
async def fake_open_chat(_ctx, conversation) -> int:
"""Stand in for the chat screen: note the stack under it, then leave."""
depths.append(len(session._stack))
return 3
monkeypatch.setattr("meshterm.ui.chat.open_chat", fake_open_chat)
async def main() -> None:
run = asyncio.ensure_future(ChatTool()._run_live(ctx))
loop = asyncio.get_running_loop()
deadline = loop.time() + 2
while not session._stack:
assert loop.time() < deadline, "the picker never opened"
await asyncio.sleep(0)
picker = session._stack[-1]
picker.resolve(Conversation(label="Ally", is_channel=False, contact=ally))
while not depths:
await asyncio.sleep(0)
picker.resolve(CANCEL) # Esc in the picker goes to the menu
result = await asyncio.wait_for(run, timeout=2)
assert result.summary == {"conversations": 1, "messages": 3}
await asyncio.wait_for(session.run(main()), timeout=5)
assert depths == [1], "the picker is still on the stack while the chat runs"
assert session._stack == [], "and the visit pops it on the way out"
# -- the badge rule: what raises the unread count of the header ----------------
class _GatedContext(_StubContext):
"""A context with the device-state cache, so the badge rule can ask the device.
The plain :class:`_StubContext` has no cache. This is the path where the app gives the
message the benefit of the doubt: a message notifies when there is no device state to
place it against.
"""
def __init__(self, device: MockDevice, repo: Repository) -> None:
super().__init__(device, repo)
from meshterm.services.device_state import DeviceState
self.devstate = DeviceState(self)
async def _record(chat: ChatService, ctx, message: Message) -> None:
"""Publish one inbound message and wait until the worker files it."""
ctx.events.publish(MeshEvent.message_event(message))
await chat._queue.join()
async def test_badge_ignores_direct_messages_from_non_recipients(repo: Repository) -> None:
"""The direct messages of a repeater go into the history, but they do not raise the badge.
A repeater is not a DM recipient, so the picker does not list it. But a repeater does
send us direct messages: each reply of the remote CLI from a repeater that you administer
is one. If the app counted them, the badge pointed at a conversation with no row, and
the user could not clear it.
"""
device = MockDevice() # its Yagi-Repeater contact is a NODE_TYPE_REPEATER
ctx = _GatedContext(device, repo)
chat = ChatService(ctx)
await chat.start()
try:
await _record(chat, ctx, Message(text="> ok", sender="a1b2c3d4"))
assert chat.unread("dm:a1b2c3d4") == 0
# Silent, but not dropped: the reply is in the transcript.
stored = repo.recent_chat_messages(is_channel=False, peer="a1b2c3d4")
assert [m.text for m in stored] == ["> ok"]
finally:
await chat.stop()
await device.disconnect()
async def test_badge_ignores_a_sender_with_no_contact(repo: Repository) -> None:
"""A sender that is not in the contact table is stored silently. There is no row to open."""
device = MockDevice()
ctx = _GatedContext(device, repo)
chat = ChatService(ctx)
await chat.start()
try:
await _record(chat, ctx, Message(text="hi", sender="ffee11223344"))
assert chat.unread("dm:ffee11223344") == 0
assert repo.recent_chat_messages(is_channel=False, peer="ffee11223344")
finally:
await chat.stop()
await device.disconnect()
async def test_badge_ignores_a_channel_the_device_has_no_slot_for(repo: Repository) -> None:
"""A channel message that has no configured slot stays silent.
There are two cases. The slot can be empty, so the message has the fallback identity from
the slot, which names no channel. Or the device does not list the slot. In both cases the
picker has no row.
"""
device = MockDevice()
await device.connect()
device._channels[0] = {
"channel_idx": 0,
"channel_name": "Public",
"channel_secret": DEFAULT_PUBLIC_SECRET,
}
ctx = _GatedContext(device, repo)
chat = ChatService(ctx)
await chat.start()
try:
await _record(chat, ctx, Message(text="Bob: yo", channel=3, is_channel=True))
assert chat.unread("chan:slot:3") == 0
assert repo.recent_chat_messages(is_channel=True, channel_id="slot:3")
finally:
await chat.stop()
await device.disconnect()
async def test_badge_counts_a_configured_channel_and_a_companion(repo: Repository) -> None:
"""The positive control: a message that the picker lists still raises the badge."""
from meshterm.core.channels import channel_identity
device = MockDevice()
await device.connect()
device._channels[0] = {
"channel_idx": 0,
"channel_name": "Public",
"channel_secret": DEFAULT_PUBLIC_SECRET,
}
ctx = _GatedContext(device, repo)
chat = ChatService(ctx)
await chat.start()
try:
await _record(chat, ctx, Message(text="Ann: hey", channel=0, is_channel=True))
await _record(chat, ctx, Message(text="hello", sender="d4e5f6a7")) # Alice, a companion
public = channel_identity("Public", DEFAULT_PUBLIC_SECRET)
assert chat.unread(f"chan:{public}") == 1
assert chat.unread("dm:d4e5f6a7") == 1
finally:
await chat.stop()
await device.disconnect()
async def test_badge_notifies_when_the_device_state_is_unavailable(repo: Repository) -> None:
"""If there is no device state to place a message against, the message notifies.
A cold or broken cache is a reason to notify too much. It is never a reason to remove a
real message.
"""
device = MockDevice()
ctx = _StubContext(device, repo) # no devstate, on purpose
chat = ChatService(ctx)
await chat.start()
try:
await _record(chat, ctx, Message(text="ping", sender="ffee11223344"))
assert chat.unread("dm:ffee11223344") == 1
finally:
await chat.stop()
await device.disconnect()
async def test_badge_still_respects_a_muted_channel(tmp_path: Path, repo: Repository) -> None:
"""If the user mutes a configured channel, it stays silent. The mute that the user set wins."""
from meshterm.core.channels import channel_identity
from meshterm.core.mute_store import MuteStore
device = MockDevice()
await device.connect()
device._channels[0] = {
"channel_idx": 0,
"channel_name": "Public",
"channel_secret": DEFAULT_PUBLIC_SECRET,
}
public = channel_identity("Public", DEFAULT_PUBLIC_SECRET)
ctx = _GatedContext(device, repo)
ctx.mute_store = MuteStore(tmp_path / "mutes.json")
ctx.mute_store.set_muted(public, True)
chat = ChatService(ctx)
await chat.start()
try:
await _record(chat, ctx, Message(text="Ann: hey", channel=0, is_channel=True))
assert chat.unread(f"chan:{public}") == 0
assert repo.recent_chat_messages(is_channel=True, channel_id=public)
finally:
await chat.stop()
await device.disconnect()
async def test_channel_slot_change_drops_the_cached_channel_list(repo: Repository) -> None:
"""If a slot resolves to a new identity, the cached channel list of the screens is invalid.
The recorder reads a slot for each message, so it sees first a channel that another
client (the phone app) changed to a new key. The screens and the badge rule above read
from the session cache, and only a channel edit in the app clears that cache. If the
cache stays old, it keeps the channels that the device had at connect time. Then a
message on the new channel goes unnoticed, because the cached list has no row for it.
"""
from meshterm.core.channels import derive_secret
device = MockDevice()
await device.connect()
device._channels[0] = {
"channel_idx": 0,
"channel_name": "Public",
"channel_secret": DEFAULT_PUBLIC_SECRET,
}
ctx = _StubContext(device, repo)
class _Devstate:
def __init__(self) -> None:
self.invalidations = 0
def invalidate_channels(self) -> None:
self.invalidations += 1
ctx.devstate = _Devstate()
chat = ChatService(ctx)
first = await chat.channel_id_for(0)
assert ctx.devstate.invalidations == 0 # the first read of a slot is not a change
assert await chat.channel_id_for(0) == first
assert ctx.devstate.invalidations == 0 # a slot that did not change leaves the cache as it is
# Another client gives slot 0 a new key, and we do not know it.
device._channels[0] = {
"channel_idx": 0,
"channel_name": "#montreal",
"channel_secret": derive_secret("#montreal"),
}
assert await chat.channel_id_for(0) != first
assert ctx.devstate.invalidations == 1
# To clear the slot is also a change. It gives the fallback identity from the slot.
device._channels.pop(0)
assert await chat.channel_id_for(0) == "slot:0"
assert ctx.devstate.invalidations == 2
await device.disconnect()
def test_chat_fkey_lane_has_no_day_jump() -> None:
"""No chip sends the day jump: JP removed ``Day ↑`` and ``Day ↓`` (2026-10-10).
The desktop keeps the jump on ^PgUp and ^PgDn.
"""
from datetime import timedelta
from meshterm.core.models import utcnow
now = utcnow()
spread = _screen(
_StubSession(),
send=None,
messages=[
ChatMessage(text="then", peer="d4e5f6a7", created_at=now - timedelta(days=2)),
ChatMessage(text="now", peer="d4e5f6a7", created_at=now),
],
)
lane = spread.picocalc_lyra_lane
actions = {action for pair in lane if pair for action in (pair.action, pair.opp_action)}
assert not actions & {"ctrl_pageup", "ctrl_pagedown"}
deck = fkeys.PICOCALC_LYRA_DECK
assert "Day" not in deck.lane_text(lane).plain + deck.lane_text(lane, shifted=True).plain
spread.handle("ctrl_pageup")
assert spread._selected == 0 # the key still moves to the first message of the older day
def test_channel_and_dm_lanes_share_f1_to_f3_and_keep_their_own_verbs_behind_shift() -> None:
"""The plain bank of F1 to F3 is the same in each conversation (JP, 2026-10-10).
The verbs that the screens share are there, in the order of the hint. The Shift bank
has the verbs of one screen: ``Retry`` in a DM and ``Resend`` in a channel, both ^R.
"""
from tests.conftest import recording_locator
messages = [ChatMessage(text="hi", peer="d4e5f6a7")]
direct = ChatScreen(
Conversation(
label="Alice",
is_channel=False,
contact=Contact(name="Alice", public_key="d4" + "0" * 62, key_prefix="d4e5f6a7"),
),
messages,
send=None,
names={},
session=_StubSession(),
locator=recording_locator([]), # type: ignore[arg-type]
)
async def resend_unscoped(message: ChatMessage) -> ChatMessage:
return message
channel = ChatScreen(
Conversation(label="#general", is_channel=True, channel_id="1"),
[ChatMessage(text="Bob: hi", is_channel=True, channel_idx=0)],
send=None,
names={},
session=_StubSession(),
resend_unscoped=resend_unscoped,
locator=recording_locator([]), # type: ignore[arg-type]
)
for screen in (direct, channel):
lane = screen.picocalc_lyra_lane
assert [pair.label for pair in lane[:3]] == ["Paths", "URL→QR", "Map"]
assert [pair.action for pair in lane[:3]] == ["paths", "url_code", "locate"]
assert direct.picocalc_lyra_lane[0].opp_label == "Retry"
assert channel.picocalc_lyra_lane[0].opp_label == "Resend"
assert channel.picocalc_lyra_lane[0].opp_action == direct.picocalc_lyra_lane[0].opp_action
# -- URL codes (the PicoCalc) ---------------------------------------------------
def _code_rows(lines: list[str]) -> list[str]:
"""The rendered rows that have braille, that is, the rows of a QR code in a transcript."""
return [ln for ln in lines if any("⠀" <= ch <= "⣿" for ch in ln)]
def _decode_braille(rows: list[str], left: int, size: int) -> list[list[bool]]:
"""The ``size``-module code drawn in braille from cell ``left`` of ``rows``."""
dots = ((0x01, 0x02, 0x04, 0x40), (0x08, 0x10, 0x20, 0x80))
return [
[
bool((ord(rows[y // 4][left + x // 2]) - 0x2800) & dots[x % 2][y % 4])
for x in range(size)
]
for y in range(size)
]
def test_on_the_picocalc_each_url_hangs_a_code_under_its_message_side_by_side() -> None:
"""Two links give two braille codes on the same rows, at the indent of the body.
Each module must stay correct in the packing. The test decodes each code from the
braille and compares it with the segno matrix for its URL, at the standard fit.
"""
import segno
from meshterm.platforms import PICOCALC_LYRA, set_platform
set_platform(PICOCALC_LYRA)
first, second = "https://meshterm.net", "https://github.com/jpmartineau/MeshTerm"
conv = Conversation(label="#ops", is_channel=True, channel_idx=1)
message = ChatMessage(text=f"Alice: map at {first}, code at {second}!", is_channel=True)
screen = ChatScreen(conv, [message], send=None, names={}, session=_StubSession())
rows = _code_rows(_strip_ansi(screen._render_grouped(53)).splitlines())
indent = 11 # " HH:MM ", the hanging indent of the body
assert rows and all(len(row) <= 53 and row.startswith(" " * indent) for row in rows)
cells = [row[indent:].ljust(53 - indent) for row in rows]
left, tallest = 0, 0
for url in (first, second):
matrix = [
[bool(v) for v in row] for row in segno.make(url, error="m").matrix_iter(border=4)
]
assert _decode_braille(cells, left, len(matrix)) == matrix, url
left += -(-len(matrix) // 2) + 1 # its cells, then the gap of one cell
tallest = max(tallest, -(-len(matrix) // 4))
assert len(rows) == tallest # one band, as tall as the taller code
def test_a_regular_terminal_hangs_no_code_under_a_url() -> None:
"""A code in half blocks has four times the area. Under each link, codes fill the chat."""
conv = Conversation(label="#ops", is_channel=True, channel_idx=1)
message = ChatMessage(text="Alice: map at https://meshterm.net", is_channel=True)
screen = ChatScreen(conv, [message], send=None, names={}, session=_StubSession())
rendered = _strip_ansi(screen._render_grouped(72))
assert "https://meshterm.net" in rendered
assert not _code_rows(rendered.splitlines())
assert not any(ch in "▀▄█" for ch in rendered)
def test_codes_that_outrun_the_line_start_a_new_band() -> None:
"""Two short links fit next to each other in the 44 cells after the indent. A third wraps."""
import segno
from meshterm.platforms import PICOCALC_LYRA, set_platform
from meshterm.ui.qr import qr_strip
set_platform(PICOCALC_LYRA)
urls = [f"https://meshterm.net/{c}" for c in "abc"]
size = segno.make(urls[0], error="m").symbol_size(border=4)[0]
cols, height = -(-size // 2), -(-size // 4)
rows = qr_strip(urls, 53, indent=9).plain.splitlines()
assert len(rows) == 2 * height and all(len(row) <= 53 for row in rows)
assert all(row[9 + cols] == " " and row[9 + cols + 1] != " " for row in rows[:height])
assert all(row[9 + cols :].strip() == "" for row in rows[height:])
def test_a_picked_message_keeps_its_codes_in_view() -> None:
"""If the selection moves down to a link, the screen shows its code, not only the first row."""
from meshterm.platforms import PICOCALC_LYRA, set_platform
from meshterm.ui.tui import frame
set_platform(PICOCALC_LYRA)
conv = Conversation(label="#ops", is_channel=True, channel_idx=1)
messages = [ChatMessage(text=f"Alice: line {i}", is_channel=True) for i in range(12)]
messages.insert(6, ChatMessage(text="Bob: map at https://meshterm.net", is_channel=True))
screen = ChatScreen(conv, messages, send=None, names={}, session=_StubSession())
viewport = 16
def paint() -> list[str]:
visible, _above, _below = frame._visible_slice(screen, screen.render_body(53), viewport)
return _strip_ansi(visible).splitlines()
paint()
screen.handle("ctrl_home") # the oldest message, then down to the link
for _ in range(6):
paint()
screen.handle("down")
shown = paint()
whole = _code_rows(_strip_ansi(screen._render_grouped(53)).splitlines())
assert any("map at https://meshterm.net" in row for row in shown)
assert len(_code_rows(shown)) == len(whole) # each row of the code is visible
# -- URLs kept whole, and their codes on ^U --------------------------------------------
#: A link that is longer than the regular body lane (72 minus the 11-cell gutter), but that
#: the screen can hold.
_LONG_URL = "https://github.com/jpmartineau/MeshTerm/blob/main/meshterm/ui/chat.py"
def _url_chat(text: str, session=None) -> ChatScreen: # noqa: ANN001
conv = Conversation(label="#ops", is_channel=True, channel_idx=1)
message = ChatMessage(text=f"Alice: {text}", is_channel=True)
return ChatScreen(conv, [message], send=None, names={}, session=session or _StubSession())
def test_a_url_too_long_to_hang_steps_out_to_the_gutter_whole() -> None:
"""A link that the body lane cannot hold, but the screen can, is drawn whole on one line.
It starts in the column after the ``❯`` of the selected message. The body
continues at its own indent after the link, so only the link leaves the block.
"""
screen = _url_chat(f"look at {_LONG_URL} and tell me what you think of it all")
rows = _strip_ansi(screen._render_grouped(72)).splitlines()
link = next(row for row in rows if _LONG_URL[:20] in row)
assert link == " " + _LONG_URL # whole, at the gutter
after = rows[rows.index(link) + 1]
assert after.startswith(" " * 11 + "and tell me") # back under the indent of the body
screen.handle("up") # selected: the ❯ is on the first line, not on the line of the link
rows = _strip_ansi(screen._render_grouped(72)).splitlines()
assert " " + _LONG_URL in rows
def test_a_short_tail_finishes_the_links_line_and_a_too_wide_link_still_folds() -> None:
"""Text that fits after a link at the gutter stays on its line. A wider link folds."""
url = "https://meshterm.net/" + "a" * 46 # 67 cells: wider than the lane of 61 cells
rows = _strip_ansi(_url_chat(f"see {url} ok")._render_grouped(72)).splitlines()
assert " " + url + " ok" in rows # 2 + 67 + 3: the width of the screen
rows = _strip_ansi(_url_chat(f"see {url} ok!")._render_grouped(72)).splitlines()
assert " " + url in rows and " " * 11 + "ok!" in rows # one cell too many: below it
wide = "https://meshterm.net/" + "b" * 60 # 81 cells: wider than the screen
rows = _strip_ansi(_url_chat(f"see {wide}")._render_grouped(72)).splitlines()
assert not any(wide in row for row in rows) # cut, as before
assert all(row.startswith(" " * 11) for row in rows if "bbb" in row) # hanging, as before
def test_ctrl_u_opens_the_picked_messages_links_on_one_share_screen() -> None:
"""^U shows all the links of the selected message as QR codes, in the order of the message."""
from meshterm.ui.qr import QrScreen
opened: list = []
class Session(_StubSession):
async def run_screen(self, screen): # noqa: ANN001, ANN202
opened.append(screen)
async def scenario() -> None:
screen = _url_chat(f"map at https://meshterm.net, code at {_LONG_URL}!", Session())
screen.handle("url_code") # nothing selected: no message to take links from
await asyncio.sleep(0)
assert opened == []
screen.handle("up")
screen.handle("url_code")
await asyncio.sleep(0)
assert len(opened) == 1 and isinstance(opened[0], QrScreen)
assert opened[0].urls == ("https://meshterm.net", _LONG_URL)
asyncio.run(scenario())
def test_ctrl_u_is_named_only_on_a_picked_message_with_a_link() -> None:
"""The hint names ^U where it acts and fits in 72 cells. The lane lights QR in the same way."""
plain = _url_chat("no links here")
linked = _url_chat("map at https://meshterm.net")
assert "^U" not in linked.footer_hint # nothing selected yet
linked.handle("up")
plain.handle("up")
assert "^U QR" in linked.footer_hint and "^U" not in plain.footer_hint
assert fkeys.PICOCALC_LYRA_DECK.action_for(linked.picocalc_lyra_lane, 2) == "url_code" # F2
assert linked.picocalc_lyra_lane[1].enabled and not plain.picocalc_lyra_lane[1].enabled
# The hint of each selection state, with a link in it, fits in the 72-cell budget.
direct = _screen(
_StubSession(),
send=None,
messages=[ChatMessage(text="see https://meshterm.net", peer="d4e5f6a7")],
)
direct.handle("up")
scoped = _url_chat("x")
scoped._retry_target = lambda: 0 # a selected message to send again: the longest hint
scoped._messages[0] = ChatMessage(text="see https://meshterm.net", is_channel=True)
scoped.handle("up")
for screen in (linked, direct, scoped):
assert "^U QR" in screen.footer_hint
assert len(screen.footer_hint) <= 72, screen.footer_hint
def test_on_the_cardputer_a_link_takes_the_share_screen_as_on_the_desktop() -> None:
"""The 14 rows have no room for a code under each message, so ^U is the way to a code.
No code hangs under the message. The lane has QR on its Shift bank (Shift+Fn+5). The
share screen draws the code in braille. The code is small enough that a link and its
code share the 53x14 panel, with the URL under the code and nothing a page down.
"""
from meshterm.platforms import CARDPUTER_ZERO, set_platform
from meshterm.ui.qr import QrScreen
set_platform(CARDPUTER_ZERO)
url = "https://github.com/jpmartineau/MeshTerm"
screen = _url_chat(f"the road list is at {url}")
assert not _code_rows(_strip_ansi(screen._render_grouped(53)).splitlines())
screen.handle("up")
assert fkeys.CARDPUTER_ZERO_DECK.action_for(screen.cardputer_zero_lane, 5) == "url_code"
share = QrScreen(url, title="Links")
share.note_viewport(14)
lines = _strip_ansi(share.render_body(53)).splitlines()
assert len(lines) <= 14 and lines[-1].strip() == url
assert _code_rows(lines), "the code is braille, one module a dot"
def test_a_link_written_without_a_scheme_gets_its_code_too() -> None:
"""``meshterm.net/map`` is a link in the same way as its https form, and it opens as that."""
from meshterm.ui.qr import QrScreen
opened: list = []
class Session(_StubSession):
async def run_screen(self, screen): # noqa: ANN001, ANN202
opened.append(screen)
async def scenario() -> None:
screen = _url_chat("map at meshterm.net/map, notes in file.txt", Session())
screen.handle("up")
assert "^U QR" in screen.footer_hint
screen.handle("url_code")
await asyncio.sleep(0)
assert isinstance(opened[0], QrScreen) and opened[0].urls == ("https://meshterm.net/map",)
asyncio.run(scenario())
# -- acks: listen early, wait long enough, and catch the late acks -------------
class _AckingMeshCore:
"""A meshcore stand-in. Its send accepts the message, then acks it as the script says.
``ack_at`` is the number of seconds after the send returns (``None`` means never).
``during_send`` pushes the ack before the send returns, as an ack from a neighbour can.
"""
def __init__(self, *, ack_at: float | None, during_send: bool = False, flood: bool = True):
self.subs: list = []
self.commands = self
self._ack_at = ack_at
self._during = during_send
self._flood = flood
def subscribe(self, event_type, callback, attribute_filters=None): # noqa: ANN001, ANN201
from types import SimpleNamespace
sub = SimpleNamespace(event_type=event_type, callback=callback)
sub.unsubscribe = lambda: self.subs.remove(sub) if sub in self.subs else None
self.subs.append(sub)
return sub
def _push_ack(self) -> None:
from types import SimpleNamespace
from meshcore import EventType
event = SimpleNamespace(type=EventType.ACK, payload={"code": "c0ffee01", "trip_time": 900})
for sub in list(self.subs):
if sub.event_type is EventType.ACK:
sub.callback(event)
async def send_msg(self, pub, text): # noqa: ANN001, ANN201
from types import SimpleNamespace
if self._during:
self._push_ack()
elif self._ack_at is not None:
asyncio.get_running_loop().call_later(self._ack_at, self._push_ack)
return SimpleNamespace(
type="MSG_SENT",
is_error=lambda: False,
payload={
"type": 1 if self._flood else 0,
"expected_ack": bytes.fromhex("c0ffee01"),
"suggested_timeout": 100, # 0.1 s: the estimate of the radio (too optimistic)
},
)
def _acking_device(mc) -> object: # noqa: ANN001
from meshterm.core.connection import MeshCoreDevice
device = MeshCoreDevice(port="mock")
device._mc = mc
return device
async def test_an_ack_quicker_than_the_send_is_still_caught() -> None:
"""The listener is ready before the send, so an instant ack from a neighbour is not missed."""
device = _acking_device(_AckingMeshCore(ack_at=None, during_send=True))
delivery = await device.send_direct_message(_contact(), "hi")
assert delivery.acked and delivery.code == "c0ffee01" and delivery.flood is True
async def test_an_ack_after_the_radios_estimate_is_still_waited_for(monkeypatch) -> None: # noqa: ANN001
"""The bug: the wait was the radio estimate times 1.2 (0.12 s here, and 5 s on hardware).
The ack of a flood returns on the path that the message found. On hardware, it took
twice the estimate. Now the wait is the estimate plus a round trip, with the same size
as the round trip for a login.
"""
import meshterm.core.connection as connection
monkeypatch.setattr(connection, "trace_timeout", lambda hops: 0.5)
device = _acking_device(_AckingMeshCore(ack_at=0.3))
delivery = await device.send_direct_message(_contact(), "hi")
assert delivery.acked, "an ack at 0.3 s, after the old 0.12 s wait, counts"
async def test_no_ack_in_the_wait_still_says_which_code_to_listen_for(monkeypatch) -> None: # noqa: ANN001
"""A message that is not yet confirmed has not failed. The send returns the code."""
import meshterm.core.connection as connection
monkeypatch.setattr(connection, "trace_timeout", lambda hops: 0.05)
device = _acking_device(_AckingMeshCore(ack_at=None, flood=False))
delivery = await device.send_direct_message(_contact(), "hi")
assert not delivery.acked and delivery.code == "c0ffee01" and delivery.flood is False
async def _late_ack_chat(tmp_path: Path, repo: Repository): # noqa: ANN202
"""A started chat service with a simulator. The acks for Alice arrive late."""
device = MockDevice()
device._late_acks.add("Alice")
ctx = _StubContext(device, repo)
chat = ChatService(ctx)
await chat.start()
alice = next(c for c in await device.get_contacts() if c.name == "Alice")
return device, chat, alice
async def test_a_late_ack_turns_the_message_delivered(tmp_path: Path, repo: Repository) -> None:
"""A late ack changes the message in the history and on the chat screen: ✗ becomes ✓."""
device, chat, alice = await _late_ack_chat(tmp_path, repo)
heard: list[ChatMessage] = []
try:
sent = await chat.send_direct(alice, "you there?", on_late_ack=heard.append)
assert sent.acked is False
for _ in range(100):
if sent.acked:
break
await asyncio.sleep(0.01)
assert sent.acked is True
stored = repo.recent_chat_messages(is_channel=False, peer=alice.key_prefix)
assert stored[-1].acked is True
assert heard == [sent]
finally:
await chat.stop()
await device.disconnect()
async def test_a_late_ack_for_an_earlier_attempt_settles_a_retried_message(
tmp_path: Path, repo: Repository
) -> None:
"""^R sends the message again with a new code. The ack of the first try still counts."""
device, chat, alice = await _late_ack_chat(tmp_path, repo)
try:
await chat.stop() # keep the late acks until both tries are sent
sent = await chat.send_direct(alice, "anyone?")
first = next(iter(chat._awaiting))
await chat.resend_direct(alice, sent)
assert len(chat._awaiting) == 2 and sent.acked is False
chat._settle_ack(first)
assert sent.acked is True and not chat._awaiting
stored = repo.recent_chat_messages(is_channel=False, peer=alice.key_prefix)
assert len(stored) == 1 and stored[0].acked is True
finally:
await device.disconnect()
def test_only_the_acks_the_radio_could_still_push_are_awaited(repo: Repository) -> None:
"""The radio keeps its last eight codes. The service keeps a few more than that, and no more."""
from meshterm.services.chat_service import _AWAITING_CAP
chat = ChatService(_StubContext(MockDevice(), repo))
messages = [ChatMessage(text=str(n), outbound=True, acked=False) for n in range(40)]
for n, message in enumerate(messages):
chat._await_ack(message, [f"{n:08x}"], None)
assert len(chat._awaiting) == _AWAITING_CAP
chat._settle_ack("00000000") # removed long ago: nothing to settle
assert messages[0].acked is False
chat._settle_ack(f"{39:08x}")
assert messages[39].acked is True
# --- ^L: the map on the author ------------------------------------------------------------
#: A node with a location, and our node.
_ALICE_AT = MapMarker(label="Alice", lat=45.5, lon=-73.6, key="a1" * 32)
_US_AT = MapMarker(label="Base", lat=45.4, lon=-73.5, is_self=True, key="ee" * 32)
async def test_chat_locate_opens_the_map_on_the_author_of_the_selected_message() -> None:
"""^L opens the map on the author of the selected message, and is named only where it acts."""
from tests.conftest import recording_locator
locator = recording_locator([_ALICE_AT, _US_AT])
keys = {"alice": "a1" * 32, "bob": "b2" * 32}
screen = ChatScreen(
Conversation(label="#public", is_channel=True, channel_idx=0),
[
ChatMessage(text="Alice: hi", is_channel=True, channel_idx=0),
ChatMessage(text="Bob: yo", is_channel=True, channel_idx=0),
ChatMessage(text="hello", outbound=True, is_channel=True, channel_idx=0),
],
send=None,
names={},
session=_StubSession(),
key_of=lambda name: keys.get(name.casefold()),
locator=locator, # type: ignore[arg-type]
)
# With no selected message, ^L does nothing. The hint does not name it, and its chip on
# F3 is dim.
screen.handle("locate")
await asyncio.sleep(0)
assert locator.shown == []
assert "^L" not in screen.footer_hint
assert screen.picocalc_lyra_lane[2].label == "Map"
assert screen.picocalc_lyra_lane[2].enabled is False
# Our own message: its author is our node.
screen.handle("up")
assert "^L map" in screen.footer_hint
screen.handle("locate")
await asyncio.sleep(0)
assert locator.shown == [_US_AT]
# Bob has a key but no location, so there is nothing to show.
screen.handle("up")
assert "^L" not in screen.footer_hint
assert screen.picocalc_lyra_lane[2].enabled is False
screen.handle("locate")
await asyncio.sleep(0)
assert locator.shown == [_US_AT]
# Alice has a location. The chip is lit, and F3 sends the same action as ^L.
screen.handle("up")
assert "^L map" in screen.footer_hint
assert screen.picocalc_lyra_lane[2].enabled is True
assert fkeys.PICOCALC_LYRA_DECK.action_for(screen.picocalc_lyra_lane, 3) == "locate"
screen.handle("locate")
await asyncio.sleep(0)
assert locator.shown == [_US_AT, _ALICE_AT]
def test_chat_without_a_locator_has_no_map_chip() -> None:
"""A chat that cannot locate leaves F3 empty, not dim."""
screen = _screen(_StubSession(), send=None, messages=[ChatMessage(text="hi", peer="d4e5f6a7")])
assert screen.picocalc_lyra_lane[2] is None
async def test_chat_author_page_opens_the_node_page_of_the_author_of_the_selection() -> None:
"""^O opens the node page of the author of the selected message, named only where it acts.
It is the twin of ^L: the same author, found by the same key. Its chip is ``Author``,
on the Shift half of F3, behind ``Map``.
"""
from tests.conftest import recording_pages
alice = Contact(name="Alice", public_key="a1" * 32, key_prefix="a1" * 6)
pages = recording_pages([alice])
keys = {"alice": "a1" * 32, "bob": "b2" * 32}
screen = ChatScreen(
Conversation(label="#public", is_channel=True, channel_idx=0),
[
ChatMessage(text="Alice: hi", is_channel=True, channel_idx=0),
ChatMessage(text="Bob: yo", is_channel=True, channel_idx=0),
ChatMessage(text="hello", outbound=True, is_channel=True, channel_idx=0),
],
send=None,
names={},
session=_StubSession(),
key_of=lambda name: keys.get(name.casefold()),
pages=pages, # type: ignore[arg-type]
)
# With no selected message, ^O does nothing. The hint does not name it, and its chip is
# dim. The screen has no locator, so the plain half of F3 is empty.
screen.handle("author_page")
await asyncio.sleep(0)
assert pages.shown == []
assert "^O" not in screen.footer_hint
lane = screen.picocalc_lyra_lane
assert (lane[2].label, lane[2].opp_label) == ("", "Author")
assert lane[2].opp_enabled is False
# Our own message: its author is our node, whose page has no contact.
screen.handle("up")
assert "^O author" in screen.footer_hint
screen.handle("author_page")
await asyncio.sleep(0)
assert pages.shown == [None]
# Bob has a key but he is not a contact, so there is no page to open.
screen.handle("up")
assert "^O" not in screen.footer_hint
assert screen.picocalc_lyra_lane[2].opp_enabled is False
screen.handle("author_page")
await asyncio.sleep(0)
assert pages.shown == [None]
# Alice is a contact. The chip is lit, and Shift+F3 sends the same action as ^O.
screen.handle("up")
assert "^O author" in screen.footer_hint
assert screen.picocalc_lyra_lane[2].opp_enabled is True
assert fkeys.PICOCALC_LYRA_DECK.action_for(screen.picocalc_lyra_lane, 8) == "author_page"
screen.handle("author_page")
await asyncio.sleep(0)
assert pages.shown == [None, alice]
def test_a_direct_chat_finds_the_page_of_its_peer() -> None:
"""In a DM, the author of an inbound message is the peer, and the peer is a contact."""
from tests.conftest import recording_pages
peer = Contact(name="Alice", public_key="d4" + "0" * 62, key_prefix="d4e5f6a7")
screen = ChatScreen(
Conversation(label="Alice", is_channel=False, contact=peer),
[ChatMessage(text="hi", peer="d4e5f6a7")],
send=None,
names={"d4e5f6a7": "Alice"},
session=_StubSession(),
pages=recording_pages([peer]), # type: ignore[arg-type]
)
screen.handle("up")
assert "^O author" in screen.footer_hint
assert cell_len(screen.footer_hint) <= 72
def test_k_locks_and_unlocks_the_author_of_the_selected_message(tmp_path: Path) -> None:
"""^K toggles the lock of the author of the selected message (JP, 2026-10-19).
It is the third verb about the author, with ^L and ^O, and it finds the same contact.
The hint and the Shift half of F2 say ``lock`` or ``unlock``, so they also show the
state. The status line tells what the key did. Our own message, and an author that is
not a contact, have nothing to lock.
"""
from meshterm.core.contact_store import ContactStore
from meshterm.ui.tui.session import _CTRL_LETTER_CHORDS
from tests.conftest import recording_pages
assert _CTRL_LETTER_CHORDS["k"] == "lock"
store = ContactStore(tmp_path / "contacts.json")
dev = "cc" * 32
alice = Contact(name="Alice", public_key="a1" * 32, key_prefix="a1" * 6)
keys = {"alice": "a1" * 32, "bob": "b2" * 32}
screen = ChatScreen(
Conversation(label="#public", is_channel=True, channel_idx=0),
[
ChatMessage(text="Alice: hi", is_channel=True, channel_idx=0),
ChatMessage(text="Bob: yo", is_channel=True, channel_idx=0),
ChatMessage(text="hello", outbound=True, is_channel=True, channel_idx=0),
],
send=None,
names={},
session=_StubSession(),
key_of=lambda name: keys.get(name.casefold()),
pages=recording_pages([alice], store=store, self_key=dev), # type: ignore[arg-type]
)
# With no selected message, ^K does nothing, and its chip is dim.
lane = screen.picocalc_lyra_lane
assert (lane[1].opp_label, lane[1].opp_action, lane[1].opp_enabled) == ("Lock", "lock", False)
screen.handle("lock")
assert store.locked_keys(dev) == frozenset()
# Our own message, then Bob, who is not a contact: nothing to lock.
for _ in range(2):
screen.handle("up")
assert "^K" not in screen.footer_hint
assert screen.picocalc_lyra_lane[1].opp_enabled is False
# Alice is a contact. Shift+F2 sends the same action as ^K.
screen.handle("up")
hint = screen.footer_hint
assert "^K lock" in hint and "^O author" in hint and "↑↓ pick" in hint
assert cell_len(hint) <= 72
lane = screen.picocalc_lyra_lane
assert lane[1].opp_enabled is True
assert fkeys.PICOCALC_LYRA_DECK.action_for(lane, 7) == "lock"
screen.handle("lock")
assert store.is_locked(dev, "a1" * 32)
assert "Locked Alice" in screen._status
assert "^K unlock" in screen.footer_hint
assert screen.picocalc_lyra_lane[1].opp_label == "Unlock"
# The same key reverses it.
screen.handle("lock")
assert not store.is_locked(dev, "a1" * 32)
assert screen._status == "Unlocked Alice"
assert "^K lock" in screen.footer_hint
@pytest.mark.parametrize("chips", [True, False])
def test_a_locked_author_has_the_padlock_on_its_sender_chip(
tmp_path: Path,
powerline, # noqa: ANN001
chips: bool,
) -> None:
"""A locked contact reads ``🔒 Name`` on its sender chip, as our chip reads ``★ Name``.
JP, 2026-10-19. The padlock is in the label, so arrow mode has it too. The transcript
keeps the lines of each message in a cache, but a change of the locks draws the chips
again: here the lock goes away as the node page would remove it, with no key press in
the chat.
"""
from meshterm.core.contact_store import ContactStore
from tests.conftest import recording_pages
powerline(chips)
store = ContactStore(tmp_path / "contacts.json")
dev = "cc" * 32
alice = Contact(name="Alice", public_key="a1" * 32, key_prefix="a1" * 6)
store.set_locked(dev, alice, True)
pages = recording_pages([alice], store=store, self_key=dev)
keys = {"alice": "a1" * 32, "bob": "b2" * 32}
screen = ChatScreen(
Conversation(label="#public", is_channel=True, channel_idx=0),
[
ChatMessage(text="Alice: hi", is_channel=True, channel_idx=0),
ChatMessage(text="Bob: yo", is_channel=True, channel_idx=0),
],
send=None,
names={},
session=_StubSession(),
key_of=lambda name: keys.get(name.casefold()),
pages=pages, # type: ignore[arg-type]
)
shown = _strip_ansi(screen.render_body(60))
assert "\U0001f512 Alice" in shown
assert "\U0001f512 Bob" not in shown # Bob is not a contact, so he has no lock
pages.set_locked(alice, False) # type: ignore[attr-defined]
shown = _strip_ansi(screen.render_body(60))
assert "\U0001f512" not in shown and "Alice" in shown
@pytest.mark.parametrize("chips", [True, False])
def test_the_chips_of_the_selected_message_turn_white_with_its_text(
powerline, # noqa: ANN001
chips: bool,
) -> None:
"""On the selected message, a mention chip has a white label, as the text around it does.
JP, 2026-10-19. A chip keeps its fill, so it still tells one node from the next. In
arrow mode, the label takes the ``cursor`` white. A node that no key can place turns
white too: its grey fill already says that the node is unknown.
"""
powerline(chips)
keys = {"alice": "a1" * 32, "bob": "b2" * 32}
body = "hi @[Bob] and @[Zed]"
screen = ChatScreen(
Conversation(label="#public", is_channel=True, channel_idx=0),
[ChatMessage(text=f"Alice: {body}", is_channel=True, channel_idx=0)],
send=None,
names={},
session=_StubSession(),
key_of=lambda name: keys.get(name.casefold()),
)
def styles(label: str, *, selected: bool) -> str:
text = screen._render_mentions(body, selected=selected)
at = text.plain.index(label)
return " ".join(str(s.style) for s in text.spans if s.start <= at < s.end)
white = "#ffffff" if chips else "cursor"
assert white in styles("Bob", selected=True)
assert white not in styles("Bob", selected=False)
assert white in styles("Zed", selected=True)
assert white not in styles("Zed", selected=False)
@pytest.mark.parametrize("chips", [True, False])
def test_the_author_chip_of_the_selected_message_turns_white(
powerline, # noqa: ANN001
chips: bool,
) -> None:
"""The author chip of the group of the selected message has a white label (JP, 2026-10-19).
The chip is drawn with the first message of the group. Thus that message draws again
when the selection is lower in the group, and again when the selection leaves the group,
although the transcript keeps each message in a cache.
"""
powerline(chips)
keys = {"alice": "a1" * 32}
screen = ChatScreen(
Conversation(label="#public", is_channel=True, channel_idx=0),
[
ChatMessage(text="Alice: one", is_channel=True, channel_idx=0),
ChatMessage(text="Alice: two", is_channel=True, channel_idx=0),
ChatMessage(text="Zed: three", is_channel=True, channel_idx=0),
],
send=None,
names={},
session=_StubSession(),
key_of=lambda name: keys.get(name.casefold()),
)
white = "#ffffff" if chips else "cursor"
for sender, message in (("Alice", screen._messages[0]), ("Zed", screen._messages[2])):
picked = screen._group_header(sender, message, picked=True)
plain = screen._group_header(sender, message)
assert white in " ".join(str(s.style) for s in picked.spans)
assert white not in " ".join(str(s.style) for s in plain.spans)
calls: list[tuple[str, bool]] = []
header = screen._group_header
def spy(sender, message, *, picked=False): # noqa: ANN001, ANN202
calls.append((sender, picked))
return header(sender, message, picked=picked)
screen._group_header = spy # type: ignore[method-assign]
screen.render_body(60)
assert calls == [("Alice", False), ("Zed", False)]
calls.clear()
screen.handle("up")
screen.handle("up") # the second message of Alice: its group is the first one
screen.render_body(60)
assert calls == [("Alice", True)]
calls.clear()
screen.handle("down") # Zed: the chip of Alice goes back, and the chip of Zed is white
screen.render_body(60)
assert calls == [("Alice", False), ("Zed", True)]
def test_the_lock_atom_never_pushes_another_key_out_of_the_hint(tmp_path: Path) -> None:
"""With a link and a location, the hint has no room for ^K, so the atom stays out.
The other keys stay as they are without it (JP, 2026-10-19: "If there's room, put
it."). The chip on the lane and the key itself still work.
"""
from meshterm.core.contact_store import ContactStore
from tests.conftest import recording_locator, recording_pages
store = ContactStore(tmp_path / "contacts.json")
dev = "cc" * 32
alice = Contact(name="Alice", public_key="a1" * 32, key_prefix="a1" * 6)
def build(pages: object) -> ChatScreen:
screen = ChatScreen(
Conversation(label="#public", is_channel=True, channel_idx=0),
[ChatMessage(text="Alice: see https://meshterm.net", is_channel=True, channel_idx=0)],
send=None,
names={},
session=_StubSession(),
key_of=lambda name: "a1" * 32 if name == "Alice" else None,
locator=recording_locator([_ALICE_AT]), # type: ignore[arg-type]
pages=pages, # type: ignore[arg-type]
)
screen.handle("up")
return screen
without = build(recording_pages([alice])).footer_hint
screen = build(recording_pages([alice], store=store, self_key=dev))
assert "^K" not in screen.footer_hint
assert screen.footer_hint == without
assert screen.picocalc_lyra_lane[1].opp_enabled is True
screen.handle("lock")
assert store.is_locked(dev, "a1" * 32)
def test_a_chat_without_a_lock_store_has_no_lock_chip() -> None:
"""Where MeshTerm cannot store a lock, the Shift half of F2 is empty, not dim."""
from tests.conftest import recording_pages
alice = Contact(name="Alice", public_key="a1" * 32, key_prefix="a1" * 6)
screen = ChatScreen(
Conversation(label="#public", is_channel=True, channel_idx=0),
[ChatMessage(text="Alice: hi", is_channel=True, channel_idx=0)],
send=None,
names={},
session=_StubSession(),
key_of=lambda name: "a1" * 32,
pages=recording_pages([alice]), # type: ignore[arg-type]
)
screen.handle("up")
assert screen.picocalc_lyra_lane[1].opp_label == ""
assert "^K" not in screen.footer_hint
def test_chat_hint_with_all_its_keys_fits_in_72_cells() -> None:
"""The longest hint (a scoped message of ours, with a link, on the map) fits in 72 cells.
The ↑↓ atom goes first, then ^P. The keys that only this message offers stay.
"""
async def resend_unscoped(message: ChatMessage) -> ChatMessage | None:
return None
from tests.conftest import recording_locator
screen = ChatScreen(
Conversation(label="#ops", is_channel=True, channel_idx=1),
[
ChatMessage(
text="map at https://meshterm.net",
outbound=True,
is_channel=True,
channel_idx=1,
scope="yul",
)
],
send=None,
names={},
session=_StubSession(),
scope="yul",
resend_unscoped=resend_unscoped,
locator=recording_locator([_US_AT]), # type: ignore[arg-type]
)
screen.handle("up")
hint = screen.footer_hint
assert cell_len(hint) <= 72
for atom in ("Enter reply", "^U QR", "^L map", "^R resend unscoped", "Esc cancel"):
assert atom in hint
assert "↑↓ pick" not in hint and "^P paths" not in hint
def test_chat_resend_hint_drops_our_own_page_third() -> None:
"""With ^O too, the resend hint drops ↑↓, then ^P, then ^O: the message is ours."""
async def resend_unscoped(message: ChatMessage) -> ChatMessage | None:
return None
from tests.conftest import recording_locator, recording_pages
screen = ChatScreen(
Conversation(label="#ops", is_channel=True, channel_idx=1),
[
ChatMessage(
text="map at https://meshterm.net",
outbound=True,
is_channel=True,
channel_idx=1,
scope="yul",
)
],
send=None,
names={},
session=_StubSession(),
scope="yul",
resend_unscoped=resend_unscoped,
locator=recording_locator([_US_AT]), # type: ignore[arg-type]
pages=recording_pages([]), # type: ignore[arg-type]
)
screen.handle("up")
hint = screen.footer_hint
assert cell_len(hint) <= 72
for atom in ("Enter reply", "^U QR", "^L map", "^R resend unscoped", "Esc cancel"):
assert atom in hint
assert "↑↓ pick" not in hint and "^P paths" not in hint and "^O" not in hint
def test_chat_reply_hint_with_every_key_of_a_selection_fits_in_72_cells() -> None:
"""A message of a located contact, with a link: the reply hint takes its short words.
With ^U, ^L, and ^O, the long words (``(@mention)``, ``^End/Esc``) do not fit. The
short words of the resend hint do, and then each key of the selection stays.
"""
from tests.conftest import recording_locator, recording_pages
alice = Contact(name="Alice", public_key="a1" * 32, key_prefix="a1" * 6)
screen = ChatScreen(
Conversation(label="#public", is_channel=True, channel_idx=0),
[ChatMessage(text="Alice: map at https://meshterm.net", is_channel=True, channel_idx=0)],
send=None,
names={},
session=_StubSession(),
key_of=lambda name: "a1" * 32 if name.casefold() == "alice" else None,
locator=recording_locator([_ALICE_AT]), # type: ignore[arg-type]
pages=recording_pages([alice]), # type: ignore[arg-type]
)
screen.handle("up")
hint = screen.footer_hint
assert cell_len(hint) <= 72
for atom in ("Enter reply", "^P paths", "^U QR", "^L map", "^O author", "Esc cancel"):
assert atom in hint
Reference in new issue View Git Blame Copy Permalink