ucvl-home-vision/test_backups.py

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)