66 lines
3.7 KiB
Python
66 lines
3.7 KiB
Python
|
import datetime as dt
|
||
|
|
from contextlib import closing
|
||
|
|
import sqlite3
|
||
|
|
import unittest
|
||
|
|
from unittest.mock import patch
|
||
|
|
from test_app import app
|
||
|
|
import test_households as fixture
|
||
|
|
from backups import Backups
|
||
|
|
|
||
|
|
|
||
|
|
class BackupTests(unittest.TestCase):
|
||
|
|
setUp=fixture.HouseholdTests.setUp
|
||
|
|
stop_server=fixture.HouseholdTests.stop_server
|
||
|
|
request=fixture.HouseholdTests.request
|
||
|
|
|
||
|
|
def source_file(self):
|
||
|
|
with closing(sqlite3.connect(app.DATA/'vision.db')) as dest:app.DB.backup(dest)
|
||
|
|
return Backups(app)
|
||
|
|
|
||
|
|
def test_backup_permissions_and_link_validation(self):
|
||
|
|
for user in (self.owner,self.members[self.aid]):
|
||
|
|
self.assertEqual(self.request('/api/backups',user)[0],403)
|
||
|
|
self.assertEqual(self.request('/api/backups/run',user,{})[0],403)
|
||
|
|
config=app.BACKUPS.config();config['networkPath']='https://pan.baidu.com/share/example'
|
||
|
|
self.assertEqual(self.request('/api/backups/config',self.root,config)[0],400)
|
||
|
|
config['networkPath']=str(app.DATA)
|
||
|
|
self.assertEqual(self.request('/api/backups/config',self.root,config)[0],400)
|
||
|
|
|
||
|
|
def test_local_snapshot_is_integral_and_does_not_copy_active_sessions(self):
|
||
|
|
backup=self.source_file();before=app.DB.execute('select count(*) from sessions').fetchone()[0]
|
||
|
|
result=backup.run();self.assertTrue(result['localOk'],result)
|
||
|
|
with closing(sqlite3.connect(app.DATA/'managed-backups'/result['file'])) as copy:
|
||
|
|
self.assertEqual(copy.execute('pragma integrity_check').fetchone()[0],'ok')
|
||
|
|
self.assertEqual(copy.execute('select count(*) from users').fetchone()[0],len(app.objects('users')))
|
||
|
|
self.assertEqual(copy.execute('select count(*) from sessions').fetchone()[0],0)
|
||
|
|
self.assertEqual(app.DB.execute('select count(*) from sessions').fetchone()[0],before)
|
||
|
|
self.assertEqual(len(result['sha256']),64)
|
||
|
|
|
||
|
|
def test_network_failure_keeps_local_success_and_is_not_reported_as_cloud_success(self):
|
||
|
|
backup=self.source_file();app.set_setting('backup_config',dict(backup.config(),networkPath=str(app.DATA/'missing')))
|
||
|
|
result=backup.run()
|
||
|
|
self.assertTrue(result['localOk']);self.assertFalse(result['networkOk']);self.assertIn('error',result)
|
||
|
|
|
||
|
|
def test_network_copy_checksum_and_retention_only_touch_owned_files(self):
|
||
|
|
backup=self.source_file();network=app.DATA/'network';network.mkdir();unrelated=network/'keep.sqlite';unrelated.write_text('unrelated')
|
||
|
|
app.set_setting('backup_config',dict(backup.config(),networkPath=str(network),keep=1))
|
||
|
|
with patch('backups.os.path.ismount',side_effect=lambda p:p==network.resolve()):
|
||
|
|
first=backup.run();second=backup.run()
|
||
|
|
self.assertTrue(first['networkOk'],first);self.assertTrue(second['networkOk'],second)
|
||
|
|
self.assertTrue(unrelated.exists());self.assertFalse((network/first['file']).exists());self.assertTrue((network/second['file']).exists())
|
||
|
|
|
||
|
|
def test_daily_schedule_once_per_beijing_date_and_disabled_by_default(self):
|
||
|
|
backup=Backups(app);now=dt.datetime(2026,10,4,4)
|
||
|
|
with patch.object(backup,'start') as start:
|
||
|
|
self.assertFalse(backup.tick(now))
|
||
|
|
app.set_setting('backup_config',dict(backup.config(),enabled=True))
|
||
|
|
self.assertTrue(backup.tick(now));self.assertFalse(backup.tick(now))
|
||
|
|
self.assertTrue(backup.tick(now+dt.timedelta(days=1)));self.assertEqual(start.call_count,2)
|
||
|
|
|
||
|
|
def test_config_revision_and_busy_guard(self):
|
||
|
|
backup=Backups(app);data=backup.config();backup.save(data)
|
||
|
|
with self.assertRaises(app.Problem):backup.save(data)
|
||
|
|
backup.running=True
|
||
|
|
with self.assertRaises(app.Problem):backup.start()
|
||
|
|
with self.assertRaises(app.Problem):backup.save(backup.config())
|