Files
local-ci-cd-system/tests/python/test_commands_wait.py
T
Simone 80f6661ad5 feat(a2): port leaf PS scripts to ci_orchestrator CLI
Implements Phase A2 of plans/implementation-plan-A-B.md:

- commands/wait.py    -> wait-ready (replaces Wait-VMReady.ps1)

- commands/vm.py      -> vm remove + vm cleanup (replaces Remove-BuildVM.ps1, Cleanup-OrphanedBuildVMs.ps1); cleanup accepts injected VmBackend (Phase C ESXi hook preserved)

- commands/monitor.py -> monitor disk + monitor runner (replaces Watch-DiskSpace.ps1, Watch-RunnerHealth.ps1)

- commands/report.py  -> report job (replaces Get-CIJobSummary.ps1)

Each PS script reduced to a 3-line shim that delegates to the Python CLI and preserves \0.

Tests: 69 pytest cases across new test_commands_*.py modules; ruff clean, mypy --strict clean, coverage 74.5% (>=70 gate).
2026-05-14 18:34:03 +02:00

190 lines
5.7 KiB
Python

"""Tests for ``ci_orchestrator wait-ready``.
Migrated from Pester ``tests/Wait-VMReady.Tests.ps1`` (now removed).
Preserves the negative cases:
* missing/invalid VMX → command surfaces a backend error path
* never-becomes-ready → exit 2 with timeout message
* IP override skips the get_ip polling step
"""
from __future__ import annotations
from typing import Any
import pytest
from click.testing import CliRunner
import ci_orchestrator.commands.wait as wait_module
from ci_orchestrator.__main__ import cli
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
self.running_calls = 0
def is_running(self, _h: VmHandle) -> bool:
self.running_calls += 1
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 _BrokenTransport(_FakeTransport):
def is_ready(self) -> bool:
return False
class _FakeStore:
def get(self, _target: str) -> Credential:
return Credential("user", "pwd")
def _patch(monkeypatch: pytest.MonkeyPatch, backend: _FakeBackend, transport: type = _FakeTransport) -> None:
monkeypatch.setattr(wait_module, "load_backend", lambda _cfg: backend)
monkeypatch.setattr(wait_module, "KeyringCredentialStore", lambda: _FakeStore())
monkeypatch.setattr(wait_module, "WinRmTransport", transport)
monkeypatch.setattr(wait_module, "SshTransport", transport)
monkeypatch.setattr(wait_module.time, "sleep", lambda _s: None)
# ── happy paths ────────────────────────────────────────────────────────────
def test_wait_ready_windows_via_transport_alias(monkeypatch: pytest.MonkeyPatch) -> None:
"""``--transport WinRM`` is a PS-shim alias of ``--guest-os windows``."""
_patch(monkeypatch, _FakeBackend())
result = CliRunner().invoke(
cli,
["wait-ready", "--vmx", "x.vmx", "--transport", "WinRM", "--timeout", "1"],
)
assert result.exit_code == 0, result.output
assert "WinRM ready" in result.output
def test_wait_ready_skip_ping_is_no_op(monkeypatch: pytest.MonkeyPatch) -> None:
_patch(monkeypatch, _FakeBackend())
result = CliRunner().invoke(
cli,
["wait-ready", "--vmx", "x.vmx", "--skip-ping", "--timeout", "1"],
)
assert result.exit_code == 0, result.output
def test_wait_ready_ip_override_skips_get_ip(monkeypatch: pytest.MonkeyPatch) -> None:
"""When --ip-address is provided, the backend is not polled for an IP."""
backend = _FakeBackend(running=True, ip=None) # would time out at IP step
_patch(monkeypatch, backend)
result = CliRunner().invoke(
cli,
[
"wait-ready",
"--vmx",
"x.vmx",
"--ip-address",
"10.0.0.55",
"--timeout",
"1",
],
)
assert result.exit_code == 0, result.output
assert "10.0.0.55" in result.output
def test_wait_ready_linux_via_ssh_alias(monkeypatch: pytest.MonkeyPatch) -> None:
_patch(monkeypatch, _FakeBackend())
result = CliRunner().invoke(
cli,
["wait-ready", "--vmx", "x.vmx", "--transport", "SSH", "--timeout", "1"],
)
assert result.exit_code == 0, result.output
assert "SSH ready" in result.output
# ── failure paths (Pester migration) ───────────────────────────────────────
def test_wait_ready_timeout_when_vm_never_running(monkeypatch: pytest.MonkeyPatch) -> None:
_patch(monkeypatch, _FakeBackend(running=False))
result = CliRunner().invoke(
cli,
[
"wait-ready",
"--vmx",
"x.vmx",
"--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(monkeypatch, _FakeBackend(running=True, ip=None))
result = CliRunner().invoke(
cli,
[
"wait-ready",
"--vmx",
"x.vmx",
"--timeout",
"0.01",
"--poll-interval",
"0.001",
],
)
assert result.exit_code == 3
assert "timeout waiting for guest IP" in result.output
def test_wait_ready_timeout_transport_never_ready(monkeypatch: pytest.MonkeyPatch) -> None:
_patch(monkeypatch, _FakeBackend(), transport=_BrokenTransport)
result = CliRunner().invoke(
cli,
[
"wait-ready",
"--vmx",
"x.vmx",
"--ip-address",
"10.0.0.55",
"--timeout",
"0.01",
"--poll-interval",
"0.001",
],
)
assert result.exit_code == 4
assert "transport to be ready" in result.output
def test_wait_ready_invalid_transport_value(monkeypatch: pytest.MonkeyPatch) -> None:
_patch(monkeypatch, _FakeBackend())
result = CliRunner().invoke(
cli,
["wait-ready", "--vmx", "x.vmx", "--transport", "bogus"],
)
assert result.exit_code != 0
assert "Unknown" in result.output or "bogus" in result.output