Add a first-class channel manager with QR sharing and reordering

Replace the primitive channel editor that lived inside the config menu with
a dedicated `channels` tool: list the device's slots and, per channel, show a
scannable QR share code, reveal the key, open it in chat, rename/re-key, or
clear it. Create channels four ways — new private (random key), public `#`
(key derived from the name), join-by-key, or import a `meshcore://` link — with
every change written to the device immediately.

- core/channels.py: pure, hardware-free logic (secret derivation, tolerant key
  parsing, share-URL round-tripping, channel hash). full_channel_hash exposes
  the complete digest; the short channel_hash is its leading byte.
- ui/qr.py: half-block QR rendering (segno), drawn black-on-white so it scans
  regardless of terminal theme.
- ui/channels.py: the interactive manager. Channels display by their real name;
  reorder rows by dragging with the arrow keys; the key view shows the full hash
  with the first byte highlighted.
- ui/tui: a ReorderScreen (grab/move/drop) plus a ctx.ui.reorder surface method.
- tools/channels.py: the tool + `channels list/add/join/import/share` CLI.
- Pin Chat + Channels to the top of the main menu via an explicit category order.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Jean-Pierre MartineauandClaude Opus 4.8 committed 2026-07-06 18:23:47 -04:00
1 parent 6856b682b1
commit 7330796829
12 files changed
+1257 -49

No files matched your search

+157
View File
@@ -0,0 +1,157 @@
"""Pure channel logic: secret derivation and MeshCore share-URL round-tripping.
A MeshCore channel is a name plus a 16-byte shared secret. *Public* channels (whose name
starts with ``#``) derive their secret deterministically from the name, so anyone naming the
channel the same way lands on the same key; *private* channels carry a random secret shared
out of band. This module holds that logic with no device or I/O dependency, so the create /
join flows — and the ``meshcore://`` share links behind the QR codes — can be built and
validated (and unit-tested) without hardware.
"""
from __future__ import annotations
import secrets
from hashlib import sha256
from typing import Optional
from urllib.parse import parse_qs, quote, urlsplit
#: A channel shared secret is exactly 16 bytes (128-bit), per the MeshCore protocol.
CHANNEL_SECRET_BYTES = 16
#: The default public channel every MeshCore device ships with on slot 0.
DEFAULT_PUBLIC_NAME = "public"
def is_public_name(name: str) -> bool:
"""Whether a channel name is a public one whose key derives from the name.
Args:
name: The channel name.
Returns:
``True`` for a ``#``-prefixed name (the firmware's convention for a name-derived
key), ``False`` otherwise.
"""
return name.startswith("#")
def derive_secret(name: str) -> bytes:
"""Derive a channel's 16-byte secret from its name, exactly as the firmware does.
A ``#``-prefixed (public) channel — or any channel created without an explicit key —
uses ``sha256(name)[:16]``, so everyone who types the same name shares the same key.
Args:
name: The channel name (including any leading ``#``).
Returns:
The derived 16-byte secret.
"""
return sha256(name.encode("utf-8")).digest()[:CHANNEL_SECRET_BYTES]
def random_secret() -> bytes:
"""Generate a fresh cryptographically-random 16-byte secret for a private channel.
Returns:
16 random bytes.
"""
return secrets.token_bytes(CHANNEL_SECRET_BYTES)
def normalize_secret(text: str) -> bytes:
"""Parse a user-supplied secret (32 hex chars / 16 bytes).
Tolerates surrounding whitespace, an ``0x`` prefix, and internal spaces so a key
pasted in any common shape is accepted.
Args:
text: The secret as a hex string.
Returns:
The 16-byte secret.
Raises:
ValueError: If ``text`` is not exactly 16 bytes of hexadecimal.
"""
cleaned = text.strip().removeprefix("0x").replace(" ", "").replace(":", "")
data = bytes.fromhex(cleaned) # raises ValueError on odd length / non-hex
if len(data) != CHANNEL_SECRET_BYTES:
raise ValueError("Channel secret must be exactly 16 bytes (32 hex characters).")
return data
def full_channel_hash(secret: bytes) -> str:
"""Return the full ``sha256(secret)`` hex digest (64 chars) for a channel.
Its leading byte is the short :func:`channel_hash` MeshCore reports; the rest is shown
only for a complete fingerprint.
Args:
secret: The channel's 16-byte secret.
Returns:
A 64-character lowercase hex string.
"""
return sha256(bytes(secret)).hexdigest()
def channel_hash(secret: bytes) -> str:
"""Return the 2-hex-character channel hash MeshCore shows for a secret.
This is the leading byte of ``sha256(secret)``, the same short fingerprint the
companion reports for a channel — handy for telling two channels apart at a glance.
Args:
secret: The channel's 16-byte secret.
Returns:
A two-character lowercase hex string.
"""
return full_channel_hash(secret)[:2]
def share_url(name: str, secret: bytes) -> str:
"""Build the MeshCore ``meshcore://channel/add`` share URL for a channel.
This is the exact scheme the MeshCore mobile apps scan, so a QR code rendered from the
returned string adds the channel directly on another device.
Args:
name: The channel name.
secret: The channel's 16-byte secret.
Returns:
A ``meshcore://channel/add?name=…&secret=…`` URL.
"""
return f"meshcore://channel/add?name={quote(name, safe='')}&secret={bytes(secret).hex()}"
def parse_share_url(url: str) -> Optional[tuple[str, bytes]]:
"""Parse a ``meshcore://channel/add`` link into ``(name, secret)``.
Args:
url: A channel-add share URL (as scanned from a QR code or pasted in).
Returns:
The channel name and its 16-byte secret, or ``None`` if the string is not a valid
MeshCore channel-add link.
"""
try:
parts = urlsplit(url.strip())
except ValueError:
return None
if parts.scheme != "meshcore":
return None
# The path splits as netloc="channel", path="/add"; recombine and compare on the route.
if f"{parts.netloc}{parts.path}".strip("/") != "channel/add":
return None
query = parse_qs(parts.query)
names = query.get("name")
keys = query.get("secret")
if not names or not keys:
return None
try:
return names[0], normalize_secret(keys[0])
except ValueError:
return None
+22 -2
View File
@@ -131,6 +131,19 @@ class Tool(ABC):
_REGISTRY: dict[str, Tool] = {}
#: Explicit menu ordering for tool categories. Messaging (Chat + Channels) leads because it
#: holds the first-class, everyday features; the rest follow in a natural workflow order.
#: Categories not listed here sort last, alphabetically, so a new category still appears.
_CATEGORY_ORDER = ["Messaging", "Device", "Diagnostics", "Optimization", "Data"]
def _category_rank(category: str) -> int:
"""Return a category's sort rank (its position in ``_CATEGORY_ORDER``, else last)."""
try:
return _CATEGORY_ORDER.index(category)
except ValueError:
return len(_CATEGORY_ORDER)
def register(cls: type[Tool]) -> type[Tool]:
"""Class decorator that instantiates a tool and adds it to the registry.
@@ -154,12 +167,19 @@ def register(cls: type[Tool]) -> type[Tool]:
def all_tools() -> list[Tool]:
"""Return all registered tools sorted by category then order then name.
"""Return all registered tools sorted by category rank, then order, then name.
Categories are ordered by :data:`_CATEGORY_ORDER` (Messaging first), so the first-class
Chat and Channels tools lead the menu; within a category, ``order`` then ``name`` break
ties.
Returns:
The registered tool instances in stable menu order.
"""
return sorted(_REGISTRY.values(), key=lambda t: (t.category, t.order, t.name))
return sorted(
_REGISTRY.values(),
key=lambda t: (_category_rank(t.category), t.category, t.order, t.name),
)
def get_tool(name: str) -> Optional[Tool]:
+217
View File
@@ -0,0 +1,217 @@
"""The ``channels`` tool: create, join, and manage mesh channels.
Interactively it opens a full-screen channel manager (see :mod:`meshtools.ui.channels`) — a
first-class, phone-app-style experience for creating private channels, adding public ``#``
channels, joining with a key, importing a scanned ``meshcore://`` link, and sharing any
channel as a QR code. On the CLI it exposes ``list``, ``add``, ``join``, ``import``, and
``share`` subcommands for scripted use.
Channels are also *listed* by the ``chat`` tool for picking a conversation; this tool owns
everything to do with configuring the slots themselves.
"""
from __future__ import annotations
from typing import Any, Optional
import typer
from rich.console import Group
from rich.table import Table
from rich.text import Text
from ..context import AppContext
from ..core.channels import (
derive_secret,
normalize_secret,
parse_share_url,
random_secret,
share_url,
)
from .base import Tool, ToolResult, register
@register
class ChannelsTool(Tool):
"""Create, join, share, and manage the device's mesh channels."""
name = "channels"
help = "Create, join, and share mesh channels (with QR codes)."
category = "Messaging"
order = 20
async def run(self, ctx: AppContext, params: dict[str, Any]) -> ToolResult:
"""Open the channel manager (menu) or run a scripted action (CLI).
Args:
ctx: Shared application context.
params: A ``cli_action`` with its arguments (CLI), or empty for the menu.
Returns:
A :class:`ToolResult` summarizing what happened.
"""
action = params.get("cli_action")
if action is not None:
return await self._run_cli(ctx, action, params)
from ..ui.channels import manage_channels
changes = await manage_channels(ctx)
message = (
f"[ok]✓[/ok] applied [brand]{changes}[/brand] channel change(s)"
if changes
else None
)
return ToolResult(summary={"changes": changes}, message=message)
# -- CLI --------------------------------------------------------------------
async def _run_cli(
self, ctx: AppContext, action: str, params: dict[str, Any]
) -> ToolResult:
"""Dispatch a scripted CLI action."""
if action == "add":
return await self._cli_add(ctx, params)
if action == "join":
return await self._cli_join(ctx, params)
if action == "import":
return await self._cli_import(ctx, params)
if action == "share":
return await self._cli_share(ctx, params)
return await self._cli_list(ctx)
async def _cli_list(self, ctx: AppContext) -> ToolResult:
"""List the configured channel slots."""
from ..ui.channels import _read_slots
device = await ctx.device()
slots = await _read_slots(device)
if not slots:
ctx.ui.note("[muted]no channels configured[/muted]")
return ToolResult(summary={"channels": 0})
table = Table(title="Channels", border_style="muted", expand=False)
table.add_column("Slot", justify="right")
table.add_column("Name")
table.add_column("Type")
table.add_column("Hash")
for slot in slots:
table.add_row(
str(slot.idx),
Text(slot.name, style="brand"),
"public" if slot.is_public else "private",
slot.hash,
)
ctx.ui.show(table)
return ToolResult(summary={"channels": len(slots)})
async def _cli_add(self, ctx: AppContext, params: dict[str, Any]) -> ToolResult:
"""Add a public, private-random, or explicitly-keyed channel on a slot."""
device = await ctx.device()
idx = int(params["index"])
name = str(params["name"])
if name.startswith("#"): # public: firmware derives the key from the name
await device.set_channel(idx, name, None)
secret = derive_secret(name)
else:
secret = (
normalize_secret(params["secret"]) if params.get("secret") else random_secret()
)
await device.set_channel(idx, name, secret)
ctx.ui.note(f"[ok]✓[/ok] channel [brand]{name}[/brand] set on slot {idx}")
self._print_share(ctx, name, secret)
return ToolResult(summary={"index": idx, "name": name})
async def _cli_join(self, ctx: AppContext, params: dict[str, Any]) -> ToolResult:
"""Join a channel from a name and an explicit key."""
device = await ctx.device()
idx = int(params["index"])
name = str(params["name"])
secret = normalize_secret(str(params["secret"]))
await device.set_channel(idx, name, secret)
ctx.ui.note(f"[ok]✓[/ok] joined [brand]{name}[/brand] on slot {idx}")
return ToolResult(summary={"index": idx, "name": name})
async def _cli_import(self, ctx: AppContext, params: dict[str, Any]) -> ToolResult:
"""Import a channel from a ``meshcore://channel/add`` link."""
parsed = parse_share_url(str(params["url"]))
if parsed is None:
raise typer.BadParameter("not a valid meshcore:// channel link")
name, secret = parsed
device = await ctx.device()
idx = int(params["index"])
await device.set_channel(idx, name, secret)
ctx.ui.note(f"[ok]✓[/ok] imported [brand]{name}[/brand] on slot {idx}")
return ToolResult(summary={"index": idx, "name": name})
async def _cli_share(self, ctx: AppContext, params: dict[str, Any]) -> ToolResult:
"""Print a channel's share link and QR code."""
from ..ui.channels import _read_slots
device = await ctx.device()
idx = int(params["index"])
slot = next((s for s in await _read_slots(device) if s.idx == idx), None)
if slot is None:
ctx.ui.note(f"[warn]slot {idx} is empty[/warn]")
return ToolResult(summary={"index": idx, "shared": False})
self._print_share(ctx, slot.name, slot.secret)
return ToolResult(summary={"index": idx, "shared": True})
@staticmethod
def _print_share(ctx: AppContext, name: str, secret: bytes) -> None:
"""Render a channel's QR code and share URL into the output surface."""
from ..ui.qr import qr_text
url = share_url(name, secret)
ctx.ui.show(Group(qr_text(url), Text(""), Text(url, style="accent")))
def register_cli(self, app: typer.Typer) -> None:
"""Register the ``channels`` subcommand group.
Args:
app: The Typer application.
"""
from ..cli import run_tool_command
channels_app = typer.Typer(
help=self.help, no_args_is_help=True, rich_markup_mode="rich"
)
@channels_app.command("list", help="List the configured channel slots.")
def _list_cmd() -> None:
run_tool_command(self, {"cli_action": "list"})
@channels_app.command("add", help="Add a channel (# name = public; else private).")
def _add_cmd(
index: int = typer.Argument(..., help="Channel slot index."),
name: str = typer.Argument(..., help="Channel name (leading # = public)."),
secret: Optional[str] = typer.Option(
None, "--secret", help="32-hex-char key (private only; random if omitted)."
),
) -> None:
run_tool_command(
self, {"cli_action": "add", "index": index, "name": name, "secret": secret}
)
@channels_app.command("join", help="Join a channel with its name and key.")
def _join_cmd(
index: int = typer.Argument(..., help="Channel slot index."),
name: str = typer.Argument(..., help="Channel name."),
secret: str = typer.Argument(..., help="32-hex-char (16-byte) key."),
) -> None:
run_tool_command(
self, {"cli_action": "join", "index": index, "name": name, "secret": secret}
)
@channels_app.command("import", help="Import a meshcore:// channel link.")
def _import_cmd(
index: int = typer.Argument(..., help="Channel slot index."),
url: str = typer.Argument(..., help="A meshcore://channel/add link."),
) -> None:
run_tool_command(self, {"cli_action": "import", "index": index, "url": url})
@channels_app.command("share", help="Print a channel's share link and QR code.")
def _share_cmd(
index: int = typer.Argument(..., help="Channel slot index."),
) -> None:
run_tool_command(self, {"cli_action": "share", "index": index})
app.add_typer(channels_app, name=self.name)
+3 -3
View File
@@ -3,7 +3,7 @@
Interactively it opens a conversation picker and then a live, full-screen chat (see
:mod:`meshtools.ui.chat`) where sent and received messages stream together. On the CLI it
exposes ``send``, ``history``, and ``list`` subcommands for scripted use. Channels are only
*listed* here for picking; creating and editing channel slots lives in the ``config`` tool.
*listed* here for picking; creating and editing channel slots lives in the ``channels`` tool.
Every inbound message is recorded to history by the always-on
:class:`~meshtools.services.chat_service.ChatService`, and outbound messages are recorded on
@@ -383,10 +383,10 @@ async def _read_channels(device: Device) -> list[Conversation]:
if channel:
name = str(channel.get("channel_name") or idx)
conversations.append(
Conversation(label=f"#{name}", is_channel=True, channel_idx=idx)
Conversation(label=name, is_channel=True, channel_idx=idx)
)
if not any(c.channel_idx == 0 for c in conversations):
conversations.insert(0, Conversation(label="#public", is_channel=True, channel_idx=0))
conversations.insert(0, Conversation(label="Public", is_channel=True, channel_idx=0))
return conversations
+477
View File
@@ -0,0 +1,477 @@
"""The interactive channel manager, rendered in the full-screen session.
This is the user-facing channel-management experience for the ``channels`` tool: a live
list of the device's channel slots with, for each, a detail view that shows the sharable
QR code and key, renames or re-keys it, opens it in chat, or clears it. New channels are
created four ways — a fresh private channel (random key), a public ``#`` channel (key
derived from the name), joining by pasting a key, or importing a scanned ``meshcore://``
link. Every change is written to the device immediately (like a phone app), so the list you
see always reflects the radio.
The module sits in the UI layer but, like the config editor and chat screen, is allowed to
depend on the context and services; it owns no persistence of its own.
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import TYPE_CHECKING, Optional
from rich.console import Group
from rich.table import Table
from rich.text import Text
from ..core.channels import (
channel_hash,
derive_secret,
full_channel_hash,
is_public_name,
normalize_secret,
parse_share_url,
random_secret,
share_url,
)
from ..core.connection import Device
from ..core.models import Conversation
from .qr import qr_text
from .tui import Choice, Separator
if TYPE_CHECKING:
from ..context import AppContext
#: How many channel slots the firmware exposes (0-based); matches the chat/backup probes.
_MAX_CHANNELS = 8
# Top-menu action sentinels (distinct from a plain slot index, which selects that channel).
_CREATE = "__create__"
_PUBLIC = "__public__"
_JOIN = "__join__"
_IMPORT = "__import__"
_REORDER = "__reorder__"
_BACK = "__back__"
# Channel-detail action sentinels.
_QR = "qr"
_KEY = "key"
_CHAT = "chat"
_EDIT = "edit"
_CLEAR = "clear"
@dataclass(slots=True)
class ChannelSlot:
"""One configured channel slot read back from the device.
Attributes:
idx: The 0-based slot index.
name: The channel's name.
secret: The channel's 16-byte shared secret.
"""
idx: int
name: str
secret: bytes
@property
def is_public(self) -> bool:
"""Whether this channel's key is derived from its name (a public channel)."""
return is_public_name(self.name) or self.secret == derive_secret(self.name)
@property
def hash(self) -> str:
"""The channel's two-character hash fingerprint (the leading byte MeshCore shows)."""
return channel_hash(self.secret)
@property
def full_hash(self) -> str:
"""The channel's complete ``sha256(secret)`` digest; its first byte is :attr:`hash`."""
return full_channel_hash(self.secret)
@property
def conversation(self) -> Conversation:
"""A :class:`~meshtools.core.models.Conversation` for opening this channel in chat."""
return Conversation(label=self.name, is_channel=True, channel_idx=self.idx)
async def manage_channels(ctx: "AppContext") -> int:
"""Run the interactive channel manager until the user backs out.
Args:
ctx: Shared application context (provides the connected device and UI surface).
Returns:
The number of channels created, changed, or cleared during the session.
"""
device = await ctx.device()
changes = 0
highlight: Optional[object] = None
while True:
slots = await _read_slots(device)
choice = await _main_menu(ctx, slots, default=highlight)
if choice in (None, _BACK):
return changes
highlight = choice
if choice == _CREATE:
changes += await _create_private(ctx, device, slots)
elif choice == _PUBLIC:
changes += await _add_public(ctx, device, slots)
elif choice == _JOIN:
changes += await _join_with_key(ctx, device, slots)
elif choice == _IMPORT:
changes += await _import_link(ctx, device, slots)
elif choice == _REORDER:
changes += await _reorder_channels(ctx, device, slots)
else: # an existing slot index
slot = next((s for s in slots if s.idx == choice), None)
if slot is not None:
changes += await _channel_detail(ctx, device, slot)
# unreachable
# --- reading -----------------------------------------------------------------
async def _read_slots(device: Device) -> list[ChannelSlot]:
"""Probe the channel slots and return the configured ones, in index order.
Args:
device: The connected device to query.
Returns:
One :class:`ChannelSlot` per configured slot (an empty slot is skipped).
"""
slots: list[ChannelSlot] = []
for idx in range(_MAX_CHANNELS):
try:
payload = await device.get_channel(idx)
except Exception: # noqa: BLE001 - firmware may not support channel reads
break
if payload and payload.get("channel_name"):
slots.append(
ChannelSlot(
idx=idx,
name=str(payload["channel_name"]),
secret=bytes(payload.get("channel_secret") or b"\x00" * 16),
)
)
return slots
def _next_free_slot(slots: list[ChannelSlot]) -> Optional[int]:
"""Return the lowest unused slot index, or ``None`` when every slot is full."""
used = {s.idx for s in slots}
return next((i for i in range(_MAX_CHANNELS) if i not in used), None)
def _slot_label(slot: ChannelSlot) -> str:
"""Format a channel for a menu row: name, public/private, and hash (no slot index)."""
lock = "# public " if slot.is_public else "🔒 private"
return f"{slot.name:<18.18} {lock} hash {slot.hash}"
# --- menus -------------------------------------------------------------------
async def _main_menu(
ctx: "AppContext", slots: list[ChannelSlot], *, default: object = None
) -> object:
"""Show the channel list and the add-a-channel actions; return the chosen value."""
items: list = [Separator("── Channels ──")]
if slots:
for slot in slots:
items.append(Choice(title=_slot_label(slot), value=slot.idx))
else:
items.append(Separator(" (no channels configured yet)"))
if len(slots) > 1:
items.append(Separator("── Channel list ──"))
items.append(Choice(title="↕ Reorder channels", value=_REORDER))
items.append(Separator("── Add a channel ──"))
items.append(Choice(title="+ New private channel (random key)", value=_CREATE))
items.append(Choice(title="# Public channel (key from its name)", value=_PUBLIC))
items.append(Choice(title="🔑 Join a channel with its key", value=_JOIN))
items.append(Choice(title="🔗 Import a meshcore:// link", value=_IMPORT))
items.append(Separator(" "))
items.append(Choice(title="Back", value=_BACK))
choice = await ctx.ui.select("Channels", items, default=default)
return _BACK if choice is None else choice
async def _channel_detail(ctx: "AppContext", device: Device, slot: ChannelSlot) -> int:
"""Show one channel's actions (QR, key, chat, rename, clear); return changes made."""
kind = "public" if slot.is_public else "private"
while True:
choice = await ctx.ui.select(
f"{slot.name} ({kind}, hash {slot.hash})",
[
Choice(title="📱 Show QR code (share)", value=_QR),
Choice(title="🔑 Show key", value=_KEY),
Choice(title="💬 Open in chat", value=_CHAT),
Choice(title="✎ Rename / change key", value=_EDIT),
Choice(title="🗑 Clear this slot", value=_CLEAR),
Choice(title="Back", value=_BACK),
],
)
if choice in (None, _BACK):
return 0
if choice == _QR:
await _show_share(ctx, slot.name, slot.secret)
elif choice == _KEY:
await _show_key(ctx, slot)
elif choice == _CHAT:
await _open_chat(ctx, slot)
elif choice == _EDIT:
if await _edit(ctx, device, slot):
return 1
elif choice == _CLEAR:
if await _clear(ctx, device, slot):
return 1
# --- create / join flows -----------------------------------------------------
async def _create_private(
ctx: "AppContext", device: Device, slots: list[ChannelSlot]
) -> int:
"""Create a private channel with a fresh random key on the next free slot."""
idx = await _pick_free_slot(ctx, slots)
if idx is None:
return 0
name = await ctx.ui.text("Channel name:", validate=_nonblank)
if not name:
return 0
secret = random_secret()
await device.set_channel(idx, name.strip(), secret)
ctx.ui.note(f"[ok]✓[/ok] created private channel [brand]{name.strip()}[/brand]")
await _show_share(ctx, name.strip(), secret, intro="Share this channel:")
return 1
async def _add_public(ctx: "AppContext", device: Device, slots: list[ChannelSlot]) -> int:
"""Create a public channel whose key is derived from its (``#``-prefixed) name."""
idx = await _pick_free_slot(ctx, slots)
if idx is None:
return 0
raw = await ctx.ui.text(
"Public channel name:",
help_text="A leading # is added automatically; the key is derived from the name.",
validate=_nonblank,
)
if not raw:
return 0
name = raw.strip()
if not name.startswith("#"):
name = f"#{name}"
secret = derive_secret(name) # what the firmware will compute; kept for the QR/share
await device.set_channel(idx, name, None) # None => firmware derives the key from name
ctx.ui.note(f"[ok]✓[/ok] added public channel [brand]{name}[/brand]")
await _show_share(ctx, name, secret, intro="Share this channel:")
return 1
async def _join_with_key(
ctx: "AppContext", device: Device, slots: list[ChannelSlot]
) -> int:
"""Join an existing private channel by entering its name and 16-byte key."""
idx = await _pick_free_slot(ctx, slots)
if idx is None:
return 0
name = await ctx.ui.text("Channel name:", validate=_nonblank)
if not name:
return 0
key = await ctx.ui.text(
"Channel key (32 hex characters / 16 bytes):", validate=_valid_secret
)
if not key:
return 0
secret = normalize_secret(key)
await device.set_channel(idx, name.strip(), secret)
ctx.ui.note(f"[ok]✓[/ok] joined [brand]{name.strip()}[/brand]")
return 1
async def _import_link(
ctx: "AppContext", device: Device, slots: list[ChannelSlot]
) -> int:
"""Import a channel from a pasted ``meshcore://channel/add`` link."""
idx = await _pick_free_slot(ctx, slots)
if idx is None:
return 0
url = await ctx.ui.text(
"Paste a meshcore:// channel link:", validate=_valid_link
)
if not url:
return 0
parsed = parse_share_url(url)
if parsed is None: # pragma: no cover - guarded by the validator
ctx.ui.note("[err]not a valid channel link[/err]")
return 0
name, secret = parsed
await device.set_channel(idx, name, secret)
ctx.ui.note(f"[ok]✓[/ok] imported [brand]{name}[/brand]")
return 1
async def _edit(ctx: "AppContext", device: Device, slot: ChannelSlot) -> bool:
"""Rename and/or re-key an existing channel; return whether it changed."""
name = await ctx.ui.text("Channel name:", default=slot.name, validate=_nonblank)
if name is None:
return False
name = name.strip()
key = await ctx.ui.text(
"Channel key (32 hex chars; blank to derive from the name):",
default="" if slot.is_public else slot.secret.hex(),
validate=_optional_secret,
)
if key is None:
return False
secret = normalize_secret(key) if key.strip() else None
await device.set_channel(slot.idx, name, secret)
ctx.ui.note(f"[ok]✓[/ok] updated channel [brand]{name}[/brand]")
return True
async def _clear(ctx: "AppContext", device: Device, slot: ChannelSlot) -> bool:
"""Clear a channel slot after confirmation; return whether it was cleared."""
if not await ctx.ui.confirm(
f"Clear [brand]{slot.name}[/brand]? This removes the channel from this device.",
default=False,
):
return False
await device.set_channel(slot.idx, "", None) # empty name => the slot reads as unused
ctx.ui.note(f"[warn]cleared channel {slot.name}[/warn]")
return True
# --- reordering --------------------------------------------------------------
async def _write_slot(device: Device, idx: int, slot: ChannelSlot) -> None:
"""Write ``slot``'s contents into slot ``idx`` (public channels re-derive their key)."""
secret = None if slot.is_public else slot.secret
await device.set_channel(idx, slot.name, secret)
async def _apply_order(
device: Device, slots: list[ChannelSlot], order: list[int]
) -> int:
"""Rewrite the channel slots so they display in ``order``; return the writes made.
``slots`` is the current list in slot-index order and ``order`` is a permutation of its
positions (as returned by the reorder screen). The channels are re-laid across the same
physical slot indices, lowest first, so their on-device order matches the new display
order. Only slots whose occupant actually changes are written. Reads come from the
in-memory snapshot, so the interleaved writes never clobber a not-yet-placed channel.
"""
indices = sorted(s.idx for s in slots) # the physical slots to fill, ascending
changes = 0
for target_idx, pos in zip(indices, order, strict=True):
slot = slots[pos]
if slot.idx == target_idx:
continue # already in place; no write needed
await _write_slot(device, target_idx, slot)
changes += 1
return changes
async def _reorder_channels(
ctx: "AppContext", device: Device, slots: list[ChannelSlot]
) -> int:
"""Let the user drag channels into a new order with the arrows; return writes made."""
if len(slots) < 2: # pragma: no cover - the menu only offers reorder with 2+ channels
return 0
labels = [_slot_label(slot) for slot in slots]
order = await ctx.ui.reorder("Reorder channels", labels)
if not order or order == list(range(len(slots))):
return 0 # cancelled or left unchanged
return await _apply_order(device, slots, order)
# --- shared views ------------------------------------------------------------
async def _show_share(
ctx: "AppContext", name: str, secret: bytes, *, intro: str = "Scan to add this channel:"
) -> None:
"""Show the channel's QR code and its share URL in a dismissable window."""
url = share_url(name, secret)
body = Group(
Text(intro, style="muted"),
Text(""),
qr_text(url),
Text(""),
Text(url, style="accent"),
)
await ctx.ui.view(body, title=f"Share {name}", footer_hint="Esc back")
def _hash_text(full_hash: str) -> Text:
"""Render a full channel hash with its first byte (the MeshCore fingerprint) highlighted."""
text = Text(no_wrap=True)
text.append(full_hash[:2], style="brand") # the leading byte the companion app reports
text.append(full_hash[2:], style="muted")
return text
async def _show_key(ctx: "AppContext", slot: ChannelSlot) -> None:
"""Show a channel's name, key, hash, and share URL in a small table."""
table = Table(border_style="muted", expand=False, show_header=False)
table.add_column("", style="muted")
table.add_column("")
table.add_row("Name", Text(slot.name, style="brand"))
table.add_row("Type", "public (key from name)" if slot.is_public else "private")
table.add_row("Hash", _hash_text(slot.full_hash))
table.add_row("Key", Text(slot.secret.hex(), style="warn"))
table.add_row("Link", Text(share_url(slot.name, slot.secret), style="accent"))
await ctx.ui.view(table, title=f"Key — {slot.name}", footer_hint="Esc back")
async def _open_chat(ctx: "AppContext", slot: ChannelSlot) -> None:
"""Open this channel in the live chat screen."""
from .chat import open_chat
await open_chat(ctx, slot.conversation)
async def _pick_free_slot(ctx: "AppContext", slots: list[ChannelSlot]) -> Optional[int]:
"""Return the next free slot, warning (and returning ``None``) if all are full."""
idx = _next_free_slot(slots)
if idx is None:
await ctx.ui.view(
Text.from_markup(
"[warn]All channel slots are full.[/warn] Clear one first, then try again."
),
title="No free slot",
)
return idx
# --- validators --------------------------------------------------------------
def _nonblank(text: str) -> bool | str:
"""Require a non-empty name."""
return bool(text.strip()) or "Enter a channel name."
def _valid_secret(text: str) -> bool | str:
"""Require a valid 16-byte hex key."""
try:
normalize_secret(text)
return True
except ValueError as exc:
return str(exc)
def _optional_secret(text: str) -> bool | str:
"""Accept a blank key (derive from name) or a valid 16-byte hex key."""
return True if not text.strip() else _valid_secret(text)
def _valid_link(text: str) -> bool | str:
"""Require a parseable ``meshcore://channel/add`` link."""
return parse_share_url(text) is not None or "Not a valid meshcore:// channel link."
+3 -43
View File
@@ -2,8 +2,9 @@
Drives the menu flow for the ``config`` tool: it presents the device's settings (each row
showing its current value and any staged change), lets the user stage changes setting-by-
setting with type-aware prompts and validation, and handles custom variables, channels,
backup/restore and the destructive "danger zone". It returns an operation list for
setting with type-aware prompts and validation, and handles custom variables,
backup/restore and the destructive "danger zone". (Channels have their own first-class
manager — see the ``channels`` tool.) It returns an operation list for
:class:`~meshtools.tools.config.ConfigTool` to execute and log; it performs no device writes
itself (except danger-zone actions, which run immediately and show their result at once).
"""
@@ -30,7 +31,6 @@ from .tui import Choice, Separator
# Menu action sentinels (distinct from setting keys, which are plain strings).
_PRESETS = "__presets__"
_CUSTOM = "__custom__"
_CHANNELS = "__channels__"
_BACKUP = "__backup__"
_RESTORE = "__restore__"
_DANGER = "__danger__"
@@ -75,8 +75,6 @@ async def edit_config(ctx: AppContext) -> Optional[list[tuple]]:
await _stage_preset(ctx, pending)
elif choice == _CUSTOM:
await _stage_custom_var(ctx, custom, extra_ops)
elif choice == _CHANNELS:
await _stage_channel(ctx, extra_ops)
elif choice == _BACKUP:
await _stage_backup(ctx, extra_ops)
elif choice == _RESTORE:
@@ -168,7 +166,6 @@ async def _main_menu(
items.append(Choice(title="View current config (full table)", value=_VIEW))
items.append(Choice(title="Radio presets (standard configs)", value=_PRESETS))
items.append(Choice(title="Custom / experimental vars", value=_CUSTOM))
items.append(Choice(title="Channels", value=_CHANNELS))
items.append(Choice(title="Backup to file", value=_BACKUP))
items.append(Choice(title="Restore from file", value=_RESTORE))
items.append(Choice(title="⚠ Danger zone", value=_DANGER))
@@ -267,24 +264,6 @@ async def _stage_custom_var(
extra_ops.append(("set_custom", key.strip(), value))
async def _stage_channel(ctx: AppContext, extra_ops: list[tuple]) -> None:
"""Prompt for a channel slot's name/secret and stage a set operation."""
idx_raw = await ctx.ui.text("Channel index:", default="0", validate=_is_int)
if idx_raw is None:
return
name = await ctx.ui.text("Channel name (leading # derives the secret from the name):")
if not name:
return
secret_hex = await ctx.ui.text(
"Secret (32 hex chars / 16 bytes; blank to derive from name):",
validate=_is_optional_secret,
)
if secret_hex is None:
return
secret = bytes.fromhex(secret_hex) if secret_hex else None
extra_ops.append(("set_channel", int(idx_raw), name, secret))
async def _stage_backup(ctx: AppContext, extra_ops: list[tuple]) -> None:
"""Prompt for a backup destination path and stage the operation."""
path = await ctx.ui.path("Write backup to:", default="meshtools-config.toml")
@@ -372,15 +351,6 @@ async def _confirm_typed(ctx: AppContext, word: str, *, warning: str = "") -> bo
# --- validators --------------------------------------------------------------
def _is_int(text: str) -> bool | str:
"""Validate that ``text`` is an integer."""
try:
int(text)
return True
except ValueError:
return "Enter a whole number."
def _is_hex(text: str) -> bool | str:
"""Validate that ``text`` is a hex string."""
try:
@@ -388,13 +358,3 @@ def _is_hex(text: str) -> bool | str:
return True
except ValueError:
return "Enter hex characters only."
def _is_optional_secret(text: str) -> bool | str:
"""Validate an optional 16-byte hex channel secret."""
if text == "":
return True
try:
return len(bytes.fromhex(text)) == 16 or "Secret must be exactly 16 bytes (32 hex chars)."
except ValueError:
return "Enter hex characters only."
+56
View File
@@ -0,0 +1,56 @@
"""Render a string as a scannable QR code in the terminal, using half-block characters.
Each character cell stacks two vertical modules — ``▀`` (upper only), ``▄`` (lower only),
``█`` (both), space (neither) — so a code renders at half the row height of a full-block
one and comfortably fits an interactive window. Modules are always drawn black on a white
background (with the mandatory quiet-zone border), so a phone camera reads the code no
matter what colour theme the terminal itself is using.
"""
from __future__ import annotations
from rich.text import Text
#: Quiet-zone width (modules) around the code. Scanners need at least 4 to lock on.
_BORDER = 4
#: Half-block glyph for each (top-module-dark, bottom-module-dark) pair. Drawn as black
#: ink, so a dark module is black and a light module shows the white cell background.
_GLYPH = {
(True, True): "█", # █ full block
(True, False): "▀", # ▀ upper half
(False, True): "▄", # ▄ lower half
(False, False): " ",
}
#: Constant style for every glyph: black ink on a white field. Naming both ends means the
#: contrast holds regardless of the surrounding terminal palette.
_STYLE = "black on white"
def qr_text(data: str, *, error: str = "m") -> Text:
"""Render ``data`` as a QR code built from half-block characters.
Args:
data: The string to encode (e.g. a ``meshcore://channel/add`` share URL).
error: QR error-correction level — ``l``/``m``/``q``/``h`` (default ``m``).
Returns:
A Rich :class:`~rich.text.Text` (no-wrap) whose lines draw the code, ready to hand
to ``ctx.ui.view`` / ``ctx.ui.show``.
"""
import segno
qr = segno.make(data, error=error)
rows = [[bool(v) for v in row] for row in qr.matrix_iter(border=_BORDER)]
# Pair rows top-to-bottom; pad an odd final row with light (white) modules so the last
# half-block cell renders cleanly.
if len(rows) % 2:
rows.append([False] * len(rows[0]))
text = Text(no_wrap=True)
for top, bottom in zip(rows[0::2], rows[1::2], strict=True):
line = "".join(_GLYPH[(t, b)] for t, b in zip(top, bottom, strict=True))
text.append(line, style=_STYLE)
text.append("\n")
return text
+12
View File
@@ -57,6 +57,10 @@ class Ui:
"""Prompt the user to choose one item; return its value or ``None`` if cancelled."""
raise NotImplementedError
async def reorder(self, title: str, labels: list[str]) -> list[int]:
"""Let the user rearrange rows with the arrows; return the new order of row indices."""
raise NotImplementedError
async def text(
self,
title: str,
@@ -129,6 +133,10 @@ class PlainUi(Ui):
"""Unsupported in scripted CLI mode."""
raise self._no_prompt()
async def reorder(self, title: str, labels: list[str]) -> list[int]:
"""Unsupported in scripted CLI mode."""
raise self._no_prompt()
async def text(
self,
title: str,
@@ -239,6 +247,10 @@ class TuiUi(Ui):
"""Delegate to the session's select screen."""
return await self.session.select(title, items, default=default)
async def reorder(self, title: str, labels: list[str]) -> list[int]:
"""Delegate to the session's reorder screen."""
return await self.session.reorder(title, labels)
async def text(
self,
title: str,
+76
View File
@@ -159,6 +159,82 @@ class SelectScreen(Screen):
self._index = 0
class ReorderScreen(Screen):
"""A list whose rows the user rearranges in place with the arrow keys.
Move the cursor with ↑/↓; press Enter to *grab* the highlighted row, then ↑/↓ carry it
up and down the list; press Enter again to *drop* it. The screen stays open until Esc,
at which point it resolves with the final order as a list of the original row indices
(so ``[2, 0, 1]`` means "the row that started third is now first").
"""
def __init__(self, title: str, labels: list[str]) -> None:
"""Build a reorder screen.
Args:
title: Heading shown above the list.
labels: The row labels, in their current order.
"""
super().__init__()
self.title = title
self._labels = list(labels)
# order[position] == the label's original index; a moved row carries its index along.
self._order = list(range(len(labels)))
self._index = 0
self._grabbed = False
@property
def footer_hint(self) -> str: # type: ignore[override]
"""Key hint, phrased for whether a row is currently grabbed."""
if self._grabbed:
return "↑↓ move row · Enter drop · Esc done"
return "↑↓ choose · Enter grab · Esc done"
def render_body(self, width: int) -> list[str]:
"""Render each row, marking the cursor (and, when grabbed, the moving row)."""
lines: list[str] = []
for pos, orig in enumerate(self._order):
is_cursor = pos == self._index
if is_cursor and self._grabbed:
pointer, style = "▸ ", "brand"
elif is_cursor:
pointer, style = "❯ ", "brand"
else:
pointer, style = " ", ""
text = Text(pointer + self._labels[orig], style=style, no_wrap=True,
overflow="ellipsis")
text.truncate(width)
lines.append(render_to_ansi(text, width))
self._cursor = self._index
return lines
def cursor_line(self) -> Optional[int]:
"""Return the body line index of the cursor row, so the session keeps it in view."""
return getattr(self, "_cursor", None)
def handle(self, action: str, data: str = "") -> None:
"""Move the cursor, carry a grabbed row, toggle grab, or finish on Esc."""
n = len(self._order)
if action == "up":
if self._grabbed and self._index > 0:
self._order[self._index - 1], self._order[self._index] = (
self._order[self._index], self._order[self._index - 1])
self._index -= 1
elif not self._grabbed and n:
self._index = (self._index - 1) % n
elif action == "down":
if self._grabbed and self._index < n - 1:
self._order[self._index + 1], self._order[self._index] = (
self._order[self._index], self._order[self._index + 1])
self._index += 1
elif not self._grabbed and n:
self._index = (self._index + 1) % n
elif action == "enter":
self._grabbed = not self._grabbed
elif action == "escape":
self.resolve(list(self._order))
def _cursor_line(rows: list, selected: Optional[Choice], filtered: bool) -> Optional[int]:
"""Compute the rendered body line index of the selected row.
+7 -1
View File
@@ -28,7 +28,7 @@ from . import frame
from .progress import TuiProgress
from .prompt import AutocompleteScreen, ConfirmScreen, TextScreen, Validator
from .screen import CANCEL, Screen, ScrollScreen
from .select import Choice, SelectScreen, Separator
from .select import Choice, ReorderScreen, SelectScreen, Separator
#: Maps prompt_toolkit keys to the normalized action names screens understand.
_KEY_ACTIONS: dict[Any, str] = {
@@ -125,6 +125,11 @@ class TuiSession:
result = await self.run_screen(SelectScreen(title, items, default=default))
return None if result is CANCEL else result
async def reorder(self, title: str, labels: list[str]) -> list[int]:
"""Show a drag-with-arrows reorder screen; return the final order of row indices."""
result = await self.run_screen(ReorderScreen(title, labels))
return list(range(len(labels))) if result is CANCEL else result
async def text(
self,
title: str,
@@ -312,5 +317,6 @@ __all__ = [
"Choice",
"Separator",
"SelectScreen",
"ReorderScreen",
"ScrollScreen",
]
+1
View File
@@ -19,6 +19,7 @@ dependencies = [
"rich>=13.7", # modern console rendering (progress, tables, panels)
"prompt_toolkit>=3.0", # full-screen interactive TUI (terminal, input, resize)
"plotly>=5.20", # interactive self-contained HTML charts
"segno>=1.6", # pure-Python QR codes (channel share links)
"pyvis>=0.3.2", # interactive HTML mesh graphs
"networkx>=3.2", # graph model + path scoring
"tomli>=2.0; python_version < '3.11'", # TOML config reader on 3.10
+226
View File
@@ -0,0 +1,226 @@
"""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 meshtools.context import AppContext
from meshtools.core.admin_store import AdminStore
from meshtools.core.channels import (
CHANNEL_SECRET_BYTES,
channel_hash,
derive_secret,
full_channel_hash,
normalize_secret,
parse_share_url,
random_secret,
share_url,
)
from meshtools.core.config import Settings
from meshtools.core.device_store import DeviceStore
from meshtools.persistence.repository import Repository
from meshtools.tools.channels import ChannelsTool
from meshtools.ui.channels import ChannelSlot, _apply_order, _next_free_slot, _read_slots
from meshtools.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
# -- 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 private.conversation.is_channel and private.conversation.channel_idx == 2
assert private.conversation.label == "Ops" # raw name, no forced leading '#'
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) == 1
full = [ChannelSlot(idx=i, name=str(i), secret=b"\x00" * 16) for i in range(8)]
assert _next_free_slot(full) is None
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_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_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))
# -- 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_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_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_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_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"})