Jean-Pierre MartineauandClaude Opus 4.8
1c267378ef
format_ago in widgets.py is now the one prose form for relative ages — bare "now"/"never", "5m ago" otherwise — and every row that stitched _format_age plus a hard-coded " ago" suffix (packet viewer heard row, Time Machine record/first-heard lines, Courier finished rows, channel detail summary) goes through it, killing "now ago" everywhere at once. The channel summary drops its one-off "just now" special case for the same app-wide wording. Also: the Device actions description no longer claims an advert action — that lives in its own main-menu entry. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
623 lines
28 KiB
Python
623 lines
28 KiB
Python
"""Tests for the channel-management feature: pure logic, QR rendering, and the tool.
|
|
|
|
The logic (secret derivation, share-URL round-tripping) is hardware-free; the tool actions
|
|
run against the :class:`MockDevice` simulator and a temporary database.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from hashlib import sha256
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from rich.console import Console
|
|
|
|
from meshterm.context import AppContext
|
|
from meshterm.core.admin_store import AdminStore
|
|
from meshterm.core.channels import (
|
|
CHANNEL_SECRET_BYTES,
|
|
DEFAULT_PUBLIC_SECRET,
|
|
channel_hash,
|
|
decrypt_channel_text,
|
|
derive_secret,
|
|
full_channel_hash,
|
|
normalize_secret,
|
|
parse_share_url,
|
|
random_secret,
|
|
share_url,
|
|
)
|
|
from meshterm.core.config import Settings
|
|
from meshterm.core.device_store import DeviceStore
|
|
from meshterm.persistence.repository import Repository
|
|
from meshterm.tools.channels import ChannelsTool
|
|
from meshterm.services.chat_service import ChatService
|
|
from meshterm.ui.channels import (
|
|
ChannelSlot,
|
|
_apply_order,
|
|
_next_free_slot,
|
|
read_channel_slots,
|
|
)
|
|
from meshterm.ui.qr import qr_text
|
|
|
|
_ANSI = re.compile(r"\x1b\[[0-9;]*m")
|
|
|
|
|
|
# -- pure channel logic -------------------------------------------------------
|
|
|
|
|
|
def test_derive_secret_matches_firmware_formula() -> None:
|
|
"""A derived key is sha256(name)[:16] — the firmware's public-channel scheme."""
|
|
assert derive_secret("#public") == sha256(b"#public").digest()[:16]
|
|
assert len(derive_secret("#anything")) == CHANNEL_SECRET_BYTES
|
|
assert derive_secret("#a") == derive_secret("#a") # deterministic
|
|
|
|
|
|
def test_random_secret_is_16_unique_bytes() -> None:
|
|
"""A random private key is 16 bytes and (practically) never repeats."""
|
|
a, b = random_secret(), random_secret()
|
|
assert len(a) == CHANNEL_SECRET_BYTES and len(b) == CHANNEL_SECRET_BYTES
|
|
assert a != b
|
|
|
|
|
|
def test_normalize_secret_accepts_common_shapes() -> None:
|
|
"""A pasted key is accepted with 0x, spaces, or colons; length is enforced."""
|
|
raw = "000102030405060708090a0b0c0d0e0f"
|
|
expected = bytes(range(16))
|
|
assert normalize_secret(raw) == expected
|
|
assert normalize_secret("0x" + raw) == expected
|
|
assert normalize_secret("0001 0203 0405 0607 0809 0a0b 0c0d 0e0f") == expected
|
|
assert normalize_secret("00:01:02:03:04:05:06:07:08:09:0a:0b:0c:0d:0e:0f") == expected
|
|
|
|
|
|
@pytest.mark.parametrize("bad", ["", "abcd", "zz" * 16, "00" * 15, "00" * 17])
|
|
def test_normalize_secret_rejects_bad_keys(bad: str) -> None:
|
|
"""Non-hex or wrong-length keys raise, so the prompt can re-ask."""
|
|
with pytest.raises(ValueError):
|
|
normalize_secret(bad)
|
|
|
|
|
|
def test_channel_hash_is_two_hex_chars() -> None:
|
|
"""The channel hash is the leading byte of sha256(secret), as the companion reports."""
|
|
secret = bytes(range(16))
|
|
assert channel_hash(secret) == sha256(secret).hexdigest()[:2]
|
|
assert len(channel_hash(secret)) == 2
|
|
|
|
|
|
def test_full_channel_hash_is_the_whole_digest_led_by_the_short_hash() -> None:
|
|
"""The full hash is the complete sha256(secret) digest, opening with the short hash."""
|
|
secret = bytes(range(16))
|
|
full = full_channel_hash(secret)
|
|
assert full == sha256(secret).hexdigest()
|
|
assert len(full) == 64
|
|
assert full[:2] == channel_hash(secret) # the highlighted first byte
|
|
|
|
|
|
def test_share_url_round_trips_with_encoding() -> None:
|
|
"""A share URL builds and parses back to the same name and secret, encoding included."""
|
|
name, secret = "Ops Team #1", bytes(range(16))
|
|
url = share_url(name, secret)
|
|
assert url.startswith("meshcore://channel/add?")
|
|
assert "Ops%20Team" in url # the space is percent-encoded
|
|
parsed = parse_share_url(url)
|
|
assert parsed == (name, secret)
|
|
|
|
|
|
def test_parse_share_url_rejects_non_channel_links() -> None:
|
|
"""Anything that isn't a valid channel-add link parses to None."""
|
|
assert parse_share_url("https://example.com") is None
|
|
assert parse_share_url("meshcore://contact/add?name=x&public_key=ab") is None
|
|
assert parse_share_url("meshcore://channel/add?name=x") is None # missing secret
|
|
assert parse_share_url("meshcore://channel/add?name=x&secret=nothex") is None
|
|
assert parse_share_url("meshcore://channel/add?secret=" + "00" * 16) is None # no name
|
|
|
|
|
|
# -- channel-text decryption ---------------------------------------------------
|
|
|
|
|
|
def _grp_txt_frame(
|
|
name: str, secret: bytes, text: str, *, timestamp: int = 1_700_000_000, attempt: int = 0
|
|
) -> tuple[str, str, str]:
|
|
"""Build a firmware-shaped GRP_TXT frame (chan_hash, cipher_mac, crypted), all hex.
|
|
|
|
Mirrors the encoding :func:`~meshterm.core.channels.decrypt_channel_text` reverses:
|
|
a 4-byte little-endian timestamp, an attempt/type byte, the text, zero-padded to a
|
|
whole AES block, encrypted ECB, then MAC'd with HMAC-SHA256 truncated to 2 bytes.
|
|
"""
|
|
from Crypto.Cipher import AES
|
|
from Crypto.Hash import HMAC, SHA256
|
|
|
|
from meshterm.core.channels import effective_secret
|
|
|
|
key = effective_secret(name, secret)
|
|
plain = timestamp.to_bytes(4, "little") + bytes([attempt & 3]) + text.encode("utf-8")
|
|
plain += b"\x00" * (-len(plain) % 16)
|
|
crypted = AES.new(key, AES.MODE_ECB).encrypt(plain)
|
|
mac = HMAC.new(key, digestmod=SHA256)
|
|
mac.update(crypted)
|
|
return channel_hash(key), mac.digest()[:2].hex(), crypted.hex()
|
|
|
|
|
|
def test_decrypt_channel_text_round_trips_a_known_channel() -> None:
|
|
"""A frame encrypted for a channel we hold the key for decodes cleanly."""
|
|
name, secret = "#general", derive_secret("#general")
|
|
chash, mac, crypted = _grp_txt_frame(name, secret, "hello mesh")
|
|
result = decrypt_channel_text(chash, mac, crypted, [(name, secret)])
|
|
assert result is not None
|
|
assert result.channel_name == name
|
|
assert result.text == "hello mesh"
|
|
assert result.attempt == 0
|
|
assert result.sent_at is not None
|
|
|
|
|
|
def test_decrypt_channel_text_reports_a_resend_attempt() -> None:
|
|
"""The sender's resend counter survives into the decrypted result."""
|
|
name, secret = "Ops", random_secret()
|
|
chash, mac, crypted = _grp_txt_frame(name, secret, "again", attempt=2)
|
|
result = decrypt_channel_text(chash, mac, crypted, [(name, secret)])
|
|
assert result is not None and result.attempt == 2
|
|
|
|
|
|
def test_decrypt_channel_text_returns_none_for_an_unknown_channel() -> None:
|
|
"""A frame from a channel we don't hold the key for decrypts to nothing."""
|
|
name, secret = "#general", derive_secret("#general")
|
|
chash, mac, crypted = _grp_txt_frame(name, secret, "hello mesh")
|
|
other = ("Ops", random_secret())
|
|
assert decrypt_channel_text(chash, mac, crypted, [other]) is None
|
|
assert decrypt_channel_text(chash, mac, crypted, []) is None
|
|
|
|
|
|
def test_decrypt_channel_text_rejects_a_fingerprint_collision() -> None:
|
|
"""A candidate sharing the frame's hash byte but not its key fails the MAC check.
|
|
|
|
The one-byte channel-hash fingerprint can collide between unrelated channels;
|
|
the MAC is what actually confirms the key, so a same-fingerprint wrong channel
|
|
must not be mistaken for a match, and decryption must fall through to try the
|
|
next (correct) candidate instead of stopping at the collision.
|
|
"""
|
|
name, secret = "#general", derive_secret("#general")
|
|
chash, mac, crypted = _grp_txt_frame(name, secret, "hello mesh")
|
|
# Brute-force a same-fingerprint, wrong-key decoy (a 1-byte hash, so cheap to find).
|
|
decoy = next(
|
|
s for s in (random_secret() for _ in range(10_000))
|
|
if channel_hash(s) == chash and s != secret
|
|
)
|
|
result = decrypt_channel_text(
|
|
chash, mac, crypted, [("decoy", decoy), (name, secret)]
|
|
)
|
|
assert result is not None and result.text == "hello mesh" # falls through to the real key
|
|
|
|
|
|
def test_decrypt_channel_text_rejects_malformed_hex() -> None:
|
|
"""Garbage hex fields fail closed instead of raising."""
|
|
assert decrypt_channel_text("zz", "zzzz", "zzzzzzzzzzzzzzzz", [("x", random_secret())]) is None
|
|
assert decrypt_channel_text("00", "0000", "00" * 15, [("x", random_secret())]) is None # not block-aligned
|
|
|
|
|
|
# -- QR rendering -------------------------------------------------------------
|
|
|
|
|
|
def test_qr_text_renders_square_with_quiet_zone() -> None:
|
|
"""qr_text produces block rows framed by a light quiet zone, wider than the data."""
|
|
console = Console(force_terminal=True, width=120, file=__import__("io").StringIO())
|
|
console.print(qr_text("meshcore://channel/add?name=Test&secret=" + "ab" * 16))
|
|
plain = _ANSI.sub("", console.file.getvalue()).rstrip("\n")
|
|
lines = plain.splitlines()
|
|
assert lines, "QR produced no output"
|
|
# A version-appropriate QR for this URL is at least ~25 modules wide plus an 8-module
|
|
# quiet zone; the finder pattern makes the code non-trivial.
|
|
assert len(lines[0]) >= 30
|
|
assert any("█" in line for line in lines) # dark modules were drawn
|
|
|
|
|
|
# -- channel slot model -------------------------------------------------------
|
|
|
|
|
|
def test_channel_slot_classifies_public_and_private() -> None:
|
|
"""A #-named or name-derived slot reads as public; a random-key slot as private."""
|
|
public = ChannelSlot(idx=0, name="#public", secret=derive_secret("#public"))
|
|
assert public.is_public
|
|
derived = ChannelSlot(idx=1, name="general", secret=derive_secret("general"))
|
|
assert derived.is_public # key derived from the name, even without a leading #
|
|
private = ChannelSlot(idx=2, name="Ops", secret=bytes(range(16)))
|
|
assert not private.is_public
|
|
assert not private.is_name_derived
|
|
# The firmware default "Public" is public but keyed by a fixed well-known secret, not by
|
|
# its name — so it reads as public without being name-derived (its key must be preserved).
|
|
default_public = ChannelSlot(idx=0, name="Public", secret=DEFAULT_PUBLIC_SECRET)
|
|
assert default_public.is_public
|
|
assert not default_public.is_name_derived
|
|
assert channel_hash(DEFAULT_PUBLIC_SECRET) == "11" # the fingerprint devices report for it
|
|
assert private.conversation.is_channel and private.conversation.channel_idx == 2
|
|
assert private.conversation.label == "Ops" # raw name, no forced leading '#'
|
|
# The conversation is keyed by the channel's identity, not its slot.
|
|
assert private.conversation.channel_id == private.identity
|
|
assert private.conversation.key == f"chan:{private.identity}"
|
|
|
|
|
|
def test_next_free_slot_finds_gaps_and_full() -> None:
|
|
"""The next free slot skips used indices and is None when every slot is taken."""
|
|
slots = [ChannelSlot(idx=0, name="a", secret=b"\x00" * 16),
|
|
ChannelSlot(idx=2, name="c", secret=b"\x00" * 16)]
|
|
assert _next_free_slot(slots, capacity=8) == 1
|
|
full = [ChannelSlot(idx=i, name=str(i), secret=b"\x00" * 16) for i in range(8)]
|
|
assert _next_free_slot(full, capacity=8) is None
|
|
# With a larger discovered capacity the same "full-at-8" set still has room.
|
|
assert _next_free_slot(full, capacity=16) == 8
|
|
|
|
|
|
async def test_channel_capacity_is_probed_not_assumed(ctx: AppContext) -> None:
|
|
"""Capacity comes from probing the device, and counts slots, not their occupants."""
|
|
device = await ctx.device()
|
|
assert await device.channel_capacity() == 8 # the mock models stock 8-slot firmware
|
|
# Occupying some slots must not change the ceiling — capacity is how many slots exist.
|
|
await device.set_channel(0, "Alpha", bytes(range(16)))
|
|
await device.set_channel(3, "Bravo", bytes(range(16)))
|
|
assert await device.channel_capacity() == 8
|
|
|
|
|
|
async def test_channel_capacity_tracks_a_larger_ceiling(ctx: AppContext) -> None:
|
|
"""Firmware with more slots is discovered as such — no hard-coded 8 anywhere."""
|
|
device = await ctx.device()
|
|
device._max_channels = 12 # simulate a firmware build with a larger slot table
|
|
assert await device.channel_capacity() == 12
|
|
|
|
|
|
async def test_apply_order_relays_channels_into_new_positions(ctx: AppContext) -> None:
|
|
"""Applying a new order re-lays channels across the same slots in display order."""
|
|
device = await ctx.device()
|
|
await device.set_channel(0, "Alpha", bytes(range(16))) # private
|
|
await device.set_channel(1, "#beta", None) # public, key derived from the name
|
|
await device.set_channel(2, "Gamma", bytes(range(16, 32))) # private
|
|
|
|
slots = await read_channel_slots(device)
|
|
# Reverse the display order: the row that was third moves first, first moves last.
|
|
writes = await _apply_order(device, slots, [2, 1, 0])
|
|
assert writes == 2 # the middle channel keeps its slot; the two ends swap
|
|
|
|
after = {s.idx: s for s in await read_channel_slots(device)}
|
|
assert after[0].name == "Gamma" and after[0].secret == bytes(range(16, 32))
|
|
assert after[1].name == "#beta" and after[1].is_public # public key re-derived in place
|
|
assert after[2].name == "Alpha" and after[2].secret == bytes(range(16))
|
|
|
|
|
|
def test_channel_identity_is_stable_across_slot_moves() -> None:
|
|
"""A channel's identity depends on its key material, not on which slot it occupies."""
|
|
at_two = ChannelSlot(idx=2, name="Ops", secret=bytes(range(16)))
|
|
at_five = ChannelSlot(idx=5, name="Ops", secret=bytes(range(16)))
|
|
assert at_two.identity == at_five.identity # same channel, different slot → same identity
|
|
|
|
other = ChannelSlot(idx=2, name="Ops", secret=bytes(range(16, 32)))
|
|
assert other.identity != at_two.identity # a different key → a different channel
|
|
|
|
|
|
class _ScriptedUi:
|
|
"""A UI surface that replays queued answers, for driving the channel manager headless.
|
|
|
|
``select``/``text``/``confirm`` each pop their next scripted answer; the display methods
|
|
are no-ops. Enough of the :class:`~meshterm.ui.surface.Ui` contract for the channel
|
|
manager's create/clear flows.
|
|
"""
|
|
|
|
def __init__(self, selects: list, texts: list, confirms: list) -> None:
|
|
self._selects = list(selects)
|
|
self._texts = list(texts)
|
|
self._confirms = list(confirms)
|
|
|
|
def show(self, *renderables) -> None: # noqa: ANN002
|
|
pass
|
|
|
|
def note(self, markup: str) -> None:
|
|
pass
|
|
|
|
async def view(self, renderable, *, title: str = "", footer_hint: str = "") -> None: # noqa: ANN001
|
|
pass
|
|
|
|
async def select(self, title: str, items: list, *, default=None): # noqa: ANN001, ANN201
|
|
return self._selects.pop(0) if self._selects else "__back__"
|
|
|
|
async def text(self, title: str, *, default: str = "", validate=None, help_text: str = "", password: bool = False): # noqa: ANN001, ANN201
|
|
return self._texts.pop(0)
|
|
|
|
async def confirm(self, title: str, *, default: bool = True): # noqa: ANN201
|
|
return self._confirms.pop(0)
|
|
|
|
|
|
async def test_recreating_a_slot_refiles_messages_to_the_new_channel(ctx: AppContext) -> None:
|
|
"""Clearing a channel and creating another that reuses its slot must not cross history.
|
|
|
|
Reproduces the reported bug: inbound channel messages carry only a slot index, which the
|
|
chat service maps to a channel identity through a cache. When a slot is cleared and a new
|
|
channel takes it, a stale cache filed the newcomer's messages under the old channel — so
|
|
opening the old channel showed the new one's transcript. The manager must refresh the
|
|
cache after the mutation so a message on the reused slot lands in the right channel.
|
|
"""
|
|
from meshterm.core.channels import channel_identity
|
|
from meshterm.core.events import MeshEvent
|
|
from meshterm.core.models import Message
|
|
from meshterm.ui.channels import _BACK, _CLEAR, _CREATE, manage_channels
|
|
|
|
device = await ctx.device()
|
|
await device.set_channel(0, "Public", DEFAULT_PUBLIC_SECRET)
|
|
public_id = channel_identity("Public", DEFAULT_PUBLIC_SECRET)
|
|
|
|
await ctx.chat.start() # prime the slot→identity cache (slot 0 == Public)
|
|
try:
|
|
# A message on Public arrives before we touch anything — files under Public. The
|
|
# inbound worker resolves it against the slot as it stands now (Public), which is
|
|
# exactly what happens live: messages are recorded on arrival, ahead of any later edit.
|
|
ctx.events.publish(
|
|
MeshEvent.message_event(Message(text="hi public", channel=0, is_channel=True))
|
|
)
|
|
await ctx.chat._queue.join()
|
|
|
|
# Drive the manager: open Public's detail → clear it, then create a new private
|
|
# channel (which reuses freed slot 0), then back out.
|
|
ctx.ui = _ScriptedUi(
|
|
selects=[0, _CLEAR, _CREATE, _BACK],
|
|
texts=["Ops"], # the new channel's name
|
|
confirms=[True], # confirm the clear
|
|
)
|
|
await manage_channels(ctx)
|
|
|
|
ops = next(s for s in await read_channel_slots(device) if s.name == "Ops")
|
|
assert ops.idx == 0 # the new channel took the freed slot
|
|
|
|
# A message now arrives on slot 0 — which is Ops, not Public.
|
|
ctx.events.publish(
|
|
MeshEvent.message_event(Message(text="ops secret", channel=0, is_channel=True))
|
|
)
|
|
await ctx.chat._queue.join()
|
|
|
|
public_msgs = ctx.repo.recent_chat_messages(is_channel=True, channel_id=public_id)
|
|
ops_msgs = ctx.repo.recent_chat_messages(is_channel=True, channel_id=ops.identity)
|
|
assert [m.text for m in public_msgs] == ["hi public"] # unchanged, no leakage
|
|
assert [m.text for m in ops_msgs] == ["ops secret"] # filed under the right channel
|
|
finally:
|
|
await ctx.chat.stop()
|
|
|
|
|
|
async def test_reorder_keeps_history_because_key_is_intrinsic(ctx: AppContext) -> None:
|
|
"""Reordering channels leaves each channel's transcript with it — no migration needed.
|
|
|
|
Reproduces the reported bug's setup (a channel moved to a new slot) and asserts the fix:
|
|
because history is keyed by channel identity rather than slot, the moved channel keeps
|
|
its own messages and the slot it landed on inherits none.
|
|
"""
|
|
device = await ctx.device()
|
|
await device.set_channel(0, "Alpha", bytes(range(16)))
|
|
await device.set_channel(1, "Beta", bytes(range(16, 32)))
|
|
|
|
chat = ChatService(ctx)
|
|
await chat.start()
|
|
try:
|
|
await chat.send_channel(0, "hello from alpha", label="Alpha")
|
|
await chat.send_channel(1, "hello from beta", label="Beta")
|
|
|
|
slots = await read_channel_slots(device)
|
|
before = {s.name: s for s in slots}
|
|
alpha_id, beta_id = before["Alpha"].identity, before["Beta"].identity
|
|
|
|
# Swap the two channels' slots, then refresh the service's slot→identity cache.
|
|
writes = await _apply_order(device, slots, [1, 0])
|
|
assert writes == 2
|
|
await chat.refresh_channels()
|
|
|
|
after = {s.name: s for s in await read_channel_slots(device)}
|
|
assert after["Alpha"].idx == 1 and after["Beta"].idx == 0 # slots actually swapped
|
|
assert after["Alpha"].identity == alpha_id # identity is unchanged by the move
|
|
|
|
# History still resolves per channel by identity — no transcript was inherited.
|
|
alpha_msgs = ctx.repo.recent_chat_messages(is_channel=True, channel_id=alpha_id)
|
|
beta_msgs = ctx.repo.recent_chat_messages(is_channel=True, channel_id=beta_id)
|
|
assert [m.text for m in alpha_msgs] == ["hello from alpha"]
|
|
assert [m.text for m in beta_msgs] == ["hello from beta"]
|
|
finally:
|
|
await chat.stop()
|
|
|
|
|
|
# -- message statistics in the manager ----------------------------------------
|
|
|
|
|
|
def test_channel_stats_aggregates_totals_window_and_recency(ctx: AppContext) -> None:
|
|
"""Per-channel stats count all messages, the trailing-2h slice, and the last time."""
|
|
from datetime import timedelta
|
|
|
|
from meshterm.core.models import ChatMessage, utcnow
|
|
|
|
stale = utcnow() - timedelta(days=30) # outside the two-hour activity window
|
|
fresh = utcnow() - timedelta(minutes=2)
|
|
for text, when in (("old a", stale), ("old b", stale), ("new", fresh)):
|
|
ctx.repo.record_chat_message(
|
|
ChatMessage(text=text, is_channel=True, channel_id="ops", created_at=when)
|
|
)
|
|
ctx.repo.record_chat_message( # a direct message must not leak into channel stats
|
|
ChatMessage(text="dm", is_channel=False, peer="abc123", created_at=fresh)
|
|
)
|
|
ctx.repo.record_chat_message( # a legacy row without an identity has nothing to count under
|
|
ChatMessage(text="legacy", is_channel=True, channel_id=None, created_at=fresh)
|
|
)
|
|
|
|
stats = ctx.repo.channel_stats()
|
|
assert set(stats) == {"ops"}
|
|
ops = stats["ops"]
|
|
assert ops.total == 3
|
|
assert ops.recent == 1 # only the fresh message falls inside the window
|
|
assert ops.last_at is not None
|
|
assert abs((ops.last_at - fresh).total_seconds()) < 1
|
|
# The histogram spans the window newest-first: the 2-minute-old message lands in the
|
|
# "now" (first) bucket and the 30-day-old ones land nowhere.
|
|
assert len(ops.histogram) == 24
|
|
assert ops.histogram[0] == 1 and sum(ops.histogram) == 1
|
|
|
|
|
|
def test_activity_sparkline_packs_24_buckets_into_braille() -> None:
|
|
"""The sparkline draws two buckets per cell at absolute heights, newest at the right."""
|
|
from meshterm.ui.channels import _activity_sparkline
|
|
|
|
assert _activity_sparkline((0,) * 24).plain == "⣀" * 12 # silent window: a flatline
|
|
assert _activity_sparkline(()).plain == "⣀" * 12 # a channel with no stats at all
|
|
assert _activity_sparkline((21,) * 24).plain == "⣿" * 12 # flat-out: every dot lit
|
|
# A lone message in the newest bucket: one tier-one dot in the *final* cell's right
|
|
# column (now sits at the right edge), riding the continuing baseline — so the glyph
|
|
# matches the flatline and the ok-green style is what marks it as traffic.
|
|
lone = _activity_sparkline((1,) + (0,) * 23)
|
|
assert lone.plain == "⣀" * 12
|
|
styles = [span.style for span in lone.spans]
|
|
assert styles[-1] == "ok" and set(styles[:-1]) == {"faint"}
|
|
# Heights step at the absolute thresholds 1 / 3 / 8 / 21 messages per bucket, the
|
|
# newest (tallest here) pair in the rightmost cell.
|
|
steps = _activity_sparkline((1, 2, 3, 7, 8, 20, 21, 999) + (0,) * 16).plain
|
|
assert steps[-4:] == "".join(
|
|
chr(0x2800 | left | right)
|
|
for left, right in ((0x47, 0xB8), (0x46, 0xB0), (0x44, 0xA0), (0x40, 0x80))
|
|
)
|
|
|
|
|
|
async def test_channel_rows_carry_stats_unread_and_lanes(ctx: AppContext) -> None:
|
|
"""A list row shows the channel's type, unread badge, counts, age, and activity meter."""
|
|
from meshterm.core.models import ChatMessage
|
|
from meshterm.ui.channels import _LiveStats, _menu_items
|
|
from meshterm.ui.tui import Choice
|
|
|
|
device = await ctx.device()
|
|
await device.set_channel(0, "Ops", bytes(range(16)))
|
|
slots = await read_channel_slots(device)
|
|
slot = slots[0]
|
|
for text in ("one", "two"):
|
|
ctx.repo.record_chat_message(
|
|
ChatMessage(text=text, is_channel=True, channel_id=slot.identity)
|
|
)
|
|
ctx.chat._unread[slot.conversation.key] = 2 # as the service would after two arrivals
|
|
|
|
title, items = _menu_items(ctx, slots, 8, _LiveStats(ctx))
|
|
assert title == "Channels — 1/8 slots"
|
|
row = next(it for it in items if isinstance(it, Choice) and it.value == 0)
|
|
plain = row.label.plain # the title is a live callable; .label resolves it
|
|
assert "Ops" in plain
|
|
# No TYPE or HASH lane: the glyph carries openness, Show key carries the hash —
|
|
# the freed cells keep the activity lane visible at 72 columns.
|
|
assert "private" not in plain and slot.hash not in plain
|
|
assert "● 2" in plain # the unread badge
|
|
assert "now" in plain # the just-recorded message's age
|
|
# Both just-recorded messages sit in the sparkline's newest bucket — the *final*
|
|
# cell's right column (one dot: two messages is still tier one), riding the baseline
|
|
# at the row's right edge — with the rest of the window on the faint flatline.
|
|
assert plain.rstrip().endswith("⣀" * 12) # the sparkline is the final lane
|
|
label = row.label
|
|
assert label.spans[-1].style == "ok" # …and its newest cell reads as live traffic
|
|
|
|
|
|
async def test_show_key_popup_wraps_values_under_header_labels() -> None:
|
|
"""The key popup is labelled blocks, not a table, and long values wrap whole."""
|
|
from types import SimpleNamespace
|
|
|
|
from meshterm.ui.channels import _show_key
|
|
from meshterm.ui.tui.render import render_lines
|
|
|
|
captured: dict = {}
|
|
|
|
async def view(renderable, *, title="", footer_hint=""): # noqa: ANN001, ANN202
|
|
captured["renderable"] = renderable
|
|
captured["title"] = title
|
|
|
|
slot = ChannelSlot(idx=0, name="Ops", secret=bytes(range(16)))
|
|
await _show_key(SimpleNamespace(ui=SimpleNamespace(view=view)), slot)
|
|
assert captured["title"] == "Key — Ops"
|
|
|
|
# Render at a deliberately narrow dialog width: every value must survive whole,
|
|
# wrapped across lines, with no ellipsis truncation anywhere.
|
|
lines = [_ANSI.sub("", line) for line in render_lines(captured["renderable"], 40)]
|
|
text = "\n".join(lines)
|
|
for label in ("NAME", "TYPE", "HASH", "KEY", "LINK"):
|
|
assert label in text
|
|
assert "…" not in text
|
|
joined = text.replace("\n", "").replace(" ", "")
|
|
assert slot.secret.hex() in joined # the 32-hex key, reassembled across wraps
|
|
assert full_channel_hash(slot.secret) in joined # the 64-hex hash likewise
|
|
assert share_url("Ops", slot.secret).replace(" ", "") in joined
|
|
|
|
|
|
async def test_detail_summary_reads_slot_totals_and_unread(ctx: AppContext) -> None:
|
|
"""The detail screen's vital-signs line covers slot, totals, unread, and recency."""
|
|
from meshterm.core.models import ChatMessage
|
|
from meshterm.ui.channels import _detail_summary, _LiveStats
|
|
|
|
device = await ctx.device()
|
|
await device.set_channel(3, "Ops", bytes(range(16)))
|
|
slot = next(s for s in await read_channel_slots(device) if s.idx == 3)
|
|
|
|
assert _detail_summary(ctx, slot, _LiveStats(ctx)) == "Slot 3 · no messages recorded yet"
|
|
|
|
ctx.repo.record_chat_message(
|
|
ChatMessage(text="hi", is_channel=True, channel_id=slot.identity)
|
|
)
|
|
ctx.chat._unread[slot.conversation.key] = 1
|
|
summary = _detail_summary(ctx, slot, _LiveStats(ctx))
|
|
# A fresh age reads as bare "now" (the app-wide format_ago grammar — never "now ago").
|
|
assert summary == "Slot 3 · 1 message · 1 unread · last message now"
|
|
|
|
|
|
# -- the tool against the simulator -------------------------------------------
|
|
|
|
|
|
@pytest.fixture()
|
|
def ctx(tmp_path: Path) -> AppContext:
|
|
"""A mock-backed application context with the plain (console) UI surface."""
|
|
settings = Settings(config_dir=tmp_path, db_path=tmp_path / "chan.db")
|
|
context = AppContext(
|
|
console=Console(file=__import__("io").StringIO()),
|
|
settings=settings,
|
|
repo=Repository(settings.db_path),
|
|
device_store=DeviceStore(tmp_path / "devices.json"),
|
|
admin_store=AdminStore(tmp_path / "admin.json"),
|
|
mock=True,
|
|
)
|
|
yield context
|
|
context.repo.close()
|
|
|
|
|
|
async def test_cli_add_private_generates_key_and_lists(ctx: AppContext) -> None:
|
|
"""`add` with no secret creates a private channel with a random key; `list` sees it."""
|
|
tool = ChannelsTool()
|
|
result = await tool.run(ctx, {"cli_action": "add", "index": 1, "name": "Ops", "secret": None})
|
|
assert result.summary == {"index": 1, "name": "Ops"}
|
|
|
|
slots = await read_channel_slots(await ctx.device())
|
|
slot = next(s for s in slots if s.idx == 1)
|
|
assert slot.name == "Ops"
|
|
assert not slot.is_public # a random key, not derived from the name
|
|
|
|
|
|
async def test_cli_add_public_derives_key(ctx: AppContext) -> None:
|
|
"""`add` with a #-name creates a public channel keyed from the name."""
|
|
tool = ChannelsTool()
|
|
await tool.run(ctx, {"cli_action": "add", "index": 2, "name": "#general", "secret": None})
|
|
slot = next(s for s in await read_channel_slots(await ctx.device()) if s.idx == 2)
|
|
assert slot.name == "#general" and slot.is_public
|
|
|
|
|
|
async def test_cli_join_and_import_round_trip(ctx: AppContext) -> None:
|
|
"""A channel joined by key can be shared and re-imported to the same secret."""
|
|
tool = ChannelsTool()
|
|
secret = bytes(range(16))
|
|
await tool.run(
|
|
ctx, {"cli_action": "join", "index": 3, "name": "Squad", "secret": secret.hex()}
|
|
)
|
|
slot = next(s for s in await read_channel_slots(await ctx.device()) if s.idx == 3)
|
|
assert slot.name == "Squad" and slot.secret == secret
|
|
|
|
# Importing the channel's own share link onto another slot reproduces it.
|
|
await tool.run(ctx, {"cli_action": "import", "index": 4, "url": share_url("Squad", secret)})
|
|
imported = next(s for s in await read_channel_slots(await ctx.device()) if s.idx == 4)
|
|
assert imported.name == "Squad" and imported.secret == secret
|
|
|
|
|
|
async def test_cli_import_rejects_bad_link(ctx: AppContext) -> None:
|
|
"""Importing a non-channel link is a clean parameter error, not a crash."""
|
|
import typer
|
|
|
|
tool = ChannelsTool()
|
|
with pytest.raises(typer.BadParameter):
|
|
await tool.run(ctx, {"cli_action": "import", "index": 5, "url": "https://nope"})
|