fix(tests): fix sqlite migrator backup and restore test

On Windows, we must ensure the connection to the database is closed before exiting the tempfile context.

Also, rejiggered the thing to use the file directly.
This commit is contained in:
psychedelicious@windows 2023-12-05 14:44:23 +11:00 committed by psychedelicious
parent 3c692018cd
commit abc9dc4d17

View File

@ -99,23 +99,35 @@ def test_run_migrations(migrator: SQLiteMigrator):
assert migrator._get_current_version() == 3 assert migrator._get_current_version() == 3
def test_backup_and_restore_db(migrator: SQLiteMigrator): def test_backup_and_restore_db():
# must do this with a file database - we don't backup/restore for memory
with TemporaryDirectory() as tempdir: with TemporaryDirectory() as tempdir:
# must do this with a file database - we don't backup/restore for memory # create test DB w/ some data
database = Path(tempdir) / "test.db" database = Path(tempdir) / "test.db"
migrator._database = database conn = sqlite3.connect(database, check_same_thread=False)
migrator._cursor.execute("CREATE TABLE test (id INTEGER PRIMARY KEY);") cursor = conn.cursor()
migrator._conn.commit() cursor.execute("CREATE TABLE test (id INTEGER PRIMARY KEY);")
conn.commit()
migrator = SQLiteMigrator(conn=conn, database=database, lock=threading.RLock(), logger=Logger("test"))
backup_path = migrator._backup_db(migrator._database) backup_path = migrator._backup_db(migrator._database)
# mangle the db
migrator._cursor.execute("DROP TABLE test;") migrator._cursor.execute("DROP TABLE test;")
migrator._conn.commit() migrator._conn.commit()
migrator._restore_db(backup_path) # this closes the connection migrator._cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='test';")
# reconnect to db assert migrator._cursor.fetchone() is None
# restore (closes the connection - must create a new one)
migrator._restore_db(backup_path)
restored_conn = sqlite3.connect(database) restored_conn = sqlite3.connect(database)
restored_cursor = restored_conn.cursor() restored_cursor = restored_conn.cursor()
restored_cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='test';") restored_cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='test';")
assert restored_cursor.fetchone() is not None assert restored_cursor.fetchone() is not None
# must manually close else tempfile throws on cleanup on windows
restored_conn.close()
def test_no_backup_and_restore_for_memory_db(migrator: SQLiteMigrator): def test_no_backup_and_restore_for_memory_db(migrator: SQLiteMigrator):
with pytest.raises(MigrationError, match="Cannot back up memory database"): with pytest.raises(MigrationError, match="Cannot back up memory database"):