Codex
125 lines
8.6 KiB
Python
125 lines
8.6 KiB
Python
import datetime as dt
|
|
from contextlib import closing
|
|
from pathlib import PurePosixPath
|
|
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.object(backup,'network_mount',return_value=True):
|
|
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())
|
|
|
|
def test_local_filesystems_are_never_reported_as_network_mounts(self):
|
|
records=['1 0 8:1 / / rw - ext4 /dev/sda1 rw',
|
|
'2 1 0:1 / /tmp rw - tmpfs tmpfs rw',
|
|
'3 1 0:2 / /mnt/cloud rw - nfs4 nas:/share rw',
|
|
'4 3 0:3 / /mnt/cloud/local rw - tmpfs tmpfs rw']
|
|
for path in ('/var/lib/backups','/tmp/backups','/mnt/cloud/local/backups'):
|
|
self.assertFalse(Backups.linux_network_mount(PurePosixPath(path),records))
|
|
self.assertTrue(Backups.linux_network_mount(PurePosixPath('/mnt/cloud/backups'),records))
|
|
|
|
def test_network_mount_paths_with_spaces_and_supported_fuse_types(self):
|
|
for filesystem in ('cifs','fuse.rclone','fuse.davfs','fuse.sshfs'):
|
|
records=['1 0 8:1 / / rw - ext4 /dev/sda1 rw',
|
|
'2 1 0:2 / /mnt/cloud\\040drive rw - '+filesystem+' remote rw']
|
|
self.assertTrue(Backups.linux_network_mount(PurePosixPath('/mnt/cloud drive/backups'),records))
|
|
self.assertFalse(Backups.linux_network_mount(PurePosixPath('/tmp'),['invalid data']))
|
|
|
|
def test_config_read_failure_always_releases_running_guard(self):
|
|
backup=self.source_file();backup.running=True
|
|
with patch.object(backup,'config',side_effect=RuntimeError('配置读取失败')):
|
|
result=backup.run()
|
|
self.assertFalse(backup.running);self.assertIn('配置读取失败',result['error'])
|
|
|
|
def test_baidu_success_is_separate_from_network_mount_and_requires_opt_in(self):
|
|
backup=self.source_file()
|
|
with patch.object(app.BAIDU,'upload',return_value={'fs_id':123}) as upload, patch.object(app.BAIDU,'public_config',return_value={'appFolder':'测试'}), patch.object(app.BAIDU,'verify_backup'), patch.object(app.BAIDU,'list_directory',return_value=[]):
|
|
result=backup.run();upload.assert_not_called();self.assertFalse(result['baiduOk'])
|
|
app.set_setting('backup_config',dict(backup.config(),baidu=True))
|
|
result=backup.run();self.assertTrue(result['baiduOk']);self.assertFalse(result['networkOk']);self.assertEqual(result['baiduFileId'],'123')
|
|
|
|
def test_cloud_cleanup_only_runs_after_verified_upload(self):
|
|
backup=self.source_file();app.set_setting('backup_config',dict(backup.config(),baidu=True))
|
|
with patch.object(app.BAIDU,'public_config',return_value={'appFolder':'测试'}), patch.object(app.BAIDU,'upload',return_value={'fs_id':123}), patch.object(backup,'prune_cloud') as prune:
|
|
with patch.object(app.BAIDU,'verify_backup',side_effect=RuntimeError('hash mismatch')):
|
|
result=backup.run();self.assertFalse(result['baiduOk']);prune.assert_not_called()
|
|
with patch.object(app.BAIDU,'verify_backup'):
|
|
result=backup.run();self.assertTrue(result['baiduOk']);prune.assert_called_once()
|
|
|
|
def test_retention_is_a_calendar_month_including_leap_and_year_boundaries(self):
|
|
for value, expected in [('2024-03-31','2024-02-29'),('2026-03-31','2026-02-28'),('2026-01-04','2025-12-04')]:
|
|
self.assertEqual(Backups.month_before(dt.datetime.fromisoformat(value)).date().isoformat(),expected)
|
|
|
|
def test_cloud_prune_limits_to_instance_and_expired_database_files(self):
|
|
backup=Backups(app);instance='a'*16;root='/apps/测试/数据库备份/'+instance
|
|
def name(stamp):return 'vision-'+instance+'-'+stamp+'.sqlite'
|
|
old=name('20260804T120000000000Z');nested=name('20260805T120000000000Z');recent=name('20261004T120000000000Z');boundary=name('20260904T120000000000Z')
|
|
rows=[dict(path=root+'/'+n,isdir=0) for n in (old,recent,boundary,'holiday.png','../'+old)]
|
|
rows+=[dict(path=root+'/'+nested,isdir=1),dict(path='/apps/其他/'+old,isdir=0)]
|
|
def listing(path):return rows if path==root else [dict(path=path+'/'+nested,isdir=0),dict(path=path+'/private.png',isdir=0)]
|
|
with patch.object(app.BAIDU,'public_config',return_value={'appFolder':'测试'}),patch.object(app.BAIDU,'list_directory',side_effect=listing),patch.object(app.BAIDU,'delete_backup_file') as delete:
|
|
result={};backup.prune_cloud(instance,root+'/'+recent,result,dt.datetime(2026,10,4,12,tzinfo=dt.timezone.utc))
|
|
self.assertEqual([c.args[0] for c in delete.call_args_list],[root+'/'+old,root+'/'+nested+'/'+nested]);self.assertEqual(result['cloudDeleted'],2)
|
|
|
|
def test_cloud_cleanup_failure_keeps_successful_new_backup(self):
|
|
backup=self.source_file();app.set_setting('backup_config',dict(backup.config(),baidu=True))
|
|
with patch.object(app.BAIDU,'public_config',return_value={'appFolder':'测试'}),patch.object(app.BAIDU,'upload',return_value={'fs_id':123}),patch.object(app.BAIDU,'verify_backup'),patch.object(backup,'prune_cloud',side_effect=RuntimeError('cleanup unavailable')):
|
|
result=backup.run()
|
|
self.assertTrue(result['baiduOk']);self.assertFalse(result['cloudCleanupOk']);self.assertNotIn('error',result)
|