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", "announce_max_stored_lxmf_delivery", "announce_max_stored_nomadnetwork_node", "announce_max_stored_lxmf_propagation", "announce_fetch_limit_lxmf_delivery", "announce_fetch_limit_nomadnetwork_node", "announce_fetch_limit_lxmf_propagation", "announce_search_max_fetch", "discovered_interfaces_max_return", "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 )