mirror of
https://github.com/pdxlocations/contact.git
synced 2026-11-02 09:00:32 +01:00
Compare commits
10
Commits
1.6.6
..
flicker-fix
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
a89d4f8c0d | ||
|
|
808fee4fe0 | ||
|
|
94d0a9053e | ||
|
|
f38167db1a | ||
|
|
57caf463bc | ||
|
|
36408a1f37 | ||
|
|
07b9ed22d8 | ||
|
|
bdba69677b | ||
|
|
1f88a37ac1 | ||
|
|
8f33cbc22d |
No files matched your search
+5
-3
@@ -121,7 +121,7 @@ def interface_is_ready(interface: object) -> bool:
|
||||
|
||||
def initialize_runtime_interface_with_retry(stdscr: curses.window, args: object):
|
||||
while True:
|
||||
interface = initialize_runtime_interface(args)
|
||||
interface = initialize_runtime_interface(args, stdscr)
|
||||
if getattr(args, "demo_screenshot", False) or interface_is_ready(interface):
|
||||
return interface
|
||||
|
||||
@@ -162,10 +162,12 @@ def initialize_globals(seed_demo: bool = False) -> None:
|
||||
load_messages_from_db()
|
||||
|
||||
|
||||
def initialize_runtime_interface(args: object):
|
||||
def initialize_runtime_interface(args: object, stdscr=None):
|
||||
if getattr(args, "demo_screenshot", False):
|
||||
configure_demo_database()
|
||||
return build_demo_interface()
|
||||
if stdscr is not None:
|
||||
return initialize_interface(args, status_callback=lambda status: draw_splash(stdscr, status))
|
||||
return initialize_interface(args)
|
||||
|
||||
|
||||
@@ -258,7 +260,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
-1
@@ -36,7 +36,7 @@ def main(stdscr: curses.window) -> None:
|
||||
|
||||
parser = setup_parser()
|
||||
args = parser.parse_args()
|
||||
interface = initialize_interface(args)
|
||||
interface = initialize_interface(args, status_callback=lambda status: draw_splash(stdscr, status))
|
||||
|
||||
if interface.localNode.localConfig.lora.region == 0:
|
||||
confirmation = get_list_input(
|
||||
|
||||
+214
-97
@@ -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:
|
||||
@@ -242,11 +246,14 @@ def compute_widths(total_w: int, focus: int):
|
||||
return MIN_COL, MIN_COL, total_w - 2 * MIN_COL
|
||||
|
||||
|
||||
def paint_frame(win, selected: bool) -> None:
|
||||
def paint_frame(win, selected: bool, refresh: bool = True) -> 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"))
|
||||
win.refresh()
|
||||
if refresh:
|
||||
win.refresh()
|
||||
|
||||
|
||||
def get_channel_row_color(index: int) -> int:
|
||||
@@ -269,7 +276,7 @@ def get_node_row_color(index: int, highlight: bool = False) -> int:
|
||||
return get_color(color, reverse=reverse)
|
||||
|
||||
|
||||
def refresh_node_selection(old_index: int = -1, highlight: bool = False) -> None:
|
||||
def refresh_node_selection(old_index: int = -1, highlight: bool = False, refresh: bool = True) -> None:
|
||||
if nodes_pad is None or not ui_state.node_list:
|
||||
return
|
||||
|
||||
@@ -288,8 +295,9 @@ def refresh_node_selection(old_index: int = -1, highlight: bool = False) -> None
|
||||
pass
|
||||
|
||||
ui_state.start_index[2] = max(0, ui_state.selected_node - (nodes_win.getmaxyx()[0] - 3))
|
||||
refresh_pad(2)
|
||||
draw_window_arrows(2)
|
||||
if refresh:
|
||||
refresh_pad(2)
|
||||
draw_window_arrows(2)
|
||||
|
||||
|
||||
def refresh_main_window(window_id: int, selected: bool) -> None:
|
||||
@@ -300,7 +308,7 @@ def refresh_main_window(window_id: int, selected: bool) -> None:
|
||||
channel_pad.chgat(ui_state.selected_channel, 1, width, get_channel_row_color(ui_state.selected_channel))
|
||||
refresh_pad(0)
|
||||
elif window_id == 1:
|
||||
paint_frame(messages_win, selected=selected)
|
||||
paint_frame(messages_win, selected=selected, refresh=False)
|
||||
refresh_pad(1)
|
||||
elif window_id == 2:
|
||||
paint_frame(nodes_win, selected=selected)
|
||||
@@ -491,6 +499,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 +565,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 +605,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 +613,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 +634,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 +659,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, refresh=False)
|
||||
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 +729,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 +742,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 +759,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 +802,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 +825,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:
|
||||
@@ -858,6 +937,28 @@ def handle_enter(input_text: str) -> str:
|
||||
return input_text
|
||||
|
||||
|
||||
def stage_node_details_background(stdscr, previous_window):
|
||||
"""Restore the panes behind a moved dialog without displaying an intermediate frame."""
|
||||
current_window = ui_state.current_window
|
||||
try:
|
||||
ui_state.current_window = previous_window
|
||||
stdscr.touchwin()
|
||||
stdscr.noutrefresh()
|
||||
windows = (channel_win, messages_win, nodes_win)
|
||||
for index, win in enumerate(windows):
|
||||
if ui_state.single_pane_mode and index != previous_window:
|
||||
continue
|
||||
win.touchwin()
|
||||
refresh_pad(index, commit=False)
|
||||
if ui_state.display_log and (not ui_state.single_pane_mode or previous_window == 1):
|
||||
packetlog_win.touchwin()
|
||||
packetlog_win.noutrefresh()
|
||||
entry_win.touchwin()
|
||||
entry_win.noutrefresh()
|
||||
finally:
|
||||
ui_state.current_window = current_window
|
||||
|
||||
|
||||
def handle_f5_key(stdscr: curses.window) -> None:
|
||||
if not ui_state.node_list:
|
||||
return
|
||||
@@ -955,6 +1056,7 @@ def handle_f5_key(stdscr: curses.window) -> None:
|
||||
ui_state.current_window = 4
|
||||
scroll_offset = 0
|
||||
dialog_win = None
|
||||
previous_layout = None
|
||||
|
||||
curses.curs_set(0)
|
||||
refresh_node_selection(highlight=True)
|
||||
@@ -974,11 +1076,15 @@ def handle_f5_key(stdscr: curses.window) -> None:
|
||||
max_scroll = max(0, len(message_lines) - viewport_h)
|
||||
scroll_offset = max(0, min(scroll_offset, max_scroll))
|
||||
|
||||
layout = (dialog_height, dialog_width, y, x, ui_state.selected_node, height, width)
|
||||
if previous_layout is not None and layout != previous_layout:
|
||||
stage_node_details_background(stdscr, previous_window)
|
||||
previous_layout = layout
|
||||
|
||||
if dialog_win is None:
|
||||
dialog_win = curses.newwin(dialog_height, dialog_width, y, x)
|
||||
else:
|
||||
dialog_win.erase()
|
||||
dialog_win.refresh()
|
||||
dialog_win.resize(dialog_height, dialog_width)
|
||||
dialog_win.mvwin(y, x)
|
||||
|
||||
@@ -1023,7 +1129,7 @@ def handle_f5_key(stdscr: curses.window) -> None:
|
||||
except curses.error:
|
||||
pass
|
||||
|
||||
dialog_win.refresh()
|
||||
dialog_win.noutrefresh()
|
||||
msg_win.noutrefresh()
|
||||
curses.doupdate()
|
||||
|
||||
@@ -1043,12 +1149,12 @@ def handle_f5_key(stdscr: curses.window) -> None:
|
||||
old_selected_node = ui_state.selected_node
|
||||
ui_state.selected_node = (ui_state.selected_node - 1) % len(ui_state.node_list)
|
||||
scroll_offset = 0
|
||||
refresh_node_selection(old_selected_node, highlight=True)
|
||||
refresh_node_selection(old_selected_node, highlight=True, refresh=False)
|
||||
elif char == curses.KEY_DOWN:
|
||||
old_selected_node = ui_state.selected_node
|
||||
ui_state.selected_node = (ui_state.selected_node + 1) % len(ui_state.node_list)
|
||||
scroll_offset = 0
|
||||
refresh_node_selection(old_selected_node, highlight=True)
|
||||
refresh_node_selection(old_selected_node, highlight=True, refresh=False)
|
||||
elif char == curses.KEY_PPAGE:
|
||||
scroll_offset = max(0, scroll_offset - viewport_h)
|
||||
elif char == curses.KEY_NPAGE:
|
||||
@@ -1107,14 +1213,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 +1238,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)
|
||||
@@ -1558,7 +1671,7 @@ def draw_messages_window(scroll_to_bottom: bool = False, preserve_selection: boo
|
||||
messages_pad.addstr(row, 1, line, color)
|
||||
ui_state.message_line_ranges[channel] = message_ranges
|
||||
|
||||
paint_frame(messages_win, selected=(ui_state.current_window == 1))
|
||||
paint_frame(messages_win, selected=(ui_state.current_window == 1), refresh=False)
|
||||
|
||||
visible_lines = get_msg_window_lines(messages_win, packetlog_win)
|
||||
|
||||
@@ -1573,7 +1686,6 @@ def draw_messages_window(scroll_to_bottom: bool = False, preserve_selection: boo
|
||||
|
||||
refresh_message_highlight()
|
||||
|
||||
messages_win.refresh()
|
||||
refresh_pad(1)
|
||||
draw_packetlog_win()
|
||||
draw_window_arrows(1)
|
||||
@@ -1587,7 +1699,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):
|
||||
@@ -1903,7 +2015,7 @@ def search(win: int) -> None:
|
||||
entry_win.erase()
|
||||
|
||||
|
||||
def refresh_pad(window: int) -> None:
|
||||
def refresh_pad(window: int, commit: bool = True) -> None:
|
||||
|
||||
# If in single-pane mode and this isn't the focused window, skip refreshing its (collapsed) pad
|
||||
if ui_state.single_pane_mode and window != ui_state.current_window:
|
||||
@@ -1920,7 +2032,7 @@ def refresh_pad(window: int) -> None:
|
||||
|
||||
if ui_state.display_log:
|
||||
packetlog_win.box()
|
||||
packetlog_win.refresh()
|
||||
packetlog_win.noutrefresh()
|
||||
|
||||
elif window == 2:
|
||||
pad = nodes_pad
|
||||
@@ -1963,9 +2075,12 @@ def refresh_pad(window: int) -> None:
|
||||
return
|
||||
|
||||
draw_frame_title(box, get_window_title(window))
|
||||
box.refresh()
|
||||
box.noutrefresh()
|
||||
|
||||
pad.refresh(
|
||||
# The frame overlaps the pad's viewport. Recompose even unchanged pad
|
||||
# rows before committing, so its text cannot flash blank between draws.
|
||||
pad.touchwin()
|
||||
pad.noutrefresh(
|
||||
start_index,
|
||||
0,
|
||||
top,
|
||||
@@ -1973,6 +2088,8 @@ def refresh_pad(window: int) -> None:
|
||||
bottom,
|
||||
right,
|
||||
)
|
||||
if commit:
|
||||
curses.doupdate()
|
||||
|
||||
|
||||
def add_notification(channel_number: int) -> None:
|
||||
|
||||
+10
-5
@@ -1,8 +1,9 @@
|
||||
import curses
|
||||
import textwrap
|
||||
from contact.ui.colors import get_color
|
||||
|
||||
|
||||
def draw_splash(stdscr: object) -> None:
|
||||
def draw_splash(stdscr: object, status: str = "connecting...") -> None:
|
||||
"""Draw the splash screen with a logo and connecting message."""
|
||||
curses.curs_set(0)
|
||||
|
||||
@@ -13,16 +14,20 @@ def draw_splash(stdscr: object) -> None:
|
||||
message_1 = "/ Λ"
|
||||
message_2 = "/ / \\"
|
||||
message_3 = "P W R D"
|
||||
message_4 = "connecting..."
|
||||
|
||||
start_x = width // 2 - len(message_1) // 2
|
||||
start_x2 = width // 2 - len(message_4) // 2
|
||||
start_y = height // 2 - 1
|
||||
stdscr.addstr(start_y, start_x, message_1, get_color("splash_logo", bold=True))
|
||||
stdscr.addstr(start_y + 1, start_x - 1, message_2, get_color("splash_logo", bold=True))
|
||||
stdscr.addstr(start_y + 2, start_x - 2, message_3, get_color("splash_logo", bold=True))
|
||||
stdscr.addstr(start_y + 4, start_x2, message_4, get_color("splash_text"))
|
||||
stdscr.move(start_y + 5, start_x2)
|
||||
status_lines = []
|
||||
for line in status.splitlines():
|
||||
status_lines.extend(textwrap.wrap(line, max(1, width - 4)) or [""])
|
||||
for offset, line in enumerate(status_lines):
|
||||
row = start_y + 4 + offset
|
||||
if row >= height - 1:
|
||||
break
|
||||
stdscr.addstr(row, max(1, (width - len(line)) // 2), line, get_color("splash_text"))
|
||||
|
||||
stdscr.attrset(get_color("window_frame"))
|
||||
stdscr.box()
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -1,4 +1,6 @@
|
||||
import logging
|
||||
import contextlib
|
||||
import io
|
||||
import time
|
||||
import meshtastic.serial_interface, meshtastic.tcp_interface, meshtastic.ble_interface
|
||||
|
||||
@@ -25,7 +27,7 @@ def interface_is_connected(interface) -> bool:
|
||||
return getattr(interface, "localNode", None) is not None
|
||||
|
||||
|
||||
def initialize_interface(args):
|
||||
def initialize_interface(args, status_callback=None):
|
||||
try:
|
||||
|
||||
if args.ble:
|
||||
@@ -43,7 +45,10 @@ def initialize_interface(args):
|
||||
logging.error(f"Error connecting to {args.host}. {ex}")
|
||||
else:
|
||||
try:
|
||||
client = meshtastic.serial_interface.SerialInterface(args.port)
|
||||
# The library prints its fallback notice directly to stdout,
|
||||
# which would overwrite the curses splash screen.
|
||||
with contextlib.redirect_stdout(io.StringIO()):
|
||||
client = meshtastic.serial_interface.SerialInterface(args.port)
|
||||
except FileNotFoundError as ex:
|
||||
logging.error(f"The serial device at '{args.port}' was not found. {ex}")
|
||||
except PermissionError as ex:
|
||||
@@ -56,6 +61,8 @@ def initialize_interface(args):
|
||||
logging.error(f"The serial device couldn't be opened, it might be in use by another process. {ex}")
|
||||
if client.devPath is None:
|
||||
try:
|
||||
if status_callback:
|
||||
status_callback("No serial device found.\nConnecting via TCP to localhost…")
|
||||
client = meshtastic.tcp_interface.TCPInterface("localhost")
|
||||
except Exception as ex:
|
||||
logging.error(f"Error connecting to localhost:{ex}")
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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
@@ -1,6 +1,6 @@
|
||||
[project]
|
||||
name = "contact"
|
||||
version = "1.6.6"
|
||||
version = "1.6.9"
|
||||
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"}
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
import unittest
|
||||
from contextlib import ExitStack
|
||||
from types import SimpleNamespace
|
||||
from unittest import mock
|
||||
|
||||
@@ -11,6 +12,52 @@ from tests.test_support import reset_singletons, restore_config, snapshot_config
|
||||
|
||||
|
||||
class ContactUiTests(unittest.TestCase):
|
||||
def test_f5_restores_background_when_switching_between_different_size_nodes(self):
|
||||
ui_state.node_list = [101, 102]
|
||||
ui_state.current_window = 2
|
||||
interface = SimpleNamespace(nodesByNum={
|
||||
101: {"user": {"shortName": "BIG", "longName": "A" * 60}, "position": {"altitude": 100}},
|
||||
102: {"user": {"shortName": "SMALL", "longName": "B"}},
|
||||
})
|
||||
win = mock.Mock()
|
||||
win.getch.side_effect = [contact_ui.curses.KEY_DOWN, contact_ui.curses.KEY_UP, 27]
|
||||
events = mock.Mock()
|
||||
events.attach_mock(win.noutrefresh, "dialog")
|
||||
stdscr = mock.Mock()
|
||||
with ExitStack() as stack:
|
||||
stack.enter_context(mock.patch.object(contact_ui.interface_state, "interface", interface))
|
||||
for name, value in (("LINES", 40), ("COLS", 100)):
|
||||
stack.enter_context(mock.patch.object(contact_ui.curses, name, value, create=True))
|
||||
for name in ("curs_set", "update_lines_cols", "doupdate"):
|
||||
stack.enter_context(mock.patch.object(contact_ui.curses, name))
|
||||
stack.enter_context(mock.patch.object(contact_ui.curses, "newwin", return_value=win))
|
||||
for name in ("get_color", "refresh_node_selection", "handle_resize"):
|
||||
stack.enter_context(mock.patch.object(contact_ui, name, return_value=0))
|
||||
restore = stack.enter_context(mock.patch.object(contact_ui, "stage_node_details_background"))
|
||||
events.attach_mock(restore, "restore")
|
||||
contact_ui.handle_f5_key(stdscr)
|
||||
self.assertEqual(events.mock_calls, [
|
||||
mock.call.dialog(), mock.call.restore(stdscr, 2), mock.call.dialog(),
|
||||
mock.call.restore(stdscr, 2), mock.call.dialog(),
|
||||
])
|
||||
self.assertNotEqual(win.resize.call_args_list[0], win.resize.call_args_list[1])
|
||||
self.assertEqual(win.refresh.call_count, 1) # Only on close.
|
||||
|
||||
def test_background_restore_stages_panes_without_flushing(self):
|
||||
ui_state.current_window = 4
|
||||
ui_state.single_pane_mode = True
|
||||
stdscr = mock.Mock()
|
||||
with ExitStack() as stack:
|
||||
for name in ("channel_win", "messages_win", "nodes_win", "entry_win"):
|
||||
stack.enter_context(mock.patch.object(contact_ui, name, mock.Mock(), create=True))
|
||||
refresh = stack.enter_context(mock.patch.object(contact_ui, "refresh_pad"))
|
||||
update = stack.enter_context(mock.patch.object(contact_ui.curses, "doupdate"))
|
||||
contact_ui.stage_node_details_background(stdscr, 2)
|
||||
refresh.assert_called_once_with(2, commit=False)
|
||||
update.assert_not_called()
|
||||
stdscr.noutrefresh.assert_called_once()
|
||||
self.assertEqual(ui_state.current_window, 4)
|
||||
|
||||
def setUp(self) -> None:
|
||||
reset_singletons()
|
||||
self.saved_config = snapshot_config("single_pane_mode")
|
||||
@@ -64,6 +111,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)
|
||||
@@ -418,6 +484,8 @@ class ContactUiTests(unittest.TestCase):
|
||||
self.assertEqual(contact_ui.get_window_title(1), "Primary")
|
||||
|
||||
def test_refresh_pad_draws_selected_channel_title_on_message_frame(self) -> None:
|
||||
self.addCleanup(mock.patch.stopall)
|
||||
mock.patch.object(contact_ui, "packetlog_win", mock.Mock(), create=True).start()
|
||||
ui_state.single_pane_mode = True
|
||||
ui_state.current_window = 1
|
||||
ui_state.channel_list = ["Primary"]
|
||||
@@ -434,9 +502,16 @@ class ContactUiTests(unittest.TestCase):
|
||||
contact_ui.messages_win.getmaxyx.return_value = (10, 20)
|
||||
|
||||
with mock.patch.object(contact_ui, "get_msg_window_lines", return_value=4):
|
||||
contact_ui.refresh_pad(1)
|
||||
with mock.patch.object(contact_ui.curses, "doupdate") as update:
|
||||
contact_ui.refresh_pad(1)
|
||||
|
||||
contact_ui.messages_win.addstr.assert_called_once_with(0, 2, " Primary ", contact_ui.curses.A_BOLD)
|
||||
contact_ui.messages_win.refresh.assert_not_called()
|
||||
contact_ui.messages_pad.refresh.assert_not_called()
|
||||
contact_ui.messages_win.noutrefresh.assert_called_once()
|
||||
contact_ui.messages_pad.touchwin.assert_called_once()
|
||||
contact_ui.messages_pad.noutrefresh.assert_called_once()
|
||||
update.assert_called_once()
|
||||
|
||||
def test_search_ignores_no_input_from_curses(self) -> None:
|
||||
ui_state.node_list = [101]
|
||||
@@ -489,6 +564,9 @@ class ContactUiTests(unittest.TestCase):
|
||||
contact_ui.handle_f5_key(stdscr)
|
||||
|
||||
self.assertEqual(dialog_win.getch.call_count, 2)
|
||||
# Idle polling must never present an erased dialog before its text.
|
||||
self.assertEqual(dialog_win.noutrefresh.call_count, 2)
|
||||
self.assertEqual(dialog_win.refresh.call_count, 1) # Closing the dialog only.
|
||||
handle_resize.assert_called_once_with(stdscr, False)
|
||||
|
||||
def test_f5_node_details_tolerates_none_metrics(self) -> None:
|
||||
|
||||
+5
-2
@@ -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)
|
||||
@@ -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)
|
||||
@@ -55,6 +55,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
@@ -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"
|
||||
|
||||
Reference in new issue
Block a user