239 lines
7.4 KiB
Python
239 lines
7.4 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
|
|
|
|
|
|
# ── helpers (wait_module._wait_running / _wait_for_ip / branches) ───────────────────
|
|
|
|
|
|
from ci_orchestrator.backends.errors import BackendError # noqa: E402
|
|
|
|
|
|
class _ErrorBackend:
|
|
def __init__(self) -> None:
|
|
self.calls = 0
|
|
|
|
def is_running(self, _h: VmHandle) -> bool:
|
|
self.calls += 1
|
|
raise BackendError("probe boom")
|
|
|
|
def get_ip(self, _h: VmHandle, timeout: float = 0.0) -> str | None:
|
|
raise BackendError("ip boom")
|
|
|
|
|
|
def test_wait_running_swallows_backend_error_and_returns_false(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
monkeypatch.setattr(wait_module.time, "sleep", lambda _s: None)
|
|
monkeypatch.setattr(wait_module.time, "monotonic", lambda: 0.0)
|
|
backend = _ErrorBackend()
|
|
# deadline already passed after first iteration
|
|
seq = iter([0.0, 1.0])
|
|
monkeypatch.setattr(wait_module.time, "monotonic", lambda: next(seq))
|
|
ok = wait_module._wait_running(backend, VmHandle("x"), deadline=0.5, poll=0.001) # type: ignore[arg-type]
|
|
assert ok is False
|
|
assert backend.calls == 1
|
|
|
|
|
|
def test_wait_for_ip_returns_none_on_backend_error(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
monkeypatch.setattr(wait_module.time, "sleep", lambda _s: None)
|
|
seq = iter([0.0, 1.0])
|
|
monkeypatch.setattr(wait_module.time, "monotonic", lambda: next(seq))
|
|
backend = _ErrorBackend()
|
|
ip = wait_module._wait_for_ip(backend, VmHandle("x"), deadline=0.5, poll=0.001) # type: ignore[arg-type]
|
|
assert ip is None
|
|
|
|
|
|
def test_normalise_guest_os_rejects_unknown() -> None:
|
|
import click
|
|
with pytest.raises(click.BadParameter):
|
|
wait_module._normalise_guest_os("freebsd")
|