83 lines
4.4 KiB
Python
83 lines
4.4 KiB
Python
"""Real rename/fsync preservation tests; not a privileged migration acceptance."""
|
|
import errno
|
|
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]))
|
|
import c01_migration_layout as layout
|
|
|
|
|
|
class LayoutTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temporary=tempfile.TemporaryDirectory();self.addCleanup(self.temporary.cleanup)
|
|
base=Path(self.temporary.name);self.source=base/'old';self.target=base/'private'
|
|
self.source.mkdir();self.target.mkdir();(self.source/'.private').mkdir()
|
|
for name in ('identity','federation','lnd'):
|
|
(self.source/name).mkdir();(self.source/name/'retained').write_bytes(('opaque '+name).encode())
|
|
self.plan={'source_root':str(self.source),'target_root':str(self.target),
|
|
'private_parent':{'path':str(self.source/'.private')},'entries':[]}
|
|
for name in ('identity','federation','lnd'):
|
|
private=name!='lnd';path=self.source/name
|
|
self.plan['entries'].append({'name':name,'disposition':'private' if private else 'runtime',
|
|
'identity':layout.identity(path.stat()),
|
|
'mode_after':path.stat().st_mode & 0o7777,
|
|
'tree_metadata_sha256':layout.private_metadata(path) if private else None})
|
|
self.source_fd=os.open(self.source,os.O_RDONLY|os.O_DIRECTORY)
|
|
self.target_fd=os.open(self.target,os.O_RDONLY|os.O_DIRECTORY)
|
|
self.addCleanup(os.close,self.source_fd);self.addCleanup(os.close,self.target_fd)
|
|
|
|
def move(self,resume=False):return layout.relocate_entries(self.plan,self.source_fd,self.target_fd,resume)
|
|
|
|
def test_private_moves_preserve_inodes_and_runtime_tree_remains(self):
|
|
before={n:(self.source/n/'retained').stat() for n in ('identity','federation','lnd')}
|
|
self.move()
|
|
for name in ('identity','federation','lnd'):
|
|
current=(self.source if name=='lnd' else self.target)/name/'retained'
|
|
after=current.stat()
|
|
self.assertEqual((before[name].st_dev,before[name].st_ino,before[name].st_uid,before[name].st_gid,before[name].st_mode),
|
|
(after.st_dev,after.st_ino,after.st_uid,after.st_gid,after.st_mode))
|
|
self.assertEqual(current.read_bytes(),('opaque '+name).encode())
|
|
self.assertFalse((self.source/'identity').exists())
|
|
|
|
def test_interrupted_move_resumes_without_recopy_or_reset(self):
|
|
original=layout.LIBC.renameat2;calls=[]
|
|
def interrupted(*args):
|
|
calls.append(args)
|
|
if len(calls)==2:raise OSError(errno.EIO,'injected interruption before second move')
|
|
return original(*args)
|
|
with patch.object(layout.LIBC,'renameat2',side_effect=interrupted):
|
|
with self.assertRaises(OSError):self.move()
|
|
retained=(self.target/'identity/retained').stat().st_ino
|
|
self.move(resume=True)
|
|
self.assertEqual((self.target/'identity/retained').stat().st_ino,retained)
|
|
|
|
def test_no_overwrite_and_ambiguous_resume_refused(self):
|
|
(self.target/'identity').mkdir();(self.target/'identity/keep').write_bytes(b'do not replace')
|
|
with self.assertRaises(OSError):self.move()
|
|
self.assertEqual((self.target/'identity/keep').read_bytes(),b'do not replace')
|
|
with self.assertRaises(ValueError):self.move(resume=True)
|
|
|
|
def test_external_hard_link_and_symlink_cannot_escape_private_custody(self):
|
|
path=self.source/'identity/retained';alias=Path(self.temporary.name)/'outside'
|
|
os.link(path,alias)
|
|
with self.assertRaisesRegex(ValueError,'hard-link'):layout.private_metadata(path.parent)
|
|
alias.unlink();alias.symlink_to(path)
|
|
with self.assertRaises(ValueError):layout.private_metadata(alias)
|
|
|
|
def test_identity_directory_is_hardened_without_changing_children_or_ownership(self):
|
|
entry=self.plan['entries'][0];path=self.source/'identity'
|
|
path.chmod(0o755);entry['identity']=layout.identity(path.stat());entry['mode_after']=0o700
|
|
entry['tree_metadata_sha256']=layout.private_metadata(path)
|
|
before=(path/'retained').stat()
|
|
self.move()
|
|
self.assertEqual((self.target/'identity').stat().st_mode & 0o7777,0o700)
|
|
self.assertEqual(layout.identity((self.target/'identity/retained').stat()),layout.identity(before))
|
|
self.move(resume=True)
|
|
|
|
|
|
if __name__=='__main__':unittest.main()
|