"""Drive filelock's lock end to end and print one key=value per line.

One interpreter start does the whole of it. Every python3 start on this arm is a
fresh load of one ~63 MB static image (99-python3 records the cost), so the suite
that drives this asserts on the OUTPUT rather than calling python3 once per case.

Every line is `key=value`; a key is emitted at most once. The suite reads keys it
names and ignores the rest, so adding a fact here cannot break it. Failure is
reported as a key too -- `*_error` -- rather than only as an exit status, because
the case that reds should say which stage stopped and why.

WHY THIS PROBE NEVER BLOCKS WITHOUT A BOUND, ANYWHERE.
======================================================
A lock is the one library whose failure mode is a HANG rather than a wrong
answer, and a suite that wedges the guest is the one outcome worth more than a
red: brintos_local_verify quarantines a wedger and every sibling on the shared
batch image loses the round with it. So every acquire here carries an explicit
`timeout=`, and the forked child is reaped with waitpid(WNOHANG) in a bounded
poll loop rather than with a blocking wait -- see the fork section for the second
reason that matters on this arm specifically.

THE LOCK FILE IS NOT DELETED ON RELEASE, AND THAT IS UPSTREAM'S DECISION.
========================================================================
filelock's UnixFileLock._release does flock(LOCK_UN) + close(2) and deliberately
leaves the file: unlinking it races another process that has already open()ed it
and is about to flock it, which would hand the lock to two holders at once
(py-filelock issue #31). BaseFileLock.release says so in its own docstring --
"the lock file itself is not automatically deleted". This probe therefore MEASURES
persistence rather than removal; the ticket's problem statement expects the file
to be gone after the context manager exits, and the library it names does not do
that on Unix. SoftFileLock is the backend that unlinks, and it is not the one this
image selects.
"""
import errno
import os
import sys
import time

W = sys.argv[1]

_seen = set()


def emit(key, value):
    if key in _seen:
        return
    _seen.add(key)
    # One line per fact: the suite greps by key, and an embedded newline would
    # land as a second, unkeyed line.
    print("%s=%s" % (key, str(value).replace("\n", " ").replace("\r", " ")))
    sys.stdout.flush()


def fail(key, exc):
    emit(key, "%s: %s" % (type(exc).__name__, exc))
    raise SystemExit(1)


def classify(exc):
    """Name an OSError by its errno rather than by its message.

    A message is glibc's and moves between releases; the errno is the kernel's
    answer and is the thing a platform gap should be reported as. Anything that
    is not an OSError reports its class name, so a NotImplementedError -- which
    is what filelock raises when flock returns ENOSYS -- is not flattened into
    the same bucket as a permission problem.
    """
    if isinstance(exc, OSError) and exc.errno is not None:
        return "%s:%s" % (type(exc).__name__, errno.errorcode.get(exc.errno, exc.errno))
    return type(exc).__name__


# ------------------------------------------------------------------ the import
try:
    import filelock
except Exception as exc:                      # noqa: BLE001 - reported, not raised
    fail("import_error", exc)

emit("version", getattr(filelock, "__version__", "<absent>"))
emit("module_file", filelock.__file__)

# The public surface, whole and sorted. filelock 3.13.4 documents exactly the
# names in its own __all__, so this is asserted as a SET rather than by presence:
# a release that grew or dropped a public entry point would be a different
# library than the one this suite describes, and nothing else here would notice.
emit("api_surface", ",".join(sorted(a for a in dir(filelock) if not a.startswith("_"))))

# ------------------------------------------------- WHICH BACKEND WAS SELECTED
# THE SINGLE MOST IMPORTANT FACT IN THIS FILE, and the one that a wasm32 image
# can get wrong SILENTLY. filelock/__init__.py picks UnixFileLock only when
# `import fcntl` succeeded; otherwise it falls back to SoftFileLock, which is an
# atomic-create lock with none of flock's kernel semantics, and warns once into a
# warnings filter nobody reads. On this arm fcntl is compiled INTO the
# interpreter (MODULE_BUILDTYPE=static -- there is no lib-dynload here), so the
# import succeeds and the real backend is selected; the empty python3-fcntl
# package is a packaging artifact of that and not a missing module.
#
# A green here is what makes every contention case below a statement about
# flock(2). A red would mean this image shipped a mutex that does not serialise
# the way its consumers assume.
emit("has_fcntl", "yes" if getattr(filelock, "has_fcntl", False) else "no")
emit("backend", filelock.FileLock.__name__)
try:
    import fcntl
except Exception as exc:                      # noqa: BLE001
    emit("fcntl_file", "<absent: %s>" % classify(exc))
else:
    # A static extension has no __file__. That absence IS the expected answer on
    # this arm and is emitted as a value rather than as a failure.
    emit("fcntl_file", getattr(fcntl, "__file__", "<builtin>"))

FileLock = filelock.FileLock
Timeout = filelock.Timeout

# The filesystem each lock below is taken on, read from the process's own view
# rather than assumed. The suite's scratch dir is tmpfs; the rootfs is a virtiofs
# cloud mount whose files report the HOST's uid (88-logrotate measured that), so
# when the rootfs case reds this is the line that says on what.
def fstype(path):
    try:
        with open("/proc/self/mountinfo") as fh:
            entries = []
            for line in fh:
                parts = line.split(" - ")
                if len(parts) < 2:
                    continue
                mp = parts[0].split()[4]
                entries.append((mp, parts[1].split()[0]))
    except Exception:                         # noqa: BLE001
        return "<unreadable>"
    best, best_ty = "", "<none>"
    real = os.path.realpath(path)
    for mp, ty in entries:
        if (real == mp or real.startswith(mp.rstrip("/") + "/")) and len(mp) >= len(best):
            best, best_ty = mp, ty
    return best_ty


emit("tmp_fstype", fstype(W))

# ------------------------------------------------------------ acquire/release
# The lifecycle the ticket names, driven on the scratch dir. Each fact is emitted
# as its own key so a red says WHICH step of the lifecycle stopped working rather
# than that the lock "did not work".
LOCK = os.path.join(W, "target.lock")
emit("lockfile_pre_exists", "yes" if os.path.exists(LOCK) else "no")

lock = FileLock(LOCK, timeout=5)
emit("lock_is_unix", "yes" if isinstance(lock, filelock.UnixFileLock) else "no")
emit("locked_before", "yes" if lock.is_locked else "no")
try:
    lock.acquire()
except Exception as exc:                      # noqa: BLE001
    fail("acquire_error", classify(exc))
emit("acquire_ok", "yes")
emit("locked_after_acquire", "yes" if lock.is_locked else "no")
emit("lockfile_after_acquire", "yes" if os.path.isfile(LOCK) else "no")
emit("counter_after_acquire", lock.lock_counter)

# RE-ENTRANT WITHIN ONE PROCESS, which is the property the counter exists for: a
# second acquire on the SAME object does not deadlock against itself, it counts.
try:
    lock.acquire()
except Exception as exc:                      # noqa: BLE001
    fail("reacquire_error", classify(exc))
emit("reacquire_ok", "yes")
emit("counter_after_reacquire", lock.lock_counter)
emit("locked_after_reacquire", "yes" if lock.is_locked else "no")

# ...and the release is counted too. The FIRST release must NOT drop the lock --
# that is the whole point of a re-entrant lock, and a library that released on
# the inner exit would hand the file to another process while the outer scope
# still believed it held it.
lock.release()
emit("counter_after_release1", lock.lock_counter)
emit("locked_after_release1", "yes" if lock.is_locked else "no")

lock.release()
emit("counter_after_release2", lock.lock_counter)
emit("locked_after_release2", "yes" if lock.is_locked else "no")

# THE FILE SURVIVES THE RELEASE. See the header: upstream leaves it on purpose,
# and this key is the measurement that says so on this image rather than a claim
# read off the source.
emit("lockfile_after_release", "yes" if os.path.isfile(LOCK) else "no")

# ------------------------------------------------------- the context manager
# The documented idiom, on its own path so it cannot be confused with the
# explicit acquire/release above.
CM = os.path.join(W, "cm.lock")
try:
    with FileLock(CM, timeout=5) as held:
        emit("cm_locked_inside", "yes" if held.is_locked else "no")
        emit("cm_file_inside", "yes" if os.path.isfile(CM) else "no")
        cm_obj = held
except Exception as exc:                      # noqa: BLE001
    fail("cm_error", classify(exc))
emit("cm_locked_after", "yes" if cm_obj.is_locked else "no")
# Persistence again, through the idiom the ticket asks about by name.
emit("cm_file_after", "yes" if os.path.isfile(CM) else "no")

# ------------------------------------------------- contention, same process
# TWO SEPARATE FileLock OBJECTS ON ONE PATH CONTEND, and they contend for the
# reason that makes flock the right primitive here: a flock is held by the OPEN
# FILE DESCRIPTION, not by the process, so a second open() of the same file gets
# a description of its own and blocks against the first even from one pid. This
# is what fcntl(F_SETLK) record locking would NOT do -- those are per-process and
# would silently succeed -- and record locking is refused outright on this arm
# anyway (brintos/linux#16, which took libutempter's utmp layer down). filelock
# picked flock; 79-lockdev's port turns on the same distinction.
outer = FileLock(LOCK, timeout=5)
outer.acquire()
emit("outer_locked", "yes" if outer.is_locked else "no")

rival = FileLock(LOCK, timeout=0.5)
t0 = time.perf_counter()
try:
    rival.acquire()
except Timeout:
    emit("contend_sameproc", "Timeout")
except Exception as exc:                      # noqa: BLE001
    emit("contend_sameproc", classify(exc))
else:
    emit("contend_sameproc", "<acquired>")
    rival.release()
waited = time.perf_counter() - t0
# It WAITED rather than refusing instantly. A backend that raised Timeout without
# polling would satisfy the case above and be a different behaviour; the bound is
# loose (a poll interval is 50 ms and this arm's clock is coarse) because the
# assertion is "it blocked", not "it blocked for exactly 0.5 s".
emit("contend_waited", "yes" if waited >= 0.3 else "no")
emit("contend_rival_locked", "yes" if rival.is_locked else "no")
# The counter was UNWOUND by the failed acquire. acquire() increments before it
# tries, so a library that forgot to decrement on the Timeout path would leave a
# lock nobody can ever fully release -- silent, and only visible here.
emit("contend_rival_counter", rival.lock_counter)

# ...and blocking=False refuses immediately, which is the other documented
# entry point and the one a caller uses when it wants to do something else.
t0 = time.perf_counter()
try:
    FileLock(LOCK).acquire(blocking=False)
except Timeout:
    emit("contend_nonblocking", "Timeout")
except Exception as exc:                      # noqa: BLE001
    emit("contend_nonblocking", classify(exc))
else:
    emit("contend_nonblocking", "<acquired>")
emit("contend_nonblocking_fast", "yes" if time.perf_counter() - t0 < 0.3 else "no")

# ------------------------------------------------------- contention, a child
# THE TICKET'S OWN CASE: a second FileLock on the same path, in a FORKED CHILD,
# times out while the parent holds it.
#
# THREE THINGS ABOUT THIS ARM SHAPE THE CODE, and none of them are paranoia:
#
#  1. fork() and waitpid() are both compiled into this interpreter (HAVE_FORK,
#     HAVE_WAITPID in the target pyconfig, measured against the build) and
#     fork+exec is known to work here -- gnupg spawns gpg-agent that way
#     (80-gnupg). What does NOT work is popen/system (brintos/linux#14), which
#     is why this is a raw fork rather than subprocess.
#  2. THE CHILD IS REAPED WITH WNOHANG IN A BOUNDED LOOP, never with a blocking
#     wait. A parent that parks on a child that never exits is exactly the wedge
#     that costs a shared batch image its round, and this arm has a live defect
#     of that shape -- a foreground child is never reaped in the sigsuspend idiom
#     (linux-in-the-browser#161), which is also why timeout(1) is used nowhere in
#     this suite. os.waitpid(WNOHANG) cannot block by construction, so an
#     unreapable child costs one key set to `no` instead of the whole guest.
#  3. THE CHILD TALKS THROUGH A FILE, NOT THE PIPE. It closes the inherited
#     stdout/stderr before doing anything, so a child that wedged could not hold
#     the probe's output pipe open and stall the reader on the other end.
#
# The child inherits the parent's LOCKED descriptor across the fork -- a flock
# travels with the open file description -- so it is contending with a lock it
# also holds a handle to. That is the honest reproduction of the case: it opens
# its OWN description, which is what a genuinely separate process would do, and
# that description is what has to be refused. It exits with os._exit so filelock's
# __del__ never runs in the child and cannot release the parent's lock.
VERDICT = os.path.join(W, "child.verdict")
emit("parent_holds_for_child", "yes" if outer.is_locked else "no")
try:
    pid = os.fork()
except Exception as exc:                      # noqa: BLE001
    emit("fork_supported", "no")
    emit("fork_error", classify(exc))
    pid = -1

if pid == 0:
    # ---- child ----------------------------------------------------------
    # THE REDIRECT IS THE FIRST STATEMENT AFTER THE FORK, and nothing above it
    # may emit: everything between fork() and this dup2 runs in BOTH processes
    # and lands twice in the parent's transcript. Measured on the x86-64 control,
    # where an emit placed before the branch duplicated its line.
    try:
        devnull = os.open(os.devnull, os.O_RDWR)
        os.dup2(devnull, 1)
        os.dup2(devnull, 2)
    except Exception:                         # noqa: BLE001
        pass
    try:
        child_lock = FileLock(LOCK, timeout=1)
        try:
            child_lock.acquire()
        except Timeout:
            verdict = "Timeout"
        except Exception as exc:              # noqa: BLE001
            verdict = classify(exc)
        else:
            verdict = "<acquired>"
        with open(VERDICT, "w") as fh:
            fh.write(verdict)
    except Exception:                         # noqa: BLE001
        pass
    os._exit(0)

if pid > 0:
    # ---- parent ---------------------------------------------------------
    emit("fork_supported", "yes")
    reaped, status = "no", None
    deadline = time.perf_counter() + 30
    while time.perf_counter() < deadline:
        try:
            done, status = os.waitpid(pid, os.WNOHANG)
        except ChildProcessError:
            reaped = "gone"
            break
        except Exception as exc:              # noqa: BLE001
            emit("child_wait_error", classify(exc))
            break
        if done == pid:
            reaped = "yes"
            break
        time.sleep(0.1)
    emit("child_reaped", reaped)
    if reaped == "yes" and status is not None:
        emit("child_exit", os.waitstatus_to_exitcode(status))
    if reaped == "no":
        # Signals are unreliable on this arm (brintos/linux#21 -- kill(2) reports
        # success and delivers nothing), so this is a best effort and its outcome
        # is not asserted anywhere. It is here so a wedged child is at least ASKED
        # to go before the probe returns.
        try:
            os.kill(pid, 9)
        except Exception:                     # noqa: BLE001
            pass
    # The verdict is read from the file whether or not the child was reaped: the
    # two facts are independent, and a child that answered correctly and then
    # failed to be collected is a platform finding rather than a lock failure.
    try:
        with open(VERDICT) as fh:
            emit("child_verdict", fh.read().strip() or "<empty>")
    except Exception as exc:                  # noqa: BLE001
        emit("child_verdict", "<unread: %s>" % classify(exc))

# THE LOCK IS ACTUALLY RELEASABLE, and the path is free afterwards. Without this
# every Timeout above would also be satisfied by a lock that can never be taken
# at all -- which is a broken mutex that passes a contention test.
outer.release()
emit("outer_released", "yes" if not outer.is_locked else "no")
after = FileLock(LOCK, timeout=2)
try:
    after.acquire()
except Exception as exc:                      # noqa: BLE001
    emit("acquire_after_release", classify(exc))
else:
    emit("acquire_after_release", "ok")
    after.release()

# ------------------------------------------------- the lock on the ROOTFS
# WHY THIS IS A SEPARATE CASE AND NOT A DUPLICATE. Everything above ran on the
# suite's scratch dir, which is tmpfs. The consumers this package exists for --
# pip, virtualenv, tox -- lock a SHARED CACHE, which lives on the real filesystem;
# and this image's real filesystem is a virtiofs cloud mount, whose semantics
# differ from tmpfs in ways this port has already met elsewhere (88-logrotate: a
# file there reports the HOST's uid while the process is uid 0).
#
# flock on a FUSE-backed mount is a capability the server has to advertise, so it
# is genuinely possible for it to be refused here while tmpfs works perfectly.
# filelock turns exactly that into NotImplementedError when flock returns ENOSYS.
# The result is emitted with its errno so a red names the kernel's own answer.
#
# The fchmod filelock does after opening is wrapped in suppress(PermissionError)
# upstream, so the host-uid mismatch on that mount cannot fail the acquire by
# itself -- which is what makes this a clean statement about flock alone.
# The directory is argv[2] where the caller names one, and /var/lib otherwise.
# The suite passes nothing, so on target this is /var/lib -- a real directory on
# the virtiofs root that the suite, running as uid 0, can write. The parameter
# exists so the x86-64 control can drive this same code path against a directory
# it is allowed to write, since an unprivileged /var/lib answers EACCES at open()
# and would say nothing about flock at all.
ROOTDIR = sys.argv[2] if len(sys.argv) > 2 else "/var/lib"
ROOTLOCK = os.path.join(ROOTDIR, "brintos-filelock-probe.%d.lock" % os.getpid())
emit("rootfs_dir", ROOTDIR)
emit("rootfs_fstype", fstype(ROOTDIR))
rl = FileLock(ROOTLOCK, timeout=2)
try:
    rl.acquire()
except Exception as exc:                      # noqa: BLE001
    emit("rootfs_lock", classify(exc))
else:
    emit("rootfs_lock", "ok")
    emit("rootfs_lockfile", "yes" if os.path.isfile(ROOTLOCK) else "no")
    rl.release()
    # Cleaned up by this probe rather than by the library, precisely because the
    # library does not remove it -- see the header. Leaving it would put a file
    # per boot into the image's own /var/lib.
    try:
        os.unlink(ROOTLOCK)
    except Exception:                         # noqa: BLE001
        pass

# ------------------------------------------------------------ the soft backend
# SoftFileLock is shipped by this package and is the fallback the selector would
# choose if fcntl ever went missing, so it is exercised once: it must work, and
# it must have the OTHER unlink behaviour, which is the cleanest available proof
# that `lockfile_after_release` above is a property of the flock backend and not
# of this filesystem.
SOFT = os.path.join(W, "soft.lock")
try:
    with filelock.SoftFileLock(SOFT, timeout=5):
        emit("soft_file_inside", "yes" if os.path.isfile(SOFT) else "no")
except Exception as exc:                      # noqa: BLE001
    emit("soft_error", classify(exc))
else:
    emit("soft_file_after", "yes" if os.path.isfile(SOFT) else "no")

emit("done", "yes")