test: add tests for fetchone, fetchall, fetchval, and protocol methods

Fills major coverage gaps:
- fetchone (returns first row, returns None for empty)
- fetchall (returns list of lists)
- fetchval (returns first column, returns None for empty)
- protocol get_leader, prepare, finalize
- protocol connection-closed-during-read error path

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: antoineleclair

tests/test_connection.py

@@ -231,3 +231,158 @@ async def test_connection_invalidated_after_protocol_error(self) -> None:
 
         # Connection should be invalidated
         assert not conn.is_connected
+
+    async def test_fetchone_returns_first_row(self) -> None:
+        conn = DqliteConnection("localhost:9001")
+
+        mock_reader = AsyncMock()
+        mock_writer = MagicMock()
+        mock_writer.drain = AsyncMock()
+        mock_writer.close = MagicMock()
+        mock_writer.wait_closed = AsyncMock()
+
+        from dqlitewire.constants import ValueType
+        from dqlitewire.messages import DbResponse, RowsResponse, WelcomeResponse
+
+        responses = [
+            WelcomeResponse(heartbeat_timeout=15000).encode(),
+            DbResponse(db_id=1).encode(),
+            RowsResponse(
+                column_names=["id", "name"],
+                column_types=[ValueType.INTEGER, ValueType.TEXT],
+                rows=[[1, "first"], [2, "second"]],
+                has_more=False,
+            ).encode(),
+        ]
+        mock_reader.read.side_effect = responses
+
+        with patch("asyncio.open_connection", return_value=(mock_reader, mock_writer)):
+            await conn.connect()
+
+        mock_reader.read.side_effect = [responses[2]]
+        result = await conn.fetchone("SELECT * FROM t")
+        assert result == {"id": 1, "name": "first"}
+
+    async def test_fetchone_returns_none_for_empty(self) -> None:
+        conn = DqliteConnection("localhost:9001")
+
+        mock_reader = AsyncMock()
+        mock_writer = MagicMock()
+        mock_writer.drain = AsyncMock()
+        mock_writer.close = MagicMock()
+        mock_writer.wait_closed = AsyncMock()
+
+        from dqlitewire.constants import ValueType
+        from dqlitewire.messages import DbResponse, RowsResponse, WelcomeResponse
+
+        responses = [
+            WelcomeResponse(heartbeat_timeout=15000).encode(),
+            DbResponse(db_id=1).encode(),
+        ]
+        mock_reader.read.side_effect = responses
+
+        with patch("asyncio.open_connection", return_value=(mock_reader, mock_writer)):
+            await conn.connect()
+
+        empty_response = RowsResponse(
+            column_names=["id"],
+            column_types=[ValueType.INTEGER],
+            rows=[],
+            has_more=False,
+        ).encode()
+        mock_reader.read.side_effect = [empty_response]
+        result = await conn.fetchone("SELECT * FROM t WHERE 1=0")
+        assert result is None
+
+    async def test_fetchall_returns_lists(self) -> None:
+        conn = DqliteConnection("localhost:9001")
+
+        mock_reader = AsyncMock()
+        mock_writer = MagicMock()
+        mock_writer.drain = AsyncMock()
+        mock_writer.close = MagicMock()
+        mock_writer.wait_closed = AsyncMock()
+
+        from dqlitewire.constants import ValueType
+        from dqlitewire.messages import DbResponse, RowsResponse, WelcomeResponse
+
+        responses = [
+            WelcomeResponse(heartbeat_timeout=15000).encode(),
+            DbResponse(db_id=1).encode(),
+        ]
+        mock_reader.read.side_effect = responses
+
+        with patch("asyncio.open_connection", return_value=(mock_reader, mock_writer)):
+            await conn.connect()
+
+        rows_response = RowsResponse(
+            column_names=["id", "name"],
+            column_types=[ValueType.INTEGER, ValueType.TEXT],
+            rows=[[1, "a"], [2, "b"]],
+            has_more=False,
+        ).encode()
+        mock_reader.read.side_effect = [rows_response]
+        result = await conn.fetchall("SELECT * FROM t")
+        assert result == [[1, "a"], [2, "b"]]
+
+    async def test_fetchval_returns_first_column(self) -> None:
+        conn = DqliteConnection("localhost:9001")
+
+        mock_reader = AsyncMock()
+        mock_writer = MagicMock()
+        mock_writer.drain = AsyncMock()
+        mock_writer.close = MagicMock()
+        mock_writer.wait_closed = AsyncMock()
+
+        from dqlitewire.constants import ValueType
+        from dqlitewire.messages import DbResponse, RowsResponse, WelcomeResponse
+
+        responses = [
+            WelcomeResponse(heartbeat_timeout=15000).encode(),
+            DbResponse(db_id=1).encode(),
+        ]
+        mock_reader.read.side_effect = responses
+
+        with patch("asyncio.open_connection", return_value=(mock_reader, mock_writer)):
+            await conn.connect()
+
+        rows_response = RowsResponse(
+            column_names=["count"],
+            column_types=[ValueType.INTEGER],
+            rows=[[42]],
+            has_more=False,
+        ).encode()
+        mock_reader.read.side_effect = [rows_response]
+        result = await conn.fetchval("SELECT count(*) FROM t")
+        assert result == 42
+
+    async def test_fetchval_returns_none_for_empty(self) -> None:
+        conn = DqliteConnection("localhost:9001")
+
+        mock_reader = AsyncMock()
+        mock_writer = MagicMock()
+        mock_writer.drain = AsyncMock()
+        mock_writer.close = MagicMock()
+        mock_writer.wait_closed = AsyncMock()
+
+        from dqlitewire.constants import ValueType
+        from dqlitewire.messages import DbResponse, RowsResponse, WelcomeResponse
+
+        responses = [
+            WelcomeResponse(heartbeat_timeout=15000).encode(),
+            DbResponse(db_id=1).encode(),
+        ]
+        mock_reader.read.side_effect = responses
+
+        with patch("asyncio.open_connection", return_value=(mock_reader, mock_writer)):
+            await conn.connect()
+
+        empty_response = RowsResponse(
+            column_names=["id"],
+            column_types=[ValueType.INTEGER],
+            rows=[],
+            has_more=False,
+        ).encode()
+        mock_reader.read.side_effect = [empty_response]
+        result = await conn.fetchval("SELECT id FROM t WHERE 1=0")
+        assert result is None

tests/test_protocol.py

@@ -138,6 +138,59 @@ async def test_query_sql_multipart(
         assert rows[0] == [1, "alice"]
         assert rows[1] == [2, "bob"]
 
+    async def test_get_leader(
+        self,
+        protocol: DqliteProtocol,
+        mock_reader: AsyncMock,
+        leader_response: bytes,
+    ) -> None:
+        mock_reader.read.return_value = leader_response
+
+        node_id, address = await protocol.get_leader()
+
+        assert node_id == 1
+        assert address == "localhost:9001"
+
+    async def test_prepare(
+        self,
+        protocol: DqliteProtocol,
+        mock_reader: AsyncMock,
+    ) -> None:
+        from dqlitewire.messages import StmtResponse
+
+        mock_reader.read.return_value = StmtResponse(
+            db_id=1, stmt_id=1, num_params=2
+        ).encode()
+
+        stmt_id, num_params = await protocol.prepare(1, "INSERT INTO t VALUES (?, ?)")
+
+        assert stmt_id == 1
+        assert num_params == 2
+
+    async def test_finalize(
+        self,
+        protocol: DqliteProtocol,
+        mock_reader: AsyncMock,
+    ) -> None:
+        from dqlitewire.messages import EmptyResponse
+
+        mock_reader.read.return_value = EmptyResponse().encode()
+
+        # Should not raise
+        await protocol.finalize(1, 1)
+
+    async def test_connection_closed_during_read(
+        self,
+        protocol: DqliteProtocol,
+        mock_reader: AsyncMock,
+    ) -> None:
+        mock_reader.read.return_value = b""
+
+        from dqliteclient.exceptions import DqliteConnectionError
+
+        with pytest.raises(DqliteConnectionError, match="Connection closed"):
+            await protocol.exec_sql(1, "SELECT 1")
+
     async def test_read_timeout(
         self,
         mock_reader: AsyncMock,