summaryrefslogtreecommitdiffstats
path: root/packages/meshbay-node/tests
diff options
context:
space:
mode:
Diffstat (limited to 'packages/meshbay-node/tests')
-rw-r--r--packages/meshbay-node/tests/test_daemon.py143
-rw-r--r--packages/meshbay-node/tests/test_hot_reload_survives_client_close.py329
-rw-r--r--packages/meshbay-node/tests/test_index_cache.py78
-rw-r--r--packages/meshbay-node/tests/test_index_progress.py165
-rw-r--r--packages/meshbay-node/tests/test_indexer.py290
-rw-r--r--packages/meshbay-node/tests/test_ops.py30
-rw-r--r--packages/meshbay-node/tests/test_scan_settings_policy.py205
7 files changed, 1239 insertions, 1 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"
diff --git a/packages/meshbay-node/tests/test_hot_reload_survives_client_close.py b/packages/meshbay-node/tests/test_hot_reload_survives_client_close.py
new file mode 100644
index 0000000..7cb74cb
--- /dev/null
+++ b/packages/meshbay-node/tests/test_hot_reload_survives_client_close.py
@@ -0,0 +1,329 @@
+"""
+Adding a group (Create Group wizard, or "add a directory" to an existing
+one) fires `_reload_config()` without awaiting it (`asyncio.ensure_future`,
+ui/app.py) — the request handler, and whatever browser tab triggered it,
+return immediately. This is deliberate: the initial scan behind it can take
+a very long time (measured at 23 minutes for a 114 GB library on a slow
+disk), and none of that work belongs to the HTTP request or the WebRTC
+session that happened to start it.
+
+This test proves the scan is genuinely independent of its caller: it starts
+the reload the same way the real endpoint does — schedules it and does not
+await it, standing in for "the browser tab that made the call was closed" —
+then does something else, and only afterwards checks that the reload
+finished and the new group became available on its own.
+"""
+
+import asyncio
+import base64
+import os
+
+import pytest
+from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey
+from unittest.mock import AsyncMock, MagicMock, patch
+
+from meshbay_node import ops
+from meshbay_node.config import Config, HubConfig, NodeConfig, GroupConfig, KeystoreConfig
+from meshbay_node.daemon import NodeDaemon
+import meshbay_node.indexer.indexer as indexer_mod
+
+
+def _free_port() -> int:
+ import socket
+ with socket.socket() as s:
+ s.bind(("127.0.0.1", 0))
+ return s.getsockname()[1]
+
+
+def _toml(data_dir, first_group_dir, second_group_id=None, second_group_dir=None) -> str:
+ # data_dir MUST come before any [section] header — TOML has no notion of
+ # "back to top-level" once a table is open, so a bare `key = value` line
+ # placed after [node] becomes node.data_dir, not the top-level data_dir
+ # load_config() actually reads. Silently falls back to the real default
+ # (~/.local/share/meshbay) instead of erroring, which is exactly how this
+ # test once ran a whole daemon — including _shutdown()'s unlink of
+ # ui-token — against the developer's real, already-running node.
+ text = f"""
+data_dir = "{data_dir}"
+
+[hub]
+url = "http://localhost:9999"
+username = "testuser"
+
+[node]
+quic_port = {_free_port()}
+ui_port = {_free_port()}
+
+[[groups]]
+id = "{"a" * 32}"
+name = "first"
+shared_dir = "{first_group_dir}"
+visibility = "private"
+"""
+ if second_group_id:
+ text += f"""
+[[groups]]
+id = "{second_group_id}"
+name = "slow-new-group"
+shared_dir = "{second_group_dir}"
+visibility = "private"
+"""
+ return text
+
+
+def _mock_keystore_keys(sk_ed):
+ sk_x = X25519PrivateKey.generate()
+ pk_x_raw = sk_x.public_key().public_bytes(
+ serialization.Encoding.Raw, serialization.PublicFormat.Raw)
+ mock_keys = MagicMock()
+ mock_keys.sk_ed25519 = sk_ed
+ mock_keys.pk_ed25519_b64 = "test"
+ mock_keys.sk_x25519 = sk_x
+ mock_keys.pk_x25519_b64 = base64.b64encode(pk_x_raw).decode()
+ return mock_keys
+
+
+@pytest.mark.asyncio
+async def test_hot_loaded_group_finishes_scanning_without_anyone_awaiting_the_reload(
+ tmp_path):
+ first_dir = tmp_path / "first"
+ first_dir.mkdir()
+ (first_dir / "readme.txt").write_bytes(b"hello")
+
+ second_dir = tmp_path / "second"
+ second_dir.mkdir()
+ for i in range(3):
+ (second_dir / f"file{i}.bin").write_bytes(os.urandom(64))
+ second_group_id = "b" * 32
+
+ data_dir = tmp_path / "data"
+ config_path = tmp_path / "node.toml"
+ config_path.write_text(_toml(data_dir, first_dir))
+
+ from meshbay_node.config import load_config
+ daemon = NodeDaemon(load_config(config_path), config_path=config_path)
+
+ sk_node = Ed25519PrivateKey.generate()
+ mock_keys = _mock_keystore_keys(sk_node)
+ mock_session = MagicMock()
+ mock_session.node_id = "node123"
+ mock_session.user_id = "user123"
+ mock_session.hub_pk_pem = sk_node.public_key().public_bytes(
+ serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo)
+
+ # Slow the second group's hashing down (a stand-in for a large/slow
+ # library) so there is a real window in which "nobody is awaiting this"
+ # actually means something, without needing a genuinely huge file.
+ real_scan_file = indexer_mod._scan_file
+
+ def slow_scan_file(root, path):
+ import time
+ time.sleep(0.15)
+ return real_scan_file(root, path)
+
+ with patch("meshbay_node.daemon.load_or_create_keystore", return_value=mock_keys), \
+ patch("meshbay_node.daemon.HubClient") as MockHub, \
+ patch.object(indexer_mod, "_scan_file", slow_scan_file):
+
+ hub_instance = AsyncMock()
+ hub_instance.startup = AsyncMock(return_value=mock_session)
+ hub_instance.send_ws = AsyncMock()
+ hub_instance._ws = None
+ hub_instance.close = AsyncMock()
+ hub_instance.__aenter__ = AsyncMock(return_value=hub_instance)
+ hub_instance.__aexit__ = AsyncMock(return_value=False)
+ MockHub.return_value = hub_instance
+
+ shutdown_event = asyncio.Event()
+
+ async def mock_maintain_ws(**kwargs):
+ await shutdown_event.wait()
+ hub_instance.maintain_ws = mock_maintain_ws
+
+ async def run_daemon():
+ with patch("signal.SIGINT", 2), patch("signal.SIGTERM", 15):
+ try:
+ await asyncio.wait_for(daemon.run(), timeout=15)
+ except (asyncio.TimeoutError, Exception):
+ pass
+
+ run_task = asyncio.create_task(run_daemon())
+ try:
+ for _ in range(50):
+ if daemon._state.get("status") == "running":
+ break
+ await asyncio.sleep(0.05)
+ assert daemon._state["status"] == "running"
+ assert second_group_id not in daemon._state.get("groups_ctx", {})
+
+ # Add the second group to the config on disk, the way the wizard's
+ # attach + /api/reload would leave it, then fire the reload exactly
+ # as ui/app.py does: scheduled, NOT awaited.
+ config_path.write_text(_toml(data_dir, first_dir,
+ second_group_id, second_dir))
+ reload_task = asyncio.ensure_future(daemon._reload_config())
+
+ # Stand in for "the browser tab is gone": do something completely
+ # unrelated to the reload, and explicitly do not await it here.
+ await asyncio.sleep(0.01)
+ assert second_group_id not in daemon._state.get("groups_ctx", {}), \
+ "the scan (3 files x 0.15s) cannot have finished yet"
+
+ # Only now catch up with the background work, from a place that
+ # has no relationship to whatever originally triggered it.
+ await asyncio.wait_for(reload_task, timeout=5)
+
+ assert second_group_id in daemon._state["groups_ctx"], \
+ "the new group must be usable once its scan finishes, " \
+ "regardless of whether anything was still watching the reload"
+ new_indexer = daemon._state["indexers"][second_group_id]
+ assert new_indexer.index.count == 3
+ assert new_indexer.progress.scanning is False
+ finally:
+ shutdown_event.set()
+ await daemon._shutdown()
+ run_task.cancel()
+ try:
+ await run_task
+ except (asyncio.CancelledError, Exception):
+ pass
+
+
+@pytest.mark.asyncio
+async def test_group_scoped_ops_404_until_listed_then_succeed(tmp_path):
+ """
+ The wizard's own sequence, reproduced against the real ops layer: attach
+ a brand-new group, fire the reload the way /api/reload now does
+ (ops.start_reload — scheduled, not awaited), and hit the group-scoped
+ calls that come right after in the UI (add a root, init the GEK) while
+ the scan is still running.
+
+ Found live: "Attaching to node" no longer times out (ops.start_reload
+ returns immediately), but the very next wizard step then failed with
+ "Group not configured on this node" / "Group not hosted on this node" —
+ the group is not in daemon._state["config"].groups or ["groups_ctx"]
+ until _reload_config_inner() finishes, scan included, which is *after*
+ ops.start_reload has already returned. This locks in both halves: the
+ 404 while the scan runs, and success once ops.list_groups() actually
+ lists the group — the exact condition the wizard's own wait
+ (platform.waitForGroupHosted, app.js) polls for.
+ """
+ first_dir = tmp_path / "first"
+ first_dir.mkdir()
+ (first_dir / "readme.txt").write_bytes(b"hello")
+
+ second_dir = tmp_path / "second"
+ second_dir.mkdir()
+ for i in range(3):
+ (second_dir / f"file{i}.bin").write_bytes(os.urandom(64))
+ second_group_id = "c" * 32
+ extra_root_dir = tmp_path / "extra_root"
+ extra_root_dir.mkdir()
+
+ data_dir = tmp_path / "data2"
+ config_path = tmp_path / "node2.toml"
+ config_path.write_text(_toml(data_dir, first_dir))
+
+ from meshbay_node.config import load_config
+ daemon = NodeDaemon(load_config(config_path), config_path=config_path)
+
+ sk_node = Ed25519PrivateKey.generate()
+ mock_keys = _mock_keystore_keys(sk_node)
+ mock_session = MagicMock()
+ mock_session.node_id = "node123"
+ mock_session.user_id = "user123"
+ mock_session.hub_pk_pem = sk_node.public_key().public_bytes(
+ serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo)
+
+ real_scan_file = indexer_mod._scan_file
+
+ def slow_scan_file(root, path):
+ import time
+ time.sleep(0.15)
+ return real_scan_file(root, path)
+
+ with patch("meshbay_node.daemon.load_or_create_keystore", return_value=mock_keys), \
+ patch("meshbay_node.daemon.HubClient") as MockHub, \
+ patch.object(indexer_mod, "_scan_file", slow_scan_file):
+
+ hub_instance = AsyncMock()
+ hub_instance.startup = AsyncMock(return_value=mock_session)
+ hub_instance.send_ws = AsyncMock()
+ hub_instance._ws = None
+ hub_instance.close = AsyncMock()
+ hub_instance.__aenter__ = AsyncMock(return_value=hub_instance)
+ hub_instance.__aexit__ = AsyncMock(return_value=False)
+ MockHub.return_value = hub_instance
+
+ shutdown_event = asyncio.Event()
+
+ async def mock_maintain_ws(**kwargs):
+ await shutdown_event.wait()
+ hub_instance.maintain_ws = mock_maintain_ws
+
+ async def run_daemon():
+ with patch("signal.SIGINT", 2), patch("signal.SIGTERM", 15):
+ try:
+ await asyncio.wait_for(daemon.run(), timeout=15)
+ except (asyncio.TimeoutError, Exception):
+ pass
+
+ run_task = asyncio.create_task(run_daemon())
+ try:
+ for _ in range(50):
+ if daemon._state.get("status") == "running":
+ break
+ await asyncio.sleep(0.05)
+ assert daemon._state["status"] == "running"
+
+ # The same file write ops.attach_group does (a raw text append),
+ # then the same fire-and-forget reload /api/reload now does.
+ config_path.write_text(_toml(data_dir, first_dir,
+ second_group_id, second_dir))
+ reload_task = asyncio.ensure_future(ops.start_reload(daemon._state))
+
+ await asyncio.sleep(0.01)
+ listing = await ops.list_groups(daemon._state)
+ assert second_group_id not in [g["id"] for g in listing["groups"]], \
+ "the scan (3 files x 0.15s) cannot have finished this fast"
+
+ # Exactly the wizard's next two steps, hit mid-scan.
+ with pytest.raises(ops.OpError) as add_root_exc:
+ await ops.add_root(daemon._state, second_group_id,
+ str(extra_root_dir))
+ assert add_root_exc.value.status == 404
+
+ with pytest.raises(ops.OpError) as gek_exc:
+ await ops.set_gek(daemon._state, second_group_id)
+ assert gek_exc.value.status == 404
+
+ # Now wait the way platform.waitForGroupHosted (app.js) does:
+ # poll list_groups(), not index-status, until the group is
+ # actually there.
+ for _ in range(100):
+ listing = await ops.list_groups(daemon._state)
+ if second_group_id in [g["id"] for g in listing["groups"]]:
+ break
+ await asyncio.sleep(0.05)
+ else:
+ pytest.fail("group never appeared in list_groups()")
+
+ await asyncio.wait_for(reload_task, timeout=5)
+
+ # Both calls that 404'd above must now succeed.
+ add_result = await ops.add_root(daemon._state, second_group_id,
+ str(extra_root_dir))
+ assert add_result["status"] == "added"
+
+ gek_result = await ops.set_gek(daemon._state, second_group_id)
+ assert gek_result["status"] == "ok"
+ finally:
+ shutdown_event.set()
+ await daemon._shutdown()
+ run_task.cancel()
+ try:
+ await run_task
+ except (asyncio.CancelledError, Exception):
+ pass
diff --git a/packages/meshbay-node/tests/test_index_cache.py b/packages/meshbay-node/tests/test_index_cache.py
new file mode 100644
index 0000000..db0e37e
--- /dev/null
+++ b/packages/meshbay-node/tests/test_index_cache.py
@@ -0,0 +1,78 @@
+"""Tests for the (path, size, mtime) -> hash cache (indexer/cache.py)."""
+
+import pytest
+
+from meshbay_node.indexer.cache import IndexCache
+
+
+@pytest.fixture
+async def cache(tmp_path):
+ c = IndexCache(db_path=tmp_path / "index_cache.db")
+ await c.open()
+ yield c
+ await c.close()
+
+
+@pytest.mark.asyncio
+async def test_put_then_lookup_hits(cache):
+ await cache.put("/lib/a.mkv", size=1000, mtime=111.0, hash="abc123",
+ type="video", added_at=42)
+
+ hit = await cache.lookup("/lib/a.mkv", size=1000, mtime=111.0)
+
+ assert hit is not None
+ assert hit.hash == "abc123"
+ assert hit.type == "video"
+ assert hit.added_at == 42
+
+
+@pytest.mark.asyncio
+async def test_lookup_misses_on_unknown_path(cache):
+ assert await cache.lookup("/lib/never-seen.mkv", size=1, mtime=1.0) is None
+
+
+@pytest.mark.asyncio
+async def test_lookup_misses_on_different_mtime(cache):
+ await cache.put("/lib/a.mkv", size=1000, mtime=111.0, hash="abc123",
+ type="video", added_at=42)
+
+ assert await cache.lookup("/lib/a.mkv", size=1000, mtime=222.0) is None
+
+
+@pytest.mark.asyncio
+async def test_lookup_misses_on_different_size(cache):
+ await cache.put("/lib/a.mkv", size=1000, mtime=111.0, hash="abc123",
+ type="video", added_at=42)
+
+ assert await cache.lookup("/lib/a.mkv", size=2000, mtime=111.0) is None
+
+
+@pytest.mark.asyncio
+async def test_put_overwrites_previous_row_for_same_path(cache):
+ await cache.put("/lib/a.mkv", size=1000, mtime=111.0, hash="old",
+ type="video", added_at=1)
+ await cache.put("/lib/a.mkv", size=2000, mtime=222.0, hash="new",
+ type="video", added_at=2)
+
+ assert await cache.lookup("/lib/a.mkv", size=1000, mtime=111.0) is None
+ hit = await cache.lookup("/lib/a.mkv", size=2000, mtime=222.0)
+ assert hit.hash == "new"
+
+
+@pytest.mark.asyncio
+async def test_cache_survives_reopen(tmp_path):
+ db_path = tmp_path / "index_cache.db"
+
+ c1 = IndexCache(db_path=db_path)
+ await c1.open()
+ await c1.put("/lib/a.mkv", size=1000, mtime=111.0, hash="abc123",
+ type="video", added_at=42)
+ await c1.close()
+
+ c2 = IndexCache(db_path=db_path)
+ await c2.open()
+ hit = await c2.lookup("/lib/a.mkv", size=1000, mtime=111.0)
+ await c2.close()
+
+ assert hit is not None
+ assert hit.hash == "abc123"
diff --git a/packages/meshbay-node/tests/test_index_progress.py b/packages/meshbay-node/tests/test_index_progress.py
new file mode 100644
index 0000000..52cd78b
--- /dev/null
+++ b/packages/meshbay-node/tests/test_index_progress.py
@@ -0,0 +1,165 @@
+"""
+Indexing status visible node -> client: handshake ack field, the loopback
+status route for the Create Group wizard / "add a directory", and the
+periodic INDEX_PROGRESS push to already-connected peers. Never the index
+itself (see test_daemon.py for that) and never anything sent to the hub.
+"""
+
+import asyncio
+
+import pytest
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+from unittest.mock import MagicMock
+from fastapi.testclient import TestClient
+
+from meshbay_node.config import Config, HubConfig, NodeConfig, GroupConfig, KeystoreConfig
+from meshbay_node.daemon import NodeDaemon
+from meshbay_node.indexer.indexer import IndexProgress
+from meshbay_node.indexer.group_index import GroupIndex
+from meshbay_node.transport.webrtc_server import WebRTCPeerSession
+from meshbay_node.ui.app import create_ui_app
+
+
+def _session_with_progress(progress: IndexProgress | None, group_id: str = "g" * 32):
+ index = GroupIndex(group_id=group_id, sk_node=Ed25519PrivateKey.generate())
+ session = WebRTCPeerSession.__new__(WebRTCPeerSession)
+ group_ctx = {"index": index}
+ if progress is not None:
+ group_ctx["progress"] = progress
+ session._ctx = {"groups": {group_id: group_ctx}}
+ session._group_id = group_id
+ return session
+
+
+# ── _indexing_status() ───────────────────────────────────────────────────────
+
+def test_indexing_status_defaults_idle_when_no_progress_tracked():
+ session = _session_with_progress(None)
+ assert session._indexing_status() == {
+ "scanning": False, "scanned_bytes": 0, "total_bytes": 0}
+
+
+def test_indexing_status_reflects_live_progress():
+ progress = IndexProgress(scanning=True, scanned_bytes=500, total_bytes=2000,
+ current_dir="StarWars")
+ session = _session_with_progress(progress)
+
+ status = session._indexing_status()
+
+ assert status == {"scanning": True, "scanned_bytes": 500, "total_bytes": 2000}
+ assert "current_dir" not in status, \
+ "the directory name is operator-local detail, never sent to a member"
+
+
+# ── /api/groups/{id}/index-status (loopback) ────────────────────────────────
+
+def _ui_client(state: dict) -> TestClient:
+ return TestClient(create_ui_app({"status": "running", "groups_ctx": {},
+ "indexes": {}, **state}))
+
+
+def test_index_status_route_idle_for_unknown_group():
+ client = _ui_client({"indexers": {}})
+ resp = client.get("/api/groups/unknown-group/index-status")
+ assert resp.status_code == 200
+ assert resp.json() == {"scanning": False, "scanned_bytes": 0,
+ "total_bytes": 0, "current_dir": ""}
+
+
+def test_index_status_route_reflects_indexer_progress():
+ fake_indexer = MagicMock()
+ fake_indexer.progress = IndexProgress(
+ scanning=True, scanned_bytes=1_000_000, total_bytes=4_000_000_000,
+ current_dir="2024")
+ client = _ui_client({"indexers": {"g" * 32: fake_indexer}})
+
+ resp = client.get(f"/api/groups/{'g' * 32}/index-status")
+
+ assert resp.json() == {
+ "scanning": True, "scanned_bytes": 1_000_000,
+ "total_bytes": 4_000_000_000, "current_dir": "2024",
+ }
+
+
+# ── _push_index_progress / _progress_pusher ─────────────────────────────────
+
+def _daemon(tmp_path) -> NodeDaemon:
+ config = Config(
+ hub=HubConfig(url="http://localhost:9999", username="testuser"),
+ node=NodeConfig(),
+ groups=[],
+ keystore=KeystoreConfig(path=tmp_path / "keystore.enc"),
+ data_dir=tmp_path / "data",
+ )
+ return NodeDaemon(config)
+
+
+@pytest.mark.asyncio
+async def test_push_index_progress_only_reaches_same_group_peers(tmp_path):
+ daemon = _daemon(tmp_path)
+
+ same_group = MagicMock()
+ same_group._group_id = "a" * 32
+ same_group._send = MagicMock()
+ other_group = MagicMock()
+ other_group._group_id = "b" * 32
+ other_group._send = MagicMock()
+
+ mock_webrtc = MagicMock()
+ mock_webrtc._sessions = {"p1": same_group, "p2": other_group}
+ daemon._webrtc = mock_webrtc
+
+ progress = IndexProgress(scanning=True, scanned_bytes=10, total_bytes=100)
+ daemon._push_index_progress("a" * 32, progress)
+
+ same_group._send.assert_called_once()
+ msg = same_group._send.call_args[0][0]
+ assert msg["type"] == "index_progress"
+ assert msg["group_id"] == "a" * 32
+ assert msg["scanning"] is True
+ assert msg["scanned_bytes"] == 10
+ assert msg["total_bytes"] == 100
+ other_group._send.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_progress_pusher_pushes_while_scanning_then_one_final_push(tmp_path):
+ daemon = _daemon(tmp_path)
+
+ session = MagicMock()
+ session._group_id = "a" * 32
+ session._send = MagicMock()
+ mock_webrtc = MagicMock()
+ mock_webrtc._sessions = {"p1": session}
+ daemon._webrtc = mock_webrtc
+
+ indexer = MagicMock()
+ indexer.group_id = "a" * 32
+ indexer.progress = IndexProgress(scanning=True, scanned_bytes=0, total_bytes=100)
+
+ task = asyncio.create_task(daemon._progress_pusher(indexer, interval=0.05))
+ try:
+ # Two ticks while still scanning.
+ await asyncio.sleep(0.12)
+ assert session._send.call_count >= 2
+ assert all(c.args[0]["scanning"] is True for c in session._send.call_args_list)
+
+ # Scan finishes between ticks.
+ indexer.progress.scanning = False
+ calls_before = session._send.call_count
+ await asyncio.sleep(0.07)
+ assert session._send.call_count == calls_before + 1, \
+ "exactly one final push must follow the False transition"
+ assert session._send.call_args.args[0]["scanning"] is False
+
+ # Nothing further once idle.
+ calls_after_final = session._send.call_count
+ await asyncio.sleep(0.15)
+ assert session._send.call_count == calls_after_final, \
+ "no more pushes once idle and already reported"
+ finally:
+ task.cancel()
+ try:
+ await task
+ except asyncio.CancelledError:
+ pass
diff --git a/packages/meshbay-node/tests/test_indexer.py b/packages/meshbay-node/tests/test_indexer.py
index c304361..68ccc4c 100644
--- a/packages/meshbay-node/tests/test_indexer.py
+++ b/packages/meshbay-node/tests/test_indexer.py
@@ -9,7 +9,8 @@ from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey
from meshbay_common.crypto import generate_gek
-from meshbay_node.indexer import DirectoryIndexer, GroupIndex
+from meshbay_node.indexer import DirectoryIndexer, GroupIndex, IndexCache
+import meshbay_node.indexer.indexer as indexer_mod
from conftest import one_root
from meshbay_node.keystore import NodeKeys
@@ -190,3 +191,290 @@ async def test_index_roundtrip_after_scan(shared_dir, sk_node, gek):
wire = indexer.index.serialize()
recovered = GroupIndex.deserialize(wire, sk_node=sk_node, gek=gek)
assert recovered.count == indexer.index.count
+
+
+# ── Cache-aware scanning ───────────────────────────────────────────────────────
+
+@pytest.fixture
+async def index_cache(tmp_path):
+ c = IndexCache(db_path=tmp_path / "index_cache.db")
+ await c.open()
+ yield c
+ await c.close()
+
+
+@pytest.mark.asyncio
+async def test_second_scan_with_same_cache_hashes_nothing(
+ shared_dir, sk_node, gek, index_cache):
+ """
+ The whole point of the cache: a "restart" (a fresh DirectoryIndexer, same
+ on-disk cache) that finds every file's (size, mtime) unchanged must not
+ read a single byte of file content.
+ """
+ first = DirectoryIndexer(roots=one_root(shared_dir), group_id="g",
+ sk_node=sk_node, gek=gek, cache=index_cache)
+ await first.initial_scan()
+ assert first.index.count == 4
+
+ calls = []
+ real_scan_file = indexer_mod._scan_file
+
+ def spy(root, path):
+ calls.append(path)
+ return real_scan_file(root, path)
+
+ indexer_mod._scan_file = spy
+ try:
+ second = DirectoryIndexer(roots=one_root(shared_dir), group_id="g",
+ sk_node=sk_node, gek=gek, cache=index_cache)
+ await second.initial_scan()
+ finally:
+ indexer_mod._scan_file = real_scan_file
+
+ assert calls == [], f"expected zero hash calls on a fully-cached rescan, got {calls}"
+ assert second.index.count == first.index.count
+ assert {e.id for e in second.index.entries} == {e.id for e in first.index.entries}
+
+
+@pytest.mark.asyncio
+async def test_modified_file_is_rehashed(tmp_path, sk_node, gek, index_cache):
+ d = tmp_path / "shared"
+ d.mkdir()
+ f = d / "movie.mkv"
+ f.write_bytes(b"original content")
+
+ first = DirectoryIndexer(roots=one_root(d), group_id="g",
+ sk_node=sk_node, gek=gek, cache=index_cache)
+ await first.initial_scan()
+ old_id = first.index.entries[0].id
+
+ # Change both content and mtime, as any real edit would.
+ f.write_bytes(b"a completely different, longer payload")
+ os.utime(f, (time.time() + 5, time.time() + 5))
+
+ second = DirectoryIndexer(roots=one_root(d), group_id="g",
+ sk_node=sk_node, gek=gek, cache=index_cache)
+ await second.initial_scan()
+
+ assert second.index.count == 1
+ assert second.index.entries[0].id != old_id
+
+
+@pytest.mark.asyncio
+async def test_scan_interrupted_partway_leaves_only_completed_files_cached(
+ tmp_path, sk_node, gek, index_cache):
+ """
+ A cache row is only ever written after a file is fully hashed (cache.py),
+ so a crash mid-scan cannot leave a stale/partial row — the next scan just
+ treats the not-yet-cached files as new, and finishes the job.
+ """
+ d = tmp_path / "shared"
+ d.mkdir()
+ names = [f"file{i}.bin" for i in range(5)]
+ for i, name in enumerate(names):
+ (d / name).write_bytes(os.urandom(64) * (i + 1))
+
+ real_scan_file = indexer_mod._scan_file
+ hashed_before_crash = []
+
+ def crash_after_three(root, path):
+ if len(hashed_before_crash) >= 3:
+ raise RuntimeError("simulated crash mid-scan")
+ entry = real_scan_file(root, path)
+ hashed_before_crash.append(path)
+ return entry
+
+ indexer_mod._scan_file = crash_after_three
+ try:
+ crashing = DirectoryIndexer(roots=one_root(d), group_id="g",
+ sk_node=sk_node, gek=gek, cache=index_cache)
+ with pytest.raises(RuntimeError):
+ await crashing.initial_scan()
+ finally:
+ indexer_mod._scan_file = real_scan_file
+
+ assert len(hashed_before_crash) == 3
+
+ # A normal rescan (same cache) must still end up with all 5 files
+ # correctly indexed, hashing only the ones the crash never got to.
+ calls = []
+
+ def spy(root, path):
+ calls.append(path)
+ return real_scan_file(root, path)
+
+ indexer_mod._scan_file = spy
+ try:
+ resumed = DirectoryIndexer(roots=one_root(d), group_id="g",
+ sk_node=sk_node, gek=gek, cache=index_cache)
+ await resumed.initial_scan()
+ finally:
+ indexer_mod._scan_file = real_scan_file
+
+ assert resumed.index.count == 5
+ assert len(calls) == 2, f"expected only the 2 not-yet-cached files to be hashed, got {len(calls)}"
+
+
+# ── Progress state ───────────────────────────────────────────────────────────
+
+@pytest.mark.asyncio
+async def test_progress_reflects_bytes_scanned(shared_dir, sk_node, gek):
+ indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g",
+ sk_node=sk_node, gek=gek)
+ assert indexer.progress.scanning is False
+
+ await indexer.initial_scan()
+
+ total_size = sum(f.stat().st_size for f in shared_dir.rglob("*") if f.is_file())
+ assert indexer.progress.scanning is False, "must end idle, not stuck scanning"
+ assert indexer.progress.scanned_bytes == total_size
+ assert indexer.progress.total_bytes == total_size
+
+
+@pytest.mark.asyncio
+async def test_progress_stops_even_when_hashing_raises(tmp_path, sk_node, gek):
+ d = tmp_path / "shared"
+ d.mkdir()
+ (d / "a.bin").write_bytes(os.urandom(64))
+ (d / "b.bin").write_bytes(os.urandom(64))
+
+ real_scan_file = indexer_mod._scan_file
+
+ def boom(root, path):
+ raise RuntimeError("simulated failure mid-scan")
+
+ indexer = DirectoryIndexer(roots=one_root(d), group_id="g",
+ sk_node=sk_node, gek=gek)
+ indexer_mod._scan_file = boom
+ try:
+ with pytest.raises(RuntimeError):
+ await indexer.initial_scan()
+ finally:
+ indexer_mod._scan_file = real_scan_file
+
+ assert indexer.progress.scanning is False, \
+ "an exception mid-scan must not leave the scanning flag stuck on"
+
+
+# ── Off-loop directory walks, reconcile backoff ─────────────────────────────
+
+@pytest.mark.asyncio
+async def test_walk_root_does_not_stall_the_event_loop(tmp_path, sk_node, gek):
+ d = tmp_path / "shared"
+ d.mkdir()
+ (d / "f.bin").write_bytes(b"x")
+
+ real_walk = indexer_mod._walk_root
+
+ def slow_walk(root):
+ time.sleep(0.2)
+ return real_walk(root)
+
+ indexer_mod._walk_root = slow_walk
+ ticks = 0
+
+ async def ticker():
+ nonlocal ticks
+ while True:
+ await asyncio.sleep(0.01)
+ ticks += 1
+
+ ticker_task = asyncio.create_task(ticker())
+ try:
+ indexer = DirectoryIndexer(roots=one_root(d), group_id="g",
+ sk_node=sk_node, gek=gek)
+ await indexer.initial_scan()
+ finally:
+ indexer_mod._walk_root = real_walk
+ ticker_task.cancel()
+ try:
+ await ticker_task
+ except asyncio.CancelledError:
+ pass
+
+ assert ticks >= 5, (
+ "the event loop must keep running other tasks while a directory "
+ f"walk is in progress in the executor — only {ticks} ticks happened "
+ "during a 0.2s walk")
+
+
+@pytest.mark.asyncio
+async def test_reconcile_backoff_grows_with_no_changes_then_caps(shared_dir, sk_node, gek):
+ indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g",
+ sk_node=sk_node, gek=gek, reconcile_secs=0.01)
+ await indexer.initial_scan()
+ assert indexer._reconcile_delay == 0.01
+
+ task = asyncio.create_task(indexer._reconcile_loop())
+ try:
+ await asyncio.sleep(0.2)
+ assert indexer._reconcile_delay > 0.01, \
+ "several no-change ticks must have grown the delay"
+ finally:
+ task.cancel()
+ try:
+ await task
+ except asyncio.CancelledError:
+ pass
+
+ # A low, instance-only cap so the clamp is observable without waiting
+ # through dozens of real doublings up to the real 7200s ceiling.
+ indexer.RECONCILE_BACKOFF_CAP = 0.05
+ indexer._reconcile_delay = 0.04
+ task = asyncio.create_task(indexer._reconcile_loop())
+ try:
+ await asyncio.sleep(0.15)
+ assert indexer._reconcile_delay <= 0.05, "delay must never exceed the cap"
+ finally:
+ task.cancel()
+ try:
+ await task
+ except asyncio.CancelledError:
+ pass
+
+
+@pytest.mark.asyncio
+async def test_note_activity_resets_backoff(shared_dir, sk_node, gek):
+ indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g",
+ sk_node=sk_node, gek=gek, reconcile_secs=10.0)
+ await indexer.initial_scan()
+ indexer._reconcile_delay = 5000.0 # simulate a long-idle backoff
+
+ indexer.note_activity()
+
+ assert indexer._reconcile_delay == 10.0
+
+
+@pytest.mark.asyncio
+async def test_reconcile_backoff_resets_when_something_actually_changes(
+ shared_dir, sk_node, gek):
+ indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g",
+ sk_node=sk_node, gek=gek, reconcile_secs=0.02)
+ await indexer.initial_scan()
+ indexer._reconcile_delay = 5.0 # pretend it had already backed off a lot
+
+ # A fake reconcile() rather than a real filesystem change: the real
+ # sweep's timing (disk I/O, the executor round trip) would race against
+ # this test's own sleeps. What matters here is only _reconcile_loop's
+ # reaction to "something changed", not reconcile()'s own detection logic
+ # — that is covered separately (test_root_availability.py).
+ reconciled_once = asyncio.Event()
+
+ async def fake_reconcile():
+ reconciled_once.set()
+ return True
+
+ indexer._reconcile_delay = 0.01
+ indexer.reconcile = fake_reconcile
+
+ task = asyncio.create_task(indexer._reconcile_loop())
+ try:
+ await asyncio.wait_for(reconciled_once.wait(), timeout=2.0)
+ assert indexer._reconcile_delay == 0.02, \
+ "a real change must reset the delay back to the base interval"
+ finally:
+ task.cancel()
+ try:
+ await task
+ except asyncio.CancelledError:
+ pass
diff --git a/packages/meshbay-node/tests/test_ops.py b/packages/meshbay-node/tests/test_ops.py
index d2ccc0d..83758ae 100644
--- a/packages/meshbay-node/tests/test_ops.py
+++ b/packages/meshbay-node/tests/test_ops.py
@@ -9,6 +9,7 @@ appeared: the adapters must be thin, and the operations must not decide who may
call them.
"""
+import asyncio
import inspect
from pathlib import Path
@@ -221,3 +222,32 @@ async def test_reload_config_without_fn_is_refused(tmp_path):
state = _state(tmp_path)
with pytest.raises(ops.OpError, match="Reload not available"):
await ops.reload_config(state)
+
+
+async def test_start_reload_returns_before_reload_fn_finishes(tmp_path):
+ """The loopback route uses this one: a brand-new group's initial scan
+ can take minutes, and the Electron bridge caps every loopback call at
+ 30s (main.js node:call) — start_reload must not block on it."""
+ state = _state(tmp_path)
+ release = asyncio.Event()
+ called = []
+
+ async def slow_reload():
+ await release.wait()
+ called.append(True)
+ state["reload_fn"] = slow_reload
+
+ out = await asyncio.wait_for(ops.start_reload(state), timeout=1.0)
+
+ assert out["status"] == "reloading"
+ assert not called, "start_reload must return before reload_fn finishes"
+
+ release.set()
+ await asyncio.sleep(0) # let the still-running reload_fn task complete
+ assert called, "reload_fn must still actually run, just not be waited on"
+
+
+async def test_start_reload_without_fn_is_refused(tmp_path):
+ state = _state(tmp_path)
+ with pytest.raises(ops.OpError, match="Reload not available"):
+ await ops.start_reload(state)
diff --git a/packages/meshbay-node/tests/test_scan_settings_policy.py b/packages/meshbay-node/tests/test_scan_settings_policy.py
new file mode 100644
index 0000000..719b988
--- /dev/null
+++ b/packages/meshbay-node/tests/test_scan_settings_policy.py
@@ -0,0 +1,205 @@
+"""
+The operator can tune how often the indexer's reconciliation backstop runs,
+and how long it waits after a file's last write before hashing it.
+
+Same shape as test_apps_enabled_policy.py / test_member_upload_policy.py:
+changed by a signed operator instruction, stored on the node rather than the
+hub. Unlike those two, there is also a *live* DirectoryIndexer object to
+update — see test_set_scan_settings_updates_the_live_indexer below.
+"""
+
+import os
+from pathlib import Path
+
+import pytest
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+
+from meshbay_common.adminop import OP_SET_SCAN_SETTINGS
+from meshbay_common.crypto import generate_gek
+from meshbay_node import ops
+from meshbay_node.indexer.group_index import GroupIndex
+from meshbay_node.indexer.indexer import DirectoryIndexer
+from meshbay_node.roster import Roster
+from meshbay_node.transport.webrtc_server import WebRTCPeerSession
+
+from conftest import one_root
+
+pytestmark = pytest.mark.asyncio
+
+
+@pytest.fixture
+def gek():
+ return generate_gek()
+
+
+@pytest.fixture
+def shared_dir(tmp_path):
+ d = tmp_path / "shared"
+ d.mkdir()
+ (d / "video.mkv").write_bytes(os.urandom(256))
+ return d
+
+
+def _session(tmp_path: Path, user_id: str, *, operator: str | None = None) -> WebRTCPeerSession:
+ shared_root = tmp_path / "shared"
+ shared_root.mkdir(exist_ok=True)
+ index = GroupIndex(group_id="g" * 32, sk_node=Ed25519PrivateKey.generate())
+ ctx = {
+ "roots": one_root(shared_root),
+ "index": index,
+ "sk_node": index.sk_node,
+ "node_user_id": operator,
+ }
+ session = WebRTCPeerSession.__new__(WebRTCPeerSession)
+ session._ctx = ctx
+ session._group_id = None
+ session._user_id = user_id
+ session._pk_user = ""
+ session.sent = []
+ session._send = session.sent.append
+ session._audit = lambda *a, **k: None
+ return session
+
+
+# ── Refused before a challenge is even issued ───────────────────────────────
+
+async def test_out_of_range_reconcile_interval_is_refused(tmp_path):
+ session = _session(tmp_path, "op", operator="op")
+ session._has_admin_authority = lambda: True
+ issued = []
+ session._issue_admin_challenge = lambda op, subject: issued.append((op, subject))
+
+ session._do_set_scan_settings(
+ {"reconcile_interval_secs": 1.0, "debounce_secs": 2.0})
+
+ assert not issued
+ assert [m for m in session.sent if m.get("type") == "error"]
+
+
+async def test_out_of_range_debounce_is_refused(tmp_path):
+ session = _session(tmp_path, "op", operator="op")
+ session._has_admin_authority = lambda: True
+ issued = []
+ session._issue_admin_challenge = lambda op, subject: issued.append((op, subject))
+
+ session._do_set_scan_settings(
+ {"reconcile_interval_secs": 600.0, "debounce_secs": 99999.0})
+
+ assert not issued
+ assert [m for m in session.sent if m.get("type") == "error"]
+
+
+async def test_non_numeric_values_are_refused(tmp_path):
+ session = _session(tmp_path, "op", operator="op")
+ session._has_admin_authority = lambda: True
+ issued = []
+ session._issue_admin_challenge = lambda op, subject: issued.append((op, subject))
+
+ session._do_set_scan_settings(
+ {"reconcile_interval_secs": "not-a-number", "debounce_secs": 2.0})
+
+ assert not issued
+ assert [m for m in session.sent if m.get("type") == "error"]
+
+
+async def test_a_request_with_nobody_to_authorize_it_is_refused(tmp_path):
+ session = _session(tmp_path, "member-1", operator="the-operator")
+ session._has_admin_authority = lambda: False
+
+ session._do_set_scan_settings(
+ {"reconcile_interval_secs": 600.0, "debounce_secs": 2.0})
+
+ assert [m for m in session.sent if m.get("type") == "error"]
+
+
+# ── Who may change it ───────────────────────────────────────────────────────
+
+async def test_changing_it_needs_a_signature(tmp_path):
+ """The request only ever produces a challenge — nothing is applied
+ until a signature over the transcript verifies."""
+ session = _session(tmp_path, "op", operator="op")
+ session._has_admin_authority = lambda: True
+ issued = []
+ session._issue_admin_challenge = lambda op, subject: issued.append((op, subject))
+
+ session._do_set_scan_settings(
+ {"reconcile_interval_secs": 600.0, "debounce_secs": 2.0})
+
+ assert issued == [(OP_SET_SCAN_SETTINGS, "600,2")]
+
+
+# ── Where it is stored ──────────────────────────────────────────────────────
+
+async def test_the_setting_lives_on_the_node_and_survives_a_restart(tmp_path):
+ roster = Roster(db_path=tmp_path / "roster.db")
+ await roster.open()
+ try:
+ defaults = await roster.scan_settings("g1")
+ assert defaults == {
+ "reconcile_interval_secs": DirectoryIndexer.DEFAULT_RECONCILE_SECS,
+ "debounce_secs": DirectoryIndexer.DEFAULT_DEBOUNCE_SECS,
+ }, "unset must mean the indexer's own defaults, or an upgrade " \
+ "changes behaviour for every existing group"
+
+ await roster.set_scan_settings("g1", 1200.0, 5.0, set_by="op")
+ assert await roster.scan_settings("g1") == {
+ "reconcile_interval_secs": 1200.0, "debounce_secs": 5.0}
+ finally:
+ await roster.close()
+
+ reopened = Roster(db_path=tmp_path / "roster.db")
+ await reopened.open()
+ try:
+ assert await reopened.scan_settings("g1") == {
+ "reconcile_interval_secs": 1200.0, "debounce_secs": 5.0}
+ assert await reopened.scan_settings("g2") == {
+ "reconcile_interval_secs": DirectoryIndexer.DEFAULT_RECONCILE_SECS,
+ "debounce_secs": DirectoryIndexer.DEFAULT_DEBOUNCE_SECS,
+ }, "one group's setting must not answer for another"
+ finally:
+ await reopened.close()
+
+
+# ── Applying it to the live indexer ─────────────────────────────────────────
+
+async def test_set_scan_settings_updates_the_live_indexer(tmp_path, shared_dir, gek):
+ roster = Roster(db_path=tmp_path / "roster.db")
+ await roster.open()
+ indexer = DirectoryIndexer(
+ roots=one_root(shared_dir), group_id="g1",
+ sk_node=Ed25519PrivateKey.generate(), gek=gek)
+ await indexer.initial_scan()
+ indexer._reconcile_delay = 5000.0 # simulate a long-idle backoff
+ state = {"roster": roster, "indexers": {"g1": indexer}}
+
+ try:
+ result = await ops.set_scan_settings(state, "g1", 1800.0, 3.0)
+
+ assert result == {"reconcile_interval_secs": 1800.0, "debounce_secs": 3.0,
+ "group_id": "g1"}
+ assert indexer.reconcile_secs == 1800.0
+ assert indexer.debounce_secs == 3.0
+ assert indexer._reconcile_delay == 1800.0, \
+ "the new interval must apply right away, not after whatever " \
+ "backoff had already stretched the wait to"
+ assert await roster.scan_settings("g1") == {
+ "reconcile_interval_secs": 1800.0, "debounce_secs": 3.0}
+ finally:
+ await roster.close()
+
+
+async def test_set_scan_settings_without_a_live_indexer_still_persists(tmp_path):
+ """A group hosted on the node but with no running indexer in this
+ process (e.g. a test, or a group not yet hot-loaded) must not crash —
+ the setting still lands in roster.db for whenever it is."""
+ roster = Roster(db_path=tmp_path / "roster.db")
+ await roster.open()
+ state = {"roster": roster, "indexers": {}}
+
+ try:
+ result = await ops.set_scan_settings(state, "g1", 1800.0, 3.0)
+ assert result["reconcile_interval_secs"] == 1800.0
+ assert await roster.scan_settings("g1") == {
+ "reconcile_interval_secs": 1800.0, "debounce_secs": 3.0}
+ finally:
+ await roster.close()
generated by cgit v1.2.3 (git 2.51.0) at 2026-09-25 21:09:23 +0000