#!/usr/bin/env python3
"""One interpreter start for the whole of 99-python3-asyncinotify.

Emits `key=value' lines, one per line, in file order; the suite reads them back
with grep.  A python3 start on this arm costs tens of seconds, so the start is
taken ONCE and every measurement below rides it.

THREE RULES THE VALUES OBEY, ALL ABOUT WHAT SURVIVES THE TRANSPORT.

The runner parses its input line by line, so no value may contain a NEWLINE.
An event sequence is reported as a joined string ("CREATE,MODIFY,DELETE"),
never as a list printed over several lines, and an exception is reported as
`<Name>: <msg>' with the message flattened.

The suite splits each line on the FIRST `=', so a VALUE containing `=' is safe
and a KEY containing one is not.  Every key emitted here is `[a-z0-9_]+'.

And no value may depend on the encoding stdout happens to be in.  Everything
below is ASCII by construction: the fixtures are short file names and numbers.

============================================================================
WHY ALMOST NOTHING HERE TOUCHES A REAL INOTIFY DESCRIPTOR.
============================================================================

It cannot.  On this arm `inotify_init1(2)' returns ENOSYS -- the subsystem is
compiled into vmlinux and its sysctls register at boot, but no member of the
inotify family has an entry in arch/wasm32/kernel/syscall_table.c and the
hardwarejs worker does not answer for it either (brintos/linux#33, open).
`Inotify()' therefore raises OSError(ENOSYS) on its FIRST statement and there
is no way to obtain a watch descriptor on this image at all.

A second, INDEPENDENT gap sits one line further into the same constructor:
`Inotify.__init__' calls `select.epoll()' unconditionally, and the epoll family
is unwired in exactly the same way (brintos/linux#53, open).  So the object is
unconstructible for two separate reasons, and this probe measures BOTH rather
than reporting the first and inferring the second -- `inotify_ctor_stage' says
which one actually fired.  Conflating them is how a port ends up attributed to
the wrong open issue.

WHAT IS LEFT IS NOT SMALL, and it is the reason this port ships rather than
being parked.  asyncinotify is a ctypes binding plus an event-stream PARSER
plus an asyncio readiness path, and only the first of those three needs the
syscall:

  * The PARSER (`Inotify._get') turns a byte stream of `struct inotify_event'
    into Event objects -- variable-length names, NUL truncation, cookie
    pairing, watch lookup, and the IGNORED/ONESHOT ownership transfer.  That is
    the most intricate code in the package and it reads from `os.read(self.fd)'
    -- it does not care what KIND of descriptor that is.  Driven below over a
    pipe carrying bytes this probe lays out by hand, so the expected values are
    exact rather than whatever the kernel happened to emit.

  * The READINESS path (`_get_events_future') registers the descriptor with
    `loop.add_reader()' and waits for the loop to wake.  That is the half the
    ticket calls out as never having been asserted on target from Python, and
    it is answerable over a pipe fd for the same reason: `add_reader' does not
    care either.

So the substitution is deliberate and it is stated in every case name that
relies on it (`pipe-' and `loop-' prefixes).  What this probe CANNOT answer is
whether the kernel's inotify would deliver the right events, because the kernel
will not open one.  That question is attributed to brintos/linux#33 and is not
dressed up as something this suite measured.

============================================================================
THE CONTROL.
============================================================================

Every expected value in 99-python3-asyncinotify.test was read off an x86-64
CPython 3.12.14 (the build's own python3-native, the SAME minor version the
target ships) running THIS BUILD'S wasm32 .deb contents on PYTHONPATH.  So a
red on target is a target-only divergence, not a miscalibrated assertion.

Nothing below asserts a repr() of an IntFlag member or a duration.  IntFlag's
repr and str changed in 3.11 and its member ordering is not a stable contract;
durations measure the host's load.  Names, integer values and orderings are
what get asserted.
"""

import os
import sys
import types

emitted = []


def emit(key, value):
    emitted.append(key)
    print('%s=%s' % (key, value))
    sys.stdout.flush()      # a hang in a later group must not lose an earlier line


def one_line(text):
    return ' '.join(str(text).split())


def guard(key, fn):
    """Emit fn()'s value, or `<ExcName>: <msg>' if it raised.

    Nothing in this probe may abort the run: one unexpected exception would cost
    every case after it, and the suite would report those as absent rather than
    as red -- which reads as a broken port instead of as one broken case.
    """
    try:
        emit(key, fn())
    except Exception as exc:
        emit(key, '%s: %s' % (type(exc).__name__, one_line(exc)))


try:
    import asyncinotify
    from asyncinotify import Inotify, Watch, Event, Mask, InitFlags
    from asyncinotify import _ffi
except Exception as exc:                                        # pragma: no cover
    print('import_error=%s: %s' % (type(exc).__name__, one_line(exc)))
    sys.exit(1)

# ------------------------------------------------------------------ identity --
emit('module_file', getattr(asyncinotify, '__file__', '<none>'))
emit('is_package', 'yes' if hasattr(asyncinotify, '__path__') else 'no')
emit('ffi_module', getattr(_ffi, '__name__', '<none>'))

# The public surface. A release that grows or drops a name is a red here rather
# than a surprise to a caller who copied this suite.
guard('public_names',
      lambda: ','.join(sorted(n for n in dir(asyncinotify) if not n.startswith('_'))))

# THE MODULE HAS NO `__version__'. The ticket asks the module to "report the
# version the recipe pins"; flit takes the version from pyproject.toml and
# asyncinotify/__init__.py never defines the attribute, so the only answer is
# the installed distribution's metadata. Both halves are asserted so the
# correction lives in a case and not only in a commit message.
guard('version_attr_absent',
      lambda: 'yes' if not hasattr(asyncinotify, '__version__') else 'no')
guard('dist_version',
      lambda: __import__('importlib.metadata', fromlist=['version']).version('asyncinotify'))

# ----------------------------------------------------------------------- abi --
# `struct inotify_event' is the kernel ABI this package reads with ctypes, and
# a wrong layout here would silently mis-parse every event rather than fail.
# It is 16 bytes on LP64 and on this arm's ILP32 alike -- int + three uint32,
# no pointer -- so the value is the same on the control and on target, which is
# exactly why a divergence would be worth knowing about.
import ctypes

guard('event_struct_size', lambda: _ffi.inotify_event_size)
guard('event_offsets', lambda: ','.join(
    '%s@%d' % (f, getattr(_ffi.inotify_event, f).offset)
    for f in ('wd', 'mask', 'cookie', 'len')))
guard('event_field_sizes', lambda: ','.join(
    '%s:%d' % (f, getattr(_ffi.inotify_event, f).size)
    for f in ('wd', 'mask', 'cookie', 'len')))
guard('name_max', lambda: _ffi.NAME_MAX)
guard('read_chunk', lambda: (_ffi.inotify_event_size + _ffi.NAME_MAX + 1) * 10)

# ------------------------------------------------------------- libc binding --
# The port's one non-Python dependency: _ffi.py resolves libc through
# ctypes.util.find_library('c') at IMPORT time and declares four prototypes on
# it. find_library() shells out to ldconfig/gcc/objdump and may well answer None
# on a target image that has none of them -- in which case CDLL(None) opens the
# main program's own handle, which works when the interpreter is dynamically
# linked against libc. Both the name and the outcome are reported, because
# `None' here is NOT a failure and must not read as one.
guard('libc_name', lambda: repr(_ffi._libcname))
guard('libc_loaded', lambda: 'yes' if _ffi.libc is not None else 'no')

for _sym in ('inotify_init', 'inotify_init1', 'inotify_add_watch', 'inotify_rm_watch'):
    def _has(s=_sym):
        try:
            getattr(_ffi.libc, s)
            return 'yes'
        except AttributeError:
            return 'no'
    guard('sym_' + _sym, _has)

# check_return is the errno helper every one of those four goes through.
guard('check_return_passthrough', lambda: _ffi.check_return(7))


def _check_return_raises():
    ctypes.set_errno(2)
    try:
        _ffi.check_return(-1)
    except OSError as exc:
        return 'OSError:%s' % exc.errno
    return 'no-raise'


guard('check_return_raises', _check_return_raises)

# ------------------------------------------------- the two platform gaps ------
# Measured separately and directly, never inferred from each other or from the
# constructor's failure. These are what the `Inotify()' cases below are
# attributed to, so they have to be facts in their own right.
import select

guard('epoll_attr_present', lambda: 'yes' if hasattr(select, 'epoll') else 'no')


def _epoll_usable():
    if not hasattr(select, 'epoll'):
        return 'absent'
    try:
        ep = select.epoll()
        ep.close()
        return 'yes'
    except OSError as exc:
        _gap['epoll_errno'] = exc.errno
        return 'no'


_gap = {}
guard('epoll_usable', _epoll_usable)
guard('epoll_errno', lambda: _gap.get('epoll_errno', '<none>'))

# The selector asyncio will actually use. `selectors._can_use' probes each
# backend by CONSTRUCTING it inside try/except OSError rather than trusting the
# build-time macro, so an epoll that exists and refuses falls through to poll --
# which is why the loop cases further down can run at all on this arm.
import selectors

guard('selector_class', lambda: selectors.DefaultSelector.__name__)
guard('selector_is_epoll',
      lambda: 'yes' if selectors.DefaultSelector is getattr(selectors, 'EpollSelector', None) else 'no')


def _inotify_init1():
    """Call the syscall the constructor calls, FIRST, and by itself.

    This is what tells brintos/linux#33 from brintos/linux#53: the constructor
    would hit both, in this order, and only the first one's errno is the one it
    reports.
    """
    ctypes.set_errno(0)
    try:
        fd = _ffi.libc.inotify_init1(InitFlags.CLOEXEC | InitFlags.NONBLOCK)
    except OSError as exc:
        _gap['ino_errno'] = exc.errno
        return 'OSError:%s' % exc.errno
    os.close(fd)
    return 'fd'


guard('inotify_init1', _inotify_init1)
guard('inotify_init1_errno', lambda: _gap.get('ino_errno', '<none>'))


def _ctor():
    """Construct an Inotify, and CATCH THE SECOND FAILURE THE FIRST ONE CAUSES.

    `Inotify.__init__' assigns `self._fd' only AFTER `inotify_init1' returns, so
    on an arm where that call raises, the half-built object reaches __del__ with
    the slot unset -- and `close()' reads `self._fd' and raises AttributeError
    inside __del__. CPython routes that to sys.unraisablehook and prints a
    traceback to stderr.

    That is an upstream wart, visible only on a platform where inotify is
    refused, which is to say only here. It is captured rather than left to
    spray the capture at whatever moment the collector runs: the hook makes it
    deterministic AND turns it into a case (`ctor_del_unraisable'), so a future
    release that fixes the ordering is a visible change rather than a silence.
    We file nothing against third-party public repositories, so recording it
    here is where it ends.
    """
    import gc
    seen = []
    prev = sys.unraisablehook
    sys.unraisablehook = lambda u: seen.append(getattr(u.exc_type, '__name__', '?'))
    try:
        try:
            ino = Inotify()
        except OSError as exc:
            _gap['ctor_errno'] = exc.errno
            return 'OSError:%s' % exc.errno
        except Exception as exc:
            return '%s: %s' % (type(exc).__name__, one_line(exc))
        ino.close()
        return 'constructed'
    finally:
        gc.collect()
        sys.unraisablehook = prev
        _gap['del_unraisable'] = ','.join(seen) if seen else '<none>'


guard('inotify_ctor', _ctor)
guard('ctor_del_unraisable', lambda: _gap.get('del_unraisable', '<none>'))


def _ctor_stage():
    """WHICH of the two gaps the constructor actually died on.

    `Inotify.__init__' runs `inotify_init1' before `select.epoll()', so on this
    arm the reported errno is inotify's and epoll is never reached. If #33 closes
    while #53 stays open this flips to `epoll' with no other case moving, which
    is precisely the discrimination the batch's other fs-notification members
    need and cannot get from the errno alone.
    """
    if _gap.get('ctor_errno') is None:
        return 'constructed'
    if _gap.get('ino_errno') is not None:
        return 'inotify_init1'
    if _gap.get('epoll_errno') is not None:
        return 'epoll'
    return 'unknown'


guard('inotify_ctor_stage', _ctor_stage)

# --------------------------------------------------------- the kernel ABI -----
# Mask is the inotify UAPI bitmask transcribed into an IntFlag. These are the
# numbers the parser matches events against, so a transcription error would
# mis-classify every event rather than fail. Values only -- no repr, no str.
_MASKS = ('ACCESS', 'MODIFY', 'ATTRIB', 'CLOSE_WRITE', 'CLOSE_NOWRITE', 'OPEN',
          'MOVED_FROM', 'MOVED_TO', 'CREATE', 'DELETE', 'DELETE_SELF',
          'MOVE_SELF', 'UNMOUNT', 'Q_OVERFLOW', 'IGNORED', 'ONLYDIR',
          'DONT_FOLLOW', 'EXCL_UNLINK', 'MASK_CREATE', 'MASK_ADD', 'ISDIR',
          'ONESHOT')
guard('mask_values', lambda: ','.join('%s:%#x' % (n, int(getattr(Mask, n))) for n in _MASKS))
guard('mask_close', lambda: '%#x' % int(Mask.CLOSE))
guard('mask_move', lambda: '%#x' % int(Mask.MOVE))
guard('mask_close_is_union',
      lambda: 'yes' if int(Mask.CLOSE) == int(Mask.CLOSE_WRITE) | int(Mask.CLOSE_NOWRITE) else 'no')
guard('mask_move_is_union',
      lambda: 'yes' if int(Mask.MOVE) == int(Mask.MOVED_FROM) | int(Mask.MOVED_TO) else 'no')
guard('initflags', lambda: 'CLOEXEC:%#x,NONBLOCK:%#x' % (int(InitFlags.CLOEXEC),
                                                         int(InitFlags.NONBLOCK)))
guard('initflags_match_os',
      lambda: 'yes' if (int(InitFlags.CLOEXEC) == os.O_CLOEXEC
                        and int(InitFlags.NONBLOCK) == os.O_NONBLOCK) else 'no')

# ================================================================== parser ====
#
# `Inotify._get' is the package's real machinery: it reads a byte stream of
# `struct inotify_event' records and turns them into Event objects. It reads
# through `os.read(self.fd, ...)' and never asks what kind of descriptor that
# is, so a PIPE carrying bytes laid out here drives it exactly as the kernel
# would -- with the expected values exact, because this file wrote them.
#
# The instance is built with `__new__' and its slots filled by hand. That is
# not a shortcut around a failing constructor so much as the only way in: the
# constructor's first statement is the ENOSYS syscall, so there is no path to a
# real instance on this arm at all.

import struct
from pathlib import Path

_EVHDR = struct.Struct('=iIII')          # wd, mask, cookie, len -- 16 bytes


def ev(wd, mask, cookie=0, name=None, pad=0):
    """One `struct inotify_event', laid out the way the kernel lays it out.

    The kernel NUL-terminates the name and pads to a multiple of the struct
    alignment, so `len' is generally LARGER than the name. `pad' reproduces
    that, because the parser's name handling is all about finding the first NUL
    inside a longer field and it would pass trivially against tight names.
    """
    if name is None:
        raw = b''
    else:
        raw = name.encode() + b'\0' * (1 + pad)
    return _EVHDR.pack(wd, int(mask), cookie, len(raw)) + raw


class _Fake:
    """An Inotify whose descriptor is a pipe. See the header."""

    def __init__(self, watches, nonblock=True):
        self.r, self.w = os.pipe()
        if nonblock:
            os.set_blocking(self.r, False)
        ino = Inotify.__new__(Inotify)
        ino._fd = self.r
        ino._watches = watches
        ino._events = []
        ino._epoll = None
        ino._sync_timeout = None
        ino._cache_size = 10
        self.ino = ino

    def feed(self, blob):
        os.write(self.w, blob)

    def __enter__(self):
        return self

    def __exit__(self, *a):
        # `_fd' is cleared FIRST so that Inotify.__del__ -> close() takes its
        # no-op arm: close() would otherwise call self._epoll.close(), and this
        # instance has no epoll (there is no epoll on this arm to give it).
        self.ino._fd = None
        for fd in (self.r, self.w):
            try:
                os.close(fd)
            except OSError:
                pass


# Watch weakrefs its owning Inotify, so it cannot be handed None. These
# fixtures get a detached owner: `_fd' None means its __del__ -> close() takes
# the no-op arm and it never touches a descriptor or the epoll it does not have.
_OWNER = Inotify.__new__(Inotify)
_OWNER._fd = None
_OWNER._sync_timeout = None


def _watch(wd, path, mask, owner=None):
    return Watch(inotify=owner or _OWNER, path=Path(path), mask=mask, wd=wd)


# THE ORDERED EVENT SET THE TICKET NAMES, parsed out of one read.
#
# create, modify, close-write, the moved-from/moved-to pair sharing one cookie,
# and delete -- each naming the file. On a working kernel this is the byte
# stream a watch on a directory would produce for
# `touch a; echo >a; mv a b; rm b'. Here it is the byte stream itself that is
# under test, which is the half of that sentence this arm can still answer.
_W = {1: _watch(1, '/watched', Mask.CREATE | Mask.MODIFY | Mask.CLOSE_WRITE
                | Mask.MOVE | Mask.DELETE)}

_SEQ = (ev(1, Mask.CREATE, 0, 'a.txt', 2)
        + ev(1, Mask.MODIFY, 0, 'a.txt', 2)
        + ev(1, Mask.CLOSE_WRITE, 0, 'a.txt', 2)
        + ev(1, Mask.MOVED_FROM, 0x2a, 'a.txt', 2)
        + ev(1, Mask.MOVED_TO, 0x2a, 'b.txt', 2)
        + ev(1, Mask.DELETE, 0, 'b.txt', 2))


def _parse_seq():
    with _Fake(dict(_W)) as f:
        f.feed(_SEQ)
        return _drain(f.ino)


def _drain(ino, n=None):
    """Pull events with the SYNCHRONOUS path.

    `sync_get' with sync_timeout None skips the epoll poll entirely and goes
    straight to `_get' -- so the parser is reachable synchronously on this arm
    even though `Inotify.sync_timeout' as a feature is not (it polls the epoll
    that brintos/linux#53 denies).
    """
    out = []
    while True:
        try:
            e = ino.sync_get()
        except BlockingIOError:
            break
        if e is None:
            break
        out.append(e)
        if n is not None and len(out) >= n:
            break
    return out


_events = []
try:
    _events = _parse_seq()
except Exception as exc:
    emit('parse_error', '%s: %s' % (type(exc).__name__, one_line(exc)))

guard('parse_count', lambda: len(_events))
guard('parse_masks', lambda: ','.join(e.mask.name or '?' for e in _events))
guard('parse_names', lambda: ','.join(str(e.name) for e in _events))
guard('parse_paths', lambda: ','.join(str(e.path) for e in _events))
guard('parse_cookies', lambda: ','.join(str(e.cookie) for e in _events))
guard('parse_wds', lambda: ','.join(str(e.watch.wd) for e in _events))

# The move pair is the one place a cookie means anything: both halves carry the
# same non-zero value and nothing else carries any.
guard('move_cookie_paired',
      lambda: 'yes' if (_events[3].cookie == _events[4].cookie != 0) else 'no')
guard('nonmove_cookies_zero',
      lambda: 'yes' if all(e.cookie == 0 for e in
                           (_events[0], _events[1], _events[2], _events[5])) else 'no')

# `in' on an Event tests its mask, and raises on anything that is not a Mask.
guard('event_contains', lambda: 'yes' if Mask.CREATE in _events[0] else 'no')
guard('event_not_contains', lambda: 'no' if Mask.DELETE in _events[0] else 'yes')


def _contains_typeerror():
    try:
        'CREATE' in _events[0]
    except TypeError:
        return 'TypeError'
    return 'no-raise'


guard('event_contains_typeerror', _contains_typeerror)


# --- name handling: the parser's fiddliest corner ---------------------------
def _names():
    out = {}
    with _Fake({1: _watch(1, '/w', Mask.CREATE)}) as f:
        f.feed(ev(1, Mask.CREATE, 0, None)              # len 0  -> name None
               + ev(1, Mask.CREATE, 0, 'short', 10)     # padded -> truncated at NUL
               + ev(1, Mask.CREATE, 0, 'x' * 255, 0))   # NAME_MAX, exact
        got = _drain(f.ino)
        out['n'] = len(got)
        out['none'] = 'None' if got[0].name is None else str(got[0].name)
        out['trunc'] = str(got[1].name)
        out['maxlen'] = len(str(got[2].name))
        # name None means the event is about the watch itself, so path is the
        # watch path rather than watchpath/None.
        out['path_none'] = str(got[0].path)
    return out


_n = {}
try:
    _n = _names()
except Exception as exc:
    emit('names_error', '%s: %s' % (type(exc).__name__, one_line(exc)))

guard('name_count', lambda: _n['n'])
guard('name_absent', lambda: _n['none'])
guard('name_truncated', lambda: _n['trunc'])
guard('name_max_len', lambda: _n['maxlen'])
guard('name_absent_path', lambda: _n['path_none'])


# --- watch lookup and the two ownership transfers ---------------------------
def _ignored():
    """IGNORED hands the Watch to the Event and drops it from the Inotify.

    The comment in rm_watch() says exactly this -- it does NOT remove from
    _watches, because the IGNORED event does. So the observable is a live
    (non-weak) watch on the event and an empty _watches afterwards.
    """
    w = _watch(3, '/w', Mask.CREATE)
    with _Fake({3: w}) as f:
        f.feed(ev(3, Mask.IGNORED, 0, None))
        got = _drain(f.ino)
        return '%d,%s,%s' % (len(got),
                             'live' if got[0].watch is w else 'other',
                             'dropped' if 3 not in f.ino._watches else 'retained')


guard('ignored_transfer', _ignored)


def _oneshot():
    w = _watch(4, '/w', Mask.ONESHOT | Mask.CREATE)
    with _Fake({4: w}) as f:
        f.feed(ev(4, Mask.CREATE, 0, 'once'))
        got = _drain(f.ino)
        return '%s,%s' % ('live' if got[0].watch is w else 'other',
                          'dropped' if 4 not in f.ino._watches else 'retained')


guard('oneshot_transfer', _oneshot)


def _weakref_watch():
    """An ordinary event holds its Watch WEAKLY -- the Inotify keeps ownership."""
    w = _watch(5, '/w', Mask.CREATE)
    with _Fake({5: w}) as f:
        f.feed(ev(5, Mask.CREATE, 0, 'k'))
        got = _drain(f.ino)
        same = got[0].watch is w
        retained = 5 in f.ino._watches
        return '%s,%s' % ('same' if same else 'other',
                          'retained' if retained else 'dropped')


guard('weakref_watch', _weakref_watch)


def _overflow():
    """Q_OVERFLOW carries wd -1, which matches no watch, so Event.watch is None
    and Event.path cannot be constructed. The docstring promises exactly that."""
    with _Fake({1: _watch(1, '/w', Mask.CREATE)}) as f:
        f.feed(_EVHDR.pack(-1, int(Mask.Q_OVERFLOW), 0, 0))
        got = _drain(f.ino)
        return '%d,%s,%s' % (len(got),
                             'None' if got[0].watch is None else 'watch',
                             'None' if got[0].path is None else 'path')


guard('overflow_event', _overflow)


def _isdir():
    """A directory event is CREATE|ISDIR -- a composite the parser must keep
    whole rather than reduce to either half."""
    with _Fake({1: _watch(1, '/w', Mask.CREATE)}) as f:
        f.feed(ev(1, Mask.CREATE | Mask.ISDIR, 0, 'sub'))
        e = _drain(f.ino)[0]
        return '%s,%s,%#x' % ('yes' if Mask.CREATE in e else 'no',
                              'yes' if Mask.ISDIR in e else 'no',
                              int(e.mask))


guard('isdir_composite', _isdir)


def _unknown_wd():
    """An event for a wd nobody is watching yields watch None, not a KeyError."""
    with _Fake({}) as f:
        f.feed(ev(99, Mask.CREATE, 0, 'orphan'))
        e = _drain(f.ino)[0]
        return '%s,%s' % ('None' if e.watch is None else 'watch', str(e.name))


guard('unknown_wd', _unknown_wd)


def _batched_read():
    """The parser must return EVERY record in one read, not just the first --
    `get()' returns one and banks the rest in `_events'."""
    with _Fake({1: _watch(1, '/w', Mask.CREATE)}) as f:
        f.feed(_SEQ)
        first = f.ino.sync_get()
        banked = len(f.ino._events)
        return '%s,%d' % (first.mask.name, banked)


guard('batched_read', _batched_read)


# ============================================== the object surface, no kernel ==
#
# Everything the classes promise that does not need a live descriptor. These are
# cheap and they are the cases that would catch a package that installed but was
# subtly wrong, as opposed to one the platform refuses.

guard('ctor_default_flags', lambda: '%#x' % int(
    __import__('inspect').signature(Inotify.__init__).parameters['flags'].default))
guard('ctor_default_flags_nonblock',
      lambda: 'yes' if (int(__import__('inspect').signature(Inotify.__init__)
                            .parameters['flags'].default) & os.O_NONBLOCK) else 'no')
guard('ctor_default_cache_size', lambda: __import__('inspect')
      .signature(Inotify.__init__).parameters['cache_size'].default)
guard('ctor_default_sync_timeout', lambda: repr(__import__('inspect')
      .signature(Inotify.__init__).parameters['sync_timeout'].default))


def _closed_fd():
    """`fd' on a closed Inotify is a ValueError with a documented message, not
    an AttributeError and not a -1."""
    ino = Inotify.__new__(Inotify)
    ino._fd = None
    try:
        ino.fd
    except ValueError as exc:
        return 'ValueError: %s' % one_line(exc)
    return 'no-raise'


guard('closed_fd', _closed_fd)


def _watch_api():
    w = _watch(11, '/a/b', Mask.CREATE | Mask.DELETE)
    before = str(w.path)
    w.path = Path('/moved')                 # documented: the caller re-points it
    return '%s,%s,%d,%#x,%s' % (before, str(w.path), w.wd, int(w.mask),
                                'live' if w.inotify is _OWNER else 'other')


guard('watch_api', _watch_api)


def _watch_weakref():
    """Watch holds its Inotify weakly, so a Watch outliving its Inotify reports
    None rather than keeping the descriptor alive."""
    with _Fake({}) as f:
        w = Watch(inotify=f.ino, path=Path('/w'), mask=Mask.CREATE, wd=1)
        alive = w.inotify is f.ino
    return 'yes' if alive else 'no'


guard('watch_weakref', _watch_weakref)


def _api_present():
    want = ('add_watch', 'rm_watch', 'close', 'get', 'sync_get',
            '__aiter__', '__anext__', '__iter__', '__next__',
            '__enter__', '__exit__')
    return ','.join(n for n in want if not hasattr(Inotify, n)) or '<all-present>'


guard('api_present', _api_present)

# ==================================================== the readiness path ======
#
# THE HALF OF THIS PORT THE TICKET SAYS HAS NEVER BEEN ASSERTED ON TARGET.
#
# `Inotify.get()' hands its descriptor to `loop.add_reader()' and awaits a
# future the loop resolves when the descriptor is readable. Whether THAT works
# is a question about this arm's selector and event loop, not about inotify --
# and it is answerable, because add_reader does not care that the descriptor is
# a pipe.
#
# It runs LAST on purpose. Every group above is already printed and flushed, so
# a platform that cannot wake its loop costs the tail of this suite rather than
# all of it. `probe_done' at the very end is what says the whole file ran.
#
# No case here measures a duration. The timer values below are deadlines to be
# beaten, never quantities to be asserted, so a loaded host cannot flap them.

import asyncio

_ASYNC_BUDGET = 20          # seconds for the whole group, enforced per-run


def _arun(coro):
    return asyncio.run(asyncio.wait_for(coro, _ASYNC_BUDGET))


async def _async_for():
    """`async for' over an Inotify, receiving the ticket's ordered event set."""
    with _Fake(dict(_W)) as f:
        async def feeder():
            await asyncio.sleep(0.02)
            f.feed(_SEQ)

        task = asyncio.ensure_future(feeder())
        got = []
        async for e in f.ino:
            got.append('%s:%s' % (e.mask.name, e.name))
            if len(got) == 6:
                break
        await task
        return ','.join(got)


guard('loop_async_for', lambda: _arun(_async_for()))


async def _interleave():
    """The loop must make progress WHILE the watch is idle.

    A second coroutine appends to a list on a timer. If `get()' blocked the loop
    inside its read the ticks could not land before the events, and this returns
    the wrong order rather than merely running slowly -- which is the point: a
    loop that is stuck looks exactly like a directory in which nothing happened.
    """
    with _Fake(dict(_W)) as f:
        order = []

        async def ticker():
            for i in range(3):
                await asyncio.sleep(0.01)
                order.append('t%d' % i)

        async def feeder():
            await asyncio.sleep(0.15)       # comfortably after all three ticks
            f.feed(ev(1, Mask.CREATE, 0, 'late'))

        t1 = asyncio.ensure_future(ticker())
        t2 = asyncio.ensure_future(feeder())
        e = await f.ino.get()
        order.append('e:%s' % e.name)
        await t1
        await t2
        return ','.join(order)


guard('loop_interleave', lambda: _arun(_interleave()))


async def _quiet_timeout():
    """A quiet watch must make `wait_for' raise TimeoutError at its deadline
    rather than hang -- the descriptor is registered and simply never fires."""
    with _Fake(dict(_W)) as f:
        try:
            await asyncio.wait_for(f.ino.get(), 0.2)
        except asyncio.TimeoutError:
            return 'TimeoutError'
        return 'no-raise'


guard('loop_quiet_timeout', lambda: _arun(_quiet_timeout()))


async def _cancel_clean():
    """A coroutine awaiting a watch is cancelled cleanly, and the reader it
    registered is removed -- `_get_events_future' does that in its `finally',
    so a cancel must not leave the loop watching a descriptor forever."""
    with _Fake(dict(_W)) as f:
        loop = asyncio.get_running_loop()
        task = asyncio.ensure_future(f.ino.get())
        await asyncio.sleep(0.05)           # let it register the reader
        task.cancel()
        try:
            await task
            outcome = 'no-raise'
        except asyncio.CancelledError:
            outcome = 'CancelledError'
        await asyncio.sleep(0)
        # remove_reader() returns True if a reader was still registered.
        leaked = loop.remove_reader(f.r)
        pending = [t for t in asyncio.all_tasks() if t is not asyncio.current_task()]
        return '%s,%s,%d' % (outcome, 'removed' if not leaked else 'leaked',
                             len(pending))


guard('loop_cancel_clean', lambda: _arun(_cancel_clean()))


async def _reader_removed():
    """After a normal get() the reader is unregistered too, not just on cancel."""
    with _Fake(dict(_W)) as f:
        loop = asyncio.get_running_loop()

        async def feeder():
            await asyncio.sleep(0.02)
            f.feed(ev(1, Mask.CREATE, 0, 'z'))

        t = asyncio.ensure_future(feeder())
        await f.ino.get()
        await t
        return 'removed' if not loop.remove_reader(f.r) else 'leaked'


guard('loop_reader_removed', lambda: _arun(_reader_removed()))


async def _fd_nonblocking():
    """The descriptor handed to the loop must be non-blocking: a blocking one in
    an event loop is the defect this group exists to catch. Asserted on the fd
    this probe hands `add_reader', since the arm denies a real inotify fd; the
    flags the CONSTRUCTOR would have used are asserted separately above, off its
    signature."""
    with _Fake(dict(_W)) as f:
        return 'yes' if not os.get_blocking(f.r) else 'no'


guard('loop_fd_nonblocking', lambda: _arun(_fd_nonblocking()))


async def _selector_in_loop():
    loop = asyncio.get_running_loop()
    sel = getattr(loop, '_selector', None)
    return type(sel).__name__ if sel is not None else '<none>'


guard('loop_selector', lambda: _arun(_selector_in_loop()))

emit('emitted_count', len(emitted) + 2)
emit('probe_done', 'yes')