xinmi/scripts/test_weknora_etl.py

"""Regression checks for the read-only SQLite snapshot used by weknora_etl."""
import sqlite3
import sys
import tempfile
import unittest
from pathlib import Path

sys.path.insert(0, str(Path(__file__).parent))
from weknora_etl import consistent_snapshot


class SnapshotTests(unittest.TestCase):
    def test_backup_includes_uncheckpointed_wal(self):
        with tempfile.TemporaryDirectory() as directory:
            path = Path(directory) / "legacy.sqlite"
            source = sqlite3.connect(path)
            source.execute("PRAGMA journal_mode=WAL")
            source.execute("CREATE TABLE item (id INTEGER PRIMARY KEY, value TEXT)")
            source.commit()
            source.execute("INSERT INTO item(value) VALUES ('only-in-wal')")
            source.commit()
            self.assertTrue(Path(str(path) + "-wal").exists())

            with consistent_snapshot(path) as (snapshot, info):
                self.assertEqual(info["snapshot_method"], "sqlite_backup")
                self.assertEqual(snapshot.execute("SELECT value FROM item").fetchone()[0], "only-in-wal")
                self.assertEqual(snapshot.execute("PRAGMA query_only").fetchone()[0], 1)
                with self.assertRaises(sqlite3.OperationalError):
                    snapshot.execute("CREATE TABLE forbidden (id INTEGER)")
            source.close()


if __name__ == "__main__":
    unittest.main()