#!/usr/bin/env python3
"""python3-absl 2.1.0 on wasm32-hwjs glibc -- ONE python3 start, N keys.

WHY A PROBE AND NOT N SHELL CASES. Everything this port has to say is said
INSIDE an interpreter: a flag is an object with a parser, a default and a help
string; `app.run' is a function that parses an argument vector and calls
another function; `absl.logging' renders records onto a stream; `absltest' runs
a unittest suite. None of it is a command with an exit status, so a shell case
per assertion would be a python3 START per assertion -- and a start on this arm
is TENS OF SECONDS (BRINTOS-1476, brintos/linux#21). This probe takes every
measurement in one interpreter and the suite scores its transcript.

NOTHING HERE FORKS, SLEEPS, SIGNALS OR OPENS A SOCKET. That is a deliberate
constraint rather than an accident of what absl needs:

  * fork(2)/exec(2) is minutes on this arm and popen(3) is ENOSYS
    (brintos/linux#14), so `absltest' is driven through `unittest' in-process
    and never through `absltest.main', which would re-enter `app.run' and, in
    the shapes a caller normally uses, spawn;
  * signal delivery does not work at all (brintos/linux#21) and alarm(2) was
    unwired before the brintos/linux#77 fix, so nothing here arms a timeout of
    its own -- the suite's `bst_probe' bound is the only one;
  * `absl.logging' reaches socket.gethostname() and getpass.getuser() ONLY on
    the log-to-FILE path (`find_log_dir_and_names'), so this probe leaves
    `--logtostderr' alone and redirects the handler's stream instead. That is
    also why no case here needs a hostname the guest may not have.

ONE THING IS NEVER CALLED, ON PURPOSE. `absl.logging.fatal' -- and any FATAL
record created from an ABSLLogger -- ends in `os.abort()' inside
`PythonHandler.emit'. There is no exception to catch and no exit code to read:
the interpreter dies mid-suite and every case after it is lost. The FATAL
severity is therefore asserted through `converter.get_initial_for_level', which
is the same function that renders the prefix, and never by emitting one.

OUTPUT CONTRACT. `key=value' lines on stdout, one per line, flushed as they are
written, so a probe killed mid-run still delivers everything before the kill.
EVERY STRING IS WRAPPED IN ANGLE BRACKETS and the empty string is `<>' --
load-bearing, because several answers here are legitimately empty (a residual
argv with no positionals, a captured stream that must stay silent) and
`grep | cut' cannot otherwise tell an empty ANSWER from a key that was never
reached. A refusal is `!<ExceptionName>'.

TWO THINGS ARE DELIBERATELY NOT ASSERTED BY VALUE. The log prefix carries a
timestamp and a thread id -- `%c%02d%02d %02d:%02d:%02d.%06d %5d %s:%d]' -- so
the probe matches it against that SHAPE here and emits the stable fields
(severity character, source file, message) separately. Asserting the rendered
line would be asserting the clock.
"""
import os
import sys

EMITTED = 0


def emit(key, value):
    global EMITTED
    EMITTED += 1
    sys.stdout.write("%s=%s\n" % (key, value))
    sys.stdout.flush()


W = None            # the probe's own scratch dir, set below


def wrap(v):
    # mkdtemp picks a fresh suffix every boot and the path appears in the
    # generated test module's name and in tracebacks, so it is replaced by a
    # literal the suite can assert against.
    text = "%s" % (v,)
    if W:
        text = text.replace(W, "WORKDIR")
    return "<%s>" % text


def emit_s(key, value):
    emit(key, wrap(value))


def fail(key, exc):
    emit(key, "!<%s>" % type(exc).__name__)


def guard(key, fn):
    """Emit fn()'s value, or `!<Exception>' -- one case never takes out the rest."""
    try:
        emit_s(key, fn())
    except Exception as exc:                       # noqa: BLE001
        fail(key, exc)


def raises(key, fn):
    """Emit the exception type fn() raises, or `<no-raise>' if it returns."""
    try:
        fn()
        emit_s(key, "no-raise")
    except Exception as exc:                       # noqa: BLE001
        fail(key, exc)


import io                                          # noqa: E402
import re                                          # noqa: E402
import tempfile                                    # noqa: E402
import unittest                                    # noqa: E402

# EVERY absl ENTRY POINT IS IMPORTED HERE, BEFORE ANY FLAG IS PARSED. Two
# reasons, and the second is not optional: `absl.testing.absltest' DEFINES
# flags on the global FLAGS at module scope (--test_srcdir, --test_tmpdir,
# --test_random_seed, --test_randomize_ordering_seed, --xml_output_file), and a
# flag defined after `FLAGS(argv)' has run is not parsed by it. Importing first
# is what makes the `app.run' section below a statement about a realistic flag
# set rather than about five flags absl never saw.
from absl import app                               # noqa: E402
from absl import command_name                      # noqa: E402
from absl import flags                             # noqa: E402
from absl import logging as alog                   # noqa: E402
from absl.flags import argparse_flags              # noqa: E402
from absl.logging import converter                 # noqa: E402
from absl.testing import absltest                  # noqa: E402
from absl.testing import flagsaver                 # noqa: E402
from absl.testing import parameterized             # noqa: E402
from absl.testing import _bazelize_command         # noqa: E402,F401
from absl.testing import _pretty_print_reporter    # noqa: E402
from absl.testing import xml_reporter              # noqa: E402

import absl                                        # noqa: E402

W = tempfile.mkdtemp(prefix="bst-absl-")
emit_s("fixture_built", "yes" if os.path.isdir(W) else "no")

# ------------------------------------------------------------- identity ------
emit_s("module_file", absl.__file__)
emit_s("is_package", "yes" if hasattr(absl, "__path__") else "no")
guard("version_attr", lambda: absl.__version__)

import importlib.metadata as _im                   # noqa: E402

guard("dist_version", lambda: _im.version("absl-py"))
guard("dist_name", lambda: _im.distribution("absl-py").metadata["Name"])
# The distribution is `absl-py' and the import name is `absl'. Asked the other
# way round, `importlib.metadata' must not know an `absl' distribution at all.
raises("dist_absl_unknown", lambda: _im.version("absl"))
guard("dist_requires", lambda: "|".join(_im.distribution("absl-py").requires or []))
guard("dist_top_level",
      lambda: (_im.distribution("absl-py").read_text("top_level.txt") or "").strip())

for _name in ("app", "command_name", "flags", "flags.argparse_flags",
              "logging", "logging.converter", "testing", "testing.absltest",
              "testing.flagsaver", "testing.parameterized",
              "testing._pretty_print_reporter", "testing._bazelize_command",
              "testing.xml_reporter"):
    _key = "submodule_" + _name.replace(".", "_")
    emit_s(_key, "yes" if ("absl." + _name) in sys.modules else "no")

# NO COMPILED EXTENSION ANYWHERE UNDER absl/. This image's python3 is ONE STATIC
# BINARY with an empty lib-dynload and no dynamic loader (BRINTOS-1476), so a
# module carrying a .so could not be imported here at all. Counted from the
# installed tree rather than from the recipe.
_pkgdir = os.path.dirname(absl.__file__)


def _count(suffix):
    n = 0
    for root, dirs, files in os.walk(_pkgdir):
        n += len([f for f in files if f.endswith(suffix)])
    return n


emit_s("tree_py_count", _count(".py"))
emit_s("tree_pyi_count", _count(".pyi"))
emit_s("tree_so_count", _count(".so"))

# ================================================================= flags ======
# EVERY FLAG CASE USES ITS OWN `FlagValues'. The global `flags.FLAGS' is a
# process-wide registry that `absltest' and `app' have both already written to,
# so a case declaring `--port' on it would be asserting against whatever else
# this interpreter has defined. A fresh FlagValues is the documented way to hold
# a flag set of one's own, and it is what makes "declared, then parsed, then
# asserted" a statement about THIS declaration.

# --------------------------------------------- one flag of each type ---------
_fv = flags.FlagValues()
flags.DEFINE_string("sflag", "sdefault", "a string flag", flag_values=_fv)
flags.DEFINE_integer("iflag", 7, "an integer flag", flag_values=_fv)
flags.DEFINE_float("fflag", 1.5, "a float flag", flag_values=_fv)
flags.DEFINE_boolean("bflag", False, "a boolean flag", flag_values=_fv)
flags.DEFINE_enum("eflag", "red", ["red", "green", "blue"], "an enum flag",
                  flag_values=_fv)
flags.DEFINE_list("lflag", ["a", "b"], "a list flag", flag_values=_fv)
flags.DEFINE_multi_string("mflag", ["x"], "a multi-string flag", flag_values=_fv)

emit_s("declared_count", len(_fv))
emit_s("declared_names", "|".join(sorted(_fv)))

# THE ARGUMENT VECTOR IS THE TEST'S, NEVER THE PROCESS'S. `FlagValues.__call__'
# returns the residual argv INCLUDING argv[0], which is the contract `app.run'
# relies on and is asserted as such below.
_argv = ["bstprog", "--sflag=hello", "--iflag=42", "--fflag=2.5", "--bflag",
         "--eflag=green", "--lflag=p,q,r", "--mflag=one", "--mflag=two",
         "pos1", "pos2"]
guard("parse_residual", lambda: "|".join(_fv(_argv)))
guard("parse_is_parsed", lambda: "yes" if _fv.is_parsed() else "no")

guard("value_string", lambda: _fv.sflag)
guard("value_integer", lambda: _fv.iflag)
guard("value_integer_type", lambda: type(_fv.iflag).__name__)
guard("value_float", lambda: _fv.fflag)
guard("value_float_type", lambda: type(_fv.fflag).__name__)
guard("value_boolean", lambda: _fv.bflag)
guard("value_boolean_type", lambda: type(_fv.bflag).__name__)
guard("value_enum", lambda: _fv.eflag)
guard("value_list", lambda: "|".join(_fv.lflag))
guard("value_list_type", lambda: type(_fv.lflag).__name__)
guard("value_multi_string", lambda: "|".join(_fv.mflag))
guard("value_multi_string_type", lambda: type(_fv.mflag).__name__)

# `present' counts how many times each flag appeared. It is the only thing that
# separates "the value equals the default" from "the flag was absent", and it is
# what a caller checks before treating a default as a choice.
guard("present_string", lambda: _fv["sflag"].present)
guard("present_multi_string", lambda: _fv["mflag"].present)

# ------------------------------------- the same flags, none of them given ----
# THE DEFAULT IS ASSERTED AS A PARSE RESULT, not by reading `.default' back. A
# flag whose default never reaches `.value' is the failure this catches, and
# reading the attribute the declaration wrote cannot see it.
_fvd = flags.FlagValues()
flags.DEFINE_string("sflag", "sdefault", "a string flag", flag_values=_fvd)
flags.DEFINE_integer("iflag", 7, "an integer flag", flag_values=_fvd)
flags.DEFINE_float("fflag", 1.5, "a float flag", flag_values=_fvd)
flags.DEFINE_boolean("bflag", True, "a boolean flag", flag_values=_fvd)
flags.DEFINE_enum("eflag", "red", ["red", "green", "blue"], "an enum flag",
                  flag_values=_fvd)
flags.DEFINE_list("lflag", ["a", "b"], "a list flag", flag_values=_fvd)
flags.DEFINE_multi_string("mflag", ["x"], "a multi-string flag", flag_values=_fvd)
guard("default_residual", lambda: "|".join(_fvd(["bstprog"])))
guard("default_string", lambda: _fvd.sflag)
guard("default_integer", lambda: _fvd.iflag)
guard("default_float", lambda: _fvd.fflag)
guard("default_boolean", lambda: _fvd.bflag)
guard("default_enum", lambda: _fvd.eflag)
guard("default_list", lambda: "|".join(_fvd.lflag))
guard("default_multi_string", lambda: "|".join(_fvd.mflag))
guard("default_present_zero", lambda: _fvd["sflag"].present)

# ------------------------------------------------- the boolean spellings -----
# FOUR SPELLINGS, and `--nobflag' is the one the ticket names. `--bflag=false'
# and the bare `--bflag' are asserted beside it because they are the same parser
# arm and a regression in it would show in only one of the three.
def _bool_run(args, default=True):
    fv = flags.FlagValues()
    flags.DEFINE_boolean("bflag", default, "a boolean flag", flag_values=fv)
    fv(["bstprog"] + args)
    return fv.bflag


guard("bool_no_form", lambda: _bool_run(["--nobflag"]))
guard("bool_bare_form", lambda: _bool_run(["--bflag"], default=False))
guard("bool_eq_false", lambda: _bool_run(["--bflag=false"]))
guard("bool_eq_true", lambda: _bool_run(["--bflag=true"], default=False))
guard("bool_eq_zero", lambda: _bool_run(["--bflag=0"]))
raises("bool_eq_garbage", lambda: _bool_run(["--bflag=maybe"]))

# ----------------------------------------------------------- validators ------
# A VALIDATOR IS ASSERTED IN BOTH DIRECTIONS, because a validator that never
# fires and a validator that always fires are both broken and only the pair
# tells them apart. It fires AT PARSE TIME, which is why each arm re-parses
# rather than assigning `.value'.
def _validated(value):
    fv = flags.FlagValues()
    flags.DEFINE_integer("port", 80, "a port number", flag_values=fv)
    flags.register_validator(
        "port", lambda v: v is None or 1 <= v <= 65535,
        message="port out of range", flag_values=fv)
    fv(["bstprog", "--port=%s" % value])
    return fv.port


guard("validator_accepts", lambda: _validated(8080))
raises("validator_rejects", lambda: _validated(70000))


def _validator_message():
    try:
        _validated(70000)
    except flags.IllegalFlagValueError as exc:
        return str(exc)
    return "no-raise"


guard("validator_message", _validator_message)

# A MULTI-FLAG VALIDATOR, the other half of the API: it sees the whole dict and
# fires on a RELATION between flags rather than on one value.
def _multi_validated(lo, hi):
    fv = flags.FlagValues()
    flags.DEFINE_integer("lo", 1, "lower bound", flag_values=fv)
    flags.DEFINE_integer("hi", 9, "upper bound", flag_values=fv)
    flags.register_multi_flags_validator(
        ["lo", "hi"], lambda d: d["lo"] <= d["hi"],
        message="lo must not exceed hi", flag_values=fv)
    fv(["bstprog", "--lo=%d" % lo, "--hi=%d" % hi])
    return "%d..%d" % (fv.lo, fv.hi)


guard("multi_validator_accepts", lambda: _multi_validated(2, 5))
raises("multi_validator_rejects", lambda: _multi_validated(5, 2))

# `mark_flag_as_required' is the third validator shape and the one with a
# distinct exception.
def _required(given):
    fv = flags.FlagValues()
    flags.DEFINE_string("must", None, "a required flag", flag_values=fv)
    flags.mark_flag_as_required("must", flag_values=fv)
    fv(["bstprog"] + (["--must=here"] if given else []))
    return fv.must


guard("required_given", lambda: _required(True))
raises("required_missing", lambda: _required(False))

# --------------------------------------------- what flag_values reports ------
# THE TICKET'S "flag_values reports the declared help text and default". Both
# are read off the registry rather than off the variable the declaration
# returned, because the registry is what `--help' renders from.
guard("report_help_string", lambda: _fv["sflag"].help)
guard("report_help_integer", lambda: _fv["iflag"].help)
guard("report_default_string", lambda: _fv["sflag"].default)
guard("report_default_integer", lambda: _fv["iflag"].default)
guard("report_default_as_str_string", lambda: _fv["sflag"].default_as_str)
guard("report_default_as_str_integer", lambda: _fv["iflag"].default_as_str)
guard("report_default_as_str_bool", lambda: _fvd["bflag"].default_as_str)
guard("report_default_list", lambda: "|".join(_fv["lflag"].default))
guard("report_enum_values", lambda: "|".join(_fv["eflag"].parser.enum_values))
guard("report_flag_type_string", lambda: _fv["sflag"].flag_type())
guard("report_flag_type_integer", lambda: _fv["iflag"].flag_type())
guard("report_flag_type_enum", lambda: _fv["eflag"].flag_type())
guard("report_flag_type_list", lambda: _fv["lflag"].flag_type())
guard("report_flag_type_multi", lambda: _fv["mflag"].flag_type())
guard("report_name", lambda: _fv["sflag"].name)
guard("report_serialize_enum", lambda: _fv["eflag"].serialize())
guard("report_serialize_bool_false", lambda: _fvd["bflag"].serialize())
guard("report_parser_string", lambda: type(_fv["sflag"].parser).__name__)
guard("report_parser_integer", lambda: type(_fv["iflag"].parser).__name__)
guard("report_parser_enum", lambda: type(_fv["eflag"].parser).__name__)
# THE RENDERED HELP BLOCK, whole. `--help' is what a caller sees when they get
# the flag wrong, so the help text and the default are asserted in the layout
# they are actually presented in and not only as attributes. Rendered from a
# two-flag registry of its own, so the literal stays readable; newlines are
# joined with `|' for the transport.
#
# THE FIRST TWO RENDERED LINES ARE DROPPED and that is not tidying. absl groups
# the help by DEFINING MODULE and heads each group with that module's path --
# here the probe's own, which is `$BST_DATA/...' on target and whatever the
# control was invoked as off it. Asserting it would be asserting the invocation.
# What is dropped is one blank line and that heading; every line that carries a
# flag is kept.
_fvh = flags.FlagValues()
flags.DEFINE_string("sflag", "sdefault", "a string flag", flag_values=_fvh)
flags.DEFINE_enum("eflag", "red", ["red", "green", "blue"], "an enum flag",
                  flag_values=_fvh)
guard("report_rendered_help",
      lambda: "|".join(_fvh.get_help(include_special_flags=False).splitlines()[2:]))

# -------------------------------------------------- the duplicate name -------
def _duplicate():
    fv = flags.FlagValues()
    flags.DEFINE_string("dup", "first", "the first declaration", flag_values=fv)
    flags.DEFINE_string("dup", "second", "the second declaration", flag_values=fv)


raises("duplicate_raises", _duplicate)
# `DuplicateFlagError' is a subclass of `flags.Error' and NOT of ValueError, so
# a caller writing `except flags.Error' catches it and one writing
# `except ValueError' does not. Both are asserted: the second is the mistake.
emit_s("duplicate_is_flags_error",
       "yes" if issubclass(flags.DuplicateFlagError, flags.Error) else "no")
emit_s("duplicate_is_value_error",
       "yes" if issubclass(flags.DuplicateFlagError, ValueError) else "no")
emit_s("illegal_is_flags_error",
       "yes" if issubclass(flags.IllegalFlagValueError, flags.Error) else "no")

# ---------------------------------------------- the other parse refusals -----
def _parse_only(decl, args):
    fv = flags.FlagValues()
    decl(fv)
    fv(["bstprog"] + args)
    return "no-raise"


raises("reject_bad_integer",
       lambda: _parse_only(
           lambda fv: flags.DEFINE_integer("n", 1, "h", flag_values=fv),
           ["--n=notanumber"]))
raises("reject_bad_enum",
       lambda: _parse_only(
           lambda fv: flags.DEFINE_enum("c", "red", ["red", "green"], "h",
                                        flag_values=fv),
           ["--c=purple"]))
raises("reject_unknown_flag",
       lambda: _parse_only(
           lambda fv: flags.DEFINE_string("x", "a", "h", flag_values=fv),
           ["--nosuchflag=1"]))

# `--' ends the flags and everything after it is positional, even a `--word'.
def _double_dash():
    fv = flags.FlagValues()
    flags.DEFINE_string("x", "a", "h", flag_values=fv)
    return "|".join(fv(["bstprog", "--x=set", "--", "--x=notaflag", "tail"]))


guard("double_dash_residual", _double_dash)

# ================================================================== app =======
# `app.run' IS DRIVEN WITH AN ARGUMENT VECTOR, never with the process's own.
# That is the ticket's wording and it is also the only way this is testable at
# all: sys.argv here is the probe's own path.
#
# WHAT app.run DOES ON THE WAY IN, and why none of it is a hazard on this arm.
# `_run_init' calls `command_name.make_process_name_useful()' (writes
# /proc/self/comm, and on EnvironmentError falls back to a ctypes prctl that
# cannot load libc.so.6 on a static interpreter -- both arms are caught
# upstream, and `command_name_survives' below asserts the call returns rather
# than what it achieved), then `logging.use_absl_handler()', then
# `faulthandler.enable()' inside a bare `except Exception'. Signal DELIVERY is
# broken here (brintos/linux#21) but faulthandler only REGISTERS handlers, so
# this is a registration that nothing will ever fire -- which is exactly why it
# is harmless rather than merely untested.
flags.DEFINE_string("bst_app_flag", "appdefault", "a flag for the app.run case")

_seen = {}


def _main_ok(argv):
    _seen["argv"] = list(argv)
    _seen["flag"] = flags.FLAGS.bst_app_flag
    return 7


def _run(main, argv):
    try:
        app.run(main, argv=argv)
    except SystemExit as exc:
        return "SystemExit:%r" % (exc.code,)
    return "returned"


emit_s("app_run_exit", _run(_main_ok, ["bstprog", "--bst_app_flag=set",
                                       "posA", "posB"]))
guard("app_main_argv", lambda: "|".join(_seen["argv"]))
guard("app_main_flag", lambda: _seen["flag"])
# The residual carries argv[0] and NOTHING that parsed as a flag. A `main' that
# indexes argv[1] is the ordinary shape, so an off-by-one here is a real bug.
guard("app_main_argv0", lambda: _seen["argv"][0])
guard("app_main_positionals", lambda: "|".join(_seen["argv"][1:]))

emit_s("app_flags_parsed", "yes" if flags.FLAGS.is_parsed() else "no")
emit_s("command_name_survives",
       "yes" if command_name.make_process_name_useful() is None else "no")


def _main_zero(argv):
    return 0


def _main_none(argv):
    return None


def _main_usage(argv):
    raise app.UsageError("a deliberate usage error")


emit_s("app_run_exit_zero", _run(_main_zero, ["bstprog"]))
emit_s("app_run_exit_none", _run(_main_none, ["bstprog"]))
# `UsageError' writes the usage block to stderr and exits with its own code.
# stderr is captured so the suite's `probe-quiet' case stays a statement about
# everything else.
_saved_err, sys.stderr = sys.stderr, io.StringIO()
try:
    _usage_exit = _run(_main_usage, ["bstprog"])
    _usage_text = sys.stderr.getvalue()
finally:
    sys.stderr = _saved_err
emit_s("app_run_exit_usage", _usage_exit)
emit_s("app_usage_mentions_flags", "yes" if "flags:" in _usage_text else "no")
emit_s("app_usage_mentions_error",
       "yes" if "a deliberate usage error" in _usage_text else "no")

# An unknown flag in the vector is `app.run's other refusal, and it exits 1
# through the same path rather than propagating the FlagsError.
_saved_err, sys.stderr = sys.stderr, io.StringIO()
try:
    _unknown_exit = _run(_main_zero, ["bstprog", "--no_such_flag_at_all"])
finally:
    sys.stderr = _saved_err
emit_s("app_run_exit_unknown_flag", _unknown_exit)

# `call_after_init' after initialisation is documented to run the callback
# immediately, in the caller's thread. There is no thread here to get that
# wrong, which is the point: it is asserted as a synchronous call.
_after = []
app.call_after_init(lambda: _after.append("ran"))
emit_s("app_call_after_init", "|".join(_after))

# ============================================================== logging =======
# CAPTURED, NOT READ OFF THE CONSOLE. `PythonHandler.emit' routes to stderr
# whenever FLAGS are unparsed or `--logtostderr' is set; app.run above has
# parsed them, so setting the handler's stream to a StringIO is enough, and
# `--stderrthreshold=fatal' keeps the copy off stderr. Nothing here logs at
# FATAL: an ABSL FATAL record calls os.abort() and would end the probe.
_h = alog.get_absl_handler()
flags.FLAGS.logtostderr = False
flags.FLAGS.alsologtostderr = False
flags.FLAGS.stderrthreshold = "fatal"

_LINE = re.compile(
    r"^([IWEF])(\d{2})(\d{2}) (\d{2}):(\d{2}):(\d{2})\.(\d{6}) +(\d+) "
    r"([^:]+):(\d+)\] (.*)$")


def _capture(fn, verbosity=None):
    """Run fn() with absl's handler pointed at a StringIO; return the text."""
    buf = io.StringIO()
    old_stream = _h.python_handler.stream
    old_verbosity = alog.get_verbosity()
    _h.python_handler.stream = buf
    if verbosity is not None:
        alog.set_verbosity(verbosity)
    try:
        fn()
        _h.flush()
    finally:
        _h.python_handler.stream = old_stream
        alog.set_verbosity(old_verbosity)
    return buf.getvalue()


emit_s("log_handler_type", type(_h).__name__)
emit_s("log_python_handler_type", type(_h.python_handler).__name__)
emit_s("log_default_verbosity", alog.get_verbosity())

# THE FOUR SEVERITY CONSTANTS ARE absl LEVELS, NOT stdlib ONES, and they run the
# other way: DEBUG is 1 and FATAL is -3, so "more verbose" is a LARGER number.
# A caller who passes `logging.INFO' from the stdlib module gets 20, which is
# `--verbosity=20' -- every debug level at once. The pair is asserted so the
# inversion is on the record.
emit_s("log_level_debug", alog.DEBUG)
emit_s("log_level_info", alog.INFO)
emit_s("log_level_warning", alog.WARNING)
emit_s("log_level_error", alog.ERROR)
emit_s("log_level_fatal", alog.FATAL)

# EACH SEVERITY RENDERS WITH ITS DOCUMENTED PREFIX. The prefix also carries a
# timestamp and a thread id, so the shape is matched here and only the stable
# fields are emitted -- asserting the rendered line would be asserting the clock.
_sev_text = _capture(lambda: (alog.info("info-msg"),
                              alog.warning("warn-msg"),
                              alog.error("error-msg"),
                              alog.debug("debug-msg")),
                     verbosity=alog.INFO)
_sev_lines = _sev_text.splitlines()
emit_s("log_severity_line_count", len(_sev_lines))
for _i, _want in enumerate(("info", "warning", "error")):
    _m = _LINE.match(_sev_lines[_i]) if _i < len(_sev_lines) else None
    emit_s("log_shape_" + _want, "yes" if _m else "no")
    emit_s("log_initial_" + _want, _m.group(1) if _m else "no-match")
    emit_s("log_message_" + _want, _m.group(11) if _m else "no-match")
    emit_s("log_source_" + _want,
           os.path.basename(_m.group(9)) if _m else "no-match")

# `debug' AT verbosity=INFO IS THE ABSENT ONE. Its absence is what makes the
# line count above a filter assertion rather than a coincidence.
emit_s("log_debug_suppressed_at_info",
       "yes" if "debug-msg" not in _sev_text else "no")

# The severity initials come from `converter.get_initial_for_level', which is
# also what renders the prefix -- so FATAL's `F' is asserted here rather than by
# emitting a FATAL record, which would os.abort() the probe.
guard("log_initials_by_level",
      lambda: "|".join(converter.get_initial_for_level(l)
                       for l in (10, 20, 30, 40, 50)))
guard("log_absl_to_standard",
      lambda: "|".join(str(converter.absl_to_standard(l))
                       for l in (alog.DEBUG, alog.INFO, alog.WARNING,
                                 alog.ERROR, alog.FATAL)))
guard("log_standard_to_absl",
      lambda: "|".join(str(converter.standard_to_absl(l))
                       for l in (10, 20, 30, 40, 50)))

# `--verbosity' FILTERS. At WARNING an info() must not reach the stream and a
# warning() must; both halves are asserted, since a filter that drops everything
# passes the first on its own.
_filt_text = _capture(lambda: (alog.info("filtered-away"),
                               alog.warning("kept")),
                      verbosity=alog.WARNING)
emit_s("log_filter_line_count", len(_filt_text.splitlines()))
emit_s("log_filter_info_absent", "yes" if "filtered-away" not in _filt_text else "no")
emit_s("log_filter_warning_present", "yes" if "kept" in _filt_text else "no")

# At `--verbosity=1' (absl DEBUG) the debug record arrives, which is the same
# filter read from the other side.
_dbg_text = _capture(lambda: alog.debug("debug-arrives"), verbosity=alog.DEBUG)
emit_s("log_debug_present_at_debug", "yes" if "debug-arrives" in _dbg_text else "no")
_dbg_m = _LINE.match(_dbg_text.splitlines()[0]) if _dbg_text.splitlines() else None
emit_s("log_initial_debug", _dbg_m.group(1) if _dbg_m else "no-match")

# `log_every_n' EMITS THE DOCUMENTED FRACTION: the 1st call and every nth after
# it, so ten calls at n=3 are four lines carrying 0, 3, 6, 9. The MESSAGES are
# asserted, not just the count -- a counter that emitted the last four calls
# would give the same count and be wrong.
def _every_n():
    for i in range(10):
        alog.log_every_n(alog.INFO, "every3 %d", 3, i)


_evn_text = _capture(_every_n, verbosity=alog.INFO)
_evn_msgs = [_LINE.match(l).group(11) for l in _evn_text.splitlines()
             if _LINE.match(l)]
emit_s("log_every_n_count", len(_evn_msgs))
emit_s("log_every_n_messages", "|".join(_evn_msgs))

# `log_first_n' is the same counter with the other stop condition.
def _first_n():
    for i in range(10):
        alog.log_first_n(alog.INFO, "first2 %d", 2, i)


_fst_text = _capture(_first_n, verbosity=alog.INFO)
_fst_msgs = [_LINE.match(l).group(11) for l in _fst_text.splitlines()
             if _LINE.match(l)]
emit_s("log_first_n_count", len(_fst_msgs))
emit_s("log_first_n_messages", "|".join(_fst_msgs))

# `%'-STYLE INTERPOLATION IS DEFERRED to the formatter, as in the stdlib module,
# so the arguments are passed and not pre-formatted by the caller.
_fmt_text = _capture(lambda: alog.info("a=%s b=%d", "str", 5), verbosity=alog.INFO)
_fmt_m = _LINE.match(_fmt_text.splitlines()[0]) if _fmt_text.splitlines() else None
emit_s("log_interpolation", _fmt_m.group(11) if _fmt_m else "no-match")

# `exception()' renders at ERROR and appends the traceback under the line.
def _exc():
    try:
        raise ValueError("deliberate")
    except ValueError:
        alog.exception("caught it")


_exc_text = _capture(_exc, verbosity=alog.INFO)
_exc_lines = _exc_text.splitlines()
_exc_m = _LINE.match(_exc_lines[0]) if _exc_lines else None
emit_s("log_exception_initial", _exc_m.group(1) if _exc_m else "no-match")
emit_s("log_exception_message", _exc_m.group(11) if _exc_m else "no-match")
emit_s("log_exception_has_traceback",
       "yes" if "ValueError: deliberate" in _exc_text else "no")

# `logging.get_absl_logger()' is an ABSLLogger and it is the one the module
# functions write through -- the level set on it is what `set_verbosity' moved.
emit_s("log_logger_type", type(alog.get_absl_logger()).__name__)
emit_s("log_logger_name", alog.get_absl_logger().name)
guard("log_level_roundtrip", lambda: (alog.set_verbosity(alog.WARNING),
                                      alog.get_verbosity())[1])
alog.set_verbosity(alog.INFO)
guard("log_level_named_roundtrip", lambda: (alog.set_verbosity("error"),
                                            alog.get_verbosity())[1])
alog.set_verbosity(alog.INFO)

# NOTHING IN THIS SUITE OPENS A LOG FILE. `use_absl_log_file' is the arm that
# reaches `getpass.getuser()' and `socket.gethostname()'; it is named here so a
# later reader can see the omission is deliberate, and the two functions it
# would call are asserted to EXIST rather than driven.
emit_s("log_file_api_present",
       "yes" if hasattr(_h, "use_absl_log_file") else "no")
emit_s("log_find_log_dir_present",
       "yes" if hasattr(alog, "find_log_dir_and_names") else "no")

# ================================================ absltest and parameterized ==
# A REAL MODULE ON DISK, imported by path. The ticket asks for "a small test
# module", and a module is what `unittest.TestLoader.loadTestsFromModule' takes;
# defining the classes inline here would exercise the loader's other arm and
# would not be the thing the ticket names.
#
# RUN IN-PROCESS, NOT THROUGH absltest.main. `absltest.main' re-enters `app.run'
# and ends in `sys.exit', and the shapes a caller normally reaches for from
# there spawn a child. fork(2)/exec(2) is minutes on this arm and popen(3) is
# ENOSYS (brintos/linux#14), so the module is loaded and run through
# `unittest.TestLoader' and absl's OWN runner -- `_pretty_print_reporter.
# TextTestRunner', which is the runner `absltest.main' installs. The reporter,
# the TestCase base and the parameterized decorators are therefore all the real
# ones; only the process boundary is absent.
_MODULE = '''\
"""A small absltest module, written by the probe and run in-process."""
import unittest

from absl.testing import absltest
from absl.testing import parameterized


class BstAbslCase(parameterized.TestCase):
  """Built to produce a KNOWN summary: 6 pass, 1 fail, 1 error, 1 skip, 1 xfail."""

  @parameterized.parameters((1, 2, 3), (2, 3, 5), (10, 5, 15))
  def test_add(self, a, b, c):
    self.assertEqual(a + b, c)

  @parameterized.named_parameters(('lower', 'ABC', 'abc'), ('same', 'abc', 'abc'))
  def test_lower(self, given, want):
    self.assertEqual(given.lower(), want)

  def test_absltest_assertions(self):
    # absltest.TestCase's own additions, which are the reason to subclass it
    # rather than unittest.TestCase.
    self.assertLen([1, 2, 3], 3)
    self.assertEmpty([])
    self.assertNotEmpty([0])
    self.assertStartsWith('abcdef', 'abc')
    self.assertEndsWith('abcdef', 'def')
    self.assertContainsSubset([1, 2], [1, 2, 3])
    self.assertSameElements([3, 1, 2], [1, 2, 3])
    self.assertCountEqual([1, 1, 2], [1, 2, 1])
    self.assertBetween(5, 1, 10)
    self.assertRaisesWithLiteralMatch(ValueError, 'boom', self._boom)

  def _boom(self):
    raise ValueError('boom')

  def test_deliberate_failure(self):
    self.assertEqual(1, 2)

  def test_deliberate_error(self):
    raise RuntimeError('deliberate error')

  @unittest.skip('deliberately skipped')
  def test_deliberate_skip(self):
    pass

  @unittest.expectedFailure
  def test_deliberate_expected_failure(self):
    self.assertEqual(1, 2)


if __name__ == '__main__':
  absltest.main()
'''

_modpath = os.path.join(W, "bst_absl_module.py")
with open(_modpath, "w") as _f:
    _f.write(_MODULE)

import importlib.util                              # noqa: E402

_spec = importlib.util.spec_from_file_location("bst_absl_module", _modpath)
_mod = importlib.util.module_from_spec(_spec)
sys.modules["bst_absl_module"] = _mod
_spec.loader.exec_module(_mod)
emit_s("testmod_imported", "yes" if hasattr(_mod, "BstAbslCase") else "no")
emit_s("testmod_case_is_absltest",
       "yes" if issubclass(_mod.BstAbslCase, absltest.TestCase) else "no")
emit_s("testmod_case_is_unittest",
       "yes" if issubclass(_mod.BstAbslCase, unittest.TestCase) else "no")
emit_s("testmod_parameterized_is_absltest",
       "yes" if issubclass(parameterized.TestCase, absltest.TestCase) else "no")

_suite = unittest.TestLoader().loadTestsFromModule(_mod)
_names = sorted(t.id().rsplit(".", 1)[-1] for t in _suite._tests[0])
emit_s("testmod_count", _suite.countTestCases())
emit_s("testmod_names", "|".join(_names))

_runbuf = io.StringIO()
_runner = _pretty_print_reporter.TextTestRunner(stream=_runbuf, verbosity=0)
_result = _runner.run(_suite)
_runtext = _runbuf.getvalue()

emit_s("testmod_runner_type", type(_runner).__name__)
emit_s("testmod_result_type", type(_result).__name__)
emit_s("testmod_tests_run", _result.testsRun)
emit_s("testmod_failures", len(_result.failures))
emit_s("testmod_errors", len(_result.errors))
emit_s("testmod_skipped", len(_result.skipped))
emit_s("testmod_expected_failures", len(_result.expectedFailures))
emit_s("testmod_unexpected_successes", len(_result.unexpectedSuccesses))
emit_s("testmod_was_successful", "yes" if _result.wasSuccessful() else "no")

# THE SUMMARY LINE THE RUNNER PRINTS, which is what a caller reading the console
# actually sees, and the counts above read from the result object. Both, because
# they are produced by different code and only the pair says the reporter agrees
# with the result.
_summary = [l for l in _runtext.splitlines() if l.startswith("FAILED")
            or l.startswith("OK")]
emit_s("testmod_summary", _summary[0] if _summary else "no-summary")
emit_s("testmod_ran_line",
       "yes" if re.search(r"^Ran 10 tests in ", _runtext, re.M) else "no")
emit_s("testmod_reports_failure",
       "yes" if "FAIL: test_deliberate_failure" in _runtext else "no")
emit_s("testmod_reports_error",
       "yes" if "ERROR: test_deliberate_error" in _runtext else "no")
emit_s("testmod_failure_detail",
       "yes" if "AssertionError: 1 != 2" in _runtext else "no")
emit_s("testmod_error_detail",
       "yes" if "RuntimeError: deliberate error" in _runtext else "no")
emit_s("testmod_failed_test_name",
       "|".join(sorted(t.id().rsplit(".", 1)[-1] for t, _ in _result.failures)))
emit_s("testmod_errored_test_name",
       "|".join(sorted(t.id().rsplit(".", 1)[-1] for t, _ in _result.errors)))
emit_s("testmod_skipped_reason",
       "|".join(r for _, r in _result.skipped))

# THE ONE TEST THAT EXERCISES absltest's OWN ASSERTIONS is named here with its
# outcome, rather than left to be inferred from the counts. assertLen,
# assertEmpty, assertStartsWith, assertContainsSubset, assertSameElements,
# assertBetween and assertRaisesWithLiteralMatch are the reason to subclass
# `absltest.TestCase' at all, and "it is not among the failures" is a claim a
# reader should not have to assemble out of two other keys.
_bad_ids = set(t.id() for t, _ in _result.failures) | set(
    t.id() for t, _ in _result.errors)
emit_s("testmod_assertions_case",
       "fail" if any(i.endswith(".test_absltest_assertions") for i in _bad_ids)
       else "pass")

# `parameterized' EXPANDS ONE METHOD INTO N TESTS, and the generated names are
# part of its contract: `@parameters' numbers them and appends the repr of the
# tuple, `@named_parameters' appends the name given. Both are in
# `testmod_names' above; these two say how many each decorator produced.
emit_s("parameterized_from_parameters",
       len([n for n in _names if n.startswith("test_add")]))
emit_s("parameterized_from_named",
       len([n for n in _names if n.startswith("test_lower")]))

# ------------------------------------------------ absltest's own helpers -----
# `create_tempdir'/`create_tempfile' are the absltest APIs that touch the
# filesystem. They are exercised HERE rather than inside the module above, so a
# filesystem refusal is one red case and not a wrong test count.
# Whether absltest's tempdir root already existed decides whether the cleanup
# below may remove it: it is `--test_tmpdir', which belongs to the image and not
# to this probe.
_ttd = flags.FLAGS.test_tmpdir
_ttd_preexisting = os.path.isdir(_ttd)

_tc = _mod.BstAbslCase("test_absltest_assertions")
_tc.setUp()
guard("helper_tempdir_created",
      lambda: "yes" if os.path.isdir(_tc.create_tempdir("d").full_path) else "no")


def _tempfile_roundtrip():
    f = _tc.create_tempfile("f.txt", content="bst-content\n")
    return f.read_text().strip()


guard("helper_tempfile_roundtrip", _tempfile_roundtrip)
guard("helper_tempdir_under_test_tmpdir",
      lambda: "yes" if _tc.create_tempdir("e").full_path.startswith(
          flags.FLAGS.test_tmpdir) else "no")

# absltest's tempdir root is `--test_tmpdir', which is NOT the probe's own
# scratch dir and so is not removed by the teardown below. It is cleaned here so
# the suite leaves nothing behind on a rootfs it does not own.
_tc.doCleanups()
if not _ttd_preexisting:
    __import__("shutil").rmtree(_ttd, ignore_errors=True)
emit_s("helper_tempdir_root_left_behind",
       "yes" if (not _ttd_preexisting and os.path.isdir(_ttd)) else "no")

# `flagsaver' RESTORES A FLAG AFTER THE BLOCK. Asserted on the global FLAGS,
# which is the only registry it is ever used on, and in three places: inside the
# block, after it, and against the value it started from.
flags.FLAGS.bst_app_flag = "before"
with flagsaver.flagsaver(bst_app_flag="inside"):
    emit_s("flagsaver_inside", flags.FLAGS.bst_app_flag)
emit_s("flagsaver_restored", flags.FLAGS.bst_app_flag)


@flagsaver.flagsaver(bst_app_flag="decorated")
def _decorated():
    return flags.FLAGS.bst_app_flag


guard("flagsaver_decorator", _decorated)
emit_s("flagsaver_restored_after_decorator", flags.FLAGS.bst_app_flag)

# --------------------------------------------------------- argparse_flags ----
# absl's argparse bridge, the fourth public entry point. It is a real
# `argparse.ArgumentParser' subclass that also understands absl flags, so both
# halves are driven: an argparse option and an absl flag in the same vector.
def _argparse_flags():
    parser = argparse_flags.ArgumentParser(
        inherited_absl_flags=_fv, prog="bstprog", description="probe parser")
    parser.add_argument("--native", default="ndefault")
    parser.add_argument("rest", nargs="*")
    ns = parser.parse_args(["--native=given", "--sflag=viaargparse", "tail"])
    return "%s|%s|%s" % (ns.native, "|".join(ns.rest), _fv.sflag)


guard("argparse_flags_mixed", _argparse_flags)
emit_s("argparse_flags_is_argparse",
       "yes" if issubclass(argparse_flags.ArgumentParser,
                           __import__("argparse").ArgumentParser) else "no")

# ------------------------------------------------------------ xml_reporter ---
# `xml_reporter' is what `--xml_output_file' installs. It is exercised as a
# renderer over the SAME module, into a StringIO -- no file, and no second run
# of anything.
def _xml_report():
    buf = io.StringIO()
    suite = unittest.TestLoader().loadTestsFromModule(_mod)
    runner = xml_reporter.TextAndXMLTestRunner(
        xml_stream=buf, stream=io.StringIO(), verbosity=0)
    runner.run(suite)
    text = buf.getvalue()
    m = re.search(r'<testsuites[^>]*\btests="(\d+)"[^>]*\bfailures="(\d+)"'
                  r'[^>]*\berrors="(\d+)"', text)
    if not m:
        m2 = re.search(r'tests="(\d+)"', text)
        return "no-match:%s" % (m2.group(0) if m2 else text[:60])
    return "tests=%s failures=%s errors=%s" % m.groups()


guard("xml_report_counts", _xml_report)

# -------------------------------------------------------------- teardown -----
import shutil                                      # noqa: E402

shutil.rmtree(W, ignore_errors=True)
emit_s("torn_down", "yes" if not os.path.isdir(W) else "no")

emit("emitted_count", EMITTED + 2)
emit_s("probe_done", "yes")