mirror of
https://github.com/pdxlocations/contact.git
synced 2026-03-28 17:12:35 +01:00
full test suite
This commit is contained in:
pdxlocations
21
tests/test_config_io.py
Normal file
21
tests/test_config_io.py
Normal file
@@ -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()))
|
||||
15
tests/test_control_utils.py
Normal file
15
tests/test_control_utils.py
Normal file
@@ -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"])
|
||||
121
tests/test_db_handler.py
Normal file
121
tests/test_db_handler.py
Normal file
@@ -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")
|
||||
38
tests/test_default_config.py
Normal file
38
tests/test_default_config.py
Normal file
@@ -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)
|
||||
57
tests/test_i18n.py
Normal file
57
tests/test_i18n.py
Normal file
@@ -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)
|
||||
40
tests/test_ini_utils.py
Normal file
40
tests/test_ini_utils.py
Normal file
@@ -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.")
|
||||
@@ -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)
|
||||
|
||||
96
tests/test_rx_handler.py
Normal file
96
tests/test_rx_handler.py
Normal file
@@ -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)
|
||||
27
tests/test_telemetry_beautifier.py
Normal file
27
tests/test_telemetry_beautifier.py
Normal file
@@ -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)
|
||||
107
tests/test_tx_handler.py
Normal file
107
tests/test_tx_handler.py
Normal file
@@ -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])
|
||||
@@ -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), "✉️")
|
||||
|
||||
14
tests/test_validation_rules.py
Normal file
14
tests/test_validation_rules.py
Normal file
@@ -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"), {})
|
||||
Reference in New Issue
Block a user