83 lines
4.6 KiB
Python
83 lines
4.6 KiB
Python
import importlib.util
|
|
from pathlib import Path
|
|
import sqlite3
|
|
import tempfile
|
|
import unittest
|
|
import hashlib
|
|
|
|
spec = importlib.util.spec_from_file_location("migration", Path(__file__).with_name("migrate_subscription_events.py"))
|
|
migration = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(migration)
|
|
|
|
|
|
class SubscriptionEventsMigrationTest(unittest.TestCase):
|
|
def test_explicit_backup_atomic_and_idempotent(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
path = Path(directory) / "main.db"
|
|
with sqlite3.connect(path) as db:
|
|
db.executescript("""
|
|
CREATE TABLE User(id INTEGER PRIMARY KEY,username TEXT);
|
|
CREATE TABLE sub_bind(key TEXT,user TEXT,user_id INTEGER);
|
|
CREATE TABLE sub_update_record(id INTEGER PRIMARY KEY,user_id INTEGER,time INTEGER);
|
|
INSERT INTO User VALUES(3,'alice');
|
|
INSERT INTO sub_bind VALUES('public-test-key','alice',3);
|
|
""")
|
|
dry = migration.migrate(path)
|
|
self.assertEqual(2, len(dry["tables_to_create"]))
|
|
with self.assertRaises(ValueError):
|
|
migration.migrate(path, True)
|
|
first = migration.migrate(path, True, True)
|
|
self.assertEqual("ok", first["integrity"])
|
|
self.assertEqual(0o600, Path(first["backup"]).stat().st_mode & 0o777)
|
|
with sqlite3.connect(first["backup"]) as db:
|
|
self.assertEqual(0, db.execute("SELECT count(*) FROM sqlite_master WHERE name='subscription_event_binding'").fetchone()[0])
|
|
with sqlite3.connect(path) as db:
|
|
row = db.execute("SELECT key_hash,user_id,user_name FROM subscription_event_binding").fetchone()
|
|
self.assertEqual((hashlib.sha256(b"public-test-key").hexdigest(), 3, "alice"), row)
|
|
db.execute("INSERT INTO subscription_event_receipt VALUES('id','hash','v2',0,'primary',1,'fingerprint','recorded')")
|
|
second = migration.migrate(path, True, True)
|
|
self.assertEqual([], second["tables_to_create"])
|
|
with sqlite3.connect(path) as db:
|
|
self.assertEqual(1, db.execute("SELECT count(*) FROM subscription_event_receipt").fetchone()[0])
|
|
self.assertEqual(1, db.execute("SELECT count(*) FROM subscription_event_binding").fetchone()[0])
|
|
|
|
def test_legacy_labels_without_user_ids_remain_null(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
path = Path(directory) / "main.db"
|
|
with sqlite3.connect(path) as db:
|
|
db.executescript("""
|
|
CREATE TABLE User(id INTEGER PRIMARY KEY,username TEXT);
|
|
CREATE TABLE sub_bind(key TEXT,user TEXT,user_id INTEGER);
|
|
CREATE TABLE sub_update_record(id INTEGER PRIMARY KEY,user_id INTEGER,time INTEGER);
|
|
INSERT INTO sub_bind VALUES('legacy-key','legacy-label',NULL);
|
|
""")
|
|
result = migration.migrate(path, True, True)
|
|
self.assertEqual(1, result["history_bindings"])
|
|
with sqlite3.connect(path) as db:
|
|
row = db.execute("SELECT key_hash,user_id,user_name FROM subscription_event_binding").fetchone()
|
|
self.assertEqual((hashlib.sha256(b"legacy-key").hexdigest(), None, "legacy-label"), row)
|
|
self.assertIsNone(db.execute("SELECT user_id FROM sub_bind").fetchone()[0])
|
|
self.assertEqual(0, db.execute("SELECT count(*) FROM User").fetchone()[0])
|
|
migration.migrate(path, True, True)
|
|
with sqlite3.connect(path) as db:
|
|
self.assertEqual(1, db.execute("SELECT count(*) FROM subscription_event_binding").fetchone()[0])
|
|
|
|
def test_orphan_binding_aborts_without_ddl(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
path = Path(directory) / "main.db"
|
|
with sqlite3.connect(path) as db:
|
|
db.executescript("""
|
|
CREATE TABLE User(id INTEGER PRIMARY KEY,username TEXT);
|
|
CREATE TABLE sub_bind(key TEXT,user TEXT,user_id INTEGER);
|
|
CREATE TABLE sub_update_record(id INTEGER PRIMARY KEY,user_id INTEGER,time INTEGER);
|
|
INSERT INTO sub_bind VALUES('test','deleted',9);
|
|
""")
|
|
with self.assertRaises(ValueError):
|
|
migration.migrate(path, True, True)
|
|
with sqlite3.connect(path) as db:
|
|
self.assertEqual(0, db.execute("SELECT count(*) FROM sqlite_master WHERE name LIKE 'subscription_event_%'").fetchone()[0])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|