From bc40dcff4e157bd8bd3fc509428b0b34778ab7e6 Mon Sep 17 00:00:00 2001 From: Sudo-Ivan Date: Sun, 4 Jan 2026 12:41:55 -0600 Subject: [PATCH] feat(tests): add comprehensive tests for blackhole integration, RNPath management, and RNStatus handling --- tests/backend/test_blackhole_logic.py | 236 +++++++++++++++++++++++ tests/backend/test_rnpath_logic.py | 121 ++++++++++++ tests/backend/test_rnstatus_blackhole.py | 62 ++++++ 3 files changed, 419 insertions(+) create mode 100644 tests/backend/test_blackhole_logic.py create mode 100644 tests/backend/test_rnpath_logic.py create mode 100644 tests/backend/test_rnstatus_blackhole.py diff --git a/tests/backend/test_blackhole_logic.py b/tests/backend/test_blackhole_logic.py new file mode 100644 index 0000000..8ffadc5 --- /dev/null +++ b/tests/backend/test_blackhole_logic.py @@ -0,0 +1,236 @@ +import shutil +import tempfile +import pytest +import json +from unittest.mock import MagicMock, patch, AsyncMock +from meshchatx.meshchat import ReticulumMeshChat +import RNS +import asyncio + + +@pytest.fixture +def temp_dir(): + dir_path = tempfile.mkdtemp() + yield dir_path + shutil.rmtree(dir_path) + + +@pytest.fixture +def mock_rns_minimal(): + with ( + patch("RNS.Reticulum") as mock_rns, + patch("RNS.Transport"), + patch("LXMF.LXMRouter"), + patch("meshchatx.meshchat.get_file_path", return_value="/tmp/mock_path"), + ): + mock_rns_instance = mock_rns.return_value + mock_rns_instance.configpath = "/tmp/mock_config" + mock_rns_instance.is_connected_to_shared_instance = False + mock_rns_instance.transport_enabled.return_value = True + # Mock the new blackhole methods + mock_rns_instance.blackhole_identity = MagicMock() + mock_rns_instance.unblackhole_identity = MagicMock() + mock_rns_instance.get_blackholed_identities.return_value = {} + + mock_id = MagicMock(spec=RNS.Identity) + mock_id.hash = b"\x00" * 32 + mock_id.hexhash = mock_id.hash.hex() + mock_id.get_private_key.return_value = b"test_private_key" + yield mock_id + + +@pytest.mark.asyncio +async def test_banish_identity_with_blackhole(mock_rns_minimal, temp_dir): + with patch("meshchatx.meshchat.generate_ssl_certificate"): + app_instance = ReticulumMeshChat( + identity=mock_rns_minimal, + storage_dir=temp_dir, + reticulum_config_dir=temp_dir, + ) + + # Ensure blackhole integration is enabled + app_instance.config.blackhole_integration_enabled.set(True) + + # Mock database + app_instance.database = MagicMock() + app_instance.database.announces.get_announce_by_hash.return_value = None + + target_hash = "a" * 32 + + # Mock request + request = MagicMock() + request.json = AsyncMock(return_value={"destination_hash": target_hash}) + + # Find handler + handler = None + for route in app_instance.get_routes(): + if route.path == "/api/v1/blocked-destinations" and route.method == "POST": + handler = route.handler + break + + assert handler is not None + + response = await handler(request) + assert response.status == 200 + + # Verify DB call + app_instance.database.misc.add_blocked_destination.assert_called_with( + target_hash + ) + + # Verify RNS blackhole call + app_instance.reticulum.blackhole_identity.assert_called() + args, kwargs = app_instance.reticulum.blackhole_identity.call_args + assert args[0] == bytes.fromhex(target_hash) + + +@pytest.mark.asyncio +async def test_banish_identity_with_resolution(mock_rns_minimal, temp_dir): + with patch("meshchatx.meshchat.generate_ssl_certificate"): + app_instance = ReticulumMeshChat( + identity=mock_rns_minimal, + storage_dir=temp_dir, + reticulum_config_dir=temp_dir, + ) + + app_instance.config.blackhole_integration_enabled.set(True) + app_instance.database = MagicMock() + + dest_hash = "d" * 32 + ident_hash = "e" * 32 + + # Mock identity resolution + app_instance.database.announces.get_announce_by_hash.return_value = { + "identity_hash": ident_hash + } + + request = MagicMock() + request.json = AsyncMock(return_value={"destination_hash": dest_hash}) + + handler = None + for route in app_instance.get_routes(): + if route.path == "/api/v1/blocked-destinations" and route.method == "POST": + handler = route.handler + break + + await handler(request) + + # Should have blackholed the IDENTITY hash, not the destination hash + app_instance.reticulum.blackhole_identity.assert_called() + args, _ = app_instance.reticulum.blackhole_identity.call_args + assert args[0] == bytes.fromhex(ident_hash) + + +@pytest.mark.asyncio +async def test_banish_identity_disabled_integration(mock_rns_minimal, temp_dir): + with patch("meshchatx.meshchat.generate_ssl_certificate"): + app_instance = ReticulumMeshChat( + identity=mock_rns_minimal, + storage_dir=temp_dir, + reticulum_config_dir=temp_dir, + ) + + # DISABLE blackhole integration + app_instance.config.blackhole_integration_enabled.set(False) + app_instance.database = MagicMock() + + target_hash = "b" * 32 + request = MagicMock() + request.json = AsyncMock(return_value={"destination_hash": target_hash}) + + handler = None + for route in app_instance.get_routes(): + if route.path == "/api/v1/blocked-destinations" and route.method == "POST": + handler = route.handler + break + + await handler(request) + + # DB call should still happen + app_instance.database.misc.add_blocked_destination.assert_called_with( + target_hash + ) + + # RNS blackhole call should NOT happen + app_instance.reticulum.blackhole_identity.assert_not_called() + + +@pytest.mark.asyncio +async def test_lift_banishment(mock_rns_minimal, temp_dir): + with patch("meshchatx.meshchat.generate_ssl_certificate"): + app_instance = ReticulumMeshChat( + identity=mock_rns_minimal, + storage_dir=temp_dir, + reticulum_config_dir=temp_dir, + ) + + app_instance.config.blackhole_integration_enabled.set(True) + app_instance.database = MagicMock() + # Mock identity resolution + app_instance.database.announces.get_announce_by_hash.return_value = None + + target_hash = "c" * 32 + + # Mock request with match_info for the variable in path + request = MagicMock() + request.match_info = {"destination_hash": target_hash} + + handler = None + for route in app_instance.get_routes(): + if ( + route.path == "/api/v1/blocked-destinations/{destination_hash}" + and route.method == "DELETE" + ): + handler = route.handler + break + + assert handler is not None + + await handler(request) + + # Verify DB call + app_instance.database.misc.delete_blocked_destination.assert_called_with( + target_hash + ) + + # Verify RNS unblackhole call + app_instance.reticulum.unblackhole_identity.assert_called() + args, _ = app_instance.reticulum.unblackhole_identity.call_args + assert args[0] == bytes.fromhex(target_hash) + + +@pytest.mark.asyncio +async def test_get_blackhole_list(mock_rns_minimal, temp_dir): + with patch("meshchatx.meshchat.generate_ssl_certificate"): + app_instance = ReticulumMeshChat( + identity=mock_rns_minimal, + storage_dir=temp_dir, + reticulum_config_dir=temp_dir, + ) + + ident_hash_bytes = b"\x01" * 32 + app_instance.reticulum.get_blackholed_identities.return_value = { + ident_hash_bytes: { + "source": b"\x02" * 32, + "until": 1234567890, + "reason": "Spam", + } + } + + request = MagicMock() + + handler = None + for route in app_instance.get_routes(): + if route.path == "/api/v1/reticulum/blackhole" and route.method == "GET": + handler = route.handler + break + + assert handler is not None + + response = await handler(request) + data = json.loads(response.body) + + assert ident_hash_bytes.hex() in data["blackholed_identities"] + info = data["blackholed_identities"][ident_hash_bytes.hex()] + assert info["reason"] == "Spam" + assert info["source"] == (b"\x02" * 32).hex() diff --git a/tests/backend/test_rnpath_logic.py b/tests/backend/test_rnpath_logic.py new file mode 100644 index 0000000..9f00286 --- /dev/null +++ b/tests/backend/test_rnpath_logic.py @@ -0,0 +1,121 @@ +import pytest +import json +from unittest.mock import MagicMock, patch, AsyncMock +from meshchatx.meshchat import ReticulumMeshChat +import RNS + + +@pytest.fixture +def temp_dir(tmp_path): + return str(tmp_path) + + +@pytest.fixture +def mock_rns_minimal(): + with ( + patch("RNS.Reticulum") as mock_rns, + patch("RNS.Transport"), + patch("LXMF.LXMRouter"), + patch("meshchatx.meshchat.get_file_path", return_value="/tmp/mock_path"), + ): + mock_rns_instance = mock_rns.return_value + mock_rns_instance.configpath = "/tmp/mock_config" + mock_rns_instance.is_connected_to_shared_instance = False + mock_rns_instance.transport_enabled.return_value = True + + # Path management mocks + mock_rns_instance.get_path_table.return_value = [] + mock_rns_instance.get_rate_table.return_value = [] + mock_rns_instance.drop_path.return_value = True + mock_rns_instance.drop_all_via.return_value = True + mock_rns_instance.drop_announce_queues = MagicMock() + + mock_id = MagicMock(spec=RNS.Identity) + mock_id.hash = b"\x00" * 32 + mock_id.hexhash = mock_id.hash.hex() + mock_id.get_private_key.return_value = b"test_private_key" + yield mock_id + + +@pytest.mark.asyncio +async def test_rnpath_table_endpoint(mock_rns_minimal, temp_dir): + with patch("meshchatx.meshchat.generate_ssl_certificate"): + app_instance = ReticulumMeshChat( + identity=mock_rns_minimal, + storage_dir=temp_dir, + reticulum_config_dir=temp_dir, + ) + + entry = { + "hash": b"\x01" * 32, + "hops": 1, + "via": b"\x02" * 32, + "interface": "UDP", + "expires": 1234567890, + } + app_instance.reticulum.get_path_table.return_value = [entry] + + request = MagicMock() + request.query = {} + + handler = next( + r.handler + for r in app_instance.get_routes() + if r.path == "/api/v1/rnpath/table" + ) + response = await handler(request) + data = json.loads(response.body) + + assert len(data["table"]) == 1 + assert data["table"][0]["hash"] == entry["hash"].hex() + + +@pytest.mark.asyncio +async def test_rnpath_request_endpoint(mock_rns_minimal, temp_dir): + with ( + patch("meshchatx.meshchat.generate_ssl_certificate"), + patch.object(RNS.Transport, "request_path") as mock_request_path, + ): + app_instance = ReticulumMeshChat( + identity=mock_rns_minimal, + storage_dir=temp_dir, + reticulum_config_dir=temp_dir, + ) + + target_hash = "a" * 32 + request = MagicMock() + request.json = AsyncMock(return_value={"destination_hash": target_hash}) + + handler = next( + r.handler + for r in app_instance.get_routes() + if r.path == "/api/v1/rnpath/request" + ) + response = await handler(request) + + mock_request_path.assert_called_with(bytes.fromhex(target_hash)) + assert json.loads(response.body)["success"] is True + + +@pytest.mark.asyncio +async def test_rnpath_drop_endpoint(mock_rns_minimal, temp_dir): + with patch("meshchatx.meshchat.generate_ssl_certificate"): + app_instance = ReticulumMeshChat( + identity=mock_rns_minimal, + storage_dir=temp_dir, + reticulum_config_dir=temp_dir, + ) + + target_hash = "b" * 32 + request = MagicMock() + request.json = AsyncMock(return_value={"destination_hash": target_hash}) + + handler = next( + r.handler + for r in app_instance.get_routes() + if r.path == "/api/v1/rnpath/drop" + ) + response = await handler(request) + + app_instance.reticulum.drop_path.assert_called_with(bytes.fromhex(target_hash)) + assert json.loads(response.body)["success"] is True diff --git a/tests/backend/test_rnstatus_blackhole.py b/tests/backend/test_rnstatus_blackhole.py new file mode 100644 index 0000000..2911616 --- /dev/null +++ b/tests/backend/test_rnstatus_blackhole.py @@ -0,0 +1,62 @@ +import pytest +from unittest.mock import MagicMock, patch +from meshchatx.src.backend.rnstatus_handler import RNStatusHandler +import RNS + + +@pytest.fixture +def mock_reticulum_instance(): + mock = MagicMock() + mock.get_interface_stats.return_value = {"interfaces": []} + mock.get_link_count.return_value = 0 + return mock + + +def test_blackhole_status_enabled(mock_reticulum_instance): + with ( + patch.object(RNS.Reticulum, "publish_blackhole_enabled", return_value=True), + patch.object( + RNS.Reticulum, + "blackhole_sources", + return_value=[b"\x01" * 16, b"\x02" * 16], + ), + ): + handler = RNStatusHandler(mock_reticulum_instance) + status = handler.get_status() + + assert status["blackhole_enabled"] is True + assert len(status["blackhole_sources"]) == 2 + assert status["blackhole_sources"][0] == (b"\x01" * 16).hex() + + +def test_blackhole_status_disabled(mock_reticulum_instance): + with ( + patch.object(RNS.Reticulum, "publish_blackhole_enabled", return_value=False), + patch.object(RNS.Reticulum, "blackhole_sources", return_value=[]), + ): + handler = RNStatusHandler(mock_reticulum_instance) + status = handler.get_status() + + assert status["blackhole_enabled"] is False + assert status["blackhole_sources"] == [] + + +def test_blackhole_status_missing_api(mock_reticulum_instance): + # Test backward compatibility or when API is missing (e.g. older RNS version simulation) + # We simulate this by making the attribute access raise AttributeError + # However, since we import RNS in the module, we need to ensure the mock raises AttributeError + + # We can't easily remove attributes from the real RNS module if it's already imported. + # But we can patch the RNS object inside rnstatus_handler module. + + with patch( + "meshchatx.src.backend.rnstatus_handler.RNS.Reticulum" + ) as mock_rns_class: + del mock_rns_class.publish_blackhole_enabled + + handler = RNStatusHandler(mock_reticulum_instance) + status = handler.get_status() + + # Should default to False/Empty on exception + assert status["blackhole_enabled"] is False + assert status["blackhole_sources"] == []