Phase A1: bootstrap Python ci_orchestrator package
- pyproject.toml (hatchling) with ruff/mypy/pytest dev deps - src/ci_orchestrator/: config, credentials, backends (Protocol + WorkstationVmrunBackend), transport (WinRM via pypsrp, SSH via paramiko) - CLI entry point with PoC 'wait-ready' subcommand (Windows + Linux guests) - tests/python/: 35 unit tests, 83% coverage, all backends/transport mocked - gitea/workflows/lint.yml: new 'python' job (ruff + mypy --strict + pytest --cov-fail-under=70) - config.example.toml + README setup section + .gitignore Python entries Neutral VmBackend Protocol prepared for Phase C ESXi extension. See plans/implementation-plan-A-B.md step A1.
This commit is contained in:
@@ -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))
|
||||
@@ -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
|
||||
@@ -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"
|
||||
@@ -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")
|
||||
@@ -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)))
|
||||
@@ -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")
|
||||
@@ -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")
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user