From 1b6d269d5051f12a0d9f1e0ca357f5aede87c13d Mon Sep 17 00:00:00 2001 From: pdxlocations Date: Thu, 19 Mar 2026 11:40:24 -0700 Subject: [PATCH] Reconnect after config changes --- contact/__main__.py | 54 ++++++++++++-- contact/settings.py | 5 +- contact/ui/contact_ui.py | 1 + contact/ui/control_ui.py | 65 +++++++++++++--- contact/utilities/interfaces.py | 19 +++++ contact/utilities/save_to_radio.py | 88 ++++++++++++++++++++-- contact/utilities/utils.py | 55 +++++++++----- tests/test_contact_ui.py | 38 ++++++++++ tests/test_control_ui.py | 65 ++++++++++++++++ tests/test_interfaces.py | 26 +++++++ tests/test_main.py | 64 ++++++++++++++-- tests/test_save_to_radio.py | 114 +++++++++++++++++++++++++++++ tests/test_utils.py | 22 ++++++ 13 files changed, 566 insertions(+), 50 deletions(-) create mode 100644 tests/test_contact_ui.py create mode 100644 tests/test_control_ui.py create mode 100644 tests/test_interfaces.py create mode 100644 tests/test_save_to_radio.py diff --git a/contact/__main__.py b/contact/__main__.py index 262236a..58b4bc2 100644 --- a/contact/__main__.py +++ b/contact/__main__.py @@ -19,6 +19,7 @@ import subprocess import sys import threading import traceback +from typing import Optional # Third-party from pubsub import pub @@ -36,7 +37,7 @@ from contact.utilities.demo_data import build_demo_interface, configure_demo_dat from contact.utilities.input_handlers import get_list_input from contact.utilities.i18n import t from contact.ui.dialog import dialog -from contact.utilities.interfaces import initialize_interface +from contact.utilities.interfaces import initialize_interface, reconnect_interface from contact.utilities.utils import get_channels, get_nodeNum, get_node_list from contact.utilities.singleton import ui_state, interface_state, app_state @@ -60,13 +61,48 @@ app_state.lock = threading.Lock() # ------------------------------------------------------------------------------ # Main Program Logic # ------------------------------------------------------------------------------ -def prompt_region_if_unset(args: object) -> None: +def prompt_region_if_unset(args: object, stdscr: Optional[curses.window] = None) -> None: """Prompt user to set region if it is unset.""" confirmation = get_list_input("Your region is UNSET. Set it now?", "Yes", ["Yes", "No"]) if confirmation == "Yes": set_region(interface_state.interface) - interface_state.interface.close() - interface_state.interface = initialize_interface(args) + close_interface(interface_state.interface) + if stdscr is not None: + draw_splash(stdscr) + interface_state.interface = reconnect_interface(args) + + +def close_interface(interface: object) -> None: + if interface is None: + return + with contextlib.suppress(Exception): + interface.close() + + +def interface_is_ready(interface: object) -> bool: + try: + return getattr(interface, "localNode", None) is not None and interface.localNode.localConfig is not None + except Exception: + return False + + +def initialize_runtime_interface_with_retry(stdscr: curses.window, args: object): + while True: + interface = initialize_runtime_interface(args) + if getattr(args, "demo_screenshot", False) or interface_is_ready(interface): + return interface + + choice = get_list_input( + t("ui.prompt.node_not_found", default="No node found. Retry connection?"), + "Retry", + ["Retry", "Close"], + mandatory=True, + ) + close_interface(interface) + if choice == "Close": + return None + + draw_splash(stdscr) def initialize_globals(seed_demo: bool = False) -> None: @@ -117,10 +153,12 @@ def main(stdscr: curses.window) -> None: logging.info("Initializing interface...") with app_state.lock: - interface_state.interface = initialize_runtime_interface(args) + interface_state.interface = initialize_runtime_interface_with_retry(stdscr, args) + if interface_state.interface is None: + return if not getattr(args, "demo_screenshot", False) and interface_state.interface.localNode.localConfig.lora.region == 0: - prompt_region_if_unset(args) + prompt_region_if_unset(args, stdscr) initialize_globals(seed_demo=getattr(args, "demo_screenshot", False)) logging.info("Starting main UI") @@ -169,10 +207,10 @@ def start() -> None: try: curses.wrapper(main) - interface_state.interface.close() + close_interface(interface_state.interface) except KeyboardInterrupt: logging.info("User exited with Ctrl+C") - interface_state.interface.close() + close_interface(interface_state.interface) sys.exit(0) except Exception as e: logging.critical("Fatal error", exc_info=True) diff --git a/contact/settings.py b/contact/settings.py index 20247ae..41b7d00 100644 --- a/contact/settings.py +++ b/contact/settings.py @@ -14,7 +14,7 @@ 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.utilities.arg_parser import setup_parser -from contact.utilities.interfaces import initialize_interface +from contact.utilities.interfaces import initialize_interface, reconnect_interface def main(stdscr: curses.window) -> None: @@ -40,7 +40,8 @@ def main(stdscr: curses.window) -> None: if confirmation == "Yes": set_region(interface) interface.close() - interface = initialize_interface(args) + draw_splash(stdscr) + interface = reconnect_interface(args) stdscr.clear() stdscr.refresh() settings_menu(stdscr, interface) diff --git a/contact/ui/contact_ui.py b/contact/ui/contact_ui.py index 17c1979..915b862 100644 --- a/contact/ui/contact_ui.py +++ b/contact/ui/contact_ui.py @@ -774,6 +774,7 @@ def handle_backtick(stdscr: curses.window) -> None: ui_state.current_window = previous_window ui_state.single_pane_mode = config.single_pane_mode.lower() == "true" curses.curs_set(1) + get_channels() refresh_node_list() handle_resize(stdscr, False) diff --git a/contact/ui/control_ui.py b/contact/ui/control_ui.py index c42820b..322ec9c 100644 --- a/contact/ui/control_ui.py +++ b/contact/ui/control_ui.py @@ -9,6 +9,7 @@ from typing import List from contact.utilities.save_to_radio import save_changes import contact.ui.default_config as config from contact.utilities.config_io import config_export, config_import +from contact.utilities.interfaces import reconnect_interface from contact.utilities.control_utils import transform_menu_path from contact.utilities.i18n import t from contact.utilities.ini_utils import parse_ini_file @@ -23,8 +24,10 @@ from contact.ui.colors import get_color from contact.ui.dialog import dialog from contact.ui.menus import generate_menu_from_protobuf from contact.ui.nav_utils import move_highlight, draw_arrows, update_help_window +from contact.ui.splash import draw_splash from contact.ui.user_config import json_editor -from contact.utilities.singleton import menu_state +from contact.utilities.arg_parser import setup_parser +from contact.utilities.singleton import interface_state, menu_state # Setup Variables MAX_MENU_WIDTH = 80 # desired max; will shrink on small terminals @@ -222,6 +225,39 @@ def get_input_type_for_field(field) -> type: return str +def reconnect_interface_with_splash(stdscr: object, interface: object) -> object: + try: + interface.close() + except Exception: + pass + + stdscr.clear() + stdscr.refresh() + draw_splash(stdscr) + new_interface = reconnect_interface(setup_parser().parse_args()) + interface_state.interface = new_interface + redraw_main_ui_after_reconnect(stdscr) + return new_interface + + +def reconnect_after_admin_action(stdscr: object, interface: object, action, log_message: str) -> object: + action() + logging.info(log_message) + return reconnect_interface_with_splash(stdscr, interface) + + +def redraw_main_ui_after_reconnect(stdscr: object) -> None: + try: + from contact.ui import contact_ui + from contact.utilities.utils import get_channels, refresh_node_list + + get_channels() + refresh_node_list() + contact_ui.handle_resize(stdscr, False) + except Exception: + logging.debug("Skipping main UI redraw after reconnect", exc_info=True) + + def settings_menu(stdscr: object, interface: object) -> None: curses.update_lines_cols() @@ -330,9 +366,12 @@ def settings_menu(stdscr: object, interface: object) -> None: help_win.refresh() if menu_state.show_save_option and menu_state.selected_index == len(options): - save_changes(interface, modified_settings, menu_state) + reconnect_required = save_changes(interface, modified_settings, menu_state) modified_settings.clear() logging.info("Changes Saved") + if reconnect_required: + interface = reconnect_interface_with_splash(stdscr, interface) + menu = generate_menu_from_protobuf(interface) if len(menu_state.menu_path) > 1: menu_state.menu_path.pop() @@ -460,8 +499,10 @@ def settings_menu(stdscr: object, interface: object) -> None: t("ui.confirm.reboot", default="Are you sure you want to Reboot?"), None, ["Yes", "No"] ) if confirmation == "Yes": - interface.localNode.reboot() - logging.info(f"Node Reboot Requested by menu") + interface = reconnect_after_admin_action( + stdscr, interface, interface.localNode.reboot, "Node Reboot Requested by menu" + ) + menu = rebuild_menu_at_current_path(interface, menu_state) menu_state.start_index.pop() continue @@ -472,8 +513,10 @@ def settings_menu(stdscr: object, interface: object) -> None: ["Yes", "No"], ) if confirmation == "Yes": - interface.localNode.resetNodeDb() - logging.info(f"Node DB Reset Requested by menu") + interface = reconnect_after_admin_action( + stdscr, interface, interface.localNode.resetNodeDb, "Node DB Reset Requested by menu" + ) + menu = rebuild_menu_at_current_path(interface, menu_state) menu_state.start_index.pop() continue @@ -494,8 +537,10 @@ def settings_menu(stdscr: object, interface: object) -> None: ["Yes", "No"], ) if confirmation == "Yes": - interface.localNode.factoryReset() - logging.info(f"Factory Reset Requested by menu") + interface = reconnect_after_admin_action( + stdscr, interface, interface.localNode.factoryReset, "Factory Reset Requested by menu" + ) + menu = rebuild_menu_at_current_path(interface, menu_state) menu_state.start_index.pop() continue @@ -655,8 +700,10 @@ def settings_menu(stdscr: object, interface: object) -> None: if save_prompt == "Cancel": continue # Stay in the menu without doing anything elif save_prompt == "Yes": - save_changes(interface, modified_settings, menu_state) + reconnect_required = save_changes(interface, modified_settings, menu_state) logging.info("Changes Saved") + if reconnect_required: + interface = reconnect_interface_with_splash(stdscr, interface) modified_settings.clear() menu = rebuild_menu_at_current_path(interface, menu_state) diff --git a/contact/utilities/interfaces.py b/contact/utilities/interfaces.py index 064e7a8..bd33ed6 100644 --- a/contact/utilities/interfaces.py +++ b/contact/utilities/interfaces.py @@ -1,4 +1,5 @@ import logging +import time import meshtastic.serial_interface, meshtastic.tcp_interface, meshtastic.ble_interface @@ -41,3 +42,21 @@ def initialize_interface(args): except Exception as ex: logging.critical(f"Fatal error initializing interface: {ex}") + + +def reconnect_interface(args, attempts: int = 15, delay_seconds: float = 1.0): + last_error = None + + for attempt in range(attempts): + try: + interface = initialize_interface(args) + if interface is not None: + return interface + last_error = RuntimeError("initialize_interface returned None") + except Exception as ex: + last_error = ex + + if attempt < attempts - 1: + time.sleep(delay_seconds) + + raise RuntimeError("Failed to reconnect to the Meshtastic node") from last_error diff --git a/contact/utilities/save_to_radio.py b/contact/utilities/save_to_radio.py index 0a42e8e..a123965 100644 --- a/contact/utilities/save_to_radio.py +++ b/contact/utilities/save_to_radio.py @@ -4,6 +4,79 @@ import logging import base64 import time +DEVICE_REBOOT_KEYS = {"button_gpio", "buzzer_gpio", "role", "rebroadcast_mode"} +POWER_REBOOT_KEYS = { + "device_battery_ina_address", + "is_power_saving", + "ls_secs", + "min_wake_secs", + "on_battery_shutdown_after_secs", + "sds_secs", + "wait_bluetooth_secs", +} +DISPLAY_REBOOT_KEYS = {"screen_on_secs", "flip_screen", "oled", "displaymode"} +LORA_REBOOT_KEYS = { + "use_preset", + "region", + "modem_preset", + "bandwidth", + "spread_factor", + "coding_rate", + "tx_power", + "frequency_offset", + "override_frequency", + "channel_num", + "sx126x_rx_boosted_gain", +} +SECURITY_NON_REBOOT_KEYS = {"debug_log_api_enabled", "serial_enabled"} +USER_RECONNECT_KEYS = {"longName", "shortName", "isLicensed", "is_licensed"} + + +def _collect_changed_keys(modified_settings): + changed = set() + for key, value in modified_settings.items(): + if isinstance(value, dict): + changed.update(_collect_changed_keys(value)) + else: + changed.add(key) + return changed + + +def _requires_reconnect(menu_state, modified_settings) -> bool: + if not modified_settings or len(menu_state.menu_path) < 2: + return False + + section = menu_state.menu_path[1] + changed_keys = _collect_changed_keys(modified_settings) + + if section == "Module Settings": + return True + if section == "User Settings": + return bool(changed_keys & USER_RECONNECT_KEYS) + if section == "Channels": + return False + if section != "Radio Settings" or len(menu_state.menu_path) < 3: + return False + + config_category = menu_state.menu_path[2].lower() + + if config_category in {"network", "bluetooth"}: + return True + if config_category == "security": + return not changed_keys.issubset(SECURITY_NON_REBOOT_KEYS) + if config_category == "device": + return bool(changed_keys & DEVICE_REBOOT_KEYS) + if config_category == "power": + return bool(changed_keys & POWER_REBOOT_KEYS) + if config_category == "display": + return bool(changed_keys & DISPLAY_REBOOT_KEYS) + if config_category == "lora": + return bool(changed_keys & LORA_REBOOT_KEYS) + + # Firmware defaults most config writes to reboot-required unless a handler + # explicitly clears that flag. + return True + def save_changes(interface, modified_settings, menu_state): """ @@ -15,7 +88,7 @@ def save_changes(interface, modified_settings, menu_state): try: if not modified_settings: logging.info("No changes to save. modified_settings is empty.") - return + return False node = interface.getNode("^local") admin_key_backup = None @@ -51,7 +124,7 @@ def save_changes(interface, modified_settings, menu_state): # Return early if there are no other settings left to process if not modified_settings: - return + return _requires_reconnect(menu_state, {"admin_key": admin_key_backup}) if menu_state.menu_path[1] == "Radio Settings" or menu_state.menu_path[1] == "Module Settings": config_category = menu_state.menu_path[2].lower() # for radio and module configs @@ -63,7 +136,7 @@ def save_changes(interface, modified_settings, menu_state): interface.localNode.setFixedPosition(lat, lon, alt) logging.info(f"Updated {config_category} with Latitude: {lat} and Longitude {lon} and Altitude {alt}") - return + return False elif menu_state.menu_path[1] == "User Settings": # for user configs config_category = "User Settings" @@ -78,7 +151,7 @@ def save_changes(interface, modified_settings, menu_state): f"Updated {config_category} with Long Name: {long_name}, Short Name: {short_name}, Licensed Mode: {is_licensed}" ) - return + return _requires_reconnect(menu_state, modified_settings) elif menu_state.menu_path[1] == "Channels": # for channel configs config_category = "Channels" @@ -107,7 +180,7 @@ def save_changes(interface, modified_settings, menu_state): logging.info(f"Updated Channel {channel_num} in {config_category}") logging.info(node.channels) - return + return False else: config_category = None @@ -120,7 +193,7 @@ def save_changes(interface, modified_settings, menu_state): config_container = getattr(node.moduleConfig, config_category) else: logging.warning(f"Config category '{config_category}' not found in config.") - return + return False if len(menu_state.menu_path) >= 4: nested_key = menu_state.menu_path[3] @@ -164,8 +237,11 @@ def save_changes(interface, modified_settings, menu_state): if admin_key_backup is not None: modified_settings["admin_key"] = admin_key_backup + return _requires_reconnect(menu_state, modified_settings) except Exception as e: logging.error(f"Failed to write configuration for category '{config_category}': {e}") + return False except Exception as e: logging.error(f"Error saving changes: {e}") + return False diff --git a/contact/utilities/utils.py b/contact/utilities/utils.py index 68e9817..0a998b0 100644 --- a/contact/utilities/utils.py +++ b/contact/utilities/utils.py @@ -10,35 +10,50 @@ from contact.utilities.singleton import ui_state, interface_state import contact.utilities.telemetry_beautifier as tb +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 + return convert_to_camel_case(modem_preset_string) + + def get_channels(): - """Retrieve channels from the node and update ui_state.channel_list and ui_state.all_messages.""" + """Retrieve channels from the node and rebuild named channel state.""" node = interface_state.interface.getNode("^local") device_channels = node.channels + previous_channel_list = list(ui_state.channel_list) + previous_messages = dict(ui_state.all_messages) - # Clear and rebuild channel list - # ui_state.channel_list = [] + named_channels = [] for device_channel in device_channels: if device_channel.role: - # Use the channel name if available, otherwise use the modem preset - if device_channel.settings.name: - channel_name = device_channel.settings.name - else: - # If channel name is blank, use the modem preset - 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 - channel_name = convert_to_camel_case(modem_preset_string) + named_channels.append(_get_channel_name(device_channel, node)) - # Add channel to ui_state.channel_list if not already present - if channel_name not in ui_state.channel_list: - ui_state.channel_list.append(channel_name) + previous_named_channels = [channel for channel in previous_channel_list if isinstance(channel, str)] + preserved_direct_channels = [channel for channel in previous_channel_list if isinstance(channel, int)] + rebuilt_messages = {} - # Initialize ui_state.all_messages[channel_name] if it doesn't exist - if channel_name not in ui_state.all_messages: - ui_state.all_messages[channel_name] = [] + for index, channel_name in enumerate(named_channels): + previous_name = previous_named_channels[index] if index < len(previous_named_channels) else channel_name + if previous_name in previous_messages: + rebuilt_messages[channel_name] = previous_messages[previous_name] + elif channel_name in previous_messages: + rebuilt_messages[channel_name] = previous_messages[channel_name] + else: + rebuilt_messages[channel_name] = [] + + for channel in preserved_direct_channels: + if channel in previous_messages: + rebuilt_messages[channel] = previous_messages[channel] + + ui_state.channel_list = named_channels + preserved_direct_channels + ui_state.all_messages = rebuilt_messages + if ui_state.channel_list: + ui_state.selected_channel = max(0, min(ui_state.selected_channel, len(ui_state.channel_list) - 1)) return ui_state.channel_list diff --git a/tests/test_contact_ui.py b/tests/test_contact_ui.py new file mode 100644 index 0000000..e4db6eb --- /dev/null +++ b/tests/test_contact_ui.py @@ -0,0 +1,38 @@ +import unittest +from unittest import mock + +import contact.ui.default_config as config +from contact.ui import contact_ui +from contact.utilities.singleton import ui_state + +from tests.test_support import reset_singletons, restore_config, snapshot_config + + +class ContactUiTests(unittest.TestCase): + def setUp(self) -> None: + reset_singletons() + self.saved_config = snapshot_config("single_pane_mode") + + def tearDown(self) -> None: + restore_config(self.saved_config) + reset_singletons() + + def test_handle_backtick_refreshes_channels_after_settings_menu(self) -> None: + stdscr = mock.Mock() + ui_state.current_window = 1 + config.single_pane_mode = "False" + + with mock.patch.object(contact_ui.curses, "curs_set") as curs_set: + with mock.patch.object(contact_ui, "settings_menu") as settings_menu: + with mock.patch.object(contact_ui, "get_channels") as get_channels: + with mock.patch.object(contact_ui, "refresh_node_list") as refresh_node_list: + with mock.patch.object(contact_ui, "handle_resize") as handle_resize: + contact_ui.handle_backtick(stdscr) + + settings_menu.assert_called_once() + get_channels.assert_called_once_with() + refresh_node_list.assert_called_once_with() + handle_resize.assert_called_once_with(stdscr, False) + 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) diff --git a/tests/test_control_ui.py b/tests/test_control_ui.py new file mode 100644 index 0000000..3020308 --- /dev/null +++ b/tests/test_control_ui.py @@ -0,0 +1,65 @@ +from argparse import Namespace +import unittest +from unittest import mock + +from contact.ui import control_ui +from contact.utilities.singleton import interface_state + +from tests.test_support import reset_singletons + + +class ControlUiTests(unittest.TestCase): + def setUp(self) -> None: + reset_singletons() + + def tearDown(self) -> None: + reset_singletons() + + def test_reconnect_interface_with_splash_replaces_interface(self) -> None: + old_interface = mock.Mock() + new_interface = mock.Mock() + stdscr = mock.Mock() + parser = mock.Mock() + parser.parse_args.return_value = Namespace() + + with mock.patch.object(control_ui, "setup_parser", return_value=parser): + with mock.patch.object(control_ui, "draw_splash") as draw_splash: + with mock.patch.object(control_ui, "reconnect_interface", return_value=new_interface) as reconnect: + with mock.patch.object(control_ui, "redraw_main_ui_after_reconnect") as redraw: + result = control_ui.reconnect_interface_with_splash(stdscr, old_interface) + + old_interface.close.assert_called_once_with() + stdscr.clear.assert_called_once_with() + stdscr.refresh.assert_called_once_with() + draw_splash.assert_called_once_with(stdscr) + reconnect.assert_called_once_with(parser.parse_args.return_value) + redraw.assert_called_once_with(stdscr) + self.assertIs(result, new_interface) + self.assertIs(interface_state.interface, new_interface) + + def test_reconnect_after_admin_action_runs_action_then_reconnects(self) -> None: + stdscr = mock.Mock() + interface = mock.Mock() + new_interface = mock.Mock() + action = mock.Mock() + + with mock.patch.object(control_ui, "reconnect_interface_with_splash", return_value=new_interface) as reconnect: + result = control_ui.reconnect_after_admin_action( + stdscr, interface, action, "Factory Reset Requested by menu" + ) + + action.assert_called_once_with() + reconnect.assert_called_once_with(stdscr, interface) + self.assertIs(result, new_interface) + + def test_redraw_main_ui_after_reconnect_refreshes_channels_nodes_and_layout(self) -> None: + stdscr = mock.Mock() + + with mock.patch("contact.utilities.utils.get_channels") as get_channels: + with mock.patch("contact.utilities.utils.refresh_node_list") as refresh_node_list: + with mock.patch("contact.ui.contact_ui.handle_resize") as handle_resize: + control_ui.redraw_main_ui_after_reconnect(stdscr) + + get_channels.assert_called_once_with() + refresh_node_list.assert_called_once_with() + handle_resize.assert_called_once_with(stdscr, False) diff --git a/tests/test_interfaces.py b/tests/test_interfaces.py new file mode 100644 index 0000000..e594329 --- /dev/null +++ b/tests/test_interfaces.py @@ -0,0 +1,26 @@ +from argparse import Namespace +import unittest +from unittest import mock + +from contact.utilities.interfaces import reconnect_interface + + +class InterfacesTests(unittest.TestCase): + def test_reconnect_interface_retries_until_connection_succeeds(self) -> None: + args = Namespace() + + with mock.patch("contact.utilities.interfaces.initialize_interface", side_effect=[None, None, "iface"]) as initialize: + with mock.patch("contact.utilities.interfaces.time.sleep") as sleep: + result = reconnect_interface(args, attempts=3, delay_seconds=0.25) + + self.assertEqual(result, "iface") + self.assertEqual(initialize.call_count, 3) + self.assertEqual(sleep.call_count, 2) + + def test_reconnect_interface_raises_after_exhausting_attempts(self) -> None: + args = Namespace() + + with mock.patch("contact.utilities.interfaces.initialize_interface", return_value=None): + with mock.patch("contact.utilities.interfaces.time.sleep"): + with self.assertRaises(RuntimeError): + reconnect_interface(args, attempts=2, delay_seconds=0) diff --git a/tests/test_main.py b/tests/test_main.py index 7ea838a..658693a 100644 --- a/tests/test_main.py +++ b/tests/test_main.py @@ -1,4 +1,5 @@ from argparse import Namespace +from types import SimpleNamespace import unittest from unittest import mock @@ -40,20 +41,58 @@ class MainRuntimeTests(unittest.TestCase): self.assertEqual(result, "live-interface") initialize_interface.assert_called_once_with(args) + def test_interface_is_ready_detects_missing_local_node(self) -> None: + self.assertFalse(entrypoint.interface_is_ready(object())) + self.assertTrue(entrypoint.interface_is_ready(SimpleNamespace(localNode=SimpleNamespace(localConfig=mock.Mock())))) + + def test_initialize_runtime_interface_with_retry_retries_until_node_is_ready(self) -> None: + args = Namespace(demo_screenshot=False) + stdscr = mock.Mock() + bad_interface = mock.Mock(spec=["close"]) + good_interface = SimpleNamespace(localNode=SimpleNamespace(localConfig=mock.Mock())) + + with mock.patch.object(entrypoint, "initialize_runtime_interface", side_effect=[bad_interface, good_interface]): + with mock.patch.object(entrypoint, "get_list_input", return_value="Retry") as get_list_input: + with mock.patch.object(entrypoint, "draw_splash") as draw_splash: + result = entrypoint.initialize_runtime_interface_with_retry(stdscr, args) + + self.assertIs(result, good_interface) + get_list_input.assert_called_once() + bad_interface.close.assert_called_once_with() + draw_splash.assert_called_once_with(stdscr) + + def test_initialize_runtime_interface_with_retry_returns_none_when_user_closes(self) -> None: + args = Namespace(demo_screenshot=False) + stdscr = mock.Mock() + bad_interface = mock.Mock(spec=["close"]) + + with mock.patch.object(entrypoint, "initialize_runtime_interface", return_value=bad_interface): + with mock.patch.object(entrypoint, "get_list_input", return_value="Close") as get_list_input: + with mock.patch.object(entrypoint, "draw_splash") as draw_splash: + result = entrypoint.initialize_runtime_interface_with_retry(stdscr, args) + + self.assertIsNone(result) + get_list_input.assert_called_once() + bad_interface.close.assert_called_once_with() + draw_splash.assert_not_called() + def test_prompt_region_if_unset_reinitializes_interface_after_confirmation(self) -> None: args = Namespace() old_interface = mock.Mock() new_interface = mock.Mock() + stdscr = mock.Mock() interface_state.interface = old_interface with mock.patch.object(entrypoint, "get_list_input", return_value="Yes"): with mock.patch.object(entrypoint, "set_region") as set_region: - with mock.patch.object(entrypoint, "initialize_interface", return_value=new_interface) as initialize: - entrypoint.prompt_region_if_unset(args) + with mock.patch.object(entrypoint, "draw_splash") as draw_splash: + with mock.patch.object(entrypoint, "reconnect_interface", return_value=new_interface) as reconnect: + entrypoint.prompt_region_if_unset(args, stdscr) set_region.assert_called_once_with(old_interface) old_interface.close.assert_called_once_with() - initialize.assert_called_once_with(args) + draw_splash.assert_called_once_with(stdscr) + reconnect.assert_called_once_with(args) self.assertIs(interface_state.interface, new_interface) def test_prompt_region_if_unset_leaves_interface_unchanged_when_declined(self) -> None: @@ -63,11 +102,11 @@ class MainRuntimeTests(unittest.TestCase): with mock.patch.object(entrypoint, "get_list_input", return_value="No"): with mock.patch.object(entrypoint, "set_region") as set_region: - with mock.patch.object(entrypoint, "initialize_interface") as initialize: + with mock.patch.object(entrypoint, "reconnect_interface") as reconnect: entrypoint.prompt_region_if_unset(args) set_region.assert_not_called() - initialize.assert_not_called() + reconnect.assert_not_called() interface.close.assert_not_called() self.assertIs(interface_state.interface, interface) @@ -147,6 +186,21 @@ class MainRuntimeTests(unittest.TestCase): wrapper.assert_called_once_with(entrypoint.main) interface.close.assert_called_once_with() + def test_main_returns_cleanly_when_user_closes_missing_node_dialog(self) -> None: + stdscr = mock.Mock() + args = Namespace(settings=False, demo_screenshot=False) + + with mock.patch.object(entrypoint, "setup_colors"): + with mock.patch.object(entrypoint, "ensure_min_rows"): + with mock.patch.object(entrypoint, "draw_splash"): + with mock.patch.object(entrypoint, "setup_parser") as setup_parser: + with mock.patch.object(entrypoint, "initialize_runtime_interface_with_retry", return_value=None): + with mock.patch.object(entrypoint, "initialize_globals") as initialize_globals: + setup_parser.return_value.parse_args.return_value = args + entrypoint.main(stdscr) + + initialize_globals.assert_not_called() + def test_start_handles_keyboard_interrupt(self) -> None: interface = mock.Mock() interface_state.interface = interface diff --git a/tests/test_save_to_radio.py b/tests/test_save_to_radio.py new file mode 100644 index 0000000..516925d --- /dev/null +++ b/tests/test_save_to_radio.py @@ -0,0 +1,114 @@ +from types import SimpleNamespace +import unittest +from unittest import mock + +from contact.utilities.save_to_radio import save_changes + + +class SaveToRadioTests(unittest.TestCase): + def build_interface(self): + node = mock.Mock() + node.localConfig = SimpleNamespace( + lora=SimpleNamespace(region=0, serial_enabled=False), + device=SimpleNamespace(role="CLIENT", name="node"), + security=SimpleNamespace(debug_log_api_enabled=False, serial_enabled=False, admin_key=[]), + display=SimpleNamespace(flip_screen=False, units=0), + power=SimpleNamespace(is_power_saving=False, adc_enabled=False), + network=SimpleNamespace(wifi_enabled=False), + bluetooth=SimpleNamespace(enabled=False), + ) + node.moduleConfig = SimpleNamespace(mqtt=SimpleNamespace(enabled=False)) + interface = mock.Mock() + interface.getNode.return_value = node + return interface, node + + def test_save_changes_returns_true_for_lora_writes_that_require_reconnect(self) -> None: + interface, node = self.build_interface() + menu_state = SimpleNamespace(menu_path=["Main Menu", "Radio Settings", "Lora"]) + + reconnect_required = save_changes(interface, {"region": 7}, menu_state) + + self.assertTrue(reconnect_required) + self.assertEqual(node.localConfig.lora.region, 7) + node.writeConfig.assert_called_once_with("lora") + + def test_save_changes_returns_false_when_nothing_changed(self) -> None: + interface = mock.Mock() + menu_state = SimpleNamespace(menu_path=["Main Menu", "Radio Settings", "Lora"]) + + self.assertFalse(save_changes(interface, {}, menu_state)) + + def test_save_changes_returns_false_for_non_rebooting_security_fields(self) -> None: + interface, node = self.build_interface() + menu_state = SimpleNamespace(menu_path=["Main Menu", "Radio Settings", "Security"]) + + reconnect_required = save_changes(interface, {"serial_enabled": True}, menu_state) + + self.assertFalse(reconnect_required) + self.assertTrue(node.localConfig.security.serial_enabled) + + def test_save_changes_returns_true_for_rebooting_security_fields(self) -> None: + interface, _node = self.build_interface() + menu_state = SimpleNamespace(menu_path=["Main Menu", "Radio Settings", "Security"]) + + reconnect_required = save_changes(interface, {"admin_key": [b"12345678"]}, menu_state) + + self.assertTrue(reconnect_required) + + 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"]) + + self.assertFalse(save_changes(interface, {"name": "renamed"}, menu_state)) + self.assertEqual(node.localConfig.device.name, "renamed") + + node.writeConfig.reset_mock() + self.assertTrue(save_changes(interface, {"role": "ROUTER"}, menu_state)) + self.assertEqual(node.localConfig.device.role, "ROUTER") + + def test_save_changes_returns_true_for_network_settings(self) -> None: + interface, node = self.build_interface() + menu_state = SimpleNamespace(menu_path=["Main Menu", "Radio Settings", "Network"]) + + reconnect_required = save_changes(interface, {"wifi_enabled": True}, menu_state) + + self.assertTrue(reconnect_required) + self.assertTrue(node.localConfig.network.wifi_enabled) + + def test_save_changes_returns_true_only_for_rebooting_power_fields(self) -> None: + interface, node = self.build_interface() + menu_state = SimpleNamespace(menu_path=["Main Menu", "Radio Settings", "Power"]) + + self.assertFalse(save_changes(interface, {"adc_enabled": True}, menu_state)) + self.assertTrue(node.localConfig.power.adc_enabled) + + node.writeConfig.reset_mock() + self.assertTrue(save_changes(interface, {"is_power_saving": True}, menu_state)) + self.assertTrue(node.localConfig.power.is_power_saving) + + def test_save_changes_returns_true_for_module_settings(self) -> None: + interface, node = self.build_interface() + menu_state = SimpleNamespace(menu_path=["Main Menu", "Module Settings", "Mqtt"]) + + reconnect_required = save_changes(interface, {"enabled": True}, menu_state) + + self.assertTrue(reconnect_required) + self.assertTrue(node.moduleConfig.mqtt.enabled) + + def test_save_changes_returns_true_for_user_name_changes(self) -> None: + interface, node = self.build_interface() + menu_state = SimpleNamespace(menu_path=["Main Menu", "User Settings"]) + + reconnect_required = save_changes(interface, {"longName": "Node"}, menu_state) + + self.assertTrue(reconnect_required) + node.setOwner.assert_called_once() + + def test_save_changes_returns_true_for_user_license_changes(self) -> None: + interface, node = self.build_interface() + menu_state = SimpleNamespace(menu_path=["Main Menu", "User Settings"]) + + reconnect_required = save_changes(interface, {"isLicensed": True}, menu_state) + + self.assertTrue(reconnect_required) + node.setOwner.assert_called_once() diff --git a/tests/test_utils.py b/tests/test_utils.py index 8965bc5..0342366 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -60,6 +60,28 @@ class UtilsTests(unittest.TestCase): self.assertIn("MediumFast", ui_state.all_messages) self.assertIn("Another Channel", ui_state.all_messages) + def test_get_channels_rebuilds_renamed_channels_and_preserves_messages(self) -> None: + interface = build_demo_interface() + interface.localNode.channels[0].settings.name = "Renamed Channel" + interface_state.interface = interface + ui_state.channel_list = ["MediumFast", "Another Channel", 2701131788] + ui_state.all_messages = { + "MediumFast": [("prefix", "first")], + "Another Channel": [("prefix", "second")], + 2701131788: [("prefix", "dm")], + } + ui_state.selected_channel = 2 + + channels = get_channels() + + self.assertEqual(channels[0], "Renamed Channel") + self.assertEqual(channels[1], "Another Channel") + self.assertEqual(channels[2], 2701131788) + self.assertEqual(ui_state.all_messages["Renamed Channel"], [("prefix", "first")]) + self.assertEqual(ui_state.all_messages["Another Channel"], [("prefix", "second")]) + self.assertEqual(ui_state.all_messages[2701131788], [("prefix", "dm")]) + self.assertNotIn("MediumFast", ui_state.all_messages) + def test_parse_protobuf_returns_string_payload_unchanged(self) -> None: packet = {"decoded": {"portnum": "TEXT_MESSAGE_APP", "payload": "hello"}}