80 lines
4.1 KiB
Python
80 lines
4.1 KiB
Python
import importlib.util
|
|
from pathlib import Path
|
|
import shutil
|
|
import tempfile
|
|
import unittest
|
|
from unittest import mock
|
|
|
|
spec = importlib.util.spec_from_file_location('repair', Path(__file__).resolve().parents[1] / 'operations/gemini_backup_repair.py')
|
|
m = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(m)
|
|
|
|
class PreservationTests(unittest.TestCase):
|
|
def test_live_budget_preserves_file_metadata_sql_and_growth_allowances(self):
|
|
raw=100 + 2*8192 + 200*3 + 67108864
|
|
self.assertEqual(m.snapshot_budget([(100,2)],[200]),raw*12//10+67108864)
|
|
with self.assertRaises(ValueError):m.snapshot_budget([(-1,2)],[])
|
|
with self.assertRaises(ValueError):m.snapshot_budget([],[-1])
|
|
|
|
def test_plan_keeps_root_reserve_and_stops_when_var_target_is_met(self):
|
|
self.assertEqual(m.choose_moves([('a', 48), ('b', 7), ('c', 7), ('d', 7)], 26, 77, 84, 12), [('a',48),('b',7),('c',7)])
|
|
with self.assertRaises(RuntimeError):
|
|
m.choose_moves([('a', 70)], 26, 77, 84, 12)
|
|
with self.assertRaises(RuntimeError):
|
|
m.choose_moves([('a', 5)], 26, 77, 84, 12)
|
|
|
|
def test_source_parent_child_overlap_is_rejected(self):
|
|
m.disjoint('/home/.staging', ['/home/johannes'])
|
|
for path,sources in [('/home/.staging',['/home']),('/srv/history',['/srv/history/source'])]:
|
|
with self.assertRaises(RuntimeError):
|
|
m.disjoint(path,sources)
|
|
|
|
def test_symlink_and_hardlink_are_not_accepted_as_archive_data(self):
|
|
with tempfile.TemporaryDirectory() as t:
|
|
root=Path(t);source=root/'source';source.write_bytes(b'data')
|
|
link=root/'link';link.symlink_to(source)
|
|
with self.assertRaises(RuntimeError):m.inventory(link)
|
|
link.unlink();link.hardlink_to(source)
|
|
with self.assertRaises(RuntimeError):m.inventory(source)
|
|
|
|
@unittest.skipUnless(shutil.which('rsync'), 'rsync required')
|
|
def test_copy_is_verified_before_original_is_removed(self):
|
|
with tempfile.TemporaryDirectory() as t:
|
|
root=Path(t);source=root/'archive';source.mkdir();(source/'data').write_bytes(b'saved backup')
|
|
state=root/'state';state.mkdir();dest=root/'preserved';dest.mkdir()
|
|
m.migrate(source,dest/'archive',state)
|
|
self.assertFalse(source.exists())
|
|
self.assertEqual((dest/'archive/data').read_bytes(),b'saved backup')
|
|
|
|
@unittest.skipUnless(shutil.which('rsync'), 'rsync required')
|
|
def test_flat_archive_is_preserved_with_its_original_filename(self):
|
|
with tempfile.TemporaryDirectory() as t:
|
|
root=Path(t);src=root/'old.tar.zst';src.write_bytes(b'old backup')
|
|
state=root/'state';state.mkdir();dstroot=root/'preserved';dstroot.mkdir()
|
|
m.migrate(src,dstroot/src.name,state)
|
|
self.assertFalse(src.exists())
|
|
self.assertEqual((dstroot/src.name).read_bytes(),b'old backup')
|
|
|
|
@unittest.skipUnless(shutil.which('rsync'), 'rsync required')
|
|
def test_corrupt_destination_keeps_original(self):
|
|
with tempfile.TemporaryDirectory() as t:
|
|
root=Path(t);src=root/'source';src.write_bytes(b'original backup')
|
|
state=root/'state';state.mkdir();dst=root/'destination'
|
|
real=m.checked
|
|
def corrupt_after_copy(args, **kw):
|
|
result=real(args, **kw)
|
|
if '--fsync' in args:dst.write_bytes(b'corrupted')
|
|
return result
|
|
with mock.patch.object(m,'checked',side_effect=corrupt_after_copy):
|
|
with self.assertRaises(RuntimeError):m.migrate(src,dst,state)
|
|
self.assertEqual(src.read_bytes(),b'original backup')
|
|
|
|
def test_unrecognized_destination_never_removes_source(self):
|
|
with tempfile.TemporaryDirectory() as t:
|
|
root=Path(t);src=root/'source';src.write_bytes(b'original');dst=root/'destination';dst.write_bytes(b'unrelated')
|
|
state=root/'state';state.mkdir()
|
|
with self.assertRaises(RuntimeError):m.migrate(src,dst,state)
|
|
self.assertEqual(src.read_bytes(),b'original')
|
|
|
|
if __name__ == '__main__':unittest.main()
|