local-ci-cd-system/src/ci_orchestrator/backends/workstation.py

"""VMware Workstation backend (``vmrun -T ws`` wrapper).

Mirrors the behaviour of ``scripts/_Common.psm1`` :func:`Invoke-Vmrun` plus
the higher-level helpers used by ``New-BuildVM.ps1`` and
``Wait-VMReady.ps1``. Only ``subprocess`` is used — no PowerShell.

Cross-platform: ``vmrun`` lives at different paths on Windows vs Linux;
when ``vmrun_path`` is not provided, :func:`shutil.which` is consulted and
finally a Windows default is tried.
"""

from __future__ import annotations

import shutil
import subprocess
from pathlib import Path

from ci_orchestrator.backends.errors import BackendNotAvailable, BackendOperationFailed
from ci_orchestrator.backends.protocol import VmHandle

_DEFAULT_WIN_VMRUN = r"C:\Program Files (x86)\VMware\VMware Workstation\vmrun.exe"


def _resolve_vmrun(explicit: str | None) -> str:
    """Pick a usable ``vmrun`` binary path."""
    if explicit:
        if not Path(explicit).is_file():
            raise BackendNotAvailable(f"vmrun not found at: {explicit}")
        return explicit
    found = shutil.which("vmrun")
    if found:
        return found
    if Path(_DEFAULT_WIN_VMRUN).is_file():
        return _DEFAULT_WIN_VMRUN
    raise BackendNotAvailable(
        "vmrun is not available. Install VMware Workstation Pro or set vmrun_path."
    )


class WorkstationVmrunBackend:
    """``vmrun -T ws`` backend for VMware Workstation Pro."""

    def __init__(self, vmrun_path: str | None = None) -> None:
        self._vmrun_path = _resolve_vmrun(vmrun_path)

    # ------------------------------------------------------------------ utils
    @property
    def vmrun_path(self) -> str:
        return self._vmrun_path

    def _run(
        self,
        operation: str,
        *args: str,
        check: bool = True,
        timeout: float | None = None,
    ) -> subprocess.CompletedProcess[str]:
        """Run ``vmrun -T ws <operation> <args...>`` and capture output."""
        cmd = [self._vmrun_path, "-T", "ws", operation, *args]
        result = subprocess.run(
            cmd,
            check=False,
            capture_output=True,
            text=True,
            encoding="utf-8",
            errors="replace",
            timeout=timeout,
        )
        if check and result.returncode != 0:
            raise BackendOperationFailed(
                operation=operation,
                returncode=result.returncode,
                output=(result.stdout or "") + (result.stderr or ""),
            )
        return result

    # ------------------------------------------------------------- VmBackend
    def clone_linked(
        self,
        template: str,
        snapshot: str,
        name: str,
        destination: str | None = None,
    ) -> VmHandle:
        if destination is None:
            destination = str(Path(template).parent.parent / name / f"{name}.vmx")
        # vmrun clone <srcVMX> <destVMX> linked -snapshot=<name> -cloneName=<name>
        self._run(
            "clone",
            template,
            destination,
            "linked",
            f"-snapshot={snapshot}",
            f"-cloneName={name}",
        )
        return VmHandle(identifier=destination, name=name)

    def start(self, handle: VmHandle, headless: bool = True) -> None:
        mode = "nogui" if headless else "gui"
        self._run("start", handle.identifier, mode)

    def stop(self, handle: VmHandle, hard: bool = False) -> None:
        mode = "hard" if hard else "soft"
        # stop may legitimately fail if the VM is already off; treat that as ok
        result = self._run("stop", handle.identifier, mode, check=False)
        if result.returncode != 0:
            stderr = (result.stderr or "").lower()
            if "not powered on" in stderr or "is not running" in stderr:
                return
            raise BackendOperationFailed(
                "stop",
                result.returncode,
                (result.stdout or "") + (result.stderr or ""),
            )

    def delete(self, handle: VmHandle) -> None:
        self._run("deleteVM", handle.identifier)

    def get_ip(self, handle: VmHandle, timeout: float = 0.0) -> str | None:
        # ``getGuestIPAddress`` requires VMware Tools; tolerate failure.
        args: list[str] = [handle.identifier]
        if timeout > 0:
            args.extend(["-wait"])
        try:
            result = self._run("getGuestIPAddress", *args, check=False, timeout=timeout or None)
        except subprocess.TimeoutExpired:
            return None
        if result.returncode != 0:
            return None
        ip = (result.stdout or "").strip()
        if not ip or ip.lower().startswith("error:"):
            return None
        return ip

    def list_snapshots(self, handle: VmHandle) -> list[str]:
        result = self._run("listSnapshots", handle.identifier)
        lines = [line.strip() for line in (result.stdout or "").splitlines() if line.strip()]
        # First line is "Total snapshots: N"
        if lines and lines[0].lower().startswith("total snapshots"):
            lines = lines[1:]
        return lines

    def is_running(self, handle: VmHandle) -> bool:
        """``vmrun list`` lookup — does NOT call ``getGuestIPAddress``.

        See ``AGENTS.md`` error #10: ``getGuestIPAddress`` blocks for 30-60s
        without VMware Tools. ``vmrun list`` returns immediately.
        """
        result = self._run("list", check=False)
        if result.returncode != 0:
            return False
        target = Path(handle.identifier).resolve()
        for line in (result.stdout or "").splitlines():
            line = line.strip()
            if not line or line.lower().startswith("total running"):
                continue
            try:
                if Path(line).resolve() == target:
                    return True
            except OSError:
                continue
        return False