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_exits(tmp_path):
    """Daemon with no valid groups exits cleanly."""
    config = Config(
        hub=HubConfig(url="http://localhost:9999", username="testuser"),
        node=NodeConfig(),
        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 = b"pem"

    mock_server = AsyncMock()
    mock_server.serve = AsyncMock()

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

        mock_uvicorn.Config = MagicMock()
        mock_uvicorn.Server = MagicMock(return_value=mock_server)

        hub_instance = AsyncMock()
        hub_instance.startup = AsyncMock(return_value=mock_session)
        hub_instance.close = AsyncMock()
        hub_instance.__aenter__ = AsyncMock(return_value=hub_instance)
        hub_instance.__aexit__ = AsyncMock(return_value=False)
        MockHub.return_value = hub_instance

        await daemon.run()

    assert len(daemon._chat_stores) == 0

@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._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)

    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._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._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)

    same_group._send.assert_called_once()
    other_group._send.assert_not_called()