"""Two readers, one stream, and a length field taken on trust.

``IfxSocket`` owns a read-ahead buffer that ``BufferedSocketReader``
fills and drains. But ``Connection._drain_to_eot``, ``_raise_sq_err``
and the login path bypass that reader and call ``IfxSocket.read_exact``
directly, which recv'd from the socket without ever looking at the
buffer. Bytes sitting in the buffer would simply be skipped, and skipped
bytes in a length-framed protocol don't announce themselves — the next
read lands mid-field and every read after it is wrong.

Nothing triggers it today. The server sends one response per request, so
recv returns exactly that response and the buffered reader consumes all
of it before control returns to a direct read. That is a property of the
traffic, not of the code, and the buffer is connection-scoped precisely
so read-ahead *can* cross response boundaries — pipelined executemany
already puts several responses in flight. A latent desync waiting on a
timing change is not a good thing to leave in a wire protocol.

Separately, ``fill_recv_buf`` took its byte count on trust, and that
count is almost always a length field straight off the wire. A corrupt
or desynced stream turned into an allocation of whatever the field
happened to say: a garbage ``0x7FFFFFFF`` reads as a 2 GB request, and
the fill loop sits in recv until the read timeout while the buffer
grows. The limit turns that into an error that names the number, which
is the actual diagnostic — a length that absurd means framing was
already lost upstream.
"""

from __future__ import annotations

import pytest

from informix_db._protocol import BufferedSocketReader, ProtocolError
from informix_db._socket import MAX_READ_BYTES, IfxSocket


class _FakeSocket:
    """Stands in for the raw socket. Records what recv actually asked for."""

    def __init__(self, data: bytes = b"") -> None:
        self.data = data
        self.pos = 0
        self.recv_calls: list[int] = []

    def recv(self, n: int) -> bytes:
        self.recv_calls.append(n)
        chunk = self.data[self.pos : self.pos + n]
        self.pos += len(chunk)
        return chunk

    def close(self) -> None:
        # The EOF path force-closes; a stand-in has to survive that.
        pass


def _socket_with(buffered: bytes, on_wire: bytes = b"") -> IfxSocket:
    """An IfxSocket with ``buffered`` already read ahead into _recv_buf."""
    sock = IfxSocket.__new__(IfxSocket)
    sock._sock = _FakeSocket(on_wire)
    sock._recv_buf = bytearray(buffered)
    sock._recv_pos = 0
    sock._recv_size = 65536
    sock._read_timeout = None
    return sock


# ---------------------------------------------------------------------------
# read_exact must not step over the buffer
# ---------------------------------------------------------------------------


def test_read_exact_consumes_the_buffer_first() -> None:
    sock = _socket_with(b"BUFFERED", on_wire=b"SOCKET")
    assert sock.read_exact(8) == b"BUFFERED"
    assert sock._sock.recv_calls == [], "must not touch the socket at all"
    assert sock._recv_pos == 8


def test_read_exact_spans_buffer_then_socket() -> None:
    """The interesting case: a read that starts in the buffer and
    finishes on the wire. Getting this wrong reorders the stream."""
    sock = _socket_with(b"HEAD", on_wire=b"TAIL")
    assert sock.read_exact(8) == b"HEADTAIL"
    assert sock._sock.recv_calls == [4], "only the shortfall comes from recv"


def test_read_exact_respects_a_partly_consumed_buffer() -> None:
    sock = _socket_with(b"XXABCD")
    sock._recv_pos = 2  # first two bytes already decoded
    assert sock.read_exact(4) == b"ABCD"
    assert sock._sock.recv_calls == []


def test_read_exact_of_zero_is_empty() -> None:
    sock = _socket_with(b"DATA")
    assert sock.read_exact(0) == b""
    assert sock.read_exact(-5) == b"", "a negative count must not rewind"
    assert sock._recv_pos == 0


def test_short_read_error_reports_the_original_request() -> None:
    """The message counts bytes; taking some from the buffer must not make
    it lie about how many were asked for."""
    from informix_db.exceptions import OperationalError

    sock = _socket_with(b"AB", on_wire=b"")  # 2 buffered, nothing on the wire
    with pytest.raises(OperationalError, match="wanted 10 bytes"):
        sock.read_exact(10)


def test_buffered_reader_and_direct_read_agree_on_one_stream() -> None:
    """End to end: a BufferedSocketReader over-reads, then a direct
    read_exact picks up exactly where it left off."""
    sock = _socket_with(b"", on_wire=b"\x00\x2aREST-OF-THE-STREAM")
    reader = BufferedSocketReader(sock)
    assert reader.read_short() == 42
    assert len(sock._recv_buf) - sock._recv_pos > 0, (
        "precondition: the reader must have over-read for this to mean "
        "anything"
    )
    assert sock.read_exact(18) == b"REST-OF-THE-STREAM"


# ---------------------------------------------------------------------------
# fill_recv_buf must not believe an arbitrary length
# ---------------------------------------------------------------------------


def test_absurd_length_is_refused_not_allocated() -> None:
    sock = _socket_with(b"", on_wire=b"")
    with pytest.raises(ProtocolError, match="refusing to read"):
        sock.fill_recv_buf(MAX_READ_BYTES + 1)
    assert sock._sock.recv_calls == [], "must refuse before any recv"


def test_refusal_names_the_knob() -> None:
    """The error has to be actionable in both directions: framing is lost,
    or the value genuinely is that big and the limit needs raising."""
    sock = _socket_with(b"", on_wire=b"")
    with pytest.raises(ProtocolError) as exc:
        sock.fill_recv_buf(2**31 - 1)
    message = str(exc.value)
    assert "2147483647" in message
    assert "IFX_MAX_READ_BYTES" in message


def test_a_normal_length_is_unaffected() -> None:
    sock = _socket_with(b"", on_wire=b"x" * 100)
    sock.fill_recv_buf(100)
    assert len(sock._recv_buf) - sock._recv_pos >= 100


# ---------------------------------------------------------------------------
# skip
# ---------------------------------------------------------------------------


def test_buffered_skip_does_not_rewind_on_a_negative_count() -> None:
    """The base reader's skip delegates to read_exact, which guards. This
    one advances the cursor arithmetically, so an unguarded negative count
    re-decodes bytes already consumed as if they were the next field."""
    sock = _socket_with(b"ABCDEFGH")
    sock._recv_pos = 4
    BufferedSocketReader(sock).skip(-4)
    assert sock._recv_pos == 4, "skip must never move the cursor backwards"


def test_buffered_skip_advances_normally() -> None:
    sock = _socket_with(b"ABCDEFGH")
    reader = BufferedSocketReader(sock)
    reader.skip(4)
    assert reader.read_exact(4) == b"EFGH"