RNS-Things/MeshChatX
2
0
Fork 0
Code Issues 8 Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(tests): formatting

Browse Source
This commit is contained in:
ivan
2026-01-07 19:51:48 -06:00
parent e02e17d712
commit 1e5564cfa3
1 changed files with 32 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
tests/backend/test_telemetry_integration.py
View File

@@ -32,7 +32,9 @@ def mock_app():
# Attach the actual method we want to test if possible, # Attach the actual method we want to test if possible,
# but since it's an instance method, we might need to bind it. # but since it's an instance method, we might need to bind it.
app.process_incoming_telemetry = ReticulumMeshChat.process_incoming_telemetry.__get__(app, ReticulumMeshChat) app.process_incoming_telemetry = (
ReticulumMeshChat.process_incoming_telemetry.__get__(app, ReticulumMeshChat)
)
return app return app
@@ -46,7 +48,9 @@ async def test_process_incoming_telemetry_single(mock_app):
mock_lxmf_message = MagicMock() mock_lxmf_message = MagicMock()
mock_lxmf_message.hash = b"msg_hash" mock_lxmf_message.hash = b"msg_hash"
mock_app.process_incoming_telemetry(source_hash, packed_telemetry, mock_lxmf_message) mock_app.process_incoming_telemetry(
source_hash, packed_telemetry, mock_lxmf_message
)
# Verify database call # Verify database call
mock_app.database.telemetry.upsert_telemetry.assert_called() mock_app.database.telemetry.upsert_telemetry.assert_called()
@@ -78,7 +82,10 @@ async def test_process_incoming_telemetry_stream(mock_app):
# to handle single entries, and on_lxmf_delivery loops over streams. # to handle single entries, and on_lxmf_delivery loops over streams.
for entry_source, entry_timestamp, entry_data in entries: for entry_source, entry_timestamp, entry_data in entries:
mock_app.process_incoming_telemetry( mock_app.process_incoming_telemetry(
entry_source, entry_data, mock_lxmf_message, timestamp_override=entry_timestamp entry_source,
entry_data,
mock_lxmf_message,
timestamp_override=entry_timestamp,
) )
assert mock_app.database.telemetry.upsert_telemetry.call_count == 2 assert mock_app.database.telemetry.upsert_telemetry.call_count == 2
@@ -100,19 +107,25 @@ async def test_telemetry_request_parsing(mock_app):
mock_app.handle_telemetry_request = MagicMock() mock_app.handle_telemetry_request = MagicMock()
# Bind on_lxmf_delivery # Bind on_lxmf_delivery
mock_app.on_lxmf_delivery = ReticulumMeshChat.on_lxmf_delivery.__get__(mock_app, ReticulumMeshChat) mock_app.on_lxmf_delivery = ReticulumMeshChat.on_lxmf_delivery.__get__(
mock_app, ReticulumMeshChat
)
# Mocking dependencies # Mocking dependencies
mock_app.is_destination_blocked.return_value = False mock_app.is_destination_blocked.return_value = False
mock_app.current_context.config.telemetry_enabled.get.return_value = True mock_app.current_context.config.telemetry_enabled.get.return_value = True
mock_app.database.contacts.get_contact_by_identity_hash.return_value = {"is_telemetry_trusted": True} mock_app.database.contacts.get_contact_by_identity_hash.return_value = {
mock_app.database.messages.get_lxmf_message_by_hash.return_value = {} # To avoid JSON error "is_telemetry_trusted": True
}
mock_app.database.messages.get_lxmf_message_by_hash.return_value = {} # To avoid JSON error
# Call it # Call it
mock_app.on_lxmf_delivery(mock_lxmf_message) mock_app.on_lxmf_delivery(mock_lxmf_message)
# Verify handle_telemetry_request was called # Verify handle_telemetry_request was called
mock_app.handle_telemetry_request.assert_called_with("736f757263655f686173685f6279746573") mock_app.handle_telemetry_request.assert_called_with(
"736f757263655f686173685f6279746573"
)
@pytest.mark.asyncio @pytest.mark.asyncio