56 lines
2.8 KiB
Python
56 lines
2.8 KiB
Python
import importlib.util
|
|
from pathlib import Path
|
|
import sqlite3
|
|
import tempfile
|
|
import unittest
|
|
|
|
spec = importlib.util.spec_from_file_location("migration", Path(__file__).with_name("migrate_user_links.py"))
|
|
migration = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(migration)
|
|
|
|
|
|
class MigrationTest(unittest.TestCase):
|
|
def test_dry_run_idempotent_migration_and_rename(self):
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
path = Path(temporary) / "test.db"
|
|
with sqlite3.connect(path) as db:
|
|
db.executescript("""
|
|
CREATE TABLE User(id INTEGER PRIMARY KEY, username TEXT);
|
|
CREATE TABLE sub_bind(key TEXT PRIMARY KEY, user TEXT, subscription_account_id INTEGER);
|
|
CREATE TABLE sub_update_record(id INTEGER PRIMARY KEY, user TEXT, location TEXT);
|
|
INSERT INTO User VALUES(1,'old-name');
|
|
INSERT INTO sub_bind VALUES('example-key','old-name',2);
|
|
INSERT INTO sub_update_record VALUES(10,'old-name','unknown');
|
|
INSERT INTO sub_update_record VALUES(11,'old-name','unknown');
|
|
""")
|
|
before = path.read_bytes()
|
|
self.assertEqual(2, len(migration.migrate(path)["schema_changes"]))
|
|
self.assertEqual(before, path.read_bytes())
|
|
self.assertEqual(0, migration.migrate(path, True)["unresolved_bindings"])
|
|
self.assertEqual([], migration.migrate(path, True)["schema_changes"])
|
|
with sqlite3.connect(path) as db:
|
|
db.execute("update User set username='new-name' where id=1")
|
|
self.assertEqual((1,), db.execute("select user_id from sub_bind").fetchone())
|
|
# A slow first lookup must not overwrite the second request's row.
|
|
db.execute("update sub_update_record set location='first-location' where id=10")
|
|
self.assertEqual(("unknown",), db.execute("select location from sub_update_record where id=11").fetchone())
|
|
|
|
def test_ambiguous_names_fail_without_modifying_database(self):
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
path = Path(temporary) / "test.db"
|
|
with sqlite3.connect(path) as db:
|
|
db.executescript("""
|
|
CREATE TABLE User(id INTEGER PRIMARY KEY, username TEXT);
|
|
CREATE TABLE sub_bind(user TEXT);
|
|
CREATE TABLE sub_update_record(user TEXT);
|
|
INSERT INTO User VALUES(1,'same'),(2,'same');
|
|
""")
|
|
before = path.read_bytes()
|
|
with self.assertRaises(ValueError):
|
|
migration.migrate(path, True)
|
|
self.assertEqual(before, path.read_bytes())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|