mirror of
https://github.com/pdxlocations/contact.git
synced 2026-10-31 00:01:57 +01:00
Add tests and demo screenshot
This commit is contained in:
1 parent
286b5a531b
commit
6c5ae3b168
10 files changed
+435
-4
No files matched your search
+23
-4
@@ -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()
|
||||
|
||||
@@ -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
|
||||
@@ -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"),
|
||||
],
|
||||
}
|
||||
@@ -0,0 +1 @@
|
||||
|
||||
@@ -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)
|
||||
@@ -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)
|
||||
@@ -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")
|
||||
@@ -0,0 +1,29 @@
|
||||
from argparse import Namespace
|
||||
import unittest
|
||||
from unittest import mock
|
||||
|
||||
import contact.__main__ as entrypoint
|
||||
|
||||
|
||||
class MainRuntimeTests(unittest.TestCase):
|
||||
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)
|
||||
@@ -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__)
|
||||
@@ -0,0 +1,49 @@
|
||||
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_node_list
|
||||
|
||||
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"),
|
||||
],
|
||||
)
|
||||
Reference in new issue
Block a user