"""Transaction control run as SQL, and the flag that didn't notice.

``Connection._in_transaction`` decides whether ``commit()`` and
``rollback()`` send anything at all, and the pool reads it to decide
whether a returned connection needs cleaning up. It was maintained
solely by the driver's own implicit ``SQ_BEGIN``, so a caller who wrote
``cursor.execute("BEGIN WORK")`` — an entirely reasonable thing to
write — walked straight past it.

With autocommit on, nothing stopped that statement reaching the server.
A transaction opened, the flag stayed False, and ``rollback()`` returned
successfully having sent nothing. The rows it was asked to discard were
still there. The connection then went back to the pool holding an open
transaction and its locks, because the pool's cleanup is guarded by the
same flag.

With autocommit off it failed instead, and for a sillier reason: the
driver's implicit ``SQ_BEGIN`` fired first, so the caller's ``BEGIN
WORK`` got ``-535``, "already in transaction". The driver and the user
competing to open the same transaction, and the user losing.

The server labels these statements: type 34 for BEGIN, 35 for COMMIT, 36
for ROLLBACK, with and without the ``WORK`` keyword. JDBC reads the same
three values off the describe and calls ``setTxBeginState`` /
``setTxEndState``. It also calls ``initiateTransaction`` *after* the
describe rather than before, which is what makes the skip possible —
until the describe lands you can't know the statement is transaction
control.
"""

from __future__ import annotations

import contextlib

import pytest

import informix_db
from informix_db.cursors import _TX_CONTROL_TYPES
from tests.conftest import ConnParams

pytestmark = pytest.mark.integration


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,
        **kw,
    )


def test_transaction_control_types_are_what_the_server_says(
    logged_db_params: ConnParams,
) -> None:
    """Pin the three constants against a live server rather than trusting
    the decompiled source. Both spellings must map to the same type."""
    with _connect(logged_db_params, autocommit=True) as conn:
        cur = conn.cursor()
        seen = {}
        for sql in ("BEGIN WORK", "COMMIT WORK", "ROLLBACK WORK",
                    "BEGIN", "COMMIT", "ROLLBACK"):
            with conn._wire_lock:
                conn._send_pdu(
                    cur._build_prepare_pdu(sql, num_qmarks=0),
                    statement_boundary=True,
                )
                cur._read_describe_response()
                cur._release_after_failure()
            seen[sql] = cur._statement_type
        assert seen["BEGIN WORK"] == seen["BEGIN"] == 34
        assert seen["COMMIT WORK"] == seen["COMMIT"] == 35
        assert seen["ROLLBACK WORK"] == seen["ROLLBACK"] == 36
        assert set(seen.values()) == _TX_CONTROL_TYPES


def test_rollback_after_sql_begin_actually_rolls_back(
    logged_db_params: ConnParams,
) -> None:
    """The data-loss case. rollback() reported success and sent nothing,
    so the row it was asked to discard survived."""
    with _connect(logged_db_params, autocommit=True) as conn:
        cur = conn.cursor()
        with contextlib.suppress(Exception):
            cur.execute("DROP TABLE t_txstate")
        cur.execute("CREATE TABLE t_txstate (k INT)")
        try:
            cur.execute("BEGIN WORK")
            assert conn._in_transaction, "SQL BEGIN must set the flag"
            cur.execute("INSERT INTO t_txstate VALUES (1)")
            conn.rollback()
            cur.execute("SELECT COUNT(*) FROM t_txstate")
            assert cur.fetchone() == (0,), "rollback() was a silent no-op"
            assert not conn._in_transaction
        finally:
            with contextlib.suppress(Exception):
                cur.execute("DROP TABLE t_txstate")


def test_sql_commit_clears_the_flag(logged_db_params: ConnParams) -> None:
    """The mirror image: with the flag stuck True after a SQL COMMIT, the
    next rollback() would send SQ_RBWORK with no transaction open and
    draw -255."""
    with _connect(logged_db_params, autocommit=True) as conn:
        cur = conn.cursor()
        with contextlib.suppress(Exception):
            cur.execute("DROP TABLE t_txstate2")
        cur.execute("CREATE TABLE t_txstate2 (k INT)")
        try:
            cur.execute("BEGIN WORK")
            cur.execute("INSERT INTO t_txstate2 VALUES (1)")
            cur.execute("COMMIT WORK")
            assert not conn._in_transaction, "SQL COMMIT must clear the flag"
            conn.rollback()  # must be a no-op, not a -255
            cur.execute("SELECT COUNT(*) FROM t_txstate2")
            assert cur.fetchone() == (1,), "the committed row must survive"
        finally:
            with contextlib.suppress(Exception):
                cur.execute("DROP TABLE t_txstate2")


def test_sql_begin_does_not_collide_with_the_implicit_one(
    logged_db_params: ConnParams,
) -> None:
    """Non-autocommit. The driver's implicit SQ_BEGIN used to fire first
    and the caller's BEGIN WORK then got -535."""
    with _connect(logged_db_params, autocommit=False) as conn:
        cur = conn.cursor()
        with contextlib.suppress(Exception):
            cur.execute("DROP TABLE t_txstate3")
            conn.commit()
        cur.execute("CREATE TABLE t_txstate3 (k INT)")
        conn.commit()
        try:
            cur.execute("BEGIN WORK")
            cur.execute("INSERT INTO t_txstate3 VALUES (1)")
            conn.rollback()
            cur.execute("SELECT COUNT(*) FROM t_txstate3")
            assert cur.fetchone() == (0,)
            conn.commit()
        finally:
            with contextlib.suppress(Exception):
                cur.execute("DROP TABLE t_txstate3")
                conn.commit()


def test_ordinary_dml_still_opens_a_transaction(
    logged_db_params: ConnParams,
) -> None:
    """_ensure_transaction moved from before the PREPARE to after the
    describe. It still has to fire for everything that isn't transaction
    control, or non-autocommit DML runs outside a transaction."""
    with _connect(logged_db_params, autocommit=False) as conn:
        cur = conn.cursor()
        with contextlib.suppress(Exception):
            cur.execute("DROP TABLE t_txstate4")
            conn.commit()
        cur.execute("CREATE TABLE t_txstate4 (k INT)")
        conn.commit()
        try:
            assert not conn._in_transaction
            cur.execute("INSERT INTO t_txstate4 VALUES (1)")
            assert conn._in_transaction, "DML must open a transaction"
            conn.rollback()
            cur.execute("SELECT COUNT(*) FROM t_txstate4")
            assert cur.fetchone() == (0,)
            conn.commit()
        finally:
            with contextlib.suppress(Exception):
                cur.execute("DROP TABLE t_txstate4")
                conn.commit()


def test_a_failed_transaction_statement_does_not_move_the_flag(
    logged_db_params: ConnParams,
) -> None:
    """The sync runs on the success path only. A COMMIT that the server
    rejects has not ended anything."""
    with _connect(logged_db_params, autocommit=True) as conn:
        cur = conn.cursor()
        assert not conn._in_transaction
        with pytest.raises(informix_db.Error):
            cur.execute("COMMIT WORK")  # -255, nothing to commit
        assert not conn._in_transaction
        cur.execute("SELECT FIRST 1 tabid FROM systables")
        assert cur.fetchone() is not None