Compare commits

..
16 changed files with 472 additions and 117 deletions

No files matched your search

+1 -1
View File
@@ -258,7 +258,7 @@ def start() -> None:
if fatal_error is not None:
print("Fatal error:", fatal_error)
traceback.print_exc()
traceback.print_exception(type(fatal_error), fatal_error, fatal_error.__traceback__)
sys.exit(1)
if interrupted:
+1
View File
@@ -160,6 +160,7 @@ node_ignored, "Node ignored", ""
[Channels.channel]
title, "Channels"
enabled, "Enabled", "Enable or disable this channel without clearing its name or encryption key."
channel_num, "Channel number", "The index number of this channel."
psk, "PSK", "The channel's encryption key."
name, "Name", "The channel's name."
+165 -81
View File
@@ -35,6 +35,8 @@ from contact.ui.nav_utils import (
wrap_text,
truncate_with_ellipsis,
pad_to_width,
text_width,
slice_to_width,
)
from contact.utilities.singleton import ui_state, interface_state, menu_state, app_state
@@ -43,6 +45,8 @@ MIN_COL = 1 # "effectively zero" without breaking curses
RESIZE_DEBOUNCE_MS = 250
root_win = None
nodes_pad = None
entry_win = None
input_text = ""
def refresh_log_viewer() -> None:
@@ -243,6 +247,8 @@ def compute_widths(total_w: int, focus: int):
def paint_frame(win, selected: bool) -> None:
if win is not entry_win and ui_state.input_focused:
selected = False
win.attrset(get_color("window_frame_selected") if selected else get_color("window_frame"))
win.box()
win.attrset(get_color("window_frame"))
@@ -491,6 +497,8 @@ def main_ui(stdscr: curses.window) -> None:
root_win = stdscr
input_text = ""
ui_state.input_cursor = 0
ui_state.input_focused = False
queued_char = None
stdscr.keypad(True)
get_channels()
@@ -555,8 +563,7 @@ def main_ui(stdscr: curses.window) -> None:
with app_state.lock:
process_pending_ui_updates(stdscr)
entry_display = f"{ui_state.reply_context}{input_text or ''}"
draw_text_field(entry_win, f"Message: {entry_display[-(stdscr.getmaxyx()[1] - 10):]}", get_color("input"))
draw_message_input(input_text)
# Get user input from entry window
try:
@@ -596,6 +603,7 @@ def main_ui(stdscr: curses.window) -> None:
elif char in (chr(curses.KEY_ENTER), chr(10), chr(13)):
input_text = handle_enter(input_text)
ui_state.input_cursor = min(ui_state.input_cursor, len(input_text))
elif char in (curses.KEY_F4, chr(20)): # Ctrl + t and F4 for Traceroute
handle_ctrl_t(stdscr)
@@ -603,9 +611,14 @@ def main_ui(stdscr: curses.window) -> None:
elif char == curses.KEY_F5:
handle_f5_key(stdscr)
elif char in (curses.KEY_BACKSPACE, chr(127)):
elif char in (curses.KEY_BACKSPACE, chr(127), chr(8)):
input_text = handle_backspace(entry_win, input_text)
elif char == curses.KEY_DC:
focus_message_input()
position = ui_state.input_cursor
input_text = input_text[:position] + input_text[position + 1:]
elif char in (curses.KEY_F12, "`"): # ` Launch the settings interface
handle_backtick(stdscr)
@@ -619,7 +632,6 @@ def main_ui(stdscr: curses.window) -> None:
entry_win.erase()
elif char == curses.KEY_RESIZE:
input_text = ""
queued_char = drain_resize_events(entry_win)
handle_resize(stdscr, False)
continue
@@ -645,15 +657,66 @@ def main_ui(stdscr: curses.window) -> None:
break
else:
# Append typed character to input text
if isinstance(char, str):
input_text += char
else:
input_text += chr(char)
if isinstance(char, str) and char.isprintable():
input_text = insert_input_text(input_text, char)
def focus_message_input() -> None:
if ui_state.input_focused:
return
# Keep the conversation visible in single-pane mode while composing.
handle_function_keys(curses.KEY_F2)
ui_state.input_focused = True
refresh_message_highlight()
refresh_main_window(1, selected=False)
def leave_message_input() -> None:
ui_state.input_focused = False
ui_state.current_window = 1
refresh_message_highlight()
refresh_main_window(1, selected=True)
def insert_input_text(draft: str, text: str) -> str:
focus_message_input()
position = min(ui_state.input_cursor, len(draft))
ui_state.input_cursor = position + len(text)
return draft[:position] + text + draft[position:]
def draw_message_input(draft: str) -> None:
"""Redraw the whole input row and scroll it to keep the draft cursor visible."""
width = entry_win.getmaxyx()[1]
prompt = slice_to_width("Message: ", max(0, width - 3))
available = max(1, width - 2 - text_width(prompt))
ui_state.input_cursor = max(0, min(ui_state.input_cursor, len(draft)))
content = ui_state.reply_context + draft
cursor = len(ui_state.reply_context) + ui_state.input_cursor
start = cursor
used = 0
while start > 0 and used + text_width(content[start - 1]) < available:
start -= 1
used += text_width(content[start])
visible = slice_to_width(content[start:], available)
draw_text_field(entry_win, prompt + visible, get_color("input"))
paint_frame(entry_win, selected=ui_state.input_focused)
try:
hint = " Ctrl+K Help "
if width >= len(hint) + 4:
entry_win.addstr(0, width - len(hint) - 2, hint, get_color("commands"))
curses.curs_set(1 if ui_state.input_focused else 0)
entry_win.move(1, min(width - 2, 1 + text_width(prompt) + used))
except curses.error:
pass
entry_win.refresh()
def handle_up() -> None:
"""Handle key up events to scroll the current window."""
if ui_state.input_focused:
leave_message_input()
return
if ui_state.current_window == 0:
scroll_channels(-1)
elif ui_state.current_window == 1:
@@ -664,6 +727,9 @@ def handle_up() -> None:
def handle_down() -> None:
"""Handle key down events to scroll the current window."""
if ui_state.input_focused:
leave_message_input()
return
if ui_state.current_window == 0:
scroll_channels(1)
elif ui_state.current_window == 1:
@@ -674,6 +740,9 @@ def handle_down() -> None:
def handle_home() -> None:
"""Handle home key events to select the first item in the current window."""
if ui_state.input_focused:
ui_state.input_cursor = 0
return
if ui_state.current_window == 0:
select_channel(0)
elif ui_state.current_window == 1:
@@ -688,6 +757,9 @@ def handle_home() -> None:
def handle_end() -> None:
"""Handle end key events to select the last item in the current window."""
if ui_state.input_focused:
ui_state.input_cursor = len(input_text)
return
if ui_state.current_window == 0:
select_channel(len(ui_state.channel_list) - 1)
elif ui_state.current_window == 1:
@@ -728,6 +800,9 @@ def handle_pagedown() -> None:
def handle_leftright(char: int) -> None:
"""Handle left/right key events to switch between windows."""
delta = -1 if char == curses.KEY_LEFT else 1
if ui_state.input_focused:
ui_state.input_cursor = max(0, min(len(input_text), ui_state.input_cursor + delta))
return
old_window = ui_state.current_window
ui_state.current_window = (ui_state.current_window + delta) % 3
if ui_state.single_pane_mode:
@@ -748,6 +823,8 @@ def handle_leftright(char: int) -> None:
def handle_function_keys(char: int) -> None:
"""Switch windows using F1/F2/F3."""
if ui_state.input_focused:
leave_message_input()
if char == curses.KEY_F1:
target = 0
elif char == curses.KEY_F2:
@@ -1107,14 +1184,12 @@ def handle_ctrl_t(stdscr: curses.window) -> None:
def handle_backspace(entry_win: curses.window, input_text: str) -> str:
"""Handle backspace key events to remove the last character from input text."""
if input_text:
input_text = input_text[:-1]
y, x = entry_win.getyx()
entry_win.move(y, x - 1)
entry_win.addch(" ") #
entry_win.move(y, x - 1)
entry_win.refresh()
"""Delete before the draft cursor; the next frame redraws the entire line."""
focus_message_input()
position = ui_state.input_cursor
if position > 0:
input_text = input_text[:position - 1] + input_text[position:]
ui_state.input_cursor -= 1
return input_text
@@ -1134,74 +1209,83 @@ def handle_backtick(stdscr: curses.window) -> None:
handle_resize(stdscr, False)
return
options = ["Local settings"]
remote_node_num = None
selected_node_num = None
if ui_state.node_list:
remote_node_num = ui_state.node_list[ui_state.selected_node]
options.append(f"Remote admin: {get_name_from_database(remote_node_num)}")
choice = get_list_input("Open settings for", None, options)
if choice == "Local settings":
settings_menu(stdscr, interface)
elif remote_node_num is not None:
wait_win = None
selected_node_num = ui_state.node_list[ui_state.selected_node]
def remote_status(message: Optional[str]) -> None:
nonlocal wait_win
if message is None:
# The local node is already the target for normal settings; do not offer a
# misleading remote-admin choice when it is selected in the node pane.
if selected_node_num == interface_state.myNodeNum:
settings_menu(stdscr, interface)
else:
options = ["Local settings"]
remote_node_num = None
if selected_node_num is not None:
remote_node_num = selected_node_num
options.append(f"Remote admin: {get_name_from_database(remote_node_num)}")
choice = get_list_input("Open settings for", None, options)
if choice == "Local settings":
settings_menu(stdscr, interface)
elif remote_node_num is not None:
wait_win = None
def remote_status(message: Optional[str]) -> None:
nonlocal wait_win
if message is None:
if wait_win is not None:
wait_win.erase()
wait_win.refresh()
wait_win = None
return
if wait_win is None:
wait_win = show_remote_admin_wait(stdscr, get_name_from_database(remote_node_num))
update_remote_admin_wait(wait_win, message)
def remote_cancel_requested() -> bool:
if wait_win is None:
return False
try:
wait_win.timeout(0)
return wait_win.getch() == 27
finally:
wait_win.timeout(-1)
try:
remote_node = interface.getNode(remote_node_num, False)
verify_remote_admin(
remote_node,
status_callback=remote_status,
cancel_callback=remote_cancel_requested,
)
remote_status(None)
settings_menu(
stdscr,
interface,
# Settings requests configs explicitly; avoid getNode's
# implicit channel request racing that setup.
node=remote_node,
remote=True,
status_callback=remote_status,
cancel_callback=remote_cancel_requested,
)
except RemoteAdminCancelled:
logging.info("Remote admin request cancelled for %s", remote_node_num)
remote_status(None)
except (SystemExit, PermissionError) as exc:
logging.warning("Remote admin was rejected for %s: %s", remote_node_num, exc)
if wait_win is not None:
wait_win.erase()
wait_win.refresh()
wait_win = None
return
if wait_win is None:
wait_win = show_remote_admin_wait(stdscr, get_name_from_database(remote_node_num))
update_remote_admin_wait(wait_win, message)
def remote_cancel_requested() -> bool:
if wait_win is None:
return False
try:
wait_win.timeout(0)
return wait_win.getch() == 27
finally:
wait_win.timeout(-1)
try:
remote_node = interface.getNode(remote_node_num, False)
verify_remote_admin(
remote_node,
status_callback=remote_status,
cancel_callback=remote_cancel_requested,
)
remote_status(None)
settings_menu(
stdscr,
interface,
# Settings requests configs explicitly; avoid getNode's
# implicit channel request racing that setup.
node=remote_node,
remote=True,
status_callback=remote_status,
cancel_callback=remote_cancel_requested,
)
except RemoteAdminCancelled:
logging.info("Remote admin request cancelled for %s", remote_node_num)
remote_status(None)
except (SystemExit, PermissionError) as exc:
logging.warning("Remote admin was rejected for %s: %s", remote_node_num, exc)
if wait_win is not None:
wait_win.erase()
wait_win.refresh()
contact.ui.dialog.dialog(
"Remote admin rejected",
"The selected node did not authorize this admin request.",
)
except Exception as exc:
logging.exception("Remote admin failed for %s", remote_node_num)
if wait_win is not None:
wait_win.erase()
wait_win.refresh()
contact.ui.dialog.dialog("Remote admin failed", str(exc))
contact.ui.dialog.dialog(
"Remote admin rejected",
"The selected node did not authorize this admin request.",
)
except Exception as exc:
logging.exception("Remote admin failed for %s", remote_node_num)
if wait_win is not None:
wait_win.erase()
wait_win.refresh()
contact.ui.dialog.dialog("Remote admin failed", str(exc))
ui_state.current_window = previous_window
ui_state.single_pane_mode = config.single_pane_mode.lower() == "true"
curses.curs_set(1)
@@ -1587,7 +1671,7 @@ def refresh_message_highlight() -> None:
previous_range = ui_state.highlighted_message_range
width = max(0, messages_win.getmaxyx()[1] - 2)
if not ui_state.channel_list or ui_state.current_window != 1:
if not ui_state.channel_list or ui_state.current_window != 1 or ui_state.input_focused:
if previous_range:
start, end, color = previous_range
for row in range(start, end):
+26 -3
View File
@@ -97,6 +97,17 @@ def get_translated_header(menu_path: List[str]) -> str:
return f"{admin_target_label} | {breadcrumb}" if admin_target_label else breadcrumb
def get_settings_option_label(option, field_info, menu_path):
if menu_path == ["Main Menu", "Channels"] and isinstance(field_info, dict):
name = field_info.get("name", (None, ""))[1]
if name:
return name
full_key = ".".join(transform_menu_path(menu_path) + [option])
if len(menu_path) == 3 and menu_path[1] == "Channels" and option == "enabled":
return field_mapping.get(full_key, "Enabled")
return field_mapping.get(full_key, option)
def display_menu() -> tuple[object, object]:
# if help_win:
# min_help_window_height = 6
@@ -140,7 +151,7 @@ def display_menu() -> tuple[object, object]:
field_info = menu_state.current_menu[option]
current_value = field_info[1] if isinstance(field_info, tuple) else ""
full_key = ".".join(transformed_path + [option])
display_name = field_mapping.get(full_key, option)
display_name = get_settings_option_label(option, field_info, menu_state.menu_path)
if full_key.startswith("config.network.ipv4_config.") and option in {"ip", "gateway", "subnet", "dns"}:
if isinstance(current_value, int):
@@ -719,7 +730,7 @@ def settings_menu(
full_key = ".".join(transformed_path + [selected_option])
# Fetch human-readable name from field_mapping
human_readable_name = field_mapping.get(full_key, selected_option)
human_readable_name = get_settings_option_label(selected_option, field_info, menu_state.menu_path)
if field is None and selected_option in ("ringtone", "messages"):
is_ringtone = selected_option == "ringtone"
@@ -779,6 +790,11 @@ def settings_menu(
new_value = current_value if new_values is None else [base64.b64decode(key) for key in new_values]
menu_state.start_index.pop()
elif field is None and selected_option == "enabled" and menu_state.menu_path[1] == "Channels":
new_value = get_list_input(human_readable_name, str(current_value), ["True", "False"])
new_value = new_value == "True" or new_value is True
menu_state.start_index.pop()
elif field.type == 8: # Handle boolean type
new_value = get_list_input(human_readable_name, str(current_value), ["True", "False"])
if new_value == "Not Set":
@@ -820,8 +836,15 @@ def settings_menu(
else: # Handle other field types
input_type = get_input_type_for_field(field)
allow_empty = (
selected_option == "name"
and len(menu_state.menu_path) == 3
and menu_state.menu_path[1] == "Channels"
and menu_state.menu_path[2] != "Channel 1"
)
new_value = get_text_input(
f"{human_readable_name} is currently: {current_value}", selected_option, input_type
f"{human_readable_name} is currently: {current_value}", selected_option, input_type,
allow_empty=allow_empty,
)
new_value = current_value if new_value is None else new_value
menu_state.start_index.pop()
+4
View File
@@ -93,6 +93,10 @@ def generate_menu_from_protobuf(interface: object, node: Any = None, include_app
current_channel = node.getChannelByChannelIndex(i)
if current_channel:
channel_config = extract_fields(channel, current_channel.settings)
channel_config = {
"enabled": (None, current_channel.role != channel_pb2.Channel.Role.DISABLED),
**channel_config,
}
menu_structure["Main Menu"]["Channels"][f"Channel {i + 1}"] = channel_config
# Add Radio Settings
+2
View File
@@ -25,6 +25,8 @@ class ChatUIState:
selected_message: int = 0
selected_node: int = 0
current_window: int = 0
input_focused: bool = False
input_cursor: int = 0
last_sent_time: float = 0.0
last_traceroute_time: float = 0.0
selected_index: int = 0
+4 -1
View File
@@ -38,7 +38,7 @@ def invalid_input(window: curses.window, message: str, redraw_func: Optional[cal
curses.curs_set(1)
def get_text_input(prompt: str, selected_config: str, input_type: str) -> Optional[str]:
def get_text_input(prompt: str, selected_config: str, input_type: str, *, allow_empty: bool = False) -> Optional[str]:
"""Handles user input with wrapped text for long prompts."""
def redraw_input_win():
@@ -127,6 +127,9 @@ def get_text_input(prompt: str, selected_config: str, input_type: str) -> Option
menu_state.need_redraw = True
if not user_input.strip():
if allow_empty:
user_input = ""
break
invalid_input(
input_win,
t("ui.error.value_empty", default="Value cannot be empty."),
+17 -23
View File
@@ -2,7 +2,6 @@ from meshtastic.protobuf import channel_pb2
from google.protobuf.message import Message
import logging
import base64
import time
DEVICE_REBOOT_KEYS = {"button_gpio", "buzzer_gpio", "role", "rebroadcast_mode"}
POWER_REBOOT_KEYS = {
@@ -100,22 +99,14 @@ def save_changes(interface, modified_settings, menu_state, node=None):
# Filter out empty keys
valid_keys = [key for key in admin_keys if key and key.strip() and key != b""]
if not valid_keys:
logging.warning("No valid admin keys provided. Skipping admin key update.")
else:
# Clear existing keys if needed
if security_config.admin_key:
logging.info("Clearing existing admin keys...")
del security_config.admin_key[:]
node.writeConfig("security")
time.sleep(2) # Give time for device to process
# Append new keys
for key in valid_keys:
logging.info(f"Adding admin key: {key}")
security_config.admin_key.append(key)
node.writeConfig("security")
logging.info("Admin keys updated successfully!")
# Update the repeated field in memory, then send one complete
# security config. Sending an empty list first can revoke the key
# authorizing this remote-admin session before a second write can
# restore the remaining keys.
del security_config.admin_key[:]
security_config.admin_key.extend(valid_keys)
node.writeConfig("security")
logging.info("Admin keys updated successfully (%d key(s)).", len(valid_keys))
# Backup 'admin_key' before removing it
admin_key_backup = modified_settings.get("admin_key", None)
@@ -169,7 +160,15 @@ def save_changes(interface, modified_settings, menu_state, node=None):
channel = node.channels[channel_num]
for key, value in modified_settings.items():
if key == "psk": # Special case: decode Base64 for psk
if key == "enabled":
if not value:
channel.role = channel_pb2.Channel.Role.DISABLED
elif channel.role == channel_pb2.Channel.Role.DISABLED:
channel.role = (
channel_pb2.Channel.Role.PRIMARY if channel_num == 0
else channel_pb2.Channel.Role.SECONDARY
)
elif key == "psk": # Special case: decode Base64 for psk
channel.settings.psk = base64.b64decode(value)
elif key in {"position_precision", "is_muted"}:
# These fields belong to ChannelSettings.module_settings,
@@ -178,11 +177,6 @@ def save_changes(interface, modified_settings, menu_state, node=None):
else:
setattr(channel.settings, key, value) # Use setattr for other fields
if channel_num == 0:
channel.role = channel_pb2.Channel.Role.PRIMARY
else:
channel.role = channel_pb2.Channel.Role.SECONDARY
node.writeChannel(channel_num)
logging.info(f"Updated Channel {channel_num} in {config_category}")
+32 -3
View File
@@ -1,6 +1,8 @@
import datetime
import logging
import re
import time
from functools import lru_cache
from typing import Optional, Union
from google.protobuf.message import DecodeError
@@ -10,14 +12,41 @@ import contact.ui.default_config as config
from contact.utilities.singleton import ui_state, interface_state
import contact.utilities.telemetry_beautifier as tb
NEW_MODEM_PRESET_NAMES = {
14: "TINY_FAST",
15: "TINY_SLOW",
16: "MEDIUM_TURBO",
}
@lru_cache(maxsize=None)
def _get_modem_preset_name(modem_preset_enum: int) -> str:
try:
return config_pb2.Config.LoRaConfig.ModemPreset.Name(modem_preset_enum)
except ValueError:
compatibility_name = NEW_MODEM_PRESET_NAMES.get(modem_preset_enum)
if compatibility_name is not None:
logging.warning(
"Modem preset %s is newer than the installed Meshtastic protobuf schema; "
"using compatibility name %s",
modem_preset_enum,
compatibility_name,
)
return compatibility_name
logging.warning(
"Modem preset %s is not recognized by the installed Meshtastic protobuf schema",
modem_preset_enum,
)
return f"UNKNOWN_PRESET_{modem_preset_enum}"
def _get_channel_name(device_channel, node):
if device_channel.settings.name:
return device_channel.settings.name
lora_config = node.localConfig.lora
modem_preset_enum = lora_config.modem_preset
modem_preset_string = config_pb2._CONFIG_LORACONFIG_MODEMPRESET.values_by_number[modem_preset_enum].name
modem_preset_enum = node.localConfig.lora.modem_preset
modem_preset_string = _get_modem_preset_name(modem_preset_enum)
return convert_to_camel_case(modem_preset_string)
+1 -1
View File
@@ -1,6 +1,6 @@
[project]
name = "contact"
version = "1.6.6"
version = "1.6.7"
description = "This Python curses client for Meshtastic is a terminal-based client designed to manage device settings, enable mesh chat communication, and handle configuration backups and restores."
authors = [
{name = "Ben Lipsey",email = "ben@pdxlocations.com"}
+19
View File
@@ -64,6 +64,25 @@ class ContactUiTests(unittest.TestCase):
)
settings_menu.assert_not_called()
def test_handle_backtick_opens_local_settings_for_selected_local_node(self) -> None:
stdscr = mock.Mock()
ui_state.current_window = 2
ui_state.node_list = [123]
ui_state.selected_node = 0
contact_ui.interface_state.myNodeNum = 123
contact_ui.interface_state.interface = mock.Mock()
with mock.patch.object(contact_ui.curses, "curs_set"):
with mock.patch.object(contact_ui, "get_list_input") as chooser:
with mock.patch.object(contact_ui, "settings_menu") as settings_menu:
with mock.patch.object(contact_ui, "get_channels"):
with mock.patch.object(contact_ui, "refresh_node_list"):
with mock.patch.object(contact_ui, "handle_resize"):
contact_ui.handle_backtick(stdscr)
chooser.assert_not_called()
settings_menu.assert_called_once_with(stdscr, contact_ui.interface_state.interface)
def test_show_remote_admin_wait_draws_status(self) -> None:
stdscr = mock.Mock()
stdscr.getmaxyx.return_value = (24, 100)
+5 -2
View File
@@ -240,7 +240,7 @@ class MainRuntimeTests(unittest.TestCase):
with mock.patch.object(entrypoint.sys, "argv", ["contact"]):
with mock.patch.object(entrypoint.curses, "wrapper", side_effect=RuntimeError("boom")):
with mock.patch.object(entrypoint.curses, "endwin") as endwin:
with mock.patch.object(entrypoint.traceback, "print_exc") as print_exc:
with mock.patch.object(entrypoint.traceback, "print_exception") as print_exception:
with mock.patch("builtins.print") as print_mock:
with mock.patch.object(entrypoint.sys, "exit", side_effect=SystemExit(1)) as exit_mock:
with self.assertRaises(SystemExit) as raised:
@@ -248,6 +248,9 @@ class MainRuntimeTests(unittest.TestCase):
self.assertEqual(raised.exception.code, 1)
endwin.assert_called_once_with()
print_exc.assert_called_once_with()
exception_type, exception, exception_traceback = print_exception.call_args_list[-1].args
self.assertIs(exception_type, RuntimeError)
self.assertEqual(str(exception), "boom")
self.assertIsNotNone(exception_traceback)
print_mock.assert_any_call("Fatal error:", mock.ANY)
exit_mock.assert_called_once_with(1)
+19 -1
View File
@@ -1,12 +1,30 @@
from types import SimpleNamespace
import unittest
from meshtastic.protobuf import config_pb2, module_config_pb2
from meshtastic.protobuf import channel_pb2, config_pb2, module_config_pb2
from contact.ui.menus import generate_menu_from_protobuf
class MenusTests(unittest.TestCase):
def test_channels_include_enabled_state_and_display_names(self):
from contact.ui.control_ui import get_settings_option_label
channels = [
channel_pb2.Channel(role=channel_pb2.Channel.PRIMARY, settings=channel_pb2.ChannelSettings(name="Local")),
channel_pb2.Channel(role=channel_pb2.Channel.DISABLED, settings=channel_pb2.ChannelSettings(name="Local")),
channel_pb2.Channel(role=channel_pb2.Channel.DISABLED),
]
node = SimpleNamespace(
localConfig=config_pb2.Config(), moduleConfig=module_config_pb2.ModuleConfig(),
getChannelByChannelIndex=lambda index: channels[index] if index < len(channels) else None,
)
menu = generate_menu_from_protobuf(None, node=node)["Main Menu"]["Channels"]
self.assertEqual(menu["Channel 1"]["enabled"], (None, True))
self.assertEqual(menu["Channel 2"]["enabled"], (None, False))
labels = [get_settings_option_label(key, value, ["Main Menu", "Channels"]) for key, value in menu.items()]
self.assertEqual(labels, ["Local", "Local", "Channel 3"])
def test_main_menu_includes_factory_reset_config_after_factory_reset(self) -> None:
local_node = SimpleNamespace(
localConfig=config_pb2.Config(),
+95
View File
@@ -0,0 +1,95 @@
import unittest
from contextlib import ExitStack
from unittest import mock
from contact.ui import contact_ui
from contact.utilities.singleton import ui_state
from tests.test_support import reset_singletons
class MessageInputTests(unittest.TestCase):
def setUp(self):
reset_singletons()
ui_state.current_window = 1
for name in ("refresh_message_highlight", "refresh_main_window", "get_color"):
patcher = mock.patch.object(contact_ui, name, return_value=1)
patcher.start()
self.addCleanup(patcher.stop)
self.addCleanup(reset_singletons)
def test_keyboard_loop_sends_edited_draft_after_resize(self):
win = mock.Mock()
win.get_wch.side_effect = [
"a", "b", "c", contact_ui.curses.KEY_LEFT,
contact_ui.curses.KEY_DC, "d", contact_ui.curses.KEY_BACKSPACE,
"e", contact_ui.curses.KEY_RESIZE, "\n", "\x1b",
]
with ExitStack() as stack:
stack.enter_context(mock.patch.object(contact_ui, "entry_win", win))
for name in ("get_channels", "handle_resize", "process_pending_ui_updates", "draw_message_input"):
stack.enter_context(mock.patch.object(contact_ui, name))
stack.enter_context(mock.patch.object(contact_ui, "interface_is_connected", return_value=True))
stack.enter_context(mock.patch.object(contact_ui, "drain_resize_events", return_value=None))
send = stack.enter_context(mock.patch.object(contact_ui, "handle_enter", return_value=""))
contact_ui.main_ui(mock.Mock())
send.assert_called_once_with("abe")
self.assertEqual(contact_ui.input_text, "")
self.assertEqual(ui_state.input_cursor, 0)
def test_typing_focuses_input_and_edits_at_cursor(self):
draft = contact_ui.insert_input_text("", "helo")
self.assertTrue(ui_state.input_focused)
with mock.patch.object(contact_ui, "input_text", draft):
contact_ui.handle_leftright(contact_ui.curses.KEY_LEFT)
draft = contact_ui.insert_input_text(draft, "l")
self.assertEqual(draft, "hello")
self.assertEqual(ui_state.input_cursor, 4)
draft = contact_ui.handle_backspace(mock.Mock(), draft)
self.assertEqual(draft, "helo")
self.assertEqual(ui_state.input_cursor, 3)
def test_up_and_down_return_to_messages_without_scrolling(self):
for handler in (contact_ui.handle_up, contact_ui.handle_down):
ui_state.input_focused = True
ui_state.input_cursor = 2
with mock.patch.object(contact_ui, "scroll_messages") as scroll:
handler()
self.assertFalse(ui_state.input_focused)
self.assertEqual(ui_state.current_window, 1)
self.assertEqual(ui_state.input_cursor, 2)
scroll.assert_not_called()
def test_cursor_stops_at_draft_boundaries(self):
ui_state.input_focused = True
with mock.patch.object(contact_ui, "input_text", "abc"):
contact_ui.handle_leftright(contact_ui.curses.KEY_LEFT)
self.assertEqual(ui_state.input_cursor, 0)
self.assertEqual(contact_ui.handle_backspace(mock.Mock(), "abc"), "abc")
contact_ui.handle_end()
contact_ui.handle_leftright(contact_ui.curses.KEY_RIGHT)
self.assertEqual(ui_state.input_cursor, 3)
contact_ui.handle_home()
self.assertEqual(ui_state.input_cursor, 0)
def test_redraw_clears_line_and_shows_text_after_cursor(self):
win = mock.Mock()
win.getmaxyx.return_value = (3, 30)
ui_state.input_focused = True
ui_state.input_cursor = 2
with mock.patch.object(contact_ui, "entry_win", win), mock.patch.object(contact_ui.curses, "curs_set"):
contact_ui.draw_message_input("hello")
win.addstr.assert_any_call(1, 1, " " * 28, 1)
win.addstr.assert_any_call(1, 1, "Message: hello", 1)
win.move.assert_called_once_with(1, 12)
contact_ui.get_color.assert_any_call("window_frame_selected")
def test_long_draft_scrolls_to_cursor_with_reply_context(self):
win = mock.Mock()
win.getmaxyx.return_value = (3, 24)
ui_state.reply_context = "Reply to someone: "
ui_state.input_cursor = 20
with mock.patch.object(contact_ui, "entry_win", win), mock.patch.object(contact_ui.curses, "curs_set"):
contact_ui.draw_message_input("x" * 20)
self.assertLess(win.move.call_args.args[1], 23)
rendered = [call.args[2] for call in win.addstr.call_args_list if call.args[:2] == (1, 1)][-1]
self.assertEqual(rendered, "Message: " + "x" * 12)
+31
View File
@@ -1,11 +1,28 @@
from types import SimpleNamespace
import unittest
from unittest import mock
from meshtastic.protobuf import channel_pb2
from contact.utilities.save_to_radio import save_changes
class SaveToRadioTests(unittest.TestCase):
def test_channel_enable_disable_and_unrelated_edits_preserve_role(self):
for index in (0, 1):
interface, node = self.build_interface()
node.channels = [channel_pb2.Channel(index=i) for i in range(2)]
channel = node.channels[index]
path = SimpleNamespace(menu_path=["Main Menu", "Channels", f"Channel {index + 1}"])
save_changes(interface, {"name": "Test"}, path)
self.assertEqual(channel.role, channel_pb2.Channel.DISABLED)
save_changes(interface, {"enabled": True}, path)
expected = channel_pb2.Channel.PRIMARY if index == 0 else channel_pb2.Channel.SECONDARY
self.assertEqual(channel.role, expected)
save_changes(interface, {"enabled": False}, path)
self.assertEqual(channel.role, channel_pb2.Channel.DISABLED)
self.assertEqual(channel.settings.name, "Test")
self.assertEqual(node.writeChannel.call_args_list, [mock.call(index)] * 3)
def build_interface(self):
node = mock.Mock()
node.localConfig = SimpleNamespace(
@@ -55,6 +72,20 @@ class SaveToRadioTests(unittest.TestCase):
self.assertTrue(reconnect_required)
def test_save_changes_replaces_remote_admin_keys_in_one_write(self) -> None:
interface, node = self.build_interface()
key_one = b"1" * 32
key_two = b"2" * 32
key_three = b"3" * 32
node.localConfig.security.admin_key = [key_one, key_two, key_three]
menu_state = SimpleNamespace(menu_path=["Main Menu", "Radio Settings", "Security"])
reconnect_required = save_changes(interface, {"admin_key": [key_one, key_two]}, menu_state, node=node)
self.assertTrue(reconnect_required)
self.assertEqual(node.localConfig.security.admin_key, [key_one, key_two])
node.writeConfig.assert_called_once_with("security")
def test_save_changes_returns_true_only_for_rebooting_device_fields(self) -> None:
interface, node = self.build_interface()
menu_state = SimpleNamespace(menu_path=["Main Menu", "Radio Settings", "Device"])
+50 -1
View File
@@ -1,10 +1,19 @@
import unittest
from unittest import mock
from meshtastic.protobuf import config_pb2
import contact.ui.default_config as config
from contact.utilities.demo_data import DEMO_LOCAL_NODE_NUM, build_demo_interface
from contact.utilities.singleton import interface_state, ui_state
from contact.utilities.utils import add_new_message, get_channels, get_node_list, parse_protobuf
from contact.utilities.utils import (
_get_channel_name,
_get_modem_preset_name,
add_new_message,
get_channels,
get_node_list,
parse_protobuf,
)
from tests.test_support import reset_singletons, restore_config, snapshot_config
@@ -12,6 +21,7 @@ from tests.test_support import reset_singletons, restore_config, snapshot_config
class UtilsTests(unittest.TestCase):
def setUp(self) -> None:
reset_singletons()
_get_modem_preset_name.cache_clear()
self.saved_config = snapshot_config("node_sort")
def tearDown(self) -> None:
@@ -60,6 +70,45 @@ class UtilsTests(unittest.TestCase):
self.assertIn("MediumFast", ui_state.all_messages)
self.assertIn("Another Channel", ui_state.all_messages)
def test_get_channel_name_handles_newer_known_modem_preset(self) -> None:
device_channel = mock.Mock()
device_channel.settings.name = ""
node = mock.Mock()
node.localConfig.lora.modem_preset = 14
with mock.patch("contact.utilities.utils.logging.warning") as warning:
self.assertEqual(_get_channel_name(device_channel, node), "TinyFast")
warning.assert_called_once()
def test_get_channel_name_handles_unknown_modem_preset(self) -> None:
device_channel = mock.Mock()
device_channel.settings.name = ""
node = mock.Mock()
node.localConfig.lora.modem_preset = 99
with mock.patch("contact.utilities.utils.logging.warning") as warning:
self.assertEqual(_get_channel_name(device_channel, node), "UnknownPreset99")
warning.assert_called_once()
def test_get_channel_name_prefers_public_protobuf_enum_over_compatibility_map(self) -> None:
device_channel = mock.Mock()
device_channel.settings.name = ""
node = mock.Mock()
node.localConfig.lora.modem_preset = 14
with mock.patch.object(
config_pb2.Config.LoRaConfig.ModemPreset,
"Name",
return_value="TINY_FAST",
) as enum_name:
with mock.patch("contact.utilities.utils.logging.warning") as warning:
self.assertEqual(_get_channel_name(device_channel, node), "TinyFast")
enum_name.assert_called_once_with(14)
warning.assert_not_called()
def test_get_channels_rebuilds_renamed_channels_and_preserves_messages(self) -> None:
interface = build_demo_interface()
interface.localNode.channels[0].settings.name = "Renamed Channel"