Drop implementation-pinning tests and collapse redundant matrices

Remove tests that asserted on source shape (inspect.getsource / ast /
__doc__) rather than behavior — each either pinned an implementation
detail with no observable contract or duplicated an existing behavioral
test. Collapse enum-constant and flag matrices that expanded one
property into many near-identical cases into single parametrized loops.
Structure pins guarding genuine invariants with no behavioral equivalent
(cancel-path ordering, KeyboardInterrupt/SystemExit propagation,
cross-package provenance) are kept. No behavioral coverage is removed.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: antoineleclair

tests/test_aio_connection_finalizer_detached_on_cancel_close.py

@@ -53,18 +53,3 @@ def test_async_execute_does_not_wrap_retry_call_in_outer_op_lock() -> None:
         "acquire ``op_lock`` between attempts instead of parking "
         "for the full backoff curve."
     )
-
-
-def test_async_execute_calls_retry_with_a_factory_that_acquires_op_lock() -> None:
-    """The retry helper must be passed a per-attempt callable that owns op_lock."""
-    src = _async_execute_source()
-    assert "async def _attempt" in src or "_attempt = " in src, (
-        "AsyncCursor.execute should define a per-attempt coroutine "
-        "(typically named ``_attempt``) that owns the ``op_lock`` "
-        "acquire — pass that callable to ``retry_async_on_busy`` so "
-        "each retry acquires + releases the lock separately."
-    )
-    assert "async with op_lock" in src, (
-        "the per-attempt coroutine must include ``async with op_lock:`` "
-        "so each retry attempt re-acquires the lock"
-    )

tests/test_async_execute_busy_retry_releases_op_lock_between_attempts.py

@@ -7,33 +7,12 @@
 
 import asyncio
 import contextlib
-import inspect
-
-from dqlitedbapi.aio.cursor import AsyncCursor
-
-
-def test_async_fetch_methods_call_non_binding_helper_not_ensure_locks() -> None:
-    """Each fetch* method must call ``_check_loop_binding``, not ``_ensure_locks``."""
-    for method_name in ("fetchone", "fetchmany", "fetchall"):
-        method = getattr(AsyncCursor, method_name)
-        src = inspect.getsource(method)
-        assert "_check_loop_binding" in src, (
-            f"AsyncCursor.{method_name} must call _check_loop_binding"
-        )
-        assert "_ensure_locks" not in src, (
-            f"AsyncCursor.{method_name} must not call _ensure_locks (lazy-bind footgun)"
-        )
-
-
-def test_async_executescript_calls_non_binding_helper_not_ensure_locks() -> None:
-    src = inspect.getsource(AsyncCursor.executescript)
-    assert "_check_loop_binding" in src
-    assert "_ensure_locks" not in src
 
 
 def test_fresh_cursor_fetch_does_not_bind_connection_loop() -> None:
     """A fresh cursor's first call being fetchone() raises without lazy-binding the loop."""
     from dqlitedbapi.aio.connection import AsyncConnection
+    from dqlitedbapi.aio.cursor import AsyncCursor
     from dqlitedbapi.exceptions import ProgrammingError
 
     aconn = AsyncConnection.__new__(AsyncConnection)

tests/test_async_fetch_does_not_lazy_bind_loop.py

@@ -13,36 +13,6 @@
 from dqlitedbapi.aio import AsyncConnection
 
 
-def test_transaction_owner_assignment_inside_try_frame_source_pin() -> None:
-    """Source-level pin: ``self._transaction_owner = token`` is INSIDE a
-    ``try:`` frame. The bytecode-boundary race can't be injected from Python,
-    so we pin the source structure instead."""
-    import ast
-    import inspect
-    import textwrap
-
-    from dqlitedbapi.aio import connection as conn_mod
-
-    src = textwrap.dedent(inspect.getsource(conn_mod.AsyncConnection.transaction))
-    tree = ast.parse(src)
-
-    def find_owner_assign_in_try(node: ast.AST) -> bool:
-        for child in ast.walk(node):
-            if isinstance(child, ast.Try):
-                for stmt in child.body:
-                    if isinstance(stmt, ast.Assign):
-                        for tgt in stmt.targets:
-                            if isinstance(tgt, ast.Attribute) and tgt.attr == "_transaction_owner":
-                                return True
-        return False
-
-    assert find_owner_assign_in_try(tree), (
-        "AsyncConnection.transaction() must set self._transaction_owner "
-        "INSIDE a try: frame so the finally clears the slot under any "
-        "BaseException that arrives at the assignment site."
-    )
-
-
 async def test_transaction_owner_cleared_after_body_baseexception() -> None:
     """A BaseException out of the body lets the finally clear the slot."""
     conn = AsyncConnection("localhost:9001")

tests/test_async_transaction_owner_set_inside_try_frame.py

@@ -5,30 +5,9 @@
 
 import sqlite3
 
-from dqlitedbapi.aio.connection import AsyncConnection
 from dqlitedbapi.connection import Connection
 
 
-def _prop_doc(cls: type, name: str) -> str:
-    desc = cls.__dict__[name]
-    assert isinstance(desc, property)
-    doc = desc.__doc__
-    assert doc is not None
-    return doc
-
-
-def test_sync_autocommit_doc_calls_out_setter_getter_round_trip() -> None:
-    doc = _prop_doc(Connection, "autocommit")
-    assert "round-trip" in doc.lower()
-    assert "LEGACY_TRANSACTION_CONTROL" in doc
-
-
-def test_async_autocommit_doc_calls_out_setter_getter_round_trip() -> None:
-    doc = _prop_doc(AsyncConnection, "autocommit")
-    assert "round-trip" in doc.lower()
-    assert "LEGACY_TRANSACTION_CONTROL" in doc
-
-
 def test_sync_autocommit_setter_minus_one_round_trips() -> None:
     conn = Connection("127.0.0.1:9999")
     try:

tests/test_async_with_exit_does_not_close.py

@@ -25,41 +25,3 @@ def test_rollback_on_unused_connection_is_noop(self) -> None:
             mock_get.assert_not_called()
 
         conn.close()
-
-
-class TestCommitRollbackAsyncNoSpuriousConnect:
-    def test_commit_async_does_not_call_get_async_connection(self) -> None:
-        """_commit_async should check _async_conn directly, not call _get_async_connection."""
-        import ast
-        import inspect
-        import textwrap
-
-        source = textwrap.dedent(inspect.getsource(Connection._commit_async))
-        tree = ast.parse(source)
-
-        for node in ast.walk(tree):
-            if isinstance(node, ast.Call):
-                func = node.func
-                if isinstance(func, ast.Attribute) and func.attr == "_get_async_connection":
-                    raise AssertionError(
-                        "_commit_async calls _get_async_connection which creates "
-                        "new connections. It should check _async_conn directly."
-                    )
-
-    def test_rollback_async_does_not_call_get_async_connection(self) -> None:
-        """_rollback_async should check _async_conn directly, not call _get_async_connection."""
-        import ast
-        import inspect
-        import textwrap
-
-        source = textwrap.dedent(inspect.getsource(Connection._rollback_async))
-        tree = ast.parse(source)
-
-        for node in ast.walk(tree):
-            if isinstance(node, ast.Call):
-                func = node.func
-                if isinstance(func, ast.Attribute) and func.attr == "_get_async_connection":
-                    raise AssertionError(
-                        "_rollback_async calls _get_async_connection which creates "
-                        "new connections. It should check _async_conn directly."
-                    )