Jean-Pierre MartineauandClaude Fable 5
7ed31c8f31
The three-threshold bar meter compressed a week of traffic into one of four states; braille cells pack two dot columns of four rows each, so the same 8-cell lane now draws a real 16-point histogram — each bucket ~10.5 hours — and a channel's rhythm (bursts, lulls, decay) is legible at a glance. channel_stats() gains the per-bucket histogram, bucketing the window's timestamps in Python off one filtered query; recent is now simply the histogram's sum. Bars scale to the row's own busiest bucket — the sparkline shows shape, not cross-channel volume (the MSGS lane does that) — with non-empty buckets floored at one dot so a lone message never vanishes. A silent window keeps the muted ······· placeholder. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
497 lines
23 KiB
Python
497 lines
23 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,
|
||
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_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
|
||
|
||
|
||
# -- 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_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))
|
||
|
||
|
||
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_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_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_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-week slice, and the last time."""
|
||
from datetime import timedelta
|
||
|
||
from meshterm.core.models import ChatMessage, utcnow
|
||
|
||
stale = utcnow() - timedelta(days=30) # outside the 7-day activity window
|
||
fresh = utcnow() - timedelta(minutes=5)
|
||
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 oldest-first: the 5-minute-old message lands in the
|
||
# newest (rightmost) bucket and the 30-day-old ones land nowhere.
|
||
assert len(ops.histogram) == 16
|
||
assert ops.histogram[-1] == 1 and sum(ops.histogram) == 1
|
||
|
||
|
||
def test_activity_sparkline_packs_16_buckets_into_braille() -> None:
|
||
"""The sparkline draws two histogram buckets per braille cell, scaled to the peak."""
|
||
from meshterm.ui.channels import _activity_sparkline
|
||
|
||
assert _activity_sparkline((0,) * 16).plain == "········" # silent window
|
||
assert _activity_sparkline(()).plain == "········" # a channel with no stats at all
|
||
assert _activity_sparkline((4,) * 16).plain == "⣿" * 8 # flat-out: every dot lit
|
||
# A lone message in the newest bucket: full-height bar (it *is* the peak) in the final
|
||
# cell's right column; every other column stays empty.
|
||
lone = _activity_sparkline((0,) * 15 + (1,)).plain
|
||
assert lone == "⠀" * 7 + "⢸"
|
||
# Heights scale to the row's own peak, keeping any non-empty bucket at least one dot:
|
||
# counts 1/8 of the peak still show, half-peak reaches half height.
|
||
ramp = _activity_sparkline((8, 0) * 7 + (1, 4)).plain
|
||
assert ramp[-1] == chr(0x2800 | 0x40 | 0xA0) # left col 1 dot, right col 2 dots
|
||
|
||
|
||
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_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 and "private" in plain and slot.hash 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: a full-height bar
|
||
# in the final braille cell's right column, with the rest of the window empty.
|
||
assert plain.rstrip().endswith("⠀" * 7 + "⢸") # the sparkline is the final lane
|
||
|
||
|
||
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_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))
|
||
assert summary == "Slot 3 · 1 message · 1 unread · last message just 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_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"})
|