Pin three client-package defensive branches that lacked direct tests

- ``DqliteProtocol.is_wire_coherent`` is a one-line forwarder over
  ``self._decoder.is_poisoned``. Existing pool-reset tests stubbed
  the property as a boolean attribute on a mock; the real getter was
  never exercised against a real ``Decoder`` instance. A future
  refactor that swapped the underlying field name would silently
  break every short-circuit caller.

- ``_starts_with_tx_verb`` and
  ``Connection._sql_is_outermost_release_or_commit`` both strip
  leading SQL comments / whitespace and return ``False`` if the
  result is empty. The empty-after-strip case was exercised
  indirectly via ``_split_top_level_statements``; pin it directly so
  a refactor of either predicate is caught at the predicate level.

- ``ConnectionPool._reset_connection`` distinguishes leader-flip
  ROLLBACK failures (DEBUG; routine cluster churn) from genuine
  server faults (WARNING + traceback). The branching was untested,
  so a future refactor that flattened the two arms to the same level
  would silently degrade the operational signal.

Author: antoineleclair

tests/test_pool_reset_leader_class_log_level.py

@@ -0,0 +1,129 @@
+"""Pin: ``ConnectionPool._reset_connection`` distinguishes leader-flip
+ROLLBACK failures (DEBUG) from genuine server faults (WARNING).
+
+A code in ``LEADER_ERROR_CODES`` represents routine cluster churn —
+the new leader has no record of our transaction, so the ROLLBACK
+fails predictably. Operators should not get a WARNING-level log line
+per leader flip on a busy pool. Codes outside that set indicate an
+actual fault and DO warrant the WARNING + traceback so the operator
+can investigate.
+
+Without this test, a future refactor that flattened the two arms to
+the same level (or swapped them) would silently degrade the
+operational signal.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from dqliteclient.cluster import ClusterClient
+from dqliteclient.exceptions import OperationalError
+from dqliteclient.pool import ConnectionPool
+from dqlitewire import LEADER_ERROR_CODES
+
+
+def _make_pool() -> ConnectionPool:
+    cluster = MagicMock(spec=ClusterClient)
+    return ConnectionPool(
+        addresses=["localhost:9001"],
+        min_size=0,
+        max_size=1,
+        timeout=1.0,
+        cluster=cluster,
+    )
+
+
+def _conn_in_tx(execute_side_effect: BaseException) -> Any:
+    """Build a fake conn that ``_reset_connection`` will route through
+    the ROLLBACK arm. ``_in_transaction=True`` selects the arm; the
+    socket-liveness check is bypassed by leaving the protocol stub
+    looking healthy."""
+    conn = MagicMock()
+    conn._address = "localhost:9001"
+    conn._in_transaction = True
+    conn._tx_owner = None
+    conn._savepoint_stack = []
+    conn._savepoint_implicit_begin = False
+    conn._has_untracked_savepoint = False
+    # Look-alive for the cheap pre-write liveness check.
+    conn._protocol = MagicMock()
+    conn._protocol._writer = MagicMock()
+    conn._protocol._writer.transport = MagicMock()
+    conn._protocol._writer.transport.is_closing = lambda: False
+    conn._protocol._reader = MagicMock()
+    conn._protocol._reader.at_eof = lambda: False
+    conn._protocol.is_wire_coherent = True
+    conn.execute = AsyncMock(side_effect=execute_side_effect)
+    return conn
+
+
+@pytest.mark.asyncio
+async def test_reset_leader_class_rollback_failure_logs_debug(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """A ROLLBACK failure with a ``LEADER_ERROR_CODES`` code logs at
+    DEBUG, not WARNING — routine cluster churn shouldn't trigger
+    operator alerting."""
+    pool = _make_pool()
+    leader_code = next(iter(LEADER_ERROR_CODES))
+    err = OperationalError(leader_code, "not leader")
+    conn = _conn_in_tx(err)
+
+    with caplog.at_level(logging.DEBUG, logger="dqliteclient.pool"):
+        result = await pool._reset_connection(conn)
+
+    assert result is False, "leader-class ROLLBACK failure must drop the conn"
+    debug_records = [
+        r
+        for r in caplog.records
+        if r.levelno == logging.DEBUG and "leader-class ROLLBACK failure" in r.message
+    ]
+    warning_records = [r for r in caplog.records if r.levelno == logging.WARNING]
+    assert debug_records, (
+        f"expected a DEBUG record about leader-class ROLLBACK failure; "
+        f"got {[(r.levelno, r.message) for r in caplog.records]!r}"
+    )
+    assert not warning_records, (
+        "leader-class ROLLBACK failure must NOT produce a WARNING — "
+        "routine cluster churn shouldn't trigger operator alerting; "
+        f"got {[(r.levelno, r.message) for r in warning_records]!r}"
+    )
+
+
+@pytest.mark.asyncio
+async def test_reset_non_leader_rollback_failure_logs_warning(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """A ROLLBACK failure with a code OUTSIDE ``LEADER_ERROR_CODES``
+    is a genuine server fault and logs at WARNING with traceback so
+    the operator can investigate."""
+    pool = _make_pool()
+    # Pick a code that is NOT in LEADER_ERROR_CODES.
+    non_leader_code = 1
+    assert non_leader_code not in LEADER_ERROR_CODES
+    err = OperationalError(non_leader_code, "disk I/O error")
+    conn = _conn_in_tx(err)
+
+    with caplog.at_level(logging.WARNING, logger="dqliteclient.pool"):
+        result = await pool._reset_connection(conn)
+
+    assert result is False
+    warning_records = [
+        r
+        for r in caplog.records
+        if r.levelno == logging.WARNING and "ROLLBACK failure" in r.message
+    ]
+    assert warning_records, (
+        f"expected a WARNING record about ROLLBACK failure; "
+        f"got {[(r.levelno, r.message) for r in caplog.records]!r}"
+    )
+    # WARNING path must include exc_info for traceback so the operator
+    # can investigate the genuine fault.
+    assert warning_records[0].exc_info is not None, (
+        "non-leader ROLLBACK failure WARNING must include exc_info=True"
+    )

tests/test_protocol_is_wire_coherent.py

@@ -0,0 +1,46 @@
+"""Pin: ``DqliteProtocol.is_wire_coherent`` routes through the
+underlying decoder's ``is_poisoned`` flag.
+
+Existing pool-reset tests stub ``protocol.is_wire_coherent`` as a
+boolean attribute on a mock; the real one-line forwarder
+(``return not self._decoder.is_poisoned``) at
+``protocol.py:190`` was never exercised against a real ``Decoder``
+instance. Without a direct test, a future refactor that swapped the
+underlying field name (e.g. ``is_poisoned`` → ``is_corrupt``) would
+silently break every pool-reset short-circuit caller.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from unittest.mock import MagicMock
+
+from dqliteclient.protocol import DqliteProtocol
+from dqlitewire.exceptions import DecodeError
+
+
+def _make_protocol() -> DqliteProtocol:
+    """Construct a ``DqliteProtocol`` against mock reader/writer.
+
+    The protocol's ``is_wire_coherent`` property reads from
+    ``self._decoder``, which is a real ``MessageDecoder`` instance —
+    so the test exercises the real getter.
+    """
+    reader = MagicMock(spec=asyncio.StreamReader)
+    writer = MagicMock(spec=asyncio.StreamWriter)
+    return DqliteProtocol(reader=reader, writer=writer)
+
+
+def test_is_wire_coherent_true_when_decoder_not_poisoned() -> None:
+    proto = _make_protocol()
+    # Fresh decoder: not poisoned. The forwarder returns True.
+    assert proto._decoder.is_poisoned is False
+    assert proto.is_wire_coherent is True
+
+
+def test_is_wire_coherent_false_when_decoder_poisoned() -> None:
+    proto = _make_protocol()
+    # Poison the underlying decoder. The forwarder returns False.
+    proto._decoder._buffer.poison(DecodeError("forced poison for test"))
+    assert proto._decoder.is_poisoned is True
+    assert proto.is_wire_coherent is False

tests/test_tx_control_helpers_empty_after_strip.py

@@ -0,0 +1,67 @@
+"""Pin: ``_starts_with_tx_verb`` and
+``_sql_is_outermost_release_or_commit`` return ``False`` on
+all-comment / whitespace-only input.
+
+Both predicates strip leading SQL comments and whitespace then
+return ``False`` if the result is empty. This is currently exercised
+indirectly via ``_split_top_level_statements`` (see
+``test_split_top_level_statements_all_comment_input.py``), but the
+predicates themselves have no direct pin. A future refactor that
+tightened the empty-after-strip case to an exception or to ``True``
+would ripple silently through both call sites.
+"""
+
+from __future__ import annotations
+
+from dqliteclient import DqliteConnection
+from dqliteclient.connection import _starts_with_tx_verb
+
+
+def test_starts_with_tx_verb_returns_false_for_pure_dash_dash_comment() -> None:
+    assert _starts_with_tx_verb("-- comment\n") is False
+
+
+def test_starts_with_tx_verb_returns_false_for_pure_block_comment() -> None:
+    assert _starts_with_tx_verb("/* x */") is False
+
+
+def test_starts_with_tx_verb_returns_false_for_whitespace_only() -> None:
+    assert _starts_with_tx_verb("   ") is False
+
+
+def test_starts_with_tx_verb_returns_false_for_empty_string() -> None:
+    assert _starts_with_tx_verb("") is False
+
+
+def test_starts_with_tx_verb_recognises_begin_after_comments() -> None:
+    """Sanity: the strip-then-match path must still recognise a verb
+    when one survives the strip."""
+    assert _starts_with_tx_verb("/* hint */ BEGIN") is True
+
+
+def _make_connection() -> DqliteConnection:
+    """Connection for direct method invocation. We never call
+    ``connect()``, so the underlying socket is never opened — only
+    the predicate's pure-string path is exercised."""
+    conn = DqliteConnection("127.0.0.1:9001")
+    return conn
+
+
+def test_sql_is_outermost_release_or_commit_false_for_pure_dash_dash_comment() -> None:
+    conn = _make_connection()
+    assert conn._sql_is_outermost_release_or_commit("-- comment\n") is False
+
+
+def test_sql_is_outermost_release_or_commit_false_for_pure_block_comment() -> None:
+    conn = _make_connection()
+    assert conn._sql_is_outermost_release_or_commit("/* x */") is False
+
+
+def test_sql_is_outermost_release_or_commit_false_for_whitespace_only() -> None:
+    conn = _make_connection()
+    assert conn._sql_is_outermost_release_or_commit("   ") is False
+
+
+def test_sql_is_outermost_release_or_commit_false_for_empty_string() -> None:
+    conn = _make_connection()
+    assert conn._sql_is_outermost_release_or_commit("") is False