"""Tests for indexer and GroupIndex.""" import asyncio import os import time import pytest from pathlib import Path 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 conftest import one_root from meshbay_node.keystore import NodeKeys @pytest.fixture def sk_node(): return Ed25519PrivateKey.generate() @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(1024)) (d / "music.mp3").write_bytes(os.urandom(512)) (d / "readme.md").write_bytes(b"# Hello MeshBay") subdir = d / "docs" subdir.mkdir() (subdir / "manual.pdf").write_bytes(os.urandom(2048)) return d # ── GroupIndex tests ────────────────────────────────────────────────────────── def test_group_index_serialize_deserialize_private(sk_node, gek, shared_dir): idx = GroupIndex(group_id="grp-001", sk_node=sk_node, gek=gek) from meshbay_common.protocol import IndexEntry idx.add_entry(IndexEntry( id="abc123", name="video.mkv", path="", size=1024, type="video", added_at=int(time.time()), duration=120)) wire = idx.serialize() recovered = GroupIndex.deserialize(wire, sk_node=sk_node, gek=gek) assert recovered.group_id == "grp-001" assert recovered.count == 1 assert recovered.entries[0].name == "video.mkv" assert recovered.entries[0].type == "video" def test_group_index_serialize_deserialize_public(sk_node): idx = GroupIndex(group_id="pub-001", sk_node=sk_node, gek=None) from meshbay_common.protocol import IndexEntry idx.add_entry(IndexEntry( id="xyz789", name="readme.txt", path="", size=42, type="document", added_at=int(time.time()))) wire = idx.serialize() recovered = GroupIndex.deserialize(wire, sk_node=sk_node, gek=None) assert recovered.count == 1 assert recovered.entries[0].id == "xyz789" def test_group_index_wrong_gek_rejected(sk_node, gek): idx = GroupIndex(group_id="grp-002", sk_node=sk_node, gek=gek) from meshbay_common.protocol import IndexEntry idx.add_entry(IndexEntry(id="a", name="f.mp3", path="", size=1, type="audio", added_at=0)) wire = idx.serialize() wrong_gek = generate_gek() with pytest.raises(Exception): # InvalidTag from AEAD GroupIndex.deserialize(wire, sk_node=sk_node, gek=wrong_gek) def test_group_index_tampered_rejected(sk_node, gek): idx = GroupIndex(group_id="grp-003", sk_node=sk_node, gek=gek) from meshbay_common.protocol import IndexEntry idx.add_entry(IndexEntry(id="b", name="f.mp4", path="", size=1, type="video", added_at=0)) wire = bytearray(idx.serialize()) wire[-5] ^= 0xFF # flip bytes at the end with pytest.raises(Exception): GroupIndex.deserialize(bytes(wire), sk_node=sk_node, gek=gek) def test_group_index_diff(sk_node, gek): from meshbay_common.protocol import IndexEntry v1 = GroupIndex(group_id="g", sk_node=sk_node, gek=gek, version=1) v1.add_entry(IndexEntry(id="aaa", name="a.mp4", path="", size=1, type="video", added_at=0)) v1.add_entry(IndexEntry(id="bbb", name="b.mp3", path="", size=1, type="audio", added_at=0)) v2 = GroupIndex(group_id="g", sk_node=sk_node, gek=gek, version=2) v2.add_entry(IndexEntry(id="aaa", name="a.mp4", path="", size=1, type="video", added_at=0)) v2.add_entry(IndexEntry(id="ccc", name="c.mkv", path="", size=1, type="video", added_at=0)) delta = v2.diff(v1) assert delta.base_version == 1 assert delta.version == 2 assert len(delta.additions) == 1 assert delta.additions[0].id == "ccc" assert "bbb" in delta.deletions # ── DirectoryIndexer tests ──────────────────────────────────────────────────── @pytest.mark.asyncio async def test_initial_scan(shared_dir, sk_node, gek): indexer = DirectoryIndexer( roots=one_root(shared_dir), group_id="scan-test", sk_node=sk_node, gek=gek, ) await indexer.initial_scan() entries = indexer.index.entries names = {e.name for e in entries} assert "video.mkv" in names assert "music.mp3" in names assert "readme.md" in names assert "manual.pdf" in names assert indexer.index.count == 4 @pytest.mark.asyncio async def test_type_detection(shared_dir, sk_node, gek): indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek) await indexer.initial_scan() by_name = {e.name: e.type for e in indexer.index.entries} assert by_name["video.mkv"] == "video" assert by_name["music.mp3"] == "audio" assert by_name["readme.md"] == "document" assert by_name["manual.pdf"] == "document" @pytest.mark.asyncio async def test_hidden_files_excluded(tmp_path, sk_node, gek): d = tmp_path / "dir" d.mkdir() (d / ".hidden").write_bytes(b"secret") (d / "visible.txt").write_bytes(b"visible") (d / "file.tmp").write_bytes(b"tmp") indexer = DirectoryIndexer(roots=one_root(d), group_id="g", sk_node=sk_node, gek=gek) await indexer.initial_scan() names = {e.name for e in indexer.index.entries} assert "visible.txt" in names assert ".hidden" not in names assert "file.tmp" not in names @pytest.mark.asyncio async def test_on_change_callback(shared_dir, sk_node, gek): changes = [] async def on_change(idx): changes.append(idx.index.count) indexer = DirectoryIndexer( roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek, on_change=on_change) await indexer.start() await asyncio.sleep(0.1) (shared_dir / "newfile.mp4").write_bytes(os.urandom(256)) await asyncio.sleep(3.0) # watchdog detect + 2s debounce await indexer.stop() assert len(changes) >= 1, "on_change should have been called" @pytest.mark.asyncio async def test_index_roundtrip_after_scan(shared_dir, sk_node, gek): indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek) await indexer.initial_scan() wire = indexer.index.serialize() recovered = GroupIndex.deserialize(wire, sk_node=sk_node, gek=gek) assert recovered.count == indexer.index.count