mirror of
https://github.com/torlando-tech/columba
synced 2026-08-12 18:07:15 -04:00
b3345e66e0
The production code was fixed to not pass None to LXMF's set_outbound_propagation_node() since LXMF doesn't support it. Updated the tests to verify that the router method is NOT called when clearing with None, only internal state is cleared. Fixes two failing tests: - test_set_propagation_node_clears_when_none - test_set_outbound_propagation_node_clears_with_none Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
1234 lines
44 KiB
Python
1234 lines
44 KiB
Python
"""
|
|
Test suite for ReticulumWrapper propagation node functionality.
|
|
|
|
Tests the four key propagation node methods:
|
|
1. set_outbound_propagation_node - Set/clear propagation node
|
|
2. get_outbound_propagation_node - Get current node configuration
|
|
3. request_messages_from_propagation_node - Request messages from node
|
|
4. get_propagation_state - Get sync state and progress
|
|
"""
|
|
|
|
import sys
|
|
import os
|
|
import unittest
|
|
from unittest.mock import Mock, MagicMock, patch, PropertyMock
|
|
|
|
# Add parent directory to path to import reticulum_wrapper
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
|
|
# Mock RNS and LXMF before importing reticulum_wrapper
|
|
sys.modules['RNS'] = MagicMock()
|
|
sys.modules['RNS.vendor'] = MagicMock()
|
|
sys.modules['RNS.vendor.platformutils'] = MagicMock()
|
|
|
|
# Create LXMF mock with proper LXMRouter constants
|
|
# These values must match the actual LXMF library constants
|
|
lxmf_mock = MagicMock()
|
|
lxmf_mock.LXMRouter.PR_IDLE = 0
|
|
lxmf_mock.LXMRouter.PR_PATH_REQUESTED = 1
|
|
lxmf_mock.LXMRouter.PR_LINK_ESTABLISHING = 2
|
|
lxmf_mock.LXMRouter.PR_LINK_ESTABLISHED = 3
|
|
lxmf_mock.LXMRouter.PR_REQUEST_SENT = 4
|
|
lxmf_mock.LXMRouter.PR_RECEIVING = 5
|
|
lxmf_mock.LXMRouter.PR_RESPONSE_RECEIVED = 6
|
|
lxmf_mock.LXMRouter.PR_COMPLETE = 7
|
|
lxmf_mock.LXMRouter.PR_NO_PATH = 8
|
|
lxmf_mock.LXMRouter.PR_LINK_FAILED = 9
|
|
lxmf_mock.LXMRouter.PR_TRANSFER_FAILED = 10
|
|
lxmf_mock.LXMRouter.PR_NO_IDENTITY_RCVD = 11
|
|
lxmf_mock.LXMRouter.PR_NO_ACCESS = 12
|
|
sys.modules['LXMF'] = lxmf_mock
|
|
|
|
# Now import after mocking
|
|
import reticulum_wrapper
|
|
|
|
|
|
class PropagationTestBase(unittest.TestCase):
|
|
"""Base class that sets up LXMF mock for propagation tests."""
|
|
|
|
def setUp(self):
|
|
# Save original LXMF value
|
|
self._original_lxmf = reticulum_wrapper.LXMF
|
|
# Set the module-level LXMF variable in reticulum_wrapper
|
|
# (it's initialized to None and only set during initialize() which we skip in tests)
|
|
reticulum_wrapper.LXMF = lxmf_mock
|
|
|
|
def tearDown(self):
|
|
# Restore original LXMF value
|
|
reticulum_wrapper.LXMF = self._original_lxmf
|
|
|
|
|
|
class TestSetOutboundPropagationNode(PropagationTestBase):
|
|
"""Test set_outbound_propagation_node method"""
|
|
|
|
def setUp(self):
|
|
"""Set up test fixtures"""
|
|
super().setUp()
|
|
import tempfile
|
|
self.temp_dir = tempfile.mkdtemp()
|
|
self.wrapper = reticulum_wrapper.ReticulumWrapper(self.temp_dir)
|
|
|
|
# Enable Reticulum
|
|
reticulum_wrapper.RETICULUM_AVAILABLE = True
|
|
|
|
# Mock router
|
|
self.mock_router = Mock()
|
|
self.wrapper.router = self.mock_router
|
|
self.wrapper.initialized = True
|
|
|
|
def tearDown(self):
|
|
"""Clean up test fixtures"""
|
|
import shutil
|
|
if os.path.exists(self.temp_dir):
|
|
shutil.rmtree(self.temp_dir)
|
|
|
|
def test_set_propagation_node_success(self):
|
|
"""Test successfully setting a propagation node"""
|
|
test_hash = b'1234567890123456' # 16-byte hash
|
|
|
|
result = self.wrapper.set_outbound_propagation_node(test_hash)
|
|
|
|
# Verify success
|
|
self.assertTrue(result['success'])
|
|
|
|
# Verify router was called with correct hash
|
|
self.mock_router.set_outbound_propagation_node.assert_called_once_with(test_hash)
|
|
|
|
# Verify internal state was updated
|
|
self.assertEqual(self.wrapper.active_propagation_node, test_hash)
|
|
|
|
def test_set_propagation_node_clears_when_none(self):
|
|
"""Test clearing propagation node by passing None"""
|
|
# First set a node
|
|
self.wrapper.active_propagation_node = b'existing_node_hash'
|
|
|
|
result = self.wrapper.set_outbound_propagation_node(None)
|
|
|
|
# Verify success
|
|
self.assertTrue(result['success'])
|
|
|
|
# Verify router was NOT called (LXMF doesn't support None, we just clear internally)
|
|
self.mock_router.set_outbound_propagation_node.assert_not_called()
|
|
|
|
# Verify internal state was cleared
|
|
self.assertIsNone(self.wrapper.active_propagation_node)
|
|
|
|
def test_set_propagation_node_converts_jarray(self):
|
|
"""Test that jarray-like objects are converted to bytes"""
|
|
# Create a mock jarray-like object (iterable but not bytes)
|
|
test_jarray = [0x12, 0x34, 0x56, 0x78, 0x90, 0xab, 0xcd, 0xef,
|
|
0x12, 0x34, 0x56, 0x78, 0x90, 0xab, 0xcd, 0xef]
|
|
|
|
result = self.wrapper.set_outbound_propagation_node(test_jarray)
|
|
|
|
# Verify success
|
|
self.assertTrue(result['success'])
|
|
|
|
# Verify router was called with bytes
|
|
call_args = self.mock_router.set_outbound_propagation_node.call_args[0][0]
|
|
self.assertIsInstance(call_args, bytes)
|
|
|
|
# Verify bytes content matches
|
|
expected_bytes = bytes(test_jarray)
|
|
self.assertEqual(call_args, expected_bytes)
|
|
|
|
def test_set_propagation_node_not_initialized(self):
|
|
"""Test error when wrapper not initialized"""
|
|
self.wrapper.initialized = False
|
|
|
|
result = self.wrapper.set_outbound_propagation_node(b'test_hash_123456')
|
|
|
|
# Verify failure
|
|
self.assertFalse(result['success'])
|
|
self.assertIn('error', result)
|
|
self.assertIn('not initialized', result['error'].lower())
|
|
|
|
def test_set_propagation_node_no_router(self):
|
|
"""Test error when router is not available"""
|
|
self.wrapper.router = None
|
|
|
|
result = self.wrapper.set_outbound_propagation_node(b'test_hash_123456')
|
|
|
|
# Verify failure
|
|
self.assertFalse(result['success'])
|
|
self.assertIn('error', result)
|
|
self.assertIn('not initialized', result['error'].lower())
|
|
|
|
def test_set_propagation_node_router_exception(self):
|
|
"""Test handling of router exceptions"""
|
|
self.mock_router.set_outbound_propagation_node.side_effect = Exception("Router error")
|
|
|
|
result = self.wrapper.set_outbound_propagation_node(b'test_hash_123456')
|
|
|
|
# Verify failure
|
|
self.assertFalse(result['success'])
|
|
self.assertIn('error', result)
|
|
self.assertEqual(result['error'], "Router error")
|
|
|
|
|
|
class TestGetOutboundPropagationNode(PropagationTestBase):
|
|
"""Test get_outbound_propagation_node method"""
|
|
|
|
def setUp(self):
|
|
"""Set up test fixtures"""
|
|
super().setUp()
|
|
import tempfile
|
|
self.temp_dir = tempfile.mkdtemp()
|
|
self.wrapper = reticulum_wrapper.ReticulumWrapper(self.temp_dir)
|
|
|
|
# Enable Reticulum
|
|
reticulum_wrapper.RETICULUM_AVAILABLE = True
|
|
|
|
# Mock router
|
|
self.mock_router = Mock()
|
|
self.wrapper.router = self.mock_router
|
|
self.wrapper.initialized = True
|
|
|
|
def tearDown(self):
|
|
"""Clean up test fixtures"""
|
|
import shutil
|
|
if os.path.exists(self.temp_dir):
|
|
shutil.rmtree(self.temp_dir)
|
|
|
|
def test_get_propagation_node_returns_hex(self):
|
|
"""Test getting propagation node returns hex string"""
|
|
test_hash = b'1234567890123456'
|
|
self.mock_router.get_outbound_propagation_node.return_value = test_hash
|
|
|
|
result = self.wrapper.get_outbound_propagation_node()
|
|
|
|
# Verify success
|
|
self.assertTrue(result['success'])
|
|
|
|
# Verify hex string returned
|
|
self.assertIn('propagation_node', result)
|
|
self.assertEqual(result['propagation_node'], test_hash.hex())
|
|
|
|
def test_get_propagation_node_returns_none_when_not_set(self):
|
|
"""Test getting propagation node when none is set"""
|
|
self.mock_router.get_outbound_propagation_node.return_value = None
|
|
|
|
result = self.wrapper.get_outbound_propagation_node()
|
|
|
|
# Verify success
|
|
self.assertTrue(result['success'])
|
|
|
|
# Verify None returned
|
|
self.assertIn('propagation_node', result)
|
|
self.assertIsNone(result['propagation_node'])
|
|
|
|
def test_get_propagation_node_not_initialized(self):
|
|
"""Test error when wrapper not initialized"""
|
|
self.wrapper.initialized = False
|
|
|
|
result = self.wrapper.get_outbound_propagation_node()
|
|
|
|
# Verify failure
|
|
self.assertFalse(result['success'])
|
|
self.assertIn('error', result)
|
|
self.assertIn('not initialized', result['error'].lower())
|
|
|
|
def test_get_propagation_node_no_router(self):
|
|
"""Test error when router is not available"""
|
|
self.wrapper.router = None
|
|
|
|
result = self.wrapper.get_outbound_propagation_node()
|
|
|
|
# Verify failure
|
|
self.assertFalse(result['success'])
|
|
self.assertIn('error', result)
|
|
|
|
def test_get_propagation_node_router_exception(self):
|
|
"""Test handling of router exceptions"""
|
|
self.mock_router.get_outbound_propagation_node.side_effect = Exception("Router error")
|
|
|
|
result = self.wrapper.get_outbound_propagation_node()
|
|
|
|
# Verify failure
|
|
self.assertFalse(result['success'])
|
|
self.assertIn('error', result)
|
|
|
|
|
|
class TestRequestMessagesFromPropagationNode(PropagationTestBase):
|
|
"""Test request_messages_from_propagation_node method"""
|
|
|
|
def setUp(self):
|
|
"""Set up test fixtures"""
|
|
super().setUp()
|
|
import tempfile
|
|
self.temp_dir = tempfile.mkdtemp()
|
|
self.wrapper = reticulum_wrapper.ReticulumWrapper(self.temp_dir)
|
|
|
|
# Mock router
|
|
self.mock_router = Mock()
|
|
self.mock_router.propagation_transfer_state = 0
|
|
self.wrapper.router = self.mock_router
|
|
self.wrapper.initialized = True
|
|
|
|
# Set active propagation node
|
|
self.wrapper.active_propagation_node = b'1234567890123456'
|
|
|
|
# Mock default identity
|
|
self.mock_identity = Mock()
|
|
self.mock_identity.hash = b'identity_hash123'
|
|
self.wrapper.default_identity = self.mock_identity
|
|
|
|
def tearDown(self):
|
|
"""Clean up test fixtures"""
|
|
import shutil
|
|
if os.path.exists(self.temp_dir):
|
|
shutil.rmtree(self.temp_dir)
|
|
|
|
def test_request_messages_success_default_identity(self):
|
|
"""Test successfully requesting messages with default identity"""
|
|
result = self.wrapper.request_messages_from_propagation_node()
|
|
|
|
# Verify success
|
|
self.assertTrue(result['success'])
|
|
|
|
# Verify router was called with default identity
|
|
self.mock_router.request_messages_from_propagation_node.assert_called_once()
|
|
call_args = self.mock_router.request_messages_from_propagation_node.call_args
|
|
self.assertEqual(call_args[0][0], self.mock_identity)
|
|
|
|
# Verify state is returned
|
|
self.assertIn('state', result)
|
|
self.assertEqual(result['state'], 0)
|
|
|
|
@patch('reticulum_wrapper.RNS')
|
|
def test_request_messages_with_custom_identity(self, mock_rns):
|
|
"""Test requesting messages with custom identity"""
|
|
# Create mock identity private key
|
|
test_private_key = b'test_private_key_bytes_32_chars!'
|
|
|
|
# Mock RNS.Identity.from_bytes
|
|
mock_custom_identity = Mock()
|
|
mock_custom_identity.hash = b'custom_identity_'
|
|
mock_rns.Identity.from_bytes.return_value = mock_custom_identity
|
|
|
|
result = self.wrapper.request_messages_from_propagation_node(
|
|
identity_private_key=test_private_key
|
|
)
|
|
|
|
# Verify success
|
|
self.assertTrue(result['success'])
|
|
|
|
# Verify from_bytes was called with the private key
|
|
mock_rns.Identity.from_bytes.assert_called_once_with(test_private_key)
|
|
|
|
# Verify router was called with custom identity
|
|
call_args = self.mock_router.request_messages_from_propagation_node.call_args
|
|
self.assertEqual(call_args[0][0], mock_custom_identity)
|
|
|
|
def test_request_messages_with_custom_max_messages(self):
|
|
"""Test requesting messages with custom max_messages"""
|
|
result = self.wrapper.request_messages_from_propagation_node(max_messages=512)
|
|
|
|
# Verify success
|
|
self.assertTrue(result['success'])
|
|
|
|
# Verify router was called with custom max_messages
|
|
call_args = self.mock_router.request_messages_from_propagation_node.call_args
|
|
self.assertEqual(call_args[1]['max_messages'], 512)
|
|
|
|
def test_request_messages_no_propagation_node(self):
|
|
"""Test error when no propagation node is configured"""
|
|
self.wrapper.active_propagation_node = None
|
|
|
|
result = self.wrapper.request_messages_from_propagation_node()
|
|
|
|
# Verify failure
|
|
self.assertFalse(result['success'])
|
|
self.assertIn('error', result)
|
|
self.assertIn('No propagation node', result['error'])
|
|
|
|
# Verify error code
|
|
self.assertEqual(result['errorCode'], 'NO_PROPAGATION_NODE')
|
|
|
|
def test_request_messages_not_initialized(self):
|
|
"""Test error when wrapper not initialized"""
|
|
self.wrapper.initialized = False
|
|
|
|
result = self.wrapper.request_messages_from_propagation_node()
|
|
|
|
# Verify failure
|
|
self.assertFalse(result['success'])
|
|
self.assertIn('error', result)
|
|
self.assertIn('not initialized', result['error'].lower())
|
|
|
|
def test_request_messages_no_router(self):
|
|
"""Test error when router is not available"""
|
|
self.wrapper.router = None
|
|
|
|
result = self.wrapper.request_messages_from_propagation_node()
|
|
|
|
# Verify failure
|
|
self.assertFalse(result['success'])
|
|
self.assertIn('error', result)
|
|
|
|
@patch('reticulum_wrapper.RNS')
|
|
def test_request_messages_converts_jarray_identity(self, mock_rns):
|
|
"""Test that jarray-like private key is converted to bytes"""
|
|
# Create a mock jarray-like object (iterable but not bytes)
|
|
test_jarray = [0x01] * 32 # 32-byte private key
|
|
|
|
# Mock RNS.Identity.from_bytes
|
|
mock_identity = Mock()
|
|
mock_identity.hash = b'converted_ident'
|
|
mock_rns.Identity.from_bytes.return_value = mock_identity
|
|
|
|
result = self.wrapper.request_messages_from_propagation_node(
|
|
identity_private_key=test_jarray
|
|
)
|
|
|
|
# Verify success
|
|
self.assertTrue(result['success'])
|
|
|
|
# Verify from_bytes was called with bytes
|
|
call_args = mock_rns.Identity.from_bytes.call_args[0][0]
|
|
self.assertIsInstance(call_args, bytes)
|
|
self.assertEqual(call_args, bytes(test_jarray))
|
|
|
|
def test_request_messages_router_exception(self):
|
|
"""Test handling of router exceptions"""
|
|
self.mock_router.request_messages_from_propagation_node.side_effect = Exception("Request failed")
|
|
|
|
result = self.wrapper.request_messages_from_propagation_node()
|
|
|
|
# Verify failure
|
|
self.assertFalse(result['success'])
|
|
self.assertIn('error', result)
|
|
self.assertEqual(result['error'], "Request failed")
|
|
|
|
def test_request_messages_returns_transfer_state(self):
|
|
"""Test that current transfer state is returned"""
|
|
# Set different transfer states
|
|
test_states = [0, 1, 2, 3, 4, 5, 7]
|
|
|
|
for state in test_states:
|
|
self.mock_router.propagation_transfer_state = state
|
|
result = self.wrapper.request_messages_from_propagation_node()
|
|
|
|
self.assertTrue(result['success'])
|
|
self.assertEqual(result['state'], state)
|
|
|
|
|
|
class TestGetPropagationState(PropagationTestBase):
|
|
"""Test get_propagation_state method"""
|
|
|
|
def setUp(self):
|
|
"""Set up test fixtures"""
|
|
super().setUp()
|
|
import tempfile
|
|
self.temp_dir = tempfile.mkdtemp()
|
|
self.wrapper = reticulum_wrapper.ReticulumWrapper(self.temp_dir)
|
|
|
|
# Enable Reticulum
|
|
reticulum_wrapper.RETICULUM_AVAILABLE = True
|
|
|
|
# Mock router
|
|
self.mock_router = Mock()
|
|
self.wrapper.router = self.mock_router
|
|
self.wrapper.initialized = True
|
|
|
|
def tearDown(self):
|
|
"""Clean up test fixtures"""
|
|
import shutil
|
|
if os.path.exists(self.temp_dir):
|
|
shutil.rmtree(self.temp_dir)
|
|
|
|
def test_get_propagation_state_idle(self):
|
|
"""Test getting state when idle (state 0)"""
|
|
self.mock_router.propagation_transfer_state = 0
|
|
self.mock_router.propagation_transfer_progress = 0.0
|
|
|
|
result = self.wrapper.get_propagation_state()
|
|
|
|
# Verify success
|
|
self.assertTrue(result['success'])
|
|
|
|
# Verify state details
|
|
self.assertEqual(result['state'], 0)
|
|
self.assertEqual(result['state_name'], 'idle')
|
|
self.assertEqual(result['progress'], 0.0)
|
|
|
|
def test_get_propagation_state_path_requested(self):
|
|
"""Test state during path discovery (state 1)"""
|
|
self.mock_router.propagation_transfer_state = 1
|
|
self.mock_router.propagation_transfer_progress = 0.0
|
|
|
|
result = self.wrapper.get_propagation_state()
|
|
|
|
self.assertTrue(result['success'])
|
|
self.assertEqual(result['state'], 1)
|
|
self.assertEqual(result['state_name'], 'path_requested')
|
|
|
|
def test_get_propagation_state_link_establishing(self):
|
|
"""Test state during link establishment (state 2)"""
|
|
self.mock_router.propagation_transfer_state = 2
|
|
self.mock_router.propagation_transfer_progress = 0.0
|
|
|
|
result = self.wrapper.get_propagation_state()
|
|
|
|
self.assertTrue(result['success'])
|
|
self.assertEqual(result['state'], 2)
|
|
self.assertEqual(result['state_name'], 'link_establishing')
|
|
|
|
def test_get_propagation_state_link_established(self):
|
|
"""Test state when link is established (state 3)"""
|
|
self.mock_router.propagation_transfer_state = 3
|
|
self.mock_router.propagation_transfer_progress = 0.0
|
|
|
|
result = self.wrapper.get_propagation_state()
|
|
|
|
self.assertTrue(result['success'])
|
|
self.assertEqual(result['state'], 3)
|
|
self.assertEqual(result['state_name'], 'link_established')
|
|
|
|
def test_get_propagation_state_request_sent(self):
|
|
"""Test state when message list requested (state 4)"""
|
|
self.mock_router.propagation_transfer_state = 4
|
|
self.mock_router.propagation_transfer_progress = 0.0
|
|
|
|
result = self.wrapper.get_propagation_state()
|
|
|
|
self.assertTrue(result['success'])
|
|
self.assertEqual(result['state'], 4)
|
|
self.assertEqual(result['state_name'], 'request_sent')
|
|
|
|
def test_get_propagation_state_receiving(self):
|
|
"""Test state during message download (state 5)"""
|
|
self.mock_router.propagation_transfer_state = 5
|
|
self.mock_router.propagation_transfer_progress = 0.65
|
|
|
|
result = self.wrapper.get_propagation_state()
|
|
|
|
self.assertTrue(result['success'])
|
|
self.assertEqual(result['state'], 5)
|
|
self.assertEqual(result['state_name'], 'receiving')
|
|
self.assertEqual(result['progress'], 0.65)
|
|
|
|
def test_get_propagation_state_complete(self):
|
|
"""Test state when transfer complete (state 7)"""
|
|
self.mock_router.propagation_transfer_state = 7
|
|
self.mock_router.propagation_transfer_progress = 1.0
|
|
self.mock_router.propagation_transfer_last_result = 5 # 5 messages received
|
|
|
|
result = self.wrapper.get_propagation_state()
|
|
|
|
self.assertTrue(result['success'])
|
|
self.assertEqual(result['state'], 7)
|
|
self.assertEqual(result['state_name'], 'complete')
|
|
self.assertEqual(result['progress'], 1.0)
|
|
self.assertEqual(result['messages_received'], 5)
|
|
|
|
def test_get_propagation_state_unknown_state(self):
|
|
"""Test handling of unknown state values"""
|
|
self.mock_router.propagation_transfer_state = 99
|
|
self.mock_router.propagation_transfer_progress = 0.0
|
|
|
|
result = self.wrapper.get_propagation_state()
|
|
|
|
self.assertTrue(result['success'])
|
|
self.assertEqual(result['state'], 99)
|
|
self.assertEqual(result['state_name'], 'unknown_99')
|
|
|
|
def test_get_propagation_state_messages_received(self):
|
|
"""Test that messages_received is included in response"""
|
|
self.mock_router.propagation_transfer_state = 0
|
|
self.mock_router.propagation_transfer_progress = 0.0
|
|
self.mock_router.propagation_transfer_last_result = 10
|
|
|
|
result = self.wrapper.get_propagation_state()
|
|
|
|
self.assertTrue(result['success'])
|
|
self.assertIn('messages_received', result)
|
|
self.assertEqual(result['messages_received'], 10)
|
|
|
|
def test_get_propagation_state_no_last_result_attribute(self):
|
|
"""Test handling when propagation_transfer_last_result doesn't exist"""
|
|
self.mock_router.propagation_transfer_state = 0
|
|
self.mock_router.propagation_transfer_progress = 0.0
|
|
|
|
# Use a special mock that raises AttributeError for propagation_transfer_last_result
|
|
del self.mock_router.propagation_transfer_last_result
|
|
|
|
result = self.wrapper.get_propagation_state()
|
|
|
|
self.assertTrue(result['success'])
|
|
self.assertIn('messages_received', result)
|
|
self.assertEqual(result['messages_received'], 0) # Should default to 0
|
|
|
|
def test_get_propagation_state_not_initialized(self):
|
|
"""Test error when wrapper not initialized"""
|
|
self.wrapper.initialized = False
|
|
|
|
result = self.wrapper.get_propagation_state()
|
|
|
|
self.assertFalse(result['success'])
|
|
self.assertIn('error', result)
|
|
self.assertIn('not initialized', result['error'].lower())
|
|
|
|
def test_get_propagation_state_no_router(self):
|
|
"""Test error when router is not available"""
|
|
self.wrapper.router = None
|
|
|
|
result = self.wrapper.get_propagation_state()
|
|
|
|
self.assertFalse(result['success'])
|
|
self.assertIn('error', result)
|
|
|
|
def test_get_propagation_state_router_exception(self):
|
|
"""Test handling of router exceptions"""
|
|
# Make accessing state raise an exception
|
|
type(self.mock_router).propagation_transfer_state = PropertyMock(
|
|
side_effect=Exception("State error")
|
|
)
|
|
|
|
result = self.wrapper.get_propagation_state()
|
|
|
|
self.assertFalse(result['success'])
|
|
self.assertIn('error', result)
|
|
|
|
|
|
class TestPropagationNodeIntegration(PropagationTestBase):
|
|
"""Integration tests for propagation node workflow"""
|
|
|
|
def setUp(self):
|
|
"""Set up test fixtures"""
|
|
super().setUp()
|
|
import tempfile
|
|
self.temp_dir = tempfile.mkdtemp()
|
|
self.wrapper = reticulum_wrapper.ReticulumWrapper(self.temp_dir)
|
|
|
|
# Enable Reticulum
|
|
reticulum_wrapper.RETICULUM_AVAILABLE = True
|
|
|
|
# Mock router
|
|
self.mock_router = Mock()
|
|
self.wrapper.router = self.mock_router
|
|
self.wrapper.initialized = True
|
|
|
|
# Mock default identity
|
|
self.mock_identity = Mock()
|
|
self.mock_identity.hash = b'identity_hash123'
|
|
self.wrapper.default_identity = self.mock_identity
|
|
|
|
def tearDown(self):
|
|
"""Clean up test fixtures"""
|
|
import shutil
|
|
if os.path.exists(self.temp_dir):
|
|
shutil.rmtree(self.temp_dir)
|
|
|
|
def test_full_propagation_workflow(self):
|
|
"""Test complete workflow: set node, request messages, check state"""
|
|
node_hash = b'propagation_node'
|
|
|
|
# Step 1: Set propagation node
|
|
self.mock_router.get_outbound_propagation_node.return_value = node_hash
|
|
result = self.wrapper.set_outbound_propagation_node(node_hash)
|
|
self.assertTrue(result['success'])
|
|
|
|
# Step 2: Verify node was set
|
|
result = self.wrapper.get_outbound_propagation_node()
|
|
self.assertTrue(result['success'])
|
|
self.assertEqual(result['propagation_node'], node_hash.hex())
|
|
|
|
# Step 3: Request messages
|
|
self.mock_router.propagation_transfer_state = 1 # path_requested
|
|
result = self.wrapper.request_messages_from_propagation_node()
|
|
self.assertTrue(result['success'])
|
|
self.assertEqual(result['state'], 1)
|
|
|
|
# Step 4: Check state during transfer
|
|
self.mock_router.propagation_transfer_state = 5 # receiving
|
|
self.mock_router.propagation_transfer_progress = 0.5
|
|
result = self.wrapper.get_propagation_state()
|
|
self.assertTrue(result['success'])
|
|
self.assertEqual(result['state'], 5)
|
|
self.assertEqual(result['state_name'], 'receiving')
|
|
self.assertEqual(result['progress'], 0.5)
|
|
|
|
# Step 5: Check state when complete
|
|
self.mock_router.propagation_transfer_state = 7 # complete
|
|
self.mock_router.propagation_transfer_progress = 1.0
|
|
self.mock_router.propagation_transfer_last_result = 3
|
|
result = self.wrapper.get_propagation_state()
|
|
self.assertTrue(result['success'])
|
|
self.assertEqual(result['state'], 7)
|
|
self.assertEqual(result['state_name'], 'complete')
|
|
self.assertEqual(result['messages_received'], 3)
|
|
|
|
def test_cannot_request_without_setting_node(self):
|
|
"""Test that requesting messages fails if no node is set"""
|
|
# Ensure no node is set
|
|
self.wrapper.active_propagation_node = None
|
|
|
|
# Try to request messages
|
|
result = self.wrapper.request_messages_from_propagation_node()
|
|
|
|
# Verify failure
|
|
self.assertFalse(result['success'])
|
|
self.assertEqual(result['errorCode'], 'NO_PROPAGATION_NODE')
|
|
|
|
def test_can_get_state_without_active_transfer(self):
|
|
"""Test that getting state works even without active transfer"""
|
|
self.mock_router.propagation_transfer_state = 0
|
|
self.mock_router.propagation_transfer_progress = 0.0
|
|
|
|
result = self.wrapper.get_propagation_state()
|
|
|
|
self.assertTrue(result['success'])
|
|
self.assertEqual(result['state'], 0)
|
|
self.assertEqual(result['state_name'], 'idle')
|
|
|
|
def test_clearing_node_resets_internal_state(self):
|
|
"""Test that clearing node resets wrapper's internal state"""
|
|
# Set a node first
|
|
node_hash = b'propagation_node'
|
|
self.wrapper.set_outbound_propagation_node(node_hash)
|
|
self.assertEqual(self.wrapper.active_propagation_node, node_hash)
|
|
|
|
# Clear the node
|
|
result = self.wrapper.set_outbound_propagation_node(None)
|
|
self.assertTrue(result['success'])
|
|
|
|
# Verify internal state cleared
|
|
self.assertIsNone(self.wrapper.active_propagation_node)
|
|
|
|
# Verify can't request messages after clearing
|
|
result = self.wrapper.request_messages_from_propagation_node()
|
|
self.assertFalse(result['success'])
|
|
self.assertEqual(result['errorCode'], 'NO_PROPAGATION_NODE')
|
|
|
|
|
|
class TestSetIncomingMessageSizeLimit(PropagationTestBase):
|
|
"""Test set_incoming_message_size_limit method"""
|
|
|
|
def setUp(self):
|
|
"""Set up test fixtures"""
|
|
super().setUp()
|
|
import tempfile
|
|
self.temp_dir = tempfile.mkdtemp()
|
|
self.wrapper = reticulum_wrapper.ReticulumWrapper(self.temp_dir)
|
|
|
|
# Enable Reticulum
|
|
reticulum_wrapper.RETICULUM_AVAILABLE = True
|
|
|
|
# Mock router
|
|
self.mock_router = Mock()
|
|
self.wrapper.router = self.mock_router
|
|
self.wrapper.initialized = True
|
|
|
|
def tearDown(self):
|
|
"""Clean up test fixtures"""
|
|
super().tearDown()
|
|
import shutil
|
|
if os.path.exists(self.temp_dir):
|
|
shutil.rmtree(self.temp_dir)
|
|
|
|
def test_set_size_limit_success(self):
|
|
"""Test successfully setting incoming message size limit"""
|
|
result = self.wrapper.set_incoming_message_size_limit(1024)
|
|
|
|
# Verify success
|
|
self.assertTrue(result['success'])
|
|
|
|
# Verify both limits were set
|
|
self.assertEqual(self.mock_router.delivery_per_transfer_limit, 1024)
|
|
self.assertEqual(self.mock_router.propagation_per_transfer_limit, 1024)
|
|
|
|
def test_set_size_limit_large_value(self):
|
|
"""Test setting large size limit (128MB for 'unlimited')"""
|
|
result = self.wrapper.set_incoming_message_size_limit(131072)
|
|
|
|
self.assertTrue(result['success'])
|
|
self.assertEqual(self.mock_router.delivery_per_transfer_limit, 131072)
|
|
self.assertEqual(self.mock_router.propagation_per_transfer_limit, 131072)
|
|
|
|
def test_set_size_limit_not_initialized(self):
|
|
"""Test error when wrapper not initialized"""
|
|
self.wrapper.initialized = False
|
|
|
|
result = self.wrapper.set_incoming_message_size_limit(1024)
|
|
|
|
self.assertFalse(result['success'])
|
|
self.assertIn('error', result)
|
|
self.assertIn('not initialized', result['error'].lower())
|
|
|
|
def test_set_size_limit_no_router(self):
|
|
"""Test error when router is not available"""
|
|
self.wrapper.router = None
|
|
|
|
result = self.wrapper.set_incoming_message_size_limit(1024)
|
|
|
|
self.assertFalse(result['success'])
|
|
self.assertIn('error', result)
|
|
|
|
def test_set_size_limit_reticulum_unavailable(self):
|
|
"""Test error when Reticulum is not available"""
|
|
reticulum_wrapper.RETICULUM_AVAILABLE = False
|
|
|
|
result = self.wrapper.set_incoming_message_size_limit(1024)
|
|
|
|
self.assertFalse(result['success'])
|
|
self.assertIn('error', result)
|
|
|
|
def test_set_size_limit_router_exception(self):
|
|
"""Test handling of router exceptions"""
|
|
# Make setting limit raise an exception
|
|
type(self.mock_router).delivery_per_transfer_limit = PropertyMock(
|
|
side_effect=Exception("Setting limit failed")
|
|
)
|
|
|
|
result = self.wrapper.set_incoming_message_size_limit(1024)
|
|
|
|
self.assertFalse(result['success'])
|
|
self.assertIn('error', result)
|
|
|
|
|
|
class TestExtractFileSummary(PropagationTestBase):
|
|
"""Test _extract_file_summary method"""
|
|
|
|
def setUp(self):
|
|
"""Set up test fixtures"""
|
|
super().setUp()
|
|
import tempfile
|
|
self.temp_dir = tempfile.mkdtemp()
|
|
self.wrapper = reticulum_wrapper.ReticulumWrapper(self.temp_dir)
|
|
|
|
def tearDown(self):
|
|
"""Clean up test fixtures"""
|
|
super().tearDown()
|
|
import shutil
|
|
if os.path.exists(self.temp_dir):
|
|
shutil.rmtree(self.temp_dir)
|
|
|
|
def test_extract_file_summary_single_file(self):
|
|
"""Test extracting summary from message with single file attachment"""
|
|
mock_message = Mock()
|
|
mock_message.fields = {
|
|
5: [['document.pdf', b'PDF content here with more data']]
|
|
}
|
|
|
|
result = self.wrapper._extract_file_summary(mock_message)
|
|
|
|
self.assertIsNotNone(result)
|
|
self.assertEqual(result['first_filename'], 'document.pdf')
|
|
self.assertEqual(result['file_count'], 1)
|
|
self.assertEqual(result['total_size'], 31) # len(b'PDF content here with more data')
|
|
|
|
def test_extract_file_summary_multiple_files(self):
|
|
"""Test extracting summary from message with multiple file attachments"""
|
|
mock_message = Mock()
|
|
mock_message.fields = {
|
|
5: [
|
|
['file1.txt', b'First file content'],
|
|
['file2.bin', b'\x00\x01\x02\x03'],
|
|
['file3.pdf', b'PDF data']
|
|
]
|
|
}
|
|
|
|
result = self.wrapper._extract_file_summary(mock_message)
|
|
|
|
self.assertIsNotNone(result)
|
|
self.assertEqual(result['first_filename'], 'file1.txt')
|
|
self.assertEqual(result['file_count'], 3)
|
|
self.assertEqual(result['total_size'], 18 + 4 + 8) # Sum of all file sizes
|
|
|
|
def test_extract_file_summary_tuple_format(self):
|
|
"""Test extracting summary from message with tuple format attachments"""
|
|
mock_message = Mock()
|
|
mock_message.fields = {
|
|
5: [('image.jpg', b'\xff\xd8\xff\xe0\x00\x10JFIF')]
|
|
}
|
|
|
|
result = self.wrapper._extract_file_summary(mock_message)
|
|
|
|
self.assertIsNotNone(result)
|
|
self.assertEqual(result['first_filename'], 'image.jpg')
|
|
self.assertEqual(result['file_count'], 1)
|
|
|
|
def test_extract_file_summary_no_fields(self):
|
|
"""Test when message has no fields"""
|
|
mock_message = Mock()
|
|
mock_message.fields = None
|
|
|
|
result = self.wrapper._extract_file_summary(mock_message)
|
|
|
|
self.assertIsNone(result)
|
|
|
|
def test_extract_file_summary_no_file_field(self):
|
|
"""Test when message has fields but no file attachments (field 5)"""
|
|
mock_message = Mock()
|
|
mock_message.fields = {6: ['jpg', b'image data']} # Image field, not file field
|
|
|
|
result = self.wrapper._extract_file_summary(mock_message)
|
|
|
|
self.assertIsNone(result)
|
|
|
|
def test_extract_file_summary_empty_attachments(self):
|
|
"""Test when file attachments list is empty"""
|
|
mock_message = Mock()
|
|
mock_message.fields = {5: []}
|
|
|
|
result = self.wrapper._extract_file_summary(mock_message)
|
|
|
|
self.assertIsNone(result)
|
|
|
|
def test_extract_file_summary_invalid_attachment_format(self):
|
|
"""Test when attachments have invalid format"""
|
|
mock_message = Mock()
|
|
mock_message.fields = {
|
|
5: [
|
|
'invalid_string', # Not a list/tuple
|
|
['valid.txt', b'content'], # Valid
|
|
[123], # Too short
|
|
]
|
|
}
|
|
|
|
result = self.wrapper._extract_file_summary(mock_message)
|
|
|
|
self.assertIsNotNone(result)
|
|
self.assertEqual(result['file_count'], 1) # Only the valid one counted
|
|
|
|
def test_extract_file_summary_none_filename(self):
|
|
"""Test handling of None filename"""
|
|
mock_message = Mock()
|
|
mock_message.fields = {
|
|
5: [[None, b'content without name']]
|
|
}
|
|
|
|
result = self.wrapper._extract_file_summary(mock_message)
|
|
|
|
self.assertIsNotNone(result)
|
|
self.assertEqual(result['first_filename'], 'file') # Default fallback
|
|
|
|
def test_extract_file_summary_no_hasattr(self):
|
|
"""Test when message doesn't have fields attribute"""
|
|
mock_message = Mock(spec=[]) # No attributes
|
|
|
|
result = self.wrapper._extract_file_summary(mock_message)
|
|
|
|
self.assertIsNone(result)
|
|
|
|
|
|
class TestFailMessagePermanently(PropagationTestBase):
|
|
"""Test _fail_message_permanently method"""
|
|
|
|
def setUp(self):
|
|
"""Set up test fixtures"""
|
|
super().setUp()
|
|
import tempfile
|
|
self.temp_dir = tempfile.mkdtemp()
|
|
self.wrapper = reticulum_wrapper.ReticulumWrapper(self.temp_dir)
|
|
|
|
def tearDown(self):
|
|
"""Clean up test fixtures"""
|
|
super().tearDown()
|
|
import shutil
|
|
if os.path.exists(self.temp_dir):
|
|
shutil.rmtree(self.temp_dir)
|
|
|
|
def test_fail_message_notifies_kotlin(self):
|
|
"""Test that permanent failure notifies Kotlin callback"""
|
|
mock_callback = Mock()
|
|
self.wrapper.kotlin_delivery_status_callback = mock_callback
|
|
|
|
mock_message = Mock()
|
|
mock_message.hash = b'messagehash12345'
|
|
|
|
self.wrapper._fail_message_permanently(mock_message, 'max_relay_retries_exceeded')
|
|
|
|
# Kotlin callback should be invoked
|
|
mock_callback.assert_called_once()
|
|
call_arg = mock_callback.call_args[0][0]
|
|
|
|
# Parse JSON
|
|
import json
|
|
status_event = json.loads(call_arg)
|
|
self.assertEqual(status_event['status'], 'failed')
|
|
self.assertEqual(status_event['reason'], 'max_relay_retries_exceeded')
|
|
self.assertEqual(status_event['message_hash'], mock_message.hash.hex())
|
|
self.assertIn('timestamp', status_event)
|
|
|
|
def test_fail_message_removes_from_pending(self):
|
|
"""Test that permanent failure removes message from pending fallback"""
|
|
mock_message = Mock()
|
|
mock_message.hash = b'messagehash12345'
|
|
|
|
# Add to pending
|
|
self.wrapper._pending_relay_fallback_messages[mock_message.hash.hex()] = mock_message
|
|
|
|
self.wrapper._fail_message_permanently(mock_message, 'no_relays_available')
|
|
|
|
# Should be removed from pending
|
|
self.assertNotIn(mock_message.hash.hex(), self.wrapper._pending_relay_fallback_messages)
|
|
|
|
def test_fail_message_no_callback(self):
|
|
"""Test that permanent failure works without Kotlin callback"""
|
|
self.wrapper.kotlin_delivery_status_callback = None
|
|
|
|
mock_message = Mock()
|
|
mock_message.hash = b'messagehash12345'
|
|
|
|
# Should not raise exception
|
|
self.wrapper._fail_message_permanently(mock_message, 'test_failure')
|
|
|
|
def test_fail_message_callback_exception(self):
|
|
"""Test handling of exception in Kotlin callback"""
|
|
mock_callback = Mock(side_effect=Exception("Callback error"))
|
|
self.wrapper.kotlin_delivery_status_callback = mock_callback
|
|
|
|
mock_message = Mock()
|
|
mock_message.hash = b'messagehash12345'
|
|
|
|
# Should not raise exception
|
|
self.wrapper._fail_message_permanently(mock_message, 'test_failure')
|
|
|
|
|
|
class TestOnAlternativeRelayReceived(PropagationTestBase):
|
|
"""Test on_alternative_relay_received method"""
|
|
|
|
def setUp(self):
|
|
"""Set up test fixtures"""
|
|
super().setUp()
|
|
import tempfile
|
|
self.temp_dir = tempfile.mkdtemp()
|
|
self.wrapper = reticulum_wrapper.ReticulumWrapper(self.temp_dir)
|
|
|
|
# Enable Reticulum
|
|
reticulum_wrapper.RETICULUM_AVAILABLE = True
|
|
|
|
# Mock router
|
|
self.mock_router = Mock()
|
|
self.wrapper.router = self.mock_router
|
|
self.wrapper.initialized = True
|
|
|
|
def tearDown(self):
|
|
"""Clean up test fixtures"""
|
|
super().tearDown()
|
|
import shutil
|
|
if os.path.exists(self.temp_dir):
|
|
shutil.rmtree(self.temp_dir)
|
|
|
|
def test_alternative_relay_no_pending_messages(self):
|
|
"""Test when no pending messages for relay fallback"""
|
|
self.wrapper._pending_relay_fallback_messages = {}
|
|
|
|
# Should not crash
|
|
self.wrapper.on_alternative_relay_received(b'newrelay12345678')
|
|
|
|
def test_alternative_relay_none_fails_all(self):
|
|
"""Test that None relay hash fails all pending messages"""
|
|
mock_callback = Mock()
|
|
self.wrapper.kotlin_delivery_status_callback = mock_callback
|
|
|
|
mock_message1 = Mock()
|
|
mock_message1.hash = b'message1hash1234'
|
|
mock_message2 = Mock()
|
|
mock_message2.hash = b'message2hash1234'
|
|
|
|
self.wrapper._pending_relay_fallback_messages = {
|
|
mock_message1.hash.hex(): mock_message1,
|
|
mock_message2.hash.hex(): mock_message2,
|
|
}
|
|
|
|
self.wrapper.on_alternative_relay_received(None)
|
|
|
|
# All pending should be cleared
|
|
self.assertEqual(len(self.wrapper._pending_relay_fallback_messages), 0)
|
|
|
|
# Callback should be called for each failed message
|
|
self.assertEqual(mock_callback.call_count, 2)
|
|
|
|
def test_alternative_relay_retries_with_new_relay(self):
|
|
"""Test that alternative relay retries pending messages"""
|
|
mock_callback = Mock()
|
|
self.wrapper.kotlin_delivery_status_callback = mock_callback
|
|
|
|
mock_message = Mock()
|
|
mock_message.hash = b'messagehash12345'
|
|
mock_message.tried_relays = []
|
|
mock_message.fields = {}
|
|
|
|
self.wrapper._pending_relay_fallback_messages = {
|
|
mock_message.hash.hex(): mock_message
|
|
}
|
|
|
|
new_relay = b'newrelay12345678'
|
|
self.wrapper.on_alternative_relay_received(new_relay)
|
|
|
|
# Active propagation node should be updated
|
|
self.assertEqual(self.wrapper.active_propagation_node, new_relay)
|
|
|
|
# Message should be resubmitted
|
|
self.mock_router.handle_outbound.assert_called_once_with(mock_message)
|
|
|
|
# Pending should be cleared
|
|
self.assertEqual(len(self.wrapper._pending_relay_fallback_messages), 0)
|
|
|
|
# Status callback should be invoked
|
|
mock_callback.assert_called()
|
|
call_arg = mock_callback.call_args[0][0]
|
|
import json
|
|
status = json.loads(call_arg)
|
|
self.assertEqual(status['status'], 'retrying_propagated')
|
|
|
|
def test_alternative_relay_converts_jarray(self):
|
|
"""Test that jarray-like relay hash is converted to bytes"""
|
|
mock_message = Mock()
|
|
mock_message.hash = b'messagehash12345'
|
|
mock_message.tried_relays = []
|
|
mock_message.fields = {}
|
|
|
|
self.wrapper._pending_relay_fallback_messages = {
|
|
mock_message.hash.hex(): mock_message
|
|
}
|
|
|
|
# Pass list instead of bytes (simulating jarray from Java)
|
|
relay_jarray = [0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08,
|
|
0x09, 0x0a, 0x0b, 0x0c, 0x0d, 0x0e, 0x0f, 0x10]
|
|
|
|
self.wrapper.on_alternative_relay_received(relay_jarray)
|
|
|
|
# Should be converted to bytes
|
|
self.assertIsInstance(self.wrapper.active_propagation_node, bytes)
|
|
self.assertEqual(self.wrapper.active_propagation_node, bytes(relay_jarray))
|
|
|
|
def test_alternative_relay_tracks_tried_relays(self):
|
|
"""Test that tried relays are tracked on the message"""
|
|
mock_message = Mock()
|
|
mock_message.hash = b'messagehash12345'
|
|
mock_message.tried_relays = [b'previousrelay123']
|
|
mock_message.fields = {}
|
|
|
|
self.wrapper._pending_relay_fallback_messages = {
|
|
mock_message.hash.hex(): mock_message
|
|
}
|
|
|
|
new_relay = b'newrelay12345678'
|
|
self.wrapper.on_alternative_relay_received(new_relay)
|
|
|
|
# New relay should be added to tried list
|
|
self.assertIn(new_relay, mock_message.tried_relays)
|
|
self.assertEqual(len(mock_message.tried_relays), 2)
|
|
|
|
def test_alternative_relay_resets_message_state(self):
|
|
"""Test that message state is reset for fresh retry"""
|
|
mock_message = Mock()
|
|
mock_message.hash = b'messagehash12345'
|
|
mock_message.tried_relays = []
|
|
mock_message.delivery_attempts = 5
|
|
mock_message.packed = b'old_packed'
|
|
mock_message.propagation_packed = b'old_prop_packed'
|
|
mock_message.propagation_stamp = 'old_stamp'
|
|
mock_message.fields = {}
|
|
|
|
self.wrapper._pending_relay_fallback_messages = {
|
|
mock_message.hash.hex(): mock_message
|
|
}
|
|
|
|
self.wrapper.on_alternative_relay_received(b'newrelay12345678')
|
|
|
|
# State should be reset
|
|
self.assertEqual(mock_message.delivery_attempts, 0)
|
|
self.assertIsNone(mock_message.packed)
|
|
self.assertIsNone(mock_message.propagation_packed)
|
|
self.assertIsNone(mock_message.propagation_stamp)
|
|
self.assertTrue(mock_message.defer_propagation_stamp)
|
|
self.assertEqual(mock_message.desired_method, lxmf_mock.LXMessage.PROPAGATED)
|
|
|
|
def test_alternative_relay_updates_router(self):
|
|
"""Test that router propagation node is updated"""
|
|
mock_message = Mock()
|
|
mock_message.hash = b'messagehash12345'
|
|
mock_message.tried_relays = []
|
|
mock_message.fields = {}
|
|
|
|
self.wrapper._pending_relay_fallback_messages = {
|
|
mock_message.hash.hex(): mock_message
|
|
}
|
|
|
|
new_relay = b'newrelay12345678'
|
|
self.wrapper.on_alternative_relay_received(new_relay)
|
|
|
|
# Router should be updated
|
|
self.mock_router.set_outbound_propagation_node.assert_called_once_with(new_relay)
|
|
|
|
|
|
class TestSendPendingFileNotification(PropagationTestBase):
|
|
"""Test _send_pending_file_notification method"""
|
|
|
|
def setUp(self):
|
|
"""Set up test fixtures"""
|
|
super().setUp()
|
|
import tempfile
|
|
self.temp_dir = tempfile.mkdtemp()
|
|
self.wrapper = reticulum_wrapper.ReticulumWrapper(self.temp_dir)
|
|
|
|
# Enable Reticulum
|
|
reticulum_wrapper.RETICULUM_AVAILABLE = True
|
|
|
|
# Mock router
|
|
self.mock_router = Mock()
|
|
self.wrapper.router = self.mock_router
|
|
self.wrapper.initialized = True
|
|
|
|
def tearDown(self):
|
|
"""Clean up test fixtures"""
|
|
super().tearDown()
|
|
import shutil
|
|
if os.path.exists(self.temp_dir):
|
|
shutil.rmtree(self.temp_dir)
|
|
|
|
def test_send_notification_for_file_message(self):
|
|
"""Test sending notification for message with file attachments"""
|
|
mock_message = Mock()
|
|
mock_message.hash = b'messagehash12345'
|
|
mock_message.destination = Mock()
|
|
mock_message.source = Mock()
|
|
mock_message.fields = {
|
|
5: [['document.pdf', b'PDF content']]
|
|
}
|
|
|
|
self.wrapper._send_pending_file_notification(mock_message)
|
|
|
|
# Router should have received a notification message
|
|
self.mock_router.handle_outbound.assert_called_once()
|
|
notification_msg = self.mock_router.handle_outbound.call_args[0][0]
|
|
|
|
# Verify it's an OPPORTUNISTIC message
|
|
self.assertEqual(notification_msg.desired_method, lxmf_mock.LXMessage.OPPORTUNISTIC)
|
|
|
|
def test_send_notification_skips_no_files(self):
|
|
"""Test that notification is skipped when no file attachments"""
|
|
mock_message = Mock()
|
|
mock_message.hash = b'messagehash12345'
|
|
mock_message.fields = None
|
|
|
|
self.wrapper._send_pending_file_notification(mock_message)
|
|
|
|
# Router should NOT have received any message
|
|
self.mock_router.handle_outbound.assert_not_called()
|
|
|
|
def test_send_notification_handles_exception(self):
|
|
"""Test that exceptions in notification sending are handled"""
|
|
mock_message = Mock()
|
|
mock_message.hash = b'messagehash12345'
|
|
mock_message.destination = Mock()
|
|
mock_message.source = Mock()
|
|
mock_message.fields = {
|
|
5: [['file.txt', b'content']]
|
|
}
|
|
|
|
# Make handle_outbound raise
|
|
self.mock_router.handle_outbound.side_effect = Exception("Send failed")
|
|
|
|
# Should not raise exception
|
|
self.wrapper._send_pending_file_notification(mock_message)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
# Run tests with verbose output
|
|
unittest.main(verbosity=2)
|