path: root/packages/meshbay-node/tests/test_daemon.py

"""
Integration test: Node daemon wires all components correctly.

Phase 11 — verifies that NodeDaemon creates chat stores, WebRTC transport,
index push on change, swarm registration, and shuts down cleanly.
Hub interaction is mocked.
"""

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_common.crypto import generate_gek
from meshbay_node.config import Config, HubConfig, NodeConfig, GroupConfig, KeystoreConfig
from conftest import one_root
from meshbay_node.daemon import NodeDaemon
from meshbay_node.indexer import DirectoryIndexer

def _mock_keystore_keys(sk_ed):
    """Create a mock keystore with real Ed25519 + X25519 key material."""
    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

def _free_port() -> int:
    """
    A port nobody else in the session is on.

    These tests start the real admin UI server. Hardcoding 28000 made them fail
    with EADDRINUSE whenever another test file had a node running — which is why
    the full suite failed while each file passed on its own.
    """
    import socket
    with socket.socket() as s:
        s.bind(("127.0.0.1", 0))
        return s.getsockname()[1]


@pytest.fixture
def sk_hub():
    return Ed25519PrivateKey.generate()

@pytest.fixture
def hub_pk_pem(sk_hub):
    return sk_hub.public_key().public_bytes(
        serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo)

@pytest.fixture
def gek():
    return generate_gek()

@pytest.fixture
def shared_dir(tmp_path):
    d = tmp_path / "shared"
    d.mkdir()
    (d / "test.bin").write_bytes(os.urandom(2048))
    (d / "hello.txt").write_bytes(b"hello daemon test " * 50)
    return d

@pytest.fixture
def node_config(tmp_path, shared_dir):
    return Config(
        hub=HubConfig(url="http://localhost:9999", username="testuser"),
        node=NodeConfig(quic_port=_free_port(), ui_port=_free_port()),
        groups=[GroupConfig(
            id="g" * 32,
            name="test-group",
            shared_dir=str(shared_dir),
            visibility="private",
            quic_port=29010,
        )],
        keystore=KeystoreConfig(path=tmp_path / "keystore.enc"),
        data_dir=tmp_path / "data",
    )

@pytest.mark.asyncio
async def test_daemon_creates_chat_store(tmp_path, node_config, gek, hub_pk_pem):
    """Daemon creates ChatStore for each group and shuts down cleanly."""
    daemon = NodeDaemon(node_config)

    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 = hub_pk_pem

    with patch("meshbay_node.daemon.load_or_create_keystore", return_value=mock_keys), \
         patch("meshbay_node.daemon.HubClient") as MockHub:

        hub_instance = AsyncMock()
        hub_instance.startup = AsyncMock(return_value=mock_session)
        hub_instance.maintain_ws = AsyncMock()
        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=5)
                except (asyncio.TimeoutError, Exception):
                    pass

        task = asyncio.create_task(run_daemon())
        await asyncio.sleep(1)

        assert daemon._state["status"] == "running"
        group_id = "g" * 32
        assert group_id in daemon._chat_stores
        assert daemon._chat_stores[group_id]._db is not None

        if daemon._webrtc:
            # Finding H1: chat_store must live in the per-group context, never on
            # the shared transport context. Hoisting the first group's store
            # transport-wide sent every group's chat to one database and served it
            # back to members of every other group.
            assert "chat_store" not in daemon._webrtc._ctx
            groups_ctx = daemon._webrtc._ctx["groups"]
            assert groups_ctx[group_id]["chat_store"] is daemon._chat_stores[group_id]

            assert "hub_ws" in daemon._webrtc._ctx
            assert "node_user_id" in daemon._webrtc._ctx
            assert daemon._webrtc._ctx["node_user_id"] == "user123"

        shutdown_event.set()
        await daemon._shutdown()
        task.cancel()
        try:
            await task
        except (asyncio.CancelledError, Exception):
            pass

        for store in daemon._chat_stores.values():
            assert store._db is None

@pytest.mark.asyncio
async def test_daemon_no_groups_stays_up(tmp_path, hub_pk_pem):
    """Daemon with no valid groups stays up (admin UI + hub connection alive)."""
    config = Config(
        hub=HubConfig(url="http://localhost:9999", username="testuser"),
        node=NodeConfig(quic_port=_free_port(), ui_port=_free_port()),
        groups=[GroupConfig(id="", name="empty", shared_dir="")],
        keystore=KeystoreConfig(path=tmp_path / "keystore.enc"),
        data_dir=tmp_path / "data",
    )
    daemon = NodeDaemon(config)

    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 = hub_pk_pem

    shutdown_event = asyncio.Event()

    with patch("meshbay_node.daemon.load_or_create_keystore", return_value=mock_keys), \
         patch("meshbay_node.daemon.HubClient") as MockHub:

        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

        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=5)
                except (asyncio.TimeoutError, Exception):
                    pass

        task = asyncio.create_task(run_daemon())
        await asyncio.sleep(1)

        assert daemon._state["status"] == "running"
        assert len(daemon._chat_stores) == 0

        shutdown_event.set()
        await daemon._shutdown()
        task.cancel()
        try:
            await task
        except (asyncio.CancelledError, Exception):
            pass

@pytest.mark.asyncio
async def test_daemon_index_change_pushes_to_peers(tmp_path, shared_dir, gek, hub_pk_pem):
    """Index change callback pushes updated index to WebRTC peers."""
    config = Config(
        hub=HubConfig(url="http://localhost:9999", username="testuser"),
        node=NodeConfig(quic_port=_free_port(), ui_port=_free_port()),
        groups=[GroupConfig(
            id="a" * 32,
            name="test-group",
            shared_dir=str(shared_dir),
            visibility="private",
            quic_port=29010,
        )],
        keystore=KeystoreConfig(path=tmp_path / "keystore.enc"),
        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"

    sk_node = Ed25519PrivateKey.generate()
    indexer = DirectoryIndexer(
        roots=one_root(shared_dir), group_id="a" * 32,
        sk_node=sk_node, gek=gek)
    await indexer.initial_scan()

    mock_session = MagicMock()
    mock_session._group_id = "a" * 32
    mock_session._send = MagicMock()

    mock_webrtc = MagicMock()
    mock_webrtc._sessions = {"peer1": mock_session}
    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]
    assert msg["type"] == "index_sync"
    assert msg["group_id"] == "a" * 32
    assert len(msg["entries"]) == indexer.index.count

    # Finding H7: this group is private, so its content hashes must NOT be
    # registered with the hub. The test previously asserted the opposite —
    # publishing a fingerprint of every private file was treated as expected
    # behaviour. Index push to members is unaffected (asserted above).
    await asyncio.sleep(0.1)
    daemon._hub.register_swarm.assert_not_called()

@pytest.mark.asyncio
async def test_daemon_index_change_registers_swarm_for_public_group(
        tmp_path, shared_dir, gek, hub_pk_pem):
    """Public groups still register content hashes with the hub swarm (H7)."""
    config = Config(
        hub=HubConfig(url="http://localhost:9999", username="testuser"),
        node=NodeConfig(quic_port=_free_port(), ui_port=_free_port()),
        groups=[GroupConfig(
            id="a" * 32,
            name="public-group",
            shared_dir=str(shared_dir),
            visibility="public",
            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=2)
    daemon._state["endpoint_hint"] = "node123"

    indexer = DirectoryIndexer(
        roots=one_root(shared_dir), group_id="a" * 32,
        sk_node=Ed25519PrivateKey.generate(), gek=gek)
    await indexer.initial_scan()

    await daemon._on_index_change(indexer)

    await asyncio.sleep(0.1)
    daemon._hub.register_swarm.assert_called_once()
    assert len(daemon._hub.register_swarm.call_args[0][0]) == indexer.index.count


@pytest.mark.asyncio
async def test_daemon_index_change_skips_other_group_peers(
    tmp_path, shared_dir, gek, hub_pk_pem
):
    """Index change only pushes to peers in the same group."""
    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",
    )
    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"

    sk_node = Ed25519PrivateKey.generate()
    indexer = DirectoryIndexer(
        roots=one_root(shared_dir), group_id="a" * 32,
        sk_node=sk_node, gek=gek)
    await indexer.initial_scan()

    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

    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_media_cache_not_pruned_when_another_group_still_has_the_content(
        tmp_path, shared_dir, gek):
    """
    media_cache.db is node-wide, keyed by content hash — a file shared into
    two groups is one row there. Removing it from ONE group's index (root
    unshared, group left) must not wipe the thumbnail/tmdb/mbid mapping the
    OTHER group's copy still needs, or that surviving group pays for a
    redundant re-fetch/re-probe/re-thumbnail for content it never lost.
    """
    daemon = _new_daemon_for_group(tmp_path, shared_dir, gek, group_id="a" * 32)
    daemon._media_cache = AsyncMock()
    daemon._webrtc = MagicMock()
    daemon._webrtc._sessions = {}

    indexer_a = DirectoryIndexer(roots=one_root(shared_dir), group_id="a" * 32,
                                 sk_node=Ed25519PrivateKey.generate(), gek=gek)
    await indexer_a.initial_scan()
    shared_id = indexer_a.index.entries[0].id

    indexer_b = DirectoryIndexer(roots=one_root(shared_dir), group_id="b" * 32,
                                 sk_node=Ed25519PrivateKey.generate(), gek=gek)
    await indexer_b.initial_scan()
    assert indexer_b.index.get_entry(shared_id) is not None

    daemon._indexers = [indexer_a, indexer_b]

    await daemon._on_index_change(indexer_a)  # establishes the snapshot
    await asyncio.sleep(0.05)

    indexer_a.index.remove_entry(shared_id)
    await daemon._on_index_change(indexer_a)
    await asyncio.sleep(0.05)

    daemon._media_cache.prune_file.assert_not_called()


@pytest.mark.asyncio
async def test_media_cache_pruned_once_no_group_has_the_content_left(
        tmp_path, shared_dir, gek):
    """Counterpart of the test above: with only one group ever having held
    the content, its removal must still prune media_cache as before — the
    fix only withholds pruning when the content genuinely survives
    elsewhere, it must not make pruning stop happening altogether."""
    daemon = _new_daemon_for_group(tmp_path, shared_dir, gek, group_id="a" * 32)
    daemon._media_cache = AsyncMock()
    daemon._webrtc = MagicMock()
    daemon._webrtc._sessions = {}

    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
    daemon._indexers = [indexer]

    await daemon._on_index_change(indexer)
    await asyncio.sleep(0.05)

    indexer.index.remove_entry(removed_id)
    await daemon._on_index_change(indexer)
    await asyncio.sleep(0.05)

    daemon._media_cache.prune_file.assert_called_once_with(removed_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"