# SPDX-FileCopyrightText: 2025-present The Firebird Projects <www.firebirdsql.org>

#

# SPDX-License-Identifier: MIT

#

# PROGRAM/MODULE: firebird-driver

# FILE: tests/test_transaction.py

# DESCRIPTION: Tests for Transaction

# CREATED: 10.4.2025

#

# Software distributed under the License is distributed AS IS,

# WITHOUT WARRANTY OF ANY KIND, either express or implied.

# See the License for the specific language governing rights

# and limitations under the License.

#

# The Original Code was created by Pavel Cisar

#

# Copyright (c) Pavel Cisar <pcisar@users.sourceforge.net>

# and all contributors signed below.

#

# All Rights Reserved.

# Contributor(s): ______________________________________.

#

# See LICENSE.TXT for details.

import pytest

from packaging.specifiers import SpecifierSet

from firebird.driver import (Isolation, connect, tpb, TransactionManager,

transaction, InterfaceError, TPB, TableShareMode,

TableAccessMode, TraInfoCode, TraInfoAccess, TraAccessMode,

DefaultAction)

def test_cursor(db_connection):

with db_connection: # Use connection context manager

tr = db_connection.main_transaction

tr.begin()

with tr.cursor() as cur:

cur.execute("insert into t (c1) values (1)")

tr.commit()

cur.execute("select * from t")

rows = cur.fetchall()

assert rows == [(1,)]

cur.execute("delete from t")

tr.commit()

assert len(tr.cursors) == 1

assert tr.cursors[0] is cur # This checks weakref behavior, might need adjustment

def test_context_manager(db_connection):

with db_connection.cursor() as cur:

with transaction(db_connection):

cur.execute("insert into t (c1) values (1)")

cur.execute("select * from t")

rows = cur.fetchall()

assert rows == [(1,)]

with pytest.raises(Exception): # Use pytest.raises

with transaction(db_connection):

cur.execute("delete from t")

raise Exception("Simulating error")

cur.execute("select * from t")

rows = cur.fetchall()

assert rows == [(1,)] # Should still be 1 due to rollback

with transaction(db_connection):

cur.execute("delete from t")

cur.execute("select * from t")

rows = cur.fetchall()

assert rows == []

def test_savepoint(db_connection):

db_connection.begin()

tr = db_connection.main_transaction

db_connection.execute_immediate("insert into t (c1) values (1)")

tr.savepoint('test')

db_connection.execute_immediate("insert into t (c1) values (2)")

tr.rollback(savepoint='test')

tr.commit()

with tr.cursor() as cur:

cur.execute("select * from t")

rows = cur.fetchall()

assert rows == [(1,)]

def test_fetch_after_commit(db_connection):

db_connection.execute_immediate("insert into t (c1) values (1)")

db_connection.commit()

with db_connection.cursor() as cur:

cur.execute("select * from t")

db_connection.commit()

with pytest.raises(InterfaceError, match='Cannot fetch from cursor that did not executed a statement.'):

cur.fetchall()

def test_fetch_after_rollback(db_connection):

db_connection.execute_immediate("insert into t (c1) values (1)")

db_connection.rollback()

with db_connection.cursor() as cur:

cur.execute("select * from t")

# Rollback implicitly happens if not committed when transaction ends

# Or explicitly:

db_connection.rollback()

with pytest.raises(InterfaceError, match='Cannot fetch from cursor that did not executed a statement.'):

cur.fetchall()

def test_tpb(db_connection):

tpb_obj = TPB(isolation=Isolation.READ_COMMITTED, no_auto_undo=True)

tpb_obj.lock_timeout = 10

tpb_obj.reserve_table('COUNTRY', TableShareMode.PROTECTED, TableAccessMode.LOCK_WRITE)

tpb_buffer = tpb_obj.get_buffer()

with db_connection.transaction_manager(tpb_buffer) as tr:

info = tr.info.get_info(TraInfoCode.ISOLATION)

# Version check might be needed here as before

engine_version = db_connection.info.engine_version

if engine_version >= 4.0:

assert info in [Isolation.READ_COMMITTED_READ_CONSISTENCY,

Isolation.READ_COMMITTED_RECORD_VERSION]

else:

assert info == Isolation.READ_COMMITTED_RECORD_VERSION

assert tr.info.get_info(TraInfoCode.ACCESS) == TraInfoAccess.READ_WRITE

assert tr.info.lock_timeout == 10

del tpb_obj

tpb_parsed = TPB()

tpb_parsed.parse_buffer(tpb_buffer)

assert tpb_parsed.access_mode == TraAccessMode.WRITE

assert tpb_parsed.isolation == Isolation.READ_COMMITTED_RECORD_VERSION

assert tpb_parsed.lock_timeout == 10

assert not tpb_parsed.auto_commit

assert tpb_parsed.no_auto_undo

assert not tpb_parsed.ignore_limbo

assert tpb_parsed._table_reservation == [('COUNTRY',

TableShareMode.PROTECTED,

TableAccessMode.LOCK_WRITE)]

def test_transaction_info(db_connection, db_file):

with db_connection.main_transaction as tr:

assert tr.is_active()

assert str(db_file) in tr.info.database # Check fixture use

assert tr.info.isolation == Isolation.SNAPSHOT

assert tr.info.id > 0

assert tr.info.oit > 0

assert tr.info.oat > 0

assert tr.info.ost > 0

assert tr.info.lock_timeout == -1

assert tr.info.isolation == Isolation.SNAPSHOT

def test_default_action_rollback(db_connection):

"""Verify TransactionManager closes with rollback if default_action is ROLLBACK."""

# Ensure table is empty first

with db_connection.cursor() as cur_clean:

cur_clean.execute("DELETE FROM t")

db_connection.commit()

tr_rollback = None # Define outside 'with' to check is_closed later

try:

# Create manager with ROLLBACK default

tr_rollback = db_connection.transaction_manager(default_action=DefaultAction.ROLLBACK)

# Use context manager for the TransactionManager itself

with tr_rollback:

tr_rollback.begin() # Start the transaction

with tr_rollback.cursor() as cur:

cur.execute("insert into t (c1) values (99)")

# Do not explicitly commit or rollback, let the 'with tr_rollback:' handle it

assert tr_rollback.is_active()

# Check transaction is no longer active and manager is closed

assert not tr_rollback.is_active()

assert tr_rollback.is_closed()

# Verify data was rolled back using a separate transaction

with db_connection.cursor() as cur_verify:

cur_verify.execute("select * from t where c1 = 99")

rows = cur_verify.fetchall()

assert rows == []

finally:

# Ensure cleanup even if assertions fail

if tr_rollback and not tr_rollback.is_closed():

tr_rollback.close()

# Clean up table again

with db_connection.cursor() as cur_clean:

cur_clean.execute("DELETE FROM t")

db_connection.commit()

def test_connection_close_with_active_transaction(dsn, db_connection):

"""Verify transaction behavior when connection is closed while active."""

# Ensure table is empty first

with db_connection.cursor() as cur_clean:

cur_clean.execute("DELETE FROM t")

db_connection.commit()

tr = db_connection.transaction_manager()

tr.begin()

with tr.cursor() as cur:

cur.execute("insert into t (c1) values (88)")

# Don't commit or rollback yet

# Close the connection while transaction is active

db_connection.close()

# Assertions on the transaction manager state

assert tr.is_closed(), "Transaction manager should be closed after connection close"

assert not tr.is_active(), "Transaction should not be active after connection close"

# Reconnect and verify data was rolled back

with connect(dsn) as new_con:

with new_con.cursor() as cur_verify:

cur_verify.execute("select * from t where c1 = 88")

rows = cur_verify.fetchall()

assert rows == [], "Data inserted before connection close should be rolled back"

def test_complex_savepoints(db_connection):

"""Test rolling back past multiple savepoints."""

# Ensure table is empty first

with db_connection.cursor() as cur_clean:

cur_clean.execute("DELETE FROM t")

db_connection.commit()

# Scenario 1: Rollback past multiple savepoints

db_connection.begin()

db_connection.savepoint('SP1')

db_connection.execute_immediate("insert into t (c1) values (1)")

db_connection.savepoint('SP2')

db_connection.execute_immediate("insert into t (c1) values (2)")

db_connection.savepoint('SP3')

db_connection.execute_immediate("insert into t (c1) values (3)")

# Rollback to the first savepoint

db_connection.rollback(savepoint='SP2')

# Commit the remaining transaction state (only includes insert 1)

db_connection.commit()

# Verify state

with db_connection.cursor() as cur:

cur.execute("select * from t order by c1")

rows = cur.fetchall()

assert rows == [(1,)], "Should only contain data before SP2"

# Scenario 2: Intermediate rollbacks

with db_connection.cursor() as cur_clean: # Reuse cursor

cur_clean.execute("DELETE FROM t")

db_connection.commit()

db_connection.begin()

db_connection.savepoint('SP_A')

db_connection.execute_immediate("insert into t (c1) values (10)")

db_connection.savepoint('SP_B')

db_connection.execute_immediate("insert into t (c1) values (20)")

# Rollback to SP_B (should effectively do nothing visible yet)

db_connection.rollback(savepoint='SP_B')

# Insert another value after rolling back to SP_B

db_connection.execute_immediate("insert into t (c1) values (30)")

db_connection.savepoint('SP_C')

db_connection.execute_immediate("insert into t (c1) values (40)")

# Rollback to SP_A

db_connection.rollback(savepoint='SP_A')

# Commit remaining transaction (should only contain insert 10)

db_connection.commit()

# Verify state

with db_connection.cursor() as cur:

cur.execute("select * from t order by c1")

rows = cur.fetchall()

assert rows == [], "Should only contain data before SP_A"

def test_tpb_at_snapshot_number(fb_vars, db_connection):

"""Test starting a transaction at a specific snapshot number (FB4+)."""

if fb_vars['version'] not in SpecifierSet('>=4.0'):

pytest.skip("Requires Firebird 4.0+ for AT SNAPSHOT NUMBER")

# Ensure table is empty first

with db_connection.cursor() as cur_clean:

cur_clean.execute("DELETE FROM t")

db_connection.commit()

# 0. Start TR0 (normal), insert different data, commit TR0

# This changes the *current* state of the database

with db_connection.cursor() as cur2:

cur2.execute("insert into t (c1) values (1)")

db_connection.commit() # Commit TR2

# 1. Start TR1, insert data, get snapshot number

tr1: TransactionManager = db_connection.transaction_manager()

tr1.begin(tpb(Isolation.SNAPSHOT)) # TR1

snapshot_no = tr1.info.snapshot_number

assert snapshot_no > 0

#db_connection.commit() # Commit TR1

# 2. Start TR2 (normal), insert different data, commit TR2

# This changes the *current* state of the database

with db_connection.cursor() as cur2:

cur2.execute("insert into t (c1) values (2)")

db_connection.commit() # Commit TR2

# 3. Start TR3 using the snapshot number from TR1

tr_snap: TransactionManager = None

try:

tr_snap = db_connection.transaction_manager()

# Create TPB with the specific snapshot number

tpb_snap = TPB(isolation=Isolation.SNAPSHOT, at_snapshot_number=snapshot_no)

tr_snap.begin(tpb=tpb_snap.get_buffer())

# 4. Select data within TR3 - should only see data from TR1's snapshot

with tr_snap.cursor() as cur_snap:

cur_snap.execute("select * from t order by c1")

rows = cur_snap.fetchall()

assert rows == [(1,)], "Transaction at snapshot should only see data from that snapshot"

tr_snap.commit() # Commit/Rollback TR3

finally:

if tr_snap and not tr_snap.is_closed():

tr_snap.close()

# Clean up table again

with db_connection.cursor() as cur_clean:

cur_clean.execute("DELETE FROM t")

db_connection.commit()