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 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():
    if len(sys.argv) != 2:
        print("usage: TEST_posix_readdir <mountpoint>", file=sys.stderr)
        return 1

    mount = sys.argv[1]

    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)


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