diff --git a/.gitignore b/.gitignore index c7b87ea..a543049 100644 --- a/.gitignore +++ b/.gitignore @@ -46,3 +46,18 @@ $RECYCLE.BIN/ # Copilot Orchestrator temporary files .orchestrator/ .worktrees/ + +# Python +__pycache__/ +*.py[cod] +*.egg-info/ +build/ +dist/ +.venv/ +venv/ +.coverage +.coverage.* +htmlcov/ +.pytest_cache/ +.mypy_cache/ +.ruff_cache/ diff --git a/README.md b/README.md index ca95e2a..cef37af 100644 --- a/README.md +++ b/README.md @@ -146,6 +146,40 @@ Variante **Windows Server 2022**: `*WinBuild2022.ps1` + VMX `F:\CI\Templates\Win --- +## Python orchestrator (Phase A — in progress) + +A new Python 3.11+ rewrite of the orchestrator lives under `src/ci_orchestrator/`. +Phase A (see [implementation-plan-A-B](plans/implementation-plan-A-B.md)) is on +branch `feature/python-rewrite-phase-a`. + +Quick setup on the Windows host: + +```powershell +# Create venv (one-time) +python -m venv F:\CI\python\venv +F:\CI\python\venv\Scripts\python.exe -m pip install --upgrade pip +F:\CI\python\venv\Scripts\python.exe -m pip install -e ".[dev]" + +# Sanity checks +F:\CI\python\venv\Scripts\python.exe -m ci_orchestrator --help +F:\CI\python\venv\Scripts\python.exe -m pytest tests\python -q +F:\CI\python\venv\Scripts\python.exe -m ruff check src tests\python +F:\CI\python\venv\Scripts\python.exe -m mypy --strict src +``` + +Configuration: copy `config.example.toml` to `F:\CI\config.toml` (or set +`$env:CI_CONFIG`). Environment variables (`CI_ROOT`, `CI_TEMPLATES`, …) override +the file. + +In Phase A1 only the `wait-ready` PoC is wired: + +```powershell +F:\CI\python\venv\Scripts\python.exe -m ci_orchestrator wait-ready ` + --vmx F:\CI\BuildVMs\smoke\smoke.vmx --guest-os windows --timeout 300 +``` + +--- + ## Setup rapido ### 1a. Template VM Windows diff --git a/config.example.toml b/config.example.toml new file mode 100644 index 0000000..51a1f6b --- /dev/null +++ b/config.example.toml @@ -0,0 +1,30 @@ +# ci-orchestrator config example. +# +# Copy to $CI_ROOT/config.toml or set $CI_CONFIG to the file path. +# Environment variables (CI_ROOT, CI_TEMPLATES, CI_BUILD_VMS, CI_ARTIFACTS, +# CI_KEYS, CI_VMRUN_PATH, CI_SSH_KEY_PATH, CI_GUEST_CRED_TARGET) override +# anything set here. + +# --------------------------------------------------------------- Windows host +[paths] +root = "F:/CI" +templates = "F:/CI/Templates" +build_vms = "F:/CI/BuildVMs" +artifacts = "F:/CI/Artifacts" +keys = "F:/CI/keys" + +# Phase B (Linux host) — uncomment after migration. +# [paths] +# root = "/var/lib/ci" +# templates = "/var/lib/ci/templates" +# build_vms = "/var/lib/ci/build-vms" +# artifacts = "/var/lib/ci/artifacts" +# keys = "/etc/ci/keys" + +# Phase C hook: backend selector. Only "workstation" is implemented in Phase A. +[backend] +type = "workstation" + +vmrun_path = "C:/Program Files (x86)/VMware/VMware Workstation/vmrun.exe" +ssh_key_path = "F:/CI/keys/ci_linux" +guest_cred_target = "BuildVMGuest" diff --git a/gitea/workflows/lint.yml b/gitea/workflows/lint.yml index 204f4fb..ff95bbc 100644 --- a/gitea/workflows/lint.yml +++ b/gitea/workflows/lint.yml @@ -5,20 +5,26 @@ # Failures block the PR. Fix warnings with: # Invoke-ScriptAnalyzer -Path scripts\ -Recurse -Fix -name: Lint (PSScriptAnalyzer) +name: Lint on: push: paths: - '**.ps1' - '**.psm1' + - '**.py' + - 'pyproject.toml' + - 'gitea/workflows/lint.yml' pull_request: paths: - '**.ps1' - '**.psm1' + - '**.py' + - 'pyproject.toml' + - 'gitea/workflows/lint.yml' jobs: - lint: + pssa: runs-on: windows-build timeout-minutes: 10 @@ -53,3 +59,55 @@ jobs: else { Write-Host "PSScriptAnalyzer: no issues found." } + + python: + runs-on: windows-build + timeout-minutes: 15 + + env: + PYTHONIOENCODING: utf-8 + VENV_DIR: F:\CI\python\venv + + steps: + - name: Checkout + uses: actions/checkout@v4 + with: + fetch-depth: 1 + + - name: Setup venv + install package + shell: powershell + run: | + $ErrorActionPreference = 'Stop' + + $py = (Get-Command python -ErrorAction Stop).Source + if (-not (Test-Path "$env:VENV_DIR\Scripts\python.exe")) { + Write-Host "Creating venv at $env:VENV_DIR" + & $py -m venv $env:VENV_DIR + if ($LASTEXITCODE -ne 0) { throw "venv creation failed" } + } + $venvPy = Join-Path $env:VENV_DIR 'Scripts\python.exe' + & $venvPy -m pip install --upgrade pip + if ($LASTEXITCODE -ne 0) { throw "pip upgrade failed" } + & $venvPy -m pip install -e ".[dev]" + if ($LASTEXITCODE -ne 0) { throw "pip install failed" } + + - name: ruff + shell: powershell + run: | + $venvPy = Join-Path $env:VENV_DIR 'Scripts\python.exe' + & $venvPy -m ruff check src tests/python + if ($LASTEXITCODE -ne 0) { throw "ruff failed" } + + - name: mypy --strict + shell: powershell + run: | + $venvPy = Join-Path $env:VENV_DIR 'Scripts\python.exe' + & $venvPy -m mypy --strict src + if ($LASTEXITCODE -ne 0) { throw "mypy failed" } + + - name: pytest (coverage >= 70%) + shell: powershell + run: | + $venvPy = Join-Path $env:VENV_DIR 'Scripts\python.exe' + & $venvPy -m pytest tests/python --cov=ci_orchestrator --cov-fail-under=70 + if ($LASTEXITCODE -ne 0) { throw "pytest failed" } diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..9ebbd08 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,89 @@ +[build-system] +requires = ["hatchling>=1.18"] +build-backend = "hatchling.build" + +[project] +name = "ci-orchestrator" +version = "0.1.0" +description = "Local CI/CD orchestrator (Phase A: Python rewrite of PowerShell scripts)" +readme = "README.md" +requires-python = ">=3.11" +license = { text = "Proprietary" } +authors = [{ name = "Local CI/CD System" }] +dependencies = [ + "click>=8.1", + "pypsrp>=0.8", + "paramiko>=3.4", + "keyring>=24.0", + "tomli>=2.0; python_version < '3.11'", +] + +[project.optional-dependencies] +dev = [ + "pytest>=8.0", + "pytest-cov>=5.0", + "pytest-mock>=3.12", + "ruff>=0.5", + "mypy>=1.10", + "types-paramiko", +] + +[project.scripts] +ci-orchestrator = "ci_orchestrator.__main__:cli" + +[tool.hatch.build.targets.wheel] +packages = ["src/ci_orchestrator"] + +[tool.ruff] +line-length = 100 +target-version = "py311" +src = ["src", "tests"] + +[tool.ruff.lint] +select = [ + "E", # pycodestyle errors + "W", # pycodestyle warnings + "F", # pyflakes + "I", # isort + "B", # flake8-bugbear + "UP", # pyupgrade + "SIM", # flake8-simplify + "RUF", # ruff-specific +] +ignore = [ + "E501", # line length handled by formatter +] + +[tool.ruff.lint.per-file-ignores] +"tests/**" = ["B011", "SIM117"] + +[tool.mypy] +python_version = "3.11" +strict = true +files = ["src"] +warn_unused_ignores = true +warn_redundant_casts = true + +[[tool.mypy.overrides]] +module = ["pypsrp.*", "keyring.*"] +ignore_missing_imports = true + +[tool.pytest.ini_options] +minversion = "8.0" +addopts = "-q --strict-markers" +testpaths = ["tests"] +pythonpath = ["src"] +markers = [ + "integration: requires real VM/network (skipped in unit-test runs)", +] + +[tool.coverage.run] +source = ["ci_orchestrator"] +branch = true + +[tool.coverage.report] +exclude_lines = [ + "pragma: no cover", + "if TYPE_CHECKING:", + "raise NotImplementedError", +] diff --git a/src/ci_orchestrator/__init__.py b/src/ci_orchestrator/__init__.py new file mode 100644 index 0000000..84646d9 --- /dev/null +++ b/src/ci_orchestrator/__init__.py @@ -0,0 +1,9 @@ +"""Local CI/CD orchestrator package. + +Phase A: Python rewrite of the PowerShell-based orchestrator. +See plans/implementation-plan-A-B.md for the full scope. +""" + +__version__ = "0.1.0" + +__all__ = ["__version__"] diff --git a/src/ci_orchestrator/__main__.py b/src/ci_orchestrator/__main__.py new file mode 100644 index 0000000..864da0b --- /dev/null +++ b/src/ci_orchestrator/__main__.py @@ -0,0 +1,123 @@ +"""CLI entry point. + +Phase A1 ships the ``wait-ready`` PoC sub-command. Subsequent steps add +``vm``, ``build``, ``artifacts``, ``monitor``, ``report``, and ``job``. +""" + +from __future__ import annotations + +import sys +import time + +import click + +from ci_orchestrator import __version__ +from ci_orchestrator.backends import load_backend +from ci_orchestrator.backends.errors import BackendError +from ci_orchestrator.backends.protocol import VmHandle +from ci_orchestrator.config import load_config +from ci_orchestrator.credentials import KeyringCredentialStore +from ci_orchestrator.transport.errors import TransportError +from ci_orchestrator.transport.ssh import SshTransport +from ci_orchestrator.transport.winrm import WinRmTransport + + +@click.group() +@click.version_option(__version__, prog_name="ci-orchestrator") +def cli() -> None: + """Local CI/CD orchestrator (Phase A).""" + + +@cli.command("wait-ready") +@click.option("--vmx", required=True, help="Path to the guest VMX.") +@click.option( + "--guest-os", + type=click.Choice(["windows", "linux"]), + default="windows", + show_default=True, +) +@click.option("--timeout", type=float, default=300.0, show_default=True) +@click.option( + "--poll-interval", type=float, default=5.0, show_default=True, help="Seconds between probes." +) +@click.option( + "--credential-target", + default=None, + help="Override credential target name (defaults to config.guest_cred_target).", +) +@click.option("--ssh-user", default="ci_build", show_default=True) +def wait_ready( + vmx: str, + guest_os: str, + timeout: float, + poll_interval: float, + credential_target: str | None, + ssh_user: str, +) -> None: + """Poll a guest until WinRM (Windows) or SSH (Linux) is responsive.""" + config = load_config() + backend = load_backend(config) + handle = VmHandle(identifier=vmx) + + deadline = time.monotonic() + timeout + + # Step 1: wait until backend reports the VM as running. + click.echo(f"[wait-ready] waiting for backend to report VM running: {vmx}") + while time.monotonic() < deadline: + try: + if backend.is_running(handle): + break + except BackendError as exc: + click.echo(f"[wait-ready] backend probe error: {exc}", err=True) + time.sleep(poll_interval) + else: + click.echo("[wait-ready] timeout waiting for VM to be running.", err=True) + sys.exit(2) + + # Step 2: wait for an IP. + ip: str | None = None + while time.monotonic() < deadline: + try: + ip = backend.get_ip(handle) + except BackendError: + ip = None + if ip: + break + time.sleep(poll_interval) + if not ip: + click.echo("[wait-ready] timeout waiting for guest IP.", err=True) + sys.exit(3) + click.echo(f"[wait-ready] guest IP: {ip}") + + # Step 3: probe the transport layer. + if guest_os == "windows": + store = KeyringCredentialStore() + target = credential_target or config.guest_cred_target + cred = store.get(target) + while time.monotonic() < deadline: + try: + with WinRmTransport(ip, cred.username, cred.password) as t: + if t.is_ready(): + click.echo("[wait-ready] WinRM ready.") + return + except TransportError as exc: + click.echo(f"[wait-ready] WinRM probe failed: {exc}", err=True) + time.sleep(poll_interval) + else: + key_path = config.ssh_key_path + while time.monotonic() < deadline: + try: + with SshTransport(ip, username=ssh_user, key_path=key_path) as t: + if t.is_ready(): + click.echo("[wait-ready] SSH ready.") + return + except TransportError as exc: + click.echo(f"[wait-ready] SSH probe failed: {exc}", err=True) + time.sleep(poll_interval) + + click.echo("[wait-ready] timeout waiting for transport to be ready.", err=True) + sys.exit(4) + + +if __name__ == "__main__": # pragma: no cover + cli() diff --git a/src/ci_orchestrator/backends/__init__.py b/src/ci_orchestrator/backends/__init__.py new file mode 100644 index 0000000..761a213 --- /dev/null +++ b/src/ci_orchestrator/backends/__init__.py @@ -0,0 +1,30 @@ +"""VM backend implementations. + +The :class:`~ci_orchestrator.backends.protocol.VmBackend` Protocol defines +the neutral interface every backend must satisfy. Phase A ships only the +:class:`~ci_orchestrator.backends.workstation.WorkstationVmrunBackend`; +Phase C will add an ESXi backend behind the same interface. +""" + +from __future__ import annotations + +from ci_orchestrator.backends.protocol import VmBackend, VmHandle, VmState +from ci_orchestrator.backends.workstation import WorkstationVmrunBackend + +__all__ = ["VmBackend", "VmHandle", "VmState", "WorkstationVmrunBackend", "load_backend"] + + +def load_backend(config: object) -> VmBackend: + """Factory: pick a backend implementation from ``config.backend.type``. + + Phase C hook: when the ESXi backend lands, this function will dispatch + on ``config.backend.type``. For now only ``workstation`` is supported. + """ + backend_type = getattr(getattr(config, "backend", None), "type", "workstation") + if backend_type != "workstation": + raise NotImplementedError( + f"Backend type '{backend_type}' is not implemented. " + "Only 'workstation' is available in Phase A." + ) + vmrun_path = getattr(config, "vmrun_path", None) + return WorkstationVmrunBackend(vmrun_path=vmrun_path) diff --git a/src/ci_orchestrator/backends/errors.py b/src/ci_orchestrator/backends/errors.py new file mode 100644 index 0000000..7b6a161 --- /dev/null +++ b/src/ci_orchestrator/backends/errors.py @@ -0,0 +1,24 @@ +"""Backend error hierarchy.""" + +from __future__ import annotations + + +class BackendError(RuntimeError): + """Base error for VM backend failures.""" + + +class BackendNotAvailable(BackendError): + """Backend tooling (e.g. ``vmrun``) is missing or unusable.""" + + +class BackendOperationFailed(BackendError): + """A backend operation (clone, start, ...) returned a non-zero status.""" + + def __init__(self, operation: str, returncode: int, output: str) -> None: + super().__init__( + f"Backend operation '{operation}' failed " + f"(exit={returncode}): {output.strip() or ''}" + ) + self.operation = operation + self.returncode = returncode + self.output = output diff --git a/src/ci_orchestrator/backends/protocol.py b/src/ci_orchestrator/backends/protocol.py new file mode 100644 index 0000000..19af3b9 --- /dev/null +++ b/src/ci_orchestrator/backends/protocol.py @@ -0,0 +1,81 @@ +"""Neutral VM backend Protocol. + +Phase C hook: keep names hypervisor-agnostic. ``clone_linked``, ``start``, +``stop``, ``delete``, ``get_ip``, ``list_snapshots`` map cleanly onto both +``vmrun`` and pyVmomi (vSphere). Identifiers are opaque strings (a VMX path +on Workstation, a managed object reference on ESXi). +""" + +from __future__ import annotations + +from dataclasses import dataclass +from enum import StrEnum +from typing import Protocol + + +class VmState(StrEnum): + """Coarse VM power state, normalised across backends.""" + + POWERED_OFF = "powered_off" + POWERED_ON = "powered_on" + SUSPENDED = "suspended" + UNKNOWN = "unknown" + + +@dataclass(frozen=True) +class VmHandle: + """Opaque reference to a VM managed by a backend. + + ``identifier`` is backend-specific (a VMX path for Workstation, a + managed object reference for vSphere). Callers MUST treat it as opaque. + """ + + identifier: str + name: str | None = None + + +class VmBackend(Protocol): + """Hypervisor-agnostic VM operations. + + All methods raise :class:`~ci_orchestrator.backends.errors.BackendError` + on failure. Implementations are free to add backend-specific kwargs but + MUST honour the documented signature for the listed parameters. + """ + + def clone_linked( + self, + template: str, + snapshot: str, + name: str, + destination: str | None = None, + ) -> VmHandle: + """Create a linked clone of ``template`` at snapshot ``snapshot``.""" + ... + + def start(self, handle: VmHandle, headless: bool = True) -> None: + """Power on the VM.""" + ... + + def stop(self, handle: VmHandle, hard: bool = False) -> None: + """Power off the VM. ``hard=True`` requests a forced power-off.""" + ... + + def delete(self, handle: VmHandle) -> None: + """Remove the VM from inventory and delete its files.""" + ... + + def get_ip(self, handle: VmHandle, timeout: float = 0.0) -> str | None: + """Return the guest's primary IP, or ``None`` if not yet known.""" + ... + + def list_snapshots(self, handle: VmHandle) -> list[str]: + """Return snapshot names defined on ``handle``.""" + ... + + def is_running(self, handle: VmHandle) -> bool: + """Return ``True`` if the VM appears in the running-VM list. + + Implementations MUST NOT rely on guest tools (e.g. ``vmrun + getGuestIPAddress``) for this check — see ``AGENTS.md`` error #10. + """ + ... diff --git a/src/ci_orchestrator/backends/workstation.py b/src/ci_orchestrator/backends/workstation.py new file mode 100644 index 0000000..652a846 --- /dev/null +++ b/src/ci_orchestrator/backends/workstation.py @@ -0,0 +1,157 @@ +"""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 `` 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 linked -snapshot= -cloneName= + 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"]) + result = self._run("getGuestIPAddress", *args, check=False, timeout=timeout or None) + if result.returncode != 0: + return None + ip = (result.stdout or "").strip() + return ip or None + + 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 diff --git a/src/ci_orchestrator/config.py b/src/ci_orchestrator/config.py new file mode 100644 index 0000000..df3744b --- /dev/null +++ b/src/ci_orchestrator/config.py @@ -0,0 +1,163 @@ +"""Configuration loading for the CI orchestrator. + +Resolution order (later wins): + 1. OS-aware defaults (Windows: F:\\CI\\..., Linux: /var/lib/ci/...). + 2. Optional ``config.toml`` file (path via ``CI_CONFIG`` env var or + ``$CI_ROOT/config.toml`` if present). + 3. Environment variables (``CI_ROOT``, ``CI_TEMPLATES``, ``CI_BUILD_VMS``, + ``CI_ARTIFACTS``, ``CI_KEYS``). + +Phase C hook: a ``[backend]`` section in ``config.toml`` selects the VM +backend implementation; the field is read but currently only the +``workstation`` value is supported. +""" + +from __future__ import annotations + +import os +import tomllib +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +_ENV_KEYS = ("CI_ROOT", "CI_TEMPLATES", "CI_BUILD_VMS", "CI_ARTIFACTS", "CI_KEYS") + + +@dataclass(frozen=True) +class Paths: + """Filesystem paths used by the orchestrator.""" + + root: Path + templates: Path + build_vms: Path + artifacts: Path + keys: Path + + +@dataclass(frozen=True) +class BackendConfig: + """Backend selector. ``type`` is currently always 'workstation'.""" + + type: str = "workstation" + options: dict[str, Any] = field(default_factory=dict) + + +@dataclass(frozen=True) +class Config: + """Top-level configuration object.""" + + paths: Paths + backend: BackendConfig + vmrun_path: str | None = None + ssh_key_path: Path | None = None + guest_cred_target: str = "BuildVMGuest" + + +def _platform_defaults() -> Paths: + root = Path(r"F:\CI") if os.name == "nt" else Path("/var/lib/ci") + return Paths( + root=root, + templates=root / "Templates" if os.name == "nt" else root / "templates", + build_vms=root / "BuildVMs" if os.name == "nt" else root / "build-vms", + artifacts=root / "Artifacts" if os.name == "nt" else root / "artifacts", + keys=root / "keys" if os.name == "nt" else Path("/etc/ci/keys"), + ) + + +def _load_toml(path: Path) -> dict[str, Any]: + with path.open("rb") as fh: + return tomllib.load(fh) + + +def _merge_paths(base: Paths, overrides: dict[str, Any]) -> Paths: + def pick(key: str, current: Path) -> Path: + val = overrides.get(key) + return Path(val) if val else current + + return Paths( + root=pick("root", base.root), + templates=pick("templates", base.templates), + build_vms=pick("build_vms", base.build_vms), + artifacts=pick("artifacts", base.artifacts), + keys=pick("keys", base.keys), + ) + + +def _apply_env(paths: Paths, env: dict[str, str]) -> Paths: + mapping = { + "CI_ROOT": "root", + "CI_TEMPLATES": "templates", + "CI_BUILD_VMS": "build_vms", + "CI_ARTIFACTS": "artifacts", + "CI_KEYS": "keys", + } + overrides: dict[str, Any] = {} + for env_key, attr in mapping.items(): + val = env.get(env_key) + if val: + overrides[attr] = val + return _merge_paths(paths, overrides) + + +def load_config( + config_path: Path | None = None, + env: dict[str, str] | None = None, +) -> Config: + """Build a :class:`Config` from defaults, optional TOML file, and env vars. + + Parameters + ---------- + config_path: + Explicit path to a TOML file. If ``None``, looks at ``$CI_CONFIG`` then + ``$CI_ROOT/config.toml``. + env: + Environment dict (defaults to :data:`os.environ`). Injected for tests. + """ + if env is None: + env = dict(os.environ) + + paths = _platform_defaults() + + # Resolve config file + toml_path: Path | None = None + if config_path is not None: + toml_path = config_path + elif env.get("CI_CONFIG"): + toml_path = Path(env["CI_CONFIG"]) + else: + candidate = paths.root / "config.toml" + if candidate.is_file(): + toml_path = candidate + + backend = BackendConfig() + vmrun_path: str | None = env.get("CI_VMRUN_PATH") + ssh_key_path = Path(env["CI_SSH_KEY_PATH"]) if env.get("CI_SSH_KEY_PATH") else None + guest_cred_target = env.get("CI_GUEST_CRED_TARGET", "BuildVMGuest") + + if toml_path and toml_path.is_file(): + data = _load_toml(toml_path) + paths = _merge_paths(paths, data.get("paths", {})) + backend_section = data.get("backend", {}) + if backend_section: + backend = BackendConfig( + type=str(backend_section.get("type", "workstation")), + options={k: v for k, v in backend_section.items() if k != "type"}, + ) + if vmrun_path is None: + vmrun_path = data.get("vmrun_path") + if ssh_key_path is None and data.get("ssh_key_path"): + ssh_key_path = Path(data["ssh_key_path"]) + guest_cred_target = data.get("guest_cred_target", guest_cred_target) + + paths = _apply_env(paths, env) + + return Config( + paths=paths, + backend=backend, + vmrun_path=vmrun_path, + ssh_key_path=ssh_key_path, + guest_cred_target=guest_cred_target, + ) + + +__all__ = ["BackendConfig", "Config", "Paths", "load_config"] diff --git a/src/ci_orchestrator/credentials.py b/src/ci_orchestrator/credentials.py new file mode 100644 index 0000000..d192e59 --- /dev/null +++ b/src/ci_orchestrator/credentials.py @@ -0,0 +1,59 @@ +"""Credential store abstraction. + +Wraps the ``keyring`` library so callers do not depend on it directly. +Phase B note: the same interface is used on both Windows (Credential +Manager) and Linux (Secret Service / file-based vault); see +``plans/implementation-plan-A-B.md`` step B3. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Protocol + + +@dataclass(frozen=True) +class Credential: + """Username + secret pair retrieved from a credential store.""" + + username: str + password: str + + +class CredentialStore(Protocol): + """Read-only credential lookup.""" + + def get(self, target: str) -> Credential: + """Return the credential associated with ``target``. + + Raises :class:`KeyError` if the target is unknown. + """ + ... + + +class KeyringCredentialStore: + """Default :class:`CredentialStore` backed by the ``keyring`` library.""" + + def __init__(self, service: str = "ci") -> None: + self._service = service + + def get(self, target: str) -> Credential: + try: + import keyring + except ImportError as exc: # pragma: no cover - dep declared + raise RuntimeError( + "keyring is not installed; run `pip install keyring`." + ) from exc + + cred = keyring.get_credential(target, None) + if cred is None: + # Fallback: some backends only support get_password and need the + # username to be the same as the target. + password = keyring.get_password(self._service, target) + if password is None: + raise KeyError(f"Credential '{target}' not found in keyring.") + return Credential(username=target, password=password) + return Credential(username=cred.username, password=cred.password) + + +__all__ = ["Credential", "CredentialStore", "KeyringCredentialStore"] diff --git a/src/ci_orchestrator/transport/__init__.py b/src/ci_orchestrator/transport/__init__.py new file mode 100644 index 0000000..ba9b3eb --- /dev/null +++ b/src/ci_orchestrator/transport/__init__.py @@ -0,0 +1,9 @@ +"""Guest transport layer (WinRM for Windows, SSH for Linux).""" + +from __future__ import annotations + +from ci_orchestrator.transport.errors import TransportError +from ci_orchestrator.transport.ssh import SshTransport +from ci_orchestrator.transport.winrm import WinRmTransport + +__all__ = ["SshTransport", "TransportError", "WinRmTransport"] diff --git a/src/ci_orchestrator/transport/errors.py b/src/ci_orchestrator/transport/errors.py new file mode 100644 index 0000000..1a5eb06 --- /dev/null +++ b/src/ci_orchestrator/transport/errors.py @@ -0,0 +1,23 @@ +"""Transport error hierarchy.""" + +from __future__ import annotations + + +class TransportError(RuntimeError): + """Base class for transport-layer (WinRM/SSH) failures.""" + + +class TransportConnectError(TransportError): + """Raised when the underlying connection cannot be established.""" + + +class TransportCommandError(TransportError): + """Raised when a remote command exits non-zero.""" + + def __init__(self, returncode: int, stdout: str, stderr: str) -> None: + super().__init__( + f"Remote command failed (exit={returncode}): {(stderr or stdout).strip()[:500]}" + ) + self.returncode = returncode + self.stdout = stdout + self.stderr = stderr diff --git a/src/ci_orchestrator/transport/ssh.py b/src/ci_orchestrator/transport/ssh.py new file mode 100644 index 0000000..141cf69 --- /dev/null +++ b/src/ci_orchestrator/transport/ssh.py @@ -0,0 +1,150 @@ +"""SSH transport via ``paramiko``. + +Replaces the ``ssh.exe`` / ``scp.exe`` calls used in +``scripts/_Transport.psm1``. By default the host-key policy is +``AutoAddPolicy`` with an empty ``known_hosts`` file (i.e. permissive), +which mirrors the ``StrictHostKeyChecking=no UserKnownHostsFile=NUL`` +default used for ephemeral CI build VMs. + +See ``AGENTS.md`` error #12: in PowerShell with ``$ErrorActionPreference = +'Stop'`` the native ``ssh-keygen -R`` call could raise terminating errors +on stderr output. With paramiko there is no equivalent issue. +""" + +from __future__ import annotations + +import contextlib +from dataclasses import dataclass +from pathlib import Path +from typing import TYPE_CHECKING + +from ci_orchestrator.transport.errors import TransportCommandError, TransportConnectError + +if TYPE_CHECKING: # pragma: no cover + import paramiko + + +@dataclass(frozen=True) +class SshResult: + """Result of a remote SSH command.""" + + stdout: str + stderr: str + returncode: int + + @property + def ok(self) -> bool: + return self.returncode == 0 + + +class SshTransport: + """Thin wrapper around :class:`paramiko.SSHClient`.""" + + def __init__( + self, + host: str, + *, + username: str = "ci_build", + key_path: str | Path | None = None, + port: int = 22, + connect_timeout: float = 10.0, + known_hosts: str | Path | None = None, + ) -> None: + self._host = host + self._username = username + self._key_path = Path(key_path) if key_path else None + self._port = port + self._connect_timeout = connect_timeout + self._known_hosts = Path(known_hosts) if known_hosts else None + self._client: paramiko.SSHClient | None = None + + # ----------------------------------------------------------- connection + def _connect(self) -> paramiko.SSHClient: + if self._client is not None: + return self._client + try: + import paramiko + except ImportError as exc: # pragma: no cover - dep declared + raise TransportConnectError( + "paramiko is not installed; run `pip install paramiko`." + ) from exc + + client = paramiko.SSHClient() + if self._known_hosts is not None and self._known_hosts.is_file(): + client.load_host_keys(str(self._known_hosts)) + client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) + else: + # Permissive mode for ephemeral CI VMs (see _Transport.psm1). + client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) + + try: + client.connect( + hostname=self._host, + port=self._port, + username=self._username, + key_filename=str(self._key_path) if self._key_path else None, + timeout=self._connect_timeout, + allow_agent=False, + look_for_keys=False, + ) + except Exception as exc: + raise TransportConnectError( + f"SSH connection to {self._username}@{self._host}:{self._port} failed: {exc}" + ) from exc + self._client = client + return client + + def close(self) -> None: + if self._client is not None: + with contextlib.suppress(Exception): + self._client.close() + self._client = None + + def __enter__(self) -> SshTransport: + self._connect() + return self + + def __exit__(self, *_exc: object) -> None: + self.close() + + # --------------------------------------------------------------- ops + def run(self, command: str, *, check: bool = True, timeout: float | None = None) -> SshResult: + """Run ``command`` in the guest shell.""" + client = self._connect() + try: + _stdin, stdout, stderr = client.exec_command(command, timeout=timeout) + out = stdout.read().decode("utf-8", errors="replace") + err = stderr.read().decode("utf-8", errors="replace") + rc = stdout.channel.recv_exit_status() + except Exception as exc: + raise TransportConnectError(f"SSH exec_command failed: {exc}") from exc + result = SshResult(stdout=out, stderr=err, returncode=rc) + if check and not result.ok: + raise TransportCommandError(result.returncode, result.stdout, result.stderr) + return result + + def copy(self, local_path: str | Path, remote_path: str) -> None: + """Upload via SFTP.""" + client = self._connect() + try: + with client.open_sftp() as sftp: + sftp.put(str(local_path), remote_path) + except Exception as exc: + raise TransportConnectError(f"SFTP put failed: {exc}") from exc + + def fetch(self, remote_path: str, local_path: str | Path) -> None: + """Download via SFTP.""" + client = self._connect() + try: + with client.open_sftp() as sftp: + sftp.get(remote_path, str(local_path)) + except Exception as exc: + raise TransportConnectError(f"SFTP get failed: {exc}") from exc + + def is_ready(self) -> bool: + """Lightweight readiness probe.""" + try: + result = self.run("true", check=False, timeout=5.0) + except (TransportConnectError, TransportCommandError): + return False + return result.ok diff --git a/src/ci_orchestrator/transport/winrm.py b/src/ci_orchestrator/transport/winrm.py new file mode 100644 index 0000000..93e7750 --- /dev/null +++ b/src/ci_orchestrator/transport/winrm.py @@ -0,0 +1,156 @@ +"""WinRM transport via ``pypsrp``. + +Replaces the ``New-PSSession`` / ``Invoke-Command`` calls used in the +PowerShell scripts. CI build VMs use a self-signed TLS certificate on +WinRM port 5986, so SSL validation is disabled by default — appropriate +for an isolated lab network (see ``scripts/_Common.psm1``). +""" + +from __future__ import annotations + +import contextlib +from dataclasses import dataclass +from pathlib import Path +from typing import TYPE_CHECKING, Any + +from ci_orchestrator.transport.errors import TransportCommandError, TransportConnectError + +if TYPE_CHECKING: # pragma: no cover + from pypsrp.client import Client + + +@dataclass(frozen=True) +class WinRmResult: + """Result of a remote PowerShell invocation.""" + + stdout: str + stderr: str + returncode: int + + @property + def ok(self) -> bool: + return self.returncode == 0 + + +class WinRmTransport: + """Thin wrapper around :class:`pypsrp.client.Client`. + + Parameters + ---------- + host: + IP or DNS name of the guest. + username, password: + Credentials. Pass via :class:`~ci_orchestrator.credentials.Credential` + in caller code; do not embed plaintext. + port: + WinRM HTTPS port (default 5986). + cert_validation: + When ``False`` (default), self-signed certs are accepted — required + for the ephemeral build VMs in this lab. + """ + + def __init__( + self, + host: str, + username: str, + password: str, + *, + port: int = 5986, + ssl: bool = True, + cert_validation: bool = False, + connection_timeout: int = 30, + ) -> None: + self._host = host + self._username = username + self._password = password + self._port = port + self._ssl = ssl + self._cert_validation = cert_validation + self._connection_timeout = connection_timeout + self._client: Client | None = None + + # ----------------------------------------------------------- connection + def _connect(self) -> Client: + if self._client is not None: + return self._client + try: + from pypsrp.client import Client # local import: optional dep at runtime + except ImportError as exc: # pragma: no cover - dep declared in pyproject + raise TransportConnectError( + "pypsrp is not installed; run `pip install pypsrp`." + ) from exc + try: + self._client = Client( + self._host, + username=self._username, + password=self._password, + port=self._port, + ssl=self._ssl, + cert_validation=self._cert_validation, + connection_timeout=self._connection_timeout, + ) + except Exception as exc: + raise TransportConnectError( + f"WinRM connection to {self._host}:{self._port} failed: {exc}" + ) from exc + return self._client + + def close(self) -> None: + if self._client is not None: + with contextlib.suppress(Exception): + self._client.wsman.close() + self._client = None + + def __enter__(self) -> WinRmTransport: + self._connect() + return self + + def __exit__(self, *_exc: object) -> None: + self.close() + + # --------------------------------------------------------------- ops + def run(self, script: str, *, check: bool = True) -> WinRmResult: + """Run a PowerShell script block in the guest.""" + client = self._connect() + try: + stdout, stderr, returncode = client.execute_ps(script) + except Exception as exc: + raise TransportConnectError(f"WinRM execute_ps failed: {exc}") from exc + # pypsrp returns stderr as a list of PSRP error records in some versions; + # normalise to a string. + stderr_any: Any = stderr + if isinstance(stderr_any, str): + stderr_str = stderr_any + else: + try: + stderr_str = "\n".join(str(s) for s in stderr_any) + except TypeError: + stderr_str = str(stderr_any) + result = WinRmResult(stdout=stdout, stderr=stderr_str, returncode=int(returncode)) + if check and not result.ok: + raise TransportCommandError(result.returncode, result.stdout, result.stderr) + return result + + def copy(self, local_path: str | Path, remote_path: str) -> None: + """Upload a file to the guest via PSRP.""" + client = self._connect() + try: + client.copy(str(local_path), remote_path) + except Exception as exc: + raise TransportConnectError(f"WinRM copy failed: {exc}") from exc + + def fetch(self, remote_path: str, local_path: str | Path) -> None: + """Download a file from the guest via PSRP.""" + client = self._connect() + try: + client.fetch(remote_path, str(local_path)) + except Exception as exc: + raise TransportConnectError(f"WinRM fetch failed: {exc}") from exc + + def is_ready(self) -> bool: + """Lightweight readiness probe: run ``$true`` and check exit code.""" + try: + result = self.run("$true | Out-Null", check=False) + except (TransportConnectError, TransportCommandError): + return False + return result.ok diff --git a/tests/python/__init__.py b/tests/python/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/python/conftest.py b/tests/python/conftest.py new file mode 100644 index 0000000..8284c8e --- /dev/null +++ b/tests/python/conftest.py @@ -0,0 +1,11 @@ +"""Shared pytest fixtures for ci_orchestrator unit tests.""" + +from __future__ import annotations + +import sys +from pathlib import Path + +# Ensure src/ is on sys.path so `import ci_orchestrator` works without install. +_SRC = Path(__file__).resolve().parents[2] / "src" +if str(_SRC) not in sys.path: + sys.path.insert(0, str(_SRC)) diff --git a/tests/python/test_cli.py b/tests/python/test_cli.py new file mode 100644 index 0000000..9739e4a --- /dev/null +++ b/tests/python/test_cli.py @@ -0,0 +1,133 @@ +"""Smoke tests for the click CLI entry point. + +These do not hit any real VM — backend, transport, credentials and +``time.sleep`` are all monkeypatched. +""" + +from __future__ import annotations + +from typing import Any + +import pytest +from click.testing import CliRunner + +import ci_orchestrator.__main__ as cli_module +from ci_orchestrator import __version__ +from ci_orchestrator.backends.protocol import VmHandle +from ci_orchestrator.credentials import Credential + + +class _FakeBackend: + def __init__( + self, + *, + running: bool = True, + ip: str | None = "10.0.0.42", + ) -> None: + self._running = running + self._ip = ip + + def is_running(self, _h: VmHandle) -> bool: + return self._running + + def get_ip(self, _h: VmHandle, timeout: float = 0.0) -> str | None: + return self._ip + + +class _FakeTransport: + def __init__(self, *_a: Any, **_kw: Any) -> None: + pass + + def __enter__(self) -> _FakeTransport: + return self + + def __exit__(self, *_e: object) -> None: + pass + + def is_ready(self) -> bool: + return True + + +class _FakeStore: + def get(self, _target: str) -> Credential: + return Credential("user", "pwd") + + +def _patch_common(monkeypatch: pytest.MonkeyPatch, backend: _FakeBackend) -> None: + monkeypatch.setattr(cli_module, "load_backend", lambda _cfg: backend) + monkeypatch.setattr(cli_module, "KeyringCredentialStore", lambda: _FakeStore()) + monkeypatch.setattr(cli_module, "WinRmTransport", _FakeTransport) + monkeypatch.setattr(cli_module, "SshTransport", _FakeTransport) + monkeypatch.setattr(cli_module.time, "sleep", lambda _s: None) + + +def test_help_lists_wait_ready() -> None: + result = CliRunner().invoke(cli_module.cli, ["--help"]) + assert result.exit_code == 0 + assert "wait-ready" in result.output + + +def test_version() -> None: + result = CliRunner().invoke(cli_module.cli, ["--version"]) + assert result.exit_code == 0 + assert __version__ in result.output + + +def test_wait_ready_windows_success(monkeypatch: pytest.MonkeyPatch) -> None: + _patch_common(monkeypatch, _FakeBackend(running=True, ip="10.0.0.42")) + result = CliRunner().invoke( + cli_module.cli, + ["wait-ready", "--vmx", "x.vmx", "--guest-os", "windows", "--timeout", "1"], + ) + assert result.exit_code == 0, result.output + assert "WinRM ready" in result.output + + +def test_wait_ready_linux_success(monkeypatch: pytest.MonkeyPatch) -> None: + _patch_common(monkeypatch, _FakeBackend(running=True, ip="10.0.0.42")) + result = CliRunner().invoke( + cli_module.cli, + ["wait-ready", "--vmx", "x.vmx", "--guest-os", "linux", "--timeout", "1"], + ) + assert result.exit_code == 0, result.output + assert "SSH ready" in result.output + + +def test_wait_ready_timeout_when_not_running(monkeypatch: pytest.MonkeyPatch) -> None: + _patch_common(monkeypatch, _FakeBackend(running=False)) + result = CliRunner().invoke( + cli_module.cli, + [ + "wait-ready", + "--vmx", + "x.vmx", + "--guest-os", + "windows", + "--timeout", + "0.01", + "--poll-interval", + "0.001", + ], + ) + assert result.exit_code == 2 + assert "timeout waiting for VM" in result.output + + +def test_wait_ready_timeout_no_ip(monkeypatch: pytest.MonkeyPatch) -> None: + _patch_common(monkeypatch, _FakeBackend(running=True, ip=None)) + result = CliRunner().invoke( + cli_module.cli, + [ + "wait-ready", + "--vmx", + "x.vmx", + "--guest-os", + "windows", + "--timeout", + "0.01", + "--poll-interval", + "0.001", + ], + ) + assert result.exit_code == 3 + assert "timeout waiting for guest IP" in result.output diff --git a/tests/python/test_config.py b/tests/python/test_config.py new file mode 100644 index 0000000..5eaf57e --- /dev/null +++ b/tests/python/test_config.py @@ -0,0 +1,62 @@ +"""Tests for ci_orchestrator.config.""" + +from __future__ import annotations + +import os +from pathlib import Path + +from ci_orchestrator.config import load_config + + +def test_defaults_are_os_aware() -> None: + cfg = load_config(env={}) + if os.name == "nt": + assert str(cfg.paths.root) == r"F:\CI" + else: + assert str(cfg.paths.root) == "/var/lib/ci" + + +def test_env_overrides_paths(tmp_path: Path) -> None: + cfg = load_config( + env={ + "CI_ROOT": str(tmp_path / "root"), + "CI_TEMPLATES": str(tmp_path / "tpl"), + "CI_BUILD_VMS": str(tmp_path / "bvm"), + "CI_ARTIFACTS": str(tmp_path / "art"), + "CI_KEYS": str(tmp_path / "keys"), + } + ) + assert cfg.paths.root == tmp_path / "root" + assert cfg.paths.templates == tmp_path / "tpl" + assert cfg.paths.build_vms == tmp_path / "bvm" + assert cfg.paths.artifacts == tmp_path / "art" + assert cfg.paths.keys == tmp_path / "keys" + + +def test_toml_overrides_defaults(tmp_path: Path) -> None: + toml = tmp_path / "config.toml" + toml.write_text( + '[paths]\n' + f'root = "{(tmp_path / "altroot").as_posix()}"\n' + '[backend]\n' + 'type = "workstation"\n' + 'extra = "value"\n', + encoding="utf-8", + ) + cfg = load_config(config_path=toml, env={}) + assert cfg.paths.root == tmp_path / "altroot" + assert cfg.backend.type == "workstation" + assert cfg.backend.options == {"extra": "value"} + + +def test_env_wins_over_toml(tmp_path: Path) -> None: + toml = tmp_path / "config.toml" + toml.write_text( + f'[paths]\nroot = "{(tmp_path / "fromtoml").as_posix()}"\n', + encoding="utf-8", + ) + cfg = load_config( + config_path=toml, + env={"CI_ROOT": str(tmp_path / "fromenv")}, + ) + assert cfg.paths.root == tmp_path / "fromenv" diff --git a/tests/python/test_credentials.py b/tests/python/test_credentials.py new file mode 100644 index 0000000..a48271c --- /dev/null +++ b/tests/python/test_credentials.py @@ -0,0 +1,58 @@ +"""Tests for KeyringCredentialStore (mocked keyring backend).""" + +from __future__ import annotations + +import sys +import types +from typing import Any + +import pytest + +from ci_orchestrator.credentials import KeyringCredentialStore + + +def _install_fake_keyring(monkeypatch: pytest.MonkeyPatch, **behaviour: Any) -> None: + fake = types.ModuleType("keyring") + + def get_credential(target: str, _username: str | None) -> Any: + return behaviour.get("credential_map", {}).get(target) + + def get_password(service: str, target: str) -> str | None: + return behaviour.get("password_map", {}).get((service, target)) + + fake.get_credential = get_credential # type: ignore[attr-defined] + fake.get_password = get_password # type: ignore[attr-defined] + monkeypatch.setitem(sys.modules, "keyring", fake) + + +class _FakeCred: + def __init__(self, username: str, password: str) -> None: + self.username = username + self.password = password + + +def test_get_returns_credential(monkeypatch: pytest.MonkeyPatch) -> None: + _install_fake_keyring( + monkeypatch, + credential_map={"BuildVMGuest": _FakeCred("Administrator", "s3cret")}, + ) + cred = KeyringCredentialStore().get("BuildVMGuest") + assert cred.username == "Administrator" + assert cred.password == "s3cret" + + +def test_get_falls_back_to_password(monkeypatch: pytest.MonkeyPatch) -> None: + _install_fake_keyring( + monkeypatch, + credential_map={}, + password_map={("ci", "GiteaPAT"): "tokentoken"}, + ) + cred = KeyringCredentialStore().get("GiteaPAT") + assert cred.username == "GiteaPAT" + assert cred.password == "tokentoken" + + +def test_get_raises_when_missing(monkeypatch: pytest.MonkeyPatch) -> None: + _install_fake_keyring(monkeypatch, credential_map={}, password_map={}) + with pytest.raises(KeyError): + KeyringCredentialStore().get("Unknown") diff --git a/tests/python/test_load_backend.py b/tests/python/test_load_backend.py new file mode 100644 index 0000000..0457c84 --- /dev/null +++ b/tests/python/test_load_backend.py @@ -0,0 +1,36 @@ +"""Factory test: load_backend dispatches on config.backend.type.""" + +from __future__ import annotations + +from dataclasses import dataclass +from pathlib import Path + +import pytest + +from ci_orchestrator.backends import load_backend +from ci_orchestrator.backends.workstation import WorkstationVmrunBackend + + +@dataclass +class _BE: + type: str + + +@dataclass +class _Cfg: + backend: _BE + vmrun_path: str | None = None + + +def test_load_backend_returns_workstation(tmp_path: Path) -> None: + fake = tmp_path / "vmrun.exe" + fake.write_bytes(b"") + backend = load_backend(_Cfg(backend=_BE("workstation"), vmrun_path=str(fake))) + assert isinstance(backend, WorkstationVmrunBackend) + + +def test_load_backend_rejects_unknown(tmp_path: Path) -> None: + fake = tmp_path / "vmrun.exe" + fake.write_bytes(b"") + with pytest.raises(NotImplementedError): + load_backend(_Cfg(backend=_BE("esxi"), vmrun_path=str(fake))) diff --git a/tests/python/test_ssh_transport.py b/tests/python/test_ssh_transport.py new file mode 100644 index 0000000..609dddb --- /dev/null +++ b/tests/python/test_ssh_transport.py @@ -0,0 +1,159 @@ +"""Tests for SshTransport (paramiko mocked). + +Covers AGENTS.md error #12 indirectly: with paramiko there is no native +``ssh-keygen`` invocation, so the PowerShell stderr-handling pitfall does +not apply. The default permissive host-key policy mirrors the +``StrictHostKeyChecking=no`` / ``UserKnownHostsFile=NUL`` defaults of +``scripts/_Transport.psm1``. +""" + +from __future__ import annotations + +import sys +import types +from typing import Any + +import pytest + +from ci_orchestrator.transport.errors import TransportCommandError, TransportConnectError +from ci_orchestrator.transport.ssh import SshTransport + + +class _FakeChannel: + def __init__(self, rc: int) -> None: + self._rc = rc + + def recv_exit_status(self) -> int: + return self._rc + + +class _FakeStream: + def __init__(self, data: bytes, rc: int = 0) -> None: + self._data = data + self.channel = _FakeChannel(rc) + + def read(self) -> bytes: + return self._data + + +class _FakeSFTP: + def __init__(self) -> None: + self.puts: list[tuple[str, str]] = [] + self.gets: list[tuple[str, str]] = [] + + def __enter__(self) -> _FakeSFTP: + return self + + def __exit__(self, *_e: object) -> None: + pass + + def put(self, local: str, remote: str) -> None: + self.puts.append((local, remote)) + + def get(self, remote: str, local: str) -> None: + self.gets.append((remote, local)) + + +class _FakeSSHClient: + def __init__(self) -> None: + self.connect_kwargs: dict[str, Any] = {} + self.policy: Any = None + self.next_stdout = b"" + self.next_stderr = b"" + self.next_rc = 0 + self.commands: list[str] = [] + self.sftp = _FakeSFTP() + self.closed = False + + def set_missing_host_key_policy(self, policy: Any) -> None: + self.policy = policy + + def load_host_keys(self, _path: str) -> None: + pass + + def connect(self, **kwargs: Any) -> None: + self.connect_kwargs = kwargs + + def exec_command(self, command: str, timeout: float | None = None) -> Any: + self.commands.append(command) + return ( + None, + _FakeStream(self.next_stdout, self.next_rc), + _FakeStream(self.next_stderr, self.next_rc), + ) + + def open_sftp(self) -> _FakeSFTP: + return self.sftp + + def close(self) -> None: + self.closed = True + + +def _install_fake_paramiko(monkeypatch: pytest.MonkeyPatch, client: _FakeSSHClient) -> Any: + fake = types.ModuleType("paramiko") + fake.SSHClient = lambda: client # type: ignore[attr-defined] + fake.AutoAddPolicy = lambda: "auto-add" # type: ignore[attr-defined] + monkeypatch.setitem(sys.modules, "paramiko", fake) + return fake + + +def test_run_returns_stdout(monkeypatch: pytest.MonkeyPatch) -> None: + client = _FakeSSHClient() + client.next_stdout = b"hello\n" + _install_fake_paramiko(monkeypatch, client) + t = SshTransport("10.0.0.2", username="ci_build", key_path="/tmp/key") + result = t.run("echo hello") + assert result.ok + assert result.stdout == "hello\n" + assert client.commands == ["echo hello"] + + +def test_run_raises_on_nonzero(monkeypatch: pytest.MonkeyPatch) -> None: + client = _FakeSSHClient() + client.next_rc = 2 + client.next_stderr = b"oops" + _install_fake_paramiko(monkeypatch, client) + t = SshTransport("10.0.0.2", key_path="/tmp/key") + with pytest.raises(TransportCommandError): + t.run("bad") + + +def test_default_policy_is_auto_add(monkeypatch: pytest.MonkeyPatch) -> None: + """Mirrors permissive `StrictHostKeyChecking=no` default for ephemeral CI VMs.""" + client = _FakeSSHClient() + _install_fake_paramiko(monkeypatch, client) + SshTransport("10.0.0.2", key_path="/tmp/key").run("true") + assert client.policy == "auto-add" + + +def test_copy_and_fetch(monkeypatch: pytest.MonkeyPatch) -> None: + client = _FakeSSHClient() + _install_fake_paramiko(monkeypatch, client) + t = SshTransport("10.0.0.2", key_path="/tmp/key") + t.copy("local.bin", "/remote/local.bin") + t.fetch("/remote/out.bin", "out.bin") + assert client.sftp.puts == [("local.bin", "/remote/local.bin")] + assert client.sftp.gets == [("/remote/out.bin", "out.bin")] + + +def test_is_ready_returns_false_on_connect_failure(monkeypatch: pytest.MonkeyPatch) -> None: + fake = types.ModuleType("paramiko") + + class _Boom: + def set_missing_host_key_policy(self, _p: object) -> None: + pass + + def connect(self, **_kw: Any) -> None: + raise OSError("refused") + + def close(self) -> None: + pass + + fake.SSHClient = lambda: _Boom() # type: ignore[attr-defined] + fake.AutoAddPolicy = lambda: object() # type: ignore[attr-defined] + monkeypatch.setitem(sys.modules, "paramiko", fake) + + t = SshTransport("10.0.0.2", key_path="/tmp/key") + assert t.is_ready() is False + with pytest.raises(TransportConnectError): + t.run("x") diff --git a/tests/python/test_winrm_transport.py b/tests/python/test_winrm_transport.py new file mode 100644 index 0000000..1fa2e23 --- /dev/null +++ b/tests/python/test_winrm_transport.py @@ -0,0 +1,100 @@ +"""Tests for WinRmTransport (pypsrp client mocked).""" + +from __future__ import annotations + +import sys +import types +from typing import Any + +import pytest + +from ci_orchestrator.transport.errors import TransportCommandError, TransportConnectError +from ci_orchestrator.transport.winrm import WinRmTransport + + +class _FakeClient: + def __init__(self, *args: Any, **kwargs: Any) -> None: + self.args = args + self.kwargs = kwargs + self.copies: list[tuple[str, str]] = [] + self.fetches: list[tuple[str, str]] = [] + self.scripts: list[str] = [] + self.next_result: tuple[str, str, int] = ("ok", "", 0) + self.wsman = types.SimpleNamespace(close=lambda: None) + + def execute_ps(self, script: str) -> tuple[str, str, int]: + self.scripts.append(script) + return self.next_result + + def copy(self, local: str, remote: str) -> None: + self.copies.append((local, remote)) + + def fetch(self, remote: str, local: str) -> None: + self.fetches.append((remote, local)) + + +def _install_fake_pypsrp(monkeypatch: pytest.MonkeyPatch, client: _FakeClient) -> None: + module = types.ModuleType("pypsrp") + client_module = types.ModuleType("pypsrp.client") + client_module.Client = lambda *a, **kw: client # type: ignore[attr-defined] + monkeypatch.setitem(sys.modules, "pypsrp", module) + monkeypatch.setitem(sys.modules, "pypsrp.client", client_module) + + +def test_run_returns_result(monkeypatch: pytest.MonkeyPatch) -> None: + fake = _FakeClient() + fake.next_result = ("hello\n", "", 0) + _install_fake_pypsrp(monkeypatch, fake) + t = WinRmTransport("10.0.0.1", "user", "pwd") + result = t.run("Write-Output 'hello'") + assert result.ok + assert result.stdout == "hello\n" + assert fake.scripts == ["Write-Output 'hello'"] + + +def test_run_raises_on_failure(monkeypatch: pytest.MonkeyPatch) -> None: + fake = _FakeClient() + fake.next_result = ("", "boom", 7) + _install_fake_pypsrp(monkeypatch, fake) + t = WinRmTransport("10.0.0.1", "user", "pwd") + with pytest.raises(TransportCommandError) as exc: + t.run("bad") + assert exc.value.returncode == 7 + + +def test_run_check_false_does_not_raise(monkeypatch: pytest.MonkeyPatch) -> None: + fake = _FakeClient() + fake.next_result = ("", "boom", 7) + _install_fake_pypsrp(monkeypatch, fake) + t = WinRmTransport("10.0.0.1", "user", "pwd") + result = t.run("bad", check=False) + assert not result.ok + + +def test_copy_and_fetch(monkeypatch: pytest.MonkeyPatch) -> None: + fake = _FakeClient() + _install_fake_pypsrp(monkeypatch, fake) + t = WinRmTransport("10.0.0.1", "user", "pwd") + t.copy("local.txt", "C:/remote.txt") + t.fetch("C:/remote.txt", "downloaded.txt") + assert fake.copies == [("local.txt", "C:/remote.txt")] + assert fake.fetches == [("C:/remote.txt", "downloaded.txt")] + + +def test_is_ready_handles_connect_failure(monkeypatch: pytest.MonkeyPatch) -> None: + # Make the local import inside _connect succeed but Client raise. + module = types.ModuleType("pypsrp") + client_module = types.ModuleType("pypsrp.client") + + def _boom(*_a: object, **_kw: object) -> None: + raise OSError("connection refused") + + client_module.Client = _boom # type: ignore[attr-defined] + monkeypatch.setitem(sys.modules, "pypsrp", module) + monkeypatch.setitem(sys.modules, "pypsrp.client", client_module) + + t = WinRmTransport("10.0.0.1", "user", "pwd") + assert t.is_ready() is False + # Direct .run should raise. + with pytest.raises(TransportConnectError): + t.run("x") diff --git a/tests/python/test_workstation_backend.py b/tests/python/test_workstation_backend.py new file mode 100644 index 0000000..9842936 --- /dev/null +++ b/tests/python/test_workstation_backend.py @@ -0,0 +1,149 @@ +"""Unit tests for WorkstationVmrunBackend. + +Cover AGENTS.md error #10: ``is_running`` MUST use ``vmrun list`` and +NOT ``getGuestIPAddress`` (which would block 30-60s without VMware Tools). +""" + +from __future__ import annotations + +import subprocess +from pathlib import Path +from typing import Any + +import pytest + +from ci_orchestrator.backends.errors import BackendNotAvailable, BackendOperationFailed +from ci_orchestrator.backends.protocol import VmHandle +from ci_orchestrator.backends.workstation import WorkstationVmrunBackend + + +@pytest.fixture +def fake_vmrun(tmp_path: Path) -> str: + """A real (empty) file we can pass as vmrun_path.""" + p = tmp_path / "vmrun.exe" + p.write_bytes(b"") + return str(p) + + +def _make_completed(returncode: int = 0, stdout: str = "", stderr: str = "") -> Any: + return subprocess.CompletedProcess(args=[], returncode=returncode, stdout=stdout, stderr=stderr) + + +def test_init_raises_when_vmrun_missing(tmp_path: Path) -> None: + missing = tmp_path / "nope.exe" + with pytest.raises(BackendNotAvailable): + WorkstationVmrunBackend(vmrun_path=str(missing)) + + +def test_init_uses_explicit_path(fake_vmrun: str) -> None: + backend = WorkstationVmrunBackend(vmrun_path=fake_vmrun) + assert backend.vmrun_path == fake_vmrun + + +def test_clone_linked_invokes_vmrun(monkeypatch: pytest.MonkeyPatch, fake_vmrun: str) -> None: + captured: dict[str, list[str]] = {} + + def fake_run(cmd: list[str], **_kwargs: object) -> Any: + captured["cmd"] = cmd + return _make_completed(0, stdout="") + + monkeypatch.setattr(subprocess, "run", fake_run) + backend = WorkstationVmrunBackend(vmrun_path=fake_vmrun) + handle = backend.clone_linked( + template=r"C:\tpl\WinBuild2025\WinBuild2025.vmx", + snapshot="BaseClean", + name="job-123", + destination=r"C:\bvm\job-123\job-123.vmx", + ) + assert handle.identifier == r"C:\bvm\job-123\job-123.vmx" + assert handle.name == "job-123" + assert captured["cmd"][0] == fake_vmrun + assert captured["cmd"][1:4] == ["-T", "ws", "clone"] + assert "linked" in captured["cmd"] + assert "-snapshot=BaseClean" in captured["cmd"] + assert "-cloneName=job-123" in captured["cmd"] + + +def test_clone_failure_raises(monkeypatch: pytest.MonkeyPatch, fake_vmrun: str) -> None: + monkeypatch.setattr( + subprocess, + "run", + lambda *a, **kw: _make_completed(1, stderr="Error: source not found"), + ) + backend = WorkstationVmrunBackend(vmrun_path=fake_vmrun) + with pytest.raises(BackendOperationFailed) as exc_info: + backend.clone_linked("src.vmx", "snap", "name", destination="dst.vmx") + assert exc_info.value.returncode == 1 + assert "source not found" in exc_info.value.output + + +def test_stop_tolerates_already_off(monkeypatch: pytest.MonkeyPatch, fake_vmrun: str) -> None: + monkeypatch.setattr( + subprocess, + "run", + lambda *a, **kw: _make_completed(255, stderr="Error: The virtual machine is not powered on"), + ) + backend = WorkstationVmrunBackend(vmrun_path=fake_vmrun) + # Should NOT raise — VM already off is treated as success. + backend.stop(VmHandle("x.vmx")) + + +def test_get_ip_returns_none_on_failure(monkeypatch: pytest.MonkeyPatch, fake_vmrun: str) -> None: + monkeypatch.setattr( + subprocess, "run", lambda *a, **kw: _make_completed(1, stderr="Tools not running") + ) + backend = WorkstationVmrunBackend(vmrun_path=fake_vmrun) + assert backend.get_ip(VmHandle("x.vmx")) is None + + +def test_get_ip_returns_address(monkeypatch: pytest.MonkeyPatch, fake_vmrun: str) -> None: + monkeypatch.setattr(subprocess, "run", lambda *a, **kw: _make_completed(0, stdout="192.168.79.42\n")) + backend = WorkstationVmrunBackend(vmrun_path=fake_vmrun) + assert backend.get_ip(VmHandle("x.vmx")) == "192.168.79.42" + + +def test_list_snapshots_strips_header(monkeypatch: pytest.MonkeyPatch, fake_vmrun: str) -> None: + monkeypatch.setattr( + subprocess, + "run", + lambda *a, **kw: _make_completed(0, stdout="Total snapshots: 2\nBaseClean\nDirty\n"), + ) + backend = WorkstationVmrunBackend(vmrun_path=fake_vmrun) + assert backend.list_snapshots(VmHandle("x.vmx")) == ["BaseClean", "Dirty"] + + +def test_is_running_uses_list_not_getip( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path, fake_vmrun: str +) -> None: + """Regression for AGENTS.md error #10.""" + vmx = tmp_path / "job1" / "job1.vmx" + vmx.parent.mkdir() + vmx.write_text("") + + calls: list[str] = [] + + def fake_run(cmd: list[str], **_kwargs: object) -> Any: + calls.append(cmd[3]) # operation name + if cmd[3] == "list": + return _make_completed( + 0, stdout=f"Total running VMs: 1\n{vmx}\n" + ) + raise AssertionError(f"Unexpected vmrun op: {cmd[3]}") + + monkeypatch.setattr(subprocess, "run", fake_run) + backend = WorkstationVmrunBackend(vmrun_path=fake_vmrun) + assert backend.is_running(VmHandle(str(vmx))) is True + assert calls == ["list"] + assert "getGuestIPAddress" not in calls + + +def test_is_running_returns_false_when_absent( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path, fake_vmrun: str +) -> None: + vmx = tmp_path / "job1.vmx" + vmx.write_text("") + monkeypatch.setattr( + subprocess, "run", lambda *a, **kw: _make_completed(0, stdout="Total running VMs: 0\n") + ) + backend = WorkstationVmrunBackend(vmrun_path=fake_vmrun) + assert backend.is_running(VmHandle(str(vmx))) is False