58 lines
2.3 KiB
Python
58 lines
2.3 KiB
Python
import os
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from archive_clients.backup import BackupError, SQLiteBackupManager
|
|
from archive_clients.config import BackupConfig
|
|
from archive_clients.locking import DatabaseLease, DatabaseLockedError
|
|
from archive_clients.state import ClientStore
|
|
|
|
|
|
class BackupTests(unittest.TestCase):
|
|
def test_create_verify_detect_corruption_and_restore(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
database = root / "state" / "client.db"
|
|
backups = root / "backups"
|
|
store = ClientStore(database)
|
|
store.initialize()
|
|
manager = SQLiteBackupManager(database, backups, BackupConfig())
|
|
record = manager.create("test")
|
|
self.assertEqual(manager.verify(record.database).sha256, record.sha256)
|
|
|
|
database.write_bytes(b"broken")
|
|
preserved = manager.restore(record.database)
|
|
self.assertIsNotNone(preserved)
|
|
self.assertEqual(ClientStore(database).list_active_job_cursors(), [])
|
|
|
|
with record.database.open("ab") as target:
|
|
target.write(b"corruption")
|
|
with self.assertRaisesRegex(BackupError, "checksum"):
|
|
manager.verify(record.database)
|
|
|
|
def test_restore_refuses_a_live_database_lease(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
database = root / "client.db"
|
|
ClientStore(database).initialize()
|
|
manager = SQLiteBackupManager(database, root / "backups", BackupConfig())
|
|
record = manager.create("test")
|
|
with DatabaseLease(database):
|
|
with self.assertRaises(DatabaseLockedError):
|
|
manager.restore(record.database)
|
|
|
|
def test_backup_database_has_private_permissions(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
database = root / "client.db"
|
|
ClientStore(database).initialize()
|
|
record = SQLiteBackupManager(
|
|
database, root / "backups", BackupConfig()
|
|
).create("test")
|
|
self.assertEqual(os.stat(record.database).st_mode & 0o777, 0o600)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|