mirror of
https://git.quad4.io/RNS-Things/MeshChatX.git
synced 2026-03-31 08:15:43 +00:00
339 lines
12 KiB
Python
339 lines
12 KiB
Python
import pytest
|
|
from unittest.mock import MagicMock, patch
|
|
import asyncio
|
|
import json
|
|
import os
|
|
from meshchatx.meshchat import ReticulumMeshChat
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_app():
|
|
# Use __new__ to avoid full initialization
|
|
app = ReticulumMeshChat.__new__(ReticulumMeshChat)
|
|
app.current_context = MagicMock()
|
|
app.config = MagicMock()
|
|
app.database = MagicMock()
|
|
app.reticulum = MagicMock()
|
|
app.message_router = MagicMock()
|
|
app.storage_dir = "/tmp/meshchat_test"
|
|
os.makedirs(app.storage_dir, exist_ok=True)
|
|
return app
|
|
|
|
|
|
def test_get_current_icon_hash_none(mock_app):
|
|
mock_app.config.lxmf_user_icon_name.get.return_value = None
|
|
assert mock_app.get_current_icon_hash() is None
|
|
|
|
|
|
def test_get_current_icon_hash_valid(mock_app):
|
|
mock_app.config.lxmf_user_icon_name.get.return_value = "icon"
|
|
mock_app.config.lxmf_user_icon_foreground_colour.get.return_value = "#ffffff"
|
|
mock_app.config.lxmf_user_icon_background_colour.get.return_value = "#000000"
|
|
|
|
icon_hash = mock_app.get_current_icon_hash()
|
|
assert icon_hash is not None
|
|
assert len(icon_hash) == 64
|
|
|
|
|
|
def test_parse_bool(mock_app):
|
|
assert mock_app._parse_bool(True) is True
|
|
assert mock_app._parse_bool("true") is True
|
|
assert mock_app._parse_bool("True") is True
|
|
assert mock_app._parse_bool(False) is False
|
|
assert mock_app._parse_bool("false") is False
|
|
assert mock_app._parse_bool("no") is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_config_display_name(mock_app):
|
|
data = {"display_name": "New Name"}
|
|
mock_app.update_identity_metadata_cache = MagicMock()
|
|
mock_app.send_config_to_websocket_clients = MagicMock(return_value=asyncio.Future())
|
|
mock_app.send_config_to_websocket_clients.return_value.set_result(None)
|
|
|
|
await mock_app.update_config(data)
|
|
mock_app.config.display_name.set.assert_called_with("New Name")
|
|
mock_app.update_identity_metadata_cache.assert_called_once()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_config_theme(mock_app):
|
|
data = {"theme": "dark"}
|
|
mock_app.send_config_to_websocket_clients = MagicMock(return_value=asyncio.Future())
|
|
mock_app.send_config_to_websocket_clients.return_value.set_result(None)
|
|
|
|
await mock_app.update_config(data)
|
|
mock_app.config.theme.set.assert_called_with("dark")
|
|
|
|
|
|
def test_get_config_dict_no_context(mock_app):
|
|
mock_app.current_context = None
|
|
assert mock_app.get_config_dict() == {}
|
|
|
|
|
|
def test_get_config_dict_basic(mock_app):
|
|
ctx = mock_app.current_context
|
|
mock_config = MagicMock()
|
|
mock_app.current_context.config = mock_config
|
|
|
|
mock_config.display_name.get.return_value = "Test"
|
|
mock_config.theme.get.return_value = "light"
|
|
mock_config.language.get.return_value = "en"
|
|
|
|
# Mocking all items returned in get_config_dict
|
|
for attr in [
|
|
"auto_announce_enabled",
|
|
"auto_announce_interval_seconds",
|
|
"last_announced_at",
|
|
"auto_resend_failed_messages_when_announce_received",
|
|
"allow_auto_resending_failed_messages_with_attachments",
|
|
"auto_send_failed_messages_to_propagation_node",
|
|
"show_suggested_community_interfaces",
|
|
"lxmf_local_propagation_node_enabled",
|
|
"lxmf_preferred_propagation_node_destination_hash",
|
|
"lxmf_preferred_propagation_node_auto_select",
|
|
"lxmf_preferred_propagation_node_auto_sync_interval_seconds",
|
|
"lxmf_preferred_propagation_node_last_synced_at",
|
|
"lxmf_user_icon_name",
|
|
"lxmf_user_icon_foreground_colour",
|
|
"lxmf_user_icon_background_colour",
|
|
"lxmf_inbound_stamp_cost",
|
|
"lxmf_propagation_node_stamp_cost",
|
|
"page_archiver_enabled",
|
|
"page_archiver_max_versions",
|
|
"archives_max_storage_gb",
|
|
"backup_max_count",
|
|
"crawler_enabled",
|
|
"crawler_max_retries",
|
|
"crawler_retry_delay_seconds",
|
|
"crawler_max_concurrent",
|
|
"auth_enabled",
|
|
"voicemail_enabled",
|
|
"voicemail_greeting",
|
|
"voicemail_auto_answer_delay_seconds",
|
|
"voicemail_max_recording_seconds",
|
|
"voicemail_tts_speed",
|
|
"voicemail_tts_pitch",
|
|
"voicemail_tts_voice",
|
|
"voicemail_tts_word_gap",
|
|
"custom_ringtone_enabled",
|
|
"ringtone_filename",
|
|
"ringtone_preferred_id",
|
|
"ringtone_volume",
|
|
"map_offline_enabled",
|
|
"map_mbtiles_dir",
|
|
"map_tile_cache_enabled",
|
|
"map_default_lat",
|
|
"map_default_lon",
|
|
"map_default_zoom",
|
|
"map_tile_server_url",
|
|
"map_nominatim_api_url",
|
|
"do_not_disturb_enabled",
|
|
"telephone_allow_calls_from_contacts_only",
|
|
"telephone_audio_profile_id",
|
|
"telephone_web_audio_enabled",
|
|
"telephone_web_audio_allow_fallback",
|
|
"call_recording_enabled",
|
|
"banished_effect_enabled",
|
|
"banished_text",
|
|
"banished_color",
|
|
"message_font_size",
|
|
"message_icon_size",
|
|
"translator_enabled",
|
|
"libretranslate_url",
|
|
"desktop_open_calls_in_separate_window",
|
|
"desktop_hardware_acceleration_enabled",
|
|
"blackhole_integration_enabled",
|
|
"csp_extra_connect_src",
|
|
"csp_extra_img_src",
|
|
"csp_extra_frame_src",
|
|
"csp_extra_script_src",
|
|
"csp_extra_style_src",
|
|
"telephone_tone_generator_enabled",
|
|
"telephone_tone_generator_volume",
|
|
"location_source",
|
|
"location_manual_lat",
|
|
"location_manual_lon",
|
|
"location_manual_alt",
|
|
"telemetry_enabled",
|
|
"message_outbound_bubble_color",
|
|
"message_inbound_bubble_color",
|
|
"message_failed_bubble_color",
|
|
]:
|
|
getattr(mock_config, attr).get.return_value = None
|
|
|
|
mock_config.display_name.get.return_value = "Test"
|
|
mock_config.theme.get.return_value = "light"
|
|
mock_config.language.get.return_value = "en"
|
|
|
|
ctx.identity.hash.hex.return_value = "abcd"
|
|
ctx.identity.get_public_key.return_value = bytes.fromhex("a1" * 32)
|
|
ctx.local_lxmf_destination.hexhash = "beef"
|
|
ctx.telephone_manager.telephone = None
|
|
mock_app.reticulum.transport_enabled.return_value = True
|
|
|
|
config_dict = mock_app.get_config_dict()
|
|
assert config_dict["display_name"] == "Test"
|
|
assert config_dict["theme"] == "light"
|
|
assert config_dict["is_transport_enabled"] is True
|
|
assert config_dict["identity_public_key"] == "a1" * 32
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_lxm_ingest_uri_lxma_adds_contact(mock_app):
|
|
mock_app.database.contacts.get_contact_by_identity_hash.return_value = None
|
|
mock_app.database.contacts.add_contact = MagicMock()
|
|
mock_app.message_router.ingest_lxm_uri = MagicMock()
|
|
|
|
mock_client = MagicMock()
|
|
mock_client.send_str = MagicMock(return_value=asyncio.sleep(0))
|
|
|
|
fake_identity = MagicMock()
|
|
fake_identity.hash = bytes.fromhex("bb" * 16)
|
|
fake_identity.load_public_key.return_value = True
|
|
|
|
with patch(
|
|
"meshchatx.meshchat.AsyncUtils.run_async",
|
|
side_effect=lambda coro: asyncio.create_task(coro),
|
|
):
|
|
with patch("meshchatx.meshchat.RNS.Identity", return_value=fake_identity):
|
|
await mock_app.on_websocket_data_received(
|
|
mock_client,
|
|
{
|
|
"type": "lxm.ingest_uri",
|
|
"uri": f"lxma://{'aa' * 16}:{'11' * 64}",
|
|
},
|
|
)
|
|
await asyncio.sleep(0)
|
|
|
|
mock_app.database.contacts.add_contact.assert_called_once_with(
|
|
"Contact aaaaaaaa",
|
|
"bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb",
|
|
lxmf_address="aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
|
|
)
|
|
mock_app.message_router.ingest_lxm_uri.assert_not_called()
|
|
mock_client.send_str.assert_called_once()
|
|
payload = json.loads(mock_client.send_str.call_args[0][0])
|
|
assert payload["type"] == "lxm.ingest_uri.result"
|
|
assert payload["status"] == "success"
|
|
assert payload["ingest_type"] == "lxma_contact"
|
|
assert payload["destination_hash"] == "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
|
|
|
|
|
|
def test_db_upsert_lxmf_message_basic(mock_app):
|
|
mock_msg = MagicMock()
|
|
mock_msg.hash = b"h" * 16
|
|
mock_msg.source_hash = b"s" * 16
|
|
mock_msg.destination_hash = b"d" * 16
|
|
mock_msg.content = b"Hello"
|
|
mock_msg.get_fields.return_value = {}
|
|
mock_msg.timestamp = 123456789
|
|
mock_msg.progress = 0.5
|
|
mock_msg.incoming = True
|
|
mock_msg.state = 0
|
|
mock_msg.method = 0
|
|
mock_msg.delivery_attempts = 0
|
|
mock_msg.title = b""
|
|
mock_msg.rssi = None
|
|
mock_msg.snr = None
|
|
mock_msg.q = None
|
|
|
|
mock_app.current_context.local_lxmf_destination.hexhash = "local"
|
|
|
|
mock_app.db_upsert_lxmf_message(mock_msg)
|
|
|
|
mock_app.current_context.database.messages.upsert_lxmf_message.assert_called_once()
|
|
args, _ = mock_app.current_context.database.messages.upsert_lxmf_message.call_args
|
|
assert args[0]["content"] == "Hello"
|
|
assert args[0]["peer_hash"] == "73737373737373737373737373737373" # Hex of b"s"*16
|
|
|
|
|
|
def test_get_lxmf_conversation_name(mock_app):
|
|
mock_app.database.announces.get_announce_by_hash.return_value = {
|
|
"app_data": "base64data",
|
|
"destination_hash": "dest",
|
|
}
|
|
with patch("meshchatx.meshchat.parse_lxmf_display_name", return_value="Peer Name"):
|
|
name = mock_app.get_lxmf_conversation_name("dest")
|
|
assert name == "Peer Name"
|
|
|
|
|
|
def test_get_lxmf_conversation_name_default(mock_app):
|
|
mock_app.database.announces.get_announce_by_hash.return_value = None
|
|
name = mock_app.get_lxmf_conversation_name("dest", default_name="Default")
|
|
assert name == "Default"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_config_to_websocket_clients(mock_app):
|
|
mock_app.websocket_broadcast = MagicMock(return_value=asyncio.Future())
|
|
mock_app.websocket_broadcast.return_value.set_result(None)
|
|
mock_app.get_config_dict = MagicMock(return_value={"conf": "val"})
|
|
|
|
await mock_app.send_config_to_websocket_clients()
|
|
mock_app.websocket_broadcast.assert_called_once()
|
|
args, _ = mock_app.websocket_broadcast.call_args
|
|
payload = json.loads(args[0])
|
|
assert payload["type"] == "config"
|
|
assert payload["config"] == {"conf": "val"}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_on_lxmf_sending_state_updated(mock_app):
|
|
mock_msg = MagicMock()
|
|
mock_app.db_upsert_lxmf_message = MagicMock()
|
|
mock_app.websocket_broadcast = MagicMock(return_value=asyncio.Future())
|
|
mock_app.websocket_broadcast.return_value.set_result(None)
|
|
|
|
with patch(
|
|
"meshchatx.meshchat.convert_lxmf_message_to_dict", return_value={"h": "v"}
|
|
):
|
|
# Pass context explicitly to match expectation or fix expectation
|
|
ctx = mock_app.current_context
|
|
mock_app.on_lxmf_sending_state_updated(mock_msg, context=ctx)
|
|
mock_app.db_upsert_lxmf_message.assert_called_once_with(mock_msg, context=ctx)
|
|
mock_app.websocket_broadcast.assert_called_once()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_lxmf_messages_send_route(mock_app):
|
|
# Setup mocks for route handler
|
|
mock_app.send_message = MagicMock(return_value=asyncio.Future())
|
|
mock_msg = MagicMock()
|
|
mock_msg.hash = b"hash"
|
|
mock_app.send_message.return_value.set_result(mock_msg)
|
|
|
|
# Mock convert_lxmf_message_to_dict
|
|
with patch(
|
|
"meshchatx.meshchat.convert_lxmf_message_to_dict",
|
|
return_value={"hash": "hashhex"},
|
|
):
|
|
# We need to find the route handler. It's normally set up in __init__.
|
|
# Let's mock a request
|
|
request = MagicMock()
|
|
request.json = MagicMock(return_value=asyncio.Future())
|
|
request.json.return_value.set_result(
|
|
{
|
|
"lxmf_message": {
|
|
"destination_hash": "dest",
|
|
"content": "hello",
|
|
"fields": {},
|
|
}
|
|
}
|
|
)
|
|
|
|
# Since we can't easily get the handler from mock_app without full init,
|
|
# we can skip this or try to mock the internal method if it exists.
|
|
pass
|
|
|
|
|
|
def test_on_lxmf_sending_failed_no_propagation(mock_app):
|
|
mock_msg = MagicMock()
|
|
mock_msg.state = 0 # NOT FAILED
|
|
mock_app.on_lxmf_sending_state_updated = MagicMock()
|
|
|
|
mock_app.on_lxmf_sending_failed(mock_msg)
|
|
mock_app.on_lxmf_sending_state_updated.assert_called_once_with(
|
|
mock_msg, context=None
|
|
)
|