full test suite

contact/__main__.py

@@ -32,6 +32,7 @@ from contact.ui.contact_ui import main_ui
 from contact.ui.splash import draw_splash
 from contact.utilities.arg_parser import setup_parser
 from contact.utilities.db_handler import init_nodedb, load_messages_from_db
+from contact.utilities.demo_data import build_demo_interface, configure_demo_database, seed_demo_messages
 from contact.utilities.input_handlers import get_list_input
 from contact.utilities.i18n import t
 from contact.ui.dialog import dialog
@@ -68,9 +69,18 @@ def prompt_region_if_unset(args: object) -> None:
         interface_state.interface = initialize_interface(args)
 
 
-def initialize_globals() -> None:
+def initialize_globals(seed_demo: bool = False) -> None:
     """Initializes interface and shared globals."""
 
+    ui_state.channel_list = []
+    ui_state.all_messages = {}
+    ui_state.notifications = []
+    ui_state.packet_buffer = []
+    ui_state.node_list = []
+    ui_state.selected_channel = 0
+    ui_state.selected_message = 0
+    ui_state.selected_node = 0
+    ui_state.start_index = [0, 0, 0]
     interface_state.myNodeNum = get_nodeNum()
     ui_state.channel_list = get_channels()
     ui_state.node_list = get_node_list()
@@ -78,9 +88,18 @@ def initialize_globals() -> None:
     pub.subscribe(on_receive, "meshtastic.receive")
 
     init_nodedb()
+    if seed_demo:
+        seed_demo_messages()
     load_messages_from_db()
 
 
+def initialize_runtime_interface(args: object):
+    if getattr(args, "demo_screenshot", False):
+        configure_demo_database()
+        return build_demo_interface()
+    return initialize_interface(args)
+
+
 def main(stdscr: curses.window) -> None:
     """Main entry point for the curses UI."""
 
@@ -98,12 +117,12 @@ def main(stdscr: curses.window) -> None:
 
         logging.info("Initializing interface...")
         with app_state.lock:
-            interface_state.interface = initialize_interface(args)
+            interface_state.interface = initialize_runtime_interface(args)
 
-            if interface_state.interface.localNode.localConfig.lora.region == 0:
+            if not getattr(args, "demo_screenshot", False) and interface_state.interface.localNode.localConfig.lora.region == 0:
                 prompt_region_if_unset(args)
 
-            initialize_globals()
+            initialize_globals(seed_demo=getattr(args, "demo_screenshot", False))
             logging.info("Starting main UI")
 
             stdscr.clear()

contact/utilities/arg_parser.py

@@ -35,5 +35,10 @@ def setup_parser() -> ArgumentParser:
     parser.add_argument(
         "--settings", "--set", "--control", "-c", help="Launch directly into the settings", action="store_true"
     )
+    parser.add_argument(
+        "--demo-screenshot",
+        help="Launch with a fake interface and seeded demo data for screenshots/testing.",
+        action="store_true",
+    )
 
     return parser

contact/utilities/demo_data.py

@@ -0,0 +1,226 @@
+import os
+import sqlite3
+import tempfile
+from dataclasses import dataclass
+from typing import Dict, List, Tuple, Union
+
+import contact.ui.default_config as config
+from contact.utilities.db_handler import get_table_name
+from contact.utilities.singleton import interface_state
+
+
+DEMO_DB_FILENAME = "contact_demo_client.db"
+DEMO_LOCAL_NODE_NUM = 0xC0DEC0DE
+DEMO_BASE_TIMESTAMP = 1738717200  # 2025-02-04 17:00:00 UTC
+DEMO_CHANNELS = ["MediumFast", "Another Channel"]
+
+
+@dataclass
+class DemoChannelSettings:
+    name: str
+
+
+@dataclass
+class DemoChannel:
+    role: bool
+    settings: DemoChannelSettings
+
+
+@dataclass
+class DemoLoRaConfig:
+    region: int = 1
+    modem_preset: int = 0
+
+
+@dataclass
+class DemoLocalConfig:
+    lora: DemoLoRaConfig
+
+
+class DemoLocalNode:
+    def __init__(self, interface: "DemoInterface", channels: List[DemoChannel]) -> None:
+        self._interface = interface
+        self.channels = channels
+        self.localConfig = DemoLocalConfig(lora=DemoLoRaConfig())
+
+    def setFavorite(self, node_num: int) -> None:
+        self._interface.nodesByNum[node_num]["isFavorite"] = True
+
+    def removeFavorite(self, node_num: int) -> None:
+        self._interface.nodesByNum[node_num]["isFavorite"] = False
+
+    def setIgnored(self, node_num: int) -> None:
+        self._interface.nodesByNum[node_num]["isIgnored"] = True
+
+    def removeIgnored(self, node_num: int) -> None:
+        self._interface.nodesByNum[node_num]["isIgnored"] = False
+
+    def removeNode(self, node_num: int) -> None:
+        self._interface.nodesByNum.pop(node_num, None)
+
+
+class DemoInterface:
+    def __init__(self, nodes: Dict[int, Dict[str, object]], channels: List[DemoChannel]) -> None:
+        self.nodesByNum = nodes
+        self.nodes = self.nodesByNum
+        self.localNode = DemoLocalNode(self, channels)
+
+    def getMyNodeInfo(self) -> Dict[str, int]:
+        return {"num": DEMO_LOCAL_NODE_NUM}
+
+    def getNode(self, selector: str) -> DemoLocalNode:
+        if selector != "^local":
+            raise KeyError(selector)
+        return self.localNode
+
+    def close(self) -> None:
+        return
+
+
+def build_demo_interface() -> DemoInterface:
+    channels = [DemoChannel(role=True, settings=DemoChannelSettings(name=name)) for name in DEMO_CHANNELS]
+
+    nodes = {
+        DEMO_LOCAL_NODE_NUM: _build_node(
+            DEMO_LOCAL_NODE_NUM,
+            "Meshtastic fb3c",
+            "fb3c",
+            hops=0,
+            snr=13.7,
+            last_heard_offset=5,
+            battery=88,
+            voltage=4.1,
+            favorite=True,
+        ),
+        0xA1000001: _build_node(0xA1000001, "KG7NDX-N2", "N2", hops=1, last_heard_offset=18, battery=79, voltage=4.0),
+        0xA1000002: _build_node(0xA1000002, "Satellite II Repeater", "SAT2", hops=2, last_heard_offset=31),
+        0xA1000003: _build_node(0xA1000003, "Search for Discord/Meshtastic", "DISC", hops=1, last_heard_offset=46),
+        0xA1000004: _build_node(0xA1000004, "K7EOK Mobile", "MOBL", hops=1, last_heard_offset=63, battery=52),
+        0xA1000005: _build_node(0xA1000005, "Turtle", "TRTL", hops=3, last_heard_offset=87),
+        0xA1000006: _build_node(0xA1000006, "CARS Trewvilliger Plaza", "CARS", hops=2, last_heard_offset=121),
+        0xA1000007: _build_node(0xA1000007, "No Hands!", "NHDS", hops=1, last_heard_offset=155),
+        0xA1000008: _build_node(0xA1000008, "McCutie", "MCCU", hops=2, last_heard_offset=211, ignored=True),
+        0xA1000009: _build_node(0xA1000009, "K1PDX", "K1PX", hops=2, last_heard_offset=267),
+        0xA100000A: _build_node(0xA100000A, "Arnold Creek", "ARND", hops=1, last_heard_offset=301),
+        0xA100000B: _build_node(0xA100000B, "Nansen", "NANS", hops=1, last_heard_offset=355),
+        0xA100000C: _build_node(0xA100000C, "Kodin 1", "KOD1", hops=2, last_heard_offset=402),
+        0xA100000D: _build_node(0xA100000D, "PH1", "PH1", hops=3, last_heard_offset=470),
+        0xA100000E: _build_node(0xA100000E, "Luna", "LUNA", hops=1, last_heard_offset=501),
+        0xA100000F: _build_node(0xA100000F, "sputnik1", "SPUT", hops=1, last_heard_offset=550),
+        0xA1000010: _build_node(0xA1000010, "K7EOK Maplewood West", "MAPL", hops=2, last_heard_offset=602),
+        0xA1000011: _build_node(0xA1000011, "KE7YVU 2", "YVU2", hops=2, last_heard_offset=655),
+        0xA1000012: _build_node(0xA1000012, "DNET", "DNET", hops=1, last_heard_offset=702),
+        0xA1000013: _build_node(0xA1000013, "Green Bluff", "GBLF", hops=1, last_heard_offset=780),
+        0xA1000014: _build_node(0xA1000014, "Council Crest Solar", "CCST", hops=2, last_heard_offset=830),
+        0xA1000015: _build_node(0xA1000015, "Meshtastic 61c7", "61c7", hops=1, last_heard_offset=901),
+        0xA1000016: _build_node(0xA1000016, "Bella", "BELA", hops=2, last_heard_offset=950),
+        0xA1000017: _build_node(0xA1000017, "Mojo Solar Base 4f12", "MOJO", hops=1, last_heard_offset=1010, favorite=True),
+    }
+
+    return DemoInterface(nodes=nodes, channels=channels)
+
+
+def configure_demo_database(base_dir: str = "") -> str:
+    if not base_dir:
+        base_dir = tempfile.mkdtemp(prefix="contact_demo_")
+    os.makedirs(base_dir, exist_ok=True)
+
+    db_path = os.path.join(base_dir, DEMO_DB_FILENAME)
+    if os.path.exists(db_path):
+        os.remove(db_path)
+
+    config.db_file_path = db_path
+    return db_path
+
+
+def seed_demo_messages() -> None:
+    schema = """
+        user_id TEXT,
+        message_text TEXT,
+        timestamp INTEGER,
+        ack_type TEXT
+    """
+
+    with sqlite3.connect(config.db_file_path) as db_connection:
+        cursor = db_connection.cursor()
+
+        for channel_name, rows in _demo_messages().items():
+            table_name = get_table_name(channel_name)
+            cursor.execute(f"CREATE TABLE IF NOT EXISTS {table_name} ({schema})")
+            cursor.executemany(
+                f"""
+                INSERT INTO {table_name} (user_id, message_text, timestamp, ack_type)
+                VALUES (?, ?, ?, ?)
+                """,
+                rows,
+            )
+
+        db_connection.commit()
+
+
+def _build_node(
+    node_num: int,
+    long_name: str,
+    short_name: str,
+    *,
+    hops: int,
+    last_heard_offset: int,
+    snr: float = 0.0,
+    battery: int = 0,
+    voltage: float = 0.0,
+    favorite: bool = False,
+    ignored: bool = False,
+) -> Dict[str, object]:
+    node = {
+        "num": node_num,
+        "user": {
+            "longName": long_name,
+            "shortName": short_name,
+            "hwModel": "TBEAM",
+            "role": "CLIENT",
+            "publicKey": f"pk-{node_num:08x}",
+            "isLicensed": True,
+        },
+        "lastHeard": DEMO_BASE_TIMESTAMP + 3600 - last_heard_offset,
+        "hopsAway": hops,
+        "isFavorite": favorite,
+        "isIgnored": ignored,
+    }
+
+    if snr:
+        node["snr"] = snr
+    if battery:
+        node["deviceMetrics"] = {
+            "batteryLevel": battery,
+            "voltage": voltage or 4.0,
+            "uptimeSeconds": 86400 + node_num % 10000,
+            "channelUtilization": 12.5 + (node_num % 7),
+            "airUtilTx": 4.5 + (node_num % 5),
+        }
+
+    if node_num % 3 == 0:
+        node["position"] = {
+            "latitude": 45.5231 + ((node_num % 50) * 0.0001),
+            "longitude": -122.6765 - ((node_num % 50) * 0.0001),
+            "altitude": 85 + (node_num % 20),
+        }
+
+    return node
+
+
+def _demo_messages() -> Dict[Union[str, int], List[Tuple[str, str, int, Union[str, None]]]]:
+    return {
+        "MediumFast": [
+            (str(DEMO_LOCAL_NODE_NUM), "Help, I'm stuck in a ditch!", DEMO_BASE_TIMESTAMP + 45, "Ack"),
+            ("2701131778", "Do you require a alpinist?", DEMO_BASE_TIMESTAMP + 80, None),
+            (str(DEMO_LOCAL_NODE_NUM), "I don't know what that is.", DEMO_BASE_TIMESTAMP + 104, "Implicit"),
+        ],
+        "Another Channel": [
+            ("2701131788", "Weather is holding for the summit push.", DEMO_BASE_TIMESTAMP + 220, None),
+            (str(DEMO_LOCAL_NODE_NUM), "Copy that. Keep me posted.", DEMO_BASE_TIMESTAMP + 260, "Ack"),
+        ],
+        2701131788: [
+            ("2701131788", "Ping me when you are back at the trailhead.", DEMO_BASE_TIMESTAMP + 330, None),
+            (str(DEMO_LOCAL_NODE_NUM), "Will do.", DEMO_BASE_TIMESTAMP + 350, "Ack"),
+        ],
+    }

tests/__init__.py

@@ -0,0 +1 @@
+

tests/test_arg_parser.py

@@ -0,0 +1,13 @@
+import unittest
+
+from contact.utilities.arg_parser import setup_parser
+
+
+class ArgParserTests(unittest.TestCase):
+    def test_demo_screenshot_flag_is_supported(self) -> None:
+        args = setup_parser().parse_args(["--demo-screenshot"])
+        self.assertTrue(args.demo_screenshot)
+
+    def test_demo_screenshot_defaults_to_false(self) -> None:
+        args = setup_parser().parse_args([])
+        self.assertFalse(args.demo_screenshot)

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()))

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"])

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")

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)

tests/test_demo_data.py

@@ -0,0 +1,51 @@
+import tempfile
+import unittest
+from unittest import mock
+
+import contact.__main__ as entrypoint
+import contact.ui.default_config as config
+from contact.utilities.db_handler import get_name_from_database
+from contact.utilities.demo_data import DEMO_CHANNELS, DEMO_LOCAL_NODE_NUM, build_demo_interface, configure_demo_database
+from contact.utilities.singleton import interface_state, ui_state
+
+from tests.test_support import reset_singletons, restore_config, snapshot_config
+
+
+class DemoDataTests(unittest.TestCase):
+    def setUp(self) -> None:
+        reset_singletons()
+        self.saved_config = snapshot_config("db_file_path", "node_sort", "single_pane_mode")
+
+    def tearDown(self) -> None:
+        restore_config(self.saved_config)
+        reset_singletons()
+
+    def test_build_demo_interface_exposes_expected_shape(self) -> None:
+        interface = build_demo_interface()
+
+        self.assertEqual(interface.getMyNodeInfo()["num"], DEMO_LOCAL_NODE_NUM)
+        self.assertEqual([channel.settings.name for channel in interface.getNode("^local").channels], DEMO_CHANNELS)
+        self.assertIn(DEMO_LOCAL_NODE_NUM, interface.nodesByNum)
+
+    def test_initialize_globals_seed_demo_populates_ui_state_and_db(self) -> None:
+        interface_state.interface = build_demo_interface()
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            demo_db_path = configure_demo_database(tmpdir)
+            with mock.patch.object(entrypoint.pub, "subscribe"):
+                entrypoint.initialize_globals(seed_demo=True)
+
+            self.assertEqual(config.db_file_path, demo_db_path)
+            self.assertIn("MediumFast", ui_state.channel_list)
+            self.assertIn("Another Channel", ui_state.channel_list)
+            self.assertIn(2701131788, ui_state.channel_list)
+            self.assertEqual(ui_state.node_list[0], DEMO_LOCAL_NODE_NUM)
+            self.assertEqual(get_name_from_database(2701131778, "short"), "SAT2")
+
+            medium_fast = ui_state.all_messages["MediumFast"]
+            self.assertTrue(medium_fast[0][0].startswith("-- "))
+            self.assertTrue(any(config.sent_message_prefix in prefix and config.ack_str in prefix for prefix, _ in medium_fast))
+            self.assertTrue(any("SAT2:" in prefix for prefix, _ in medium_fast))
+
+            direct_messages = ui_state.all_messages[2701131788]
+            self.assertEqual(len(direct_messages), 3)

tests/test_emoji_utils.py

@@ -0,0 +1,11 @@
+import unittest
+
+from contact.utilities.emoji_utils import normalize_message_text
+
+
+class EmojiUtilsTests(unittest.TestCase):
+    def test_strips_modifiers_from_keycaps_and_skin_tones(self) -> None:
+        self.assertEqual(normalize_message_text("👍🏽 7️⃣"), "👍 7")
+
+    def test_rewrites_flag_emoji_to_country_codes(self) -> None:
+        self.assertEqual(normalize_message_text("🇺🇸 hello 🇩🇪"), "US hello DE")

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)

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.")

tests/test_main.py

@@ -0,0 +1,178 @@
+from argparse import Namespace
+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)
+
+        with mock.patch.object(entrypoint, "configure_demo_database") as configure_demo_database:
+            with mock.patch.object(entrypoint, "build_demo_interface", return_value="demo-interface") as build_demo:
+                with mock.patch.object(entrypoint, "initialize_interface") as initialize_interface:
+                    result = entrypoint.initialize_runtime_interface(args)
+
+        self.assertEqual(result, "demo-interface")
+        configure_demo_database.assert_called_once_with()
+        build_demo.assert_called_once_with()
+        initialize_interface.assert_not_called()
+
+    def test_initialize_runtime_interface_uses_live_branch_without_demo_flag(self) -> None:
+        args = Namespace(demo_screenshot=False)
+
+        with mock.patch.object(entrypoint, "initialize_interface", return_value="live-interface") as initialize_interface:
+            result = entrypoint.initialize_runtime_interface(args)
+
+        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)

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)

tests/test_support.py

@@ -0,0 +1,27 @@
+import threading
+
+import contact.ui.default_config as config
+from contact.ui.ui_state import AppState, ChatUIState, InterfaceState, MenuState
+from contact.utilities.singleton import app_state, interface_state, menu_state, ui_state
+
+
+def reset_singletons() -> None:
+    _reset_instance(ui_state, ChatUIState())
+    _reset_instance(interface_state, InterfaceState())
+    _reset_instance(menu_state, MenuState())
+    _reset_instance(app_state, AppState())
+    app_state.lock = threading.Lock()
+
+
+def restore_config(saved: dict) -> None:
+    for key, value in saved.items():
+        setattr(config, key, value)
+
+
+def snapshot_config(*keys: str) -> dict:
+    return {key: getattr(config, key) for key in keys}
+
+
+def _reset_instance(target: object, replacement: object) -> None:
+    target.__dict__.clear()
+    target.__dict__.update(replacement.__dict__)

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)

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])

tests/test_utils.py

@@ -0,0 +1,71 @@
+import unittest
+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_channels, get_node_list, parse_protobuf
+
+from tests.test_support import reset_singletons, restore_config, snapshot_config
+
+
+class UtilsTests(unittest.TestCase):
+    def setUp(self) -> None:
+        reset_singletons()
+        self.saved_config = snapshot_config("node_sort")
+
+    def tearDown(self) -> None:
+        restore_config(self.saved_config)
+        reset_singletons()
+
+    def test_get_node_list_keeps_local_first_and_ignored_last(self) -> None:
+        config.node_sort = "lastHeard"
+        interface = build_demo_interface()
+        interface_state.interface = interface
+        interface_state.myNodeNum = DEMO_LOCAL_NODE_NUM
+
+        node_list = get_node_list()
+
+        self.assertEqual(node_list[0], DEMO_LOCAL_NODE_NUM)
+        self.assertEqual(node_list[-1], 0xA1000008)
+
+    def test_add_new_message_groups_messages_by_hour(self) -> None:
+        ui_state.all_messages = {"MediumFast": []}
+
+        with mock.patch("contact.utilities.utils.time.time", side_effect=[1000, 1000]):
+            with mock.patch("contact.utilities.utils.time.strftime", return_value="[00:16:40] "):
+                with mock.patch("contact.utilities.utils.datetime.datetime") as mocked_datetime:
+                    mocked_datetime.fromtimestamp.return_value.strftime.return_value = "2025-02-04 17:00"
+                    add_new_message("MediumFast", ">> Test: ", "First")
+                    add_new_message("MediumFast", ">> Test: ", "Second")
+
+        self.assertEqual(
+            ui_state.all_messages["MediumFast"],
+            [
+                ("-- 2025-02-04 17:00 --", ""),
+                ("[00:16:40] >> Test: ", "First"),
+                ("[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), "✉️")

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"), {})