view release on metacpan or  search on metacpan

bundled/tools/scripts/test_container.py  view on Meta::CPAN

    },
    "native-32bit": {
        "description": "Host Native 32-bit x86 (-m32 multilib execution directly on Linux host)",
        "image": None,
        "platform": None,
        "arch": "i686",
        "endian": "little",
        "bits": 32,
    },
}

ALIASES = {
    "alpine": "musl",
    "big-endian": "s390x",
    "bigendian": "s390x",
    "32bit": "i386",
    "x86": "i386",
    "aarch64": "arm64",
    "arm": "armv7",
    "riscv": "riscv64",
    "32bit-native": "native-32bit",
}

ALL_LOCAL_TARGETS = ["hermetic", "musl", "s390x", "i386", "native-32bit", "arm64", "armv7", "riscv64"]

# Flag whether container execution requires "sg docker -c" group elevation
USE_SG_DOCKER = False


def log(msg, color="1;34"):
    if sys.stdout.isatty():
        print(f"\033[{color}m==>\033[0m \033[1m{msg}\033[0m")
    else:
        print(f"==> {msg}")


def log_success(msg):
    log(msg, color="1;32")


def log_error(msg):
    log(msg, color="1;31")


def exec_container_cmd(cmd, **kwargs):
    """Execute container command, applying sg docker elevation if necessary."""
    if USE_SG_DOCKER and cmd[0] == "docker":
        wrapped = ["sg", "docker", "-c", " ".join(shlex.quote(a) for a in cmd)]
        return subprocess.run(wrapped, **kwargs)
    return subprocess.run(cmd, **kwargs)


def test_engine_connectivity(engine):
    """Test if container engine binary exists and can communicate with daemon/socket."""
    if not shutil.which(engine):
        return False, f"'{engine}' binary not found in PATH", False
    try:
        res = subprocess.run(
            [engine, "info"],
            stdout=subprocess.DEVNULL,
            stderr=subprocess.PIPE,
            text=True,
            timeout=10
        )
        if res.returncode == 0:
            return True, "", False
        if engine == "docker" and shutil.which("sg") and "permission denied" in res.stderr.lower():
            sg_res = subprocess.run(
                ["sg", "docker", "-c", f"{engine} info"],
                stdout=subprocess.DEVNULL,
                stderr=subprocess.PIPE,
                text=True,
                timeout=10
            )
            if sg_res.returncode == 0:
                return True, "", True
        return False, res.stderr.strip(), False
    except Exception as e:
        return False, str(e), False


def detect_container_engine(preferred="auto"):
    """Detect an available and working container engine (docker or podman)."""
    global USE_SG_DOCKER
    candidates = ["docker", "podman"] if preferred == "auto" else [preferred]
    errors = {}

    for engine in candidates:
        if shutil.which(engine):
            ok, err, use_sg = test_engine_connectivity(engine)
            if ok:
                USE_SG_DOCKER = use_sg
                return engine
            errors[engine] = err

    # If we reached here, no engine was fully working
    print("\n" + "=" * 70)
    log_error("No working container engine (Docker or Podman) accessible.")
    print("=" * 70)
    for engine, err in errors.items():
        print(f"  * {engine}: {err}")
    print("\nTroubleshooting & Fixes:")
    print("  1. If using Docker: Ensure your user has permission to access the Docker daemon:")
    print("     $ sudo usermod -aG docker $USER")
    print("     (Log out and log back in, or run with 'newgrp docker')")
    print("  2. Or use Podman (rootless by default without daemon permissions):")
    print("     $ sudo apt-get install -y podman   # Debian/Ubuntu")
    print("     $ sudo dnf install -y podman       # Fedora/RHEL")
    print("=" * 70 + "\n")
    return None


def is_foreign_arch(target_arch):
    """Check if target architecture differs from host and requires QEMU emulation."""
    host = platform.machine().lower()
    if host in ["x86_64", "amd64"]:
        return target_arch not in ["x86_64", "i386", "i686"]
    if host in ["aarch64", "arm64"]:
        return target_arch not in ["aarch64", "arm64", "armv7l"]
    return target_arch != host


def ensure_binfmt_registered(engine, targets_to_run):
    """Ensure QEMU binfmt handlers are registered only if running foreign architectures."""
    if platform.system() != "Linux":
        return

    foreign_targets = [
        t for t in targets_to_run
        if TARGETS[t].get("arch") and is_foreign_arch(TARGETS[t]["arch"])
    ]
    if not foreign_targets:
        return

    binfmt_path = Path("/proc/sys/fs/binfmt_misc")
    if binfmt_path.exists():
        entries = list(binfmt_path.iterdir())
        has_qemu = any("qemu-" in e.name for e in entries)
        if has_qemu:
            return

    log("Registering QEMU user-mode binfmt handlers via multiarch/qemu-user-static...")
    try:
        cmd = [engine, "run", "--rm", "--privileged", "multiarch/qemu-user-static", "--reset", "-p", "yes"]
        exec_container_cmd(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=True)
    except Exception as e:
        log_error(f"Warning: Failed to auto-register binfmt handlers ({e}). Foreign architectures may fail if not pre-configured.")


def run_native_32bit(repo_root, jobs, build_type="Release", full=False):
    """Run native 32-bit test on host using -m32 flags."""
    build_dir = repo_root / "build-x86_32"
    log(f"Running Host Native 32-bit Build & Test in {build_dir}...")

    cmake_args = [
        "cmake",
        f"-B{build_dir}",
        f"-S{repo_root}",
        f"-DCMAKE_BUILD_TYPE={build_type}",
        "-DCMAKE_C_FLAGS=-m32",
        "-DCMAKE_EXE_LINKER_FLAGS=-m32",
        "-DCMAKE_SHARED_LINKER_FLAGS=-m32",
        "-DCMAKE_MODULE_LINKER_FLAGS=-m32",
    ]

    t0 = time.time()
    try:
        subprocess.run(cmake_args, check=True)
    except subprocess.CalledProcessError as e:
        log_error("Host 32-bit multilib build configuration failed.")
        print("\nTroubleshooting 32-Bit Multilib:")
        print("  Install multilib development packages on your host:")
        print("    $ sudo apt-get install -y gcc-multilib libc6-dev-i386   # Debian/Ubuntu")
        print("    $ sudo dnf install -y glibc-devel.i686 libgcc.i686      # Fedora/RHEL")
        print("    $ sudo pacman -S lib32-glibc lib32-gcc-libs            # Arch Linux")
        print("  Alternatively, run 32-bit tests inside an isolated container:")
        print("    $ make test-32bit\n")
        raise e

    subprocess.run(["cmake", "--build", str(build_dir), f"-j{jobs}"], check=True)
    subprocess.run(["ctest", "--test-dir", str(build_dir), f"-j{jobs}", "--output-on-failure"], check=True)

    elapsed = time.time() - t0
    log_success(f"Host Native 32-bit Test PASSED in {elapsed:.2f}s")
    return True


def run_container_target(repo_root, target_name, config, engine, jobs, build_type="Release", full=False):
    """Run build and test inside a container target with enhanced crash reporting."""
    image = config["image"]
    build_dir_name = f"build-container-{target_name}"
    install_cmd = config["install_full_cmd"] if full else config["install_cmd"]

    effective_jobs = jobs
    if config.get("arch") and is_foreign_arch(config["arch"]):
        max_emulated_jobs = 2 if config.get("arch") in ["s390x", "armv7l"] else 4
        if jobs > max_emulated_jobs:
            effective_jobs = max_emulated_jobs
            log(f"Notice: Clamped parallel concurrency to -j{effective_jobs} for emulated target '{target_name}' (QEMU user-mode stability safeguard).")

    extra_cmake_flags = config.get("cmake_flags", "")
    cmake_config_cmd = f"cmake -B {build_dir_name} -S . -DCMAKE_BUILD_TYPE={build_type}"
    if extra_cmake_flags:
        cmake_config_cmd += f" {extra_cmake_flags}"

bundled/tools/scripts/test_container.py  view on Meta::CPAN

    )
    parser.add_argument(
        "--engine",
        choices=["auto", "docker", "podman"],
        default="auto",
        help="Container runtime engine (default: auto-detect)"
    )
    parser.add_argument(
        "--full",
        action="store_true",
        help="Run full test suite (Core C + Python & Perl bindings) inside the target"
    )
    parser.add_argument(
        "-j", "--jobs",
        type=int,
        default=os.cpu_count() or 4,
        help="Number of parallel compilation jobs"
    )
    parser.add_argument(
        "--build-type",
        default="Release",
        choices=["Release", "Debug"],
        help="CMake build type (default: Release)"
    )
    parser.add_argument(
        "--list",
        action="store_true",
        help="List available target architectures and profiles"
    )
    parser.add_argument(
        "--clean",
        action="store_true",
        help="Clean up container build directories"
    )

    args = parser.parse_args()

    repo_root = Path(__file__).resolve().parent.parent.parent

    if args.list:
        print("Available Portability & Multi-Architecture Targets:")
        print("=" * 70)
        for name, cfg in TARGETS.items():
            print(f"  * {name:<14} : {cfg['description']}")
        print("\nAliases:")
        for alias, target in ALIASES.items():
            print(f"  * {alias:<14} -> {target}")
        print("\nAggregates:")
        print(f"  * {'all / matrix':<14} : {', '.join(ALL_LOCAL_TARGETS)}")
        return 0

    if args.clean:
        log("Cleaning container build directories...")
        for p in repo_root.glob("build-container-*"):
            if p.is_dir():
                shutil.rmtree(p, ignore_errors=True)
                if p.is_dir():
                    engine = detect_container_engine()
                    if engine:
                        try:
                            exec_container_cmd([engine, "run", "--rm", "-v", f"{repo_root}:/workspace", "alpine:latest", "rm", "-rf", f"/workspace/{p.name}"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
                        except Exception:
                            pass
        if (repo_root / "build-x86_32").is_dir():
            shutil.rmtree(repo_root / "build-x86_32", ignore_errors=True)
        log_success("Clean complete.")
        return 0

    # Resolve target
    target_key = args.target.lower()
    target_key = ALIASES.get(target_key, target_key)

    if target_key in ["all", "matrix", "test-matrix-local"]:
        targets_to_run = ALL_LOCAL_TARGETS
    elif target_key in TARGETS:
        targets_to_run = [target_key]
    else:
        log_error(f"Unknown target '{args.target}'. Run with --list to see available targets.")
        return 1

    needs_container = any(TARGETS[t]["image"] is not None for t in targets_to_run)
    engine = None
    if needs_container:
        engine = detect_container_engine(args.engine)
        if not engine:
            return 1
        ensure_binfmt_registered(engine, targets_to_run)

    start_time = time.time()
    results = {}

    for t in targets_to_run:
        cfg = TARGETS[t]
        try:
            if cfg["image"] is None:
                ok = run_native_32bit(repo_root, args.jobs, args.build_type, args.full)
            else:
                ok = run_container_target(repo_root, t, cfg, engine, args.jobs, args.build_type, args.full)
            results[t] = "PASSED"
        except subprocess.CalledProcessError as e:
            log_error(f"Target '{t}' FAILED with exit code {e.returncode}")
            results[t] = f"FAILED (exit code {e.returncode})"
        except Exception as e:
            log_error(f"Target '{t}' FAILED: {e}")
            results[t] = f"FAILED ({e})"

    total_elapsed = time.time() - start_time

    print("\n" + "=" * 70)
    print(" PORTABILITY & MULTI-ARCHITECTURE TEST SUMMARY")
    print("=" * 70)
    all_passed = True
    for t, status in results.items():
        if "PASSED" in status:
            mark = "\033[1;32m[PASS]\033[0m" if sys.stdout.isatty() else "[PASS]"
        else:
            mark = "\033[1;31m[FAIL]\033[0m" if sys.stdout.isatty() else "[FAIL]"
            all_passed = False
        print(f"  {mark}  {t:<16} : {status}")
    print("=" * 70)
    print(f"Total time elapsed: {total_elapsed:.2f}s\n")