Compare commits

...

11 Commits

Author SHA1 Message Date
pdxlocations
b53dab1840 Add factory reset config option and tryfix factory reset 2026-03-21 21:28:15 -07:00
pdxlocations
f2904eb550 Merge pull request #259 from pdxlocations:fix-shutdown
Fix interface shutdown handling
2026-03-21 21:14:38 -07:00
pdxlocations
480c32ba56 try fix shutdown 2026-03-21 21:13:53 -07:00
pdxlocations
b4b084b627 bump version to 1.5.4 in pyproject.toml 2026-03-19 15:47:51 -07:00
pdxlocations
5940c9b02b fix content margins 2026-03-19 15:19:24 -07:00
pdxlocations
c492c96685 bump version to 1.5.3 in pyproject.toml 2026-03-19 14:37:29 -07:00
pdxlocations
90376d35f3 Single pane mode fix 2026-03-19 14:37:14 -07:00
pdxlocations
4b35a74e2c bump version to 1.5.2 in pyproject.toml 2026-03-19 14:07:16 -07:00
pdxlocations
ecc5308dad Refactor UI redraw handling and improve message drawing logic 2026-03-19 14:06:01 -07:00
pdxlocations
8f376edabe bump version to 1.5.1 in pyproject.toml 2026-03-19 11:41:08 -07:00
pdxlocations
e5ef87ec19 Merge pull request #258 from pdxlocations:reconnect
Reconnect after config changes
2026-03-19 11:40:41 -07:00
18 changed files with 721 additions and 170 deletions

View File

@@ -12,6 +12,7 @@ Reboot, "Reboot", ""
Reset Node DB, "Reset Node DB", ""
Shutdown, "Shutdown", ""
Factory Reset, "Factory Reset", ""
factory_reset_config, "Factory Reset Config", ""
Exit, "Exit", ""
Yes, "Yes", ""
No, "No", ""
@@ -55,6 +56,7 @@ confirm.reboot, "Are you sure you want to Reboot?", ""
confirm.reset_node_db, "Are you sure you want to Reset Node DB?", ""
confirm.shutdown, "Are you sure you want to Shutdown?", ""
confirm.factory_reset, "Are you sure you want to Factory Reset?", ""
confirm.factory_reset_config, "Are you sure you want to Factory Reset Config?", ""
confirm.save_before_exit_section, "You have unsaved changes in {section}. Save before exiting?", ""
prompt.select_region, "Select your region:", ""
dialog.slow_down_title, "Slow down", ""

View File

@@ -12,6 +12,7 @@ Reboot, "Перезагрузить", ""
Reset Node DB, "Сбросить БД узлов", ""
Shutdown, "Выключить", ""
Factory Reset, "Сброс до заводских", ""
factory_reset_config, "Сбросить только конфигурацию", ""
Exit, "Выход", ""
Yes, "Да", ""
No, "Нет", ""
@@ -55,6 +56,7 @@ confirm.reboot, "Перезагрузить устройство?", ""
confirm.reset_node_db, "Сбросить БД узлов?", ""
confirm.shutdown, "Выключить устройство?", ""
confirm.factory_reset, "Сбросить до заводских настроек?", ""
confirm.factory_reset_config, "Сбросить только конфигурацию?", ""
confirm.save_before_exit_section, "Есть несохраненные изменения в {section}. Сохранить перед выходом?", ""
prompt.select_region, "Выберите ваш регион:", ""
dialog.slow_down_title, "Подождите", ""

View File

@@ -48,11 +48,8 @@ from contact.utilities.utils import (
add_new_message,
)
from contact.ui.contact_ui import (
draw_packetlog_win,
draw_node_list,
draw_messages_window,
draw_channel_list,
add_notification,
request_ui_redraw,
)
from contact.utilities.db_handler import (
save_message_to_db,
@@ -121,7 +118,7 @@ def on_receive(packet: Dict[str, Any], interface: Any) -> None:
ui_state.packet_buffer = ui_state.packet_buffer[-20:]
if ui_state.display_log:
draw_packetlog_win()
request_ui_redraw(packetlog=True)
if ui_state.current_window == 4:
menu_state.need_redraw = True
@@ -132,7 +129,7 @@ def on_receive(packet: Dict[str, Any], interface: Any) -> None:
# Assume any incoming packet could update the last seen time for a node
changed = refresh_node_list()
if changed:
draw_node_list()
request_ui_redraw(nodes=True)
if packet["decoded"]["portnum"] == "NODEINFO_APP":
if "user" in packet["decoded"] and "longName" in packet["decoded"]["user"]:
@@ -186,9 +183,9 @@ def on_receive(packet: Dict[str, Any], interface: Any) -> None:
add_new_message(channel_id, f"{config.message_prefix} [{hops}] {message_from_string} ", message_string)
if refresh_channels:
draw_channel_list()
request_ui_redraw(channels=True)
if refresh_messages:
draw_messages_window(True)
request_ui_redraw(messages=True, scroll_messages_to_bottom=True)
save_message_to_db(channel_id, message_from_id, message_string)

View File

@@ -15,7 +15,7 @@ from contact.utilities.db_handler import (
)
import contact.ui.default_config as config
from contact.utilities.singleton import ui_state, interface_state
from contact.utilities.singleton import ui_state, interface_state, app_state
from contact.utilities.utils import add_new_message
@@ -28,145 +28,141 @@ def onAckNak(packet: Dict[str, Any]) -> None:
"""
Handles incoming ACK/NAK response packets.
"""
from contact.ui.contact_ui import draw_messages_window
from contact.ui.contact_ui import request_ui_redraw
request = packet["decoded"]["requestId"]
if request not in ack_naks:
return
with app_state.lock:
request = packet["decoded"]["requestId"]
if request not in ack_naks:
return
acknak = ack_naks.pop(request)
message = ui_state.all_messages[acknak["channel"]][acknak["messageIndex"]][1]
acknak = ack_naks.pop(request)
message = ui_state.all_messages[acknak["channel"]][acknak["messageIndex"]][1]
confirm_string = " "
ack_type = None
if packet["decoded"]["routing"]["errorReason"] == "NONE":
if packet["from"] == interface_state.myNodeNum: # Ack "from" ourself means implicit ACK
confirm_string = config.ack_implicit_str
ack_type = "Implicit"
confirm_string = " "
ack_type = None
if packet["decoded"]["routing"]["errorReason"] == "NONE":
if packet["from"] == interface_state.myNodeNum: # Ack "from" ourself means implicit ACK
confirm_string = config.ack_implicit_str
ack_type = "Implicit"
else:
confirm_string = config.ack_str
ack_type = "Ack"
else:
confirm_string = config.ack_str
ack_type = "Ack"
else:
confirm_string = config.nak_str
ack_type = "Nak"
confirm_string = config.nak_str
ack_type = "Nak"
ui_state.all_messages[acknak["channel"]][acknak["messageIndex"]] = (
time.strftime("[%H:%M:%S] ") + config.sent_message_prefix + confirm_string + ": ",
message,
)
ui_state.all_messages[acknak["channel"]][acknak["messageIndex"]] = (
time.strftime("[%H:%M:%S] ") + config.sent_message_prefix + confirm_string + ": ",
message,
)
update_ack_nak(acknak["channel"], acknak["timestamp"], message, ack_type)
update_ack_nak(acknak["channel"], acknak["timestamp"], message, ack_type)
channel_number = ui_state.channel_list.index(acknak["channel"])
if ui_state.channel_list[channel_number] == ui_state.channel_list[ui_state.selected_channel]:
draw_messages_window()
channel_number = ui_state.channel_list.index(acknak["channel"])
if ui_state.channel_list[channel_number] == ui_state.channel_list[ui_state.selected_channel]:
request_ui_redraw(messages=True)
def on_response_traceroute(packet: Dict[str, Any]) -> None:
"""
Handle traceroute response packets and render the route visually in the UI.
"""
from contact.ui.contact_ui import draw_channel_list, draw_messages_window, add_notification
from contact.ui.contact_ui import add_notification, request_ui_redraw
refresh_channels = False
refresh_messages = False
with app_state.lock:
refresh_channels = False
refresh_messages = False
UNK_SNR = -128 # Value representing unknown SNR
UNK_SNR = -128 # Value representing unknown SNR
route_discovery = mesh_pb2.RouteDiscovery()
route_discovery.ParseFromString(packet["decoded"]["payload"])
msg_dict = google.protobuf.json_format.MessageToDict(route_discovery)
route_discovery = mesh_pb2.RouteDiscovery()
route_discovery.ParseFromString(packet["decoded"]["payload"])
msg_dict = google.protobuf.json_format.MessageToDict(route_discovery)
msg_str = "Traceroute to:\n"
msg_str = "Traceroute to:\n"
route_str = (
get_name_from_database(packet["to"], "short") or f"{packet['to']:08x}"
) # Start with destination of response
# SNR list should have one more entry than the route, as the final destination adds its SNR also
lenTowards = 0 if "route" not in msg_dict else len(msg_dict["route"])
snrTowardsValid = "snrTowards" in msg_dict and len(msg_dict["snrTowards"]) == lenTowards + 1
if lenTowards > 0: # Loop through hops in route and add SNR if available
for idx, node_num in enumerate(msg_dict["route"]):
route_str += (
" --> "
+ (get_name_from_database(node_num, "short") or f"{node_num:08x}")
+ " ("
+ (
str(msg_dict["snrTowards"][idx] / 4)
if snrTowardsValid and msg_dict["snrTowards"][idx] != UNK_SNR
else "?"
)
+ "dB)"
)
# End with origin of response
route_str += (
" --> "
+ (get_name_from_database(packet["from"], "short") or f"{packet['from']:08x}")
+ " ("
+ (str(msg_dict["snrTowards"][-1] / 4) if snrTowardsValid and msg_dict["snrTowards"][-1] != UNK_SNR else "?")
+ "dB)"
)
msg_str += route_str + "\n" # Print the route towards destination
# Only if hopStart is set and there is an SNR entry (for the origin) it's valid, even though route might be empty (direct connection)
lenBack = 0 if "routeBack" not in msg_dict else len(msg_dict["routeBack"])
backValid = "hopStart" in packet and "snrBack" in msg_dict and len(msg_dict["snrBack"]) == lenBack + 1
if backValid:
msg_str += "Back:\n"
route_str = (
get_name_from_database(packet["from"], "short") or f"{packet['from']:08x}"
) # Start with origin of response
get_name_from_database(packet["to"], "short") or f"{packet['to']:08x}"
) # Start with destination of response
if lenBack > 0: # Loop through hops in routeBack and add SNR if available
for idx, node_num in enumerate(msg_dict["routeBack"]):
lenTowards = 0 if "route" not in msg_dict else len(msg_dict["route"])
snrTowardsValid = "snrTowards" in msg_dict and len(msg_dict["snrTowards"]) == lenTowards + 1
if lenTowards > 0:
for idx, node_num in enumerate(msg_dict["route"]):
route_str += (
" --> "
+ (get_name_from_database(node_num, "short") or f"{node_num:08x}")
+ " ("
+ (str(msg_dict["snrBack"][idx] / 4) if msg_dict["snrBack"][idx] != UNK_SNR else "?")
+ (
str(msg_dict["snrTowards"][idx] / 4)
if snrTowardsValid and msg_dict["snrTowards"][idx] != UNK_SNR
else "?"
)
+ "dB)"
)
# End with destination of response (us)
route_str += (
" --> "
+ (get_name_from_database(packet["to"], "short") or f"{packet['to']:08x}")
+ (get_name_from_database(packet["from"], "short") or f"{packet['from']:08x}")
+ " ("
+ (str(msg_dict["snrBack"][-1] / 4) if msg_dict["snrBack"][-1] != UNK_SNR else "?")
+ (str(msg_dict["snrTowards"][-1] / 4) if snrTowardsValid and msg_dict["snrTowards"][-1] != UNK_SNR else "?")
+ "dB)"
)
msg_str += route_str + "\n" # Print the route back to us
msg_str += route_str + "\n"
if packet["from"] not in ui_state.channel_list:
ui_state.channel_list.append(packet["from"])
refresh_channels = True
lenBack = 0 if "routeBack" not in msg_dict else len(msg_dict["routeBack"])
backValid = "hopStart" in packet and "snrBack" in msg_dict and len(msg_dict["snrBack"]) == lenBack + 1
if backValid:
msg_str += "Back:\n"
route_str = get_name_from_database(packet["from"], "short") or f"{packet['from']:08x}"
if is_chat_archived(packet["from"]):
update_node_info_in_db(packet["from"], chat_archived=False)
if lenBack > 0:
for idx, node_num in enumerate(msg_dict["routeBack"]):
route_str += (
" --> "
+ (get_name_from_database(node_num, "short") or f"{node_num:08x}")
+ " ("
+ (str(msg_dict["snrBack"][idx] / 4) if msg_dict["snrBack"][idx] != UNK_SNR else "?")
+ "dB)"
)
channel_number = ui_state.channel_list.index(packet["from"])
channel_id = ui_state.channel_list[channel_number]
route_str += (
" --> "
+ (get_name_from_database(packet["to"], "short") or f"{packet['to']:08x}")
+ " ("
+ (str(msg_dict["snrBack"][-1] / 4) if msg_dict["snrBack"][-1] != UNK_SNR else "?")
+ "dB)"
)
if channel_id == ui_state.channel_list[ui_state.selected_channel]:
refresh_messages = True
else:
add_notification(channel_number)
refresh_channels = True
msg_str += route_str + "\n"
message_from_string = get_name_from_database(packet["from"], type="short") + ":\n"
if packet["from"] not in ui_state.channel_list:
ui_state.channel_list.append(packet["from"])
refresh_channels = True
add_new_message(channel_id, f"{config.message_prefix} {message_from_string}", msg_str)
if is_chat_archived(packet["from"]):
update_node_info_in_db(packet["from"], chat_archived=False)
if refresh_channels:
draw_channel_list()
if refresh_messages:
draw_messages_window(True)
channel_number = ui_state.channel_list.index(packet["from"])
channel_id = ui_state.channel_list[channel_number]
save_message_to_db(channel_id, packet["from"], msg_str)
if channel_id == ui_state.channel_list[ui_state.selected_channel]:
refresh_messages = True
else:
add_notification(channel_number)
refresh_channels = True
message_from_string = get_name_from_database(packet["from"], type="short") + ":\n"
add_new_message(channel_id, f"{config.message_prefix} {message_from_string}", msg_str)
if refresh_channels:
request_ui_redraw(channels=True)
if refresh_messages:
request_ui_redraw(messages=True, scroll_messages_to_bottom=True)
save_message_to_db(channel_id, packet["from"], msg_str)
def send_message(message: str, destination: int = BROADCAST_NUM, channel: int = 0) -> None:
"""

View File

@@ -6,19 +6,26 @@ import sys
import traceback
import contact.ui.default_config as config
from contact.utilities.input_handlers import get_list_input
from contact.utilities.i18n import t
from contact.ui.dialog import dialog
from contact.utilities.i18n import t
from contact.ui.colors import setup_colors
from contact.ui.splash import draw_splash
from contact.ui.control_ui import set_region, settings_menu
from contact.ui.dialog import dialog
from contact.ui.splash import draw_splash
from contact.utilities.arg_parser import setup_parser
from contact.utilities.i18n import t
from contact.utilities.input_handlers import get_list_input
from contact.utilities.interfaces import initialize_interface, reconnect_interface
def close_interface(interface: object) -> None:
if interface is None:
return
with contextlib.suppress(Exception):
interface.close()
def main(stdscr: curses.window) -> None:
output_capture = io.StringIO()
interface = None
try:
with contextlib.redirect_stdout(output_capture), contextlib.redirect_stderr(output_capture):
setup_colors()
@@ -39,7 +46,7 @@ def main(stdscr: curses.window) -> None:
)
if confirmation == "Yes":
set_region(interface)
interface.close()
close_interface(interface)
draw_splash(stdscr)
interface = reconnect_interface(args)
stdscr.clear()
@@ -52,6 +59,8 @@ def main(stdscr: curses.window) -> None:
logging.error("Traceback: %s", traceback.format_exc())
logging.error("Console output before crash:\n%s", console_output)
raise
finally:
close_interface(interface)
def ensure_min_rows(stdscr: curses.window, min_rows: int = 11) -> None:

View File

@@ -15,8 +15,15 @@ from contact.utilities.i18n import t
from contact.utilities.emoji_utils import normalize_message_text
import contact.ui.default_config as config
import contact.ui.dialog
from contact.ui.nav_utils import move_main_highlight, draw_main_arrows, get_msg_window_lines, wrap_text
from contact.utilities.singleton import ui_state, interface_state, menu_state
from contact.ui.nav_utils import (
move_main_highlight,
draw_main_arrows,
get_msg_window_lines,
wrap_text,
truncate_with_ellipsis,
pad_to_width,
)
from contact.utilities.singleton import ui_state, interface_state, menu_state, app_state
MIN_COL = 1 # "effectively zero" without breaking curses
@@ -25,6 +32,53 @@ root_win = None
nodes_pad = None
def request_ui_redraw(
*,
channels: bool = False,
messages: bool = False,
nodes: bool = False,
packetlog: bool = False,
full: bool = False,
scroll_messages_to_bottom: bool = False,
) -> None:
ui_state.redraw_channels = ui_state.redraw_channels or channels
ui_state.redraw_messages = ui_state.redraw_messages or messages
ui_state.redraw_nodes = ui_state.redraw_nodes or nodes
ui_state.redraw_packetlog = ui_state.redraw_packetlog or packetlog
ui_state.redraw_full_ui = ui_state.redraw_full_ui or full
ui_state.scroll_messages_to_bottom = ui_state.scroll_messages_to_bottom or scroll_messages_to_bottom
def process_pending_ui_updates(stdscr: curses.window) -> None:
if ui_state.redraw_full_ui:
ui_state.redraw_full_ui = False
ui_state.redraw_channels = False
ui_state.redraw_messages = False
ui_state.redraw_nodes = False
ui_state.redraw_packetlog = False
ui_state.scroll_messages_to_bottom = False
handle_resize(stdscr, False)
return
if ui_state.redraw_channels:
ui_state.redraw_channels = False
draw_channel_list()
if ui_state.redraw_nodes:
ui_state.redraw_nodes = False
draw_node_list()
if ui_state.redraw_messages:
scroll_to_bottom = ui_state.scroll_messages_to_bottom
ui_state.redraw_messages = False
ui_state.scroll_messages_to_bottom = False
draw_messages_window(scroll_to_bottom)
if ui_state.redraw_packetlog:
ui_state.redraw_packetlog = False
draw_packetlog_win()
# Draw arrows for a specific window id (0=channel,1=messages,2=nodes).
def draw_window_arrows(window_id: int) -> None:
@@ -131,6 +185,43 @@ def get_node_display_name(node_num: int, node: dict) -> str:
return user.get("longName") or get_name_from_database(node_num, "long")
def get_selected_channel_title() -> str:
if not ui_state.channel_list:
return ""
channel = ui_state.channel_list[min(ui_state.selected_channel, len(ui_state.channel_list) - 1)]
if isinstance(channel, int):
return get_name_from_database(channel, "long") or get_name_from_database(channel, "short") or str(channel)
return str(channel)
def get_window_title(window: int) -> str:
if window == 2:
return f"Nodes: {len(ui_state.node_list)}"
if ui_state.single_pane_mode and window == 1:
return get_selected_channel_title()
return ""
def draw_frame_title(box: curses.window, title: str) -> None:
if not title:
return
_, box_w = box.getmaxyx()
max_title_width = max(0, box_w - 6)
if max_title_width <= 0:
return
clipped_title = truncate_with_ellipsis(title, max_title_width).rstrip()
if not clipped_title:
return
try:
box.addstr(0, 2, f" {clipped_title} ", curses.A_BOLD)
except curses.error:
pass
def handle_resize(stdscr: curses.window, firstrun: bool) -> None:
"""Handle terminal resize events and redraw the UI accordingly."""
global messages_pad, messages_win, nodes_pad, nodes_win, channel_pad, channel_win, packetlog_win, entry_win
@@ -139,11 +230,19 @@ def handle_resize(stdscr: curses.window, firstrun: bool) -> None:
height, width = stdscr.getmaxyx()
if ui_state.single_pane_mode:
channel_width, messages_width, nodes_width = compute_widths(width, ui_state.current_window)
channel_width = width
messages_width = width
nodes_width = width
channel_x = 0
messages_x = 0
nodes_x = 0
else:
channel_width = int(config.channel_list_16ths) * (width // 16)
nodes_width = int(config.node_list_16ths) * (width // 16)
messages_width = width - channel_width - nodes_width
channel_x = 0
messages_x = channel_width
nodes_x = channel_width + messages_width
channel_width = max(MIN_COL, channel_width)
messages_width = max(MIN_COL, messages_width)
@@ -151,7 +250,7 @@ def handle_resize(stdscr: curses.window, firstrun: bool) -> None:
# Ensure the three widths sum exactly to the terminal width by adjusting the focused pane
total = channel_width + messages_width + nodes_width
if total != width:
if not ui_state.single_pane_mode and total != width:
delta = total - width
if ui_state.current_window == 0:
channel_width = max(MIN_COL, channel_width - delta)
@@ -168,11 +267,11 @@ def handle_resize(stdscr: curses.window, firstrun: bool) -> None:
if firstrun:
entry_win = curses.newwin(entry_height, width, height - entry_height, 0)
channel_win = curses.newwin(content_h, channel_width, 0, 0)
messages_win = curses.newwin(content_h, messages_width, 0, channel_width)
nodes_win = curses.newwin(content_h, nodes_width, 0, channel_width + messages_width)
channel_win = curses.newwin(content_h, channel_width, 0, channel_x)
messages_win = curses.newwin(content_h, messages_width, 0, messages_x)
nodes_win = curses.newwin(content_h, nodes_width, 0, nodes_x)
packetlog_win = curses.newwin(pkt_h, messages_width, height - pkt_h - entry_height, channel_width)
packetlog_win = curses.newwin(pkt_h, messages_width, height - pkt_h - entry_height, messages_x)
# Will be resized to what we need when drawn
messages_pad = curses.newpad(1, 1)
@@ -199,23 +298,30 @@ def handle_resize(stdscr: curses.window, firstrun: bool) -> None:
entry_win.mvwin(height - entry_height, 0)
channel_win.resize(content_h, channel_width)
channel_win.mvwin(0, 0)
channel_win.mvwin(0, channel_x)
messages_win.resize(content_h, messages_width)
messages_win.mvwin(0, channel_width)
messages_win.mvwin(0, messages_x)
nodes_win.resize(content_h, nodes_width)
nodes_win.mvwin(0, channel_width + messages_width)
nodes_win.mvwin(0, nodes_x)
packetlog_win.resize(pkt_h, messages_width)
packetlog_win.mvwin(height - pkt_h - entry_height, channel_width)
packetlog_win.mvwin(height - pkt_h - entry_height, messages_x)
# Draw window borders
for win in [channel_win, entry_win, nodes_win, messages_win]:
windows_to_draw = [entry_win]
if ui_state.single_pane_mode:
windows_to_draw.append([channel_win, messages_win, nodes_win][ui_state.current_window])
else:
windows_to_draw.extend([channel_win, nodes_win, messages_win])
for win in windows_to_draw:
win.box()
win.refresh()
entry_win.keypad(True)
entry_win.timeout(200)
curses.curs_set(1)
try:
@@ -261,14 +367,19 @@ def main_ui(stdscr: curses.window) -> None:
handle_resize(stdscr, True)
while True:
with app_state.lock:
process_pending_ui_updates(stdscr)
draw_text_field(entry_win, f"Message: {(input_text or '')[-(stdscr.getmaxyx()[1] - 10):]}", get_color("input"))
# Get user input from entry window
if queued_char is None:
char = entry_win.get_wch()
else:
char = queued_char
queued_char = None
try:
if queued_char is None:
char = entry_win.get_wch()
else:
char = queued_char
queued_char = None
except curses.error:
continue
# draw_debug(f"Keypress: {char}")
@@ -961,7 +1072,7 @@ def draw_channel_list() -> None:
channel_pad.erase()
win_width = channel_win.getmaxyx()[1]
channel_pad.resize(len(ui_state.all_messages), channel_win.getmaxyx()[1])
channel_pad.resize(max(1, len(ui_state.channel_list)), channel_win.getmaxyx()[1])
idx = 0
for channel in ui_state.channel_list:
@@ -978,9 +1089,7 @@ def draw_channel_list() -> None:
notification = " " + config.notification_symbol if idx in ui_state.notifications else ""
# Truncate the channel name if it's too long to fit in the window
truncated_channel = (
(channel[: win_width - 5] + "-" if len(channel) > win_width - 5 else channel) + notification
).ljust(win_width - 3)
truncated_channel = truncate_with_ellipsis(f"{channel}{notification}", win_width - 4)
color = get_color("channel_list")
if idx == ui_state.selected_channel:
@@ -1075,8 +1184,7 @@ def draw_node_list() -> None:
node_name = get_node_display_name(node_num, node)
# Future node name custom formatting possible
node_str = f"{status_icon} {node_name}"
node_str = node_str.ljust(box_width - 4)[: box_width - 2]
node_str = truncate_with_ellipsis(f"{status_icon} {node_name}", box_width - 4)
nodes_pad.addstr(i, 1, node_str, get_node_row_color(i))
paint_frame(nodes_win, selected=(ui_state.current_window == 2))
@@ -1298,8 +1406,7 @@ def refresh_pad(window: int) -> None:
pad = messages_pad
box = messages_win
lines = get_msg_window_lines(messages_win, packetlog_win)
selected_item = ui_state.selected_message
start_index = ui_state.selected_message
start_index = ui_state.start_index[1]
if ui_state.display_log:
packetlog_win.box()
@@ -1309,7 +1416,6 @@ def refresh_pad(window: int) -> None:
pad = nodes_pad
box = nodes_win
lines = box.getmaxyx()[0] - 2
box.addstr(0, 2, (f"Nodes: {len(ui_state.node_list)}"), curses.A_BOLD)
selected_item = ui_state.selected_node
start_index = max(0, selected_item - (win_height - 3)) # Leave room for borders
@@ -1346,6 +1452,9 @@ def refresh_pad(window: int) -> None:
if bottom < top or right < left:
return
draw_frame_title(box, get_window_title(window))
box.refresh()
pad.refresh(
start_index,
0,

View File

@@ -5,6 +5,7 @@ import logging
import os
import sys
from typing import List
from meshtastic.protobuf import admin_pb2
from contact.utilities.save_to_radio import save_changes
import contact.ui.default_config as config
@@ -34,7 +35,7 @@ MAX_MENU_WIDTH = 80 # desired max; will shrink on small terminals
save_option = "Save Changes"
max_help_lines = 0
help_win = None
sensitive_settings = ["Reboot", "Reset Node DB", "Shutdown", "Factory Reset"]
sensitive_settings = ["Reboot", "Reset Node DB", "Shutdown", "Factory Reset", "factory_reset_config"]
# Compute the effective menu width for the current terminal
@@ -43,7 +44,7 @@ def get_menu_width() -> int:
return max(20, min(MAX_MENU_WIDTH, curses.COLS - 2))
sensitive_settings = ["Reboot", "Reset Node DB", "Shutdown", "Factory Reset"]
sensitive_settings = ["Reboot", "Reset Node DB", "Shutdown", "Factory Reset", "factory_reset_config"]
# Get the parent directory of the script
script_dir = os.path.dirname(os.path.abspath(__file__))
@@ -246,6 +247,26 @@ def reconnect_after_admin_action(stdscr: object, interface: object, action, log_
return reconnect_interface_with_splash(stdscr, interface)
def request_factory_reset(node: object, full: bool = False):
try:
return node.factoryReset(full=full)
except TypeError as ex:
field_name = "factory_reset_device" if full else "factory_reset_config"
field = admin_pb2.AdminMessage.DESCRIPTOR.fields_by_name[field_name]
if field.cpp_type != field.CPPTYPE_INT32:
raise
node.ensureSessionKey()
message = admin_pb2.AdminMessage()
setattr(message, field_name, 1)
if node == node.iface.localNode:
on_response = None
else:
on_response = node.onAckNak
return node._sendAdmin(message, onResponse=on_response)
def redraw_main_ui_after_reconnect(stdscr: object) -> None:
try:
from contact.ui import contact_ui
@@ -538,7 +559,27 @@ def settings_menu(stdscr: object, interface: object) -> None:
)
if confirmation == "Yes":
interface = reconnect_after_admin_action(
stdscr, interface, interface.localNode.factoryReset, "Factory Reset Requested by menu"
stdscr,
interface,
lambda: request_factory_reset(interface.localNode, full=True),
"Factory Reset Requested by menu",
)
menu = rebuild_menu_at_current_path(interface, menu_state)
menu_state.start_index.pop()
continue
elif selected_option == "factory_reset_config":
confirmation = get_list_input(
t("ui.confirm.factory_reset_config", default="Are you sure you want to Factory Reset Config?"),
None,
["Yes", "No"],
)
if confirmation == "Yes":
interface = reconnect_after_admin_action(
stdscr,
interface,
lambda: request_factory_reset(interface.localNode, full=False),
"Factory Reset Config Requested by menu",
)
menu = rebuild_menu_at_current_path(interface, menu_state)
menu_state.start_index.pop()

View File

@@ -133,6 +133,7 @@ def generate_menu_from_protobuf(interface: object) -> Dict[str, Any]:
"Reset Node DB": None,
"Shutdown": None,
"Factory Reset": None,
"factory_reset_config": None,
"Exit": None,
}
)

View File

@@ -331,6 +331,51 @@ def text_width(text: str) -> int:
return sum(2 if east_asian_width(c) in "FW" else 1 for c in text)
def slice_to_width(text: str, max_width: int) -> str:
if max_width <= 0:
return ""
width = 0
chars = []
for char in text:
char_width = text_width(char)
if width + char_width > max_width:
break
chars.append(char)
width += char_width
return "".join(chars)
def pad_to_width(text: str, width: int) -> str:
clipped = slice_to_width(text, width)
return clipped + (" " * max(0, width - text_width(clipped)))
def truncate_with_ellipsis(text: str, width: int) -> str:
if width <= 0:
return ""
if text_width(text) <= width:
return pad_to_width(text, width)
if width == 1:
return ""
return pad_to_width(slice_to_width(text, width - 1) + "", width)
def split_text_to_width_chunks(text: str, width: int) -> List[str]:
if width <= 0:
return [""]
chunks = []
remaining = text
while remaining:
chunk = slice_to_width(remaining, width)
if not chunk:
break
chunks.append(chunk)
remaining = remaining[len(chunk) :]
return chunks or [""]
def wrap_text(text: str, wrap_width: int) -> List[str]:
"""Wraps text while preserving spaces and breaking long words."""
@@ -353,8 +398,7 @@ def wrap_text(text: str, wrap_width: int) -> List[str]:
wrapped_lines.append(line_buffer.strip())
line_buffer = ""
line_length = 0
for i in range(0, word_length, wrap_width):
wrapped_lines.append(word[i : i + wrap_width])
wrapped_lines.extend(split_text_to_width_chunks(word, wrap_width))
continue
if line_length + word_length > wrap_width and word.strip():

View File

@@ -33,6 +33,12 @@ class ChatUIState:
show_save_option: bool = False
menu_path: List[str] = field(default_factory=list)
single_pane_mode: bool = False
redraw_channels: bool = False
redraw_messages: bool = False
redraw_nodes: bool = False
redraw_packetlog: bool = False
redraw_full_ui: bool = False
scroll_messages_to_bottom: bool = False
@dataclass

View File

@@ -1,6 +1,6 @@
[project]
name = "contact"
version = "1.5.0"
version = "1.5.4"
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"}

View File

@@ -3,6 +3,7 @@ from unittest import mock
import contact.ui.default_config as config
from contact.ui import contact_ui
from contact.ui.nav_utils import text_width
from contact.utilities.singleton import ui_state
from tests.test_support import reset_singletons, restore_config, snapshot_config
@@ -36,3 +37,166 @@ class ContactUiTests(unittest.TestCase):
self.assertEqual(curs_set.call_args_list[0].args, (0,))
self.assertEqual(curs_set.call_args_list[-1].args, (1,))
self.assertEqual(ui_state.current_window, 1)
def test_process_pending_ui_updates_draws_requested_windows(self) -> None:
stdscr = mock.Mock()
ui_state.redraw_channels = True
ui_state.redraw_messages = True
ui_state.redraw_nodes = True
ui_state.redraw_packetlog = True
ui_state.scroll_messages_to_bottom = True
with mock.patch.object(contact_ui, "draw_channel_list") as draw_channel_list:
with mock.patch.object(contact_ui, "draw_messages_window") as draw_messages_window:
with mock.patch.object(contact_ui, "draw_node_list") as draw_node_list:
with mock.patch.object(contact_ui, "draw_packetlog_win") as draw_packetlog_win:
contact_ui.process_pending_ui_updates(stdscr)
draw_channel_list.assert_called_once_with()
draw_messages_window.assert_called_once_with(True)
draw_node_list.assert_called_once_with()
draw_packetlog_win.assert_called_once_with()
def test_process_pending_ui_updates_full_redraw_uses_handle_resize(self) -> None:
stdscr = mock.Mock()
ui_state.redraw_full_ui = True
ui_state.redraw_channels = True
ui_state.redraw_messages = True
with mock.patch.object(contact_ui, "handle_resize") as handle_resize:
contact_ui.process_pending_ui_updates(stdscr)
handle_resize.assert_called_once_with(stdscr, False)
self.assertFalse(ui_state.redraw_channels)
self.assertFalse(ui_state.redraw_messages)
def test_refresh_node_selection_reserves_scroll_arrow_column(self) -> None:
ui_state.node_list = [101, 202]
ui_state.selected_node = 1
ui_state.start_index = [0, 0, 0]
contact_ui.nodes_pad = mock.Mock()
contact_ui.nodes_pad.getmaxyx.return_value = (4, 20)
contact_ui.nodes_win = mock.Mock()
contact_ui.nodes_win.getmaxyx.return_value = (10, 20)
interface = mock.Mock()
interface.nodesByNum = {101: {}, 202: {}}
with mock.patch.object(contact_ui, "refresh_pad") as refresh_pad:
with mock.patch.object(contact_ui, "draw_window_arrows") as draw_window_arrows:
with mock.patch.object(contact_ui, "get_node_row_color", side_effect=[11, 22]):
with mock.patch("contact.ui.contact_ui.interface_state.interface", interface):
contact_ui.refresh_node_selection(old_index=0, highlight=True)
self.assertEqual(
contact_ui.nodes_pad.chgat.call_args_list,
[mock.call(0, 1, 16, 11), mock.call(1, 1, 16, 22)],
)
refresh_pad.assert_called_once_with(2)
draw_window_arrows.assert_called_once_with(2)
def test_draw_channel_list_reserves_scroll_arrow_column(self) -> None:
ui_state.channel_list = ["VeryLongChannelName"]
ui_state.notifications = []
ui_state.selected_channel = 0
ui_state.current_window = 0
contact_ui.channel_pad = mock.Mock()
contact_ui.channel_win = mock.Mock()
contact_ui.channel_win.getmaxyx.return_value = (10, 20)
with mock.patch.object(contact_ui, "get_color", return_value=1):
with mock.patch.object(contact_ui, "paint_frame"):
with mock.patch.object(contact_ui, "refresh_pad"):
with mock.patch.object(contact_ui, "draw_window_arrows"):
with mock.patch.object(contact_ui, "remove_notification"):
contact_ui.draw_channel_list()
text = contact_ui.channel_pad.addstr.call_args.args[2]
self.assertEqual(len(text), 16)
def test_draw_node_list_reserves_scroll_arrow_column(self) -> None:
ui_state.node_list = [101]
ui_state.current_window = 2
contact_ui.nodes_pad = mock.Mock()
contact_ui.nodes_win = mock.Mock()
contact_ui.nodes_win.getmaxyx.return_value = (10, 20)
contact_ui.entry_win = mock.Mock()
interface = mock.Mock()
interface.nodesByNum = {101: {"user": {"longName": "VeryLongNodeName", "publicKey": ""}}}
with mock.patch("contact.ui.contact_ui.interface_state.interface", interface):
with mock.patch.object(contact_ui, "get_node_row_color", return_value=1):
with mock.patch.object(contact_ui.curses, "curs_set"):
with mock.patch.object(contact_ui, "paint_frame"):
with mock.patch.object(contact_ui, "refresh_pad"):
with mock.patch.object(contact_ui, "draw_window_arrows"):
contact_ui.draw_node_list()
text = contact_ui.nodes_pad.addstr.call_args.args[2]
self.assertEqual(text_width(text), 16)
self.assertIn("", text)
def test_handle_resize_single_pane_keeps_full_width_windows(self) -> None:
stdscr = mock.Mock()
stdscr.getmaxyx.return_value = (24, 80)
ui_state.single_pane_mode = True
ui_state.current_window = 1
contact_ui.entry_win = mock.Mock()
contact_ui.channel_win = mock.Mock()
contact_ui.messages_win = mock.Mock()
contact_ui.nodes_win = mock.Mock()
contact_ui.packetlog_win = mock.Mock()
contact_ui.messages_pad = mock.Mock()
contact_ui.nodes_pad = mock.Mock()
contact_ui.channel_pad = mock.Mock()
with mock.patch.object(contact_ui.curses, "curs_set"):
with mock.patch.object(contact_ui, "draw_channel_list") as draw_channel_list:
with mock.patch.object(contact_ui, "draw_messages_window") as draw_messages_window:
with mock.patch.object(contact_ui, "draw_node_list") as draw_node_list:
with mock.patch.object(contact_ui, "draw_window_arrows") as draw_window_arrows:
contact_ui.handle_resize(stdscr, False)
contact_ui.channel_win.resize.assert_called_once_with(21, 80)
contact_ui.messages_win.resize.assert_called_once_with(21, 80)
contact_ui.nodes_win.resize.assert_called_once_with(21, 80)
contact_ui.channel_win.mvwin.assert_called_once_with(0, 0)
contact_ui.messages_win.mvwin.assert_called_once_with(0, 0)
contact_ui.nodes_win.mvwin.assert_called_once_with(0, 0)
contact_ui.channel_win.box.assert_not_called()
contact_ui.nodes_win.box.assert_not_called()
contact_ui.messages_win.box.assert_called_once_with()
draw_channel_list.assert_called_once_with()
draw_messages_window.assert_called_once_with(True)
draw_node_list.assert_called_once_with()
draw_window_arrows.assert_called_once_with(1)
def test_get_window_title_uses_selected_channel_only_for_messages_in_single_pane_mode(self) -> None:
ui_state.single_pane_mode = True
ui_state.channel_list = ["Primary"]
ui_state.selected_channel = 0
self.assertEqual(contact_ui.get_window_title(0), "")
self.assertEqual(contact_ui.get_window_title(1), "Primary")
def test_refresh_pad_draws_selected_channel_title_on_message_frame(self) -> None:
ui_state.single_pane_mode = True
ui_state.current_window = 1
ui_state.channel_list = ["Primary"]
ui_state.selected_channel = 0
ui_state.start_index = [0, 0, 0]
ui_state.display_log = False
contact_ui.channel_win = mock.Mock()
contact_ui.channel_win.getmaxyx.return_value = (10, 20)
contact_ui.messages_pad = mock.Mock()
contact_ui.messages_pad.getmaxyx.return_value = (5, 20)
contact_ui.messages_win = mock.Mock()
contact_ui.messages_win.getbegyx.return_value = (0, 0)
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)
contact_ui.messages_win.addstr.assert_called_once_with(0, 2, " Primary ", contact_ui.curses.A_BOLD)

View File

@@ -1,4 +1,5 @@
from argparse import Namespace
from types import SimpleNamespace
import unittest
from unittest import mock
@@ -63,3 +64,49 @@ class ControlUiTests(unittest.TestCase):
get_channels.assert_called_once_with()
refresh_node_list.assert_called_once_with()
handle_resize.assert_called_once_with(stdscr, False)
def test_request_factory_reset_uses_library_helper_when_supported(self) -> None:
node = mock.Mock()
control_ui.request_factory_reset(node)
node.factoryReset.assert_called_once_with(full=False)
node.ensureSessionKey.assert_not_called()
node._sendAdmin.assert_not_called()
def test_request_factory_reset_uses_library_helper_for_full_reset_when_supported(self) -> None:
node = mock.Mock()
control_ui.request_factory_reset(node, full=True)
node.factoryReset.assert_called_once_with(full=True)
node.ensureSessionKey.assert_not_called()
node._sendAdmin.assert_not_called()
def test_request_factory_reset_falls_back_to_int_valued_admin_message(self) -> None:
node = mock.Mock()
node.factoryReset.side_effect = TypeError(
"Field meshtastic.protobuf.AdminMessage.factory_reset_config: Expected an int, got a boolean."
)
node.iface = SimpleNamespace(localNode=node)
control_ui.request_factory_reset(node)
node.ensureSessionKey.assert_called_once_with()
sent_message = node._sendAdmin.call_args.args[0]
self.assertEqual(sent_message.factory_reset_config, 1)
self.assertIsNone(node._sendAdmin.call_args.kwargs["onResponse"])
def test_request_factory_reset_full_falls_back_to_int_valued_admin_message(self) -> None:
node = mock.Mock()
node.factoryReset.side_effect = TypeError(
"Field meshtastic.protobuf.AdminMessage.factory_reset_device: Expected an int, got a boolean."
)
node.iface = SimpleNamespace(localNode=node)
control_ui.request_factory_reset(node, full=True)
node.ensureSessionKey.assert_called_once_with()
sent_message = node._sendAdmin.call_args.args[0]
self.assertEqual(sent_message.factory_reset_device, 1)
self.assertIsNone(node._sendAdmin.call_args.kwargs["onResponse"])

28
tests/test_menus.py Normal file
View File

@@ -0,0 +1,28 @@
from types import SimpleNamespace
import unittest
from meshtastic.protobuf import config_pb2, module_config_pb2
from contact.ui.menus import generate_menu_from_protobuf
class MenusTests(unittest.TestCase):
def test_main_menu_includes_factory_reset_config_after_factory_reset(self) -> None:
local_node = SimpleNamespace(
localConfig=config_pb2.Config(),
moduleConfig=module_config_pb2.ModuleConfig(),
getChannelByChannelIndex=lambda _: None,
)
interface = SimpleNamespace(
localNode=local_node,
getMyNodeInfo=lambda: {
"user": {"longName": "Test User", "shortName": "TU", "isLicensed": False},
"position": {"latitude": 0.0, "longitude": 0.0, "altitude": 0},
},
)
menu = generate_menu_from_protobuf(interface)
keys = list(menu["Main Menu"].keys())
self.assertLess(keys.index("Factory Reset"), keys.index("factory_reset_config"))
self.assertEqual(keys[keys.index("Factory Reset") + 1], "factory_reset_config")

36
tests/test_nav_utils.py Normal file
View File

@@ -0,0 +1,36 @@
import unittest
from unittest import mock
from contact.ui import nav_utils
from contact.ui.nav_utils import truncate_with_ellipsis, wrap_text
from contact.utilities.singleton import ui_state
class NavUtilsTests(unittest.TestCase):
def setUp(self) -> None:
ui_state.current_window = 0
ui_state.node_list = []
ui_state.start_index = [0, 0, 0]
def test_wrap_text_splits_wide_characters_by_display_width(self) -> None:
self.assertEqual(wrap_text("🔐🔐🔐", 4), ["🔐", "🔐", "🔐"])
def test_truncate_with_ellipsis_respects_display_width(self) -> None:
self.assertEqual(truncate_with_ellipsis("🔐Alpha", 5), "🔐Al…")
def test_highlight_line_reserves_scroll_arrow_column_for_nodes(self) -> None:
ui_state.current_window = 2
ui_state.start_index = [0, 0, 0]
menu_win = mock.Mock()
menu_win.getbegyx.return_value = (0, 0)
menu_win.getmaxyx.return_value = (8, 20)
menu_pad = mock.Mock()
menu_pad.getmaxyx.return_value = (4, 20)
with mock.patch.object(nav_utils, "get_node_color", side_effect=[11, 22]):
nav_utils.highlight_line(menu_win, menu_pad, 0, 1, 5)
self.assertEqual(
menu_pad.chgat.call_args_list,
[mock.call(0, 1, 16, 11), mock.call(1, 1, 16, 22)],
)

View File

@@ -34,17 +34,13 @@ class RxHandlerTests(unittest.TestCase):
}
with mock.patch.object(rx_handler, "refresh_node_list", return_value=True):
with mock.patch.object(rx_handler, "draw_node_list") as draw_node_list:
with mock.patch.object(rx_handler, "draw_messages_window") as draw_messages_window:
with mock.patch.object(rx_handler, "draw_channel_list") as draw_channel_list:
with mock.patch.object(rx_handler, "add_notification") as add_notification:
with mock.patch.object(rx_handler, "save_message_to_db") as save_message_to_db:
with mock.patch.object(rx_handler, "get_name_from_database", return_value="SAT2"):
rx_handler.on_receive(packet, interface=None)
with mock.patch.object(rx_handler, "request_ui_redraw") as request_ui_redraw:
with mock.patch.object(rx_handler, "add_notification") as add_notification:
with mock.patch.object(rx_handler, "save_message_to_db") as save_message_to_db:
with mock.patch.object(rx_handler, "get_name_from_database", return_value="SAT2"):
rx_handler.on_receive(packet, interface=None)
draw_node_list.assert_called_once_with()
draw_messages_window.assert_called_once_with(True)
draw_channel_list.assert_not_called()
self.assertEqual(request_ui_redraw.call_args_list, [mock.call(nodes=True), mock.call(messages=True, scroll_messages_to_bottom=True)])
add_notification.assert_not_called()
save_message_to_db.assert_called_once_with("Primary", 222, "hello")
self.assertEqual(ui_state.all_messages["Primary"][-1][1], "hello")
@@ -66,18 +62,16 @@ class RxHandlerTests(unittest.TestCase):
}
with mock.patch.object(rx_handler, "refresh_node_list", return_value=False):
with mock.patch.object(rx_handler, "draw_messages_window") as draw_messages_window:
with mock.patch.object(rx_handler, "draw_channel_list") as draw_channel_list:
with mock.patch.object(rx_handler, "add_notification") as add_notification:
with mock.patch.object(rx_handler, "update_node_info_in_db") as update_node_info_in_db:
with mock.patch.object(rx_handler, "save_message_to_db") as save_message_to_db:
with mock.patch.object(rx_handler, "get_name_from_database", return_value="SAT2"):
rx_handler.on_receive(packet, interface=None)
with mock.patch.object(rx_handler, "request_ui_redraw") as request_ui_redraw:
with mock.patch.object(rx_handler, "add_notification") as add_notification:
with mock.patch.object(rx_handler, "update_node_info_in_db") as update_node_info_in_db:
with mock.patch.object(rx_handler, "save_message_to_db") as save_message_to_db:
with mock.patch.object(rx_handler, "get_name_from_database", return_value="SAT2"):
rx_handler.on_receive(packet, interface=None)
self.assertIn(222, ui_state.channel_list)
self.assertIn(222, ui_state.all_messages)
draw_messages_window.assert_not_called()
draw_channel_list.assert_called_once_with()
request_ui_redraw.assert_called_once_with(channels=True)
add_notification.assert_called_once_with(1)
update_node_info_in_db.assert_called_once_with(222, chat_archived=False)
save_message_to_db.assert_called_once_with(222, 222, "dm")
@@ -87,10 +81,10 @@ class RxHandlerTests(unittest.TestCase):
ui_state.display_log = True
ui_state.current_window = 4
with mock.patch.object(rx_handler, "draw_packetlog_win") as draw_packetlog_win:
with mock.patch.object(rx_handler, "request_ui_redraw") as request_ui_redraw:
rx_handler.on_receive({"id": "new"}, interface=None)
draw_packetlog_win.assert_called_once_with()
request_ui_redraw.assert_called_once_with(packetlog=True)
self.assertEqual(len(ui_state.packet_buffer), 20)
self.assertEqual(ui_state.packet_buffer[-1], {"id": "new"})
self.assertTrue(menu_state.need_redraw)

75
tests/test_settings.py Normal file
View File

@@ -0,0 +1,75 @@
from argparse import Namespace
from types import SimpleNamespace
import unittest
from unittest import mock
import contact.settings as settings
class SettingsRuntimeTests(unittest.TestCase):
def test_main_closes_interface_after_normal_settings_exit(self) -> None:
stdscr = mock.Mock()
args = Namespace()
interface = mock.Mock()
interface.localNode = SimpleNamespace(localConfig=SimpleNamespace(lora=SimpleNamespace(region=1)))
with mock.patch.object(settings, "setup_colors"):
with mock.patch.object(settings, "ensure_min_rows"):
with mock.patch.object(settings, "draw_splash"):
with mock.patch.object(settings.curses, "curs_set"):
with mock.patch.object(settings, "setup_parser") as setup_parser:
with mock.patch.object(settings, "initialize_interface", return_value=interface):
with mock.patch.object(settings, "settings_menu") as settings_menu:
setup_parser.return_value.parse_args.return_value = args
settings.main(stdscr)
settings_menu.assert_called_once_with(stdscr, interface)
interface.close.assert_called_once_with()
def test_main_closes_reconnected_interface_after_region_reset(self) -> None:
stdscr = mock.Mock()
args = Namespace()
old_interface = mock.Mock()
old_interface.localNode = SimpleNamespace(localConfig=SimpleNamespace(lora=SimpleNamespace(region=0)))
new_interface = mock.Mock()
new_interface.localNode = SimpleNamespace(localConfig=SimpleNamespace(lora=SimpleNamespace(region=1)))
with mock.patch.object(settings, "setup_colors"):
with mock.patch.object(settings, "ensure_min_rows"):
with mock.patch.object(settings, "draw_splash"):
with mock.patch.object(settings.curses, "curs_set"):
with mock.patch.object(settings, "setup_parser") as setup_parser:
with mock.patch.object(settings, "initialize_interface", return_value=old_interface):
with mock.patch.object(settings, "get_list_input", return_value="Yes"):
with mock.patch.object(settings, "set_region") as set_region:
with mock.patch.object(
settings, "reconnect_interface", return_value=new_interface
) as reconnect_interface:
with mock.patch.object(settings, "settings_menu") as settings_menu:
setup_parser.return_value.parse_args.return_value = args
settings.main(stdscr)
set_region.assert_called_once_with(old_interface)
reconnect_interface.assert_called_once_with(args)
settings_menu.assert_called_once_with(stdscr, new_interface)
old_interface.close.assert_called_once_with()
new_interface.close.assert_called_once_with()
def test_main_closes_interface_when_settings_menu_raises(self) -> None:
stdscr = mock.Mock()
args = Namespace()
interface = mock.Mock()
interface.localNode = SimpleNamespace(localConfig=SimpleNamespace(lora=SimpleNamespace(region=1)))
with mock.patch.object(settings, "setup_colors"):
with mock.patch.object(settings, "ensure_min_rows"):
with mock.patch.object(settings, "draw_splash"):
with mock.patch.object(settings.curses, "curs_set"):
with mock.patch.object(settings, "setup_parser") as setup_parser:
with mock.patch.object(settings, "initialize_interface", return_value=interface):
with mock.patch.object(settings, "settings_menu", side_effect=RuntimeError("boom")):
setup_parser.return_value.parse_args.return_value = args
with self.assertRaises(RuntimeError):
settings.main(stdscr)
interface.close.assert_called_once_with()

View File

@@ -81,11 +81,11 @@ class TxHandlerTests(unittest.TestCase):
with mock.patch.object(tx_handler, "update_ack_nak") as update_ack_nak:
with mock.patch("contact.message_handlers.tx_handler.time.strftime", return_value="[01:02:03] "):
with mock.patch("contact.ui.contact_ui.draw_messages_window") as draw_messages_window:
with mock.patch("contact.ui.contact_ui.request_ui_redraw") as request_ui_redraw:
tx_handler.onAckNak(packet)
update_ack_nak.assert_called_once_with("Primary", 55, "hello", "Ack")
draw_messages_window.assert_called_once_with()
request_ui_redraw.assert_called_once_with(messages=True)
self.assertIn(config.sent_message_prefix, ui_state.all_messages["Primary"][0][0])
self.assertIn(config.ack_str, ui_state.all_messages["Primary"][0][0])
@@ -100,7 +100,7 @@ class TxHandlerTests(unittest.TestCase):
with mock.patch.object(tx_handler, "update_ack_nak") as update_ack_nak:
with mock.patch("contact.message_handlers.tx_handler.time.strftime", return_value="[01:02:03] "):
with mock.patch("contact.ui.contact_ui.draw_messages_window"):
with mock.patch("contact.ui.contact_ui.request_ui_redraw"):
tx_handler.onAckNak(packet)
update_ack_nak.assert_called_once_with("Primary", 55, "hello", "Implicit")