leefer/xiaobai-review
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

test: cover m0004 add_mentor_note migration paths

Browse Source 
Co-authored-by: multica-agent <github@multica.ai>
This commit is contained in:
MS-01-Codex
2026-08-18 20:12:18 +08:00
co-authored by multica-agent
parent c00582da7c
commit 71b3ab1d5d
1 changed files with 81 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_database_migrations.py
+81 -9
View File
@@ -18,20 +18,92 @@ class DatabaseMigrationTests(unittest.TestCase):
rows = connection.execute( rows = connection.execute(
"SELECT version, name FROM schema_migrations" "SELECT version, name FROM schema_migrations"
).fetchall() ).fetchall()
self.assertEqual( self.assertEqual(
[(row["version"], row["name"]) for row in rows], [(row["version"], row["name"]) for row in rows],
[ [
("0001", "adopt_legacy_schema"), ("0001", "adopt_legacy_schema"),
("0002", "create_job_runs"), ("0002", "create_job_runs"),
("0003", "extend_llm_audit"), ("0003", "extend_llm_audit"),
], ("0004", "add_mentor_note"),
) ],
)
columns = {
str(row["name"])
for row in connection.execute(
"PRAGMA table_info(mentor_preferences)"
)
}
self.assertIn("note", columns)
ReviewDatabase(path) ReviewDatabase(path)
with database.connect() as connection: with database.connect() as connection:
count = connection.execute( count = connection.execute(
"SELECT COUNT(*) AS count FROM schema_migrations" "SELECT COUNT(*) AS count FROM schema_migrations"
).fetchone()["count"] ).fetchone()["count"]
self.assertEqual(count, 3) self.assertEqual(count, 4)
def test_database_with_recorded_0004_and_note_column_starts_without_reapply(
self,
) -> None:
with tempfile.TemporaryDirectory() as root:
path = Path(root) / "review.db"
database = ReviewDatabase(path)
with database.connect() as connection:
note_rows = [
str(row["name"])
for row in connection.execute(
"PRAGMA table_info(mentor_preferences)"
)
]
self.assertIn("note", note_rows)
ReviewDatabase(path)
with database.connect() as connection:
count = connection.execute(
"SELECT COUNT(*) AS count FROM schema_migrations"
).fetchone()["count"]
self.assertEqual(count, 4)
def test_old_database_without_0004_upgrades_and_adds_note_column(self) -> None:
with tempfile.TemporaryDirectory() as root:
path = Path(root) / "review.db"
database = ReviewDatabase(path)
with database.connect() as connection:
connection.execute(
"DELETE FROM schema_migrations WHERE version = '0004'"
)
connection.execute(
"ALTER TABLE mentor_preferences DROP COLUMN note"
)
ReviewDatabase(path)
with database.connect() as connection:
versions = {
str(row["version"])
for row in connection.execute(
"SELECT version FROM schema_migrations"
)
}
note_rows = [
str(row["name"])
for row in connection.execute(
"PRAGMA table_info(mentor_preferences)"
)
]
self.assertEqual(versions, {"0001", "0002", "0003", "0004"})
self.assertIn("note", note_rows)
def test_database_with_unknown_migration_is_rejected(self) -> None:
with tempfile.TemporaryDirectory() as root:
path = Path(root) / "review.db"
database = ReviewDatabase(path)
with database.connect() as connection:
connection.execute(
"""
INSERT INTO schema_migrations
(version, name, checksum, applied_at)
VALUES ('9999', 'unknown_legacy', 'x', '2026-08-01T00:00:00+00:00')
"""
)
with self.assertRaises(MigrationError):
ReviewDatabase(path)
def test_connection_factory_enables_required_pragmas(self) -> None: def test_connection_factory_enables_required_pragmas(self) -> None:
with tempfile.TemporaryDirectory() as root: with tempfile.TemporaryDirectory() as root: