Prepare MariaDB restore tests and recover backup disk space safely

tests/test_space.py

@@ -0,0 +1,81 @@
+"""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())