"""Tests for PeerDrop Daemon server module using trio.""" import struct import tempfile from pathlib import Path from unittest.mock import MagicMock import pytest import trio from peerdrop.core.service import ServiceInterface, IdentityInfo from peerdrop.daemon.protocol import make_request, unpack_response from peerdrop.daemon.server import DaemonServer from peerdrop.utils.framing import length_prefix_pack, length_prefix_unpack def _make_mock_service(peer_id: str = "QmTest", addrs: list[str] | None = None) -> MagicMock: """Create a mock ServiceInterface.""" mock = MagicMock(spec=ServiceInterface) mock.get_identity.return_value = IdentityInfo(peer_id=peer_id, addrs=addrs or []) return mock class TestLengthPrefix: def test_pack_unpack(self): data = b"hello world" packed = length_prefix_pack(data) assert len(packed) == 4 + len(data) result = length_prefix_unpack(packed) assert result is not None unpacked, remaining = result assert unpacked == data assert remaining == b"" def test_multiple_messages_in_buffer(self): msg1 = length_prefix_pack(b"first") msg2 = length_prefix_pack(b"second") buffer = msg1 + msg2 result = length_prefix_unpack(buffer) assert result is not None unpacked1, buffer = result assert unpacked1 == b"first" result = length_prefix_unpack(buffer) assert result is not None unpacked2, buffer = result assert unpacked2 == b"second" assert buffer == b"" def test_incomplete_message(self): data = b"hello" packed = length_prefix_pack(data) incomplete = packed[:6] result = length_prefix_unpack(incomplete) assert result is None def test_empty_buffer(self): result = length_prefix_unpack(b"") assert result is None def test_header_only(self): result = length_prefix_unpack(struct.pack(">I", 10)) assert result is None @pytest.mark.trio async def test_server_start_stop(): """Test that server can start and stop cleanly.""" service = _make_mock_service() with tempfile.TemporaryDirectory() as tmpdir: sock_path = Path(tmpdir) / "test.sock" server = DaemonServer(service, sock_path=sock_path) async with trio.open_nursery() as nursery: await nursery.start(server.serve) assert server._running is True assert sock_path.exists() nursery.cancel_scope.cancel() assert server._running is False assert not sock_path.exists() @pytest.mark.trio async def test_server_client_connection(): """Test that a client can connect and send a request.""" service = _make_mock_service(peer_id="QmTest") with tempfile.TemporaryDirectory() as tmpdir: sock_path = Path(tmpdir) / "test.sock" server = DaemonServer(service, sock_path=sock_path) async with trio.open_nursery() as nursery: await nursery.start(server.serve) # Connect as client stream = await trio.open_unix_socket(str(sock_path)) # Send a health request request = make_request("health", request_id="test-1") await stream.send_all(length_prefix_pack(request)) # Read response response_data = await stream.receive_some(4096) result = length_prefix_unpack(response_data) assert result is not None msg, _ = result response = unpack_response(msg) assert response["ok"] is True assert response["data"]["status"] == "ok" assert response["data"]["peer_id"] == "QmTest" await stream.aclose() nursery.cancel_scope.cancel() @pytest.mark.trio async def test_server_multiple_clients(): """Test that multiple clients can connect simultaneously.""" service = _make_mock_service() with tempfile.TemporaryDirectory() as tmpdir: sock_path = Path(tmpdir) / "test.sock" server = DaemonServer(service, sock_path=sock_path) async with trio.open_nursery() as nursery: await nursery.start(server.serve) # Connect two clients s1 = await trio.open_unix_socket(str(sock_path)) s2 = await trio.open_unix_socket(str(sock_path)) # Both send health requests await s1.send_all(length_prefix_pack(make_request("health", request_id="c1"))) await s2.send_all(length_prefix_pack(make_request("health", request_id="c2"))) # Both get responses resp1_data = await s1.receive_some(4096) resp2_data = await s2.receive_some(4096) msg1 = length_prefix_unpack(resp1_data) msg2 = length_prefix_unpack(resp2_data) assert msg1 is not None assert msg2 is not None resp1 = unpack_response(msg1[0]) resp2 = unpack_response(msg2[0]) assert resp1["id"] == "c1" assert resp2["id"] == "c2" await s1.aclose() await s2.aclose() nursery.cancel_scope.cancel() @pytest.mark.trio async def test_server_client_disconnect(): """Test that server handles client disconnect gracefully.""" service = _make_mock_service() with tempfile.TemporaryDirectory() as tmpdir: sock_path = Path(tmpdir) / "test.sock" server = DaemonServer(service, sock_path=sock_path) async with trio.open_nursery() as nursery: await nursery.start(server.serve) # Connect and immediately disconnect s = await trio.open_unix_socket(str(sock_path)) await s.aclose() # Give server time to handle disconnect await trio.sleep(0.1) assert server.client_count == 0 nursery.cancel_scope.cancel() @pytest.mark.trio async def test_server_multiple_requests(): """Test that a client can send multiple sequential requests.""" service = _make_mock_service() with tempfile.TemporaryDirectory() as tmpdir: sock_path = Path(tmpdir) / "test.sock" server = DaemonServer(service, sock_path=sock_path) async with trio.open_nursery() as nursery: await nursery.start(server.serve) s = await trio.open_unix_socket(str(sock_path)) # Send first request await s.send_all(length_prefix_pack(make_request("health", request_id="req1"))) resp1 = await s.receive_some(4096) msg1 = length_prefix_unpack(resp1) assert msg1 is not None assert unpack_response(msg1[0])["id"] == "req1" # Send second request await s.send_all(length_prefix_pack(make_request("get_identity", request_id="req2"))) resp2 = await s.receive_some(4096) msg2 = length_prefix_unpack(resp2) assert msg2 is not None assert unpack_response(msg2[0])["id"] == "req2" await s.aclose() nursery.cancel_scope.cancel() @pytest.mark.trio async def test_broadcast_event(): """Test event broadcasting (async method).""" service = _make_mock_service() server = DaemonServer(service) server._clients = set() # Should not raise even with no clients await server.broadcast_event("test_event", {"key": "value"})