diff --git a/src/ci_orchestrator/commands/template.py b/src/ci_orchestrator/commands/template.py index e99ed5d..a62f90a 100644 --- a/src/ci_orchestrator/commands/template.py +++ b/src/ci_orchestrator/commands/template.py @@ -55,18 +55,25 @@ template prepare-win from __future__ import annotations +import base64 import contextlib import fnmatch +import hashlib import json +import re +import shutil import socket import subprocess import sys +import tempfile import threading import time +import urllib.request from datetime import UTC, datetime from pathlib import Path import click +import paramiko from ci_orchestrator.backends.protocol import VmHandle from ci_orchestrator.backends.workstation import WorkstationVmrunBackend @@ -759,3 +766,987 @@ Register-ScheduledTask -TaskName 'CI-StaticIp' ` ) click.echo("[prepare-win] Provisioning complete.") + + +# ═══════════════════════════════════════════════════════════════════════════ +# template deploy-linux helpers +# ═══════════════════════════════════════════════════════════════════════════ + + +def _deploy_step(msg: str, step: int, start_from: int) -> None: + if start_from > step: + click.echo(f"\n=== {msg} === [SKIP]") + else: + click.echo(f"\n=== {msg} ===") + + +def _find_vdisk_manager(explicit: str | None) -> str: + if explicit: + p = Path(explicit) + if not p.is_file(): + raise click.ClickException(f"vmware-vdiskmanager not found: {p}") + return str(p) + found = shutil.which("vmware-vdiskmanager") + if found: + return found + raise click.ClickException( + "vmware-vdiskmanager not in PATH. Pass --vdisk-manager or install VMware Workstation." + ) + + +def _download_vmdk(url: str, dest: Path) -> None: + click.echo(f" Downloading {url} ...") + tmp = dest.with_suffix(dest.suffix + ".part") + try: + urllib.request.urlretrieve(url, str(tmp)) + tmp.rename(dest) + except Exception: + if tmp.exists(): + tmp.unlink() + raise + + +def _sha256_file(path: Path) -> str: + h = hashlib.sha256() + with path.open("rb") as fh: + for chunk in iter(lambda: fh.read(65536), b""): + h.update(chunk) + return h.hexdigest().upper() + + +def _build_seed_iso(staging_dir: Path, iso_path: Path) -> None: + tool = shutil.which("genisoimage") or shutil.which("mkisofs") + if not tool: + raise click.ClickException("genisoimage/mkisofs not found. Install genisoimage.") + if iso_path.exists(): + iso_path.unlink() + subprocess.run( + [tool, "-o", str(iso_path), "-V", "cidata", "-J", "-quiet", str(staging_dir)], + check=True, + capture_output=True, + ) + + +def _poll_guest_ip_deploy(vmrun: str, vmx: str, timeout: float) -> str | None: + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + try: + result = subprocess.run( + [vmrun, "getGuestIPAddress", vmx], + capture_output=True, + text=True, + timeout=30, + check=False, + ) + ip = result.stdout.strip() + if re.match(r"^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$", ip): + return ip + except subprocess.TimeoutExpired: + pass + click.echo(" ... waiting for guest IP (open-vm-tools) ...") + time.sleep(15) + return None + + +def _wait_ssh_port_deploy( + host: str, port: int, timeout: float, interval: float +) -> bool: + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + try: + with socket.create_connection((host, port), timeout=5.0): + return True + except OSError: + click.echo(f" ... waiting for SSH port {port} on {host} ...") + time.sleep(interval) + return False + + +def _verify_cloud_init_done( + host: str, + cloud_user: str, + ssh_key: str, + timeout: float, + interval: float = 15.0, +) -> bool: + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + client = paramiko.SSHClient() + client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) + try: + client.connect( + host, + username=cloud_user, + key_filename=ssh_key, + timeout=5.0, + auth_timeout=5.0, + banner_timeout=10.0, + look_for_keys=False, + allow_agent=False, + ) + _, stdout_fh, _ = client.exec_command( + "test -f /var/lib/cloud/instance/boot-finished && echo OK", + timeout=10.0, + ) + if stdout_fh.read().decode().strip() == "OK": + return True + except Exception as exc: + click.echo( + f" ... not ready yet ({str(exc)[:80]}) -- retrying in {interval:.0f}s ..." + ) + finally: + client.close() + time.sleep(interval) + return False + + +def _patch_vmx_key(vmx_path: Path, key: str, value: str) -> None: + text = vmx_path.read_text(encoding="ascii", errors="replace") + pattern = re.compile(rf"^\s*{re.escape(key)}\s*=.*$", re.MULTILINE) + line = f'{key} = "{value}"' + text = pattern.sub(line, text) if pattern.search(text) else text.rstrip("\n") + f"\n{line}\n" + vmx_path.write_text(text, encoding="ascii") + + +# ═══════════════════════════════════════════════════════════════════════════ +# template deploy-linux command +# ═══════════════════════════════════════════════════════════════════════════ + + +@template.command("deploy-linux") +@click.option( + "--vmx", "vmx_path_str", default=None, metavar="PATH", + help="VMX file to create (default: /LinuxBuild2404/LinuxBuild2404.vmx).", +) +@click.option("--vm-name", default="LinuxBuild2404", show_default=True, help="VM display name.") +@click.option("--cloud-user", default="ci_build", show_default=True, help="cloud-init username.") +@click.option( + "--hostname", "guest_hostname", default="ci-linux-template", show_default=True, + help="Guest hostname.", +) +@click.option("--timezone", default="Europe/Rome", show_default=True, help="IANA timezone (reserved).") +@click.option("--disk-size-gb", default=40, show_default=True, help="Disk size after resize (GB).") +@click.option("--memory-mb", default=4096, show_default=True, help="Guest RAM (MiB).") +@click.option("--vcpu-count", default=4, show_default=True, help="vCPU count.") +@click.option("--cores-per-socket", default=2, show_default=True, help="Cores per socket.") +@click.option( + "--snapshot-name", default="PostInstall-Linux", show_default=True, + help="Snapshot name recorded in summary (taken by prepare-linux).", +) +@click.option( + "--ssh-key", "ssh_key_str", default=None, metavar="PATH", + help="SSH private key (default: from config ssh_key_path or /ci_linux).", +) +@click.option( + "--vmdk-url", + default="https://cloud-images.ubuntu.com/noble/current/noble-server-cloudimg-amd64.vmdk", + show_default=True, + help="Ubuntu cloud VMDK download URL.", +) +@click.option( + "--vmdk-cache", "vmdk_cache_str", default=None, metavar="PATH", + help="Local VMDK cache path (default: /iso/noble-server-cloudimg-amd64.vmdk).", +) +@click.option( + "--vmdk-sha256", "vmdk_sha256", default="", + help="Expected SHA256 of cached VMDK (empty = skip verification).", +) +@click.option( + "--seed-iso", "seed_iso_str", default=None, metavar="PATH", + help="Seed ISO path (default: /cloud-init-seed.iso).", +) +@click.option( + "--ssh-timeout", default=300, show_default=True, help="Seconds to wait for SSH port 22.", +) +@click.option( + "--ssh-poll-interval", default=10, show_default=True, + help="Seconds between SSH port poll attempts.", +) +@click.option("--show-gui", is_flag=True, help="Power on with VMware Workstation GUI visible.") +@click.option("--force", is_flag=True, help="Allow overwriting an existing VM directory.") +@click.option( + "--diag-password", "diag_password", default="", + help="Set plaintext password on ubuntu account for console login (diagnostic only).", +) +@click.option( + "--vdisk-manager", "vdisk_manager_str", default=None, metavar="PATH", + help="Path to vmware-vdiskmanager (auto-detected if omitted).", +) +@click.option( + "--start-from-step", "start_from_step", default=1, + type=click.IntRange(1, 10), show_default=True, + help=( + "Skip to this step (artifacts from skipped steps must exist). " + "Steps: 1=prereqs+vmdk 2=cloud-init 3=seed-ISO 4=vmdk-copy+resize " + "5=VMX 6=power-on 7=guest-IP 8=SSH-port 9=cloud-init-verify 10=detach-ISO" + ), +) +def template_deploy_linux( + vmx_path_str: str | None, + vm_name: str, + cloud_user: str, + guest_hostname: str, + timezone: str, + disk_size_gb: int, + memory_mb: int, + vcpu_count: int, + cores_per_socket: int, + snapshot_name: str, + ssh_key_str: str | None, + vmdk_url: str, + vmdk_cache_str: str | None, + vmdk_sha256: str, + seed_iso_str: str | None, + ssh_timeout: int, + ssh_poll_interval: int, + show_gui: bool, + force: bool, + diag_password: str, + vdisk_manager_str: str | None, + start_from_step: int, +) -> None: + """Deploy Ubuntu 24.04 LTS build VM template (Linux VMware Workstation host). + + Ports Deploy-LinuxBuild2404.ps1 for Linux hosts. Steps 1-10 mirror + the PowerShell original; ISO creation uses genisoimage instead of IMAPI2FS, + cloud-init verification uses paramiko instead of ssh.exe. + """ + _ = timezone # accepted for parity with PS script; set timezone via prepare-linux + + config = load_config() + + # ── Resolve paths ────────────────────────────────────────────────────────── + vmx_path = ( + Path(vmx_path_str).resolve() + if vmx_path_str + else config.paths.templates / "LinuxBuild2404" / "LinuxBuild2404.vmx" + ) + vm_dir = vmx_path.parent + + ssh_key_path: Path + if ssh_key_str: + ssh_key_path = Path(ssh_key_str) + elif config.ssh_key_path: + ssh_key_path = config.ssh_key_path + else: + ssh_key_path = config.paths.keys / "ci_linux" + + ssh_pub_key_path = ssh_key_path.with_suffix(ssh_key_path.suffix + ".pub") + + vmdk_cache = ( + Path(vmdk_cache_str) + if vmdk_cache_str + else config.paths.root / "iso" / "noble-server-cloudimg-amd64.vmdk" + ) + seed_iso_path = Path(seed_iso_str) if seed_iso_str else vm_dir / "cloud-init-seed.iso" + staging_dir = vm_dir / "_cloud_init_staging" + dest_vmdk = vm_dir / "LinuxBuild2404.vmdk" + + # ── Resolve tools ────────────────────────────────────────────────────────── + vmrun = config.vmrun_path or shutil.which("vmrun") or "vmrun" + vdisk_mgr = _find_vdisk_manager(vdisk_manager_str) + + # guestinfo base64 payloads — populated in step 2, consumed in step 5; + # empty when step 2 is skipped (guestinfo then omitted from VMX). + guest_user_data_b64 = "" + guest_meta_data_b64 = "" + guest_ip: str | None = None + + if start_from_step > 1: + click.echo( + f" [SKIP] Steps 1..{start_from_step - 1} -- starting from Step {start_from_step}" + ) + + # ────────────────────────────────────────────────────────────────────────── + # Step 1: validate prerequisites + download cloud VMDK + # ────────────────────────────────────────────────────────────────────────── + _deploy_step("Step 1: validate prerequisites + download cloud VMDK", 1, start_from_step) + if start_from_step <= 1: + checks: list[tuple[str, bool]] = [ + (f"vmrun found: {vmrun}", bool(shutil.which(vmrun) or Path(vmrun).is_file())), + (f"vmware-vdiskmanager found: {vdisk_mgr}", Path(vdisk_mgr).is_file()), + ( + "genisoimage/mkisofs in PATH", + bool(shutil.which("genisoimage") or shutil.which("mkisofs")), + ), + (f"SSH private key: {ssh_key_path}", ssh_key_path.is_file()), + (f"SSH public key: {ssh_pub_key_path}", ssh_pub_key_path.is_file()), + ] + for label, ok in checks: + if not ok: + raise click.ClickException(f"[PreFlight] {label}") + click.echo(f" [OK] {label}") + + if vm_dir.exists(): + if not force: + raise click.ClickException( + f"[PreFlight] VM directory already exists: {vm_dir}\n" + " Use --force to overwrite." + ) + click.echo(f" [WARN] --force set -- reusing existing VM directory: {vm_dir}") + else: + vm_dir.mkdir(parents=True, exist_ok=True) + click.echo(f" [OK] VM directory ready: {vm_dir}") + + if not vmdk_cache.is_file(): + vmdk_cache.parent.mkdir(parents=True, exist_ok=True) + try: + _download_vmdk(vmdk_url, vmdk_cache) + click.echo(f" [OK] Download complete: {vmdk_cache}") + except Exception as exc: + raise click.ClickException(f"[Deploy] VMDK download failed: {exc}") from exc + else: + click.echo(f" [OK] VMDK found in cache: {vmdk_cache}") + + if vmdk_sha256: + click.echo(" Verifying VMDK SHA256 ...") + actual = _sha256_file(vmdk_cache) + if actual != vmdk_sha256.upper(): + raise click.ClickException( + f"[Deploy] VMDK SHA256 mismatch!\n" + f" Expected: {vmdk_sha256.upper()}\n" + f" Actual: {actual}" + ) + click.echo(f" [OK] VMDK SHA256: {actual}") + + # ────────────────────────────────────────────────────────────────────────── + # Step 2: generate cloud-init user-data + meta-data + # ────────────────────────────────────────────────────────────────────────── + _deploy_step("Step 2: generate cloud-init user-data + meta-data", 2, start_from_step) + if start_from_step <= 2: + pub_key = ssh_pub_key_path.read_text(encoding="utf-8").strip() + + if staging_dir.exists(): + shutil.rmtree(staging_dir) + staging_dir.mkdir(parents=True) + + meta_data = f"instance-id: linuxbuild-001\nlocal-hostname: {guest_hostname}\n" + user_data = ( + "#cloud-config\n" + f"hostname: {guest_hostname}\n" + "users:\n" + f" - name: {cloud_user}\n" + " shell: /bin/bash\n" + " sudo: 'ALL=(ALL) NOPASSWD:ALL'\n" + " ssh_authorized_keys:\n" + f" - {pub_key}\n" + ) + if diag_password: + click.echo( + " [WARN] --diag-password set: console login enabled for ubuntu " + "(diagnostic mode)" + ) + user_data += ( + f"chpasswd:\n list: |\n ubuntu:{diag_password}\n expire: false\n" + ) + + (staging_dir / "meta-data").write_text(meta_data, encoding="utf-8") + (staging_dir / "user-data").write_text(user_data, encoding="utf-8") + click.echo(f" [OK] meta-data written: {staging_dir / 'meta-data'}") + click.echo(f" [OK] user-data written: {staging_dir / 'user-data'}") + + guest_meta_data_b64 = base64.b64encode(meta_data.encode()).decode() + guest_user_data_b64 = base64.b64encode(user_data.encode()).decode() + + # ────────────────────────────────────────────────────────────────────────── + # Step 3: build cloud-init seed ISO (nocloud, label: cidata) + # ────────────────────────────────────────────────────────────────────────── + _deploy_step( + "Step 3: build cloud-init seed ISO (nocloud, label: cidata)", 3, start_from_step + ) + if start_from_step <= 3: + try: + _build_seed_iso(staging_dir, seed_iso_path) + except subprocess.CalledProcessError as exc: + raise click.ClickException( + f"[Deploy Step 3] ISO creation failed:\n{exc.stderr}" + ) from exc + if not seed_iso_path.is_file() or seed_iso_path.stat().st_size <= 4096: + raise click.ClickException("[Deploy Step 3] Seed ISO missing or too small.") + click.echo( + f" [OK] Seed ISO: {seed_iso_path} " + f"({seed_iso_path.stat().st_size // 1024} KB)" + ) + + # ────────────────────────────────────────────────────────────────────────── + # Step 4: copy + resize cloud VMDK + # ────────────────────────────────────────────────────────────────────────── + _deploy_step(f"Step 4: copy + resize cloud VMDK to {disk_size_gb} GB", 4, start_from_step) + if start_from_step <= 4: + if dest_vmdk.exists() and force: + dest_vmdk.unlink() + click.echo( + f" Converting {vmdk_cache} -> {dest_vmdk} " + "(streamOptimized -> monolithicSparse) ..." + ) + r = subprocess.run( + [vdisk_mgr, "-r", str(vmdk_cache), "-t", "0", str(dest_vmdk)], + check=False, capture_output=True, text=True, + ) + if r.returncode != 0: + raise click.ClickException( + f"[Deploy Step 4] vmware-vdiskmanager convert failed " + f"(exit {r.returncode}):\n{r.stderr}" + ) + click.echo(f" [OK] VMDK converted: {dest_vmdk}") + + click.echo(f" Resizing VMDK to {disk_size_gb} GB ...") + r = subprocess.run( + [vdisk_mgr, "-x", f"{disk_size_gb}GB", str(dest_vmdk)], + check=False, capture_output=True, text=True, + ) + if r.returncode != 0: + raise click.ClickException( + f"[Deploy Step 4] vmware-vdiskmanager resize failed " + f"(exit {r.returncode}):\n{r.stderr}" + ) + click.echo(f" [OK] VMDK resized: {dest_vmdk}") + + # ────────────────────────────────────────────────────────────────────────── + # Step 5: generate VMX + # ────────────────────────────────────────────────────────────────────────── + _deploy_step("Step 5: generate VMX", 5, start_from_step) + if start_from_step <= 5: + vmx_lines = [ + '.encoding = "UTF-8"', + 'config.version = "8"', + 'virtualHW.version = "19"', + f'displayName = "{vm_name}"', + 'guestOS = "ubuntu-64"', + 'firmware = "efi"', + f'memsize = "{memory_mb}"', + f'numvcpus = "{vcpu_count}"', + f'cpuid.coresPerSocket = "{cores_per_socket}"', + 'scsi0.present = "TRUE"', + 'scsi0.virtualDev = "lsilogic"', + 'scsi0:0.present = "TRUE"', + 'scsi0:0.fileName = "LinuxBuild2404.vmdk"', + 'scsi0:0.deviceType = "scsi-hardDisk"', + 'ethernet0.present = "TRUE"', + 'ethernet0.connectionType = "nat"', + 'ethernet0.virtualDev = "vmxnet3"', + 'ethernet0.wakeOnPcktRcv = "FALSE"', + 'ethernet0.addressType = "generated"', + f'ide1:0.fileName = "{seed_iso_path}"', + 'ide1:0.present = "TRUE"', + 'ide1:0.deviceType = "cdrom-image"', + 'ide1:0.startConnected = "TRUE"', + 'pciBridge0.present = "TRUE"', + 'pciBridge4.present = "TRUE"', + 'pciBridge4.virtualDev = "pcieRootPort"', + 'pciBridge4.functions = "8"', + 'pciBridge5.present = "TRUE"', + 'pciBridge5.virtualDev = "pcieRootPort"', + 'pciBridge5.functions = "8"', + 'pciBridge6.present = "TRUE"', + 'pciBridge6.virtualDev = "pcieRootPort"', + 'pciBridge6.functions = "8"', + 'pciBridge7.present = "TRUE"', + 'pciBridge7.virtualDev = "pcieRootPort"', + 'pciBridge7.functions = "8"', + 'floppy0.present = "FALSE"', + 'usb.present = "FALSE"', + 'sound.present = "FALSE"', + 'tools.syncTime = "TRUE"', + 'uuid.action = "create"', + ] + if guest_user_data_b64 and guest_meta_data_b64: + vmx_lines += [ + f'guestinfo.userdata = "{guest_user_data_b64}"', + 'guestinfo.userdata.encoding = "base64"', + f'guestinfo.metadata = "{guest_meta_data_b64}"', + 'guestinfo.metadata.encoding = "base64"', + ] + click.echo( + " [INFO] guestinfo cloud-init properties embedded in VMX (backup datasource)." + ) + vmx_path.write_text("\n".join(vmx_lines) + "\n", encoding="ascii") + click.echo(f" [OK] VMX written: {vmx_path}") + + # ────────────────────────────────────────────────────────────────────────── + # Step 6: power on VM + # ────────────────────────────────────────────────────────────────────────── + _deploy_step("Step 6: power on VM", 6, start_from_step) + if start_from_step <= 6: + gui_mode = "gui" if show_gui else "nogui" + proc = subprocess.Popen([vmrun, "-T", "ws", "start", str(vmx_path), gui_mode]) + click.echo(f" [Deploy] VM power-on issued (pid={proc.pid}).") + time.sleep(10) + list_r = subprocess.run( + [vmrun, "-T", "ws", "list"], capture_output=True, text=True, check=False, + ) + if str(vmx_path) not in (list_r.stdout or ""): + raise click.ClickException( + "[Deploy Step 6] VM not found in vmrun list after power-on." + ) + click.echo(" [OK] VM appears in vmrun list.") + + # ────────────────────────────────────────────────────────────────────────── + # Step 7: detect guest IP via vmrun getGuestIPAddress (max 5 min) + # ────────────────────────────────────────────────────────────────────────── + _deploy_step( + "Step 7: detect guest IP via vmrun getGuestIPAddress (max 5 min)", 7, start_from_step + ) + if start_from_step <= 7: + guest_ip = _poll_guest_ip_deploy(vmrun, str(vmx_path), timeout=300.0) + if not guest_ip: + raise click.ClickException("[Deploy Step 7] Timed out waiting for guest IP.") + click.echo(f" [OK] Guest IP: {guest_ip}") + + # ────────────────────────────────────────────────────────────────────────── + # Step 8: poll SSH port 22 until ready + # ────────────────────────────────────────────────────────────────────────── + _deploy_step(f"Step 8: poll SSH port 22 (max {ssh_timeout}s)", 8, start_from_step) + if start_from_step <= 8: + if not guest_ip: + raise click.ClickException( + "[Deploy Step 8] Guest IP not available. Run from Step 7 or earlier." + ) + if not _wait_ssh_port_deploy( + guest_ip, 22, float(ssh_timeout), float(ssh_poll_interval) + ): + raise click.ClickException( + f"[Deploy Step 8] SSH port 22 not reachable on {guest_ip} " + f"within {ssh_timeout}s." + ) + click.echo(f" [OK] SSH port 22 open on {guest_ip}") + + # ────────────────────────────────────────────────────────────────────────── + # Step 9: verify cloud-init completed (boot-finished flag) + # ────────────────────────────────────────────────────────────────────────── + _deploy_step( + "Step 9: verify cloud-init completed (boot-finished flag)", 9, start_from_step + ) + if start_from_step <= 9: + if not guest_ip: + raise click.ClickException( + "[Deploy Step 9] Guest IP not available. Run from Step 7 or earlier." + ) + click.echo(" Polling cloud-init completion (max 600s) ...") + if not _verify_cloud_init_done( + guest_ip, cloud_user, str(ssh_key_path), timeout=600.0 + ): + raise click.ClickException( + "[Deploy Step 9] Timed out (600s) waiting for cloud-init to complete." + ) + click.echo(" [OK] cloud-init completed.") + + # ────────────────────────────────────────────────────────────────────────── + # Step 10: remove seed ISO from VMX (prevent re-run on clone) + # ────────────────────────────────────────────────────────────────────────── + _deploy_step( + "Step 10: remove seed ISO from VMX (prevent re-run on clone)", 10, start_from_step + ) + if start_from_step <= 10: + _patch_vmx_key(vmx_path, "ide1:0.present", "FALSE") + _patch_vmx_key(vmx_path, "ide1:0.startConnected", "FALSE") + click.echo(" [OK] ide1:0.present = FALSE") + click.echo(" [OK] ide1:0.startConnected = FALSE") + + # Cleanup staging dir + if staging_dir.exists(): + shutil.rmtree(staging_dir, ignore_errors=True) + + # Summary + click.echo("\n=== Done ===\n") + click.echo("[Deploy] Summary") + click.echo(f" VMX: {vmx_path}") + click.echo(f" Guest IP: {guest_ip or '(not obtained)'}") + click.echo(f" SSH key: {ssh_key_path}") + click.echo(f" Snapshot name: {snapshot_name} (run template prepare-linux next)") + click.echo("") + if guest_ip: + click.echo(f" SSH: ssh -i {ssh_key_path} {cloud_user}@{guest_ip}") + click.echo("") + click.echo(" VM is powered on. cloud-init provisioning complete.") + + +# ═══════════════════════════════════════════════════════════════════════════ +# template prepare-linux helpers +# ═══════════════════════════════════════════════════════════════════════════ + +_PREPARE_LINUX_VALIDATION = ( + '( id ci_build | grep -q sudo || sudo -l -U ci_build 2>/dev/null | grep -q NOPASSWD )' + ' && echo "UserSudo:OK" || echo "UserSudo:FAIL"\n' + 'sudo -n true 2>/dev/null && echo "SudoNoPass:OK" || echo "SudoNoPass:FAIL"\n' + 'for t in gcc g++ clang cmake python3 git 7z; do' + ' command -v "$t" >/dev/null && echo "Tool:$t:OK" || echo "Tool:$t:FAIL"; done\n' + 'test -d /opt/ci/build && echo "Dir:build:OK" || echo "Dir:build:FAIL"\n' + 'test -d /opt/ci/output && echo "Dir:output:OK" || echo "Dir:output:FAIL"\n' + 'test -d /opt/ci/scripts && echo "Dir:scripts:OK" || echo "Dir:scripts:FAIL"\n' + 'swapon --show | grep -q . && echo "Swap:ACTIVE:FAIL" || echo "Swap:off:OK"\n' + 'systemctl is-active ssh >/dev/null && echo "SSHD:active:OK" || echo "SSHD:inactive:FAIL"\n' + 'grep -q "^PasswordAuthentication no" /etc/ssh/sshd_config' + ' && echo "SSHPwdAuth:disabled:OK" || echo "SSHPwdAuth:enabled:FAIL"\n' + 'systemctl is-enabled ci-report-ip.service >/dev/null 2>&1' + ' && echo "CIReportIP:enabled:OK" || echo "CIReportIP:enabled:FAIL"\n' + 'test -x /usr/local/bin/ci-report-ip.sh' + ' && echo "CIReportIPScript:exists:OK" || echo "CIReportIPScript:exists:FAIL"\n' +) + + +def _pl_connect( + host: str, cloud_user: str, ssh_key: str, timeout: float = 10.0 +) -> paramiko.SSHClient: + client = paramiko.SSHClient() + client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) + client.connect( + host, + username=cloud_user, + key_filename=ssh_key, + timeout=timeout, + auth_timeout=timeout, + banner_timeout=timeout * 2, + look_for_keys=False, + allow_agent=False, + ) + return client + + +def _pl_run( + client: paramiko.SSHClient, cmd: str, timeout: float = 60.0 +) -> tuple[int, str, str]: + _, stdout_fh, stderr_fh = client.exec_command(cmd, timeout=timeout) + out = stdout_fh.read().decode(errors="replace") + err = stderr_fh.read().decode(errors="replace") + rc = stdout_fh.channel.recv_exit_status() + return rc, out, err + + +def _pl_stream(client: paramiko.SSHClient, cmd: str) -> int: + """Run cmd and stream stdout/stderr to console in real time. Returns exit code.""" + transport = client.get_transport() + assert transport is not None, "SSH transport closed" + channel = transport.open_session() + channel.set_combine_stderr(False) + channel.exec_command(cmd) + while True: + if channel.recv_ready(): + data = channel.recv(65536) + if data: + click.echo(data.decode(errors="replace"), nl=False) + if channel.recv_stderr_ready(): + data = channel.recv_stderr(65536) + if data: + click.echo(data.decode(errors="replace"), nl=False, err=True) + if channel.exit_status_ready(): + while channel.recv_ready(): + data = channel.recv(65536) + if data: + click.echo(data.decode(errors="replace"), nl=False) + while channel.recv_stderr_ready(): + data = channel.recv_stderr(65536) + if data: + click.echo(data.decode(errors="replace"), nl=False, err=True) + break + time.sleep(0.1) + return channel.recv_exit_status() + + +def _pl_upload(client: paramiko.SSHClient, local: Path, remote: str) -> None: + sftp = client.open_sftp() + try: + sftp.put(str(local), remote) + finally: + sftp.close() + + +# ═══════════════════════════════════════════════════════════════════════════ +# template prepare-linux command +# ═══════════════════════════════════════════════════════════════════════════ + + +@template.command("prepare-linux") +@click.option( + "--vmx", "vmx_path_str", default=None, metavar="PATH", + help=( + "Template VM .vmx file. Enables auto power-on, IP detection, " + "and automatic shutdown + snapshot at the end." + ), +) +@click.option( + "--ip", "guest_ip", default="", + help="Guest IP address (auto-detected via vmrun if --vmx is supplied).", +) +@click.option("--cloud-user", default="ci_build", show_default=True, help="SSH username in guest.") +@click.option( + "--ssh-key", "ssh_key_str", default=None, metavar="PATH", + help="SSH private key (default: from config ssh_key_path or /ci_linux).", +) +@click.option( + "--snapshot", default="BaseClean-Linux", show_default=True, + help="Snapshot name (must match the name used by vm new).", +) +@click.option( + "--toolchain-script", "toolchain_script_str", default=None, metavar="PATH", + help=( + "Path to Install-CIToolchain-Linux2404.sh " + "(auto-detected from /template/ if omitted)." + ), +) +@click.option("--skip-update", is_flag=True, help="Pass --skip-update to toolchain script.") +@click.option("--install-dotnet", is_flag=True, help="Pass --dotnet to toolchain script.") +@click.option("--no-mingw", is_flag=True, help="Pass --no-mingw to toolchain script.") +@click.option( + "--ssh-timeout", default=180, show_default=True, + help="Seconds to wait for SSH port 22 to become reachable.", +) +@click.option( + "--take-snapshot", is_flag=True, + help="Shut down and take snapshot after provisioning (always implied when --vmx is set).", +) +def template_prepare_linux( + vmx_path_str: str | None, + guest_ip: str, + cloud_user: str, + ssh_key_str: str | None, + snapshot: str, + toolchain_script_str: str | None, + skip_update: bool, + install_dotnet: bool, + no_mingw: bool, + ssh_timeout: int, + take_snapshot: bool, +) -> None: + """Provision Linux template VM via SSH (ports Prepare-LinuxBuild2404.ps1). + + Requires either --vmx (auto power-on + IP detection + snapshot) or + --ip (explicit IP, no auto VM lifecycle), or both. + """ + config = load_config() + + # ── Resolve SSH key ──────────────────────────────────────────────────────── + if ssh_key_str: + ssh_key_path = Path(ssh_key_str) + elif config.ssh_key_path: + ssh_key_path = config.ssh_key_path + else: + ssh_key_path = config.paths.keys / "ci_linux" + + # ── Resolve VMX ─────────────────────────────────────────────────────────── + vmx_path = Path(vmx_path_str).resolve() if vmx_path_str else None + + # ── Resolve toolchain script ────────────────────────────────────────────── + if toolchain_script_str is None: + candidate = Path.cwd() / "template" / "Install-CIToolchain-Linux2404.sh" + if not candidate.is_file(): + raise click.ClickException( + f"Toolchain script not found at {candidate}. " + "Run from the repo root or pass --toolchain-script." + ) + toolchain_path = candidate + else: + toolchain_path = Path(toolchain_script_str) + if not toolchain_path.is_file(): + raise click.ClickException(f"Toolchain script not found: {toolchain_path}") + + if vmx_path is None and not guest_ip: + raise click.ClickException( + "Provide --vmx (auto power-on + IP detection) or --ip (explicit IP), or both." + ) + + do_snapshot = take_snapshot or vmx_path is not None + + # ── Step 1: Pre-flight ──────────────────────────────────────────────────── + click.echo("[prepare-linux] Step 1 — Pre-flight validation") + if not ssh_key_path.is_file(): + raise click.ClickException(f"SSH private key not found: {ssh_key_path}") + click.echo(f" [OK] SSH key: {ssh_key_path}") + if vmx_path is not None and not vmx_path.is_file(): + raise click.ClickException(f"VMX not found: {vmx_path}") + if vmx_path is not None: + click.echo(f" [OK] VMX: {vmx_path}") + click.echo(f" [OK] Toolchain script: {toolchain_path}") + + # ── Step 2: Auto power-on + IP detection ────────────────────────────────── + vmrun = config.vmrun_path or shutil.which("vmrun") or "vmrun" + backend: WorkstationVmrunBackend | None = None + handle: VmHandle | None = None + + if vmx_path is not None: + backend = WorkstationVmrunBackend(vmrun_path=config.vmrun_path) + handle = VmHandle(identifier=str(vmx_path), name=vmx_path.stem) + + if not backend.is_running(handle): + click.echo(f"[prepare-linux] Starting VM: {vmx_path.name} ...") + backend.start(handle, headless=True) + else: + click.echo(f"[prepare-linux] VM already running: {vmx_path.name}") + + if not guest_ip: + click.echo("[prepare-linux] Detecting guest IP (vmrun getGuestIPAddress) ...") + detected = _poll_guest_ip_deploy(vmrun, str(vmx_path), timeout=300.0) + if not detected: + raise click.ClickException( + "Could not get guest IP. Ensure open-vm-tools is running in the guest." + ) + guest_ip = detected + click.echo(f"[prepare-linux] Guest IP: {guest_ip}") + + # ── Step 3: Wait for SSH port 22 ────────────────────────────────────────── + click.echo( + f"[prepare-linux] Step 3 — Waiting for SSH port 22 on {guest_ip} " + f"(timeout {ssh_timeout}s) ..." + ) + if not _wait_ssh_port_deploy(guest_ip, 22, float(ssh_timeout), 10.0): + raise click.ClickException( + f"SSH port 22 not reachable on {guest_ip} within {ssh_timeout}s." + ) + click.echo(f" [OK] TCP/22 reachable on {guest_ip}") + + # ── Step 4: SSH connectivity test ───────────────────────────────────────── + click.echo("[prepare-linux] Step 4 — Testing SSH connectivity ...") + try: + client = _pl_connect(guest_ip, cloud_user, str(ssh_key_path)) + except Exception as exc: + raise click.ClickException( + f"SSH connection to {cloud_user}@{guest_ip} failed: {exc}" + ) from exc + rc, out, _ = _pl_run(client, "echo connected") + if rc != 0 or "connected" not in out: + client.close() + raise click.ClickException("SSH connectivity test failed.") + click.echo(" [OK] SSH connectivity confirmed.") + + try: + # ── Step 5: Upload toolchain script ─────────────────────────────────── + click.echo( + f"[prepare-linux] Step 5 — Uploading {toolchain_path.name} to guest /tmp/ ..." + ) + raw = toolchain_path.read_bytes() + if raw[:3] == b"\xef\xbb\xbf": + click.echo(" [WARN] BOM detected — stripping") + raw = raw[3:] + normalized = raw.replace(b"\r\n", b"\n").replace(b"\r", b"\n") + remote_script = "/tmp/Install-CIToolchain-Linux2404.sh" + + with tempfile.NamedTemporaryFile(suffix=".sh", delete=False) as tf: + tf.write(normalized) + tmp_local = Path(tf.name) + try: + _pl_upload(client, tmp_local, remote_script) + finally: + tmp_local.unlink(missing_ok=True) + + rc, size_out, _ = _pl_run(client, f"stat -c %s {remote_script}") + remote_size = int(size_out.strip()) if size_out.strip().isdigit() else -1 + local_size = len(normalized) + if abs(remote_size - local_size) > 2: + raise click.ClickException( + f"Upload size mismatch: local={local_size}, remote={remote_size}." + ) + click.echo(f" [OK] {toolchain_path.name} uploaded ({local_size} bytes)") + + # ── Step 6: Execute toolchain script ────────────────────────────────── + flags: list[str] = [] + if skip_update: + flags.append("--skip-update") + if install_dotnet: + flags.append("--dotnet") + if no_mingw: + flags.append("--no-mingw") + flags_str = " ".join(flags) + remote_cmd = f"chmod +x {remote_script} && sudo {remote_script} {flags_str}".strip() + + click.echo("[prepare-linux] Step 6 — Running toolchain script (may take several minutes)") + click.echo(f" Command: {remote_cmd}") + click.echo("-" * 60) + exit_code = _pl_stream(client, remote_cmd) + click.echo("-" * 60) + if exit_code != 0: + raise click.ClickException( + f"Toolchain script exited {exit_code} — see output above." + ) + click.echo("[prepare-linux] Toolchain script completed (exit 0).") + + # ── Step 7: Post-setup validation ───────────────────────────────────── + click.echo("[prepare-linux] Step 7 — Running post-setup validation ...") + _, val_out, _ = _pl_run(client, _PREPARE_LINUX_VALIDATION, timeout=120.0) + click.echo("[prepare-linux] Validation results:") + failures: list[str] = [] + for line in val_out.splitlines(): + line = line.strip() + if not line: + continue + click.echo(f" {line}") + if ":FAIL" in line: + failures.append(line) + if failures: + raise click.ClickException( + "Post-setup validation failed:\n" + + "\n".join(f" {f}" for f in failures) + ) + click.echo("[prepare-linux] All post-setup checks passed.") + + # ── Step 7b: Clear machine-id (DHCP uniqueness for clones) ─────────── + click.echo("[prepare-linux] Step 7b — Clearing machine-id ...") + rc2, _, err2 = _pl_run( + client, + "sudo truncate -s 0 /etc/machine-id && sudo rm -f /var/lib/dbus/machine-id", + ) + if rc2 != 0: + raise click.ClickException(f"machine-id clear failed (exit {rc2}):\n{err2}") + _, size_out2, _ = _pl_run(client, "stat -c %s /etc/machine-id") + if size_out2.strip() != "0": + raise click.ClickException( + f"/etc/machine-id not empty after truncate (size={size_out2.strip()!r})." + ) + click.echo("[prepare-linux] machine-id cleared — clones get unique DHCP ID.") + + # ── Step 8: Graceful shutdown ───────────────────────────────────────── + if do_snapshot: + click.echo("[prepare-linux] Step 8 — Sending graceful shutdown ...") + with contextlib.suppress(Exception): + _pl_run(client, "sudo shutdown -h now", timeout=10.0) + else: + click.echo( + "[prepare-linux] --take-snapshot not set — VM left running. " + "Shut down and snapshot manually when ready." + ) + + finally: + client.close() + + # ── Post-SSH: wait for power-off + snapshot ─────────────────────────────── + if do_snapshot and vmx_path is not None and backend is not None and handle is not None: + click.echo("[prepare-linux] Waiting for VM to power off (120s) ...") + off_deadline = time.monotonic() + 120 + while time.monotonic() < off_deadline: + if not backend.is_running(handle): + break + time.sleep(5) + click.echo(" ... waiting for shutdown ...") + else: + raise click.ClickException("VM did not power off within 120s.") + click.echo("[prepare-linux] VM powered off.") + + existing = backend.list_snapshots(handle) + if snapshot in existing: + click.echo(f"[prepare-linux] Deleting existing snapshot '{snapshot}' ...") + subprocess.run( + [vmrun, "-T", "ws", "deleteSnapshot", str(vmx_path), snapshot], + check=True, + ) + + click.echo(f"[prepare-linux] Taking snapshot '{snapshot}' ...") + subprocess.run( + [vmrun, "-T", "ws", "snapshot", str(vmx_path), snapshot], + check=True, + ) + click.echo(f"[prepare-linux] Snapshot '{snapshot}' created.") + + # ── Summary ─────────────────────────────────────────────────────────────── + click.echo("") + click.echo("[prepare-linux] Summary") + click.echo(f" Guest IP: {guest_ip}") + click.echo(f" SSH key: {ssh_key_path}") + if vmx_path: + click.echo(f" VMX: {vmx_path}") + if do_snapshot and vmx_path is not None: + click.echo( + f" Snapshot: {snapshot} — ready for " + f"`ci-orchestrator vm new --template {vmx_path.name}`" + ) + else: + click.echo(f" Snapshot: not taken — name it '{snapshot}' when done manually.") + click.echo("[prepare-linux] Provisioning complete.") diff --git a/tests/python/test_commands_job.py b/tests/python/test_commands_job.py index 2ea2738..cb71b63 100644 --- a/tests/python/test_commands_job.py +++ b/tests/python/test_commands_job.py @@ -1588,3 +1588,45 @@ def test_job_sigterm_triggers_keyboard_interrupt( assert registered_handler, "signal.signal(SIGTERM, ...) should have been called" with pytest.raises(KeyboardInterrupt): registered_handler[-1](_signal.SIGTERM, None) + + +# ── _probe_transport: SSH connects but is_ready()=False ────────────────────── + + +def test_probe_transport_ssh_not_ready_returns_false( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """SSH transport __enter__ succeeds but is_ready()=False → deadline passes → False. + + Covers the ``time.sleep(poll)`` at line 189 in the SSH branch of + _probe_transport (branch 185->189). + """ + + class _NotReadySsh: + def __enter__(self) -> _NotReadySsh: + return self + + def __exit__(self, *_: object) -> None: + return None + + def is_ready(self) -> bool: + return False + + monkeypatch.setattr(job_module, "SshTransport", lambda *a, **k: _NotReadySsh()) + monkeypatch.setattr(job_module.time, "sleep", lambda _s: None) + + # Two-tick clock: first iteration inside deadline, second outside. + seq = iter([0.0, 2.0]) + monkeypatch.setattr(job_module, "_now", lambda: next(seq)) + + result = job_module._probe_transport( + guest_os="linux", + ip_address="10.0.0.1", + deadline=1.0, + poll=0.001, + credential_target="", + ssh_user="ci_build", + ssh_key_path=None, + ssh_known_hosts=None, + ) + assert result is False diff --git a/tests/python/test_commands_template.py b/tests/python/test_commands_template.py index 71d4b5c..49d5f87 100644 --- a/tests/python/test_commands_template.py +++ b/tests/python/test_commands_template.py @@ -1142,3 +1142,892 @@ def test_prepare_win_cleanup_only_fails( ) assert result.exit_code != 0 assert "Disk cleanup failed" in result.output + + +# ═══════════════════════════════════════════════════════════════════════════ +# template deploy-linux — helper unit tests +# ═══════════════════════════════════════════════════════════════════════════ + +from ci_orchestrator.config import BackendConfig, Config, IpPoolConfig, Paths # noqa: E402 + + +def _deploy_config(tmp_path: Path) -> Config: + return Config( + paths=Paths( + root=tmp_path, + templates=tmp_path / "templates", + build_vms=tmp_path / "build-vms", + artifacts=tmp_path / "artifacts", + keys=tmp_path / "keys", + ), + backend=BackendConfig(), + vmrun_path="/usr/bin/vmrun", + ssh_key_path=tmp_path / "keys" / "ci_linux", + ) + + +def _make_keys(keys_dir: Path) -> Path: + keys_dir.mkdir(parents=True, exist_ok=True) + key = keys_dir / "ci_linux" + key.write_text("FAKE PRIVATE KEY") + (keys_dir / "ci_linux.pub").write_text("ssh-rsa AAAA test@host") + return key + + +def _make_vmdk_cache(root: Path) -> Path: + iso_dir = root / "iso" + iso_dir.mkdir(exist_ok=True) + vmdk = iso_dir / "noble-server-cloudimg-amd64.vmdk" + vmdk.write_bytes(b"\x00" * 100) + return vmdk + + +@pytest.fixture() +def deploy_env(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> dict: + ssh_key = _make_keys(tmp_path / "keys") + vmdk_cache = _make_vmdk_cache(tmp_path) + vm_dir = tmp_path / "templates" / "LinuxBuild2404" + vm_dir.mkdir(parents=True) + config = _deploy_config(tmp_path) + monkeypatch.setattr(tmpl_mod, "load_config", lambda: config) + monkeypatch.setattr(tmpl_mod, "_find_vdisk_manager", lambda _: "/usr/bin/vmware-vdiskmanager") + monkeypatch.setattr( + tmpl_mod.shutil, "which", + lambda n: f"/usr/bin/{n}" if n in ("vmrun", "vmware-vdiskmanager", "genisoimage") else None, + ) + monkeypatch.setattr(tmpl_mod.time, "sleep", lambda _: None) + return { + "tmp_path": tmp_path, + "ssh_key": ssh_key, + "vmdk_cache": vmdk_cache, + "vm_dir": vm_dir, + "vmx": vm_dir / "LinuxBuild2404.vmx", + } + + +# ── _patch_vmx_key ──────────────────────────────────────────────────────────── + + +def test_patch_vmx_key_replaces_existing(tmp_path: Path) -> None: + vmx = tmp_path / "t.vmx" + vmx.write_text('ide1:0.present = "TRUE"\nother = "val"\n', encoding="ascii") + tmpl_mod._patch_vmx_key(vmx, "ide1:0.present", "FALSE") + content = vmx.read_text() + assert 'ide1:0.present = "FALSE"' in content + assert '"TRUE"' not in content + + +def test_patch_vmx_key_appends_when_absent(tmp_path: Path) -> None: + vmx = tmp_path / "t.vmx" + vmx.write_text('existing = "val"\n', encoding="ascii") + tmpl_mod._patch_vmx_key(vmx, "newkey", "nv") + content = vmx.read_text() + assert 'newkey = "nv"' in content + assert 'existing = "val"' in content + + +# ── _sha256_file ────────────────────────────────────────────────────────────── + + +def test_sha256_file_correct(tmp_path: Path) -> None: + import hashlib + data = b"sha256 test data" + f = tmp_path / "data.bin" + f.write_bytes(data) + assert tmpl_mod._sha256_file(f) == hashlib.sha256(data).hexdigest().upper() + + +# ── _find_vdisk_manager ─────────────────────────────────────────────────────── + + +def test_find_vdisk_manager_uses_which(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(tmpl_mod.shutil, "which", lambda n: "/usr/bin/vmware-vdiskmanager") + assert tmpl_mod._find_vdisk_manager(None) == "/usr/bin/vmware-vdiskmanager" + + +def test_find_vdisk_manager_explicit_missing(tmp_path: Path) -> None: + with pytest.raises(click.ClickException): + tmpl_mod._find_vdisk_manager(str(tmp_path / "nonexistent")) + + +def test_find_vdisk_manager_not_in_path(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(tmpl_mod.shutil, "which", lambda n: None) + with pytest.raises(click.ClickException, match="PATH"): + tmpl_mod._find_vdisk_manager(None) + + +# ── _build_seed_iso ─────────────────────────────────────────────────────────── + + +def test_build_seed_iso_calls_genisoimage( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + staging = tmp_path / "staging" + staging.mkdir() + iso = tmp_path / "seed.iso" + calls: list[list[str]] = [] + + import subprocess as _sub + + def fake_run(cmd: list[str], **kw: object) -> _sub.CompletedProcess[bytes]: + calls.append(list(cmd)) + iso.write_bytes(b"\x00" * 8192) + return _sub.CompletedProcess(cmd, 0) + + monkeypatch.setattr(tmpl_mod.subprocess, "run", fake_run) + monkeypatch.setattr( + tmpl_mod.shutil, "which", + lambda n: "/usr/bin/genisoimage" if n == "genisoimage" else None, + ) + tmpl_mod._build_seed_iso(staging, iso) + assert calls and "genisoimage" in calls[0][0] + assert "cidata" in calls[0] + + +def test_build_seed_iso_no_tool_raises( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr(tmpl_mod.shutil, "which", lambda n: None) + with pytest.raises(click.ClickException, match="genisoimage"): + tmpl_mod._build_seed_iso(tmp_path, tmp_path / "seed.iso") + + +# ── _poll_guest_ip_deploy ───────────────────────────────────────────────────── + + +def test_poll_guest_ip_returns_valid(monkeypatch: pytest.MonkeyPatch) -> None: + import subprocess as _sub + + monkeypatch.setattr( + tmpl_mod.subprocess, "run", + lambda cmd, **kw: _sub.CompletedProcess(cmd, 0, "192.168.1.100", ""), + ) + monkeypatch.setattr(tmpl_mod.time, "sleep", lambda _: None) + assert tmpl_mod._poll_guest_ip_deploy("/vmrun", "/vm.vmx", 60.0) == "192.168.1.100" + + +def test_poll_guest_ip_timeout(monkeypatch: pytest.MonkeyPatch) -> None: + import subprocess as _sub + + monkeypatch.setattr( + tmpl_mod.subprocess, "run", + lambda cmd, **kw: _sub.CompletedProcess(cmd, 0, "Error: not ready", ""), + ) + monkeypatch.setattr(tmpl_mod.time, "sleep", lambda _: None) + assert tmpl_mod._poll_guest_ip_deploy("/vmrun", "/vm.vmx", 0.0) is None + + +# ── _wait_ssh_port_deploy ───────────────────────────────────────────────────── + + +def test_wait_ssh_port_success(monkeypatch: pytest.MonkeyPatch) -> None: + ctx = MagicMock() + ctx.__enter__ = lambda s: s + ctx.__exit__ = lambda s, *a: False + monkeypatch.setattr(tmpl_mod.socket, "create_connection", lambda *a, **kw: ctx) + monkeypatch.setattr(tmpl_mod.time, "sleep", lambda _: None) + assert tmpl_mod._wait_ssh_port_deploy("192.168.1.100", 22, 10.0, 1.0) is True + + +def test_wait_ssh_port_timeout(monkeypatch: pytest.MonkeyPatch) -> None: + def _raise(*a: object, **kw: object) -> None: + raise OSError("refused") + + monkeypatch.setattr(tmpl_mod.socket, "create_connection", _raise) + monkeypatch.setattr(tmpl_mod.time, "sleep", lambda _: None) + assert tmpl_mod._wait_ssh_port_deploy("192.168.1.100", 22, 0.0, 1.0) is False + + +# ── _verify_cloud_init_done ─────────────────────────────────────────────────── + + +def test_verify_cloud_init_success(monkeypatch: pytest.MonkeyPatch) -> None: + stdout_m = MagicMock() + stdout_m.read.return_value = b"OK" + client_m = MagicMock() + client_m.exec_command.return_value = (MagicMock(), stdout_m, MagicMock()) + monkeypatch.setattr(tmpl_mod.paramiko, "SSHClient", lambda: client_m) + monkeypatch.setattr(tmpl_mod.paramiko, "AutoAddPolicy", MagicMock) + monkeypatch.setattr(tmpl_mod.time, "sleep", lambda _: None) + assert tmpl_mod._verify_cloud_init_done("host", "user", "/key", 30.0) is True + + +def test_verify_cloud_init_exception_retries(monkeypatch: pytest.MonkeyPatch) -> None: + calls = [0] + stdout_m = MagicMock() + stdout_m.read.return_value = b"OK" + client_m = MagicMock() + + def _connect(*a: object, **kw: object) -> None: + calls[0] += 1 + if calls[0] == 1: + raise OSError("not yet") + + client_m.connect.side_effect = _connect + client_m.exec_command.return_value = (MagicMock(), stdout_m, MagicMock()) + monkeypatch.setattr(tmpl_mod.paramiko, "SSHClient", lambda: client_m) + monkeypatch.setattr(tmpl_mod.paramiko, "AutoAddPolicy", MagicMock) + monkeypatch.setattr(tmpl_mod.time, "sleep", lambda _: None) + # Give enough time budget for 2 tries + assert tmpl_mod._verify_cloud_init_done("host", "user", "/key", 60.0) is True + + +def test_verify_cloud_init_timeout(monkeypatch: pytest.MonkeyPatch) -> None: + client_m = MagicMock() + client_m.connect.side_effect = OSError("refused") + monkeypatch.setattr(tmpl_mod.paramiko, "SSHClient", lambda: client_m) + monkeypatch.setattr(tmpl_mod.paramiko, "AutoAddPolicy", MagicMock) + monkeypatch.setattr(tmpl_mod.time, "sleep", lambda _: None) + assert tmpl_mod._verify_cloud_init_done("host", "user", "/key", 0.0) is False + + +# ═══════════════════════════════════════════════════════════════════════════ +# template deploy-linux — CLI tests +# ═══════════════════════════════════════════════════════════════════════════ + + +def test_deploy_linux_help() -> None: + result = CliRunner().invoke(cli, ["template", "deploy-linux", "--help"]) + assert result.exit_code == 0 + assert "Ubuntu" in result.output + + +def test_deploy_linux_preflight_missing_pub_key( + deploy_env: dict, monkeypatch: pytest.MonkeyPatch +) -> None: + (deploy_env["ssh_key"].parent / "ci_linux.pub").unlink() + result = CliRunner().invoke( + cli, ["template", "deploy-linux", "--vmx", str(deploy_env["vmx"])] + ) + assert result.exit_code != 0 + assert "ci_linux.pub" in result.output or "public key" in result.output.lower() + + +def test_deploy_linux_vm_dir_exists_no_force( + deploy_env: dict, monkeypatch: pytest.MonkeyPatch +) -> None: + result = CliRunner().invoke( + cli, ["template", "deploy-linux", "--vmx", str(deploy_env["vmx"])] + ) + assert result.exit_code != 0 + assert "force" in result.output.lower() + + +def test_deploy_linux_sha256_mismatch( + deploy_env: dict, monkeypatch: pytest.MonkeyPatch +) -> None: + result = CliRunner().invoke( + cli, + [ + "template", "deploy-linux", + "--vmx", str(deploy_env["vmx"]), + "--force", + "--vmdk-sha256", "DEADBEEF" + "0" * 56, + ], + ) + assert result.exit_code != 0 + assert "mismatch" in result.output.lower() + + +def test_deploy_linux_step3_iso_fail( + deploy_env: dict, monkeypatch: pytest.MonkeyPatch +) -> None: + import subprocess as _sub + + def fake_run(cmd: list[str], **kw: object) -> _sub.CompletedProcess[bytes]: + if "genisoimage" in str(cmd[0]): + raise _sub.CalledProcessError(1, cmd, b"", b"genisoimage error") + return _sub.CompletedProcess(cmd, 0) + + monkeypatch.setattr(tmpl_mod.subprocess, "run", fake_run) + result = CliRunner().invoke( + cli, + [ + "template", "deploy-linux", + "--vmx", str(deploy_env["vmx"]), + "--force", + "--start-from-step", "2", + ], + ) + assert result.exit_code != 0 + assert "ISO creation failed" in result.output + + +def test_deploy_linux_step4_convert_fail( + deploy_env: dict, monkeypatch: pytest.MonkeyPatch +) -> None: + import subprocess as _sub + + def fake_run(cmd: list[str], **kw: object) -> _sub.CompletedProcess[bytes]: + if "vmware-vdiskmanager" in str(cmd[0]): + return _sub.CompletedProcess(cmd, 1, b"", b"convert error") + return _sub.CompletedProcess(cmd, 0) + + monkeypatch.setattr(tmpl_mod.subprocess, "run", fake_run) + result = CliRunner().invoke( + cli, + [ + "template", "deploy-linux", + "--vmx", str(deploy_env["vmx"]), + "--force", + "--start-from-step", "4", + ], + ) + assert result.exit_code != 0 + assert "convert failed" in result.output + + +def test_deploy_linux_step5_vmx_content( + deploy_env: dict, monkeypatch: pytest.MonkeyPatch +) -> None: + import subprocess as _sub + + popen_m = MagicMock() + popen_m.pid = 1 + monkeypatch.setattr(tmpl_mod.subprocess, "Popen", lambda *a, **kw: popen_m) + + # vmrun list returns empty so step 6 fails — we can still check VMX written in step 5 + monkeypatch.setattr( + tmpl_mod.subprocess, "run", + lambda cmd, **kw: _sub.CompletedProcess(list(cmd), 0, "Total running VMs: 0\n", ""), + ) + + result = CliRunner().invoke( + cli, + [ + "template", "deploy-linux", + "--vmx", str(deploy_env["vmx"]), + "--force", + "--start-from-step", "5", + ], + ) + # Step 6 should fail (VM not in list), but VMX should have been written in step 5 + assert deploy_env["vmx"].exists() + content = deploy_env["vmx"].read_text() + assert 'guestOS = "ubuntu-64"' in content + assert 'firmware = "efi"' in content + assert 'ethernet0.virtualDev = "vmxnet3"' in content + + +def test_deploy_linux_step5_guestinfo_skipped_when_step2_skipped( + deploy_env: dict, monkeypatch: pytest.MonkeyPatch +) -> None: + import subprocess as _sub + + popen_m = MagicMock() + popen_m.pid = 1 + monkeypatch.setattr(tmpl_mod.subprocess, "Popen", lambda *a, **kw: popen_m) + monkeypatch.setattr( + tmpl_mod.subprocess, "run", + lambda cmd, **kw: _sub.CompletedProcess(list(cmd), 0, "Total running VMs: 0\n", ""), + ) + + result = CliRunner().invoke( + cli, + [ + "template", "deploy-linux", + "--vmx", str(deploy_env["vmx"]), + "--force", + "--start-from-step", "5", # step 2 was skipped → no b64 strings + ], + ) + if deploy_env["vmx"].exists(): + content = deploy_env["vmx"].read_text() + assert "guestinfo.userdata" not in content + + +def test_deploy_linux_step7_ip_timeout( + deploy_env: dict, monkeypatch: pytest.MonkeyPatch +) -> None: + import subprocess as _sub + + popen_m = MagicMock() + popen_m.pid = 1 + monkeypatch.setattr(tmpl_mod.subprocess, "Popen", lambda *a, **kw: popen_m) + + vmx_str = str(deploy_env["vmx"]) + monkeypatch.setattr( + tmpl_mod.subprocess, "run", + lambda cmd, **kw: _sub.CompletedProcess( + list(cmd), 0, f"Total running VMs: 1\n{vmx_str}\n", "" + ), + ) + # patch the poller so it returns None immediately (no 300-second spin loop) + monkeypatch.setattr(tmpl_mod, "_poll_guest_ip_deploy", lambda *a, **kw: None) + + deploy_env["vmx"].write_text('dummy = "vmx"', encoding="ascii") + + result = CliRunner().invoke( + cli, + [ + "template", "deploy-linux", + "--vmx", str(deploy_env["vmx"]), + "--force", + "--start-from-step", "6", + ], + ) + assert result.exit_code != 0 + assert "guest IP" in result.output + + +def test_deploy_linux_step8_no_guest_ip( + deploy_env: dict, monkeypatch: pytest.MonkeyPatch +) -> None: + deploy_env["vmx"].write_text('dummy = "vmx"', encoding="ascii") + result = CliRunner().invoke( + cli, + [ + "template", "deploy-linux", + "--vmx", str(deploy_env["vmx"]), + "--force", + "--start-from-step", "8", + ], + ) + assert result.exit_code != 0 + assert "Guest IP not available" in result.output + + +def test_deploy_linux_step10_patches_vmx( + deploy_env: dict, monkeypatch: pytest.MonkeyPatch +) -> None: + vmx = deploy_env["vmx"] + vmx.write_text( + 'ide1:0.present = "TRUE"\nide1:0.startConnected = "TRUE"\n', + encoding="ascii", + ) + result = CliRunner().invoke( + cli, + [ + "template", "deploy-linux", + "--vmx", str(vmx), + "--force", + "--start-from-step", "10", + ], + ) + assert result.exit_code == 0, result.output + content = vmx.read_text() + assert 'ide1:0.present = "FALSE"' in content + assert 'ide1:0.startConnected = "FALSE"' in content + + +def test_deploy_linux_full_happy_path( + deploy_env: dict, monkeypatch: pytest.MonkeyPatch +) -> None: + import subprocess as _sub + + vmx = deploy_env["vmx"] + + def fake_run(cmd: list[str], **kw: object) -> _sub.CompletedProcess[str]: + cmd_str = " ".join(str(c) for c in cmd) + if "genisoimage" in cmd_str: + for i, arg in enumerate(cmd): + if i > 0 and cmd[i - 1] == "-o": + Path(arg).write_bytes(b"\x00" * 8192) + return _sub.CompletedProcess(cmd, 0, "", "") + if "vmware-vdiskmanager" in cmd_str and "-r" in cmd: + dest = str(cmd[-1]) + Path(dest).write_bytes(b"\x00" * 1024) + return _sub.CompletedProcess(cmd, 0, "", "") + if "vmware-vdiskmanager" in cmd_str: + return _sub.CompletedProcess(cmd, 0, "", "") + if "list" in cmd: + return _sub.CompletedProcess(cmd, 0, f"Total running VMs: 1\n{vmx}\n", "") + if "getGuestIPAddress" in cmd: + return _sub.CompletedProcess(cmd, 0, "192.168.79.200", "") + return _sub.CompletedProcess(cmd, 0, "", "") + + monkeypatch.setattr(tmpl_mod.subprocess, "run", fake_run) + + popen_m = MagicMock() + popen_m.pid = 99 + monkeypatch.setattr(tmpl_mod.subprocess, "Popen", lambda *a, **kw: popen_m) + + sock_m = MagicMock() + sock_m.__enter__ = lambda s: s + sock_m.__exit__ = lambda s, *a: False + monkeypatch.setattr(tmpl_mod.socket, "create_connection", lambda *a, **kw: sock_m) + + stdout_m = MagicMock() + stdout_m.read.return_value = b"OK" + client_m = MagicMock() + client_m.exec_command.return_value = (MagicMock(), stdout_m, MagicMock()) + monkeypatch.setattr(tmpl_mod.paramiko, "SSHClient", lambda: client_m) + monkeypatch.setattr(tmpl_mod.paramiko, "AutoAddPolicy", MagicMock) + + result = CliRunner().invoke( + cli, + ["template", "deploy-linux", "--vmx", str(vmx), "--force"], + ) + assert result.exit_code == 0, result.output + + assert vmx.exists() + vmx_content = vmx.read_text() + assert 'guestOS = "ubuntu-64"' in vmx_content + assert 'ide1:0.present = "FALSE"' in vmx_content + assert 'ide1:0.startConnected = "FALSE"' in vmx_content + assert "guestinfo.userdata" in vmx_content # b64 from step 2 + assert "192.168.79.200" in result.output # guest IP in summary + + +# ═══════════════════════════════════════════════════════════════════════════ +# template prepare-linux — helper unit tests +# ═══════════════════════════════════════════════════════════════════════════ + + +def test_pl_run_returns_stdout_and_rc(monkeypatch: pytest.MonkeyPatch) -> None: + stdout_m = MagicMock() + stdout_m.read.return_value = b"hello\n" + stdout_m.channel.recv_exit_status.return_value = 0 + stderr_m = MagicMock() + stderr_m.read.return_value = b"" + client_m = MagicMock() + client_m.exec_command.return_value = (MagicMock(), stdout_m, stderr_m) + + rc, out, err = tmpl_mod._pl_run(client_m, "echo hello") + assert rc == 0 + assert "hello" in out + assert err == "" + + +def test_pl_run_nonzero_rc(monkeypatch: pytest.MonkeyPatch) -> None: + stdout_m = MagicMock() + stdout_m.read.return_value = b"" + stdout_m.channel.recv_exit_status.return_value = 1 + stderr_m = MagicMock() + stderr_m.read.return_value = b"error" + client_m = MagicMock() + client_m.exec_command.return_value = (MagicMock(), stdout_m, stderr_m) + + rc, _, err = tmpl_mod._pl_run(client_m, "false") + assert rc == 1 + assert "error" in err + + +def test_pl_stream_success(monkeypatch: pytest.MonkeyPatch) -> None: + channel_m = MagicMock() + # First call: stdout data; second: empty → exit_status_ready True on second loop + channel_m.recv_ready.side_effect = [True, False, False] + channel_m.recv.return_value = b"output line\n" + channel_m.recv_stderr_ready.return_value = False + channel_m.exit_status_ready.side_effect = [False, True] + channel_m.recv_exit_status.return_value = 0 + + transport_m = MagicMock() + transport_m.open_session.return_value = channel_m + client_m = MagicMock() + client_m.get_transport.return_value = transport_m + + monkeypatch.setattr(tmpl_mod.time, "sleep", lambda _: None) + rc = tmpl_mod._pl_stream(client_m, "cat file") + assert rc == 0 + + +def test_pl_upload_calls_sftp_put(tmp_path: Path) -> None: + local = tmp_path / "script.sh" + local.write_text("#!/bin/bash\n") + sftp_m = MagicMock() + client_m = MagicMock() + client_m.open_sftp.return_value = sftp_m + + tmpl_mod._pl_upload(client_m, local, "/tmp/script.sh") + sftp_m.put.assert_called_once_with(str(local), "/tmp/script.sh") + sftp_m.close.assert_called_once() + + +# ═══════════════════════════════════════════════════════════════════════════ +# template prepare-linux — CLI tests +# ═══════════════════════════════════════════════════════════════════════════ + + +@pytest.fixture() +def prepare_env( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> dict: + """Environment with SSH keys, VMX, and toolchain script.""" + ssh_key = _make_keys(tmp_path / "keys") + vm_dir = tmp_path / "templates" / "LinuxBuild2404" + vm_dir.mkdir(parents=True) + vmx = vm_dir / "LinuxBuild2404.vmx" + vmx.write_text('guestOS = "ubuntu-64"\n', encoding="ascii") + + # toolchain script in repo-relative location + tmpl_dir = tmp_path / "template" + tmpl_dir.mkdir() + toolchain = tmpl_dir / "Install-CIToolchain-Linux2404.sh" + toolchain.write_text("#!/bin/bash\necho done\n") + + config = _deploy_config(tmp_path) + monkeypatch.setattr(tmpl_mod, "load_config", lambda: config) + monkeypatch.setattr(tmpl_mod.time, "sleep", lambda _: None) + monkeypatch.chdir(tmp_path) + + return { + "tmp_path": tmp_path, + "ssh_key": ssh_key, + "vmx": vmx, + "toolchain": toolchain, + } + + +def _stub_prepare( + monkeypatch: pytest.MonkeyPatch, + env: dict, + *, + toolchain_exit: int = 0, + val_output: str = ( + "UserSudo:OK\nSudoNoPass:OK\nTool:gcc:OK\nTool:g++:OK\n" + "Tool:clang:OK\nTool:cmake:OK\nTool:python3:OK\nTool:git:OK\n" + "Tool:7z:OK\nDir:build:OK\nDir:output:OK\nDir:scripts:OK\n" + "Swap:off:OK\nSshd:active:OK\nSSHPwdAuth:disabled:OK\n" + "CIReportIP:enabled:OK\nCIReportIPScript:exists:OK\n" + ), + machine_id_size: str = "0", +) -> MagicMock: + """Stub all SSH / vmrun calls for prepare-linux happy-path testing.""" + client_m = MagicMock() + raw = env["toolchain"].read_bytes() + if raw[:3] == b"\xef\xbb\xbf": + raw = raw[3:] + _toolchain_size = str(len(raw.replace(b"\r\n", b"\n").replace(b"\r", b"\n"))) + + def fake_pl_run( + client: object, cmd: str, timeout: float = 60.0 + ) -> tuple[int, str, str]: + if "stat -c %s /tmp/" in cmd: + return 0, _toolchain_size, "" + if "stat -c %s /etc/machine-id" in cmd: + return 0, machine_id_size, "" + if "truncate" in cmd: + return 0, "", "" + if "shutdown" in cmd: + return 0, "", "" + if "echo connected" in cmd: + return 0, "connected", "" + # validation script + return 0, val_output, "" + + monkeypatch.setattr(tmpl_mod, "_pl_connect", lambda *a, **kw: client_m) + monkeypatch.setattr(tmpl_mod, "_pl_run", fake_pl_run) + monkeypatch.setattr(tmpl_mod, "_pl_stream", lambda *a, **kw: toolchain_exit) + monkeypatch.setattr(tmpl_mod, "_pl_upload", lambda *a, **kw: None) + monkeypatch.setattr(tmpl_mod, "_wait_ssh_port_deploy", lambda *a, **kw: True) + return client_m + + +def test_prepare_linux_missing_vmx_and_ip( + prepare_env: dict, monkeypatch: pytest.MonkeyPatch +) -> None: + result = CliRunner().invoke(cli, ["template", "prepare-linux"]) + assert result.exit_code != 0 + assert "--vmx" in result.output or "--ip" in result.output + + +def test_prepare_linux_toolchain_not_found( + prepare_env: dict, monkeypatch: pytest.MonkeyPatch +) -> None: + prepare_env["toolchain"].unlink() + result = CliRunner().invoke( + cli, + ["template", "prepare-linux", "--ip", "192.168.1.100"], + ) + assert result.exit_code != 0 + assert "not found" in result.output.lower() + + +def test_prepare_linux_ssh_key_missing( + prepare_env: dict, monkeypatch: pytest.MonkeyPatch +) -> None: + prepare_env["ssh_key"].unlink() + result = CliRunner().invoke( + cli, + ["template", "prepare-linux", "--ip", "192.168.1.100"], + ) + assert result.exit_code != 0 + assert "not found" in result.output.lower() + + +def test_prepare_linux_ssh_connect_fail( + prepare_env: dict, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr( + tmpl_mod, "_wait_ssh_port_deploy", lambda *a, **kw: True + ) + monkeypatch.setattr( + tmpl_mod, "_pl_connect", + lambda *a, **kw: (_ for _ in ()).throw(OSError("refused")), + ) + result = CliRunner().invoke( + cli, + ["template", "prepare-linux", "--ip", "192.168.1.100"], + ) + assert result.exit_code != 0 + assert "failed" in result.output.lower() + + +def test_prepare_linux_ssh_port_timeout( + prepare_env: dict, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr(tmpl_mod, "_wait_ssh_port_deploy", lambda *a, **kw: False) + result = CliRunner().invoke( + cli, + ["template", "prepare-linux", "--ip", "192.168.1.100"], + ) + assert result.exit_code != 0 + assert "not reachable" in result.output or "SSH" in result.output + + +def test_prepare_linux_toolchain_fail( + prepare_env: dict, monkeypatch: pytest.MonkeyPatch +) -> None: + _stub_prepare(monkeypatch, prepare_env, toolchain_exit=1) + result = CliRunner().invoke( + cli, + ["template", "prepare-linux", "--ip", "192.168.1.100"], + ) + assert result.exit_code != 0 + assert "exited 1" in result.output or "Toolchain" in result.output + + +def test_prepare_linux_validation_fail( + prepare_env: dict, monkeypatch: pytest.MonkeyPatch +) -> None: + _stub_prepare( + monkeypatch, prepare_env, + val_output="UserSudo:OK\nTool:gcc:FAIL\nDir:build:OK\n", + ) + result = CliRunner().invoke( + cli, + ["template", "prepare-linux", "--ip", "192.168.1.100"], + ) + assert result.exit_code != 0 + assert "FAIL" in result.output + + +def test_prepare_linux_machine_id_not_cleared( + prepare_env: dict, monkeypatch: pytest.MonkeyPatch +) -> None: + _stub_prepare(monkeypatch, prepare_env, machine_id_size="36") + result = CliRunner().invoke( + cli, + ["template", "prepare-linux", "--ip", "192.168.1.100"], + ) + assert result.exit_code != 0 + assert "machine-id" in result.output.lower() + + +def test_prepare_linux_happy_path_ip_only( + prepare_env: dict, monkeypatch: pytest.MonkeyPatch +) -> None: + """No --vmx: provisions but skips auto power-on, IP detect, and snapshot.""" + _stub_prepare(monkeypatch, prepare_env) + result = CliRunner().invoke( + cli, + ["template", "prepare-linux", "--ip", "192.168.1.100"], + ) + assert result.exit_code == 0, result.output + assert "Provisioning complete" in result.output + assert "not taken" in result.output # snapshot skipped + + +def test_prepare_linux_happy_path_with_vmx( + prepare_env: dict, monkeypatch: pytest.MonkeyPatch +) -> None: + """--vmx: auto IP detect + shutdown + snapshot.""" + _stub_prepare(monkeypatch, prepare_env) + monkeypatch.setattr(tmpl_mod, "_poll_guest_ip_deploy", lambda *a, **kw: "192.168.1.200") + + backend_m = MagicMock() + backend_m.is_running.side_effect = [False, True, False] # start, check, powered off + backend_m.list_snapshots.return_value = [] + monkeypatch.setattr(tmpl_mod, "WorkstationVmrunBackend", lambda **kw: backend_m) + + import subprocess as _sub + monkeypatch.setattr( + tmpl_mod.subprocess, "run", + lambda cmd, **kw: _sub.CompletedProcess(list(cmd), 0, "", ""), + ) + + result = CliRunner().invoke( + cli, + ["template", "prepare-linux", "--vmx", str(prepare_env["vmx"])], + ) + assert result.exit_code == 0, result.output + assert "Snapshot" in result.output + assert "192.168.1.200" in result.output + + +def test_prepare_linux_happy_path_skip_flags( + prepare_env: dict, monkeypatch: pytest.MonkeyPatch +) -> None: + """--skip-update, --install-dotnet, --no-mingw reach the toolchain command.""" + streamed: list[str] = [] + client_m = MagicMock() + + _raw = prepare_env["toolchain"].read_bytes() + _tsize = str(len(_raw.replace(b"\r\n", b"\n").replace(b"\r", b"\n"))) + + def fake_pl_run(client: object, cmd: str, timeout: float = 60.0) -> tuple[int, str, str]: + if "stat -c %s /tmp/" in cmd: + return 0, _tsize, "" + if "stat -c %s /etc/machine-id" in cmd: + return 0, "0", "" + if "truncate" in cmd: + return 0, "", "" + if "echo connected" in cmd: + return 0, "connected", "" + return 0, "UserSudo:OK\nSudoNoPass:OK\n", "" + + def fake_pl_stream(client: object, cmd: str) -> int: + streamed.append(cmd) + return 0 + + monkeypatch.setattr(tmpl_mod, "_pl_connect", lambda *a, **kw: client_m) + monkeypatch.setattr(tmpl_mod, "_pl_run", fake_pl_run) + monkeypatch.setattr(tmpl_mod, "_pl_stream", fake_pl_stream) + monkeypatch.setattr(tmpl_mod, "_pl_upload", lambda *a, **kw: None) + monkeypatch.setattr(tmpl_mod, "_wait_ssh_port_deploy", lambda *a, **kw: True) + + result = CliRunner().invoke( + cli, + [ + "template", "prepare-linux", + "--ip", "192.168.1.100", + "--skip-update", "--install-dotnet", "--no-mingw", + ], + ) + assert result.exit_code == 0, result.output + assert streamed + assert "--skip-update" in streamed[0] + assert "--dotnet" in streamed[0] + assert "--no-mingw" in streamed[0] + + +def test_prepare_linux_existing_snapshot_deleted( + prepare_env: dict, monkeypatch: pytest.MonkeyPatch +) -> None: + """If snapshot already exists, delete before recreating.""" + _stub_prepare(monkeypatch, prepare_env) + monkeypatch.setattr(tmpl_mod, "_poll_guest_ip_deploy", lambda *a, **kw: "192.168.1.200") + + backend_m = MagicMock() + backend_m.is_running.side_effect = [True, False] + backend_m.list_snapshots.return_value = ["BaseClean-Linux"] + monkeypatch.setattr(tmpl_mod, "WorkstationVmrunBackend", lambda **kw: backend_m) + + run_calls: list[list[str]] = [] + import subprocess as _sub + + def fake_run(cmd: list[str], **kw: object) -> _sub.CompletedProcess[str]: + run_calls.append(list(cmd)) + return _sub.CompletedProcess(cmd, 0, "", "") + + monkeypatch.setattr(tmpl_mod.subprocess, "run", fake_run) + + result = CliRunner().invoke( + cli, + ["template", "prepare-linux", "--vmx", str(prepare_env["vmx"])], + ) + assert result.exit_code == 0, result.output + cmds = [" ".join(c) for c in run_calls] + assert any("deleteSnapshot" in c for c in cmds) + assert any("snapshot" in c and "deleteSnapshot" not in c for c in cmds) diff --git a/tests/python/test_ssh_transport.py b/tests/python/test_ssh_transport.py index e4dc373..a7e81d3 100644 --- a/tests/python/test_ssh_transport.py +++ b/tests/python/test_ssh_transport.py @@ -443,3 +443,38 @@ def test_connect_reuses_existing_client(monkeypatch: pytest.MonkeyPatch) -> None assert client.commands == ["first", "second"] # Only one connect attempt. assert client.connect_kwargs["hostname"] == "10.0.0.2" + + +# ── run_streaming: progressed=True but exit not yet ready → loop back ───────── + + +def test_run_streaming_progressed_no_sleep_loops_back( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Data received (progressed=True) but exit_status_ready()=False on first check. + + Covers the ``if not progressed: time.sleep(0.05)`` branch when progressed + is True — no sleep occurs, loop continues (branch 170->150). + """ + slept: list[float] = [] + monkeypatch.setattr("ci_orchestrator.transport.ssh.time.sleep", slept.append) + + class _ProgressedNoBreakChannel(_FakeStreamingChannel): + def __init__(self) -> None: + super().__init__([b"hello"], [], rc=0) + self._ready_count = 0 + + def exit_status_ready(self) -> bool: + # First call: stdout already drained but signal exit not yet ready. + # Subsequent calls: ready → break. + self._ready_count += 1 + return self._ready_count > 1 + + chan = _ProgressedNoBreakChannel() + client = _FakeSSHClientWithTransport(chan) + _install_fake_paramiko(monkeypatch, client) + t = SshTransport("10.0.0.2", key_path="/tmp/key") + result = t.run_streaming("cmd") + assert result.stdout == "hello" + # sleep must NOT have been called (progressed=True skips sleep branch) + assert not slept