""" 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 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 @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=29010, ui_port=28000), 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=29010, ui_port=28000), 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( 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 await asyncio.sleep(0.1) daemon._hub.register_swarm.assert_called_once() call_args = daemon._hub.register_swarm.call_args assert len(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( 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()