"""Exercise the shipped helper against real Linux descriptors and metadata.

Only account-home lookup is redirected into an isolated test directory. No SSH,
sudo, or external server is involved in this filesystem integration fixture.
"""
import base64
import contextlib
import errno
import hashlib
import io
import json
import os
import pwd
import stat
import struct
import sys
import tempfile
from types import SimpleNamespace
from unittest.mock import patch

source = open(sys.argv[1], encoding='utf-8').read()
scenario = sys.argv[2]

test_parent = pwd.getpwuid(os.getuid()).pw_dir if os.geteuid() == 0 else None
with tempfile.TemporaryDirectory(prefix='chj-editor-test-', dir=test_parent) as root, contextlib.ExitStack() as stack:
    home = root + '/actual home'
    if scenario == 'cross-filesystem-strategy':
        if not os.path.isdir('/dev/shm') or os.stat('/dev/shm').st_dev == os.stat(root).st_dev:
            print('No separate writable filesystem available')
            sys.exit(77)
        home = stack.enter_context(tempfile.TemporaryDirectory(prefix='chj-editor-home-', dir='/dev/shm')) + '/actual home'
    os.mkdir(home, 0o700)
    parent = root + '/target directory'
    os.mkdir(parent, 0o700)
    path = parent + '/config\' ; $(touch unsafe) `test` ü'
    original = b'original\n'
    modified = b'modified\n'
    with open(path, 'wb') as out:
        out.write(original)
    os.chmod(path, 0o6750)

    def command(request):
        captured = io.StringIO()
        with patch.object(pwd, 'getpwuid', return_value=SimpleNamespace(pw_dir=home)), patch.object(sys, 'argv', ['remoteEditor.py', json.dumps(request)]), contextlib.redirect_stdout(captured):
            exec(compile(source, 'remoteEditor.py', 'exec'), {})
        return json.loads(captured.getvalue())

    def value(request):
        result = command(request)
        assert result['ok'], result
        return result['value']

    baseline = value(dict(operation='read', path=path))['baseline']
    ident = 'a' * 32
    digest = hashlib.sha256(modified).hexdigest()
    request = dict(operation='finalize', id=ident, uid=os.getuid(), path=path, baseline=baseline, hash=digest)

    def stage(data=modified):
        prepared = value(dict(operation='prepare', id=ident, path=path, baseline=request['baseline'], hash=digest))
        fd = os.open(prepared['temporary'], os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
        with os.fdopen(fd, 'wb') as out:
            out.write(data)
        assert stat.S_IMODE(os.stat(home + '/ch-j-sm').st_mode) == 0o700
        assert stat.S_IMODE(os.stat(prepared['temporary']).st_mode) == 0o600
        return prepared['temporary']

    def intact(staged):
        assert open(path, 'rb').read() == original
        assert open(staged, 'rb').read() == modified

    if scenario == 'metadata':
        os.setxattr(path, 'user.chj-test', b'attribute')
        request['baseline'] = value(dict(operation='read', path=path))['baseline']
        staged = stage()
        result = value(request)
        final = os.stat(path)
        assert result['status'] == 'saved'
        assert final.st_uid == baseline['uid'] and final.st_gid == baseline['gid']
        assert stat.S_IMODE(final.st_mode) == 0o6750
        assert os.getxattr(path, 'user.chj-test') == b'attribute'
        assert open(path, 'rb').read() == modified
        assert open(staged, 'rb').read() == modified
        assert final.st_mtime_ns != int(baseline['mtimeNs'])
        assert value(dict(operation='inspect', id=ident))['status'] == 'confirmed'
        value(dict(operation='cleanup', id=ident))
        assert not os.path.exists(staged)
        assert not os.listdir(home + '/ch-j-sm')
    elif scenario == 'move-failure':
        staged = stage()
        with patch.object(os, 'replace', side_effect=OSError('failed rename')):
            assert not command(request)['ok']
        intact(staged)
        assert os.listdir(parent) == [os.path.basename(path)]
    elif scenario == 'disk-full':
        staged = stage()
        with patch.object(os, 'fsync', side_effect=OSError('disk full')):
            assert not command(request)['ok']
        intact(staged)
    elif scenario == 'post-rename-disconnect':
        staged = stage()
        real_replace = os.replace

        def replace(*args, **kwargs):
            real_replace(*args, **kwargs)
            # Failure to confirm after the actual rename must be reported unknown.
        real_fsync = os.fsync
        count = [0]

        def fsync(fd):
            count[0] += 1
            if count[0] == 3:
                raise OSError('lost completion')
            return real_fsync(fd)
        with patch.object(os, 'replace', replace), patch.object(os, 'fsync', fsync):
            result = command(request)
        assert result['code'] == 'FILE_RESULT_UNKNOWN', result
        assert open(path, 'rb').read() == modified
        assert open(staged, 'rb').read() == modified
        assert value(dict(operation='list'))['items'][0]['status'] == 'confirmed'
    elif scenario == 'conflict':
        staged = stage()
        with open(path, 'wb') as out:
            out.write(b'external\n')
        assert command(request)['code'] == 'FILE_CONFLICT'
        assert open(path, 'rb').read() == b'external\n'
        assert open(staged, 'rb').read() == modified
        assert value(dict(operation='inspect', id=ident))['status'] == 'conflict'
        assert command(dict(operation='cleanup', id=ident))['code'] == 'FILE_RESULT_UNCONFIRMED'
        assert os.path.exists(staged)
    elif scenario == 'upload-incomplete':
        staged = stage(b'part')
        assert command(request)['code'] == 'FILE_UPLOAD_INCOMPLETE'
        assert open(path, 'rb').read() == original
        assert value(dict(operation='list'))['items'][0]['status'] == 'upload-incomplete'
    elif scenario == 'existing-temporary':
        staged = stage()
        assert not command(dict(operation='prepare', id=ident, path=path, baseline=baseline, hash=digest))['ok']
        intact(staged)
    elif scenario in ('target-hardlink', 'target-fifo'):
        os.unlink(path)
        other = parent + '/other'
        with open(other, 'wb') as out:
            out.write(original)
        if scenario == 'target-hardlink':
            os.link(other, path)
            assert command(dict(operation='read', path=path))['code'] == 'FILE_HARDLINK_UNSUPPORTED'
        else:
            os.mkfifo(path)
            assert not command(dict(operation='read', path=path))['ok']
        assert open(other, 'rb').read() == original
    elif scenario in ('target-symlink', 'symlink-retarget'):
        # sites-enabled/x.conf -> ../sites-available/x.conf: edit the target, keep the link.
        available = root + '/sites-available'
        os.mkdir(available, 0o755)
        real = available + '/example.conf'
        with open(real, 'wb') as out:
            out.write(original)
        os.unlink(path)
        os.symlink('../sites-available/example.conf', path)
        opened = value(dict(operation='read', path=path))
        assert opened['link'] is True and opened['resolvedPath'] == real
        assert base64.b64decode(opened['data']) == original
        request['baseline'] = opened['baseline']
        request['resolvedPath'] = opened['resolvedPath']
        staged = stage()
        if scenario == 'symlink-retarget':
            other = available + '/other.conf'
            with open(other, 'wb') as out:
                out.write(original)
            os.unlink(path)
            os.symlink(other, path)
            assert command(request)['code'] == 'FILE_CONFLICT'
            assert open(real, 'rb').read() == original
        else:
            result = value(request)
            assert result['status'] == 'saved' and result['resolvedPath'] == real
            assert os.path.islink(path) and os.readlink(path) == '../sites-available/example.conf'
            assert open(real, 'rb').read() == modified
            assert sorted(os.listdir(available)) == ['example.conf']
    elif scenario == 'directory-symlink':
        os.symlink(parent, root + '/linked-dir')
        via = root + '/linked-dir/' + os.path.basename(path)
        opened = value(dict(operation='read', path=via))
        assert opened['link'] is True and opened['resolvedPath'] == path
        request.update(path=via, baseline=opened['baseline'], resolvedPath=opened['resolvedPath'])
        stage()
        assert value(request)['status'] == 'saved'
        assert open(path, 'rb').read() == modified and os.path.islink(root + '/linked-dir')
    elif scenario == 'no-space':
        staged = stage()
        real_fdopen = os.fdopen

        class Full:
            def __init__(self, fd, *args, **kwargs):
                self.inner = real_fdopen(fd, *args, **kwargs)
            def write(self, data):
                raise OSError(errno.ENOSPC, 'No space left on device')
            def __getattr__(self, name):
                return getattr(self.inner, name)
            def __enter__(self):
                return self
            def __exit__(self, *args):
                self.inner.close()
        with patch.object(os, 'fdopen', Full):
            assert command(request)['code'] == 'FILE_NO_SPACE'
        intact(staged)
        assert os.listdir(parent) == [os.path.basename(path)]
    elif scenario == 'xattr-unsupported':
        with patch.object(os, 'listxattr', side_effect=OSError(errno.ENOTSUP, 'Operation not supported')):
            opened = value(dict(operation='read', path=path))
            assert opened['baseline']['xattrs'] == {}
            request['baseline'] = opened['baseline']
            stage()
            assert value(request)['status'] == 'saved'
        assert open(path, 'rb').read() == modified
    elif scenario == 'trusted-xattr':
        # Root-only attributes (XFS ACL mirror, overlay metadata) are missing from a baseline captured by
        # the SSH account; saving with sudo must neither report a conflict nor drop them.
        if os.geteuid() != 0:
            print('trusted.* attributes require root')
            sys.exit(77)
        try:
            os.setxattr(path, 'trusted.chj-test', b'root-only')
        except PermissionError:
            print('trusted.* needs CAP_SYS_ADMIN (not available in this container)')
            sys.exit(77)
        full = value(dict(operation='read', path=path))['baseline']
        user_view = dict(full, xattrs={k: v for k, v in full['xattrs'].items() if not k.startswith('trusted.')})
        request['baseline'] = user_view
        stage()
        assert value(request)['status'] == 'saved'
        assert os.getxattr(path, 'trusted.chj-test') == b'root-only'
        assert open(path, 'rb').read() == modified
    elif scenario == 'backup':
        stage()
        result = value(dict(request, backup=True))
        assert open(result['backupPath'], 'rb').read() == original
        assert stat.S_IMODE(os.stat(result['backupPath']).st_mode) == 0o600
        assert open(path, 'rb').read() == modified
    elif scenario.startswith('publish-') or scenario == 'stage-out':
        payload = os.urandom(3 * 1024 * 1024 + 17)
        digest = hashlib.sha256(payload).hexdigest()
        os.setxattr(path, 'user.chj-test', b'kept')
        os.chmod(path, 0o640)
        uid = os.getuid()
        target_stat = value(dict(operation='stat', path=path))
        assert target_stat['exists'] and target_stat['type'] == 'file'
        direct = parent + '/.part-direct'
        fd = os.open(direct, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
        with os.fdopen(fd, 'wb') as out:
            out.write(payload)
        publish = dict(operation='publish', source=direct, uid=uid, path=path, hash=digest, size=len(payload),
                       baseline=target_stat['baseline'], strategy='rename')
        if scenario == 'publish-existing-rename':
            result = value(publish)
            assert result['status'] == 'published' and result['metadataPreserved'] and not result['created']
            assert open(path, 'rb').read() == payload
            assert os.getxattr(path, 'user.chj-test') == b'kept' and stat.S_IMODE(os.stat(path).st_mode) == 0o640
            assert not os.path.exists(direct)
        elif scenario == 'publish-new-rename':
            fresh = parent + '/fresh.conf'
            result = value(dict(publish, path=fresh, baseline=None, mode=0o640))
            assert result['created'] and open(fresh, 'rb').read() == payload
            assert stat.S_IMODE(os.stat(fresh).st_mode) == 0o640 and not os.path.exists(direct)
        elif scenario == 'publish-copy':
            ws = value(dict(operation='transfer-workspace'))['path']
            staged = ws + '/' + 'b' * 32 + '.part'
            fd = os.open(staged, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
            with os.fdopen(fd, 'wb') as out:
                out.write(payload)
            nested = root + '/new parent/deeper/file.bin'
            created = value(dict(publish, source=staged, path=nested, baseline=None, strategy='copy', parents=True))
            assert created['created'] and created['sourceRemoved'] and open(nested, 'rb').read() == payload
            assert not os.path.exists(staged)
            assert [n for n in os.listdir(os.path.dirname(nested))] == ['file.bin']
            fd = os.open(staged, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
            with os.fdopen(fd, 'wb') as out:
                out.write(payload)
            replaced = value(dict(publish, source=staged, strategy='copy'))
            assert replaced['metadataPreserved'] and open(path, 'rb').read() == payload
            assert os.getxattr(path, 'user.chj-test') == b'kept'
        elif scenario == 'publish-conflict':
            with open(path, 'ab') as out:
                out.write(b'external')
            assert command(publish)['code'] == 'FILE_CONFLICT'
            assert open(path, 'rb').read() == original + b'external' and os.path.exists(direct)
        elif scenario == 'publish-verify':
            assert command(dict(publish, hash='0' * 64))['code'] == 'FILE_VERIFY_FAILED'
            assert open(path, 'rb').read() == original
        elif scenario == 'publish-exists':
            assert command(dict(publish, baseline=None))['code'] == 'FILE_EXISTS'
            assert value(dict(publish, baseline=None, overwrite=True))['status'] == 'published'
        elif scenario == 'publish-retry':
            value(publish)
            again = value(dict(publish, retry=True))
            assert again['alreadyPublished'] is True
        elif scenario == 'stage-out':
            with open(path, 'wb') as out:
                out.write(payload)
            staged = value(dict(operation='stage-out', path=path, id='c' * 32, uid=uid))
            assert staged['hash'] == digest and staged['size'] == len(payload)
            assert open(staged['path'], 'rb').read() == payload
            assert stat.S_IMODE(os.stat(staged['path']).st_mode) == 0o600
            listed = value(dict(operation='transfer-list'))['items']
            assert [item['id'] for item in listed] == ['c' * 32]
            assert value(dict(operation='hash', path=staged['path']))['hash'] == digest
            assert value(dict(operation='transfer-remove', id='c' * 32))['removed'] is True
            assert not os.path.exists(staged['path'])
        else:
            raise AssertionError('Unknown scenario: ' + scenario)
    elif scenario == 'file-ops':
        work = root + '/ops'
        value(dict(operation='mkdir', path=work + '/a/b', parents=True))
        value(dict(operation='create', path=work + '/a/b/new.txt', mode=0o600))
        assert stat.S_IMODE(os.stat(work + '/a/b/new.txt').st_mode) == 0o600
        assert command(dict(operation='create', path=work + '/a/b/new.txt'))['code'] == 'FILE_EXISTS'
        os.symlink('b', work + '/a/link')
        listed = value(dict(operation='listdir', path=work + '/a'))['entries']
        assert {e['name']: (e['type'], e.get('targetType')) for e in listed} == {'b': ('directory', None), 'link': ('symlink', 'directory')}
        value(dict(operation='create', path=work + '/a/other.txt'))
        assert command(dict(operation='rename', **{'from': work + '/a/b/new.txt', 'to': work + '/a/other.txt'}))['code'] == 'FILE_EXISTS'
        value(dict(operation='rename', **{'from': work + '/a/b/new.txt', 'to': work + '/a/renamed.txt'}))
        assert os.path.exists(work + '/a/renamed.txt')
        value(dict(operation='remove', path=work + '/a/link'))
        assert os.path.isdir(work + '/a/b') and not os.path.lexists(work + '/a/link')
        assert command(dict(operation='remove', path=work + '/a'))['code'] == 'FILE_DIRECTORY_NOT_EMPTY'
        removed = value(dict(operation='remove', path=work + '/a', recursive=True))
        assert removed['entries'] >= 3 and not os.path.exists(work + '/a')
    elif scenario == 'workspace-symlink':
        os.symlink(parent, home + '/ch-j-sm')
        assert not command(dict(operation='prepare', id=ident, path=path, baseline=baseline, hash=digest))['ok']
        assert os.listdir(parent) == [os.path.basename(path)]
    elif scenario == 'temporary-symlink':
        staged = stage()
        os.unlink(staged)
        os.symlink(path, staged)
        assert not command(request)['ok']
        assert open(path, 'rb').read() == original
    elif scenario == 'metadata-failure':
        os.setxattr(path, 'user.chj-test', b'attribute')
        request['baseline'] = value(dict(operation='read', path=path))['baseline']
        staged = stage()
        with patch.object(os, 'setxattr', side_effect=PermissionError('metadata denied')):
            assert not command(request)['ok']
        intact(staged)
    elif scenario == 'acl':
        # Linux POSIX ACL xattr format: version, followed by tag/perm/uid entries.
        acl = struct.pack('<I', 2) + b''.join(struct.pack('<HHI', tag, perm, who) for tag, perm, who in [
            (1, 7, 0xffffffff), (2, 4, os.getuid() + 1), (4, 5, 0xffffffff), (16, 5, 0xffffffff), (32, 0, 0xffffffff)])
        try:
            os.setxattr(path, 'system.posix_acl_access', acl)
        except OSError:
            print('POSIX ACLs are unavailable on the test filesystem')
            sys.exit(77)
        request['baseline'] = value(dict(operation='read', path=path))['baseline']
        staged = stage()
        assert value(request)['status'] == 'saved'
        assert os.getxattr(path, 'system.posix_acl_access') == acl
        assert open(staged, 'rb').read() == modified
    elif scenario == 'concurrent-target-replacement':
        staged = stage()
        real_fsync = os.fsync
        count = [0]

        def fsync(fd):
            count[0] += 1
            if count[0] == 2:
                replacement = parent + '/external'
                with open(replacement, 'wb') as out:
                    out.write(b'external')
                os.replace(replacement, path)
            return real_fsync(fd)
        with patch.object(os, 'fsync', fsync):
            assert command(request)['code'] == 'FILE_CONFLICT'
        assert open(path, 'rb').read() == b'external'
        assert open(staged, 'rb').read() == modified
    elif scenario == 'cross-filesystem-strategy':
        staged = stage()
        assert os.stat(staged).st_dev != os.stat(path).st_dev
        real_replace = os.replace

        def replace(src, dst, **kwargs):
            assert src != staged and '/' not in src
            assert kwargs['src_dir_fd'] == kwargs['dst_dir_fd']
            return real_replace(src, dst, **kwargs)
        with patch.object(os, 'replace', replace):
            assert value(request)['status'] == 'saved'
        assert os.path.exists(staged)
    else:
        raise AssertionError('Unknown scenario: ' + scenario)

print('ok')