feat(tests): add comprehensive LXMF propagation and sync tests
This commit is contained in:
223
tests/backend/test_lxmf_propagation_full.py
Normal file
223
tests/backend/test_lxmf_propagation_full.py
Normal file
@@ -0,0 +1,223 @@
|
||||
import shutil
|
||||
import tempfile
|
||||
import pytest
|
||||
import json
|
||||
from unittest.mock import MagicMock, patch
|
||||
from meshchatx.meshchat import ReticulumMeshChat
|
||||
import RNS
|
||||
import LXMF
|
||||
|
||||
# Store original constants
|
||||
PR_IDLE = LXMF.LXMRouter.PR_IDLE
|
||||
PR_PATH_REQUESTED = LXMF.LXMRouter.PR_PATH_REQUESTED
|
||||
PR_RECEIVING = LXMF.LXMRouter.PR_RECEIVING
|
||||
PR_COMPLETE = LXMF.LXMRouter.PR_COMPLETE
|
||||
PR_FAILED = LXMF.LXMRouter.PR_FAILED
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def temp_dir():
|
||||
dir_path = tempfile.mkdtemp()
|
||||
yield dir_path
|
||||
shutil.rmtree(dir_path)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_app(temp_dir):
|
||||
with (
|
||||
patch("RNS.Reticulum") as mock_rns,
|
||||
patch("RNS.Transport"),
|
||||
patch("LXMF.LXMRouter") as mock_router_class,
|
||||
patch("meshchatx.meshchat.get_file_path", return_value="/tmp/mock_path"),
|
||||
patch("meshchatx.meshchat.generate_ssl_certificate"),
|
||||
):
|
||||
# Set up constants on the mock class
|
||||
mock_router_class.PR_IDLE = PR_IDLE
|
||||
mock_router_class.PR_PATH_REQUESTED = PR_PATH_REQUESTED
|
||||
mock_router_class.PR_RECEIVING = PR_RECEIVING
|
||||
mock_router_class.PR_COMPLETE = PR_COMPLETE
|
||||
mock_router_class.PR_FAILED = PR_FAILED
|
||||
|
||||
mock_router = mock_router_class.return_value
|
||||
mock_router.PR_IDLE = PR_IDLE
|
||||
mock_router.PR_PATH_REQUESTED = PR_PATH_REQUESTED
|
||||
mock_router.PR_RECEIVING = PR_RECEIVING
|
||||
mock_router.PR_COMPLETE = PR_COMPLETE
|
||||
mock_router.PR_FAILED = PR_FAILED
|
||||
|
||||
mock_router.propagation_transfer_state = PR_IDLE
|
||||
mock_router.propagation_transfer_progress = 0.0
|
||||
mock_router.propagation_transfer_last_result = 0
|
||||
|
||||
mock_dest = MagicMock()
|
||||
mock_dest.hash = b"local_dest_hash_16b"
|
||||
mock_dest.hexhash = "6c6f63616c5f646573745f686173685f"
|
||||
mock_router.register_delivery_identity.return_value = mock_dest
|
||||
|
||||
mock_rns_inst = mock_rns.return_value
|
||||
mock_rns_inst.transport_enabled.return_value = False
|
||||
|
||||
with patch(
|
||||
"meshchatx.src.backend.meshchat_utils.LXMRouter"
|
||||
) as mock_utils_router:
|
||||
mock_utils_router.PR_IDLE = PR_IDLE
|
||||
mock_utils_router.PR_PATH_REQUESTED = PR_PATH_REQUESTED
|
||||
mock_utils_router.PR_RECEIVING = PR_RECEIVING
|
||||
mock_utils_router.PR_COMPLETE = PR_COMPLETE
|
||||
mock_utils_router.PR_FAILED = PR_FAILED
|
||||
|
||||
real_id = RNS.Identity()
|
||||
app = ReticulumMeshChat(
|
||||
identity=real_id,
|
||||
storage_dir=temp_dir,
|
||||
reticulum_config_dir=temp_dir,
|
||||
)
|
||||
|
||||
app.current_context.message_router = mock_router
|
||||
|
||||
with patch.object(
|
||||
app, "send_config_to_websocket_clients", return_value=None
|
||||
):
|
||||
yield app
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_lxmf_propagation_config(mock_app):
|
||||
node_hash_hex = "d81255ae2ff367d4883b16c9cc8c6178"
|
||||
node_hash_bytes = bytes.fromhex(node_hash_hex)
|
||||
|
||||
await mock_app.update_config(
|
||||
{"lxmf_preferred_propagation_node_destination_hash": node_hash_hex}
|
||||
)
|
||||
|
||||
mock_app.current_context.message_router.set_outbound_propagation_node.assert_called_with(
|
||||
node_hash_bytes
|
||||
)
|
||||
assert (
|
||||
mock_app.config.lxmf_preferred_propagation_node_destination_hash.get()
|
||||
== node_hash_hex
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_lxmf_sync_flow(mock_app):
|
||||
mock_router = mock_app.current_context.message_router
|
||||
mock_router.get_outbound_propagation_node.return_value = b"somehash"
|
||||
|
||||
# Trigger sync
|
||||
for route in mock_app.get_routes():
|
||||
if route.path == "/api/v1/lxmf/propagation-node/sync":
|
||||
await route.handler(None)
|
||||
break
|
||||
|
||||
mock_router.request_messages_from_propagation_node.assert_called_once()
|
||||
|
||||
# Check status (Receiving)
|
||||
mock_router.propagation_transfer_state = PR_RECEIVING
|
||||
mock_router.propagation_transfer_progress = 0.75
|
||||
|
||||
status_handler = next(
|
||||
r.handler
|
||||
for r in mock_app.get_routes()
|
||||
if r.path == "/api/v1/lxmf/propagation-node/status"
|
||||
)
|
||||
response = await status_handler(None)
|
||||
data = json.loads(response.body)
|
||||
assert data["propagation_node_status"]["state"] == "receiving"
|
||||
assert data["propagation_node_status"]["progress"] == 75.0
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_hosting_prop_node(mock_app):
|
||||
mock_router = mock_app.current_context.message_router
|
||||
|
||||
await mock_app.update_config({"lxmf_local_propagation_node_enabled": True})
|
||||
mock_router.enable_propagation.assert_called_once()
|
||||
|
||||
await mock_app.update_config({"lxmf_local_propagation_node_enabled": False})
|
||||
mock_router.disable_propagation.assert_called_once()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_send_failed_via_prop_node(mock_app):
|
||||
mock_router = mock_app.current_context.message_router
|
||||
mock_router.get_outbound_propagation_node.return_value = b"active_prop_node"
|
||||
|
||||
# Create a mock failed message with required attributes
|
||||
mock_msg = MagicMock(spec=LXMF.LXMessage)
|
||||
mock_msg.state = LXMF.LXMessage.FAILED
|
||||
mock_msg.source_hash = b"source_hash_16b"
|
||||
mock_msg.destination_hash = b"dest_hash_16b"
|
||||
mock_msg.hash = b"msg_hash_16b"
|
||||
|
||||
mock_app.send_failed_message_via_propagation_node(mock_msg)
|
||||
|
||||
assert mock_msg.desired_method == LXMF.LXMessage.PROPAGATED
|
||||
mock_router.handle_outbound.assert_called_with(mock_msg)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_auto_sync_interval_config(mock_app):
|
||||
await mock_app.update_config(
|
||||
{"lxmf_preferred_propagation_node_auto_sync_interval_seconds": 3600}
|
||||
)
|
||||
assert (
|
||||
mock_app.config.lxmf_preferred_propagation_node_auto_sync_interval_seconds.get()
|
||||
== 3600
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_propagation_node_status_mapping(mock_app):
|
||||
mock_router = mock_app.current_context.message_router
|
||||
status_handler = next(
|
||||
r.handler
|
||||
for r in mock_app.get_routes()
|
||||
if r.path == "/api/v1/lxmf/propagation-node/status"
|
||||
)
|
||||
|
||||
states_to_test = [
|
||||
(PR_IDLE, "idle"),
|
||||
(PR_PATH_REQUESTED, "path_requested"),
|
||||
(PR_RECEIVING, "receiving"),
|
||||
(PR_COMPLETE, "complete"),
|
||||
(PR_FAILED, "failed"),
|
||||
]
|
||||
|
||||
for state_val, state_str in states_to_test:
|
||||
mock_router.propagation_transfer_state = state_val
|
||||
response = await status_handler(None)
|
||||
data = json.loads(response.body)
|
||||
assert data["propagation_node_status"]["state"] == state_str
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_user_provided_node_hash(mock_app):
|
||||
"""Specifically test the node hash provided by the user."""
|
||||
node_hash_hex = "d81255ae2ff367d4883b16c9cc8c6178"
|
||||
|
||||
# Set this node as preferred
|
||||
await mock_app.update_config(
|
||||
{"lxmf_preferred_propagation_node_destination_hash": node_hash_hex}
|
||||
)
|
||||
|
||||
# Check if the router was updated with the correct bytes
|
||||
mock_app.current_context.message_router.set_outbound_propagation_node.assert_called_with(
|
||||
bytes.fromhex(node_hash_hex)
|
||||
)
|
||||
|
||||
# Trigger a sync request
|
||||
mock_app.current_context.message_router.get_outbound_propagation_node.return_value = bytes.fromhex(
|
||||
node_hash_hex
|
||||
)
|
||||
sync_handler = next(
|
||||
r.handler
|
||||
for r in mock_app.get_routes()
|
||||
if r.path == "/api/v1/lxmf/propagation-node/sync"
|
||||
)
|
||||
await sync_handler(None)
|
||||
|
||||
# Verify the router was told to sync for our identity
|
||||
mock_app.current_context.message_router.request_messages_from_propagation_node.assert_called_with(
|
||||
mock_app.current_context.identity
|
||||
)
|
||||
140
tests/backend/test_lxmf_sync.py
Normal file
140
tests/backend/test_lxmf_sync.py
Normal file
@@ -0,0 +1,140 @@
|
||||
import shutil
|
||||
import tempfile
|
||||
import pytest
|
||||
import json
|
||||
from unittest.mock import MagicMock, patch
|
||||
from meshchatx.meshchat import ReticulumMeshChat
|
||||
import RNS
|
||||
import LXMF
|
||||
|
||||
# Store original constants
|
||||
PR_IDLE = LXMF.LXMRouter.PR_IDLE
|
||||
PR_COMPLETE = LXMF.LXMRouter.PR_COMPLETE
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def temp_dir():
|
||||
dir_path = tempfile.mkdtemp()
|
||||
yield dir_path
|
||||
shutil.rmtree(dir_path)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_app(temp_dir):
|
||||
with (
|
||||
patch("RNS.Reticulum") as mock_rns,
|
||||
patch("RNS.Transport"),
|
||||
patch("LXMF.LXMRouter") as mock_router_class,
|
||||
patch("meshchatx.meshchat.get_file_path", return_value="/tmp/mock_path"),
|
||||
patch("meshchatx.meshchat.generate_ssl_certificate"),
|
||||
):
|
||||
# Set up constants on the mock class
|
||||
mock_router_class.PR_IDLE = PR_IDLE
|
||||
mock_router_class.PR_COMPLETE = PR_COMPLETE
|
||||
|
||||
mock_router = mock_router_class.return_value
|
||||
mock_router.PR_IDLE = PR_IDLE
|
||||
mock_router.PR_COMPLETE = PR_COMPLETE
|
||||
|
||||
mock_router.propagation_transfer_state = PR_IDLE
|
||||
mock_router.propagation_transfer_progress = 0
|
||||
mock_router.propagation_transfer_last_result = 0
|
||||
|
||||
mock_dest = MagicMock()
|
||||
mock_dest.hexhash = "mock_hash"
|
||||
mock_router.register_delivery_identity.return_value = mock_dest
|
||||
|
||||
mock_rns_inst = mock_rns.return_value
|
||||
mock_rns_inst.transport_enabled.return_value = False
|
||||
|
||||
with patch(
|
||||
"meshchatx.src.backend.meshchat_utils.LXMRouter"
|
||||
) as mock_utils_router:
|
||||
mock_utils_router.PR_IDLE = PR_IDLE
|
||||
mock_utils_router.PR_COMPLETE = PR_COMPLETE
|
||||
|
||||
real_id = RNS.Identity()
|
||||
app = ReticulumMeshChat(
|
||||
identity=real_id,
|
||||
storage_dir=temp_dir,
|
||||
reticulum_config_dir=temp_dir,
|
||||
)
|
||||
app.current_context.message_router = mock_router
|
||||
|
||||
with patch.object(
|
||||
app, "send_config_to_websocket_clients", return_value=None
|
||||
):
|
||||
yield app
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_lxmf_sync_endpoints(mock_app):
|
||||
# 1. Test status endpoint initially idle
|
||||
handler = None
|
||||
for route in mock_app.get_routes():
|
||||
if (
|
||||
route.path == "/api/v1/lxmf/propagation-node/status"
|
||||
and route.method == "GET"
|
||||
):
|
||||
handler = route.handler
|
||||
break
|
||||
|
||||
assert handler is not None
|
||||
response = await handler(None)
|
||||
data = json.loads(response.body)
|
||||
assert data["propagation_node_status"]["state"] == "idle"
|
||||
|
||||
# 2. Test sync initiation
|
||||
sync_handler = None
|
||||
for route in mock_app.get_routes():
|
||||
if route.path == "/api/v1/lxmf/propagation-node/sync" and route.method == "GET":
|
||||
sync_handler = route.handler
|
||||
break
|
||||
|
||||
assert sync_handler is not None
|
||||
|
||||
# Mock outbound propagation node configured
|
||||
mock_app.current_context.message_router.get_outbound_propagation_node.return_value = b"somehash"
|
||||
|
||||
response = await sync_handler(None)
|
||||
assert response.status == 200
|
||||
mock_app.current_context.message_router.request_messages_from_propagation_node.assert_called_once()
|
||||
|
||||
# 3. Test status change to complete
|
||||
mock_app.current_context.message_router.propagation_transfer_state = (
|
||||
LXMF.LXMRouter.PR_COMPLETE
|
||||
)
|
||||
response = await handler(None)
|
||||
data = json.loads(response.body)
|
||||
assert data["propagation_node_status"]["state"] == "complete"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_specific_node_hash_validation(mock_app):
|
||||
node_hash_hex = "d81255ae2ff367d4883b16c9cc8c6178"
|
||||
|
||||
# Ensure update_config doesn't crash due to mock serialization
|
||||
with patch.object(mock_app, "send_config_to_websocket_clients", return_value=None):
|
||||
# Set the preferred propagation node
|
||||
await mock_app.update_config(
|
||||
{"lxmf_preferred_propagation_node_destination_hash": node_hash_hex}
|
||||
)
|
||||
|
||||
# Verify it was set on the router correctly as 16 bytes
|
||||
expected_bytes = bytes.fromhex(node_hash_hex)
|
||||
mock_app.current_context.message_router.set_outbound_propagation_node.assert_called_with(
|
||||
expected_bytes
|
||||
)
|
||||
|
||||
# Trigger sync
|
||||
sync_handler = None
|
||||
for route in mock_app.get_routes():
|
||||
if route.path == "/api/v1/lxmf/propagation-node/sync" and route.method == "GET":
|
||||
sync_handler = route.handler
|
||||
break
|
||||
|
||||
# Ensure it's considered configured
|
||||
mock_app.current_context.message_router.get_outbound_propagation_node.return_value = expected_bytes
|
||||
|
||||
await sync_handler(None)
|
||||
mock_app.current_context.message_router.request_messages_from_propagation_node.assert_called_once()
|
||||
Reference in New Issue
Block a user