1 files changed, 139 insertions, 0 deletions
diff --git a/packages/meshbay-node/tests/test_ops_index_cache.py b/packages/meshbay-node/tests/test_ops_index_cache.py
new file mode 100644
index 0000000..0b28d69
--- /dev/null
+++ b/packages/meshbay-node/tests/test_ops_index_cache.py
@@ -0,0 +1,139 @@
+"""Tests for ops.index_cache_stats / ops.prune_index_cache."""
+
+import types
+
+import pytest
+from meshbay_node import ops
+from meshbay_node.indexer.cache import IndexCache
+from meshbay_node.roots import Root
+
+
+@pytest.fixture
+async def cache(tmp_path):
+    c = IndexCache(db_path=tmp_path / "index_cache.db")
+    await c.open()
+    yield c
+    await c.close()
+
+
+def _indexer(roots):
+    """A stand-in for DirectoryIndexer — prune_index_cache only ever reads
+    `.roots`, so a real one (with its watchdog Observer, executor, etc.)
+    would be pure overhead here."""
+    return types.SimpleNamespace(roots=roots)
+
+
+@pytest.mark.asyncio
+async def test_stats_with_no_cache_reports_zero():
+    assert await ops.index_cache_stats({}) == {"count": 0}
+
+
+@pytest.mark.asyncio
+async def test_stats_reports_row_count(cache):
+    await cache.put("/lib/a.mkv", size=1, mtime=1.0, hash="a", type="video", added_at=1)
+
+    assert await ops.index_cache_stats({"index_cache": cache}) == {"count": 1}
+
+
+@pytest.mark.asyncio
+async def test_prune_without_a_cache_raises():
+    with pytest.raises(ops.OpError):
+        await ops.prune_index_cache({})
+
+
+@pytest.mark.asyncio
+async def test_prune_removes_a_path_no_group_claims_any_more(cache):
+    """A folder no group's roots reference at all any more — every group
+    that once shared it was deleted, or the root was removed everywhere."""
+    await cache.put("/gone/orphan.mkv", size=1, mtime=1.0, hash="a",
+                     type="video", added_at=1)
+    state = {"index_cache": cache, "indexers": {}}
+
+    result = await ops.prune_index_cache(state)
+
+    assert result == {"status": "pruned", "removed": 1, "kept": 0}
+    assert await cache.count() == 0
+
+
+@pytest.mark.asyncio
+async def test_prune_keeps_a_path_whose_file_still_exists_under_an_available_root(
+        tmp_path, cache):
+    root_dir = tmp_path / "lib"
+    root_dir.mkdir()
+    f = root_dir / "still-here.mkv"
+    f.write_bytes(b"x")
+    await cache.put(str(f), size=1, mtime=1.0, hash="a", type="video", added_at=1)
+
+    root = Root(name="lib", path=root_dir, available=True)
+    state = {"index_cache": cache, "indexers": {"g": _indexer([root])}}
+
+    result = await ops.prune_index_cache(state)
+
+    assert result == {"status": "pruned", "removed": 0, "kept": 1}
+    assert await cache.count() == 1
+
+
+@pytest.mark.asyncio
+async def test_prune_removes_a_path_deleted_from_an_available_root(tmp_path, cache):
+    root_dir = tmp_path / "lib"
+    root_dir.mkdir()
+    gone = root_dir / "deleted.mkv"
+    # Never written to disk — simulates a file that was there when cached
+    # and has since been deleted.
+    await cache.put(str(gone), size=1, mtime=1.0, hash="a", type="video", added_at=1)
+
+    root = Root(name="lib", path=root_dir, available=True)
+    state = {"index_cache": cache, "indexers": {"g": _indexer([root])}}
+
+    result = await ops.prune_index_cache(state)
+
+    assert result == {"status": "pruned", "removed": 1, "kept": 0}
+    assert await cache.count() == 0
+
+
+@pytest.mark.asyncio
+async def test_prune_never_touches_a_path_under_an_unavailable_root(tmp_path, cache):
+    """
+    indexer.py's own rule: a root that goes away freezes, it never empties.
+    A disconnected drive must not have its cache wiped just because the
+    files can't be verified right now — that would force a full rehash of
+    the whole drive the moment it comes back, exactly the cost this cache
+    exists to avoid. Checked with a root directory that does not even exist
+    on disk (the strongest form of "can't verify") to prove availability,
+    not on-disk state, is what decides this.
+    """
+    root_dir = tmp_path / "usb"  # never created — simulates "unplugged"
+    unreachable = root_dir / "movie.mkv"
+    await cache.put(str(unreachable), size=1, mtime=1.0, hash="a",
+                     type="video", added_at=1)
+
+    root = Root(name="usb", path=root_dir, available=False)
+    state = {"index_cache": cache, "indexers": {"g": _indexer([root])}}
+
+    result = await ops.prune_index_cache(state)
+
+    assert result == {"status": "pruned", "removed": 0, "kept": 1}
+    assert await cache.count() == 1
+
+
+@pytest.mark.asyncio
+async def test_prune_across_two_groups_sharing_one_root_path(tmp_path, cache):
+    """The scenario this whole cache redesign is for: the same folder is a
+    root of two groups. A path under it must be kept as long as *either*
+    group's root still resolves it, available."""
+    root_dir = tmp_path / "shared"
+    root_dir.mkdir()
+    f = root_dir / "track.mp3"
+    f.write_bytes(b"x")
+    await cache.put(str(f), size=1, mtime=1.0, hash="a", type="audio", added_at=1)
+
+    root_a = Root(name="shared", path=root_dir, available=True)
+    root_b = Root(name="shared", path=root_dir, available=True)
+    state = {
+        "index_cache": cache,
+        "indexers": {"g1": _indexer([root_a]), "g2": _indexer([root_b])},
+    }
+
+    result = await ops.prune_index_cache(state)
+
+    assert result == {"status": "pruned", "removed": 0, "kept": 1}