Extract shared Python test utilities into tests/helpers.py

Deduplicates exec(), vec0_shadow_table_contents(), _f32(), _i64(), and
_int8() helpers that were copied across six test files.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

tests/helpers.py

@@ -0,0 +1,50 @@
+import sqlite3
+import struct
+from collections import OrderedDict
+
+
+def _f32(list):
+    return struct.pack("%sf" % len(list), *list)
+
+
+def _i64(list):
+    return struct.pack("%sq" % len(list), *list)
+
+
+def _int8(list):
+    return struct.pack("%sb" % len(list), *list)
+
+
+def exec(db, sql, parameters=[]):
+    try:
+        rows = db.execute(sql, parameters).fetchall()
+    except (sqlite3.OperationalError, sqlite3.DatabaseError) as e:
+        return {
+            "error": e.__class__.__name__,
+            "message": str(e),
+        }
+    a = []
+    for row in rows:
+        o = OrderedDict()
+        for k in row.keys():
+            o[k] = row[k]
+        a.append(o)
+    result = OrderedDict()
+    result["sql"] = sql
+    result["rows"] = a
+    return result
+
+
+def vec0_shadow_table_contents(db, v, skip_info=True):
+    shadow_tables = [
+        row[0]
+        for row in db.execute(
+            "select name from sqlite_master where name like ? order by 1", [f"{v}_%"]
+        ).fetchall()
+    ]
+    o = {}
+    for shadow_table in shadow_tables:
+        if skip_info and shadow_table.endswith("_info"):
+            continue
+        o[shadow_table] = exec(db, f"select * from {shadow_table}")
+    return o

tests/test-auxiliary.py

@@ -1,5 +1,5 @@
 import sqlite3
-from collections import OrderedDict
+from helpers import exec, vec0_shadow_table_contents
 
 
 def test_constructor_limit(db, snapshot):
@@ -126,36 +126,3 @@ def test_knn(db, snapshot):
     ) == snapshot(name="illegal KNN w/ aux")
 
 
-def exec(db, sql, parameters=[]):
-    try:
-        rows = db.execute(sql, parameters).fetchall()
-    except (sqlite3.OperationalError, sqlite3.DatabaseError) as e:
-        return {
-            "error": e.__class__.__name__,
-            "message": str(e),
-        }
-    a = []
-    for row in rows:
-        o = OrderedDict()
-        for k in row.keys():
-            o[k] = row[k]
-        a.append(o)
-    result = OrderedDict()
-    result["sql"] = sql
-    result["rows"] = a
-    return result
-
-
-def vec0_shadow_table_contents(db, v):
-    shadow_tables = [
-        row[0]
-        for row in db.execute(
-            "select name from sqlite_master where name like ? order by 1", [f"{v}_%"]
-        ).fetchall()
-    ]
-    o = {}
-    for shadow_table in shadow_tables:
-        if shadow_table.endswith("_info"):
-            continue
-        o[shadow_table] = exec(db, f"select * from {shadow_table}")
-    return o

tests/test-general.py

@@ -1,6 +1,6 @@
 import sqlite3
-from collections import OrderedDict
 import pytest
+from helpers import exec
 
 
 @pytest.mark.skipif(
@@ -27,34 +27,3 @@ def test_info(db, snapshot):
     assert exec(db, "select key, typeof(value) from v_info order by 1") == snapshot()
 
 
-def exec(db, sql, parameters=[]):
-    try:
-        rows = db.execute(sql, parameters).fetchall()
-    except (sqlite3.OperationalError, sqlite3.DatabaseError) as e:
-        return {
-            "error": e.__class__.__name__,
-            "message": str(e),
-        }
-    a = []
-    for row in rows:
-        o = OrderedDict()
-        for k in row.keys():
-            o[k] = row[k]
-        a.append(o)
-    result = OrderedDict()
-    result["sql"] = sql
-    result["rows"] = a
-    return result
-
-
-def vec0_shadow_table_contents(db, v):
-    shadow_tables = [
-        row[0]
-        for row in db.execute(
-            "select name from sqlite_master where name like ? order by 1", [f"{v}_%"]
-        ).fetchall()
-    ]
-    o = {}
-    for shadow_table in shadow_tables:
-        o[shadow_table] = exec(db, f"select * from {shadow_table}")
-    return o

tests/test-insert-delete.py

@@ -1,31 +1,7 @@
 import sqlite3
 import struct
-from collections import OrderedDict
 import pytest
-
+from helpers import _f32, exec
-
-def _f32(list):
-    return struct.pack("%sf" % len(list), *list)
-
-
-def exec(db, sql, parameters=[]):
-    try:
-        rows = db.execute(sql, parameters).fetchall()
-    except (sqlite3.OperationalError, sqlite3.DatabaseError) as e:
-        return {
-            "error": e.__class__.__name__,
-            "message": str(e),
-        }
-    a = []
-    for row in rows:
-        o = OrderedDict()
-        for k in row.keys():
-            o[k] = row[k]
-        a.append(o)
-    result = OrderedDict()
-    result["sql"] = sql
-    result["rows"] = a
-    return result
 
 
 def test_insert_creates_chunks_and_vectors(db, snapshot):

tests/test-knn-distance-constraints.py

@@ -1,5 +1,5 @@
 import sqlite3
-from collections import OrderedDict
+from helpers import exec
 
 
 def test_normal(db, snapshot):
@@ -47,36 +47,3 @@ class Row:
         return repr()
 
 
-def exec(db, sql, parameters=[]):
-    try:
-        rows = db.execute(sql, parameters).fetchall()
-    except (sqlite3.OperationalError, sqlite3.DatabaseError) as e:
-        return {
-            "error": e.__class__.__name__,
-            "message": str(e),
-        }
-    a = []
-    for row in rows:
-        o = OrderedDict()
-        for k in row.keys():
-            o[k] = row[k]
-        a.append(o)
-    result = OrderedDict()
-    result["sql"] = sql
-    result["rows"] = a
-    return result
-
-
-def vec0_shadow_table_contents(db, v):
-    shadow_tables = [
-        row[0]
-        for row in db.execute(
-            "select name from sqlite_master where name like ? order by 1", [f"{v}_%"]
-        ).fetchall()
-    ]
-    o = {}
-    for shadow_table in shadow_tables:
-        if shadow_table.endswith("_info"):
-            continue
-        o[shadow_table] = exec(db, f"select * from {shadow_table}")
-    return o

tests/test-metadata.py

@@ -2,6 +2,7 @@ import pytest
 import sqlite3
 from collections import OrderedDict
 import json
+from helpers import exec, vec0_shadow_table_contents
 
 
 def test_constructor_limit(db, snapshot):
@@ -594,36 +595,3 @@ def authorizer_deny_on(operation, x1, x2=None):
     return _auth
 
 
-def exec(db, sql, parameters=[]):
-    try:
-        rows = db.execute(sql, parameters).fetchall()
-    except (sqlite3.OperationalError, sqlite3.DatabaseError) as e:
-        return {
-            "error": e.__class__.__name__,
-            "message": str(e),
-        }
-    a = []
-    for row in rows:
-        o = OrderedDict()
-        for k in row.keys():
-            o[k] = row[k]
-        a.append(o)
-    result = OrderedDict()
-    result["sql"] = sql
-    result["rows"] = a
-    return result
-
-
-def vec0_shadow_table_contents(db, v):
-    shadow_tables = [
-        row[0]
-        for row in db.execute(
-            "select name from sqlite_master where name like ? order by 1", [f"{v}_%"]
-        ).fetchall()
-    ]
-    o = {}
-    for shadow_table in shadow_tables:
-        if shadow_table.endswith("_info"):
-            continue
-        o[shadow_table] = exec(db, f"select * from {shadow_table}")
-    return o

tests/test-partition-keys.py

@@ -1,5 +1,5 @@
 import sqlite3
-from collections import OrderedDict
+from helpers import exec, vec0_shadow_table_contents
 
 
 def test_constructor_limit(db, snapshot):
@@ -82,36 +82,3 @@ class Row:
         return repr()
 
 
-def exec(db, sql, parameters=[]):
-    try:
-        rows = db.execute(sql, parameters).fetchall()
-    except (sqlite3.OperationalError, sqlite3.DatabaseError) as e:
-        return {
-            "error": e.__class__.__name__,
-            "message": str(e),
-        }
-    a = []
-    for row in rows:
-        o = OrderedDict()
-        for k in row.keys():
-            o[k] = row[k]
-        a.append(o)
-    result = OrderedDict()
-    result["sql"] = sql
-    result["rows"] = a
-    return result
-
-
-def vec0_shadow_table_contents(db, v):
-    shadow_tables = [
-        row[0]
-        for row in db.execute(
-            "select name from sqlite_master where name like ? order by 1", [f"{v}_%"]
-        ).fetchall()
-    ]
-    o = {}
-    for shadow_table in shadow_tables:
-        if shadow_table.endswith("_info"):
-            continue
-        o[shadow_table] = exec(db, f"select * from {shadow_table}")
-    return o