"""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()