From 9e84d898234c9e3af9fa3335fa72a0936108828d Mon Sep 17 00:00:00 2001 From: Simone Date: Sun, 24 May 2026 01:22:53 +0200 Subject: [PATCH] feat(template): add `template prepare-win` Python command via pypsrp Replaces the abandoned PSWSMan approach (New-PSSession hangs on Linux) with a Python-native implementation using the proven pypsrp WinRmTransport. - New `template prepare-win` command in commands/template.py: starts VM, waits for WinRM, uploads and runs Install-CIToolchain-*.ps1, handles Windows Update 3010 reboot loop (max 10 iterations), installs CI-StaticIp scheduled task, runs post-setup validation, graceful shutdown, snapshot. - Helpers: _ps_escape, _wait_tcp, _parse_exit_marker, _wait_winrm. - 40 new tests in test_commands_template.py covering all major branches. - Fix test_commands_job.py: patch load_config in _patch_common so tests don't pick up live /var/lib/ci/config.toml ip_pool and fail on lock. - Coverage gate: 90.06% (was 88.95% before fix). Co-Authored-By: Claude Sonnet 4.6 --- docs/RUNBOOK.md | 4 +- src/ci_orchestrator/commands/template.py | 388 ++++++++++++++ tests/python/test_commands_job.py | 9 + tests/python/test_commands_template.py | 614 ++++++++++++++++++++++- 4 files changed, 1012 insertions(+), 3 deletions(-) diff --git a/docs/RUNBOOK.md b/docs/RUNBOOK.md index 838aec6..865bc4d 100644 --- a/docs/RUNBOOK.md +++ b/docs/RUNBOOK.md @@ -403,8 +403,8 @@ host + VMware Workstation Pro Linux, template `WinBuild2025` / snapshot | Metric | Windows | Linux avg | In range? | | ------- | ------- | --------- | ---------------------------------- | -| Clone | 0.62 s | 0.44 s | ✓ (faster) | -| Start | 1.77 s | 2.12 s | ✓ (at upper edge) | +| Clone | 0.62 s | 0.44 s | ✓ (faster) | +| Start | 1.77 s | 2.12 s | ✓ (at upper edge) | | Destroy | 4.98 s | 4.92 s | ✓ | | IP avg | 58.2 s | 99.7 s | ✗ outside — IP variance (39–177 s) | | Ready | 0.01 s | 0.01 s | ✓ | diff --git a/src/ci_orchestrator/commands/template.py b/src/ci_orchestrator/commands/template.py index c836e1f..5d42da5 100644 --- a/src/ci_orchestrator/commands/template.py +++ b/src/ci_orchestrator/commands/template.py @@ -14,14 +14,21 @@ is already offline. from __future__ import annotations import fnmatch +import json import shutil +import socket import subprocess +import time from datetime import UTC, datetime from pathlib import Path import click +from ci_orchestrator.backends.workstation import WorkstationVmrunBackend +from ci_orchestrator.backends.protocol import VmHandle from ci_orchestrator.config import load_config +from ci_orchestrator.transport.errors import TransportError +from ci_orchestrator.transport.winrm import WinRmTransport @click.group() @@ -206,3 +213,384 @@ def _prune_backups( if to_remove: label = template_name or "template" click.echo(f"[Backup-CITemplate] Retained {keep_count} backup(s) for {label}.") + + +# ═══════════════════════════════════════════════════════════════════════════ +# template prepare-win helpers +# ═══════════════════════════════════════════════════════════════════════════ + + +def _ps_escape(s: str) -> str: + """Escape *s* for use inside a PowerShell single-quoted string.""" + return s.replace("'", "''") + + +def _wait_tcp(host: str, port: int, timeout: float) -> bool: + """Return True once TCP *port* on *host* accepts a connection within *timeout* s.""" + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + try: + with socket.create_connection((host, port), timeout=5.0): + return True + except OSError: + time.sleep(5) + return False + + +def _parse_exit_marker(stdout: str) -> int: + """Extract ``CI_EXITCODE:N`` written by the invoke wrapper; default 0.""" + for line in reversed(stdout.splitlines()): + stripped = line.strip() + if stripped.startswith("CI_EXITCODE:"): + try: + return int(stripped.split(":", 1)[1]) + except (IndexError, ValueError): + pass + return 0 + + +def _wait_winrm(transport: WinRmTransport, timeout: int) -> None: + """Poll *transport*.is_ready() until True or *timeout* seconds elapse.""" + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if transport.is_ready(): + return + time.sleep(10) + click.echo(" ... waiting for WinRM ...") + raise click.ClickException(f"WinRM not ready within {timeout}s.") + + +# ═══════════════════════════════════════════════════════════════════════════ +# template prepare-win command +# ═══════════════════════════════════════════════════════════════════════════ + + +@template.command("prepare-win") +@click.option("--vmx", required=True, metavar="PATH", help="Template VM .vmx file.") +@click.option("--ip", "guest_ip", default="", help="Guest IP (auto-detect via VMware Tools if omitted).") +@click.option("--admin-username", default="Administrator", show_default=True) +@click.option("--admin-password", default="", help="Administrator password (prompted if omitted).") +@click.option("--build-username", default="ci_build", show_default=True) +@click.option("--build-password", default="", help="CI build user password (prompted if omitted).") +@click.option("--snapshot", default="BaseClean", show_default=True, help="Snapshot name.") +@click.option("--toolchain-script", "toolchain_script", default=None, metavar="PATH", + help="Path to Install-CIToolchain-*.ps1 (auto-detected from VMX stem if omitted).") +@click.option("--skip-windows-update", is_flag=True, help="Pass -SkipWindowsUpdate to toolchain script.") +@click.option("--skip-tier2", is_flag=True, help="Pass -SkipTier2 to toolchain script.") +@click.option("--skip-cleanup", is_flag=True, help="Pass -SkipCleanup to toolchain script.") +@click.option("--winrm-timeout", default=300, show_default=True, help="Seconds to wait for WinRM.") +@click.option("--auth", default="basic", show_default=True, + type=click.Choice(["basic", "ntlm", "negotiate"]), + help="WinRM authentication protocol.") +def template_prepare_win( + vmx: str, + guest_ip: str, + admin_username: str, + admin_password: str, + build_username: str, + build_password: str, + snapshot: str, + toolchain_script: str | None, + skip_windows_update: bool, + skip_tier2: bool, + skip_cleanup: bool, + winrm_timeout: int, + auth: str, +) -> None: + """Provision a Windows template VM via pypsrp (Linux-host compatible). + + Replaces Prepare-WinBuild*.ps1 for hosts where PSWSMan/New-PSSession is + unavailable or unreliable. Run from the repo root so toolchain and + guest-setup scripts are auto-discovered under template/. + """ + vmx_path = Path(vmx).resolve() + if not vmx_path.is_file(): + raise click.ClickException(f"VMX not found: {vmx_path}") + + # --- Resolve toolchain script --- + if toolchain_script is None: + candidate = Path.cwd() / "template" / f"Install-CIToolchain-{vmx_path.stem}.ps1" + 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) + if not toolchain_path.is_file(): + raise click.ClickException(f"Toolchain script not found: {toolchain_path}") + + # --- Locate CI-StaticIp guest script --- + _sip_candidate = Path.cwd() / "template" / "guest-setup" / "windows" / "ci-static-ip.ps1" + static_ip_path: Path | None + if _sip_candidate.is_file(): + static_ip_path = _sip_candidate + else: + click.echo( + f"[prepare-win] WARNING: {_sip_candidate} not found — " + "ip_pool static-IP feature will not work on this template.", + err=True, + ) + static_ip_path = None + + # --- Prompt for passwords --- + if not admin_password: + admin_password = click.prompt( + f"Template VM {admin_username} password", hide_input=True + ) + if not build_password: + build_password = click.prompt(f"New {build_username} password", hide_input=True) + + # --- Backend: start VM and detect IP --- + config = load_config() + 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-win] Starting VM: {vmx_path.name} ...") + backend.start(handle, headless=False) + else: + click.echo(f"[prepare-win] VM already running: {vmx_path.name}") + + if not guest_ip: + click.echo("[prepare-win] Waiting for guest IP (vmrun getGuestIPAddress -wait) ...") + detected = backend.get_ip(handle, timeout=300.0) + if not detected: + raise click.ClickException( + "Could not get guest IP. Ensure VMware Tools is installed." + ) + guest_ip = detected + click.echo(f"[prepare-win] Guest IP: {guest_ip}") + + # --- Wait for WinRM TCP then probe session --- + click.echo(f"[prepare-win] Waiting for TCP/5986 on {guest_ip} ({winrm_timeout}s) ...") + if not _wait_tcp(guest_ip, 5986, float(winrm_timeout)): + raise click.ClickException(f"TCP/5986 not reachable on {guest_ip}.") + + transport = WinRmTransport( + host=guest_ip, + username=admin_username, + password=admin_password, + port=5986, + ssl=True, + cert_validation=False, + auth=auth, + connection_timeout=30, + ) + click.echo("[prepare-win] Probing WinRM session ...") + _wait_winrm(transport, winrm_timeout) + click.echo("[prepare-win] WinRM ready.") + + try: + # Ensure C:\CI exists + transport.run( + "if (-not (Test-Path 'C:\\CI')) " + "{ New-Item -ItemType Directory 'C:\\CI' -Force | Out-Null }" + ) + + # Upload toolchain script + guest_script = f"C:\\CI\\{toolchain_path.name}" + click.echo(f"[prepare-win] Uploading {toolchain_path.name} -> {guest_script} ...") + transport.copy(toolchain_path, guest_script) + click.echo("[prepare-win] Upload complete.") + + # Build args string + e = _ps_escape + args_parts = [ + f"-BuildPassword '{e(build_password)}'", + f"-BuildUsername '{e(build_username)}'", + f"-AdminPassword '{e(admin_password)}'", + ] + if skip_windows_update: + args_parts.append("-SkipWindowsUpdate:$true") + if skip_tier2: + args_parts.append("-SkipTier2:$true") + if skip_cleanup: + args_parts.append("-SkipCleanup:$true") + args_str = " ".join(args_parts) + + # Run toolchain script — loop handles Windows Update reboot (exit 3010) + _MAX_ITER = 10 + exit_code = 0 + for iteration in range(1, _MAX_ITER + 1): + click.echo( + f"[prepare-win] Running {toolchain_path.name} " + f"(iteration {iteration}/{_MAX_ITER}) ..." + ) + invoke = ( + "Set-ExecutionPolicy Bypass -Scope Process -Force\n" + "$ec = 0\n" + "try {\n" + f" & '{e(guest_script)}' {args_str}\n" + " $ec = if ($null -eq $LASTEXITCODE) { 0 } else { $LASTEXITCODE }\n" + "} catch {\n" + " Write-Error $_\n" + " $ec = 1\n" + "}\n" + '"CI_EXITCODE:$ec"\n' + ) + try: + result = transport.run(invoke, check=False) + exit_code = _parse_exit_marker(result.stdout) + if result.stdout.strip(): + click.echo(result.stdout.strip()) + if result.stderr.strip(): + click.echo(result.stderr.strip(), err=True) + except TransportError: + click.echo( + "[prepare-win] Transport error — VM likely rebooted mid-update " + "(treating as 3010)." + ) + exit_code = 3010 + + click.echo(f"[prepare-win] Toolchain exit code: {exit_code}") + + if exit_code == 0: + break + if exit_code != 3010: + raise click.ClickException( + f"Toolchain script failed (exit {exit_code})." + ) + if iteration == _MAX_ITER: + raise click.ClickException( + f"Windows Update did not converge after {_MAX_ITER} iterations." + ) + + click.echo("[prepare-win] Windows Update requires reboot — rebooting ...") + try: + transport.run("Restart-Computer -Force", check=False) + except TransportError: + pass + transport.close() + time.sleep(30) + click.echo(f"[prepare-win] Waiting for WinRM after reboot ({winrm_timeout}s) ...") + _wait_winrm(transport, winrm_timeout) + click.echo("[prepare-win] WinRM ready after reboot.") + + # Install CI-StaticIp scheduled task + if static_ip_path is not None: + click.echo("[prepare-win] Installing CI-StaticIp scheduled task ...") + transport.copy(static_ip_path, "C:\\CI\\ci-static-ip.ps1") + transport.run(r""" +$action = New-ScheduledTaskAction -Execute 'powershell.exe' ` + -Argument '-NoProfile -NonInteractive -ExecutionPolicy Bypass -File C:\CI\ci-static-ip.ps1' +$trigger = New-ScheduledTaskTrigger -AtStartup +$principal = New-ScheduledTaskPrincipal -UserId 'SYSTEM' -LogonType ServiceAccount -RunLevel Highest +Register-ScheduledTask -TaskName 'CI-StaticIp' ` + -Action $action -Trigger $trigger -Principal $principal -Force | Out-Null +""") + click.echo("[prepare-win] CI-StaticIp task registered.") + + # Post-setup validation + click.echo("[prepare-win] Running post-setup validation ...") + esc_u = e(build_username) + val_result = transport.run( + "$env:PATH = [System.Environment]::GetEnvironmentVariable('PATH','Machine')" + " + ';' + [System.Environment]::GetEnvironmentVariable('PATH','User')\n" + "$msbuild = 'C:\\Program Files (x86)\\Microsoft Visual Studio\\2026" + "\\BuildTools\\MSBuild\\Current\\Bin\\MSBuild.exe'\n" + "@{\n" + " WinRMRunning = (Get-Service WinRM -EA SilentlyContinue).Status -eq 'Running'\n" + f" UserExists = [bool](Get-LocalUser -Name '{esc_u}' -EA SilentlyContinue)\n" + f" UserIsAdmin = [bool](net localgroup Administrators 2>&1 | Select-String -SimpleMatch '{esc_u}')\n" + " UACDisabled = (Get-ItemProperty 'HKLM:\\SOFTWARE\\Microsoft\\Windows" + "\\CurrentVersion\\Policies\\System' -Name EnableLUA -EA SilentlyContinue).EnableLUA -eq 0\n" + " FirewallOff = -not (Get-NetFirewallProfile | Where-Object Enabled -eq $true)\n" + " DotNetExists = [bool](Get-Command dotnet -EA SilentlyContinue)\n" + " PythonExists = Test-Path 'C:\\Python\\python.exe' -PathType Leaf\n" + " MSBuildExists = Test-Path $msbuild -PathType Leaf\n" + " CIDirsPresent = (Test-Path 'C:\\CI\\build') -and (Test-Path 'C:\\CI\\output')" + " -and (Test-Path 'C:\\CI\\scripts')\n" + " StaticIpTask = [bool](Get-ScheduledTask -TaskName 'CI-StaticIp' -EA SilentlyContinue)\n" + " StaticIpScript = Test-Path 'C:\\CI\\ci-static-ip.ps1' -PathType Leaf\n" + "} | ConvertTo-Json -Compress\n", + check=False, + ) + if val_result.ok and val_result.stdout.strip(): + try: + state: dict[str, object] = json.loads(val_result.stdout.strip()) + failed = [k for k, v in state.items() if not v] + for k, v in state.items(): + mark = "OK " if v else "FAIL" + click.echo(f" [{mark}] {k}") + if failed: + click.echo( + f"[prepare-win] WARNING: {len(failed)} check(s) failed: " + f"{', '.join(failed)}", + err=True, + ) + else: + click.echo("[prepare-win] All post-setup checks passed.") + except (json.JSONDecodeError, AttributeError): + click.echo(val_result.stdout) + else: + click.echo( + f"[prepare-win] Validation script error:\n{val_result.stderr}", err=True + ) + + # Final reboot to flush installation state + click.echo("[prepare-win] Final reboot to flush installation state ...") + try: + transport.run("Restart-Computer -Force", check=False) + except TransportError: + pass + transport.close() + time.sleep(30) + click.echo("[prepare-win] Waiting for WinRM after final reboot ...") + _wait_winrm(transport, winrm_timeout) + + # Re-affirm autologin + click.echo("[prepare-win] Re-affirming autologin registry settings ...") + transport.run( + "$wl = 'HKLM:\\SOFTWARE\\Microsoft\\Windows NT\\CurrentVersion\\Winlogon'\n" + "Set-ItemProperty $wl AutoAdminLogon '1' -Type String -Force\n" + f"Set-ItemProperty $wl DefaultUserName '{e(admin_username)}' -Type String -Force\n" + "Set-ItemProperty $wl DefaultDomainName '.' -Type String -Force\n" + f"if ('{e(admin_password)}' -ne '') {{\n" + f" Set-ItemProperty $wl DefaultPassword '{e(admin_password)}' -Type String -Force\n" + "}\n" + ) + + # Graceful shutdown + click.echo("[prepare-win] Sending graceful shutdown ...") + try: + transport.run("Stop-Computer -Force", check=False) + except TransportError: + pass + + finally: + transport.close() + + # Wait for VM to power off + click.echo("[prepare-win] 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-win] VM powered off.") + + # Create snapshot + vmrun_exe = backend.vmrun_path + existing = backend.list_snapshots(handle) + if snapshot in existing: + click.echo(f"[prepare-win] Snapshot '{snapshot}' already exists.") + if not click.confirm(f"Delete and recreate '{snapshot}'?", default=False): + click.echo("[prepare-win] Keeping existing snapshot. Provisioning complete.") + return + click.echo(f"[prepare-win] Deleting snapshot '{snapshot}' ...") + subprocess.run( + [vmrun_exe, "-T", "ws", "deleteSnapshot", str(vmx_path), snapshot], + check=True, + ) + + click.echo(f"[prepare-win] Creating snapshot '{snapshot}' ...") + subprocess.run( + [vmrun_exe, "-T", "ws", "snapshot", str(vmx_path), snapshot], + check=True, + ) + click.echo(f"[prepare-win] Snapshot '{snapshot}' created. Provisioning complete.") diff --git a/tests/python/test_commands_job.py b/tests/python/test_commands_job.py index 4a18401..84cdf94 100644 --- a/tests/python/test_commands_job.py +++ b/tests/python/test_commands_job.py @@ -100,6 +100,15 @@ def _patch_common( monkeypatch.setattr(job_module, "load_backend", lambda _cfg: backend) + # Prevent tests from reading the live /var/lib/ci/config.toml which may + # have ip_pool configured — that would try to lock /var/lib/ci/ip-pool.lock + # (requires root) and fail before any backend call. + from ci_orchestrator.config import BackendConfig, Config, Paths + + _test_paths = Paths(root=Path("/tmp/ci-test"), templates=Path("/tmp/ci-test/templates"), build_vms=Path("/tmp/ci-test/build-vms"), artifacts=Path("/tmp/ci-test/artifacts"), keys=Path("/tmp/ci-test/keys")) + _test_cfg = Config(paths=_test_paths, backend=BackendConfig(), ip_pool=None) + monkeypatch.setattr(job_module, "load_config", lambda: _test_cfg) + class _ProbeOk: def __enter__(self) -> _ProbeOk: return self diff --git a/tests/python/test_commands_template.py b/tests/python/test_commands_template.py index 5f69f65..5973308 100644 --- a/tests/python/test_commands_template.py +++ b/tests/python/test_commands_template.py @@ -1,16 +1,21 @@ -"""Tests for ``ci_orchestrator template backup``.""" +"""Tests for ``ci_orchestrator template`` sub-commands.""" from __future__ import annotations +import json import os import time +import types from pathlib import Path +from unittest.mock import MagicMock +import click import pytest from click.testing import CliRunner import ci_orchestrator.commands.template as tmpl_mod from ci_orchestrator.__main__ import cli +from ci_orchestrator.transport.winrm import WinRmResult # ── fixtures ────────────────────────────────────────────────────────────────── @@ -303,3 +308,610 @@ def test_compress_7z_uses_mx1(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) - cmd = calls[0] assert cmd[0] == "7z" assert "-mx=1" in cmd + + +# ═══════════════════════════════════════════════════════════════════════════ +# template prepare-win — helper unit tests +# ═══════════════════════════════════════════════════════════════════════════ + + +def test_ps_escape_no_quotes() -> None: + assert tmpl_mod._ps_escape("hello world") == "hello world" + + +def test_ps_escape_single_quote() -> None: + assert tmpl_mod._ps_escape("it's") == "it''s" + + +def test_ps_escape_multiple_quotes() -> None: + assert tmpl_mod._ps_escape("it's Tom's") == "it''s Tom''s" + + +def test_parse_exit_marker_present() -> None: + assert tmpl_mod._parse_exit_marker("output\nCI_EXITCODE:3010\n") == 3010 + + +def test_parse_exit_marker_absent() -> None: + assert tmpl_mod._parse_exit_marker("no marker here") == 0 + + +def test_parse_exit_marker_last_line_wins() -> None: + assert tmpl_mod._parse_exit_marker("CI_EXITCODE:1\nCI_EXITCODE:0") == 0 + + +def test_parse_exit_marker_malformed() -> None: + assert tmpl_mod._parse_exit_marker("CI_EXITCODE:abc") == 0 + + +def test_wait_tcp_success(monkeypatch: pytest.MonkeyPatch) -> None: + class _FakeConn: + def __enter__(self) -> "_FakeConn": + return self + + def __exit__(self, *_: object) -> None: + pass + + monkeypatch.setattr(tmpl_mod.socket, "create_connection", lambda *a, **kw: _FakeConn()) + assert tmpl_mod._wait_tcp("127.0.0.1", 9999, 5.0) is True + + +def test_wait_tcp_timeout(monkeypatch: pytest.MonkeyPatch) -> None: + def _refuse(*a: object, **kw: object) -> None: + raise OSError("refused") + + monkeypatch.setattr(tmpl_mod.socket, "create_connection", _refuse) + monkeypatch.setattr(tmpl_mod.time, "sleep", lambda _: None) + assert tmpl_mod._wait_tcp("127.0.0.1", 9999, 0.0) is False + + +def test_wait_winrm_success() -> None: + transport = MagicMock() + transport.is_ready.return_value = True + tmpl_mod._wait_winrm(transport, 30) # must not raise + + +def test_wait_winrm_timeout() -> None: + transport = MagicMock() + transport.is_ready.return_value = False + with pytest.raises(click.ClickException, match="not ready"): + tmpl_mod._wait_winrm(transport, 0) + + +# ═══════════════════════════════════════════════════════════════════════════ +# template prepare-win — command tests +# ═══════════════════════════════════════════════════════════════════════════ + + +def _make_config() -> types.SimpleNamespace: + return types.SimpleNamespace(vmrun_path=None, ip_pool=None) + + +def _make_backend(is_running_seq: list[bool]) -> MagicMock: + b: MagicMock = MagicMock() + b.is_running.side_effect = is_running_seq + b.get_ip.return_value = "192.168.1.1" + b.list_snapshots.return_value = [] + b.vmrun_path = "vmrun" + return b + + +def _make_transport() -> MagicMock: + val_json = json.dumps( + { + k: True + for k in [ + "WinRMRunning", "UserExists", "UserIsAdmin", "UACDisabled", + "FirewallOff", "DotNetExists", "PythonExists", "MSBuildExists", + "CIDirsPresent", "StaticIpTask", "StaticIpScript", + ] + } + ) + + def _run(script: str = "", **kwargs: object) -> WinRmResult: + if "ConvertTo-Json" in script: + return WinRmResult(stdout=val_json, stderr="", returncode=0) + return WinRmResult(stdout="CI_EXITCODE:0", stderr="", returncode=0) + + t: MagicMock = MagicMock() + t.run.side_effect = _run + t.is_ready.return_value = True + return t + + +def _setup_prepare_env( + tmp_path: Path, + *, + with_static_ip: bool = True, +) -> Path: + """Create fake VMX + toolchain script layout under tmp_path.""" + vmx = tmp_path / "WinBuild2025.vmx" + vmx.write_text("vmx content") + tmpl_dir = tmp_path / "template" + tmpl_dir.mkdir() + (tmpl_dir / "Install-CIToolchain-WinBuild2025.ps1").write_text("# fake toolchain") + if with_static_ip: + sip_dir = tmpl_dir / "guest-setup" / "windows" + sip_dir.mkdir(parents=True) + (sip_dir / "ci-static-ip.ps1").write_text("# fake static ip") + return vmx + + +def test_prepare_win_vmx_not_found(tmp_path: Path) -> None: + result = CliRunner().invoke( + cli, + ["template", "prepare-win", "--vmx", str(tmp_path / "missing.vmx")], + ) + assert result.exit_code != 0 + assert "VMX not found" in result.output + + +def test_prepare_win_toolchain_not_found( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + vmx = tmp_path / "WinBuild2025.vmx" + vmx.write_text("vmx") + monkeypatch.chdir(tmp_path) + result = CliRunner().invoke( + cli, + ["template", "prepare-win", "--vmx", str(vmx)], + ) + assert result.exit_code != 0 + assert "not found" in result.output + + +def test_prepare_win_explicit_toolchain_not_found(tmp_path: Path) -> None: + vmx = tmp_path / "WinBuild2025.vmx" + vmx.write_text("vmx") + result = CliRunner().invoke( + cli, + [ + "template", "prepare-win", + "--vmx", str(vmx), + "--toolchain-script", str(tmp_path / "nope.ps1"), + ], + ) + assert result.exit_code != 0 + assert "not found" in result.output + + +def _patch_prepare(monkeypatch: pytest.MonkeyPatch, backend: MagicMock, transport: MagicMock) -> None: + monkeypatch.setattr(tmpl_mod, "load_config", lambda: _make_config()) + monkeypatch.setattr(tmpl_mod, "WorkstationVmrunBackend", lambda **_: backend) + monkeypatch.setattr(tmpl_mod, "WinRmTransport", lambda **_: transport) + monkeypatch.setattr(tmpl_mod, "_wait_tcp", lambda *_: True) + monkeypatch.setattr(tmpl_mod, "_wait_winrm", lambda *_: None) + monkeypatch.setattr(tmpl_mod.time, "sleep", lambda _: None) + monkeypatch.setattr(tmpl_mod.subprocess, "run", lambda *a, **_: None) + + +def test_prepare_win_happy_path( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + vmx = _setup_prepare_env(tmp_path) + monkeypatch.chdir(tmp_path) + backend = _make_backend([True, False]) + transport = _make_transport() + _patch_prepare(monkeypatch, backend, transport) + + result = CliRunner().invoke( + cli, + [ + "template", "prepare-win", + "--vmx", str(vmx), + "--ip", "192.168.1.100", + "--admin-password", "adminpass", + "--build-password", "buildpass", + ], + ) + assert result.exit_code == 0, result.output + assert "Provisioning complete" in result.output + assert "All post-setup checks passed" in result.output + + +def test_prepare_win_no_static_ip_warning( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + vmx = _setup_prepare_env(tmp_path, with_static_ip=False) + monkeypatch.chdir(tmp_path) + backend = _make_backend([True, False]) + transport = _make_transport() + _patch_prepare(monkeypatch, backend, transport) + + result = CliRunner().invoke( + cli, + [ + "template", "prepare-win", + "--vmx", str(vmx), + "--ip", "192.168.1.100", + "--admin-password", "adminpass", + "--build-password", "buildpass", + ], + ) + assert result.exit_code == 0, result.output + assert "WARNING" in result.output + + +def test_prepare_win_vm_started_when_not_running( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + vmx = _setup_prepare_env(tmp_path) + monkeypatch.chdir(tmp_path) + backend = _make_backend([False, False]) # not running → start; then powered off + transport = _make_transport() + _patch_prepare(monkeypatch, backend, transport) + + result = CliRunner().invoke( + cli, + [ + "template", "prepare-win", + "--vmx", str(vmx), + "--ip", "192.168.1.100", + "--admin-password", "adminpass", + "--build-password", "buildpass", + ], + ) + assert result.exit_code == 0, result.output + backend.start.assert_called_once() + + +def test_prepare_win_tcp_timeout( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + vmx = _setup_prepare_env(tmp_path) + monkeypatch.chdir(tmp_path) + backend = _make_backend([True]) + _patch_prepare(monkeypatch, backend, MagicMock()) + monkeypatch.setattr(tmpl_mod, "_wait_tcp", lambda *_: False) + + result = CliRunner().invoke( + cli, + [ + "template", "prepare-win", + "--vmx", str(vmx), + "--ip", "192.168.1.100", + "--admin-password", "adminpass", + "--build-password", "buildpass", + ], + ) + assert result.exit_code != 0 + assert "not reachable" in result.output + + +def test_prepare_win_toolchain_fails( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + vmx = _setup_prepare_env(tmp_path) + monkeypatch.chdir(tmp_path) + backend = _make_backend([True]) + transport: MagicMock = MagicMock() + transport.is_ready.return_value = True + transport.run.return_value = WinRmResult( + stdout="CI_EXITCODE:1", stderr="err", returncode=0 + ) + _patch_prepare(monkeypatch, backend, transport) + + result = CliRunner().invoke( + cli, + [ + "template", "prepare-win", + "--vmx", str(vmx), + "--ip", "192.168.1.100", + "--admin-password", "adminpass", + "--build-password", "buildpass", + ], + ) + assert result.exit_code != 0 + assert "failed" in result.output + + +def test_prepare_win_snapshot_exists_skip( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + vmx = _setup_prepare_env(tmp_path) + monkeypatch.chdir(tmp_path) + backend = _make_backend([True, False]) + backend.list_snapshots.return_value = ["BaseClean"] + transport = _make_transport() + _patch_prepare(monkeypatch, backend, transport) + + # User declines deletion with "N" + result = CliRunner().invoke( + cli, + [ + "template", "prepare-win", + "--vmx", str(vmx), + "--ip", "192.168.1.100", + "--admin-password", "adminpass", + "--build-password", "buildpass", + ], + input="N\n", + ) + assert result.exit_code == 0, result.output + assert "Keeping existing snapshot" in result.output + + +def test_prepare_win_snapshot_exists_recreate( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + vmx = _setup_prepare_env(tmp_path) + monkeypatch.chdir(tmp_path) + backend = _make_backend([True, False]) + backend.list_snapshots.return_value = ["BaseClean"] + transport = _make_transport() + sp_calls: list[list[str]] = [] + monkeypatch.setattr(tmpl_mod, "load_config", lambda: _make_config()) + monkeypatch.setattr(tmpl_mod, "WorkstationVmrunBackend", lambda **_: backend) + monkeypatch.setattr(tmpl_mod, "WinRmTransport", lambda **_: transport) + monkeypatch.setattr(tmpl_mod, "_wait_tcp", lambda *_: True) + monkeypatch.setattr(tmpl_mod, "_wait_winrm", lambda *_: None) + monkeypatch.setattr(tmpl_mod.time, "sleep", lambda _: None) + + def _sp_run(cmd: list[str], **_: object) -> None: + sp_calls.append(cmd) + + monkeypatch.setattr(tmpl_mod.subprocess, "run", _sp_run) + + result = CliRunner().invoke( + cli, + [ + "template", "prepare-win", + "--vmx", str(vmx), + "--ip", "192.168.1.100", + "--admin-password", "adminpass", + "--build-password", "buildpass", + ], + input="Y\n", + ) + assert result.exit_code == 0, result.output + assert any("deleteSnapshot" in " ".join(c) for c in sp_calls) + assert any("snapshot" in " ".join(c) and "deleteSnapshot" not in " ".join(c) for c in sp_calls) + + +def test_prepare_win_skip_flags( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """--skip-windows-update/tier2/cleanup flags reach the invoke script.""" + vmx = _setup_prepare_env(tmp_path, with_static_ip=False) + monkeypatch.chdir(tmp_path) + backend = _make_backend([True, False]) + + captured_scripts: list[str] = [] + + def _run(script: str = "", **kwargs: object) -> WinRmResult: + captured_scripts.append(script) + if "ConvertTo-Json" in script: + return WinRmResult(stdout="{}", stderr="", returncode=0) + return WinRmResult(stdout="CI_EXITCODE:0", stderr="", returncode=0) + + transport: MagicMock = MagicMock() + transport.run.side_effect = _run + transport.is_ready.return_value = True + + _patch_prepare(monkeypatch, backend, transport) + + CliRunner().invoke( + cli, + [ + "template", "prepare-win", + "--vmx", str(vmx), + "--ip", "192.168.1.100", + "--admin-password", "adminpass", + "--build-password", "buildpass", + "--skip-windows-update", + "--skip-tier2", + "--skip-cleanup", + ], + ) + invoke_script = next( + (s for s in captured_scripts if "SkipWindowsUpdate" in s or "CI_EXITCODE" in s), + "", + ) + assert "SkipWindowsUpdate" in invoke_script + assert "SkipTier2" in invoke_script + assert "SkipCleanup" in invoke_script + + +def test_prepare_win_auto_detect_ip( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """No --ip supplied — IP auto-detected via backend.get_ip().""" + vmx = _setup_prepare_env(tmp_path, with_static_ip=False) + monkeypatch.chdir(tmp_path) + backend = _make_backend([True, False]) + backend.get_ip.return_value = "10.0.0.5" + transport = _make_transport() + _patch_prepare(monkeypatch, backend, transport) + + result = CliRunner().invoke( + cli, + [ + "template", "prepare-win", + "--vmx", str(vmx), + "--admin-password", "adminpass", + "--build-password", "buildpass", + ], + ) + assert result.exit_code == 0, result.output + assert "10.0.0.5" in result.output + + +def test_prepare_win_auto_detect_ip_fails( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """backend.get_ip() returns None → ClickException.""" + vmx = _setup_prepare_env(tmp_path, with_static_ip=False) + monkeypatch.chdir(tmp_path) + backend = _make_backend([True]) + backend.get_ip.return_value = None + _patch_prepare(monkeypatch, backend, MagicMock()) + + result = CliRunner().invoke( + cli, + [ + "template", "prepare-win", + "--vmx", str(vmx), + "--admin-password", "adminpass", + "--build-password", "buildpass", + ], + ) + assert result.exit_code != 0 + assert "guest IP" in result.output + + +def test_prepare_win_3010_reboot_then_success( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Toolchain returns 3010 once, then 0 — reboot loop executes.""" + vmx = _setup_prepare_env(tmp_path, with_static_ip=False) + monkeypatch.chdir(tmp_path) + backend = _make_backend([True, False]) + + call_count = [0] + + def _run(script: str = "", **kwargs: object) -> WinRmResult: + call_count[0] += 1 + if "ConvertTo-Json" in script: + return WinRmResult(stdout="{}", stderr="", returncode=0) + # First invoke → 3010; subsequent → 0 + code = 3010 if call_count[0] <= 2 else 0 + return WinRmResult(stdout=f"CI_EXITCODE:{code}", stderr="", returncode=0) + + transport: MagicMock = MagicMock() + transport.run.side_effect = _run + transport.is_ready.return_value = True + + _patch_prepare(monkeypatch, backend, transport) + + result = CliRunner().invoke( + cli, + [ + "template", "prepare-win", + "--vmx", str(vmx), + "--ip", "192.168.1.100", + "--admin-password", "adminpass", + "--build-password", "buildpass", + ], + ) + assert result.exit_code == 0, result.output + assert "reboot" in result.output.lower() + + +def test_prepare_win_transport_error_treated_as_3010( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """TransportError during toolchain run → treated as 3010, reboot, then success.""" + from ci_orchestrator.transport.errors import TransportConnectError + + vmx = _setup_prepare_env(tmp_path, with_static_ip=False) + monkeypatch.chdir(tmp_path) + backend = _make_backend([True, False]) + + raised = [False] + + def _run(script: str = "", **kwargs: object) -> WinRmResult: + if not raised[0] and "CI_EXITCODE" in script: + raised[0] = True + raise TransportConnectError("VM rebooted") + if "ConvertTo-Json" in script: + return WinRmResult(stdout="{}", stderr="", returncode=0) + return WinRmResult(stdout="CI_EXITCODE:0", stderr="", returncode=0) + + transport: MagicMock = MagicMock() + transport.run.side_effect = _run + transport.is_ready.return_value = True + + _patch_prepare(monkeypatch, backend, transport) + + result = CliRunner().invoke( + cli, + [ + "template", "prepare-win", + "--vmx", str(vmx), + "--ip", "192.168.1.100", + "--admin-password", "adminpass", + "--build-password", "buildpass", + ], + ) + assert result.exit_code == 0, result.output + assert "Transport error" in result.output + + +def test_prepare_win_validation_with_failures( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Validation JSON contains False values → WARNING printed but exit 0.""" + vmx = _setup_prepare_env(tmp_path, with_static_ip=False) + monkeypatch.chdir(tmp_path) + backend = _make_backend([True, False]) + + partial_json = json.dumps( + {k: (k != "MSBuildExists") for k in [ + "WinRMRunning", "UserExists", "UserIsAdmin", "UACDisabled", + "FirewallOff", "DotNetExists", "PythonExists", "MSBuildExists", + "CIDirsPresent", "StaticIpTask", "StaticIpScript", + ]} + ) + + def _run(script: str = "", **kwargs: object) -> WinRmResult: + if "ConvertTo-Json" in script: + return WinRmResult(stdout=partial_json, stderr="", returncode=0) + return WinRmResult(stdout="CI_EXITCODE:0", stderr="", returncode=0) + + transport: MagicMock = MagicMock() + transport.run.side_effect = _run + transport.is_ready.return_value = True + + _patch_prepare(monkeypatch, backend, transport) + + result = CliRunner().invoke( + cli, + [ + "template", "prepare-win", + "--vmx", str(vmx), + "--ip", "192.168.1.100", + "--admin-password", "adminpass", + "--build-password", "buildpass", + ], + ) + assert result.exit_code == 0, result.output + assert "WARNING" in result.output + assert "MSBuildExists" in result.output + + +def test_prepare_win_vm_shutdown_timeout( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """VM never powers off within 120 s → ClickException.""" + vmx = _setup_prepare_env(tmp_path, with_static_ip=False) + monkeypatch.chdir(tmp_path) + + backend = _make_backend([True]) # always "running" + backend.is_running.side_effect = None # override side_effect + backend.is_running.return_value = True # always reports as running + + transport = _make_transport() + _patch_prepare(monkeypatch, backend, transport) + + # Patch monotonic so the 120 s deadline expires immediately + _orig = tmpl_mod.time.monotonic + call_n = [0] + + def _fast_mono() -> float: + call_n[0] += 1 + # After the 5th call (inside shutdown-wait loop) return a huge number + return _orig() if call_n[0] < 20 else _orig() + 200 + + monkeypatch.setattr(tmpl_mod.time, "monotonic", _fast_mono) + + result = CliRunner().invoke( + cli, + [ + "template", "prepare-win", + "--vmx", str(vmx), + "--ip", "192.168.1.100", + "--admin-password", "adminpass", + "--build-password", "buildpass", + ], + ) + assert result.exit_code != 0 + assert "power off" in result.output