82 lines
3.2 KiB
Python
82 lines
3.2 KiB
Python
"""Recovery policy tests: reclaim only owned scratch, protect recovery/last copies."""
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]/'lib'))
|
|
import space
|
|
|
|
|
|
@unittest.skipUnless(os.geteuid() == 0, 'root-only cleanup policy')
|
|
class SpaceRecovery(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.work = Path(self.temp.name)
|
|
self.staging = self.work/'staging'
|
|
self.archives = self.work/'archives'
|
|
self.restore = self.work/'restore'
|
|
for p in (self.staging, self.archives, self.restore): p.mkdir(mode=0o700)
|
|
self.run = 'appbackup_2026-09-12_17-07-56_bd1f6193'
|
|
self.stage = self.staging/(self.run+'.5ICERN')
|
|
self.stage.mkdir(mode=0o700)
|
|
(self.stage/'meta').mkdir()
|
|
(self.stage/'meta/run-id.txt').write_text(self.run+'\n')
|
|
(self.stage/'meta/format.txt').write_text('format=2\ncompressor=zstd\n')
|
|
self.partial = self.archives/('.partial-'+self.run)
|
|
self.partial.mkdir(mode=0o700)
|
|
(self.partial/'meta.tar.zst').write_text('incomplete scratch')
|
|
|
|
def clean(self):
|
|
space.clean(self.work, self.staging, self.archives, 'appbackup', self.restore)
|
|
|
|
def test_gemini_failed_compression_reclaims_both_trees(self):
|
|
unknown = self.staging/'keep-manual-files'
|
|
unknown.mkdir(mode=0o700)
|
|
self.clean()
|
|
self.assertFalse(self.stage.exists())
|
|
self.assertFalse(self.partial.exists())
|
|
self.assertTrue(unknown.exists())
|
|
|
|
def test_recovery_journal_blocks_all_cleanup(self):
|
|
(self.work/('recovery-'+self.run+'.json')).write_text('{}')
|
|
with self.assertRaisesRegex(ValueError, 'recovery journal'): self.clean()
|
|
self.assertTrue(self.stage.exists())
|
|
self.assertTrue(self.partial.exists())
|
|
|
|
def test_complete_partial_is_preserved(self):
|
|
(self.partial/'COMPLETE').write_text('possibly only complete snapshot')
|
|
self.clean()
|
|
self.assertTrue(self.stage.exists())
|
|
self.assertTrue(self.partial.exists())
|
|
|
|
def test_nested_symlink_does_not_delete_target(self):
|
|
target = self.work/'production'
|
|
target.mkdir()
|
|
(target/'data').write_text('live')
|
|
(self.stage/'link').symlink_to(target)
|
|
self.clean()
|
|
self.assertEqual((target/'data').read_text(), 'live')
|
|
|
|
def test_mount_is_never_removed(self):
|
|
mount = self.stage/'mounted'
|
|
mount.mkdir()
|
|
with patch('space.os.path.ismount', side_effect=lambda p: p == mount):
|
|
with self.assertRaisesRegex(ValueError, 'mounted'): self.clean()
|
|
self.assertTrue(mount.exists())
|
|
|
|
def test_verification_download_recovered_manual_restore_kept(self):
|
|
verify = self.restore/(self.run+'.abcdef')
|
|
verify.mkdir(mode=0o700)
|
|
(verify/'.verify-only').write_text(self.run+'\n')
|
|
manual = self.restore/(self.run+'.ghijkl')
|
|
manual.mkdir(mode=0o700)
|
|
self.clean()
|
|
self.assertFalse(verify.exists())
|
|
self.assertTrue(manual.exists())
|