mirror of
https://github.com/pdxlocations/contact.git
synced 2026-07-28 04:23:00 +02:00
Enhance settings menu and remote admin handling
- Refactor settings_menu to support remote node configurations. - Add show_remote_admin_wait function for displaying progress during remote admin requests. - Update save_changes to accept a node parameter for better flexibility. - Improve error handling for remote admin rejections in handle_backtick. - Add unit tests for new remote admin functionalities and error scenarios.
This commit is contained in:
@@ -938,7 +938,45 @@ def handle_backtick(stdscr: curses.window) -> None:
|
||||
curses.curs_set(0)
|
||||
previous_window = ui_state.current_window
|
||||
ui_state.current_window = 4
|
||||
settings_menu(stdscr, interface_state.interface)
|
||||
interface = interface_state.interface
|
||||
if previous_window != 2:
|
||||
settings_menu(stdscr, interface)
|
||||
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)
|
||||
return
|
||||
|
||||
options = ["Local settings"]
|
||||
remote_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
|
||||
try:
|
||||
wait_win = show_remote_admin_wait(stdscr, get_name_from_database(remote_node_num))
|
||||
settings_menu(stdscr, interface, node=interface.getNode(remote_node_num), remote=True)
|
||||
except SystemExit 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))
|
||||
ui_state.current_window = previous_window
|
||||
ui_state.single_pane_mode = config.single_pane_mode.lower() == "true"
|
||||
curses.curs_set(1)
|
||||
@@ -947,6 +985,25 @@ def handle_backtick(stdscr: curses.window) -> None:
|
||||
handle_resize(stdscr, False)
|
||||
|
||||
|
||||
def show_remote_admin_wait(stdscr: curses.window, node_name: str) -> curses.window | None:
|
||||
"""Show progress while Meshtastic retrieves remote-admin settings."""
|
||||
message = f"Attempting remote admin of {node_name}, waiting for response..."
|
||||
try:
|
||||
height, width = stdscr.getmaxyx()
|
||||
box_width = min(width - 4, max(len(message) + 4, 36))
|
||||
box_height = 5
|
||||
wait_win = curses.newwin(box_height, box_width, max(0, (height - box_height) // 2), max(0, (width - box_width) // 2))
|
||||
wait_win.bkgd(get_color("background"))
|
||||
wait_win.attrset(get_color("window_frame"))
|
||||
wait_win.border()
|
||||
wait_win.addstr(0, 2, " Remote Admin ", get_color("settings_default"))
|
||||
wait_win.addstr(2, 2, message[: box_width - 4], get_color("settings_default"))
|
||||
wait_win.refresh()
|
||||
return wait_win
|
||||
except curses.error:
|
||||
return None
|
||||
|
||||
|
||||
def handle_ctrl_p() -> None:
|
||||
"""Handle Ctrl + P key events to toggle the packet log display."""
|
||||
# Display packet log
|
||||
|
||||
+24
-13
@@ -279,10 +279,21 @@ def redraw_main_ui_after_reconnect(stdscr: object) -> None:
|
||||
logging.debug("Skipping main UI redraw after reconnect", exc_info=True)
|
||||
|
||||
|
||||
def settings_menu(stdscr: object, interface: object) -> None:
|
||||
def settings_menu(stdscr: object, interface: object, node: object = None, remote: bool = False) -> None:
|
||||
curses.update_lines_cols()
|
||||
node = node or interface.localNode
|
||||
|
||||
menu = generate_menu_from_protobuf(interface)
|
||||
if remote:
|
||||
# Remote nodes do not populate config automatically; request every
|
||||
# section before building the menu. Authorization failures surface as
|
||||
# Meshtastic admin errors and are handled by the caller.
|
||||
for config_type in list(node.localConfig.DESCRIPTOR.fields):
|
||||
node.requestConfig(config_type)
|
||||
for config_type in list(node.moduleConfig.DESCRIPTOR.fields):
|
||||
node.requestConfig(config_type)
|
||||
node.requestChannels()
|
||||
|
||||
menu = generate_menu_from_protobuf(interface, node=node, include_app_settings=not remote)
|
||||
menu_state.current_menu = menu["Main Menu"]
|
||||
menu_state.menu_path = ["Main Menu"]
|
||||
|
||||
@@ -387,12 +398,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):
|
||||
reconnect_required = save_changes(interface, modified_settings, menu_state)
|
||||
reconnect_required = save_changes(interface, modified_settings, menu_state, node=node)
|
||||
modified_settings.clear()
|
||||
logging.info("Changes Saved")
|
||||
if reconnect_required:
|
||||
interface = reconnect_interface_with_splash(stdscr, interface)
|
||||
menu = generate_menu_from_protobuf(interface)
|
||||
menu = generate_menu_from_protobuf(interface, node=node, include_app_settings=not remote)
|
||||
|
||||
if len(menu_state.menu_path) > 1:
|
||||
menu_state.menu_path.pop()
|
||||
@@ -492,7 +503,7 @@ def settings_menu(stdscr: object, interface: object) -> None:
|
||||
continue
|
||||
|
||||
elif selected_option == "Config URL":
|
||||
current_value = interface.localNode.getURL()
|
||||
current_value = node.getURL()
|
||||
new_value = get_text_input(
|
||||
t(
|
||||
"ui.prompt.config_url_current",
|
||||
@@ -510,7 +521,7 @@ def settings_menu(stdscr: object, interface: object) -> None:
|
||||
["Yes", "No"],
|
||||
)
|
||||
if overwrite == "Yes":
|
||||
interface.localNode.setURL(new_value)
|
||||
node.setURL(new_value)
|
||||
logging.info(f"New Config URL sent to node")
|
||||
menu_state.start_index.pop()
|
||||
continue
|
||||
@@ -521,7 +532,7 @@ def settings_menu(stdscr: object, interface: object) -> None:
|
||||
)
|
||||
if confirmation == "Yes":
|
||||
interface = reconnect_after_admin_action(
|
||||
stdscr, interface, interface.localNode.reboot, "Node Reboot Requested by menu"
|
||||
stdscr, interface, node.reboot, "Node Reboot Requested by menu"
|
||||
)
|
||||
menu = rebuild_menu_at_current_path(interface, menu_state)
|
||||
menu_state.start_index.pop()
|
||||
@@ -535,7 +546,7 @@ def settings_menu(stdscr: object, interface: object) -> None:
|
||||
)
|
||||
if confirmation == "Yes":
|
||||
interface = reconnect_after_admin_action(
|
||||
stdscr, interface, interface.localNode.resetNodeDb, "Node DB Reset Requested by menu"
|
||||
stdscr, interface, node.resetNodeDb, "Node DB Reset Requested by menu"
|
||||
)
|
||||
menu = rebuild_menu_at_current_path(interface, menu_state)
|
||||
menu_state.start_index.pop()
|
||||
@@ -546,7 +557,7 @@ def settings_menu(stdscr: object, interface: object) -> None:
|
||||
t("ui.confirm.shutdown", default="Are you sure you want to Shutdown?"), None, ["Yes", "No"]
|
||||
)
|
||||
if confirmation == "Yes":
|
||||
interface.localNode.shutdown()
|
||||
node.shutdown()
|
||||
logging.info(f"Node Shutdown Requested by menu")
|
||||
menu_state.start_index.pop()
|
||||
continue
|
||||
@@ -561,7 +572,7 @@ def settings_menu(stdscr: object, interface: object) -> None:
|
||||
interface = reconnect_after_admin_action(
|
||||
stdscr,
|
||||
interface,
|
||||
lambda: request_factory_reset(interface.localNode, full=True),
|
||||
lambda: request_factory_reset(node, full=True),
|
||||
"Factory Reset Requested by menu",
|
||||
)
|
||||
menu = rebuild_menu_at_current_path(interface, menu_state)
|
||||
@@ -578,7 +589,7 @@ def settings_menu(stdscr: object, interface: object) -> None:
|
||||
interface = reconnect_after_admin_action(
|
||||
stdscr,
|
||||
interface,
|
||||
lambda: request_factory_reset(interface.localNode, full=False),
|
||||
lambda: request_factory_reset(node, full=False),
|
||||
"Factory Reset Config Requested by menu",
|
||||
)
|
||||
menu = rebuild_menu_at_current_path(interface, menu_state)
|
||||
@@ -613,8 +624,8 @@ def settings_menu(stdscr: object, interface: object) -> None:
|
||||
is_ringtone = selected_option == "ringtone"
|
||||
getter_name = "get_ringtone" if is_ringtone else "get_canned_message"
|
||||
setter_name = "set_ringtone" if is_ringtone else "set_canned_message"
|
||||
getter = getattr(interface.localNode, getter_name, None)
|
||||
setter = getattr(interface.localNode, setter_name, None)
|
||||
getter = getattr(node, getter_name, None)
|
||||
setter = getattr(node, setter_name, None)
|
||||
fetched_value = getter() if callable(getter) else None
|
||||
current_value = fetched_value if fetched_value is not None else current_value
|
||||
new_value = get_text_input(
|
||||
|
||||
+15
-13
@@ -57,14 +57,15 @@ def extract_fields(
|
||||
return menu
|
||||
|
||||
|
||||
def generate_menu_from_protobuf(interface: object) -> Dict[str, Any]:
|
||||
def generate_menu_from_protobuf(interface: object, node: Any = None, include_app_settings: bool = True) -> Dict[str, Any]:
|
||||
"""
|
||||
Builds the full settings menu structure from the protobuf definitions.
|
||||
"""
|
||||
menu_structure = {"Main Menu": {}}
|
||||
|
||||
# Add User Settings
|
||||
current_node_info = interface.getMyNodeInfo() if interface else None
|
||||
node = node or (interface.localNode if interface else None)
|
||||
current_node_info = interface.getMyNodeInfo() if interface and node is interface.localNode else None
|
||||
|
||||
if current_node_info:
|
||||
current_user_config = current_node_info.get("user", None)
|
||||
@@ -84,23 +85,23 @@ def generate_menu_from_protobuf(interface: object) -> Dict[str, Any]:
|
||||
# Add Channels
|
||||
channel = channel_pb2.ChannelSettings()
|
||||
menu_structure["Main Menu"]["Channels"] = {}
|
||||
if interface:
|
||||
if node:
|
||||
for i in range(8):
|
||||
current_channel = interface.localNode.getChannelByChannelIndex(i)
|
||||
current_channel = node.getChannelByChannelIndex(i)
|
||||
if current_channel:
|
||||
channel_config = extract_fields(channel, current_channel.settings)
|
||||
menu_structure["Main Menu"]["Channels"][f"Channel {i + 1}"] = channel_config
|
||||
|
||||
# Add Radio Settings
|
||||
radio = config_pb2.Config()
|
||||
current_radio_config = interface.localNode.localConfig if interface else None
|
||||
current_radio_config = node.localConfig if node else None
|
||||
menu_structure["Main Menu"]["Radio Settings"] = extract_fields(radio, current_radio_config)
|
||||
|
||||
# Add Lat/Lon/Alt
|
||||
position_data = {
|
||||
"latitude": (None, current_node_info["position"].get("latitude", 0.0)),
|
||||
"longitude": (None, current_node_info["position"].get("longitude", 0.0)),
|
||||
"altitude": (None, current_node_info["position"].get("altitude", 0)),
|
||||
"latitude": (None, (current_node_info or {}).get("position", {}).get("latitude", 0.0)),
|
||||
"longitude": (None, (current_node_info or {}).get("position", {}).get("longitude", 0.0)),
|
||||
"altitude": (None, (current_node_info or {}).get("position", {}).get("altitude", 0)),
|
||||
}
|
||||
|
||||
existing_position_menu = menu_structure["Main Menu"]["Radio Settings"].get("position", {})
|
||||
@@ -117,23 +118,24 @@ def generate_menu_from_protobuf(interface: object) -> Dict[str, Any]:
|
||||
|
||||
# Add Module Settings
|
||||
module = module_config_pb2.ModuleConfig()
|
||||
current_module_config = interface.localNode.moduleConfig if interface else None
|
||||
current_module_config = node.moduleConfig if node else None
|
||||
module_settings = extract_fields(module, current_module_config)
|
||||
if interface and interface.localNode:
|
||||
if node:
|
||||
# These values use dedicated admin requests rather than ModuleConfig
|
||||
# fields, so expose them alongside their related module settings.
|
||||
module_settings.setdefault("external_notification", {})["ringtone"] = (
|
||||
None,
|
||||
getattr(interface.localNode, "ringtone", "") or "",
|
||||
getattr(node, "ringtone", "") or "",
|
||||
)
|
||||
module_settings.setdefault("canned_message", {})["messages"] = (
|
||||
None,
|
||||
getattr(interface.localNode, "cannedPluginMessage", "") or "",
|
||||
getattr(node, "cannedPluginMessage", "") or "",
|
||||
)
|
||||
menu_structure["Main Menu"]["Module Settings"] = module_settings
|
||||
|
||||
# Add App Settings
|
||||
menu_structure["Main Menu"]["App Settings"] = {"Open": "app_settings"}
|
||||
if include_app_settings:
|
||||
menu_structure["Main Menu"]["App Settings"] = {"Open": "app_settings"}
|
||||
|
||||
# Additional settings options
|
||||
menu_structure["Main Menu"].update(
|
||||
|
||||
@@ -78,7 +78,7 @@ def _requires_reconnect(menu_state, modified_settings) -> bool:
|
||||
return True
|
||||
|
||||
|
||||
def save_changes(interface, modified_settings, menu_state):
|
||||
def save_changes(interface, modified_settings, menu_state, node=None):
|
||||
"""
|
||||
Save changes to the device based on modified settings.
|
||||
:param interface: Meshtastic interface instance
|
||||
@@ -90,7 +90,7 @@ def save_changes(interface, modified_settings, menu_state):
|
||||
logging.info("No changes to save. modified_settings is empty.")
|
||||
return False
|
||||
|
||||
node = interface.getNode("^local")
|
||||
node = node or interface.getNode("^local")
|
||||
admin_key_backup = None
|
||||
if "admin_key" in modified_settings:
|
||||
# Get reference to security config
|
||||
@@ -134,7 +134,7 @@ def save_changes(interface, modified_settings, menu_state):
|
||||
lon = float(modified_settings.get("longitude", 0.0))
|
||||
alt = int(modified_settings.get("altitude", 0))
|
||||
|
||||
interface.localNode.setFixedPosition(lat, lon, alt)
|
||||
node.setFixedPosition(lat, lon, alt)
|
||||
logging.info(f"Updated {config_category} with Latitude: {lat} and Longitude {lon} and Altitude {alt}")
|
||||
return False
|
||||
|
||||
|
||||
@@ -39,6 +39,41 @@ class ContactUiTests(unittest.TestCase):
|
||||
self.assertEqual(curs_set.call_args_list[-1].args, (1,))
|
||||
self.assertEqual(ui_state.current_window, 1)
|
||||
|
||||
def test_handle_backtick_shows_error_when_remote_admin_is_rejected(self) -> None:
|
||||
stdscr = mock.Mock()
|
||||
ui_state.current_window = 2
|
||||
ui_state.node_list = [123]
|
||||
ui_state.selected_node = 0
|
||||
contact_ui.interface_state.interface = mock.Mock()
|
||||
contact_ui.interface_state.interface.getNode.return_value = mock.Mock()
|
||||
|
||||
with mock.patch.object(contact_ui.curses, "curs_set"):
|
||||
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"):
|
||||
with mock.patch.object(contact_ui, "get_list_input", return_value="Remote admin: node"):
|
||||
with mock.patch.object(contact_ui, "get_name_from_database", return_value="node"):
|
||||
with mock.patch.object(contact_ui, "settings_menu", side_effect=SystemExit(1)):
|
||||
with mock.patch("contact.ui.dialog.dialog") as dialog:
|
||||
contact_ui.handle_backtick(stdscr)
|
||||
|
||||
dialog.assert_called_once_with(
|
||||
"Remote admin rejected",
|
||||
"The selected node did not authorize this admin request.",
|
||||
)
|
||||
|
||||
def test_show_remote_admin_wait_draws_status(self) -> None:
|
||||
stdscr = mock.Mock()
|
||||
stdscr.getmaxyx.return_value = (24, 100)
|
||||
wait_win = mock.Mock()
|
||||
|
||||
with mock.patch.object(contact_ui.curses, "newwin", return_value=wait_win):
|
||||
result = contact_ui.show_remote_admin_wait(stdscr, "Remote")
|
||||
|
||||
self.assertIs(result, wait_win)
|
||||
self.assertIn("Attempting remote admin of Remote", wait_win.addstr.call_args_list[-1].args[2])
|
||||
wait_win.refresh.assert_called_once()
|
||||
|
||||
def test_process_pending_ui_updates_draws_requested_windows(self) -> None:
|
||||
stdscr = mock.Mock()
|
||||
ui_state.redraw_channels = True
|
||||
|
||||
Reference in New Issue
Block a user