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()