test(state): the no-bare-connect guard covers the tracked helper

Gate review: `test_repair_path_has_no_bare_connects` pinned "the helper owns
exactly one sqlite3.connect(str(db_path))"; the helper now opens through
`connect_tracked`, so the guard is "no bare connect anywhere in the module".
Also drops the `connect_fn=sqlite3.connect` kwarg: `connect_tracked` late-binds
the same module attribute, so the kwarg (and its comment) said nothing true.
This commit is contained in:
kshitijk4poor
2026-09-19 02:43:03 +05:30
committed by kshitij
parent 471738da27
commit 1c5798939a
2 changed files with 7 additions and 20 deletions

View File

@@ -589,9 +589,7 @@ def _connect_repair_durable(db_path: Path, *, timeout: float = 5.0) -> sqlite3.C
"""
from hermes_cli.sqlite_safe_read import connect_tracked
# Through THIS module's sqlite3.connect so tests patching it keep control of the fd.
conn = connect_tracked(db_path, tracking_path=db_path, connect_fn=sqlite3.connect,
timeout=timeout, isolation_level=None)
conn = connect_tracked(db_path, tracking_path=db_path, timeout=timeout, isolation_level=None)
_reapply_durability_barriers(conn)
return conn

View File

@@ -130,23 +130,12 @@ def test_repair_path_has_no_bare_connects() -> None:
and first.args[0].id == "db_path"
)
helper = next(
node
for node in tree.body
if isinstance(node, ast.FunctionDef)
and node.name == "_connect_repair_durable"
)
helper_calls = [node for node in ast.walk(helper) if is_db_path_connect(node)]
assert len(helper_calls) == 1, (
"_connect_repair_durable must own exactly one sqlite3.connect(str(db_path), ...)"
)
all_calls = [node for node in ast.walk(tree) if is_db_path_connect(node)]
elsewhere = [node for node in all_calls if node not in helper_calls]
assert elsewhere == [], (
f"{len(elsewhere)} repair/probe connection(s) still bypass "
"_connect_repair_durable() and write state.db without the macOS "
"fsync barriers"
# The helper itself opens through ``connect_tracked`` (byte-probe registry, #63386), so no
# ``sqlite3.connect(str(db_path), ...)`` may remain anywhere in the module.
bare = [node for node in ast.walk(tree) if is_db_path_connect(node)]
assert bare == [], (
f"{len(bare)} repair/probe connection(s) bypass _connect_repair_durable() and write "
"state.db without the macOS fsync barriers or the live-connection registry"
)