67 lines
2.3 KiB
Python
67 lines
2.3 KiB
Python
"""#103339 item 2: ``hermes doctor --fix`` never checkpoints state.db through a bare writable ``sqlite3.connect``
|
|
— the checkpoint runs on the exclusive repair guard, so an opener arriving after the holder scan is refused."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sqlite3
|
|
from pathlib import Path
|
|
|
|
import hermes_state_repair
|
|
from hermes_cli.doctor_report import Finding
|
|
from hermes_cli.doctor_state import _state_db_wal
|
|
|
|
|
|
def test_doctor_checkpoint_runs_only_on_the_exclusive_repair_guard(tmp_path, monkeypatch):
|
|
db = tmp_path / "state.db"
|
|
setup = sqlite3.connect(str(db))
|
|
setup.execute("CREATE TABLE t(x)")
|
|
setup.execute("PRAGMA journal_mode=WAL")
|
|
setup.execute("INSERT INTO t VALUES (1)")
|
|
setup.commit()
|
|
setup.close()
|
|
wal = Path(f"{db}-wal")
|
|
with open(wal, "ab") as handle:
|
|
handle.truncate(51 * 1024 * 1024)
|
|
|
|
bare_connects: list[str] = []
|
|
real_connect = sqlite3.connect
|
|
|
|
def _spy(database, *args, **kwargs):
|
|
if not str(database).startswith("file:") or "mode=ro" not in str(database):
|
|
bare_connects.append(str(database))
|
|
return real_connect(database, *args, **kwargs)
|
|
|
|
monkeypatch.setattr(sqlite3, "connect", _spy)
|
|
guard_connects: list[Path] = []
|
|
real_durable = hermes_state_repair._connect_repair_durable
|
|
|
|
def _durable(path, **kwargs):
|
|
guard_connects.append(Path(path))
|
|
return real_durable(path, **kwargs)
|
|
|
|
monkeypatch.setattr(hermes_state_repair, "_connect_repair_durable", _durable)
|
|
|
|
finding = Finding()
|
|
_state_db_wal(finding, True, db)
|
|
|
|
assert finding.fixed == 1 and not finding.issues
|
|
# Every writable open went through the repair connector (probe + exclusive guard); none was a bare connect.
|
|
assert bare_connects and len(bare_connects) == len(guard_connects) >= 2
|
|
|
|
|
|
def test_session_count_reads_a_home_with_uri_reserved_characters(tmp_path):
|
|
"""`file:` URIs treat '?' and '#' as delimiters; a home named `profile?blue` must still count."""
|
|
import sqlite3
|
|
|
|
from hermes_cli.doctor_state import _session_count
|
|
|
|
home = tmp_path / "profile?blue#x"
|
|
home.mkdir()
|
|
db = home / "state.db"
|
|
conn = sqlite3.connect(db)
|
|
conn.execute("CREATE TABLE sessions(id TEXT)")
|
|
conn.execute("INSERT INTO sessions VALUES ('a'), ('b')")
|
|
conn.commit()
|
|
conn.close()
|
|
assert _session_count(db) == 2
|