xiaoxue_admin
/
mempalace
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

chore(tests): wrap sqlite3 connections in contextlib.closing 

Several tests opened sqlite3 connections without try/finally or
context-manager cleanup, relying on a flat conn.close() after the
work. Any assertion failure or exception between connect and close
leaked the connection until GC, producing
``ResourceWarning: unclosed database`` in CI logs.

On Python 3.13 / macOS the ResourceWarning isn't just noise: a
leaked connection can hold a SQLite advisory lock long enough for
later test setup to block on it, which appears to be the cause of
recent intermittent CI hangs on those two runners.

Wrap each affected ``conn = sqlite3.connect(...)`` block in
``contextlib.closing(...)`` so cleanup runs on the failure path too.
Mirrors the try/finally pattern already used in production code
(searcher.py, repair.py, backends/chroma.py).

No behavior change — same operations, same assertions, just
deterministic cleanup. All 162 affected tests pass locally.
This commit is contained in:
Igor Lins e Silva 2026-05-09 19:37:49 -03:00
parent 7d5e6c4e85
commit df5ca114b2
3 changed files with 144 additions and 160 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

121
tests/test_backends.py
View File

@ -2,6 +2,7 @@ import os
import pickle
import shutil
import sqlite3
from contextlib import closing
from pathlib import Path
import chromadb
@ -452,37 +453,33 @@ def test_get_collection_create_true_preserves_existing_metadata(tmp_path):
def test_fix_blob_seq_ids_converts_blobs_to_integers(tmp_path):
"""Simulate a ChromaDB 0.6.x database with BLOB seq_ids and verify repair."""
db_path = tmp_path / "chroma.sqlite3"
conn = sqlite3.connect(str(db_path))
conn.execute("CREATE TABLE embeddings (rowid INTEGER PRIMARY KEY, seq_id)")
# Insert BLOB seq_id like ChromaDB 0.6.x would
blob_42 = (42).to_bytes(8, byteorder="big")
conn.execute("INSERT INTO embeddings (seq_id) VALUES (?)", (blob_42,))
conn.commit()
conn.close()
with closing(sqlite3.connect(str(db_path))) as conn:
conn.execute("CREATE TABLE embeddings (rowid INTEGER PRIMARY KEY, seq_id)")
# Insert BLOB seq_id like ChromaDB 0.6.x would
blob_42 = (42).to_bytes(8, byteorder="big")
conn.execute("INSERT INTO embeddings (seq_id) VALUES (?)", (blob_42,))
conn.commit()
_fix_blob_seq_ids(str(tmp_path))
conn = sqlite3.connect(str(db_path))
row = conn.execute("SELECT seq_id, typeof(seq_id) FROM embeddings").fetchone()
assert row == (42, "integer")
conn.close()
with closing(sqlite3.connect(str(db_path))) as conn:
row = conn.execute("SELECT seq_id, typeof(seq_id) FROM embeddings").fetchone()
assert row == (42, "integer")
def test_fix_blob_seq_ids_noop_without_blobs(tmp_path):
"""No error when seq_ids are already integers."""
db_path = tmp_path / "chroma.sqlite3"
conn = sqlite3.connect(str(db_path))
conn.execute("CREATE TABLE embeddings (rowid INTEGER PRIMARY KEY, seq_id INTEGER)")
conn.execute("INSERT INTO embeddings (seq_id) VALUES (42)")
conn.commit()
conn.close()
with closing(sqlite3.connect(str(db_path))) as conn:
conn.execute("CREATE TABLE embeddings (rowid INTEGER PRIMARY KEY, seq_id INTEGER)")
conn.execute("INSERT INTO embeddings (seq_id) VALUES (42)")
conn.commit()
_fix_blob_seq_ids(str(tmp_path))
conn = sqlite3.connect(str(db_path))
row = conn.execute("SELECT seq_id, typeof(seq_id) FROM embeddings").fetchone()
assert row == (42, "integer")
conn.close()
with closing(sqlite3.connect(str(db_path))) as conn:
row = conn.execute("SELECT seq_id, typeof(seq_id) FROM embeddings").fetchone()
assert row == (42, "integer")
def test_fix_blob_seq_ids_noop_without_database(tmp_path):
@ -499,60 +496,56 @@ def test_fix_blob_seq_ids_does_not_touch_max_seq_id(tmp_path):
silently suppressed every subsequent embeddings_queue write.
"""
db_path = tmp_path / "chroma.sqlite3"
conn = sqlite3.connect(str(db_path))
conn.execute("CREATE TABLE embeddings (rowid INTEGER PRIMARY KEY, seq_id)")
conn.execute("CREATE TABLE max_seq_id (rowid INTEGER PRIMARY KEY, seq_id)")
sysdb10_blob = b"\x11\x11502607"
conn.execute("INSERT INTO max_seq_id (seq_id) VALUES (?)", (sysdb10_blob,))
conn.commit()
conn.close()
with closing(sqlite3.connect(str(db_path))) as conn:
conn.execute("CREATE TABLE embeddings (rowid INTEGER PRIMARY KEY, seq_id)")
conn.execute("CREATE TABLE max_seq_id (rowid INTEGER PRIMARY KEY, seq_id)")
sysdb10_blob = b"\x11\x11502607"
conn.execute("INSERT INTO max_seq_id (seq_id) VALUES (?)", (sysdb10_blob,))
conn.commit()
_fix_blob_seq_ids(str(tmp_path))
conn = sqlite3.connect(str(db_path))
row = conn.execute("SELECT seq_id, typeof(seq_id) FROM max_seq_id").fetchone()
assert row == (sysdb10_blob, "blob")
conn.close()
with closing(sqlite3.connect(str(db_path))) as conn:
row = conn.execute("SELECT seq_id, typeof(seq_id) FROM max_seq_id").fetchone()
assert row == (sysdb10_blob, "blob")
def test_fix_blob_seq_ids_skips_sysdb10_prefix_in_embeddings(tmp_path):
"""Defense-in-depth: sysdb-10 prefix in embeddings.seq_id is skipped."""
db_path = tmp_path / "chroma.sqlite3"
conn = sqlite3.connect(str(db_path))
conn.execute("CREATE TABLE embeddings (rowid INTEGER PRIMARY KEY, seq_id)")
sysdb10_blob = b"\x11\x11502607"
conn.execute("INSERT INTO embeddings (seq_id) VALUES (?)", (sysdb10_blob,))
conn.commit()
conn.close()
with closing(sqlite3.connect(str(db_path))) as conn:
conn.execute("CREATE TABLE embeddings (rowid INTEGER PRIMARY KEY, seq_id)")
sysdb10_blob = b"\x11\x11502607"
conn.execute("INSERT INTO embeddings (seq_id) VALUES (?)", (sysdb10_blob,))
conn.commit()
_fix_blob_seq_ids(str(tmp_path))
conn = sqlite3.connect(str(db_path))
row = conn.execute("SELECT seq_id, typeof(seq_id) FROM embeddings").fetchone()
# Still a BLOB — not converted to 1.23e18.
assert row == (sysdb10_blob, "blob")
conn.close()
with closing(sqlite3.connect(str(db_path))) as conn:
row = conn.execute("SELECT seq_id, typeof(seq_id) FROM embeddings").fetchone()
# Still a BLOB — not converted to 1.23e18.
assert row == (sysdb10_blob, "blob")
def test_fix_blob_seq_ids_still_converts_legacy_blobs_in_embeddings(tmp_path):
"""Regression guard: pure big-endian u64 BLOBs still convert for genuine 0.6.x."""
db_path = tmp_path / "chroma.sqlite3"
conn = sqlite3.connect(str(db_path))
conn.execute("CREATE TABLE embeddings (rowid INTEGER PRIMARY KEY, seq_id)")
conn.execute("INSERT INTO embeddings (seq_id) VALUES (?)", ((42).to_bytes(8, "big"),))
conn.execute("INSERT INTO embeddings (seq_id) VALUES (?)", (b"\x11\x11502607",))
conn.execute("INSERT INTO embeddings (seq_id) VALUES (?)", ((7).to_bytes(8, "big"),))
conn.commit()
conn.close()
with closing(sqlite3.connect(str(db_path))) as conn:
conn.execute("CREATE TABLE embeddings (rowid INTEGER PRIMARY KEY, seq_id)")
conn.execute("INSERT INTO embeddings (seq_id) VALUES (?)", ((42).to_bytes(8, "big"),))
conn.execute("INSERT INTO embeddings (seq_id) VALUES (?)", (b"\x11\x11502607",))
conn.execute("INSERT INTO embeddings (seq_id) VALUES (?)", ((7).to_bytes(8, "big"),))
conn.commit()
_fix_blob_seq_ids(str(tmp_path))
conn = sqlite3.connect(str(db_path))
rows = conn.execute("SELECT seq_id, typeof(seq_id) FROM embeddings ORDER BY rowid").fetchall()
assert rows[0] == (42, "integer")
assert rows[1] == (b"\x11\x11502607", "blob") # sysdb-10 row left alone
assert rows[2] == (7, "integer")
conn.close()
with closing(sqlite3.connect(str(db_path))) as conn:
rows = conn.execute(
"SELECT seq_id, typeof(seq_id) FROM embeddings ORDER BY rowid"
).fetchall()
assert rows[0] == (42, "integer")
assert rows[1] == (b"\x11\x11502607", "blob") # sysdb-10 row left alone
assert rows[2] == (7, "integer")
def test_fix_blob_seq_ids_writes_marker_after_blob_path(tmp_path):
@ -560,11 +553,10 @@ def test_fix_blob_seq_ids_writes_marker_after_blob_path(tmp_path):
from mempalace.backends.chroma import _BLOB_FIX_MARKER
db_path = tmp_path / "chroma.sqlite3"
conn = sqlite3.connect(str(db_path))
conn.execute("CREATE TABLE embeddings (rowid INTEGER PRIMARY KEY, seq_id)")
conn.execute("INSERT INTO embeddings (seq_id) VALUES (?)", ((42).to_bytes(8, "big"),))
conn.commit()
conn.close()
with closing(sqlite3.connect(str(db_path))) as conn:
conn.execute("CREATE TABLE embeddings (rowid INTEGER PRIMARY KEY, seq_id)")
conn.execute("INSERT INTO embeddings (seq_id) VALUES (?)", ((42).to_bytes(8, "big"),))
conn.commit()
marker = tmp_path / _BLOB_FIX_MARKER
assert not marker.exists()
@ -585,11 +577,10 @@ def test_fix_blob_seq_ids_writes_marker_when_already_integer(tmp_path):
from mempalace.backends.chroma import _BLOB_FIX_MARKER
db_path = tmp_path / "chroma.sqlite3"
conn = sqlite3.connect(str(db_path))
conn.execute("CREATE TABLE embeddings (rowid INTEGER PRIMARY KEY, seq_id INTEGER)")
conn.execute("INSERT INTO embeddings (seq_id) VALUES (42)")
conn.commit()
conn.close()
with closing(sqlite3.connect(str(db_path))) as conn:
conn.execute("CREATE TABLE embeddings (rowid INTEGER PRIMARY KEY, seq_id INTEGER)")
conn.execute("INSERT INTO embeddings (seq_id) VALUES (42)")
conn.commit()
marker = tmp_path / _BLOB_FIX_MARKER
assert not marker.exists()

102
tests/test_repair.py
View File

@ -2,6 +2,7 @@
import os
import sqlite3
from contextlib import closing
from unittest.mock import MagicMock, call, patch
import pytest
@ -809,63 +810,62 @@ def _seed_poisoned_max_seq_id(
closets_vec = "seg-closets-vec-0000-1111-2222-333344445555"
closets_meta = "seg-closets-meta-0000-1111-2222-33334444555"
conn = sqlite3.connect(db_path)
conn.executescript(
"""
CREATE TABLE segments(
id TEXT PRIMARY KEY, type TEXT, scope TEXT, collection TEXT
);
CREATE TABLE max_seq_id(segment_id TEXT PRIMARY KEY, seq_id);
CREATE TABLE embeddings(
id INTEGER PRIMARY KEY AUTOINCREMENT,
segment_id TEXT,
embedding_id TEXT,
seq_id
);
CREATE TABLE embeddings_queue(seq_id INTEGER PRIMARY KEY, topic TEXT, id TEXT);
CREATE TABLE collection_metadata(collection_id TEXT, key TEXT, str_value TEXT);
"""
)
conn.executemany(
"INSERT INTO segments VALUES (?, ?, ?, ?)",
[
(drawers_vec, "urn:vector", "VECTOR", drawers_coll),
(drawers_meta, "urn:metadata", "METADATA", drawers_coll),
(closets_vec, "urn:vector", "VECTOR", closets_coll),
(closets_meta, "urn:metadata", "METADATA", closets_coll),
],
)
conn.executemany(
"INSERT INTO max_seq_id(segment_id, seq_id) VALUES (?, ?)",
[
(drawers_vec, drawers_vec_poison),
(drawers_meta, drawers_meta_poison),
(closets_vec, closets_vec_poison),
(closets_meta, closets_meta_poison),
],
)
# Populate embeddings so the collection-MAX heuristic has data to work with.
# drawers METADATA owns the max at drawers_meta_max; closets likewise.
for i in range(1, drawers_meta_max + 1, max(drawers_meta_max // 5, 1)):
with closing(sqlite3.connect(db_path)) as conn:
conn.executescript(
"""
CREATE TABLE segments(
id TEXT PRIMARY KEY, type TEXT, scope TEXT, collection TEXT
);
CREATE TABLE max_seq_id(segment_id TEXT PRIMARY KEY, seq_id);
CREATE TABLE embeddings(
id INTEGER PRIMARY KEY AUTOINCREMENT,
segment_id TEXT,
embedding_id TEXT,
seq_id
);
CREATE TABLE embeddings_queue(seq_id INTEGER PRIMARY KEY, topic TEXT, id TEXT);
CREATE TABLE collection_metadata(collection_id TEXT, key TEXT, str_value TEXT);
"""
)
conn.executemany(
"INSERT INTO segments VALUES (?, ?, ?, ?)",
[
(drawers_vec, "urn:vector", "VECTOR", drawers_coll),
(drawers_meta, "urn:metadata", "METADATA", drawers_coll),
(closets_vec, "urn:vector", "VECTOR", closets_coll),
(closets_meta, "urn:metadata", "METADATA", closets_coll),
],
)
conn.executemany(
"INSERT INTO max_seq_id(segment_id, seq_id) VALUES (?, ?)",
[
(drawers_vec, drawers_vec_poison),
(drawers_meta, drawers_meta_poison),
(closets_vec, closets_vec_poison),
(closets_meta, closets_meta_poison),
],
)
# Populate embeddings so the collection-MAX heuristic has data to work with.
# drawers METADATA owns the max at drawers_meta_max; closets likewise.
for i in range(1, drawers_meta_max + 1, max(drawers_meta_max // 5, 1)):
conn.execute(
"INSERT INTO embeddings(segment_id, embedding_id, seq_id) VALUES (?, ?, ?)",
(drawers_meta, f"d-{i}", i),
)
conn.execute(
"INSERT INTO embeddings(segment_id, embedding_id, seq_id) VALUES (?, ?, ?)",
(drawers_meta, f"d-{i}", i),
(drawers_meta, "d-max", drawers_meta_max),
)
conn.execute(
"INSERT INTO embeddings(segment_id, embedding_id, seq_id) VALUES (?, ?, ?)",
(drawers_meta, "d-max", drawers_meta_max),
)
for i in range(1, closets_meta_max + 1, max(closets_meta_max // 5, 1)):
for i in range(1, closets_meta_max + 1, max(closets_meta_max // 5, 1)):
conn.execute(
"INSERT INTO embeddings(segment_id, embedding_id, seq_id) VALUES (?, ?, ?)",
(closets_meta, f"c-{i}", i),
)
conn.execute(
"INSERT INTO embeddings(segment_id, embedding_id, seq_id) VALUES (?, ?, ?)",
(closets_meta, f"c-{i}", i),
(closets_meta, "c-max", closets_meta_max),
)
conn.execute(
"INSERT INTO embeddings(segment_id, embedding_id, seq_id) VALUES (?, ?, ?)",
(closets_meta, "c-max", closets_meta_max),
)
conn.commit()
conn.close()
conn.commit()
return {
"drawers_vec": drawers_vec,
"drawers_meta": drawers_meta,

81
tests/test_sources.py
View File

@ -1,5 +1,8 @@
"""Tests for the RFC 002 source-adapter scaffolding."""
import sqlite3
from contextlib import closing
import pytest
from mempalace.sources import (
@ -362,16 +365,13 @@ def test_knowledge_graph_add_triple_accepts_source_drawer_id_and_adapter_name(tm
)
assert triple_id is not None
import sqlite3
conn = sqlite3.connect(str(tmp_path / "kg.sqlite3"))
conn.row_factory = sqlite3.Row
row = conn.execute(
"SELECT source_drawer_id, adapter_name FROM triples WHERE id=?", (triple_id,)
).fetchone()
assert row["source_drawer_id"] == "abc123_0"
assert row["adapter_name"] == "git"
conn.close()
with closing(sqlite3.connect(str(tmp_path / "kg.sqlite3"))) as conn:
conn.row_factory = sqlite3.Row
row = conn.execute(
"SELECT source_drawer_id, adapter_name FROM triples WHERE id=?", (triple_id,)
).fetchone()
assert row["source_drawer_id"] == "abc123_0"
assert row["adapter_name"] == "git"
finally:
kg.close()
@ -380,15 +380,12 @@ def test_knowledge_graph_fresh_schema_includes_new_columns(tmp_path):
"""Brand-new palaces should get source_drawer_id / adapter_name directly
from CREATE TABLE, not via a post-hoc ALTER. _migrate_schema exists only
for legacy palaces."""
import sqlite3
from mempalace.knowledge_graph import KnowledgeGraph
kg = KnowledgeGraph(db_path=str(tmp_path / "fresh.sqlite3"))
try:
conn = sqlite3.connect(str(tmp_path / "fresh.sqlite3"))
cols = {row[1] for row in conn.execute("PRAGMA table_info(triples)")}
conn.close()
with closing(sqlite3.connect(str(tmp_path / "fresh.sqlite3"))) as conn:
cols = {row[1] for row in conn.execute("PRAGMA table_info(triples)")}
assert "source_drawer_id" in cols
assert "adapter_name" in cols
finally:
@ -397,42 +394,38 @@ def test_knowledge_graph_fresh_schema_includes_new_columns(tmp_path):
def test_knowledge_graph_migration_adds_missing_columns_to_old_schema(tmp_path):
"""An old-schema triples table (pre-RFC 002) should auto-migrate on open."""
import sqlite3
db_path = tmp_path / "legacy.sqlite3"
conn = sqlite3.connect(str(db_path))
conn.executescript("""
CREATE TABLE entities (
id TEXT PRIMARY KEY,
name TEXT NOT NULL,
type TEXT DEFAULT 'unknown',
properties TEXT DEFAULT '{}',
created_at TEXT DEFAULT CURRENT_TIMESTAMP
);
CREATE TABLE triples (
id TEXT PRIMARY KEY,
subject TEXT NOT NULL,
predicate TEXT NOT NULL,
object TEXT NOT NULL,
valid_from TEXT,
valid_to TEXT,
confidence REAL DEFAULT 1.0,
source_closet TEXT,
source_file TEXT,
extracted_at TEXT DEFAULT CURRENT_TIMESTAMP
);
""")
conn.commit()
conn.close()
with closing(sqlite3.connect(str(db_path))) as conn:
conn.executescript("""
CREATE TABLE entities (
id TEXT PRIMARY KEY,
name TEXT NOT NULL,
type TEXT DEFAULT 'unknown',
properties TEXT DEFAULT '{}',
created_at TEXT DEFAULT CURRENT_TIMESTAMP
);
CREATE TABLE triples (
id TEXT PRIMARY KEY,
subject TEXT NOT NULL,
predicate TEXT NOT NULL,
object TEXT NOT NULL,
valid_from TEXT,
valid_to TEXT,
confidence REAL DEFAULT 1.0,
source_closet TEXT,
source_file TEXT,
extracted_at TEXT DEFAULT CURRENT_TIMESTAMP
);
""")
conn.commit()
from mempalace.knowledge_graph import KnowledgeGraph
kg = KnowledgeGraph(db_path=str(db_path))
try:
# New columns must be present after _init_db runs the migration.
conn = sqlite3.connect(str(db_path))
cols = {row[1] for row in conn.execute("PRAGMA table_info(triples)")}
conn.close()
with closing(sqlite3.connect(str(db_path))) as conn:
cols = {row[1] for row in conn.execute("PRAGMA table_info(triples)")}
assert "source_drawer_id" in cols
assert "adapter_name" in cols