mirror of
https://github.com/pdxlocations/contact.git
synced 2026-11-02 09:00:32 +01:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
07b9ed22d8 | ||
|
|
bdba69677b | ||
|
|
1f88a37ac1 | ||
|
|
8f33cbc22d |
No files matched your search
+1
-1
@@ -258,7 +258,7 @@ def start() -> None:
|
||||
|
||||
if fatal_error is not None:
|
||||
print("Fatal error:", fatal_error)
|
||||
traceback.print_exc()
|
||||
traceback.print_exception(type(fatal_error), fatal_error, fatal_error.__traceback__)
|
||||
sys.exit(1)
|
||||
|
||||
if interrupted:
|
||||
|
||||
+72
-63
@@ -1134,74 +1134,83 @@ def handle_backtick(stdscr: curses.window) -> None:
|
||||
handle_resize(stdscr, False)
|
||||
return
|
||||
|
||||
options = ["Local settings"]
|
||||
remote_node_num = None
|
||||
selected_node_num = None
|
||||
if ui_state.node_list:
|
||||
remote_node_num = ui_state.node_list[ui_state.selected_node]
|
||||
options.append(f"Remote admin: {get_name_from_database(remote_node_num)}")
|
||||
choice = get_list_input("Open settings for", None, options)
|
||||
if choice == "Local settings":
|
||||
settings_menu(stdscr, interface)
|
||||
elif remote_node_num is not None:
|
||||
wait_win = None
|
||||
selected_node_num = ui_state.node_list[ui_state.selected_node]
|
||||
|
||||
def remote_status(message: Optional[str]) -> None:
|
||||
nonlocal wait_win
|
||||
if message is None:
|
||||
# The local node is already the target for normal settings; do not offer a
|
||||
# misleading remote-admin choice when it is selected in the node pane.
|
||||
if selected_node_num == interface_state.myNodeNum:
|
||||
settings_menu(stdscr, interface)
|
||||
else:
|
||||
options = ["Local settings"]
|
||||
remote_node_num = None
|
||||
if selected_node_num is not None:
|
||||
remote_node_num = selected_node_num
|
||||
options.append(f"Remote admin: {get_name_from_database(remote_node_num)}")
|
||||
choice = get_list_input("Open settings for", None, options)
|
||||
if choice == "Local settings":
|
||||
settings_menu(stdscr, interface)
|
||||
elif remote_node_num is not None:
|
||||
wait_win = None
|
||||
|
||||
def remote_status(message: Optional[str]) -> None:
|
||||
nonlocal wait_win
|
||||
if message is None:
|
||||
if wait_win is not None:
|
||||
wait_win.erase()
|
||||
wait_win.refresh()
|
||||
wait_win = None
|
||||
return
|
||||
if wait_win is None:
|
||||
wait_win = show_remote_admin_wait(stdscr, get_name_from_database(remote_node_num))
|
||||
update_remote_admin_wait(wait_win, message)
|
||||
|
||||
def remote_cancel_requested() -> bool:
|
||||
if wait_win is None:
|
||||
return False
|
||||
try:
|
||||
wait_win.timeout(0)
|
||||
return wait_win.getch() == 27
|
||||
finally:
|
||||
wait_win.timeout(-1)
|
||||
|
||||
try:
|
||||
remote_node = interface.getNode(remote_node_num, False)
|
||||
verify_remote_admin(
|
||||
remote_node,
|
||||
status_callback=remote_status,
|
||||
cancel_callback=remote_cancel_requested,
|
||||
)
|
||||
remote_status(None)
|
||||
settings_menu(
|
||||
stdscr,
|
||||
interface,
|
||||
# Settings requests configs explicitly; avoid getNode's
|
||||
# implicit channel request racing that setup.
|
||||
node=remote_node,
|
||||
remote=True,
|
||||
status_callback=remote_status,
|
||||
cancel_callback=remote_cancel_requested,
|
||||
)
|
||||
except RemoteAdminCancelled:
|
||||
logging.info("Remote admin request cancelled for %s", remote_node_num)
|
||||
remote_status(None)
|
||||
except (SystemExit, PermissionError) as exc:
|
||||
logging.warning("Remote admin was rejected for %s: %s", remote_node_num, exc)
|
||||
if wait_win is not None:
|
||||
wait_win.erase()
|
||||
wait_win.refresh()
|
||||
wait_win = None
|
||||
return
|
||||
if wait_win is None:
|
||||
wait_win = show_remote_admin_wait(stdscr, get_name_from_database(remote_node_num))
|
||||
update_remote_admin_wait(wait_win, message)
|
||||
|
||||
def remote_cancel_requested() -> bool:
|
||||
if wait_win is None:
|
||||
return False
|
||||
try:
|
||||
wait_win.timeout(0)
|
||||
return wait_win.getch() == 27
|
||||
finally:
|
||||
wait_win.timeout(-1)
|
||||
|
||||
try:
|
||||
remote_node = interface.getNode(remote_node_num, False)
|
||||
verify_remote_admin(
|
||||
remote_node,
|
||||
status_callback=remote_status,
|
||||
cancel_callback=remote_cancel_requested,
|
||||
)
|
||||
remote_status(None)
|
||||
settings_menu(
|
||||
stdscr,
|
||||
interface,
|
||||
# Settings requests configs explicitly; avoid getNode's
|
||||
# implicit channel request racing that setup.
|
||||
node=remote_node,
|
||||
remote=True,
|
||||
status_callback=remote_status,
|
||||
cancel_callback=remote_cancel_requested,
|
||||
)
|
||||
except RemoteAdminCancelled:
|
||||
logging.info("Remote admin request cancelled for %s", remote_node_num)
|
||||
remote_status(None)
|
||||
except (SystemExit, PermissionError) as exc:
|
||||
logging.warning("Remote admin was rejected for %s: %s", remote_node_num, exc)
|
||||
if wait_win is not None:
|
||||
wait_win.erase()
|
||||
wait_win.refresh()
|
||||
contact.ui.dialog.dialog(
|
||||
"Remote admin rejected",
|
||||
"The selected node did not authorize this admin request.",
|
||||
)
|
||||
except Exception as exc:
|
||||
logging.exception("Remote admin failed for %s", remote_node_num)
|
||||
if wait_win is not None:
|
||||
wait_win.erase()
|
||||
wait_win.refresh()
|
||||
contact.ui.dialog.dialog("Remote admin failed", str(exc))
|
||||
contact.ui.dialog.dialog(
|
||||
"Remote admin rejected",
|
||||
"The selected node did not authorize this admin request.",
|
||||
)
|
||||
except Exception as exc:
|
||||
logging.exception("Remote admin failed for %s", remote_node_num)
|
||||
if wait_win is not None:
|
||||
wait_win.erase()
|
||||
wait_win.refresh()
|
||||
contact.ui.dialog.dialog("Remote admin failed", str(exc))
|
||||
ui_state.current_window = previous_window
|
||||
ui_state.single_pane_mode = config.single_pane_mode.lower() == "true"
|
||||
curses.curs_set(1)
|
||||
|
||||
@@ -2,7 +2,6 @@ from meshtastic.protobuf import channel_pb2
|
||||
from google.protobuf.message import Message
|
||||
import logging
|
||||
import base64
|
||||
import time
|
||||
|
||||
DEVICE_REBOOT_KEYS = {"button_gpio", "buzzer_gpio", "role", "rebroadcast_mode"}
|
||||
POWER_REBOOT_KEYS = {
|
||||
@@ -100,22 +99,14 @@ def save_changes(interface, modified_settings, menu_state, node=None):
|
||||
# Filter out empty keys
|
||||
valid_keys = [key for key in admin_keys if key and key.strip() and key != b""]
|
||||
|
||||
if not valid_keys:
|
||||
logging.warning("No valid admin keys provided. Skipping admin key update.")
|
||||
else:
|
||||
# Clear existing keys if needed
|
||||
if security_config.admin_key:
|
||||
logging.info("Clearing existing admin keys...")
|
||||
del security_config.admin_key[:]
|
||||
node.writeConfig("security")
|
||||
time.sleep(2) # Give time for device to process
|
||||
|
||||
# Append new keys
|
||||
for key in valid_keys:
|
||||
logging.info(f"Adding admin key: {key}")
|
||||
security_config.admin_key.append(key)
|
||||
node.writeConfig("security")
|
||||
logging.info("Admin keys updated successfully!")
|
||||
# Update the repeated field in memory, then send one complete
|
||||
# security config. Sending an empty list first can revoke the key
|
||||
# authorizing this remote-admin session before a second write can
|
||||
# restore the remaining keys.
|
||||
del security_config.admin_key[:]
|
||||
security_config.admin_key.extend(valid_keys)
|
||||
node.writeConfig("security")
|
||||
logging.info("Admin keys updated successfully (%d key(s)).", len(valid_keys))
|
||||
|
||||
# Backup 'admin_key' before removing it
|
||||
admin_key_backup = modified_settings.get("admin_key", None)
|
||||
|
||||
@@ -1,6 +1,8 @@
|
||||
import datetime
|
||||
import logging
|
||||
import re
|
||||
import time
|
||||
from functools import lru_cache
|
||||
from typing import Optional, Union
|
||||
from google.protobuf.message import DecodeError
|
||||
|
||||
@@ -10,14 +12,41 @@ import contact.ui.default_config as config
|
||||
from contact.utilities.singleton import ui_state, interface_state
|
||||
import contact.utilities.telemetry_beautifier as tb
|
||||
|
||||
NEW_MODEM_PRESET_NAMES = {
|
||||
14: "TINY_FAST",
|
||||
15: "TINY_SLOW",
|
||||
16: "MEDIUM_TURBO",
|
||||
}
|
||||
|
||||
|
||||
@lru_cache(maxsize=None)
|
||||
def _get_modem_preset_name(modem_preset_enum: int) -> str:
|
||||
try:
|
||||
return config_pb2.Config.LoRaConfig.ModemPreset.Name(modem_preset_enum)
|
||||
except ValueError:
|
||||
compatibility_name = NEW_MODEM_PRESET_NAMES.get(modem_preset_enum)
|
||||
if compatibility_name is not None:
|
||||
logging.warning(
|
||||
"Modem preset %s is newer than the installed Meshtastic protobuf schema; "
|
||||
"using compatibility name %s",
|
||||
modem_preset_enum,
|
||||
compatibility_name,
|
||||
)
|
||||
return compatibility_name
|
||||
|
||||
logging.warning(
|
||||
"Modem preset %s is not recognized by the installed Meshtastic protobuf schema",
|
||||
modem_preset_enum,
|
||||
)
|
||||
return f"UNKNOWN_PRESET_{modem_preset_enum}"
|
||||
|
||||
|
||||
def _get_channel_name(device_channel, node):
|
||||
if device_channel.settings.name:
|
||||
return device_channel.settings.name
|
||||
|
||||
lora_config = node.localConfig.lora
|
||||
modem_preset_enum = lora_config.modem_preset
|
||||
modem_preset_string = config_pb2._CONFIG_LORACONFIG_MODEMPRESET.values_by_number[modem_preset_enum].name
|
||||
modem_preset_enum = node.localConfig.lora.modem_preset
|
||||
modem_preset_string = _get_modem_preset_name(modem_preset_enum)
|
||||
return convert_to_camel_case(modem_preset_string)
|
||||
|
||||
|
||||
|
||||
+1
-1
@@ -1,6 +1,6 @@
|
||||
[project]
|
||||
name = "contact"
|
||||
version = "1.6.6"
|
||||
version = "1.6.7"
|
||||
description = "This Python curses client for Meshtastic is a terminal-based client designed to manage device settings, enable mesh chat communication, and handle configuration backups and restores."
|
||||
authors = [
|
||||
{name = "Ben Lipsey",email = "ben@pdxlocations.com"}
|
||||
|
||||
@@ -64,6 +64,25 @@ class ContactUiTests(unittest.TestCase):
|
||||
)
|
||||
settings_menu.assert_not_called()
|
||||
|
||||
def test_handle_backtick_opens_local_settings_for_selected_local_node(self) -> None:
|
||||
stdscr = mock.Mock()
|
||||
ui_state.current_window = 2
|
||||
ui_state.node_list = [123]
|
||||
ui_state.selected_node = 0
|
||||
contact_ui.interface_state.myNodeNum = 123
|
||||
contact_ui.interface_state.interface = mock.Mock()
|
||||
|
||||
with mock.patch.object(contact_ui.curses, "curs_set"):
|
||||
with mock.patch.object(contact_ui, "get_list_input") as chooser:
|
||||
with mock.patch.object(contact_ui, "settings_menu") as settings_menu:
|
||||
with mock.patch.object(contact_ui, "get_channels"):
|
||||
with mock.patch.object(contact_ui, "refresh_node_list"):
|
||||
with mock.patch.object(contact_ui, "handle_resize"):
|
||||
contact_ui.handle_backtick(stdscr)
|
||||
|
||||
chooser.assert_not_called()
|
||||
settings_menu.assert_called_once_with(stdscr, contact_ui.interface_state.interface)
|
||||
|
||||
def test_show_remote_admin_wait_draws_status(self) -> None:
|
||||
stdscr = mock.Mock()
|
||||
stdscr.getmaxyx.return_value = (24, 100)
|
||||
|
||||
+5
-2
@@ -240,7 +240,7 @@ class MainRuntimeTests(unittest.TestCase):
|
||||
with mock.patch.object(entrypoint.sys, "argv", ["contact"]):
|
||||
with mock.patch.object(entrypoint.curses, "wrapper", side_effect=RuntimeError("boom")):
|
||||
with mock.patch.object(entrypoint.curses, "endwin") as endwin:
|
||||
with mock.patch.object(entrypoint.traceback, "print_exc") as print_exc:
|
||||
with mock.patch.object(entrypoint.traceback, "print_exception") as print_exception:
|
||||
with mock.patch("builtins.print") as print_mock:
|
||||
with mock.patch.object(entrypoint.sys, "exit", side_effect=SystemExit(1)) as exit_mock:
|
||||
with self.assertRaises(SystemExit) as raised:
|
||||
@@ -248,6 +248,9 @@ class MainRuntimeTests(unittest.TestCase):
|
||||
|
||||
self.assertEqual(raised.exception.code, 1)
|
||||
endwin.assert_called_once_with()
|
||||
print_exc.assert_called_once_with()
|
||||
exception_type, exception, exception_traceback = print_exception.call_args_list[-1].args
|
||||
self.assertIs(exception_type, RuntimeError)
|
||||
self.assertEqual(str(exception), "boom")
|
||||
self.assertIsNotNone(exception_traceback)
|
||||
print_mock.assert_any_call("Fatal error:", mock.ANY)
|
||||
exit_mock.assert_called_once_with(1)
|
||||
@@ -55,6 +55,20 @@ class SaveToRadioTests(unittest.TestCase):
|
||||
|
||||
self.assertTrue(reconnect_required)
|
||||
|
||||
def test_save_changes_replaces_remote_admin_keys_in_one_write(self) -> None:
|
||||
interface, node = self.build_interface()
|
||||
key_one = b"1" * 32
|
||||
key_two = b"2" * 32
|
||||
key_three = b"3" * 32
|
||||
node.localConfig.security.admin_key = [key_one, key_two, key_three]
|
||||
menu_state = SimpleNamespace(menu_path=["Main Menu", "Radio Settings", "Security"])
|
||||
|
||||
reconnect_required = save_changes(interface, {"admin_key": [key_one, key_two]}, menu_state, node=node)
|
||||
|
||||
self.assertTrue(reconnect_required)
|
||||
self.assertEqual(node.localConfig.security.admin_key, [key_one, key_two])
|
||||
node.writeConfig.assert_called_once_with("security")
|
||||
|
||||
def test_save_changes_returns_true_only_for_rebooting_device_fields(self) -> None:
|
||||
interface, node = self.build_interface()
|
||||
menu_state = SimpleNamespace(menu_path=["Main Menu", "Radio Settings", "Device"])
|
||||
|
||||
+50
-1
@@ -1,10 +1,19 @@
|
||||
import unittest
|
||||
from unittest import mock
|
||||
|
||||
from meshtastic.protobuf import config_pb2
|
||||
|
||||
import contact.ui.default_config as config
|
||||
from contact.utilities.demo_data import DEMO_LOCAL_NODE_NUM, build_demo_interface
|
||||
from contact.utilities.singleton import interface_state, ui_state
|
||||
from contact.utilities.utils import add_new_message, get_channels, get_node_list, parse_protobuf
|
||||
from contact.utilities.utils import (
|
||||
_get_channel_name,
|
||||
_get_modem_preset_name,
|
||||
add_new_message,
|
||||
get_channels,
|
||||
get_node_list,
|
||||
parse_protobuf,
|
||||
)
|
||||
|
||||
from tests.test_support import reset_singletons, restore_config, snapshot_config
|
||||
|
||||
@@ -12,6 +21,7 @@ from tests.test_support import reset_singletons, restore_config, snapshot_config
|
||||
class UtilsTests(unittest.TestCase):
|
||||
def setUp(self) -> None:
|
||||
reset_singletons()
|
||||
_get_modem_preset_name.cache_clear()
|
||||
self.saved_config = snapshot_config("node_sort")
|
||||
|
||||
def tearDown(self) -> None:
|
||||
@@ -60,6 +70,45 @@ class UtilsTests(unittest.TestCase):
|
||||
self.assertIn("MediumFast", ui_state.all_messages)
|
||||
self.assertIn("Another Channel", ui_state.all_messages)
|
||||
|
||||
def test_get_channel_name_handles_newer_known_modem_preset(self) -> None:
|
||||
device_channel = mock.Mock()
|
||||
device_channel.settings.name = ""
|
||||
node = mock.Mock()
|
||||
node.localConfig.lora.modem_preset = 14
|
||||
|
||||
with mock.patch("contact.utilities.utils.logging.warning") as warning:
|
||||
self.assertEqual(_get_channel_name(device_channel, node), "TinyFast")
|
||||
|
||||
warning.assert_called_once()
|
||||
|
||||
def test_get_channel_name_handles_unknown_modem_preset(self) -> None:
|
||||
device_channel = mock.Mock()
|
||||
device_channel.settings.name = ""
|
||||
node = mock.Mock()
|
||||
node.localConfig.lora.modem_preset = 99
|
||||
|
||||
with mock.patch("contact.utilities.utils.logging.warning") as warning:
|
||||
self.assertEqual(_get_channel_name(device_channel, node), "UnknownPreset99")
|
||||
|
||||
warning.assert_called_once()
|
||||
|
||||
def test_get_channel_name_prefers_public_protobuf_enum_over_compatibility_map(self) -> None:
|
||||
device_channel = mock.Mock()
|
||||
device_channel.settings.name = ""
|
||||
node = mock.Mock()
|
||||
node.localConfig.lora.modem_preset = 14
|
||||
|
||||
with mock.patch.object(
|
||||
config_pb2.Config.LoRaConfig.ModemPreset,
|
||||
"Name",
|
||||
return_value="TINY_FAST",
|
||||
) as enum_name:
|
||||
with mock.patch("contact.utilities.utils.logging.warning") as warning:
|
||||
self.assertEqual(_get_channel_name(device_channel, node), "TinyFast")
|
||||
|
||||
enum_name.assert_called_once_with(14)
|
||||
warning.assert_not_called()
|
||||
|
||||
def test_get_channels_rebuilds_renamed_channels_and_preserves_messages(self) -> None:
|
||||
interface = build_demo_interface()
|
||||
interface.localNode.channels[0].settings.name = "Renamed Channel"
|
||||
|
||||
Reference in new issue
Block a user