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']))