mergerfs/tests/TEST_posix_readdir

#!/usr/bin/env python3

import ctypes
import os
import sys
import tempfile

from posix_parity import cleanup_dir
from posix_parity import compare_calls
from posix_parity import fail
from posix_parity import join
from posix_parity import mergerfs_mount
from posix_parity import temp_dir
from posix_parity import touch


class Dirent(ctypes.Structure):
    _fields_ = [
        ("d_ino", ctypes.c_ulong),
        ("d_off", ctypes.c_long),
        ("d_reclen", ctypes.c_ushort),
        ("d_type", ctypes.c_ubyte),
        ("d_name", ctypes.c_char * 256),
    ]


libc = ctypes.CDLL(None, use_errno=True)
libc.opendir.argtypes = [ctypes.c_char_p]
libc.opendir.restype = ctypes.c_void_p
libc.readdir.argtypes = [ctypes.c_void_p]
libc.readdir.restype = ctypes.POINTER(Dirent)
libc.telldir.argtypes = [ctypes.c_void_p]
libc.telldir.restype = ctypes.c_long
libc.seekdir.argtypes = [ctypes.c_void_p, ctypes.c_long]
libc.seekdir.restype = None
libc.rewinddir.argtypes = [ctypes.c_void_p]
libc.rewinddir.restype = None
libc.closedir.argtypes = [ctypes.c_void_p]
libc.closedir.restype = ctypes.c_int


def opendir_or_raise(path):
    ctypes.set_errno(0)
    dp = libc.opendir(path.encode())
    if not dp:
        err = ctypes.get_errno()
        raise OSError(err, os.strerror(err), path)
    return dp


def decode_name(ent):
    return bytes(ent.d_name).split(b"\0", 1)[0].decode("utf-8", errors="surrogateescape")


def read_all_names(dp):
    names = []
    while True:
        entp = libc.readdir(dp)
        if not entp:
            break
        names.append(decode_name(entp.contents))
    return names


def next_non_dot(dp):
    while True:
        entp = libc.readdir(dp)
        if not entp:
            return None
        name = decode_name(entp.contents)
        if name not in (".", ".."):
            return name


def names_set(path):
    dp = opendir_or_raise(path)
    try:
        return set(n for n in read_all_names(dp) if n not in (".", ".."))
    finally:
        libc.closedir(dp)


def seek_tell_consistent(path):
    dp = opendir_or_raise(path)
    try:
        _first = next_non_dot(dp)
        if _first is None:
            return True

        pos = libc.telldir(dp)
        second = next_non_dot(dp)
        if second is None:
            return True

        libc.seekdir(dp, pos)
        again = next_non_dot(dp)
        return second == again
    finally:
        libc.closedir(dp)


def deleted_visibility_signature(dirpath, victim_path):
    dp = opendir_or_raise(dirpath)
    try:
        before = set(n for n in read_all_names(dp) if n not in (".", ".."))
        os.unlink(victim_path)
        libc.rewinddir(dp)
        after = set(n for n in read_all_names(dp) if n not in (".", ".."))
        return ("victim" in before, "victim" in after)
    finally:
        libc.closedir(dp)


def main():
    try:
        with mergerfs_mount() as (mount, _):
            with tempfile.TemporaryDirectory() as native:
                merge_base = temp_dir(mount)
                try:
                    native_base = join(native, os.path.basename(merge_base))
                    os.makedirs(native_base, exist_ok=True)

                    merge_dir = join(merge_base, "dir")
                    native_dir = join(native_base, "dir")
                    merge_notdir = join(merge_base, "notdir")
                    native_notdir = join(native_base, "notdir")

                    for idx in range(8):
                        touch(join(merge_dir, f"f{idx}"), b"x")
                        touch(join(native_dir, f"f{idx}"), b"x")

                    touch(merge_notdir, b"x")
                    touch(native_notdir, b"x")

                    err = compare_calls(
                        "opendir ENOENT",
                        lambda: opendir_or_raise(join(merge_base, "missing")),
                        lambda: opendir_or_raise(join(native_base, "missing")),
                    )
                    if err:
                        return fail(err)

                    err = compare_calls(
                        "opendir ENOTDIR",
                        lambda: opendir_or_raise(join(merge_notdir, "child")),
                        lambda: opendir_or_raise(join(native_notdir, "child")),
                    )
                    if err:
                        return fail(err)

                    merge_names = names_set(merge_dir)
                    native_names = names_set(native_dir)
                    if merge_names != native_names:
                        return fail(f"readdir names mismatch mergerfs={sorted(merge_names)} native={sorted(native_names)}")

                    merge_consistent = seek_tell_consistent(merge_dir)
                    native_consistent = seek_tell_consistent(native_dir)
                    if merge_consistent != native_consistent:
                        return fail(
                            f"seekdir/telldir consistency mismatch mergerfs={merge_consistent} native={native_consistent}"
                        )

                    touch(join(merge_dir, "victim"), b"x")
                    touch(join(native_dir, "victim"), b"x")
                    merge_sig = deleted_visibility_signature(merge_dir, join(merge_dir, "victim"))
                    native_sig = deleted_visibility_signature(native_dir, join(native_dir, "victim"))
                    if merge_sig != native_sig:
                        return fail(f"deleted-entry visibility mismatch mergerfs={merge_sig} native={native_sig}")

                    return 0
                finally:
                    cleanup_dir(merge_base)
    except RuntimeError as exc:
        print(str(exc), end="")
        return 77


if __name__ == "__main__":
    raise SystemExit(main())