From 705b25192cd206d61772c274065f0a43f5676451 Mon Sep 17 00:00:00 2001 From: pdxlocations Date: Thu, 19 Mar 2026 09:49:11 -0700 Subject: [PATCH] full test suite --- tests/test_config_io.py | 21 ++++ tests/test_control_utils.py | 15 +++ tests/test_db_handler.py | 121 +++++++++++++++++++++++ tests/test_default_config.py | 38 ++++++++ tests/test_i18n.py | 57 +++++++++++ tests/test_ini_utils.py | 40 ++++++++ tests/test_main.py | 149 +++++++++++++++++++++++++++++ tests/test_rx_handler.py | 96 +++++++++++++++++++ tests/test_telemetry_beautifier.py | 27 ++++++ tests/test_tx_handler.py | 107 +++++++++++++++++++++ tests/test_utils.py | 24 ++++- tests/test_validation_rules.py | 14 +++ 12 files changed, 708 insertions(+), 1 deletion(-) create mode 100644 tests/test_config_io.py create mode 100644 tests/test_control_utils.py create mode 100644 tests/test_db_handler.py create mode 100644 tests/test_default_config.py create mode 100644 tests/test_i18n.py create mode 100644 tests/test_ini_utils.py create mode 100644 tests/test_rx_handler.py create mode 100644 tests/test_telemetry_beautifier.py create mode 100644 tests/test_tx_handler.py create mode 100644 tests/test_validation_rules.py diff --git a/tests/test_config_io.py b/tests/test_config_io.py new file mode 100644 index 0000000..f354f57 --- /dev/null +++ b/tests/test_config_io.py @@ -0,0 +1,21 @@ +import unittest + +from contact.utilities.config_io import _is_repeated_field, splitCompoundName + + +class ConfigIoTests(unittest.TestCase): + def test_split_compound_name_preserves_multi_part_values(self) -> None: + self.assertEqual(splitCompoundName("config.device.role"), ["config", "device", "role"]) + + def test_split_compound_name_duplicates_single_part_values(self) -> None: + self.assertEqual(splitCompoundName("owner"), ["owner", "owner"]) + + def test_is_repeated_field_prefers_new_style_attribute(self) -> None: + field = type("Field", (), {"is_repeated": True})() + + self.assertTrue(_is_repeated_field(field)) + + def test_is_repeated_field_falls_back_to_label_comparison(self) -> None: + field_type = type("Field", (), {"label": 3, "LABEL_REPEATED": 3}) + + self.assertTrue(_is_repeated_field(field_type())) diff --git a/tests/test_control_utils.py b/tests/test_control_utils.py new file mode 100644 index 0000000..6188cc9 --- /dev/null +++ b/tests/test_control_utils.py @@ -0,0 +1,15 @@ +import unittest + +from contact.utilities.control_utils import transform_menu_path + + +class ControlUtilsTests(unittest.TestCase): + def test_transform_menu_path_applies_replacements_and_normalization(self) -> None: + transformed = transform_menu_path(["Main Menu", "Radio Settings", "Channel 2", "Detail"]) + + self.assertEqual(transformed, ["config", "channel", "Detail"]) + + def test_transform_menu_path_preserves_unmatched_entries(self) -> None: + transformed = transform_menu_path(["Main Menu", "Module Settings", "WiFi"]) + + self.assertEqual(transformed, ["module", "WiFi"]) diff --git a/tests/test_db_handler.py b/tests/test_db_handler.py new file mode 100644 index 0000000..d7ba662 --- /dev/null +++ b/tests/test_db_handler.py @@ -0,0 +1,121 @@ +import os +import sqlite3 +import tempfile +import unittest + +import contact.ui.default_config as config +from contact.utilities import db_handler +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 decimal_to_hex + +from tests.test_support import reset_singletons, restore_config, snapshot_config + + +class DbHandlerTests(unittest.TestCase): + def setUp(self) -> None: + reset_singletons() + self.saved_config = snapshot_config( + "db_file_path", + "message_prefix", + "sent_message_prefix", + "ack_str", + "ack_implicit_str", + "ack_unknown_str", + "nak_str", + ) + self.tempdir = tempfile.TemporaryDirectory() + config.db_file_path = os.path.join(self.tempdir.name, "client.db") + interface_state.myNodeNum = 123 + + def tearDown(self) -> None: + self.tempdir.cleanup() + restore_config(self.saved_config) + reset_singletons() + + def test_save_message_to_db_and_update_ack_roundtrip(self) -> None: + timestamp = db_handler.save_message_to_db("Primary", "123", "hello") + + self.assertIsInstance(timestamp, int) + + db_handler.update_ack_nak("Primary", timestamp, "hello", "Ack") + + with sqlite3.connect(config.db_file_path) as conn: + row = conn.execute('SELECT user_id, message_text, ack_type FROM "123_Primary_messages"').fetchone() + + self.assertEqual(row, ("123", "hello", "Ack")) + + def test_update_node_info_in_db_fills_defaults_and_preserves_existing_values(self) -> None: + db_handler.update_node_info_in_db(999, short_name="ABCD") + + original_long_name = db_handler.get_name_from_database(999, "long") + self.assertTrue(original_long_name.startswith("Meshtastic ")) + self.assertEqual(db_handler.get_name_from_database(999, "short"), "ABCD") + self.assertEqual(db_handler.is_chat_archived(999), 0) + + db_handler.update_node_info_in_db(999, chat_archived=1) + + self.assertEqual(db_handler.get_name_from_database(999, "long"), original_long_name) + self.assertEqual(db_handler.get_name_from_database(999, "short"), "ABCD") + self.assertEqual(db_handler.is_chat_archived(999), 1) + + def test_get_name_from_database_returns_hex_when_user_is_missing(self) -> None: + user_id = 0x1234ABCD + db_handler.ensure_node_table_exists() + + self.assertEqual(db_handler.get_name_from_database(user_id, "short"), decimal_to_hex(user_id)) + self.assertEqual(db_handler.is_chat_archived(user_id), 0) + + def test_load_messages_from_db_populates_channels_and_messages(self) -> None: + db_handler.update_node_info_in_db(123, long_name="Local Node", short_name="ME") + db_handler.update_node_info_in_db(456, long_name="Remote Node", short_name="RM") + db_handler.update_node_info_in_db(789, long_name="Archived", short_name="AR", chat_archived=1) + + db_handler.ensure_table_exists( + '"123_Primary_messages"', + """ + user_id TEXT, + message_text TEXT, + timestamp INTEGER, + ack_type TEXT + """, + ) + db_handler.ensure_table_exists( + '"123_789_messages"', + """ + user_id TEXT, + message_text TEXT, + timestamp INTEGER, + ack_type TEXT + """, + ) + + with sqlite3.connect(config.db_file_path) as conn: + conn.execute('INSERT INTO "123_Primary_messages" VALUES (?, ?, ?, ?)', ("123", "sent", 1700000000, "Ack")) + conn.execute('INSERT INTO "123_Primary_messages" VALUES (?, ?, ?, ?)', ("456", "reply", 1700000001, None)) + conn.execute('INSERT INTO "123_789_messages" VALUES (?, ?, ?, ?)', ("789", "hidden", 1700000002, None)) + conn.commit() + + ui_state.channel_list = [] + ui_state.all_messages = {} + + db_handler.load_messages_from_db() + + self.assertIn("Primary", ui_state.channel_list) + self.assertNotIn(789, ui_state.channel_list) + self.assertIn("Primary", ui_state.all_messages) + self.assertIn(789, ui_state.all_messages) + + messages = ui_state.all_messages["Primary"] + self.assertTrue(messages[0][0].startswith("-- ")) + self.assertTrue(any(config.sent_message_prefix in prefix and config.ack_str in prefix for prefix, _ in messages)) + self.assertTrue(any("RM:" in prefix for prefix, _ in messages)) + self.assertEqual(ui_state.all_messages[789][-1][1], "hidden") + + def test_init_nodedb_inserts_nodes_from_interface(self) -> None: + interface_state.interface = build_demo_interface() + interface_state.myNodeNum = DEMO_LOCAL_NODE_NUM + + db_handler.init_nodedb() + + self.assertEqual(db_handler.get_name_from_database(2701131778, "short"), "SAT2") diff --git a/tests/test_default_config.py b/tests/test_default_config.py new file mode 100644 index 0000000..072d6aa --- /dev/null +++ b/tests/test_default_config.py @@ -0,0 +1,38 @@ +import tempfile +import unittest + +from contact.ui import default_config + + +class DefaultConfigTests(unittest.TestCase): + def test_get_localisation_options_filters_hidden_and_non_ini_files(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + for filename in ("en.ini", "ru.ini", ".hidden.ini", "notes.txt"): + with open(f"{tmpdir}/{filename}", "w", encoding="utf-8") as handle: + handle.write("") + + self.assertEqual(default_config.get_localisation_options(tmpdir), ["en", "ru"]) + + def test_get_localisation_file_normalizes_extensions_and_falls_back_to_english(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + for filename in ("en.ini", "ru.ini"): + with open(f"{tmpdir}/{filename}", "w", encoding="utf-8") as handle: + handle.write("") + + self.assertTrue(default_config.get_localisation_file("RU.ini", tmpdir).endswith("/ru.ini")) + self.assertTrue(default_config.get_localisation_file("missing", tmpdir).endswith("/en.ini")) + + def test_update_dict_only_adds_missing_values(self) -> None: + default = {"theme": "dark", "nested": {"language": "en", "sound": True}} + actual = {"nested": {"language": "ru"}} + + updated = default_config.update_dict(default, actual) + + self.assertTrue(updated) + self.assertEqual(actual, {"theme": "dark", "nested": {"language": "ru", "sound": True}}) + + def test_format_json_single_line_arrays_keeps_arrays_inline(self) -> None: + rendered = default_config.format_json_single_line_arrays({"items": [1, 2], "nested": {"flags": ["a", "b"]}}) + + self.assertIn('"items": [1, 2]', rendered) + self.assertIn('"flags": ["a", "b"]', rendered) diff --git a/tests/test_i18n.py b/tests/test_i18n.py new file mode 100644 index 0000000..60bcfc9 --- /dev/null +++ b/tests/test_i18n.py @@ -0,0 +1,57 @@ +import os +import tempfile +import unittest +from unittest import mock + +import contact.ui.default_config as config +from contact.utilities import i18n + +from tests.test_support import restore_config, snapshot_config + + +class I18nTests(unittest.TestCase): + def setUp(self) -> None: + self.saved_config = snapshot_config("language") + i18n._translations = {} + i18n._language = None + + def tearDown(self) -> None: + restore_config(self.saved_config) + i18n._translations = {} + i18n._language = None + + def test_t_loads_translation_file_and_formats_placeholders(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + translation_file = os.path.join(tmpdir, "xx.ini") + with open(translation_file, "w", encoding="utf-8") as handle: + handle.write('[ui]\n') + handle.write('greeting,"Hello {name}"\n') + + config.language = "xx" + with mock.patch.object(config, "get_localisation_file", return_value=translation_file): + self.assertEqual(i18n.t("ui.greeting", name="Ben"), "Hello Ben") + + def test_t_falls_back_to_default_and_returns_unformatted_text_on_error(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + translation_file = os.path.join(tmpdir, "xx.ini") + with open(translation_file, "w", encoding="utf-8") as handle: + handle.write('[ui]\n') + handle.write('greeting,"Hello {name}"\n') + + config.language = "xx" + with mock.patch.object(config, "get_localisation_file", return_value=translation_file): + self.assertEqual(i18n.t("ui.greeting"), "Hello {name}") + self.assertEqual(i18n.t("ui.missing", default="Fallback"), "Fallback") + self.assertEqual(i18n.t_text("Literal {value}", value=7), "Literal 7") + + def test_loader_cache_is_reused_until_language_changes(self) -> None: + config.language = "en" + + with mock.patch.object(i18n, "parse_ini_file", return_value=({"key": "value"}, {})) as parse_ini_file: + self.assertEqual(i18n.t("key"), "value") + self.assertEqual(i18n.t("key"), "value") + self.assertEqual(parse_ini_file.call_count, 1) + + config.language = "ru" + self.assertEqual(i18n.t("missing", default="fallback"), "fallback") + self.assertEqual(parse_ini_file.call_count, 2) diff --git a/tests/test_ini_utils.py b/tests/test_ini_utils.py new file mode 100644 index 0000000..2557b0f --- /dev/null +++ b/tests/test_ini_utils.py @@ -0,0 +1,40 @@ +import os +import tempfile +import unittest +from unittest import mock + +from contact.utilities.ini_utils import parse_ini_file + + +class IniUtilsTests(unittest.TestCase): + def test_parse_ini_file_reads_sections_fields_and_help_text(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + ini_path = os.path.join(tmpdir, "settings.ini") + with open(ini_path, "w", encoding="utf-8") as handle: + handle.write('; comment\n') + handle.write('[config.device]\n') + handle.write('title,"Device","Device help"\n') + handle.write('name,"Node Name","Node help"\n') + handle.write('empty_help,"Fallback",""\n') + + with mock.patch("contact.utilities.ini_utils.i18n.t", return_value="No help available."): + mapping, help_text = parse_ini_file(ini_path) + + self.assertEqual(mapping["config.device"], "Device") + self.assertEqual(help_text["config.device"], "Device help") + self.assertEqual(mapping["config.device.name"], "Node Name") + self.assertEqual(help_text["config.device.name"], "Node help") + self.assertEqual(help_text["config.device.empty_help"], "No help available.") + + def test_parse_ini_file_uses_builtin_help_fallback_when_i18n_fails(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + ini_path = os.path.join(tmpdir, "settings.ini") + with open(ini_path, "w", encoding="utf-8") as handle: + handle.write('[section]\n') + handle.write('name,"Name"\n') + + with mock.patch("contact.utilities.ini_utils.i18n.t", side_effect=RuntimeError("boom")): + mapping, help_text = parse_ini_file(ini_path) + + self.assertEqual(mapping["section.name"], "Name") + self.assertEqual(help_text["section.name"], "No help available.") diff --git a/tests/test_main.py b/tests/test_main.py index 180e4fc..7ea838a 100644 --- a/tests/test_main.py +++ b/tests/test_main.py @@ -3,9 +3,21 @@ import unittest from unittest import mock import contact.__main__ as entrypoint +import contact.ui.default_config as config +from contact.utilities.singleton import interface_state, ui_state + +from tests.test_support import reset_singletons, restore_config, snapshot_config class MainRuntimeTests(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_initialize_runtime_interface_uses_demo_branch(self) -> None: args = Namespace(demo_screenshot=True) @@ -27,3 +39,140 @@ class MainRuntimeTests(unittest.TestCase): self.assertEqual(result, "live-interface") initialize_interface.assert_called_once_with(args) + + def test_prompt_region_if_unset_reinitializes_interface_after_confirmation(self) -> None: + args = Namespace() + old_interface = mock.Mock() + new_interface = 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) + + set_region.assert_called_once_with(old_interface) + old_interface.close.assert_called_once_with() + initialize.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: + args = Namespace() + interface = mock.Mock() + interface_state.interface = interface + + 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: + entrypoint.prompt_region_if_unset(args) + + set_region.assert_not_called() + initialize.assert_not_called() + interface.close.assert_not_called() + self.assertIs(interface_state.interface, interface) + + def test_initialize_globals_resets_and_populates_runtime_state(self) -> None: + ui_state.channel_list = ["stale"] + ui_state.all_messages = {"stale": [("old", "message")]} + ui_state.notifications = [1] + ui_state.packet_buffer = ["packet"] + ui_state.node_list = [99] + ui_state.selected_channel = 3 + ui_state.selected_message = 4 + ui_state.selected_node = 5 + ui_state.start_index = [9, 9, 9] + config.single_pane_mode = "True" + + with mock.patch.object(entrypoint, "get_nodeNum", return_value=123): + with mock.patch.object(entrypoint, "get_channels", return_value=["Primary"]) as get_channels: + with mock.patch.object(entrypoint, "get_node_list", return_value=[123, 456]) as get_node_list: + with mock.patch.object(entrypoint.pub, "subscribe") as subscribe: + with mock.patch.object(entrypoint, "init_nodedb") as init_nodedb: + with mock.patch.object(entrypoint, "seed_demo_messages") as seed_demo_messages: + with mock.patch.object(entrypoint, "load_messages_from_db") as load_messages: + entrypoint.initialize_globals(seed_demo=True) + + self.assertEqual(ui_state.channel_list, ["Primary"]) + self.assertEqual(ui_state.all_messages, {}) + self.assertEqual(ui_state.notifications, []) + self.assertEqual(ui_state.packet_buffer, []) + self.assertEqual(ui_state.node_list, [123, 456]) + self.assertEqual(ui_state.selected_channel, 0) + self.assertEqual(ui_state.selected_message, 0) + self.assertEqual(ui_state.selected_node, 0) + self.assertEqual(ui_state.start_index, [0, 0, 0]) + self.assertTrue(ui_state.single_pane_mode) + self.assertEqual(interface_state.myNodeNum, 123) + get_channels.assert_called_once_with() + get_node_list.assert_called_once_with() + subscribe.assert_called_once_with(entrypoint.on_receive, "meshtastic.receive") + init_nodedb.assert_called_once_with() + seed_demo_messages.assert_called_once_with() + load_messages.assert_called_once_with() + + def test_ensure_min_rows_retries_until_terminal_is_large_enough(self) -> None: + stdscr = mock.Mock() + stdscr.getmaxyx.side_effect = [(10, 80), (11, 80)] + + with mock.patch.object(entrypoint, "dialog") as dialog: + with mock.patch.object(entrypoint.curses, "update_lines_cols") as update_lines_cols: + entrypoint.ensure_min_rows(stdscr, min_rows=11) + + dialog.assert_called_once() + update_lines_cols.assert_called_once_with() + stdscr.clear.assert_called_once_with() + stdscr.refresh.assert_called_once_with() + + def test_start_prints_help_and_exits_zero(self) -> None: + parser = mock.Mock() + + with mock.patch.object(entrypoint.sys, "argv", ["contact", "--help"]): + with mock.patch.object(entrypoint, "setup_parser", return_value=parser): + with mock.patch.object(entrypoint.sys, "exit", side_effect=SystemExit(0)) as exit_mock: + with self.assertRaises(SystemExit) as raised: + entrypoint.start() + + self.assertEqual(raised.exception.code, 0) + parser.print_help.assert_called_once_with() + exit_mock.assert_called_once_with(0) + + def test_start_runs_curses_wrapper_and_closes_interface(self) -> None: + interface = mock.Mock() + interface_state.interface = interface + + with mock.patch.object(entrypoint.sys, "argv", ["contact"]): + with mock.patch.object(entrypoint.curses, "wrapper") as wrapper: + entrypoint.start() + + wrapper.assert_called_once_with(entrypoint.main) + interface.close.assert_called_once_with() + + def test_start_handles_keyboard_interrupt(self) -> None: + interface = mock.Mock() + interface_state.interface = interface + + with mock.patch.object(entrypoint.sys, "argv", ["contact"]): + with mock.patch.object(entrypoint.curses, "wrapper", side_effect=KeyboardInterrupt): + with mock.patch.object(entrypoint.sys, "exit", side_effect=SystemExit(0)) as exit_mock: + with self.assertRaises(SystemExit) as raised: + entrypoint.start() + + self.assertEqual(raised.exception.code, 0) + interface.close.assert_called_once_with() + exit_mock.assert_called_once_with(0) + + def test_start_handles_fatal_exception_and_exits_one(self) -> None: + 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("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: + entrypoint.start() + + self.assertEqual(raised.exception.code, 1) + endwin.assert_called_once_with() + print_exc.assert_called_once_with() + print_mock.assert_any_call("Fatal error:", mock.ANY) + exit_mock.assert_called_once_with(1) diff --git a/tests/test_rx_handler.py b/tests/test_rx_handler.py new file mode 100644 index 0000000..9ddd0dc --- /dev/null +++ b/tests/test_rx_handler.py @@ -0,0 +1,96 @@ +import unittest +from unittest import mock + +import contact.ui.default_config as config +from contact.message_handlers import rx_handler +from contact.utilities.singleton import interface_state, menu_state, ui_state + +from tests.test_support import reset_singletons, restore_config, snapshot_config + + +class RxHandlerTests(unittest.TestCase): + def setUp(self) -> None: + reset_singletons() + self.saved_config = snapshot_config("notification_sound", "message_prefix") + config.notification_sound = "False" + + def tearDown(self) -> None: + restore_config(self.saved_config) + reset_singletons() + + def test_on_receive_text_message_refreshes_selected_channel(self) -> None: + interface_state.myNodeNum = 111 + ui_state.channel_list = ["Primary"] + ui_state.all_messages = {"Primary": []} + ui_state.selected_channel = 0 + + packet = { + "from": 222, + "to": 999, + "channel": 0, + "hopStart": 3, + "hopLimit": 1, + "decoded": {"portnum": "TEXT_MESSAGE_APP", "payload": b"hello"}, + } + + 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) + + draw_node_list.assert_called_once_with() + draw_messages_window.assert_called_once_with(True) + draw_channel_list.assert_not_called() + 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") + self.assertIn("SAT2:", ui_state.all_messages["Primary"][-1][0]) + self.assertIn("[2]", ui_state.all_messages["Primary"][-1][0]) + + def test_on_receive_direct_message_adds_channel_and_notification(self) -> None: + interface_state.myNodeNum = 111 + ui_state.channel_list = ["Primary"] + ui_state.all_messages = {"Primary": []} + ui_state.selected_channel = 0 + + packet = { + "from": 222, + "to": 111, + "hopStart": 1, + "hopLimit": 1, + "decoded": {"portnum": "TEXT_MESSAGE_APP", "payload": b"dm"}, + } + + 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) + + 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() + 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") + + def test_on_receive_trims_packet_buffer_even_when_packet_is_undecoded(self) -> None: + ui_state.packet_buffer = list(range(25)) + ui_state.display_log = True + ui_state.current_window = 4 + + with mock.patch.object(rx_handler, "draw_packetlog_win") as draw_packetlog_win: + rx_handler.on_receive({"id": "new"}, interface=None) + + draw_packetlog_win.assert_called_once_with() + self.assertEqual(len(ui_state.packet_buffer), 20) + self.assertEqual(ui_state.packet_buffer[-1], {"id": "new"}) + self.assertTrue(menu_state.need_redraw) diff --git a/tests/test_telemetry_beautifier.py b/tests/test_telemetry_beautifier.py new file mode 100644 index 0000000..f05b531 --- /dev/null +++ b/tests/test_telemetry_beautifier.py @@ -0,0 +1,27 @@ +import unittest +from unittest import mock + +from contact.utilities.telemetry_beautifier import get_chunks, humanize_wind_direction + + +class TelemetryBeautifierTests(unittest.TestCase): + def test_humanize_wind_direction_handles_boundaries(self) -> None: + self.assertEqual(humanize_wind_direction(0), "N") + self.assertEqual(humanize_wind_direction(90), "E") + self.assertEqual(humanize_wind_direction(225), "SW") + self.assertIsNone(humanize_wind_direction(-1)) + + def test_get_chunks_formats_known_and_unknown_values(self) -> None: + rendered = get_chunks("uptime_seconds:7200\nwind_direction:90\nlatitude_i:123456789\nunknown:abc\n") + + self.assertIn("🆙 2.0h", rendered) + self.assertIn("⮆ E", rendered) + self.assertIn("🌍 12.345679", rendered) + self.assertIn("unknown:abc", rendered) + + def test_get_chunks_formats_time_values(self) -> None: + with mock.patch("contact.utilities.telemetry_beautifier.datetime.datetime") as mocked_datetime: + mocked_datetime.fromtimestamp.return_value.strftime.return_value = "01.01.1970 00:00" + rendered = get_chunks("time:0\n") + + self.assertIn("🕔 01.01.1970 00:00", rendered) diff --git a/tests/test_tx_handler.py b/tests/test_tx_handler.py new file mode 100644 index 0000000..5e9e348 --- /dev/null +++ b/tests/test_tx_handler.py @@ -0,0 +1,107 @@ +from types import SimpleNamespace +import unittest +from unittest import mock + +from meshtastic import BROADCAST_NUM + +import contact.ui.default_config as config +from contact.message_handlers import tx_handler +from contact.utilities.singleton import interface_state, ui_state + +from tests.test_support import reset_singletons, restore_config, snapshot_config + + +class TxHandlerTests(unittest.TestCase): + def setUp(self) -> None: + reset_singletons() + tx_handler.ack_naks.clear() + self.saved_config = snapshot_config("sent_message_prefix", "ack_str", "ack_implicit_str", "nak_str", "ack_unknown_str") + + def tearDown(self) -> None: + tx_handler.ack_naks.clear() + restore_config(self.saved_config) + reset_singletons() + + def test_send_message_on_named_channel_tracks_ack_request(self) -> None: + interface = mock.Mock() + interface.sendText.return_value = SimpleNamespace(id="req-1") + interface_state.interface = interface + interface_state.myNodeNum = 111 + ui_state.channel_list = ["Primary"] + ui_state.all_messages = {"Primary": []} + + with mock.patch.object(tx_handler, "save_message_to_db", return_value=999) as save_message_to_db: + with mock.patch("contact.message_handlers.tx_handler.time.strftime", return_value="[00:00:00] "): + tx_handler.send_message("hello", channel=0) + + interface.sendText.assert_called_once_with( + text="hello", + destinationId=BROADCAST_NUM, + wantAck=True, + wantResponse=False, + onResponse=tx_handler.onAckNak, + channelIndex=0, + ) + save_message_to_db.assert_called_once_with("Primary", 111, "hello") + self.assertEqual(tx_handler.ack_naks["req-1"]["channel"], "Primary") + self.assertEqual(tx_handler.ack_naks["req-1"]["messageIndex"], 1) + self.assertEqual(tx_handler.ack_naks["req-1"]["timestamp"], 999) + self.assertEqual(ui_state.all_messages["Primary"][-1][1], "hello") + + def test_send_message_to_direct_node_uses_node_as_destination(self) -> None: + interface = mock.Mock() + interface.sendText.return_value = SimpleNamespace(id="req-2") + interface_state.interface = interface + interface_state.myNodeNum = 111 + ui_state.channel_list = [222] + ui_state.all_messages = {222: []} + + with mock.patch.object(tx_handler, "save_message_to_db", return_value=123): + with mock.patch("contact.message_handlers.tx_handler.time.strftime", return_value="[00:00:00] "): + tx_handler.send_message("dm", channel=0) + + interface.sendText.assert_called_once_with( + text="dm", + destinationId=222, + wantAck=True, + wantResponse=False, + onResponse=tx_handler.onAckNak, + channelIndex=0, + ) + self.assertEqual(tx_handler.ack_naks["req-2"]["channel"], 222) + + def test_on_ack_nak_updates_message_for_explicit_ack(self) -> None: + interface_state.myNodeNum = 111 + ui_state.channel_list = ["Primary"] + ui_state.selected_channel = 0 + ui_state.all_messages = {"Primary": [("pending", "hello")]} + tx_handler.ack_naks["req"] = {"channel": "Primary", "messageIndex": 0, "timestamp": 55} + + packet = {"from": 222, "decoded": {"requestId": "req", "routing": {"errorReason": "NONE"}}} + + 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: + tx_handler.onAckNak(packet) + + update_ack_nak.assert_called_once_with("Primary", 55, "hello", "Ack") + draw_messages_window.assert_called_once_with() + 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]) + + def test_on_ack_nak_uses_implicit_marker_for_self_ack(self) -> None: + interface_state.myNodeNum = 111 + ui_state.channel_list = ["Primary"] + ui_state.selected_channel = 0 + ui_state.all_messages = {"Primary": [("pending", "hello")]} + tx_handler.ack_naks["req"] = {"channel": "Primary", "messageIndex": 0, "timestamp": 55} + + packet = {"from": 111, "decoded": {"requestId": "req", "routing": {"errorReason": "NONE"}}} + + 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"): + tx_handler.onAckNak(packet) + + update_ack_nak.assert_called_once_with("Primary", 55, "hello", "Implicit") + self.assertIn(config.ack_implicit_str, ui_state.all_messages["Primary"][0][0]) diff --git a/tests/test_utils.py b/tests/test_utils.py index 4aec39b..8965bc5 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -4,7 +4,7 @@ from unittest import mock 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_node_list +from contact.utilities.utils import add_new_message, get_channels, get_node_list, parse_protobuf from tests.test_support import reset_singletons, restore_config, snapshot_config @@ -47,3 +47,25 @@ class UtilsTests(unittest.TestCase): ("[00:16:40] >> Test: ", "Second"), ], ) + + def test_get_channels_populates_message_buckets_for_device_channels(self) -> None: + interface_state.interface = build_demo_interface() + ui_state.channel_list = [] + ui_state.all_messages = {} + + channels = get_channels() + + self.assertIn("MediumFast", channels) + self.assertIn("Another Channel", channels) + self.assertIn("MediumFast", ui_state.all_messages) + self.assertIn("Another Channel", ui_state.all_messages) + + def test_parse_protobuf_returns_string_payload_unchanged(self) -> None: + packet = {"decoded": {"portnum": "TEXT_MESSAGE_APP", "payload": "hello"}} + + self.assertEqual(parse_protobuf(packet), "hello") + + def test_parse_protobuf_returns_placeholder_for_text_messages(self) -> None: + packet = {"decoded": {"portnum": "TEXT_MESSAGE_APP", "payload": b"hello"}} + + self.assertEqual(parse_protobuf(packet), "✉️") diff --git a/tests/test_validation_rules.py b/tests/test_validation_rules.py new file mode 100644 index 0000000..265e0f6 --- /dev/null +++ b/tests/test_validation_rules.py @@ -0,0 +1,14 @@ +import unittest + +from contact.utilities.validation_rules import get_validation_for + + +class ValidationRulesTests(unittest.TestCase): + def test_get_validation_for_matches_exact_keys(self) -> None: + self.assertEqual(get_validation_for("shortName"), {"max_length": 4}) + + def test_get_validation_for_matches_substrings(self) -> None: + self.assertEqual(get_validation_for("config.position.latitude"), {"min_value": -90, "max_value": 90}) + + def test_get_validation_for_returns_empty_dict_for_unknown_key(self) -> None: + self.assertEqual(get_validation_for("totally_unknown"), {})