64 lines
3.0 KiB
Python
64 lines
3.0 KiB
Python
import io
|
|
import sqlite3
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from server.db import Database
|
|
|
|
try:
|
|
from scripts.restore import restore_database
|
|
except ImportError:
|
|
restore_database=None
|
|
|
|
|
|
class RestoreTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.assertIsNotNone(restore_database,'Container restore must work as the application user')
|
|
self.temp=tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.root=Path(self.temp.name)
|
|
original=self.root/'backup.db'
|
|
db=Database(original)
|
|
with db.connect(write=True) as conn:
|
|
conn.execute('INSERT INTO system_state VALUES(?,?)',('restore-fixture','preserved'))
|
|
self.incoming=original.read_bytes()
|
|
self.destination=self.root/'data'/'simkeep.db'
|
|
|
|
def test_stream_import_preserves_database_contents_and_private_mode(self):
|
|
restore_database(self.destination,io.BytesIO(self.incoming))
|
|
with sqlite3.connect(self.destination) as conn:
|
|
self.assertEqual(conn.execute('SELECT value FROM system_state WHERE key=?',('restore-fixture',)).fetchone()[0],'preserved')
|
|
self.assertEqual(conn.execute('PRAGMA integrity_check').fetchone()[0],'ok')
|
|
self.assertEqual(self.destination.stat().st_mode&0o777,0o600)
|
|
|
|
def test_existing_database_requires_explicit_replace(self):
|
|
restore_database(self.destination,io.BytesIO(self.incoming))
|
|
before=self.destination.read_bytes()
|
|
with self.assertRaises(ValueError):
|
|
restore_database(self.destination,io.BytesIO(self.incoming))
|
|
self.assertEqual(self.destination.read_bytes(),before)
|
|
|
|
def test_invalid_or_unrelated_backup_preserves_existing_database(self):
|
|
restore_database(self.destination,io.BytesIO(self.incoming))
|
|
before=self.destination.read_bytes()
|
|
other=self.root/'other.db'
|
|
with sqlite3.connect(other) as conn:
|
|
conn.execute('CREATE TABLE unrelated(value TEXT)')
|
|
for incoming in (b'not a database',b'',other.read_bytes()):
|
|
with self.subTest(incoming_size=len(incoming)),self.assertRaises(ValueError):
|
|
restore_database(self.destination,io.BytesIO(incoming),replace=True)
|
|
self.assertEqual(self.destination.read_bytes(),before)
|
|
|
|
def test_explicit_restore_removes_old_sidecars_and_replaces_contents(self):
|
|
restore_database(self.destination,io.BytesIO(self.incoming))
|
|
with sqlite3.connect(self.destination) as conn:
|
|
conn.execute('DELETE FROM system_state')
|
|
for suffix in ('-wal','-shm'):
|
|
Path(str(self.destination)+suffix).write_bytes(b'old stopped database sidecar')
|
|
restore_database(self.destination,io.BytesIO(self.incoming),replace=True)
|
|
for suffix in ('-wal','-shm'):
|
|
self.assertFalse(Path(str(self.destination)+suffix).exists())
|
|
with sqlite3.connect(self.destination) as conn:
|
|
self.assertEqual(conn.execute('SELECT value FROM system_state').fetchone()[0],'preserved')
|