1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
|
import sqlite3
import pytest
from personal_agent.database.database import Database
def test_migrate_creates_schema(tmp_path):
database = Database(tmp_path / "test.db")
database.migrate()
with database.connect() as connection:
tables = {
row[0]
for row in connection.execute(
"""
SELECT name
FROM sqlite_master
WHERE type = 'table'
"""
)
}
assert "schema_migrations" in tables
assert "memories" in tables
def test_migrate_records_version(tmp_path):
database = Database(tmp_path / "test.db")
database.migrate()
with database.connect() as connection:
rows = connection.execute("SELECT version FROM schema_migrations").fetchall()
assert rows == [(1,), (2,), (3,)]
def test_migrate_is_idempotent(tmp_path):
database = Database(tmp_path / "test.db")
database.migrate()
database.migrate()
with database.connect() as connection:
rows = connection.execute("SELECT version FROM schema_migrations").fetchall()
assert rows == [(1,), (2,), (3,)]
def test_failed_migration_rolls_back(tmp_path):
database = Database(tmp_path / "test.db")
migration = database.migrations_dir / "999_failing_migration.sql"
migration.write_text(
"""
THIS IS INVALID SQL;
""",
encoding="utf-8",
)
try:
with pytest.raises(sqlite3.OperationalError):
database.migrate()
with database.connect() as connection:
migrations = connection.execute(
"SELECT version FROM schema_migrations"
).fetchall()
assert migrations == [(1,), (2,), (3,)]
finally:
migration.unlink()
def test_duplicate_migration_versions_are_rejected(tmp_path):
database = Database(tmp_path / "test.db")
duplicate = database.migrations_dir / "001_duplicate.sql"
duplicate.write_text(
"CREATE TABLE duplicate_table (id INTEGER PRIMARY KEY);",
encoding="utf-8",
)
try:
with pytest.raises(ValueError, match="Duplicate migration version"):
database.migrate()
finally:
duplicate.unlink()
|
