diff --git a/src/mountainash_data/backends/ibis/_transaction.py b/src/mountainash_data/backends/ibis/_transaction.py index fa48544..ac9f708 100644 --- a/src/mountainash_data/backends/ibis/_transaction.py +++ b/src/mountainash_data/backends/ibis/_transaction.py @@ -132,3 +132,16 @@ def _exec(sql: str) -> None: finally: with _LOCK: state.depth -= 1 + + +def is_active(raw_handle: t.Any) -> bool: + """True if a unit of work is registered on this raw handle (any depth). + + Read-only companion to run_transaction(): reads the ambient registry under + the same lock the register/unregister critical sections take, so the read + observes a fully-committed registry mutation. Never mutates _ACTIVE. Keyed + on id(raw_handle), matching run_transaction's reentrancy key, so distinct + IbisBackend wrappers of one raw connection agree. + """ + with _LOCK: + return id(raw_handle) in _ACTIVE diff --git a/src/mountainash_data/backends/ibis/backend.py b/src/mountainash_data/backends/ibis/backend.py index 4ec51e7..f936966 100644 --- a/src/mountainash_data/backends/ibis/backend.py +++ b/src/mountainash_data/backends/ibis/backend.py @@ -11,7 +11,7 @@ import typing as t from mountainash_data.backends.ibis.dialects._registry import DIALECTS, DialectSpec, TransactionSupport -from mountainash_data.backends.ibis._transaction import run_transaction +from mountainash_data.backends.ibis._transaction import run_transaction, is_active from mountainash_data.backends.ibis._adoption import ( apply_options, snapshot_options, restore_options, ) @@ -567,6 +567,30 @@ def transaction(self, *, required: bool = True) -> t.ContextManager[None]: raw_execute_hook=self._spec.raw_execute_hook, ) + def in_transaction(self) -> bool: + """True if a unit of work opened via transaction() is currently active + on this backend's raw connection (any nesting depth). + + Runtime companion to the static supports_transactions flag. Total: + returns False — never raises — for NONE dialects, a backend that was + never connected or has been closed, and a connection whose native + handle has gone away. A point-in-time snapshot, not a lock. + """ + if self._spec.transaction_support is TransactionSupport.NONE: + return False + if self._conn is None: + return False + try: + raw = self.raw_driver_connection() + except Exception: + # Unresolvable/absent native handle == no live unit of work. + # raw_driver_connection() raises RuntimeError on an absent handle + # attr; a property-backed handle could raise a driver-specific + # error on a dropped connection. Either way the answer is "no + # active tx"; a genuine fault surfaces on the caller's next op. + return False + return is_active(raw) + # --- Inspection (terminal — delegates to IbisConnection) --- def list_tables(self, namespace: NamespaceLike = None) -> list[str]: diff --git a/src/mountainash_data/backends/iceberg/backend.py b/src/mountainash_data/backends/iceberg/backend.py index 2504363..5f57a69 100644 --- a/src/mountainash_data/backends/iceberg/backend.py +++ b/src/mountainash_data/backends/iceberg/backend.py @@ -103,3 +103,9 @@ def transaction(self, *, required: bool = True): ) warn_once("iceberg", "iceberg has no transaction support; transaction() is a no-op.") yield + + def in_transaction(self) -> bool: + """Iceberg has no connection-level unit of work (supports_transactions + is False); nothing can be active. pyiceberg table-scoped transactions + are a separate future capability.""" + return False diff --git a/src/mountainash_data/core/protocol.py b/src/mountainash_data/core/protocol.py index 085f8a9..757abb5 100644 --- a/src/mountainash_data/core/protocol.py +++ b/src/mountainash_data/core/protocol.py @@ -70,3 +70,19 @@ def transaction(self, *, required: bool = True) -> t.ContextManager[None]: (an adopted autocommit-off connection is refused with TransactionIntegrityError). See spec §5.1–5.3.""" ... + + def in_transaction(self) -> bool: + """True if a unit of work opened via transaction() is currently active + on this backend's connection (any nesting depth). Runtime companion to + the static supports_transactions flag. + + Total: returns False — never raises — for backends with no transaction + concept, a backend that was never connected or has been closed, and a + connection whose native handle has gone away. This is a point-in-time + snapshot, NOT a lock: an answer may be stale the instant it returns, so + it must not be used as a correctness-critical mutual-exclusion gate + without external coordination. Intended use is an ownership guard + ("refuse to run nested inside an ambient unit of work"). Note: True + means a unit of work is open, not that a fresh transaction() join would + succeed — a poisoned unit is still reported active until it unwinds.""" + ... diff --git a/tests/test_integration/test_transaction_live.py b/tests/test_integration/test_transaction_live.py index f435087..bf2b8b5 100644 --- a/tests/test_integration/test_transaction_live.py +++ b/tests/test_integration/test_transaction_live.py @@ -1,29 +1,21 @@ -import os import pytest -from mountainash_data import IbisBackend pytestmark = pytest.mark.integration -REQUIRE = os.environ.get("MOUNTAINASH_REQUIRE_LIVE_DB") == "1" -PG_URL = os.environ.get("MOUNTAINASH_TEST_POSTGRES_URL") - -def _skip_or_fail(reason): - if REQUIRE: - pytest.fail(reason) - pytest.skip(reason) - - -def test_postgres_transaction_rollback(): - if not PG_URL: - _skip_or_fail("MOUNTAINASH_TEST_POSTGRES_URL not set") - with IbisBackend(PG_URL) as be: - raw = be.raw_driver_connection() - cur = raw.cursor() - cur.execute("CREATE TEMP TABLE t_tx (x INT)") - with pytest.raises(ValueError): - with be.transaction(): - cur.execute("INSERT INTO t_tx VALUES (1)") - raise ValueError("boom") - cur.execute("SELECT count(*) FROM t_tx") - assert cur.fetchone()[0] == 0 +def test_postgres_transaction_rollback(postgres_backend): + """A transaction() that raises must roll the whole unit of work back on a + real postgres connection. Uses the shared postgres_backend fixture (reads + IBIS_TEST_POSTGRES_*, connects to the live service, and honours + MOUNTAINASH_REQUIRE_LIVE_DB=1 by fail-closing only when the service is + genuinely unreachable) — the same convention as every other live test.""" + be = postgres_backend + raw = be.raw_driver_connection() + cur = raw.cursor() + cur.execute("CREATE TEMP TABLE t_tx (x INT)") + with pytest.raises(ValueError): + with be.transaction(): + cur.execute("INSERT INTO t_tx VALUES (1)") + raise ValueError("boom") + cur.execute("SELECT count(*) FROM t_tx") + assert cur.fetchone()[0] == 0 diff --git a/tests/test_unit/backends/ibis/test_backend.py b/tests/test_unit/backends/ibis/test_backend.py index e36f0d5..add63f5 100644 --- a/tests/test_unit/backends/ibis/test_backend.py +++ b/tests/test_unit/backends/ibis/test_backend.py @@ -572,3 +572,84 @@ def test_transaction_nested_joins(tmp_path): with be.transaction(): raw.execute("INSERT INTO t VALUES (1)") assert raw.execute("SELECT count(*) FROM t").fetchone()[0] == 1 + + +def test_in_transaction_false_outside_and_true_inside(): + with IbisBackend(dialect="duckdb", database=":memory:") as be: + assert be.in_transaction() is False + with be.transaction(): + assert be.in_transaction() is True + assert be.in_transaction() is False + + +def test_in_transaction_true_at_nested_depth(): + with IbisBackend(dialect="duckdb", database=":memory:") as be: + with be.transaction(): + with be.transaction(): + assert be.in_transaction() is True + assert be.in_transaction() is True + assert be.in_transaction() is False + + +def test_in_transaction_false_after_rollback(): + with IbisBackend(dialect="duckdb", database=":memory:") as be: + with pytest.raises(ValueError): + with be.transaction(): + raise ValueError("boom") + assert be.in_transaction() is False + + +def test_in_transaction_true_while_poisoned_before_unwind(): + # Public-surface pin of spec §5.4: poisoned-but-open reads True. + from mountainash_data.core.errors import TransactionPoisonedError + with IbisBackend(dialect="duckdb", database=":memory:") as be: + with pytest.raises(TransactionPoisonedError): + with be.transaction(): + try: + with be.transaction(): + raise ValueError("inner") + except ValueError: + pass + assert be.in_transaction() is True # poisoned, still open + assert be.in_transaction() is False + + +def test_in_transaction_none_dialect_returns_false(): + be = IbisBackend(dialect="clickhouse") # TransactionSupport.NONE, not connected + assert be.in_transaction() is False + + +def test_in_transaction_not_connected_returns_false(): + be = IbisBackend(dialect="duckdb", database=":memory:") # never connect()ed + assert be.in_transaction() is False + + +def test_in_transaction_after_close_returns_false(): + be = IbisBackend(dialect="duckdb", database=":memory:") + be.connect() + be.close() + assert be.in_transaction() is False + + +def test_in_transaction_returns_false_when_handle_resolution_raises(monkeypatch): + # A property-backed raw_handle_attr could raise a driver-specific (non- + # RuntimeError) error on a dropped connection. The predicate must swallow + # it and answer False, never propagate. monkeypatch replaces the bound + # method with a zero-arg callable; in_transaction() calls it with no args. + be = IbisBackend(dialect="duckdb", database=":memory:") + be.connect() + + def _boom(): + raise OSError("driver connection dropped") + + monkeypatch.setattr(be, "raw_driver_connection", _boom) + assert be.in_transaction() is False + + +@pytest.mark.parametrize("dialect", ["duckdb", "sqlite"]) +def test_raw_driver_connection_identity_is_stable(dialect): + # Load-bearing for cross-wrapper correctness (spec §3.4/§5.6): the handle + # must be the SAME object on every call so two wrappers over one raw conn + # compute one id. Pinned for every locally testable dialect. + with IbisBackend(dialect=dialect, database=":memory:") as be: + assert be.raw_driver_connection() is be.raw_driver_connection() diff --git a/tests/test_unit/backends/ibis/test_backend_adopt.py b/tests/test_unit/backends/ibis/test_backend_adopt.py index b5352ab..bc28261 100644 --- a/tests/test_unit/backends/ibis/test_backend_adopt.py +++ b/tests/test_unit/backends/ibis/test_backend_adopt.py @@ -166,3 +166,29 @@ def test_from_ibis_connection_default_unchanged(): # no apply -> ibis default stands assert raw.execute("SELECT current_setting('python_enable_replacements')").fetchone()[0] is False be.close() + + +def test_in_transaction_visible_across_wrappers_from_ibis_connection(raw_db): + # Two IbisBackends adopting ONE raw connection must agree on tx state — + # the pointbreak F-09 scenario (guard runs on a different wrapper). + ibis_conn = ibis.duckdb.from_connection(raw_db) + a = IbisBackend.from_ibis_connection(ibis_conn, dialect="duckdb") + b = IbisBackend.from_ibis_connection(ibis_conn, dialect="duckdb") + assert a.in_transaction() is False + assert b.in_transaction() is False + with a.transaction(): + assert b.in_transaction() is True # B sees A's open unit of work + assert a.in_transaction() is False + assert b.in_transaction() is False + + +def test_in_transaction_visible_across_wrappers_from_raw_connection(raw_db): + # from_raw_connection resolves the handle differently; must still agree. + # (duckdb has raw_adoption_verified=True.) ibis stores the passed raw conn + # directly as .con, so both wrappers key on id(raw_db). + a = IbisBackend.from_raw_connection(raw_db, dialect="duckdb") + b = IbisBackend.from_raw_connection(raw_db, dialect="duckdb") + with a.transaction(): + assert a.in_transaction() is True + assert b.in_transaction() is True + assert b.in_transaction() is False diff --git a/tests/test_unit/backends/ibis/test_transaction.py b/tests/test_unit/backends/ibis/test_transaction.py index 7233ece..fe3469f 100644 --- a/tests/test_unit/backends/ibis/test_transaction.py +++ b/tests/test_unit/backends/ibis/test_transaction.py @@ -142,3 +142,58 @@ def cursor(self): return Cursor(self.log) pass assert ("cur", "BEGIN") in h.log and ("cur", "COMMIT") in h.log assert ("close", None) in h.log + + +from mountainash_data.backends.ibis._transaction import is_active + + +def test_is_active_false_when_not_registered(): + h = FakeHandle() + assert is_active(h) is False + + +def test_is_active_true_inside_outer_transaction(): + h = FakeHandle() + with _tx(h): + assert is_active(h) is True + assert is_active(h) is False + + +def test_is_active_true_at_nested_depth(): + h = FakeHandle() + with _tx(h): + with _tx(h): + assert is_active(h) is True + assert is_active(h) is True # still open at depth 1 + assert is_active(h) is False + + +def test_is_active_false_after_rollback(): + h = FakeHandle() + with pytest.raises(ValueError): + with _tx(h): + assert is_active(h) is True + raise ValueError("boom") + assert is_active(h) is False + + +def test_is_active_true_while_poisoned_before_unwind(): + # A caught nested failure poisons the unit; while the outer block is still + # open the handle stays registered, so is_active must report True. + h = FakeHandle() + with pytest.raises(TransactionPoisonedError): + with _tx(h): + try: + with _tx(h): + raise ValueError("inner") + except ValueError: + pass + assert is_active(h) is True # poisoned but still an open unit of work + assert is_active(h) is False # cleared after outer unwind + + +def test_is_active_does_not_mutate_registry(): + h = FakeHandle() + before = dict(_ACTIVE) + assert is_active(h) is False + assert _ACTIVE == before # pure read diff --git a/tests/test_unit/backends/iceberg/test_backend.py b/tests/test_unit/backends/iceberg/test_backend.py index 9ef7a17..01822c3 100644 --- a/tests/test_unit/backends/iceberg/test_backend.py +++ b/tests/test_unit/backends/iceberg/test_backend.py @@ -67,3 +67,17 @@ def test_transaction_not_required_noops(): with be.transaction(required=False): pass assert any("iceberg" in str(x.message).lower() for x in w) + + +def test_in_transaction_always_false(): + be = IcebergBackend(catalog="rest", uri="http://localhost:8181") + # No connection-level unit of work — consistent with supports_transactions. + assert be.in_transaction() is False + assert be.supports_transactions is False + + +def test_iceberg_satisfies_protocol_including_in_transaction(): + from mountainash_data.core.protocol import Backend + be = IcebergBackend(catalog="rest", uri="http://localhost:8181") + assert isinstance(be, Backend) + assert callable(be.in_transaction) diff --git a/tests/test_unit/core/test_protocol.py b/tests/test_unit/core/test_protocol.py index 0848790..5893dc4 100644 --- a/tests/test_unit/core/test_protocol.py +++ b/tests/test_unit/core/test_protocol.py @@ -78,3 +78,14 @@ def test_iceberg_backend_satisfies_widened_protocol(): def test_protocol_declares_transaction(): from mountainash_data.core.protocol import Backend assert hasattr(Backend, "transaction") + + +def test_backend_protocol_declares_in_transaction(): + assert hasattr(Backend, "in_transaction") + + +def test_ibis_backend_satisfies_protocol_including_in_transaction(): + from mountainash_data.backends.ibis.backend import IbisBackend + be = IbisBackend(dialect="duckdb", database=":memory:") + assert isinstance(be, Backend) # runtime_checkable: presence of protocol methods + assert callable(be.in_transaction)