view release on metacpan or  search on metacpan

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

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"]

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

    parser = argparse.ArgumentParser(
        description="libhisto Multi-Architecture & Portability Test Runner"
    )
    parser.add_argument(
        "-t", "--target",
        default="musl",
        help="Target profile name, alias, 'all', or 'matrix' (default: musl)"
    )
    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: