"""python3-installer on wasm32-hwjs: build a wheel, install it, verify the RECORD.

ONE INTERPRETER START FOR THE WHOLE OF THE BEHAVIOURAL SECTION. An exec on this
arm is a fresh wasm module instantiation at ~18 s and a boot is worth roughly
85-100 s of guest work (BRINTOS-1682), so a suite that spawned python3 once per
assertion would not finish. Every measurement below is taken in this process and
printed as one `R<n>_<name> <value>` line; 99-python3-installer.test names and
compares them, so a single wrong answer is one red case rather than one red
suite.

NOTHING HERE SETS AN ENVIRONMENT VARIABLE. setenv(3) returns EINVAL for every
name on this arm (brintos/glibc#11), so os.environ assignment raises OSError
[Errno 22]; the scratch roots arrive as argv instead.

THE FIXTURE WHEEL IS BUILT HERE, not shipped, and that is deliberate: a wheel
checked into the image would be bytes this port never produced, while a wheel
zipfile writes in this process exercises the same zlib the installer then reads
back through. It is a wheel in the PEP 427 sense and nothing more -- no build
backend runs, because the backend is the interpreter's business and not this
package's.

WHAT THE INSTALL ACTUALLY EXERCISES, and why each is worth a case here rather
than on the host: zipfile+zlib decompression of the archive, hashlib sha256 and
urlsafe base64 for every RECORD entry, csv for the RECORD itself, the email
parser for METADATA and WHEEL, os.chmod under the process umask for the script
bit, compileall for the bytecode levels, and the .data/scripts scheme rewrite
that turns `#!python` into this image's own interpreter path.
"""

import base64
import csv
import hashlib
import io
import os
import sys
import stat
import zipfile

OUT = []


def say(tag, value):
    OUT.append("%s %s" % (tag, value))


def guard(tag):
    """Report an exception as this case's value instead of ending the run.

    A payload that died at its first surprise printed nothing for the forty
    measurements after it, and the suite then reported forty empties naming a
    symptom each. Every section below is wrapped, so one broken call costs one
    case.
    """
    def deco(fn):
        try:
            fn()
        except Exception as exc:                      # noqa: BLE001
            say(tag, "EXC:%s:%s" % (type(exc).__name__, exc))
    return deco


ROOT = sys.argv[1]
WHEEL_DIR = os.path.join(ROOT, "wheel")
PREFIX1 = os.path.join(ROOT, "p1")
PREFIX2 = os.path.join(ROOT, "p2")
for d in (WHEEL_DIR, PREFIX1, PREFIX2):
    os.makedirs(d, exist_ok=True)

# --------------------------------------------------------- R0: the import ---
try:
    import installer
    import installer.__main__ as installer_main
    from installer.destinations import SchemeDictionaryDestination
    from installer.records import RecordEntry
    from installer.sources import WheelFile
    from installer.utils import get_launcher_kind
    say("R0_import", "ok")
except Exception as exc:                              # noqa: BLE001
    say("R0_import", "EXC:%s:%s" % (type(exc).__name__, exc))
    print("\n".join(OUT))
    sys.exit(1)

say("R0_version", installer.__version__)
say("R0_all", ",".join(sorted(installer.__all__)))
say("R0_install_callable", "yes" if callable(installer.install) else "no")
say("R0_launcher_kind", get_launcher_kind())

# The two things poky's interpreter.patch adds. Both are behaviour this image
# depends on -- Yocto installs every wheel it builds with this CLI and passes
# the target interpreter explicitly -- and neither exists in upstream 0.7.0, so
# a version bump that drops the patch is caught here rather than by a rootfs
# full of host shebangs.
@guard("R0_cli_interpreter_flag")
def _cli_flag():
    ns = installer_main._get_main_parser().parse_args(["w.whl", "-i", "/usr/bin/python3"])
    say("R0_cli_interpreter_flag", ns.interpreter)


@guard("R0_shebang_plain")
def _shebang_plain():
    from installer.scripts import _build_shebang
    # Upstream wraps an interpreter path containing a space in a /bin/sh
    # trampoline; the patch removes that arm, so the answer is the plain form
    # for a spaced path too. Asserting the SPACED case is the whole point --
    # the unspaced one is identical either way and would prove nothing.
    say("R0_shebang_plain",
        _build_shebang("/opt/py 3/bin/python3", False).decode("utf-8").replace("\n", "|"))


# ------------------------------------------------- R1: build the fixture ---
DIST = "bstfix"
VER = "1.2.3"
WHEEL_NAME = "%s-%s-py3-none-any.whl" % (DIST, VER)
WHEEL_PATH = os.path.join(WHEEL_DIR, WHEEL_NAME)
DISTINFO = "%s-%s.dist-info" % (DIST, VER)
DATADIR = "%s-%s.data" % (DIST, VER)

INIT_PY = "BSTFIX_MARK = 'bst-installer-fixture-ok'\n__version__ = '1.2.3'\n"
DEEP_PY = "def answer():\n    return 42\n"
SCRIPT_TEXT = "#!python\nimport sys\nprint('bst-fixture-script', sys.argv[1:])\n"
METADATA = (
    "Metadata-Version: 2.1\n"
    "Name: bstfix\n"
    "Version: 1.2.3\n"
    "Summary: brintOS selftest fixture\n"
    "\n"
    "fixture\n"
)
WHEEL_MD = "Wheel-Version: 1.0\nGenerator: brintos-selftest\nRoot-Is-Purelib: true\nTag: py3-none-any\n"
ENTRYPOINTS = "[console_scripts]\nbst-fixture-cli = bstfix:main\n"

PLAIN_TEXT = "bst-fixture-data\n"

# (archive path, contents, unix mode). THE MODE IS PART OF THE FIXTURE, not
# decoration: installer takes is_executable from the archive member's own
# external_attr and passes it through to os.chmod, so a wheel written with
# zipfile's default 0o600 arrives non-executable however correct the installer
# is. Written 0o600 first, this suite reported the script bit absent and read as
# a platform finding; it was the fixture. The two modes below are asserted as a
# PAIR -- 0o755 on the script and 0o644 on the data file -- because either one
# alone is also passed by an installer that sets the bit unconditionally, or
# never.
MEMBERS = [
    ("%s/__init__.py" % DIST, INIT_PY, 0o644),
    ("%s/deep/mod.py" % DIST, DEEP_PY, 0o644),
    ("%s/scripts/bst-fixture-script" % DATADIR, SCRIPT_TEXT, 0o755),
    ("%s/data/share/bstfix/plain.txt" % DATADIR, PLAIN_TEXT, 0o644),
    ("%s/METADATA" % DISTINFO, METADATA, 0o644),
    ("%s/WHEEL" % DISTINFO, WHEEL_MD, 0o644),
    ("%s/entry_points.txt" % DISTINFO, ENTRYPOINTS, 0o644),
]


def rec_hash(data):
    d = hashlib.sha256(data).digest()
    return "sha256=" + base64.urlsafe_b64encode(d).decode("ascii").rstrip("=")


@guard("R1_wheel_built")
def _build_wheel():
    rows = []
    buf = io.StringIO()
    with zipfile.ZipFile(WHEEL_PATH, "w", zipfile.ZIP_DEFLATED) as z:
        for name, text, mode in MEMBERS:
            data = text.encode("utf-8")
            zi = zipfile.ZipInfo(name, date_time=(1980, 1, 1, 0, 0, 0))
            # S_IFREG MUST BE PRESENT. installer derives is_executable as
            # `mode and S_ISREG(mode) and mode & 0o111` (sources.py, borrowed
            # from pip), so a member whose external_attr carries permission bits
            # WITHOUT the file-type bits is never executable however 0o755 it
            # looks. Real wheels carry them because ZipInfo.from_file copies
            # st_mode whole; a hand-written ZipInfo does not.
            zi.external_attr = ((0o100000 | mode) & 0xFFFF) << 16
            zi.compress_type = zipfile.ZIP_DEFLATED
            z.writestr(zi, data)
            rows.append((name, rec_hash(data), str(len(data))))
        rows.append(("%s/RECORD" % DISTINFO, "", ""))
        w = csv.writer(buf, lineterminator="\n")
        for r in rows:
            w.writerow(r)
        z.writestr("%s/RECORD" % DISTINFO, buf.getvalue())
    say("R1_wheel_built", "ok" if os.path.isfile(WHEEL_PATH) else "missing")
    say("R1_wheel_members", str(len(rows)))


# The archive must survive a round trip through this image's zlib before the
# installer is blamed for anything it reads out of it.
@guard("R1_zip_roundtrip")
def _roundtrip():
    with zipfile.ZipFile(WHEEL_PATH) as z:
        bad = z.testzip()
        got = z.read("%s/__init__.py" % DIST).decode("utf-8")
    say("R1_zip_roundtrip", "ok" if bad is None else "corrupt:%s" % bad)
    say("R1_zip_content", "ok" if got == INIT_PY else "differs")


# installer's own reader, over the same file. WheelFile parses the name, finds
# the .dist-info dir and validates the RECORD; a disagreement between this and
# the block above is the package's fault rather than the archive's.
@guard("R1_source_distribution")
def _source():
    with WheelFile.open(WHEEL_PATH) as src:
        say("R1_source_distribution", src.distribution)
        say("R1_source_version", src.version)
        say("R1_source_dist_info", src.dist_info_dir)
        say("R1_source_data_dir", src.data_dir)
        names = sorted(p for (p, _h, _s) in
                       (e for e, _st, _x in src.get_contents()))
        say("R1_source_contents", ",".join(names))


# ------------------------------------------ R2: install it, via the API ---
SCHEME1 = {}


@guard("R2_installed")
def _install():
    scheme = installer_main._get_scheme_dict(DIST, prefix=PREFIX1)
    SCHEME1.update(scheme)
    with WheelFile.open(WHEEL_PATH) as src:
        dest = SchemeDictionaryDestination(
            scheme_dict=scheme,
            interpreter=sys.executable,
            script_kind=get_launcher_kind(),
            bytecode_optimization_levels=[0, 1],
        )
        installer.install(src, dest, {"INSTALLER": b"brintos-selftest\n"})
    say("R2_installed", "ok")
    say("R2_purelib_under_prefix",
        "yes" if scheme["purelib"].startswith(PREFIX1 + os.sep) else
        "no:%s" % scheme["purelib"])


def purelib():
    return SCHEME1.get("purelib", "")


# THE ASSERTION THE TICKET NAMES: every path the generated RECORD promises is
# on disk, at the scheme the installer chose, with the size and hash it wrote.
# Validating the hash is what separates "a file of that name exists" from "the
# bytes arrived" -- a truncated write through virtiofs would pass the first.
@guard("R2_record_all_present")
def _record():
    rp = os.path.join(purelib(), DISTINFO, "RECORD")
    if not os.path.isfile(rp):
        say("R2_record_all_present", "no-RECORD")
        return
    missing = []
    bad = []
    n = 0
    with open(rp, "r", newline="", encoding="utf-8") as f:
        for row in csv.reader(f):
            if not row:
                continue
            n += 1
            path, h, size = (row + ["", ""])[:3]
            full = os.path.join(purelib(), path)
            if not os.path.exists(full):
                missing.append(path)
                continue
            if not h:
                continue
            entry = RecordEntry.from_elements(path, h, size)
            with open(full, "rb") as fh:
                if not entry.validate(fh.read()):
                    bad.append(path)
    say("R2_record_rows", str(n))
    say("R2_record_all_present", "ok" if not missing else "missing:" + ",".join(sorted(missing)))
    say("R2_record_all_valid", "ok" if not bad else "bad:" + ",".join(sorted(bad)))


@guard("R2_module_file")
def _files():
    say("R2_module_file",
        "ok" if os.path.isfile(os.path.join(purelib(), DIST, "__init__.py")) else "missing")
    say("R2_deep_module",
        "ok" if os.path.isfile(os.path.join(purelib(), DIST, "deep", "mod.py")) else "missing")
    say("R2_metadata",
        "ok" if os.path.isfile(os.path.join(purelib(), DISTINFO, "METADATA")) else "missing")
    ip = os.path.join(purelib(), DISTINFO, "INSTALLER")
    if os.path.isfile(ip):
        with open(ip, "r", encoding="utf-8") as f:
            say("R2_installer_marker", f.read().strip())
    else:
        say("R2_installer_marker", "missing")


# The `data` scheme, which is the OTHER half of the .data/ split and lands at
# the prefix root rather than under site-packages. Its mode is the control for
# the script's: same install, same chmod path, 0o644 in the archive.
@guard("R2_data_scheme")
def _data_scheme():
    dp = os.path.join(SCHEME1.get("data", ""), "share", "bstfix", "plain.txt")
    if not os.path.isfile(dp):
        say("R2_data_scheme", "missing:%s" % dp)
        return
    say("R2_data_scheme", "ok")
    say("R2_data_mode", oct(os.stat(dp).st_mode & 0o111))
    say("R2_data_not_executable", "no" if not os.access(dp, os.X_OK) else "yes")


# THE .data/scripts SCHEME. `#!python` must come back as this interpreter, the
# file must land in the scripts scheme rather than beside the modules, and the
# execute bit must be set -- three separate ways a script install goes wrong and
# each is silent until something tries to run it.
@guard("R2_script_scheme")
def _script():
    sp = os.path.join(SCHEME1.get("scripts", ""), "bst-fixture-script")
    if not os.path.isfile(sp):
        say("R2_script_scheme", "missing:%s" % sp)
        return
    say("R2_script_scheme", "ok")
    with open(sp, "rb") as f:
        first = f.readline().decode("utf-8", "replace").rstrip("\n")
    say("R2_script_shebang", "ok" if first == "#!" + sys.executable else "got:%s" % first)
    say("R2_script_executable", "yes" if os.access(sp, os.X_OK) else "no")
    # THE EXEC BITS ONLY, and that is not timidity. make_file_executable does
    # `chmod(0o777 & ~umask | 0o111)`, so the read/write bits are the runner's
    # umask talking and would make this case report the harness rather than the
    # port; the 0o111 is unconditional and is the whole of the assertion.
    say("R2_script_mode", oct(os.stat(sp).st_mode & 0o111))
    say("R2_script_not_in_purelib",
        "ok" if not os.path.exists(os.path.join(purelib(), "bst-fixture-script")) else "leaked")


# The console-script the entry_points.txt declares. installer generates it from
# the metadata rather than copying anything, so its presence proves the
# entry-point parser and the script writer, and its shebang proves they share
# the interpreter the destination was given.
@guard("R2_console_script")
def _console():
    cp = os.path.join(SCHEME1.get("scripts", ""), "bst-fixture-cli")
    if not os.path.isfile(cp):
        say("R2_console_script", "missing")
        return
    say("R2_console_script", "ok")
    with open(cp, "rb") as f:
        body = f.read().decode("utf-8", "replace")
    say("R2_console_shebang",
        "ok" if body.startswith("#!" + sys.executable + "\n") else "got:%s" % body.split("\n")[0])
    say("R2_console_calls_main", "yes" if "bstfix" in body and "main" in body else "no")


# BYTECODE AT THE LEVELS ASKED FOR. compileall is the reason this recipe
# RDEPENDS on python3-compile, and level 1 lands beside level 0 under a
# different tag -- so counting both is what says the level list was honoured
# rather than the default taken.
@guard("R2_pyc_level0")
def _pyc():
    pc = os.path.join(purelib(), DIST, "__pycache__")
    names = sorted(os.listdir(pc)) if os.path.isdir(pc) else []
    say("R2_pyc_level0",
        "yes" if any(n.endswith(".pyc") and ".opt-" not in n for n in names) else "no:%s" % ",".join(names))
    say("R2_pyc_level1",
        "yes" if any(".opt-1.pyc" in n for n in names) else "no:%s" % ",".join(names))


# IMPORTING WHAT WAS JUST INSTALLED, out of the scratch prefix and not the
# rootfs. This is the end of the chain the ticket asks about: the wheel's own
# module, read back through the interpreter that installed it.
@guard("R2_import_installed")
def _import_installed():
    sys.path.insert(0, purelib())
    try:
        import bstfix
        import bstfix.deep.mod as deep
    finally:
        pass
    say("R2_import_installed", bstfix.BSTFIX_MARK)
    say("R2_import_installed_from",
        "ok" if os.path.dirname(os.path.abspath(bstfix.__file__)).startswith(purelib())
        else "elsewhere:%s" % bstfix.__file__)
    say("R2_import_deep", str(deep.answer()))


# --------------------------------------- R3: the CLI, and --no-compile ---
# `python -m installer` is _main(), and this calls exactly that with an argv --
# in this process rather than through a second exec, which on this arm would
# cost another interpreter start for a code path that is one function call away.
@guard("R3_cli_install")
def _cli():
    installer_main._main([WHEEL_PATH, "--prefix", PREFIX2, "--no-compile-bytecode"])
    scheme = installer_main._get_scheme_dict(DIST, prefix=PREFIX2)
    pl = scheme["purelib"]
    say("R3_cli_install", "ok" if os.path.isfile(os.path.join(pl, DIST, "__init__.py")) else "missing")
    say("R3_cli_record",
        "ok" if os.path.isfile(os.path.join(pl, DISTINFO, "RECORD")) else "missing")
    pc = os.path.join(pl, DIST, "__pycache__")
    names = sorted(os.listdir(pc)) if os.path.isdir(pc) else []
    say("R3_cli_no_pyc", "ok" if not names else "found:" + ",".join(names))
    say("R3_cli_two_prefixes_differ", "ok" if pl != purelib() else "same")


# THE NEGATIVE THE PORT'S SHAPE DEPENDS ON. This image's python3 has an empty
# lib-dynload and no dynamic loader (BRINTOS-1476), so a version of this package
# that grew a compiled accelerator could not be imported at all. Asserting the
# absence over the INSTALLED tree is cheaper and sharper than reading the
# recipe: it is a property of what shipped.
@guard("R4_no_so_installed")
def _no_so():
    base = os.path.dirname(os.path.abspath(installer.__file__))
    found = []
    for dirpath, _dirs, files in os.walk(base):
        for f in files:
            if f.endswith(".so") or ".cpython-" in f and f.endswith(".so"):
                found.append(os.path.join(dirpath, f))
    say("R4_no_so_installed", "ok" if not found else "found:" + ",".join(sorted(found)))
    say("R4_pure_python_dir", base)


if all(not o.split(" ", 1)[1].startswith("EXC:") for o in OUT if " " in o):
    OUT.append("ALL-INSTALLER-OK")
print("\n".join(OUT))