"""One statement per session, and what happens when we forget that.

SQLI gives a session a single statement slot. ``SQ_CLOSE``,
``SQ_RELEASE`` and ``SQ_SFETCH`` all act on whatever statement is
current — none of them names one. Ordinary use never notices, because a
non-scrollable cursor materializes its rows and releases the statement
before returning, so the slot is free again by the time anyone looks.

A scrollable cursor is the exception: it holds the slot open on purpose.
Two things then went wrong, and neither said so.

**Another statement on the same connection.** The server returns ``-285``
for the new statement and *also* destroys the scrollable cursor — its
next fetch comes back ``-267`` "the transaction has been rolled back,
all locks released". Two unattributable errors from code that reads as
completely ordinary: iterate a large result set, run a lookup partway
through. It is now a ``ProgrammingError`` that says what happened.

**The deferred-cleanup queue drained at the wrong moment.** A cursor
finalizer that can't get the wire lock hands its CLOSE/RELEASE to a
queue for the next operation to flush. That queue was flushed before
*every* PDU. But a finalizer enqueues precisely because another thread
holds the lock, i.e. is mid-statement — so the flush landed inside that
thread's own statement and released it. The victim saw ``-208`` when it
happened before the first fetch, ``-267`` between fetch batches. The
flush now happens only at a statement boundary, where the queued CLOSE
addresses the orphan it was meant for.

A stale queue entry was fatal too: the finalizer enqueues, the cursor
then gets closed properly, and the leftover CLOSE draws ``-267`` — an
``OperationalError``, which is in ``WIRE_ERRORS``, which force-closed a
perfectly healthy connection. Server-reported errors are now told apart
from wire failures by the presence of a ``sqlcode``.
"""

from __future__ import annotations

import contextlib

import pytest

import informix_db
from informix_db.cursors import _CLOSE_PDU, _RELEASE_PDU
from tests.conftest import ConnParams


def _connect(conn_params: ConnParams, **kw) -> informix_db.Connection:
    return informix_db.connect(
        host=conn_params.host,
        port=conn_params.port,
        user=conn_params.user,
        password=conn_params.password,
        database=conn_params.database,
        server=conn_params.server,
        connect_timeout=10.0,
        read_timeout=25.0,
        autocommit=True,
        **kw,
    )


# ---------------------------------------------------------------------------
# The conflict check — bookkeeping, no server
# ---------------------------------------------------------------------------


class _FakeScroll:
    def __init__(self, open_: bool = True) -> None:
        self._server_cursor_open = open_


def _conflict_check(conn: informix_db.Connection, requester: object) -> None:
    conn._check_scroll_cursor_conflict(requester)


def test_conflict_check_is_silent_with_no_scroll_cursor() -> None:
    conn = informix_db.Connection.__new__(informix_db.Connection)
    conn._open_scroll_cursor = None
    _conflict_check(conn, object())


def test_conflict_check_forgets_a_collected_cursor() -> None:
    """The ref is weak so an abandoned scrollable cursor can still be
    finalized. A dead ref means the slot is free."""
    import weakref

    conn = informix_db.Connection.__new__(informix_db.Connection)
    victim = _FakeScroll()
    conn._open_scroll_cursor = weakref.ref(victim)
    del victim
    _conflict_check(conn, object())
    assert conn._open_scroll_cursor is None


def test_conflict_check_lets_the_owner_through() -> None:
    """Re-executing the *same* scrollable cursor is allowed — it closes
    its own server-side cursor first."""
    import weakref

    conn = informix_db.Connection.__new__(informix_db.Connection)
    owner = _FakeScroll()
    conn._open_scroll_cursor = weakref.ref(owner)
    _conflict_check(conn, owner)


def test_conflict_check_forgets_a_closed_cursor() -> None:
    import weakref

    conn = informix_db.Connection.__new__(informix_db.Connection)
    done = _FakeScroll(open_=False)
    conn._open_scroll_cursor = weakref.ref(done)
    _conflict_check(conn, object())
    assert conn._open_scroll_cursor is None


# ---------------------------------------------------------------------------
# Against a real server
# ---------------------------------------------------------------------------


@pytest.mark.integration
def test_second_statement_is_refused_while_scroll_cursor_is_open(
    conn_params: ConnParams,
) -> None:
    """Used to be -285 for the new statement plus -267 for the scrollable
    cursor. Now it's one error that names the cause, and the scrollable
    cursor is untouched."""
    with _connect(conn_params) as conn:
        scroll = conn.cursor(scrollable=True)
        scroll.execute("SELECT tabid FROM systables ORDER BY tabid")
        first = scroll.fetch_first()
        assert first is not None

        other = conn.cursor()
        with pytest.raises(informix_db.ProgrammingError, match="scrollable"):
            other.execute("SELECT COUNT(*) FROM systables")

        assert scroll.fetch_absolute(1) is not None, (
            "the refused statement must not have disturbed the cursor"
        )
        scroll.close()
        other.execute("SELECT COUNT(*) FROM systables")
        assert other.fetchone() is not None, "slot must free on close"


@pytest.mark.integration
def test_executemany_is_refused_too(conn_params: ConnParams) -> None:
    with _connect(conn_params) as conn:
        cur = conn.cursor()
        cur.execute("CREATE TEMP TABLE t_slot (k INT)")
        scroll = conn.cursor(scrollable=True)
        scroll.execute("SELECT tabid FROM systables ORDER BY tabid")
        scroll.fetch_first()
        with pytest.raises(informix_db.ProgrammingError, match="scrollable"):
            cur.executemany("INSERT INTO t_slot VALUES (?)", [(1,), (2,)])
        scroll.close()
        cur.executemany("INSERT INTO t_slot VALUES (?)", [(1,), (2,)])
        cur.execute("SELECT COUNT(*) FROM t_slot")
        assert cur.fetchone() == (2,)


@pytest.mark.integration
def test_scroll_cursor_can_be_re_executed(conn_params: ConnParams) -> None:
    """The owner is the one caller allowed past the conflict check, and
    that is only safe because it closes its own server-side cursor
    first. Without that it collides with itself."""
    with _connect(conn_params) as conn:
        scroll = conn.cursor(scrollable=True)
        for _ in range(4):
            scroll.execute("SELECT tabid FROM systables ORDER BY tabid")
            assert scroll.fetch_first() is not None
        scroll.close()


@pytest.mark.integration
def test_abandoning_a_scroll_cursor_frees_the_slot(
    conn_params: ConnParams,
) -> None:
    """Dropping the last reference must let the connection be used again
    — the finalizer closes the cursor and the weak ref goes dead."""
    import gc

    with _connect(conn_params) as conn:
        scroll = conn.cursor(scrollable=True)
        scroll.execute("SELECT tabid FROM systables ORDER BY tabid")
        scroll.fetch_first()
        del scroll
        gc.collect()

        cur = conn.cursor()
        cur.execute("SELECT COUNT(*) FROM systables")
        assert cur.fetchone() is not None


# ---------------------------------------------------------------------------
# Deferred cleanup
# ---------------------------------------------------------------------------


@pytest.mark.integration
def test_queued_cleanup_does_not_land_inside_a_running_statement(
    conn_params: ConnParams,
) -> None:
    """Exactly what a cross-thread finalizer does: it lost the wire lock,
    so it queued its CLOSE/RELEASE while another thread was mid-statement.
    The flush used to happen before that thread's very next PDU — its own
    CURNAME/NFETCH — releasing the statement out from under it."""
    with _connect(conn_params) as conn:
        cur = conn.cursor()
        cur.execute("CREATE TEMP TABLE t_defer (k INT)")
        cur.executemany(
            "INSERT INTO t_defer VALUES (?)", [(i,) for i in range(50)]
        )

        original = cur._read_describe_response

        def enqueue_between_prepare_and_fetch() -> None:
            result = original()
            conn._enqueue_cleanup([_CLOSE_PDU, _RELEASE_PDU])
            return result

        cur._read_describe_response = enqueue_between_prepare_and_fetch
        try:
            cur.execute("SELECT k FROM t_defer ORDER BY k")
            assert len(cur.fetchall()) == 50, (
                "the queued cleanup released our own statement"
            )
        finally:
            cur._read_describe_response = original


@pytest.mark.integration
def test_stale_queued_cleanup_does_not_kill_the_connection(
    conn_params: ConnParams,
) -> None:
    """A queue entry goes stale whenever the cursor gets closed properly
    between enqueue and flush. The server answers the leftover CLOSE with
    -267, which is an OperationalError, which is in WIRE_ERRORS — so a
    stale entry used to force-close a healthy connection."""
    with _connect(conn_params) as conn:
        cur = conn.cursor()
        cur.execute("SELECT FIRST 1 tabid FROM systables")
        cur.fetchall()

        conn._enqueue_cleanup([_CLOSE_PDU, _RELEASE_PDU])
        cur.execute("SELECT FIRST 1 tabid FROM systables")
        assert cur.fetchone() is not None, "stale cleanup killed the connection"
        assert not conn.closed
        assert conn._pending_cleanup == [], "queue should have drained"


@pytest.mark.integration
def test_connection_closes_cleanly_with_a_scroll_cursor_open(
    conn_params: ConnParams,
) -> None:
    conn = _connect(conn_params)
    scroll = conn.cursor(scrollable=True)
    scroll.execute("SELECT tabid FROM systables ORDER BY tabid")
    scroll.fetch_first()
    with contextlib.suppress(Exception):
        conn.close()
    assert conn.closed


# ---------------------------------------------------------------------------
# The wire lock's blind spot
# ---------------------------------------------------------------------------


def test_wire_lock_reports_its_own_thread() -> None:
    """The whole point. ``RLock.acquire(blocking=False)`` returns True for
    the owning thread, so the finalizer's probe could not tell "nobody is
    using the wire" from "I am, right now, mid-statement"."""
    from informix_db.connections import _WireLock

    lock = _WireLock()
    assert not lock.held_by_current_thread
    with lock:
        assert lock.held_by_current_thread
        assert lock.acquire(blocking=False), "must still be reentrant"
        lock.release()
        assert lock.held_by_current_thread, "still held at depth 1"
    assert not lock.held_by_current_thread


def test_wire_lock_is_not_held_by_other_threads() -> None:
    import threading

    from informix_db.connections import _WireLock

    lock = _WireLock()
    seen: list[bool] = []
    entered = threading.Event()
    done = threading.Event()

    def holder() -> None:
        with lock:
            entered.set()
            done.wait(5)

    t = threading.Thread(target=holder)
    t.start()
    entered.wait(5)
    seen.append(lock.held_by_current_thread)
    done.set()
    t.join(5)
    assert seen == [False], "another thread's hold must not read as ours"


@pytest.mark.integration
def test_finalizer_defers_when_gc_runs_on_the_locking_thread(
    conn_params: ConnParams,
) -> None:
    """GC fires on whatever thread allocated. When that is the thread
    holding the wire lock, the finalizer must write nothing — it used to
    acquire the RLock reentrantly and send CLOSE/RELEASE into the running
    statement, killing it with -208."""
    import gc

    with _connect(conn_params) as conn:
        gc.disable()
        try:
            victim = conn.cursor(scrollable=True)
            victim.execute("SELECT tabid FROM systables ORDER BY tabid")
            victim.fetch_first()
            # A reference cycle, so collection waits for gc rather than
            # happening at the drop. Cycles are ordinary in Python.
            cycle = [victim]
            cycle.append(cycle)
            del victim, cycle

            writes: list[bytes] = []
            original_write = conn._sock.write_all
            conn._sock.write_all = lambda b: (
                writes.append(b),
                original_write(b),
            )[1]
            try:
                with conn._wire_lock:  # stand in for "mid-statement"
                    gc.collect()
                    assert writes == [], (
                        "finalizer wrote to the wire while a statement "
                        "owned it"
                    )
                    assert conn._pending_cleanup, (
                        "cleanup should have been deferred, not dropped"
                    )
            finally:
                conn._sock.write_all = original_write
        finally:
            gc.enable()

        cur = conn.cursor()
        cur.execute("SELECT COUNT(*) FROM systables")
        assert cur.fetchone() is not None