test.py

# Copyright(c) The Maintainers of Nanvix.
# Licensed under the MIT License.

"""Test orchestration for Nanvix CPython.

Replaces test-common.mk, test-standalone.mk, test-multi-process.mk,
test-single-process.mk, test-hyperlight.mk, test-microvm.mk, and
test-run-host.py.

Provides staging, hello-world validation, and regrtest dispatch for
all deployment modes and platforms.
"""

from __future__ import annotations

import json
import os
import shutil
import subprocess
import sys
import tarfile
import time
import urllib.request
from pathlib import Path
from typing import Any

import sys as _sys

_sys.path.insert(0, str(Path(__file__).resolve().parent))
from _loader import load_sibling

config = load_sibling("config", __file__)
build_mod = load_sibling("build", __file__)
lxml_mod = load_sibling("lxml", __file__)
ramfs_mod = load_sibling("ramfs", __file__)


# ---------------------------------------------------------------------------
# Initrd creation helper (standalone mode)
# ---------------------------------------------------------------------------


def _create_initrd(
    bin_dir: Path,
    app_path: Path,
    app_args: list[str] | None = None,
    app_env: str | None = None,
    output: Path | None = None,
) -> Path:
    """Create an initrd image bundling *app_path* with system daemons.

    Mirrors :meth:`~nanvix_zutil.ZScript.make_initrd` for use in
    standalone module-level functions that lack a ZScript instance.

    Args:
        bin_dir: Directory containing the system daemon ELFs and mkimage.
        app_path: Absolute path to the application ELF binary.
        app_args: Optional CLI arguments for the app entry.
        app_env: Optional space-separated env vars (e.g.
            ``"PYTHONHOME=/ TMPDIR=/tmp"``).  Appended after a bare
            semicolon in the cmdline so the kernel's ``split_cmdline``
            can separate args from env.
        output: Destination path for the image.  Defaults to
            ``app_path.parent / "<stem>.img"``.

    Returns:
        Path to the generated image file.
    """
    app_stem = app_path.stem
    if output is None:
        output = app_path.parent / f"{app_stem}.img"

    mkimage = bin_dir / config.mkimage_binary()

    def _escape(arg: str) -> str:
        return arg.replace(";", "\\;")

    def _entry(elf: Path, argv0: str, extra: list[str] | None, env: str | None) -> str:
        parts = [_escape(argv0)] + [_escape(a) for a in (extra or [])]
        argv = " ".join(parts)
        # The entry format for mkimage is:
        #   <escaped_elf_path>;<cmdline>
        # Within cmdline, the kernel splits on the first unescaped ';':
        #   <escaped_args>;<env_vars>
        cmdline = argv
        if env:
            cmdline += f";{env}"
        return f"{_escape(str(elf))};{cmdline}"

    # Daemons are *guest* binaries — always .elf, even on Windows.
    cmd: list[str] = [
        str(mkimage),
        "-o",
        str(output),
        _entry(bin_dir / "procd.elf", "procd", None, None),
        _entry(bin_dir / "memd.elf", "memd", None, None),
        _entry(bin_dir / "vfsd.elf", "vfsd", None, None),
        _entry(app_path, app_stem, app_args, app_env),
    ]

    subprocess.run(cmd, check=True, timeout=60)
    return output


# ---------------------------------------------------------------------------
# Windows: download release artifacts as install cache
# ---------------------------------------------------------------------------


def _download_release_as_cache(
    repo_root: Path,
    platform: str,
    process_mode: str,
    memory_size: str,
) -> Path:
    """Download the latest cpython release tarball and extract it as _install_cache.

    This lets ``./z test`` work on Windows without a prior ``./z build``
    (which requires Docker). The release tarball contains the same
    sysroot tree that ``./z build`` would produce.
    """
    cache_dir = repo_root / ".nanvix" / "_install_cache"
    if cache_dir.exists():
        shutil.rmtree(cache_dir)
    cache_dir.mkdir(parents=True, exist_ok=True)

    # Resolve the latest release from nanvix/cpython.
    gh_token = os.environ.get("GH_TOKEN") or os.environ.get("GITHUB_TOKEN")
    api_url = "https://api.github.com/repos/nanvix/cpython/releases/latest"
    req = urllib.request.Request(api_url)
    req.add_header("Accept", "application/vnd.github+json")
    if gh_token:
        req.add_header("Authorization", f"Bearer {gh_token}")

    with urllib.request.urlopen(req, timeout=30) as resp:
        release = json.loads(resp.read())

    tag = release["tag_name"]
    print(f"  Resolved cpython release: {tag}")

    # Find a standalone tarball asset (.tar.gz preferred, .tar.bz2 fallback).
    asset_prefix = f"cpython-{platform}-{process_mode}-{memory_size}"
    asset_url = None
    asset_name = None
    for ext in (".tar.gz", ".tar.bz2"):
        for a in release.get("assets", []):
            name = a.get("name", "")
            if (
                name.startswith(asset_prefix)
                and name.endswith(ext)
                and "buildroot" not in name
            ):
                asset_url = a["browser_download_url"]
                asset_name = name
                break
        if asset_url:
            break

    if not asset_url:
        raise FileNotFoundError(
            f"No cpython release asset matching '{asset_prefix}*.tar.gz' or '*.tar.bz2' "
            f"in release {tag}. Available assets: "
            + ", ".join(a["name"] for a in release.get("assets", []))
        )

    # Download.
    dl_dir = repo_root / ".nanvix" / "cache"
    dl_dir.mkdir(parents=True, exist_ok=True)
    assert asset_name is not None
    tarball = dl_dir / asset_name
    if not tarball.is_file():
        print(f"  Downloading {asset_name}...")
        urllib.request.urlretrieve(asset_url, str(tarball))

    # Extract into _install_cache with path-traversal protection.
    print(f"  Extracting to {cache_dir}...")
    with tarfile.open(tarball, "r:*") as tf:
        base = cache_dir.resolve()
        for member in tf.getmembers():
            if member.issym() or member.islnk():
                raise tarfile.TarError(f"refusing to extract link entry: {member.name}")
            resolved = (base / member.name).resolve()
            if os.path.commonpath([str(base), str(resolved)]) != str(base):
                raise tarfile.TarError(
                    f"refusing to extract path outside destination: {member.name}"
                )
        tf.extractall(cache_dir)

    # The tarball contains {bin/, sysroot/, cpython-ramfs.img}.
    # Restructure if needed so that sysroot/ is at cache_dir/sysroot/.
    sysroot = cache_dir / "sysroot"
    if not sysroot.is_dir():
        python_lib_dir = Path(config.PYTHON_LIB_DIR)
        for candidate in cache_dir.rglob(str(python_lib_dir)):
            parent = candidate
            for _ in python_lib_dir.parts:
                parent = parent.parent
            if parent != cache_dir:
                sysroot.mkdir(exist_ok=True)
                for item in parent.iterdir():
                    shutil.move(str(item), str(sysroot / item.name))
                break

    # Copy the stripped python binary into sysroot/bin/ if present.
    bin_dir = sysroot / "bin"
    bin_dir.mkdir(exist_ok=True)
    python_elf = cache_dir / "bin" / "python.elf"
    if python_elf.is_file():
        shutil.copy2(python_elf, bin_dir / config.python_binary())
        print(f"  Installed python binary ({python_elf.stat().st_size // 1024}K)")

    # Copy the test suite from the source tree into the sysroot.
    # The release tarball is trimmed (no Lib/test/), but regrtest
    # needs it. The source checkout has the full Lib/test/.
    pylib_dir = sysroot / "lib" / config.PYTHON_LIB_DIR
    test_dst = pylib_dir / "test"
    test_src = repo_root / "Lib" / "test"
    if test_src.is_dir() and not test_dst.is_dir():
        shutil.copytree(test_src, test_dst)
        test_count = sum(1 for _ in test_dst.rglob("*.py"))
        print(f"  Copied test suite from source tree ({test_count} files)")

    print(f"  Install cache ready at {cache_dir}")
    return cache_dir


# ---------------------------------------------------------------------------
# Staging
# ---------------------------------------------------------------------------


def _manual_install(
    repo_root: Path,
    staging: Path,
    install_prefix: str,
) -> None:
    """Create a minimal install tree without invoking make.

    Used when the Makefile was configured inside Docker and cannot be
    used natively.  Copies the Python binary and standard library from
    the source/build tree into the staging directory.
    """
    # install_prefix is e.g. "/sysroot" — strip leading slash for relative path.
    prefix_rel = install_prefix.lstrip("/")
    sysroot_dir = staging / prefix_rel
    bin_dir = sysroot_dir / "bin"
    lib_dir = sysroot_dir / "lib" / config.PYTHON_LIB_DIR

    bin_dir.mkdir(parents=True, exist_ok=True)
    lib_dir.mkdir(parents=True, exist_ok=True)

    # Copy the built python binary.
    python_bin = repo_root / f"python{config.EXE}"
    if python_bin.is_file():
        shutil.copy2(python_bin, bin_dir / config.python_binary())

    # Copy the standard library from Lib/.
    lib_src = repo_root / "Lib"
    if lib_src.is_dir():
        shutil.copytree(lib_src, lib_dir, dirs_exist_ok=True)

    # Copy sysconfigdata from the build directory.
    scdata_name = f"{config.SYSCONFIGDATA_NAME}.py"
    pybuilddir_file = repo_root / "pybuilddir.txt"
    if pybuilddir_file.is_file():
        bdir = repo_root / pybuilddir_file.read_text().strip()
        scdata_src = bdir / scdata_name
        if scdata_src.is_file():
            shutil.copy2(scdata_src, lib_dir / scdata_name)

    # Copy libpython archive (needed by some install validation).
    libpython = repo_root / f"libpython{config.PYTHON_VERSION}.a"
    lib_parent = sysroot_dir / "lib"
    if libpython.is_file():
        shutil.copy2(libpython, lib_parent / libpython.name)

    print(f"  Manual install complete ({sysroot_dir})")


def stage(
    sysroot: str | Path,
    toolchain: str | Path,
    repo_root: Path,
    *,
    platform: str = config.DEFAULT_PLATFORM,
    process_mode: str = config.DEFAULT_PROCESS_MODE,
    memory_size: str = config.DEFAULT_MEMORY_SIZE,
    install_prefix: str = config.DEFAULT_INSTALL_PREFIX,
    release: bool = False,
    run_fn: Any = None,
    docker: bool = False,
) -> Path:
    """Build, install, and stage CPython for testing.

    Returns the test staging directory.
    """
    staging = repo_root / ".nanvix" / "_test_staging"

    print("Running CPython tests on Nanvix...")
    if staging.exists():
        shutil.rmtree(staging)

    if config.IS_WINDOWS:
        # On Windows, use the cached install tree produced by ``./z build``
        # so that no Docker invocation is needed during testing.
        install_cache = repo_root / ".nanvix" / "_install_cache"
        if install_cache.is_dir():
            shutil.copytree(install_cache, staging)
            print("  Using cached install from ./z build")
        else:
            # Fallback: download the release tarball and use it as the
            # install cache. This lets ``./z test`` work on Windows
            # without a prior ``./z build`` (which requires Docker).
            print("  Install cache not found — downloading release artifacts...")
            install_cache = _download_release_as_cache(
                repo_root,
                platform,
                process_mode,
                memory_size,
            )
            shutil.copytree(install_cache, staging)
            print("  Using downloaded release as install cache")
    else:
        # Linux: build and install directly.
        # Only skip the rebuild when the previously built binary exists,
        # the build tree is properly configured, and the host cannot use
        # BUILD_PYTHON (e.g. after a prior Docker build where that tool
        # is unavailable outside the container).
        python_binary = repo_root / f"python{config.EXE}"
        configured_marker = repo_root / ".nanvix-configured"
        pybuilddir = repo_root / "pybuilddir.txt"

        # Determine whether BUILD_PYTHON is usable on the host.
        build_python_path = Path(toolchain) / "bin" / "python3"
        build_python_available = (
            build_python_path.is_file()
            or shutil.which(str(build_python_path)) is not None
        )

        # Detect if ./configure was run inside Docker (paths like
        # /mnt/sysroot baked into Makefile).  A native rebuild would
        # fail because those paths don't exist on the host.
        docker_configured = False
        makefile = repo_root / "Makefile"
        if makefile.is_file() and not docker:
            try:
                header = makefile.read_text(encoding="utf-8", errors="replace")[:8192]
                docker_configured = config.DOCKER_SYSROOT_PATH in header
            except OSError:
                pass

        can_skip_rebuild = (
            python_binary.is_file()
            and configured_marker.is_file()
            and pybuilddir.is_file()
            and (not build_python_available or docker_configured)
        )

        if can_skip_rebuild:
            skip_reason = (
                "Docker-configured Makefile (native rebuild would fail)"
                if docker_configured and build_python_available
                else "BUILD_PYTHON is unavailable"
            )
            print(
                f"  Skipping rebuild ({python_binary.name} already exists"
                f" — {skip_reason})"
            )
            if docker_configured and build_python_available:
                # Cannot run make install natively when configure used
                # Docker paths and the native toolchain would trigger a
                # rebuild — do a manual install instead.
                _manual_install(repo_root, staging, install_prefix)
            else:
                # Install into staging, skipping the outer build prereq
                # and stubbing PYTHON_FOR_BUILD for the inner make.
                build_mod.install(
                    sysroot,
                    toolchain,
                    repo_root,
                    staging,
                    platform=platform,
                    process_mode=process_mode,
                    memory_size=memory_size,
                    install_prefix=install_prefix,
                    release=release,
                    run_fn=run_fn,
                    extra_make_flags=["-o", "build", "PYTHON_FOR_BUILD=:"],
                    docker=docker,
                )
        else:
            build_mod.build(
                sysroot,
                toolchain,
                repo_root,
                platform=platform,
                process_mode=process_mode,
                memory_size=memory_size,
                install_prefix=install_prefix,
                release=release,
                run_fn=run_fn,
                docker=docker,
            )

            build_mod.install(
                sysroot,
                toolchain,
                repo_root,
                staging,
                platform=platform,
                process_mode=process_mode,
                memory_size=memory_size,
                install_prefix=install_prefix,
                release=release,
                run_fn=run_fn,
                docker=docker,
            )

    sysroot_dir = staging / "sysroot"

    # Ensure _sysconfigdata module is present in the installed sysroot.
    # make install should copy it from build/<pybuilddir>/ but this can
    # silently fail when PYTHON_FOR_BUILD is not available or when the
    # install recipe is interrupted.  Fall back to copying from the build
    # directory directly.
    scdata_name = f"{config.SYSCONFIGDATA_NAME}.py"
    scdata_dst = sysroot_dir / "lib" / config.PYTHON_LIB_DIR / scdata_name
    if not scdata_dst.is_file():
        pybuilddir = repo_root / "pybuilddir.txt"
        if pybuilddir.is_file():
            bdir = repo_root / pybuilddir.read_text().strip()
            scdata_src = bdir / scdata_name
            if scdata_src.is_file():
                shutil.copy2(scdata_src, scdata_dst)
                print(f"  Copied {scdata_name} from build dir (make install missed it)")
            else:
                print(f"  WARNING: {scdata_name} not found in build dir {bdir}")
        else:
            print(f"  WARNING: pybuilddir.txt not found; cannot locate {scdata_name}")
    else:
        print(
            f"  Verified: {scdata_name} installed ({scdata_dst.stat().st_size} bytes)"
        )

    # Copy test script — a simple smoke test that validates the interpreter.
    # The lxml import test is only included for standalone mode because
    # xmlInitParser() hangs in multi-process/single-process modes where
    # filesystem I/O goes through nanvixd's virtualized host-FS layer.
    hello_script = sysroot_dir / "test_hello.py"
    standalone = process_mode == "standalone"
    # Phase 0 of the .a -> .so migration: `array` is now a shared
    # extension at lib/python3.12/lib-dynload/array.cpython-312.so
    # (built from `*shared* array arraymodule.c` in Setup.local).
    # Asserting it is NOT in `sys.builtin_module_names` proves the
    # dlopen path is exercised end-to-end; if the .so failed to load,
    # the import would raise.
    array_snippet = (
        "import array\n"
        "assert 'array' not in sys.builtin_module_names, "
        "'array still built-in!'\n"
        "_a = array.array('i', [1, 2, 3])\n"
        "assert _a.tolist() == [1, 2, 3], f'array contents wrong: {_a.tolist()}'\n"
        "print(f'CPYTHON_TEST_ARRAY_SO: array loaded via dlopen from "
        "{array.__file__}')\n"
    )
    # Phase 1A: Tier-1 data-primitive modules now built as .so. Import
    # each one and exercise a trivial operation to ensure dlopen +
    # PyInit_<name> succeed end-to-end.
    phase1a_snippet = (
        "_phase1a = [\n"
        "    ('_bisect', lambda m: m.bisect_left([1, 3, 5], 4) == 2),\n"
        "    ('_heapq', lambda m: (m.heappush([], 1) is None)),\n"
        "    ('_struct', lambda m: m.pack('i', 42) == b'\\x2a\\x00\\x00\\x00'),\n"
        "    ('_random', lambda m: hasattr(m, 'Random')),\n"
        "    ('_opcode', lambda m: hasattr(m, 'stack_effect')),\n"
        "    ('_queue', lambda m: hasattr(m, 'SimpleQueue')),\n"
        "    ('_csv', lambda m: hasattr(m, 'reader')),\n"
        "    ('binascii', lambda m: m.hexlify(b'\\xab') == b'ab'),\n"
        "    ('_json', lambda m: hasattr(m, 'encode_basestring_ascii')),\n"
        "    ('_pickle', lambda m: hasattr(m, 'Pickler')),\n"
        "    ('_zoneinfo', lambda m: hasattr(m, 'ZoneInfo')),\n"
        "]\n"
        "for _name, _check in _phase1a:\n"
        "    _mod = __import__(_name)\n"
        "    assert _name not in sys.builtin_module_names, "
        "f'{_name} still built-in!'\n"
        "    assert _check(_mod), f'{_name} sanity check failed'\n"
        "    print(f'CPYTHON_TEST_PHASE1A: {_name} loaded via dlopen from "
        "{_mod.__file__}')\n"
    )
    lxml_snippet = (
        "try:\n"
        "    import lxml.etree\n"
        "    doc = lxml.etree.fromstring(b'<root><child>lxml OK</child></root>')\n"
        "    assert doc.tag == 'root'\n"
        "    assert doc[0].text == 'lxml OK'\n"
        "    print('CPYTHON_TEST_LXML: lxml.etree import and parse OK')\n"
        "except ImportError as e:\n"
        "    print(f'CPYTHON_TEST_LXML_SKIP: {e}')\n"
        "except Exception as e:\n"
        "    print(f'CPYTHON_TEST_LXML_FAIL: {e}')\n"
        "    sys.exit(1)\n"
    )
    hello_script.write_text(
        "import sys\n"
        "print('CPYTHON_TEST_HELLO: Hello from Python', sys.version_info[:2])\n"
        "print('CPYTHON_TEST_PLATFORM:', sys.platform)\n"
        + array_snippet
        + phase1a_snippet
        + (lxml_snippet if standalone else ""),
    )

    # Copy the HTTP server smoke-test script from the repo root into the
    # sysroot so it ends up in the ramfs image built downstream by
    # stage_ramfs().  Standalone mode mounts the ramfs as /, so the
    # script must already be present at this point — copying it later
    # (e.g. from run_smoke_httpserver) is too late.
    httpserver_src = repo_root / "httpserver.py"
    if httpserver_src.is_file():
        shutil.copy2(httpserver_src, sysroot_dir / "httpserver.py")

    # Copy Nanvix runtime binaries.
    bin_dir = sysroot_dir / "bin"
    bin_dir.mkdir(parents=True, exist_ok=True)
    nanvix_home = Path(sysroot)
    for binary in [
        "nanvixd.elf",
        "kernel.elf",
        "linuxd.elf",
        "uservm.elf",
        "nanvixd.exe",
        "kernel.exe",
        # Host tools for initrd creation (standalone mode).
        config.mkramfs_binary(),
        config.mkimage_binary(),
        # Guest daemon binaries — always .elf, even on Windows.
        "procd.elf",
        "memd.elf",
        "vfsd.elf",
    ]:
        src = nanvix_home / "bin" / binary
        if src.is_file():
            shutil.copy2(src, bin_dir / binary)

    # Replace unstripped python binary with stripped python.elf.
    stripped = repo_root / f"python{config.EXE}"
    if stripped.is_file():
        target = bin_dir / config.python_binary()
        shutil.copy2(stripped, target)
        size = target.stat().st_size
        print(f"  Installed stripped python.elf into staging ({size // 1024}K)")

    # Copy guest-side test runner.
    regrtest_runner = repo_root / ".nanvix" / "run-regrtest.py"
    if regrtest_runner.is_file():
        shutil.copy2(regrtest_runner, sysroot_dir / "run-regrtest.py")

    # Invalidate stale ramfs image and cache from previous runs.
    stale_ramfs = repo_root / ".nanvix" / "cpython-rootfs.img"
    if stale_ramfs.is_file():
        stale_ramfs.unlink()
    stale_cache = repo_root / ".nanvix" / "_ramfs_cache"
    if stale_cache.is_dir():
        shutil.rmtree(stale_cache)

    return staging


# ---------------------------------------------------------------------------
# Ramfs staging (standalone mode)
# ---------------------------------------------------------------------------


def stage_ramfs(
    staging: Path,
    nanvix_home: Path,
    repo_root: Path,
    ramfs_img: Path | None = None,
) -> Path:
    """Build a ramfs image for standalone mode testing.

    Uses a cached ramfs directory under .nanvix/_ramfs_cache/.

    Returns the path to the ramfs image.
    """
    if ramfs_img is None:
        ramfs_img = repo_root / ".nanvix" / "cpython-rootfs.img"

    ramfs_cache = repo_root / ".nanvix" / "_ramfs_cache"

    if ramfs_img.is_file() and (ramfs_cache / "sysroot").is_dir():
        print(f"  Using cached ramfs: {ramfs_img}")
        return ramfs_img

    # Build fresh ramfs.
    if ramfs_cache.exists():
        shutil.rmtree(ramfs_cache)
    ramfs_cache.mkdir(parents=True)

    # Copy sysroot from test staging.
    sysroot_src = staging / "sysroot"
    sysroot_dst = ramfs_cache / "sysroot"
    shutil.copytree(sysroot_src, sysroot_dst)

    # Create /tmp for tempfile.gettempdir().
    (sysroot_dst / "tmp").mkdir(exist_ok=True)

    # Trim and build ramfs image (keep tests for test pipeline).
    ramfs_mod.trim_and_build(
        ramfs_cache,
        nanvix_home,
        ramfs_img,
        keep_tests=True,
    )

    return ramfs_img


# ---------------------------------------------------------------------------
# Hello-world test
# ---------------------------------------------------------------------------


def _run_nanvixd_script(
    staging: Path,
    script_name: str,
    *,
    process_mode: str = config.DEFAULT_PROCESS_MODE,
    platform: str = config.DEFAULT_PLATFORM,
    nanvixd_extra: list[str] | None = None,
    ramfs_img: Path | None = None,
    nanvix_home: Path | None = None,
    timeout: int = 120,
    label: str = "script",
) -> tuple[int, str, int]:
    """Run a Python script on nanvixd and return (returncode, output, elapsed_ms).

    This is the low-level execution primitive shared by the hello-world
    test and the benchmark.
    """
    sysroot = staging / "sysroot"
    resolved_extra: list[str] = (
        nanvixd_extra
        if nanvixd_extra is not None
        else config.PLATFORM_NANVIXD_ARGS.get(platform, [])
    )
    # On Windows, CreateProcess searches for the executable relative to the
    # *parent's* CWD, not the child's cwd. Use an absolute path to avoid this.
    nanvixd = str((sysroot / "bin" / config.nanvixd_binary()).resolve())
    python_bin = f"./bin/{config.python_binary()}"
    standalone = process_mode == "standalone"

    if standalone:
        if ramfs_img is None:
            raise ValueError("ramfs_img is required for standalone mode")

        # Copy host tools and daemon ELFs into the staging sysroot.
        # mkramfs is needed for ramfs generation; mkimage and the daemons
        # (procd, memd, vfsd) are needed for initrd creation.
        if nanvix_home:
            # Daemons are *guest* binaries — always .elf, even on
            # Windows.  Only host tools use the platform extension.
            _staging_bins = [
                config.mkramfs_binary(),
                config.mkimage_binary(),
                "procd.elf",
                "memd.elf",
                "vfsd.elf",
            ]
            for name in _staging_bins:
                src = nanvix_home / "bin" / name
                if src.is_file():
                    shutil.copy2(src, sysroot / "bin" / name)

    initrd_img: Path | None = None
    if standalone:
        # Standalone: bundle python binary with system daemons into an
        # initrd image.  Env vars are passed via app_env so the kernel's
        # split_cmdline sees them after the bare ';' separator.
        bin_dir = sysroot / "bin"
        app_path = sysroot / "bin" / config.python_binary()
        app_args = ["-B", f"./{script_name}"]
        app_env = (
            f"PYTHONHOME=/ PYTHONDONTWRITEBYTECODE=1"
            f" _PYTHON_SYSCONFIGDATA_NAME={config.SYSCONFIGDATA_NAME}"
        )
        initrd_img = _create_initrd(
            bin_dir, app_path, app_args=app_args, app_env=app_env
        )

        cmd = [
            nanvixd,
            "-bin-dir",
            str(bin_dir),
            "-ramfs",
            str(ramfs_img),
            *resolved_extra,
            "--",
            str(initrd_img),
        ]
    else:
        # Direct mode: guest accesses host filesystem, no ramfs.
        cmd = [
            nanvixd,
            *resolved_extra,
            "--",
            python_bin,
            f"./{script_name}",
        ]

    start = time.monotonic()
    try:
        result = subprocess.run(
            cmd,
            stdin=subprocess.DEVNULL,
            capture_output=True,
            text=True,
            timeout=timeout,
            cwd=sysroot,
        )
    except subprocess.TimeoutExpired:
        raise RuntimeError(f"{label} timed out after {timeout}s")
    finally:
        if initrd_img is not None and initrd_img.exists():
            initrd_img.unlink()

    elapsed_ms = int((time.monotonic() - start) * 1000)
    output = (result.stdout + "\n" + result.stderr).strip()
    return result.returncode, output, elapsed_ms


def run_hello(
    staging: Path,
    *,
    process_mode: str = config.DEFAULT_PROCESS_MODE,
    platform: str = config.DEFAULT_PLATFORM,
    nanvixd_extra: list[str] | None = None,
    ramfs_img: Path | None = None,
    nanvix_home: Path | None = None,
) -> None:
    """Run the hello-world test via nanvixd.

    Standalone mode uses ramfs + ``-bin-dir`` + the semicolon-delimited
    environment variable syntax.  Multi-process and single-process modes
    use direct host-filesystem access (no ramfs).
    """
    standalone = process_mode == "standalone"

    print(f"Test: Hello world ({process_mode})...")

    returncode, output, elapsed_ms = _run_nanvixd_script(
        staging,
        "test_hello.py",
        process_mode=process_mode,
        platform=platform,
        nanvixd_extra=nanvixd_extra,
        ramfs_img=ramfs_img,
        nanvix_home=nanvix_home,
        label="Hello test",
    )
    print(f"  Execution time: {elapsed_ms} ms")

    if returncode != 0:
        print(f"  FAIL: Hello test exited with status {returncode}")
        print(output)
        raise RuntimeError(f"Hello test exited with status {returncode}")

    # Validate output.
    found_hello = False
    found_lxml = False
    for line in output.splitlines():
        if line.startswith("CPYTHON_TEST_"):
            tag = line.split(":")[0].replace("CPYTHON_TEST_", "")
            print(f"  {tag}: {line.strip()}")
            if tag == "HELLO":
                found_hello = True
            elif tag in ("LXML", "LXML_SKIP"):
                found_lxml = True

    if not found_hello:
        print("  FAIL: Hello test did not produce expected output")
        print(output)
        raise RuntimeError("Hello test did not produce expected output")

    if standalone and not found_lxml:
        # lxml staging is best-effort — if the runtime package was not
        # available (e.g. release asset missing), the test is non-fatal.
        print("  WARNING: lxml import/parse test did not produce expected output")

    print("  PASS")


# ---------------------------------------------------------------------------
# HTTP server smoke test
# ---------------------------------------------------------------------------


def run_smoke_httpserver(
    staging: Path,
    repo_root: Path,
    *,
    process_mode: str = config.DEFAULT_PROCESS_MODE,
    platform: str = config.DEFAULT_PLATFORM,
    nanvixd_extra: list[str] | None = None,
    ramfs_img: Path | None = None,
    nanvix_home: Path | None = None,
    host: str = "127.0.0.1",
    port: int = 9999,
    boot_timeout: float = 60.0,
    request_timeout: float = 10.0,
    listening_marker: str = "HTTP server listening",
) -> None:
    """Launch ``httpserver.py`` on nanvixd and probe it from the host.

    Assumes ``httpserver.py`` has already been staged into the sysroot
    by :func:`stage` (and therefore into the ramfs image for standalone
    mode).  Starts nanvixd as a background process, waits for the
    server's "listening" log line on stdout, issues a single HTTP/1.0
    GET, and validates the response body.  The nanvixd process is
    always terminated before this function returns.

    The smoke test only runs in *standalone* mode.  In multi-process
    and single-process modes the standalone networking stack is not
    exposed to the host (see ``HOSTED_EXCLUDE`` in ``.nanvix/config.py``),
    so the test is skipped with a "SKIP" message.
    """
    import socket as _socket
    import tempfile

    standalone = process_mode == "standalone"
    if not standalone:
        print(
            f"Test: HTTP server smoke ({process_mode})... "
            "SKIP (networking only available in standalone mode)"
        )
        return

    sysroot = staging / "sysroot"
    script_name = "httpserver.py"
    if not (sysroot / script_name).is_file():
        raise RuntimeError(
            f"{script_name} not found in staging sysroot ({sysroot}); "
            "stage() did not copy it"
        )

    resolved_extra: list[str] = (
        nanvixd_extra
        if nanvixd_extra is not None
        else config.PLATFORM_NANVIXD_ARGS.get(platform, [])
    )
    nanvixd = str((sysroot / "bin" / config.nanvixd_binary()).resolve())

    if ramfs_img is None:
        raise ValueError("ramfs_img is required for standalone mode")
    if nanvix_home is not None:
        for name in (
            config.mkramfs_binary(),
            config.mkimage_binary(),
            "procd.elf",
            "memd.elf",
            "vfsd.elf",
        ):
            hp = nanvix_home / "bin" / name
            if hp.is_file():
                shutil.copy2(hp, sysroot / "bin" / name)

    bin_dir = sysroot / "bin"
    app_path = sysroot / "bin" / config.python_binary()
    app_args = ["-B", f"./{script_name}"]
    app_env = (
        f"PYTHONHOME=/ PYTHONDONTWRITEBYTECODE=1"
        f" _PYTHON_SYSCONFIGDATA_NAME={config.SYSCONFIGDATA_NAME}"
    )
    initrd_img = _create_initrd(bin_dir, app_path, app_args=app_args, app_env=app_env)
    cmd = [
        nanvixd,
        "-bin-dir",
        str(bin_dir),
        "-ramfs",
        str(ramfs_img),
        *resolved_extra,
        "--",
        str(initrd_img),
    ]

    print(f"Test: HTTP server smoke ({process_mode}) on {host}:{port}...")

    # Capture stdout/stderr to a file so we can both poll for the
    # "listening" marker without risking PIPE deadlock and include the
    # output in error messages.
    log_fd, log_path_str = tempfile.mkstemp(prefix="nanvixd-smoke-", suffix=".log")
    os.close(log_fd)
    log_path = Path(log_path_str)
    log_fh = open(log_path, "wb")
    proc = subprocess.Popen(
        cmd,
        stdin=subprocess.DEVNULL,
        stdout=log_fh,
        stderr=subprocess.STDOUT,
        cwd=sysroot,
    )

    def _read_log() -> str:
        try:
            return log_path.read_text(encoding="utf-8", errors="replace")
        except OSError:
            return ""

    try:
        # Wait for the server to log that it is listening.  Only then
        # is it safe to attempt a TCP connection (otherwise we might
        # race with the kernel's own host stack or unrelated services
        # on the same port).
        deadline = time.monotonic() + boot_timeout
        ready = False
        while time.monotonic() < deadline:
            if proc.poll() is not None:
                raise RuntimeError(
                    f"nanvixd exited prematurely (rc={proc.returncode}) "
                    f"before server became ready:\n{_read_log()}"
                )
            if listening_marker in _read_log():
                ready = True
                break
            time.sleep(0.5)
        if not ready:
            raise RuntimeError(
                f"HTTP server did not log '{listening_marker}' "
                f"within {boot_timeout:.0f}s:\n{_read_log()}"
            )

        # Issue a minimal HTTP/1.0 request.
        try:
            with _socket.create_connection((host, port), timeout=request_timeout) as s:
                s.sendall(b"GET / HTTP/1.0\r\nHost: nanvix\r\n\r\n")
                s.settimeout(request_timeout)
                chunks: list[bytes] = []
                while True:
                    try:
                        data = s.recv(4096)
                    except OSError:
                        break
                    if not data:
                        break
                    chunks.append(data)
        except OSError as e:
            raise RuntimeError(
                f"HTTP smoke test failed to connect to {host}:{port}: {e}\n"
                f"nanvixd output:\n{_read_log()}"
            )
        response = b"".join(chunks)

        if b"200 OK" not in response or b"Hello from Nanvix!" not in response:
            raise RuntimeError(
                "HTTP smoke test received unexpected response:\n"
                + response.decode("utf-8", errors="replace")
                + "\nnanvixd output:\n"
                + _read_log()
            )

        print("  PASS")
    finally:
        if proc.poll() is None:
            proc.terminate()
            try:
                proc.wait(timeout=10)
            except subprocess.TimeoutExpired:
                proc.kill()
                proc.wait(timeout=5)
        log_fh.close()
        try:
            log_path.unlink()
        except OSError:
            pass
        if initrd_img.exists():
            initrd_img.unlink()


# ---------------------------------------------------------------------------
# Regression tests
# ---------------------------------------------------------------------------


def run_regrtest(
    staging: Path,
    repo_root: Path,
    *,
    process_mode: str = config.DEFAULT_PROCESS_MODE,
    platform: str = config.DEFAULT_PLATFORM,
    test_list: list[str] | None = None,
    batch_size: int = config.DEFAULT_TEST_BATCH_SIZE,
    nanvixd_extra: list[str] | None = None,
    ramfs_img: Path | None = None,
    release: bool = False,
) -> None:
    """Run stdlib regression tests via run-tests.py."""
    if release:
        print("Test: regrtest skipped (NANVIX_RELEASE=yes)")
        return

    if test_list is None:
        test_list = list(config.NANVIX_TEST_LIST)

    resolved_nanvixd_extra: list[str] = (
        nanvixd_extra
        if nanvixd_extra is not None
        else config.PLATFORM_NANVIXD_ARGS.get(platform, [])
    )
    standalone = process_mode == "standalone"

    sysroot = staging / "sysroot"
    run_tests_script = repo_root / ".nanvix" / "run-tests.py"

    env = os.environ.copy()
    env["NANVIX_TEST_BATCH_SIZE"] = str(batch_size)
    env["NANVIX_PYTHON_BIN"] = f"./bin/{config.python_binary()}"

    if standalone:
        # Standalone: ramfs + initrd-based invocation.
        if ramfs_img is None:
            ramfs_img = repo_root / ".nanvix" / "cpython-rootfs.img"
        bin_dir = sysroot / "bin"
        extra_str = f"-bin-dir {bin_dir} -ramfs {ramfs_img}"
        if resolved_nanvixd_extra:
            extra_str += " " + " ".join(resolved_nanvixd_extra)
        env["NANVIXD_EXTRA_ARGS"] = extra_str
        env["NANVIX_STANDALONE"] = "1"
        env["NANVIX_BIN_DIR"] = str(bin_dir)
        exclude_set = set(config.STANDALONE_EXCLUDE)
        test_list = [m for m in test_list if m not in exclude_set]
    else:
        # Direct mode: host filesystem, no ramfs.
        if resolved_nanvixd_extra:
            env["NANVIXD_EXTRA_ARGS"] = " ".join(resolved_nanvixd_extra)
        else:
            env.pop("NANVIXD_EXTRA_ARGS", None)
        env.pop("NANVIX_STANDALONE", None)
        exclude_set = set(config.HOSTED_EXCLUDE)
        test_list = [m for m in test_list if m not in exclude_set]

    cmd = [sys.executable, str(run_tests_script)] + test_list

    print(f"Test: regrtest ({len(test_list)} modules, {process_mode})...")
    result = subprocess.run(cmd, cwd=sysroot, env=env)
    if result.returncode != 0:
        raise RuntimeError(f"regrtest failed with exit code {result.returncode}")


# ---------------------------------------------------------------------------
# Cleanup
# ---------------------------------------------------------------------------


def cleanup(repo_root: Path) -> None:
    """Clean up test artifacts."""
    staging = repo_root / ".nanvix" / "_test_staging"
    if staging.is_dir():
        shutil.rmtree(staging)
    for name in [
        "cpython_test.log",
        "cpython_regrtest.log",
        "cpython_regrtest_batch.log",
    ]:
        p = repo_root / ".nanvix" / name
        if p.is_file():
            p.unlink()


def deep_cleanup(repo_root: Path) -> None:
    """Deep clean: remove cached ramfs template."""
    cleanup(repo_root)
    ramfs_cache = repo_root / ".nanvix" / "_ramfs_cache"
    if ramfs_cache.is_dir():
        shutil.rmtree(ramfs_cache)
    ramfs_img = repo_root / ".nanvix" / "cpython-rootfs.img"
    if ramfs_img.is_file():
        ramfs_img.unlink()


# ---------------------------------------------------------------------------
# Aggregate test runner
# ---------------------------------------------------------------------------


def run_all(
    sysroot: str | Path,
    toolchain: str | Path,
    repo_root: Path,
    *,
    platform: str = config.DEFAULT_PLATFORM,
    process_mode: str = config.DEFAULT_PROCESS_MODE,
    memory_size: str = config.DEFAULT_MEMORY_SIZE,
    install_prefix: str = config.DEFAULT_INSTALL_PREFIX,
    release: bool = False,
    test_list: list[str] | None = None,
    batch_size: int = config.DEFAULT_TEST_BATCH_SIZE,
    nanvixd_extra: list[str] | None = None,
    run_fn: Any = None,
    docker: bool = False,
) -> None:
    """Run the complete test pipeline: stage → hello → regrtest → cleanup."""
    nanvix_home = Path(sysroot)
    standalone = process_mode == "standalone"

    # Stage.
    staging = stage(
        sysroot,
        toolchain,
        repo_root,
        platform=platform,
        process_mode=process_mode,
        memory_size=memory_size,
        install_prefix=install_prefix,
        release=release,
        run_fn=run_fn,
        docker=docker,
    )
    lxml_mod.stage_lxml_runtime(repo_root, staging / "sysroot")

    # Ramfs — only needed for standalone mode.  Multi-process and
    # single-process use host-filesystem access (no ramfs).
    ramfs_img = None
    if standalone:
        ramfs_img = stage_ramfs(staging, nanvix_home, repo_root)

    # Hello test.
    run_hello(
        staging,
        process_mode=process_mode,
        platform=platform,
        nanvixd_extra=nanvixd_extra,
        ramfs_img=ramfs_img,
        nanvix_home=nanvix_home,
    )

    # HTTP server smoke test.
    run_smoke_httpserver(
        staging,
        repo_root,
        process_mode=process_mode,
        platform=platform,
        nanvixd_extra=nanvixd_extra,
        ramfs_img=ramfs_img,
        nanvix_home=nanvix_home,
    )

    # Regression tests.
    run_regrtest(
        staging,
        repo_root,
        process_mode=process_mode,
        platform=platform,
        test_list=test_list,
        batch_size=batch_size,
        nanvixd_extra=nanvixd_extra,
        ramfs_img=ramfs_img,
        release=release,
    )

    # Cleanup.
    cleanup(repo_root)

    print("\t\t*** CPython tests PASSED ***")


# ---------------------------------------------------------------------------
# Benchmark (hello-world only, release ramfs)
# ---------------------------------------------------------------------------


def run_benchmark(
    sysroot: str | Path,
    toolchain: str | Path,
    repo_root: Path,
    *,
    platform: str = config.DEFAULT_PLATFORM,
    process_mode: str = config.DEFAULT_PROCESS_MODE,
    memory_size: str = config.DEFAULT_MEMORY_SIZE,
    install_prefix: str = config.DEFAULT_INSTALL_PREFIX,
    nanvixd_extra: list[str] | None = None,
    run_fn: Any = None,
    docker: bool = False,
) -> None:
    """Run a hello-world benchmark.

    In *standalone* mode the benchmark builds a ramfs with the same
    trimming applied during ``./z release`` (no test/ directory, no dev
    artifacts) so that the image size and boot time reflect a production
    deployment.  Other process modes run without a ramfs and therefore
    do not exercise the production image/boot path.

    No regression tests are executed.
    """
    nanvix_home = Path(sysroot)
    standalone = process_mode == "standalone"

    try:
        _run_benchmark_impl(
            sysroot,
            toolchain,
            repo_root,
            nanvix_home=nanvix_home,
            standalone=standalone,
            platform=platform,
            process_mode=process_mode,
            memory_size=memory_size,
            install_prefix=install_prefix,
            nanvixd_extra=nanvixd_extra,
            run_fn=run_fn,
            docker=docker,
        )
    finally:
        cleanup(repo_root)


def _run_benchmark_impl(
    sysroot: str | Path,
    toolchain: str | Path,
    repo_root: Path,
    *,
    nanvix_home: Path,
    standalone: bool,
    platform: str,
    process_mode: str,
    memory_size: str,
    install_prefix: str,
    nanvixd_extra: list[str] | None = None,
    run_fn: Any = None,
    docker: bool = False,
) -> None:
    """Inner implementation of :func:`run_benchmark`."""
    # Stage (reuses cached build).
    staging = stage(
        sysroot,
        toolchain,
        repo_root,
        platform=platform,
        process_mode=process_mode,
        memory_size=memory_size,
        install_prefix=install_prefix,
        release=False,
        run_fn=run_fn,
        docker=docker,
    )

    # Write a minimal benchmark script.
    bench_script = "bench_hello.py"
    (staging / "sysroot" / bench_script).write_text("print('hello world')\n")

    # Build ramfs with release trimming (keep_tests=False).
    ramfs_img = None
    if standalone:
        ramfs_img = repo_root / ".nanvix" / "cpython-benchmark.img"
        bench_cache = repo_root / ".nanvix" / "_benchmark_cache"

        # Always rebuild to reflect the current sysroot.
        if bench_cache.exists():
            shutil.rmtree(bench_cache)
        bench_cache.mkdir(parents=True)

        sysroot_src = staging / "sysroot"
        sysroot_dst = bench_cache / "sysroot"
        shutil.copytree(sysroot_src, sysroot_dst)
        (sysroot_dst / "tmp").mkdir(exist_ok=True)

        # Release-style trim: no test/ dir, no dev artifacts.
        ramfs_mod.trim_and_build(
            bench_cache,
            nanvix_home,
            ramfs_img,
            keep_tests=False,
        )

        # Scratch directory is no longer needed.
        shutil.rmtree(bench_cache, ignore_errors=True)

    # Run benchmark.
    print(f"Benchmark: Hello world ({process_mode})...")

    returncode, output, elapsed_ms = _run_nanvixd_script(
        staging,
        bench_script,
        process_mode=process_mode,
        platform=platform,
        nanvixd_extra=nanvixd_extra,
        ramfs_img=ramfs_img,
        nanvix_home=nanvix_home,
        label="Benchmark",
    )
    print(f"  Execution time: {elapsed_ms} ms")

    if returncode != 0:
        print(f"  FAIL: Benchmark exited with status {returncode}")
        print(output)
        raise RuntimeError(f"Benchmark exited with status {returncode}")

    if "hello world" not in output:
        print(f"  FAIL: expected 'hello world' in output")
        print(output)
        raise RuntimeError("Benchmark did not produce expected output")

    print("  PASS")