diff options
Diffstat (limited to 'packages/meshbay-node/tests/test_daemon.py')
| -rw-r--r-- | packages/meshbay-node/tests/test_daemon.py | 143 |
1 files changed, 143 insertions, 0 deletions
diff --git a/packages/meshbay-node/tests/test_daemon.py b/packages/meshbay-node/tests/test_daemon.py index ab1613d..b367a20 100644 --- a/packages/meshbay-node/tests/test_daemon.py +++ b/packages/meshbay-node/tests/test_daemon.py @@ -237,6 +237,7 @@ async def test_daemon_index_change_pushes_to_peers(tmp_path, shared_dir, gek, hu data_dir=tmp_path / "data", ) daemon = NodeDaemon(config) + daemon._broadcast_coalesce_secs = 0.01 # real value would make this test wait 0.5s daemon._hub = AsyncMock() daemon._hub.register_swarm = AsyncMock(return_value=2) daemon._state["endpoint_hint"] = "node123" @@ -256,6 +257,7 @@ async def test_daemon_index_change_pushes_to_peers(tmp_path, shared_dir, gek, hu daemon._webrtc = mock_webrtc await daemon._on_index_change(indexer) + await asyncio.sleep(0.05) # let the coalescing timer fire mock_session._send.assert_called_once() msg = mock_session._send.call_args[0][0] @@ -288,6 +290,7 @@ async def test_daemon_index_change_registers_swarm_for_public_group( data_dir=tmp_path / "data", ) daemon = NodeDaemon(config) + daemon._broadcast_coalesce_secs = 0.01 daemon._hub = AsyncMock() daemon._hub.register_swarm = AsyncMock(return_value=2) daemon._state["endpoint_hint"] = "node123" @@ -317,6 +320,7 @@ async def test_daemon_index_change_skips_other_group_peers( data_dir=tmp_path / "data", ) daemon = NodeDaemon(config) + daemon._broadcast_coalesce_secs = 0.01 daemon._hub = AsyncMock() daemon._hub.register_swarm = AsyncMock(return_value=0) daemon._state["endpoint_hint"] = "node123" @@ -340,6 +344,145 @@ async def test_daemon_index_change_skips_other_group_peers( daemon._webrtc = mock_webrtc await daemon._on_index_change(indexer) + await asyncio.sleep(0.05) same_group._send.assert_called_once() other_group._send.assert_not_called() + + +# ── INDEX_DELTA (phase 4) ──────────────────────────────────────────────────── + +def _new_daemon_for_group(tmp_path, shared_dir, gek, group_id="a" * 32, + visibility="private"): + config = Config( + hub=HubConfig(url="http://localhost:9999", username="testuser"), + node=NodeConfig(quic_port=_free_port(), ui_port=_free_port()), + groups=[GroupConfig( + id=group_id, name="test-group", shared_dir=str(shared_dir), + visibility=visibility, quic_port=29010, + )], + keystore=KeystoreConfig(path=tmp_path / "keystore.enc"), + data_dir=tmp_path / "data", + ) + daemon = NodeDaemon(config) + daemon._broadcast_coalesce_secs = 0.01 + daemon._hub = AsyncMock() + daemon._hub.register_swarm = AsyncMock(return_value=0) + daemon._state["endpoint_hint"] = "node123" + return daemon + + +@pytest.mark.asyncio +async def test_first_broadcast_is_full_sync_second_is_delta(tmp_path, shared_dir, gek): + daemon = _new_daemon_for_group(tmp_path, shared_dir, gek) + indexer = DirectoryIndexer( + roots=one_root(shared_dir), group_id="a" * 32, + sk_node=Ed25519PrivateKey.generate(), gek=gek) + await indexer.initial_scan() + + session = MagicMock() + session._group_id = "a" * 32 + session._send = MagicMock() + mock_webrtc = MagicMock() + mock_webrtc._sessions = {"p1": session} + daemon._webrtc = mock_webrtc + + await daemon._on_index_change(indexer) + await asyncio.sleep(0.05) + first = session._send.call_args_list[0].args[0] + assert first["type"] == "index_sync" + assert len(first["entries"]) == indexer.index.count + + # Nothing actually changed in the index between the two calls, but + # _on_index_change does not know or care why it was called — the + # SECOND broadcast must still be a delta, now that there is a + # previous snapshot to diff against. + await daemon._on_index_change(indexer) + await asyncio.sleep(0.05) + second = session._send.call_args_list[1].args[0] + assert second["type"] == "index_delta" + assert second["additions"] == [] + assert second["deletions"] == [] + + +@pytest.mark.asyncio +async def test_delta_reflects_additions_and_deletions(tmp_path, shared_dir, gek): + daemon = _new_daemon_for_group(tmp_path, shared_dir, gek) + indexer = DirectoryIndexer( + roots=one_root(shared_dir), group_id="a" * 32, + sk_node=Ed25519PrivateKey.generate(), gek=gek) + await indexer.initial_scan() + removed_id = indexer.index.entries[0].id + + session = MagicMock() + session._group_id = "a" * 32 + session._send = MagicMock() + daemon._webrtc = MagicMock() + daemon._webrtc._sessions = {"p1": session} + + await daemon._on_index_change(indexer) # first: full sync, establishes the snapshot + await asyncio.sleep(0.05) + + # A real change: one entry removed, one added. + indexer.index.remove_entry(removed_id) + from meshbay_common.protocol import IndexEntry + new_entry = IndexEntry(id="new-file-id", name="new.mp4", path="shared", + size=10, type="video", added_at=0) + indexer.index.add_entry(new_entry) + + await daemon._on_index_change(indexer) + await asyncio.sleep(0.05) + + delta_msg = session._send.call_args_list[1].args[0] + assert delta_msg["type"] == "index_delta" + assert delta_msg["deletions"] == [removed_id] + assert [a["id"] for a in delta_msg["additions"]] == ["new-file-id"] + + +@pytest.mark.asyncio +async def test_a_burst_of_changes_produces_one_broadcast(tmp_path, shared_dir, gek): + """Coalescing: several _on_index_change calls in quick succession (one + per debounced watchdog event) must collapse into a single push.""" + daemon = _new_daemon_for_group(tmp_path, shared_dir, gek) + indexer = DirectoryIndexer( + roots=one_root(shared_dir), group_id="a" * 32, + sk_node=Ed25519PrivateKey.generate(), gek=gek) + await indexer.initial_scan() + + session = MagicMock() + session._group_id = "a" * 32 + session._send = MagicMock() + daemon._webrtc = MagicMock() + daemon._webrtc._sessions = {"p1": session} + + for _ in range(5): + await daemon._on_index_change(indexer) + await asyncio.sleep(0.05) + + session._send.assert_called_once() + + +@pytest.mark.asyncio +async def test_swarm_registration_only_sends_new_hashes_after_the_first( + tmp_path, shared_dir, gek): + daemon = _new_daemon_for_group(tmp_path, shared_dir, gek, visibility="public") + indexer = DirectoryIndexer( + roots=one_root(shared_dir), group_id="a" * 32, + sk_node=Ed25519PrivateKey.generate(), gek=gek) + await indexer.initial_scan() + total_files = indexer.index.count + + await daemon._on_index_change(indexer) + await asyncio.sleep(0.05) + assert len(daemon._hub.register_swarm.call_args_list[0].args[0]) == total_files + + from meshbay_common.protocol import IndexEntry + indexer.index.add_entry(IndexEntry(id="new-file-id", name="new.mp4", + path="shared", size=10, type="video", + added_at=0)) + await daemon._on_index_change(indexer) + await asyncio.sleep(0.05) + + assert daemon._hub.register_swarm.call_count == 2 + assert daemon._hub.register_swarm.call_args_list[1].args[0] == ["new-file-id"], \ + "only the newly added hash must be (re-)registered, not the whole library" |