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

refactor(tests): replace db.close() with db.close_all() in multiple test files and ensure proper teardown of ReticulumMeshChat instances

Browse Source
This commit is contained in:
ivan
2026-01-03 21:36:42 -06:00
parent 151b69ad50
commit 8c5a68a01f
7 changed files with 69 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
tests/backend/test_memory_profiling.py
View File

@@ -16,7 +16,7 @@ class TestMemoryProfiling(unittest.TestCase):
self.db.initialize()
def tearDown(self):
self.db.close()
self.db.close_all()
if os.path.exists(self.test_dir):
shutil.rmtree(self.test_dir)

18
tests/backend/test_notifications.py
View File

@@ -29,7 +29,8 @@ def db(temp_db):
schema.initialize()
database = Database(temp_db)
yield database
database.close()
database.close_all()
provider.close_all()
@pytest.fixture
@@ -110,6 +111,18 @@ def mock_app(db, tmp_path):
)
)
stack.enter_context(
patch.object(
ReticulumMeshChat, "auto_backup_loop", new=MagicMock(return_value=None)
)
)
# Prevent JSON serialization issues with MagicMocks
stack.enter_context(
patch.object(
ReticulumMeshChat, "send_config_to_websocket_clients", return_value=None
)
)
app = ReticulumMeshChat(
identity=mock_id,
storage_dir=str(tmp_path),
@@ -120,7 +133,8 @@ def mock_app(db, tmp_path):
app.database = db
app.websocket_broadcast = MagicMock(side_effect=lambda data: None)
return app
yield app
app.teardown_identity()
def test_add_get_notifications(db):

2
tests/backend/test_performance_bottlenecks.py
View File

@@ -22,7 +22,7 @@ class TestPerformanceBottlenecks(unittest.TestCase):
self.reticulum_mock.get_packet_q.return_value = 3
def tearDown(self):
self.db.close()
self.db.close_all()
shutil.rmtree(self.test_dir)
def test_message_pagination_performance(self):

12
tests/backend/test_rns_lifecycle.py
View File

@@ -39,6 +39,14 @@ def mock_rns():
"crawler_loop",
new=MagicMock(return_value=None),
),
patch.object(
ReticulumMeshChat,
"auto_backup_loop",
new=MagicMock(return_value=None),
),
patch.object(
ReticulumMeshChat, "send_config_to_websocket_clients", return_value=None
),
):
# Setup mock instance
mock_id_instance = MockIdentityClass()
@@ -116,6 +124,7 @@ async def test_cleanup_rns_state_for_identity(mock_rns, temp_dir):
# Verify deregistration and teardown were called
mock_rns["Transport"].deregister_destination.assert_called_with(mock_dest)
mock_link.teardown.assert_called()
app.teardown_identity()
@pytest.mark.asyncio
@@ -198,6 +207,7 @@ async def test_reload_reticulum(mock_rns, temp_dir):
assert mock_rns["Reticulum"]._Reticulum__instance is None
# Verify setup_identity was called again
app.setup_identity.assert_called()
app.teardown_identity()
@pytest.mark.asyncio
@@ -245,6 +255,7 @@ async def test_reload_reticulum_failure_recovery(mock_rns, temp_dir):
assert result is False
# Verify recovery: setup_identity should be called because hasattr(self, "reticulum") is False
app.setup_identity.assert_called()
app.teardown_identity()
@pytest.mark.asyncio
@@ -294,3 +305,4 @@ async def test_hotswap_identity(mock_rns, temp_dir):
# Check if the broadcast contains identity_switched
broadcast_call = app.websocket_broadcast.call_args[0][0]
assert "identity_switched" in broadcast_call
app.teardown_identity()

13
tests/backend/test_security_fuzzing.py
View File

@@ -63,6 +63,8 @@ def mock_app():
stack.enter_context(patch("RNS.Identity", MockIdentityClass))
stack.enter_context(patch("RNS.Reticulum"))
stack.enter_context(patch("RNS.Transport"))
mock_packet = stack.enter_context(patch("RNS.Packet"))
mock_packet.MTU = 500 # Set a default MTU to avoid MagicMock comparison errors
stack.enter_context(patch("threading.Thread"))
stack.enter_context(
patch.object(
@@ -81,6 +83,16 @@ def mock_app():
ReticulumMeshChat, "crawler_loop", new=MagicMock(return_value=None)
),
)
stack.enter_context(
patch.object(
ReticulumMeshChat, "auto_backup_loop", new=MagicMock(return_value=None)
),
)
stack.enter_context(
patch.object(
ReticulumMeshChat, "send_config_to_websocket_clients", return_value=None
),
)
mock_id = MockIdentityClass()
mock_id.get_private_key = MagicMock(return_value=b"test_private_key")
@@ -132,6 +144,7 @@ def mock_app():
app.websocket_broadcast = MagicMock()
yield app
app.teardown_identity()
@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)

15
tests/backend/test_startup.py
View File

@@ -32,6 +32,15 @@ def mock_rns():
patch("threading.Thread") as mock_thread,
patch("LXMF.LXMRouter") as mock_lxmf_router,
patch("meshchatx.meshchat.get_file_path", return_value="/tmp/mock_path"),
patch.object(ReticulumMeshChat, "announce_loop", return_value=None),
patch.object(
ReticulumMeshChat, "announce_sync_propagation_nodes", return_value=None
),
patch.object(ReticulumMeshChat, "crawler_loop", return_value=None),
patch.object(ReticulumMeshChat, "auto_backup_loop", return_value=None),
patch.object(
ReticulumMeshChat, "send_config_to_websocket_clients", return_value=None
),
):
# Setup mock instance
mock_id_instance = MockIdentityClass()
@@ -140,6 +149,8 @@ def test_reticulum_meshchat_init(mock_rns, temp_dir):
# There should be at least 3 threads: announce_loop, announce_sync_propagation_nodes, crawler_loop
assert mock_rns["Thread"].call_count >= 3
app.teardown_identity()
def test_reticulum_meshchat_init_with_auth(mock_rns, temp_dir):
with (
@@ -173,6 +184,7 @@ def test_reticulum_meshchat_init_with_auth(mock_rns, temp_dir):
)
assert app.auth_enabled is True
app.teardown_identity()
def test_reticulum_meshchat_init_database_failure_recovery(mock_rns, temp_dir):
@@ -199,7 +211,7 @@ def test_reticulum_meshchat_init_database_failure_recovery(mock_rns, temp_dir):
# Fail the first initialize call
mock_db_instance.initialize.side_effect = [Exception("DB Error"), None]
_ = ReticulumMeshChat(
app = ReticulumMeshChat(
identity=mock_rns["id_instance"],
storage_dir=temp_dir,
reticulum_config_dir=temp_dir,
@@ -208,3 +220,4 @@ def test_reticulum_meshchat_init_database_failure_recovery(mock_rns, temp_dir):
assert mock_recovery.called
assert mock_db_instance.initialize.call_count == 2
app.teardown_identity()

13
tests/backend/test_startup_advanced.py
View File

@@ -32,6 +32,15 @@ def mock_rns():
patch("RNS.Identity", MockIdentityClass),
patch("threading.Thread"),
patch("LXMF.LXMRouter"),
patch.object(ReticulumMeshChat, "announce_loop", return_value=None),
patch.object(
ReticulumMeshChat, "announce_sync_propagation_nodes", return_value=None
),
patch.object(ReticulumMeshChat, "crawler_loop", return_value=None),
patch.object(ReticulumMeshChat, "auto_backup_loop", return_value=None),
patch.object(
ReticulumMeshChat, "send_config_to_websocket_clients", return_value=None
),
):
mock_id_instance = MockIdentityClass()
mock_id_instance.get_private_key = MagicMock(return_value=b"test_private_key")
@@ -112,6 +121,7 @@ def test_run_https_logic(mock_rns, temp_dir):
app.run(host="127.0.0.1", port=8000, launch_browser=False, enable_https=False)
args, kwargs = mock_run_app.call_args
assert kwargs.get("ssl_context") is None
app.teardown_identity()
# 2. Test specific database integrity failure recovery
@@ -152,7 +162,7 @@ def test_database_integrity_recovery(mock_rns, temp_dir):
mock_config.auth_session_secret.get.return_value = "test_secret"
mock_config.display_name.get.return_value = "Test"
_ = ReticulumMeshChat(
app = ReticulumMeshChat(
identity=mock_rns["id_instance"],
storage_dir=temp_dir,
reticulum_config_dir=temp_dir,
@@ -164,6 +174,7 @@ def test_database_integrity_recovery(mock_rns, temp_dir):
assert mock_db_instance.provider.integrity_check.called
assert mock_db_instance.provider.vacuum.called
assert mock_db_instance._tune_sqlite_pragmas.called
app.teardown_identity()
# 3. Test missing critical files (identity)