2122 lines
73 KiB
Python
2122 lines
73 KiB
Python
"""Tests for ``ci_orchestrator template`` sub-commands."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import time
|
|
import types
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock
|
|
|
|
import click
|
|
import pytest
|
|
from click.testing import CliRunner
|
|
|
|
import ci_orchestrator.commands.template as tmpl_mod
|
|
from ci_orchestrator.__main__ import cli
|
|
from ci_orchestrator.transport.winrm import WinRmResult
|
|
|
|
# ── fixtures ──────────────────────────────────────────────────────────────────
|
|
|
|
|
|
@pytest.fixture()
|
|
def template_dir(tmp_path: Path) -> Path:
|
|
"""A fake template directory with a couple of files."""
|
|
t = tmp_path / "templates" / "WinBuild2025"
|
|
t.mkdir(parents=True)
|
|
(t / "WinBuild2025.vmx").write_text("vmx content")
|
|
(t / "WinBuild2025.vmdk").write_bytes(b"\x00" * 1024)
|
|
return t
|
|
|
|
|
|
@pytest.fixture()
|
|
def backup_dir(tmp_path: Path) -> Path:
|
|
d = tmp_path / "backups"
|
|
d.mkdir()
|
|
return d
|
|
|
|
|
|
def _fake_compress(source: Path, archive: Path) -> None:
|
|
"""Simulate 7z by writing a small placeholder file."""
|
|
archive.write_bytes(b"fake-7z-archive")
|
|
|
|
|
|
@pytest.fixture()
|
|
def _no_systemctl(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Prevent real systemctl and 7z calls in tests."""
|
|
monkeypatch.setattr(tmpl_mod, "_stop_runner", lambda **_kw: False)
|
|
monkeypatch.setattr(tmpl_mod, "_start_runner", lambda **_kw: None)
|
|
monkeypatch.setattr(tmpl_mod, "_compress_7z", _fake_compress)
|
|
|
|
|
|
# ── basic backup ─────────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_backup_creates_timestamped_archive(
|
|
template_dir: Path, backup_dir: Path, _no_systemctl: None
|
|
) -> None:
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
[
|
|
"template",
|
|
"backup",
|
|
"--template-path",
|
|
str(template_dir),
|
|
"--backup-base-dir",
|
|
str(backup_dir),
|
|
"--skip-runner-stop",
|
|
],
|
|
)
|
|
assert result.exit_code == 0, result.output
|
|
archives = list(backup_dir.iterdir())
|
|
assert len(archives) == 1
|
|
assert archives[0].name.startswith("Template_")
|
|
assert archives[0].suffix == ".7z"
|
|
|
|
|
|
def test_backup_missing_template_raises(
|
|
backup_dir: Path, tmp_path: Path, _no_systemctl: None
|
|
) -> None:
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
[
|
|
"template",
|
|
"backup",
|
|
"--template-path",
|
|
str(tmp_path / "nonexistent"),
|
|
"--backup-base-dir",
|
|
str(backup_dir),
|
|
"--skip-runner-stop",
|
|
],
|
|
)
|
|
assert result.exit_code != 0
|
|
assert "not found" in result.output
|
|
|
|
|
|
# ── --all-templates ───────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_all_templates_backs_up_every_subdir(
|
|
tmp_path: Path, backup_dir: Path, monkeypatch: pytest.MonkeyPatch, _no_systemctl: None
|
|
) -> None:
|
|
tmpl_root = tmp_path / "templates"
|
|
for name in ("WinBuild2025", "WinBuild2022", "LinuxBuild2404"):
|
|
d = tmpl_root / name
|
|
d.mkdir(parents=True)
|
|
(d / f"{name}.vmx").write_text("vmx")
|
|
|
|
monkeypatch.setenv("CI_TEMPLATES", str(tmpl_root))
|
|
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
[
|
|
"template",
|
|
"backup",
|
|
"--all-templates",
|
|
"--backup-base-dir",
|
|
str(backup_dir),
|
|
"--skip-runner-stop",
|
|
],
|
|
)
|
|
assert result.exit_code == 0, result.output
|
|
names = {f.name for f in backup_dir.iterdir() if f.suffix == ".7z"}
|
|
assert any("WinBuild2025" in n for n in names)
|
|
assert any("WinBuild2022" in n for n in names)
|
|
assert any("LinuxBuild2404" in n for n in names)
|
|
|
|
|
|
def test_all_templates_empty_dir_raises(
|
|
tmp_path: Path, backup_dir: Path, monkeypatch: pytest.MonkeyPatch, _no_systemctl: None
|
|
) -> None:
|
|
tmpl_root = tmp_path / "templates"
|
|
tmpl_root.mkdir()
|
|
monkeypatch.setenv("CI_TEMPLATES", str(tmpl_root))
|
|
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
[
|
|
"template",
|
|
"backup",
|
|
"--all-templates",
|
|
"--backup-base-dir",
|
|
str(backup_dir),
|
|
"--skip-runner-stop",
|
|
],
|
|
)
|
|
assert result.exit_code != 0
|
|
assert "No template" in result.output
|
|
|
|
|
|
# ── pruning ───────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_prune_keeps_only_keep_count(
|
|
template_dir: Path, backup_dir: Path, _no_systemctl: None
|
|
) -> None:
|
|
for i in range(3):
|
|
old = backup_dir / f"Template_2025010{i}_120000.7z"
|
|
old.write_bytes(b"old-archive")
|
|
mtime = time.time() - (3 - i) * 3600
|
|
os.utime(old, (mtime, mtime))
|
|
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
[
|
|
"template",
|
|
"backup",
|
|
"--template-path",
|
|
str(template_dir),
|
|
"--backup-base-dir",
|
|
str(backup_dir),
|
|
"--keep-count",
|
|
"2",
|
|
"--skip-runner-stop",
|
|
],
|
|
)
|
|
assert result.exit_code == 0, result.output
|
|
remaining = [f for f in backup_dir.iterdir() if f.suffix == ".7z"]
|
|
assert len(remaining) == 2, f"Expected 2 archives, got {len(remaining)}: {[f.name for f in remaining]}"
|
|
assert "Pruning" in result.output
|
|
|
|
|
|
# ── --what-if ────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_what_if_no_copy(template_dir: Path, backup_dir: Path, _no_systemctl: None) -> None:
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
[
|
|
"template",
|
|
"backup",
|
|
"--template-path",
|
|
str(template_dir),
|
|
"--backup-base-dir",
|
|
str(backup_dir),
|
|
"--skip-runner-stop",
|
|
"--what-if",
|
|
],
|
|
)
|
|
assert result.exit_code == 0, result.output
|
|
assert not list(backup_dir.iterdir()), "--what-if must not create anything"
|
|
assert "WhatIf" in result.output
|
|
|
|
|
|
# ── runner stop/start ─────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_stop_runner_called_when_active(
|
|
template_dir: Path, backup_dir: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
stopped: list[bool] = []
|
|
started: list[bool] = []
|
|
|
|
monkeypatch.setattr(tmpl_mod, "_stop_runner", lambda **kw: stopped.append(True) or True)
|
|
monkeypatch.setattr(tmpl_mod, "_start_runner", lambda **kw: started.append(True))
|
|
monkeypatch.setattr(tmpl_mod, "_compress_7z", _fake_compress)
|
|
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
[
|
|
"template",
|
|
"backup",
|
|
"--template-path",
|
|
str(template_dir),
|
|
"--backup-base-dir",
|
|
str(backup_dir),
|
|
],
|
|
)
|
|
assert result.exit_code == 0, result.output
|
|
assert stopped, "_stop_runner should have been called"
|
|
assert started, "_start_runner should have been called"
|
|
|
|
|
|
def test_skip_runner_stop_skips_systemctl(
|
|
template_dir: Path, backup_dir: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
stopped: list[bool] = []
|
|
monkeypatch.setattr(tmpl_mod, "_stop_runner", lambda **kw: stopped.append(True) or True)
|
|
monkeypatch.setattr(tmpl_mod, "_compress_7z", _fake_compress)
|
|
|
|
CliRunner().invoke(
|
|
cli,
|
|
[
|
|
"template",
|
|
"backup",
|
|
"--template-path",
|
|
str(template_dir),
|
|
"--backup-base-dir",
|
|
str(backup_dir),
|
|
"--skip-runner-stop",
|
|
],
|
|
)
|
|
assert not stopped, "--skip-runner-stop should prevent _stop_runner from being called"
|
|
|
|
|
|
# ── unit helpers: _stop_runner / _start_runner ───────────────────────────────
|
|
|
|
|
|
def test_stop_runner_noop_when_inactive(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setattr(
|
|
tmpl_mod.subprocess,
|
|
"run",
|
|
lambda cmd, **_kw: type("R", (), {"stdout": "inactive\n", "returncode": 0})(),
|
|
)
|
|
assert tmpl_mod._stop_runner(what_if=False) is False
|
|
|
|
|
|
def test_stop_runner_returns_true_when_active(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
calls: list[list[str]] = []
|
|
|
|
def _fake_run(cmd: list[str], **_kw: object) -> object:
|
|
calls.append(cmd)
|
|
return type("R", (), {"stdout": "active\n", "returncode": 0})()
|
|
|
|
monkeypatch.setattr(tmpl_mod.subprocess, "run", _fake_run)
|
|
result = tmpl_mod._stop_runner(what_if=False)
|
|
assert result is True
|
|
assert any("stop" in c for c in calls)
|
|
|
|
|
|
def test_stop_runner_what_if_does_not_call_stop(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
calls: list[list[str]] = []
|
|
|
|
def _fake_run(cmd: list[str], **_kw: object) -> object:
|
|
calls.append(cmd)
|
|
return type("R", (), {"stdout": "active\n", "returncode": 0})()
|
|
|
|
monkeypatch.setattr(tmpl_mod.subprocess, "run", _fake_run)
|
|
result = tmpl_mod._stop_runner(what_if=True)
|
|
assert result is True
|
|
assert not any("stop" in c for c in calls), "what_if=True must not call systemctl stop"
|
|
|
|
|
|
# ── _compress_7z args ─────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_compress_7z_uses_mx1(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
|
|
calls: list[list[str]] = []
|
|
|
|
def _fake_run(cmd: list[str], **_kw: object) -> object:
|
|
calls.append(cmd)
|
|
return type("R", (), {"returncode": 0})()
|
|
|
|
monkeypatch.setattr(tmpl_mod.subprocess, "run", _fake_run)
|
|
tmpl_mod._compress_7z(tmp_path / "src", tmp_path / "out.7z")
|
|
assert calls, "subprocess.run should have been called"
|
|
cmd = calls[0]
|
|
assert cmd[0] == "7z"
|
|
assert "-mx=1" in cmd
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
# template prepare-win — helper unit tests
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
|
|
|
|
def test_ps_escape_no_quotes() -> None:
|
|
assert tmpl_mod._ps_escape("hello world") == "hello world"
|
|
|
|
|
|
def test_ps_escape_single_quote() -> None:
|
|
assert tmpl_mod._ps_escape("it's") == "it''s"
|
|
|
|
|
|
def test_ps_escape_multiple_quotes() -> None:
|
|
assert tmpl_mod._ps_escape("it's Tom's") == "it''s Tom''s"
|
|
|
|
|
|
def test_parse_exit_marker_present() -> None:
|
|
assert tmpl_mod._parse_exit_marker("output\nCI_EXITCODE:3010\n") == 3010
|
|
|
|
|
|
def test_parse_exit_marker_absent() -> None:
|
|
assert tmpl_mod._parse_exit_marker("no marker here") == 0
|
|
|
|
|
|
def test_parse_exit_marker_last_line_wins() -> None:
|
|
assert tmpl_mod._parse_exit_marker("CI_EXITCODE:1\nCI_EXITCODE:0") == 0
|
|
|
|
|
|
def test_parse_exit_marker_malformed() -> None:
|
|
assert tmpl_mod._parse_exit_marker("CI_EXITCODE:abc") == 0
|
|
|
|
|
|
def test_wait_tcp_success(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
class _FakeConn:
|
|
def __enter__(self) -> _FakeConn:
|
|
return self
|
|
|
|
def __exit__(self, *_: object) -> None:
|
|
pass
|
|
|
|
monkeypatch.setattr(tmpl_mod.socket, "create_connection", lambda *a, **kw: _FakeConn())
|
|
assert tmpl_mod._wait_tcp("127.0.0.1", 9999, 5.0) is True
|
|
|
|
|
|
def test_wait_tcp_timeout(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
def _refuse(*a: object, **kw: object) -> None:
|
|
raise OSError("refused")
|
|
|
|
monkeypatch.setattr(tmpl_mod.socket, "create_connection", _refuse)
|
|
monkeypatch.setattr(tmpl_mod.time, "sleep", lambda _: None)
|
|
assert tmpl_mod._wait_tcp("127.0.0.1", 9999, 0.0) is False
|
|
|
|
|
|
def test_wait_winrm_success() -> None:
|
|
transport = MagicMock()
|
|
transport.is_ready.return_value = True
|
|
tmpl_mod._wait_winrm(transport, 30) # must not raise
|
|
|
|
|
|
def test_wait_winrm_timeout() -> None:
|
|
transport = MagicMock()
|
|
transport.is_ready.return_value = False
|
|
with pytest.raises(click.ClickException, match="not ready"):
|
|
tmpl_mod._wait_winrm(transport, 0)
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
# template prepare-win — command tests
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
|
|
|
|
def _make_config() -> types.SimpleNamespace:
|
|
return types.SimpleNamespace(vmrun_path=None, ip_pool=None)
|
|
|
|
|
|
def _make_backend(is_running_seq: list[bool]) -> MagicMock:
|
|
b: MagicMock = MagicMock()
|
|
b.is_running.side_effect = is_running_seq
|
|
b.get_ip.return_value = "192.168.1.1"
|
|
b.list_snapshots.return_value = []
|
|
b.vmrun_path = "vmrun"
|
|
return b
|
|
|
|
|
|
def _make_transport() -> MagicMock:
|
|
val_json = json.dumps(
|
|
{
|
|
k: True
|
|
for k in [
|
|
"WinRMRunning", "UserExists", "UserIsAdmin", "UACDisabled",
|
|
"FirewallOff", "DotNetExists", "PythonExists", "MSBuildExists",
|
|
"CIDirsPresent", "StaticIpTask", "StaticIpScript",
|
|
]
|
|
}
|
|
)
|
|
|
|
def _run(script: str = "", **kwargs: object) -> WinRmResult:
|
|
if "ConvertTo-Json" in script:
|
|
return WinRmResult(stdout=val_json, stderr="", returncode=0)
|
|
return WinRmResult(stdout="CI_EXITCODE:0", stderr="", returncode=0)
|
|
|
|
t: MagicMock = MagicMock()
|
|
t.run.side_effect = _run
|
|
t.is_ready.return_value = True
|
|
return t
|
|
|
|
|
|
def _setup_prepare_env(
|
|
tmp_path: Path,
|
|
*,
|
|
with_static_ip: bool = True,
|
|
) -> Path:
|
|
"""Create fake VMX + toolchain script layout under tmp_path."""
|
|
vmx = tmp_path / "WinBuild2025.vmx"
|
|
vmx.write_text("vmx content")
|
|
tmpl_dir = tmp_path / "template"
|
|
tmpl_dir.mkdir()
|
|
(tmpl_dir / "Install-CIToolchain-WinBuild2025.ps1").write_text("# fake toolchain")
|
|
if with_static_ip:
|
|
sip_dir = tmpl_dir / "guest-setup" / "windows"
|
|
sip_dir.mkdir(parents=True)
|
|
(sip_dir / "ci-static-ip.ps1").write_text("# fake static ip")
|
|
return vmx
|
|
|
|
|
|
def test_prepare_win_vmx_not_found(tmp_path: Path) -> None:
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
["template", "prepare-win", "--vmx", str(tmp_path / "missing.vmx")],
|
|
)
|
|
assert result.exit_code != 0
|
|
assert "VMX not found" in result.output
|
|
|
|
|
|
def test_prepare_win_toolchain_not_found(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
vmx = tmp_path / "WinBuild2025.vmx"
|
|
vmx.write_text("vmx")
|
|
monkeypatch.chdir(tmp_path)
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
["template", "prepare-win", "--vmx", str(vmx)],
|
|
)
|
|
assert result.exit_code != 0
|
|
assert "not found" in result.output
|
|
|
|
|
|
def test_prepare_win_explicit_toolchain_not_found(tmp_path: Path) -> None:
|
|
vmx = tmp_path / "WinBuild2025.vmx"
|
|
vmx.write_text("vmx")
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
[
|
|
"template", "prepare-win",
|
|
"--vmx", str(vmx),
|
|
"--toolchain-script", str(tmp_path / "nope.ps1"),
|
|
],
|
|
)
|
|
assert result.exit_code != 0
|
|
assert "not found" in result.output
|
|
|
|
|
|
def _patch_prepare(monkeypatch: pytest.MonkeyPatch, backend: MagicMock, transport: MagicMock) -> None:
|
|
monkeypatch.setattr(tmpl_mod, "load_config", lambda: _make_config())
|
|
monkeypatch.setattr(tmpl_mod, "WorkstationVmrunBackend", lambda **_: backend)
|
|
monkeypatch.setattr(tmpl_mod, "WinRmTransport", lambda **_: transport)
|
|
monkeypatch.setattr(tmpl_mod, "_wait_tcp", lambda *_: True)
|
|
monkeypatch.setattr(tmpl_mod, "_wait_winrm", lambda *_: None)
|
|
monkeypatch.setattr(tmpl_mod.time, "sleep", lambda _: None)
|
|
monkeypatch.setattr(tmpl_mod.subprocess, "run", lambda *a, **_: None)
|
|
|
|
|
|
def test_prepare_win_happy_path(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
vmx = _setup_prepare_env(tmp_path)
|
|
monkeypatch.chdir(tmp_path)
|
|
backend = _make_backend([True, False])
|
|
transport = _make_transport()
|
|
_patch_prepare(monkeypatch, backend, transport)
|
|
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
[
|
|
"template", "prepare-win",
|
|
"--vmx", str(vmx),
|
|
"--ip", "192.168.1.100",
|
|
"--admin-password", "adminpass",
|
|
"--build-password", "buildpass",
|
|
],
|
|
)
|
|
assert result.exit_code == 0, result.output
|
|
assert "Provisioning complete" in result.output
|
|
assert "All post-setup checks passed" in result.output
|
|
|
|
|
|
def test_prepare_win_no_static_ip_warning(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
vmx = _setup_prepare_env(tmp_path, with_static_ip=False)
|
|
monkeypatch.chdir(tmp_path)
|
|
backend = _make_backend([True, False])
|
|
transport = _make_transport()
|
|
_patch_prepare(monkeypatch, backend, transport)
|
|
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
[
|
|
"template", "prepare-win",
|
|
"--vmx", str(vmx),
|
|
"--ip", "192.168.1.100",
|
|
"--admin-password", "adminpass",
|
|
"--build-password", "buildpass",
|
|
],
|
|
)
|
|
assert result.exit_code == 0, result.output
|
|
assert "WARNING" in result.output
|
|
|
|
|
|
def test_prepare_win_vm_started_when_not_running(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
vmx = _setup_prepare_env(tmp_path)
|
|
monkeypatch.chdir(tmp_path)
|
|
backend = _make_backend([False, False]) # not running → start; then powered off
|
|
transport = _make_transport()
|
|
_patch_prepare(monkeypatch, backend, transport)
|
|
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
[
|
|
"template", "prepare-win",
|
|
"--vmx", str(vmx),
|
|
"--ip", "192.168.1.100",
|
|
"--admin-password", "adminpass",
|
|
"--build-password", "buildpass",
|
|
],
|
|
)
|
|
assert result.exit_code == 0, result.output
|
|
backend.start.assert_called_once()
|
|
|
|
|
|
def test_prepare_win_tcp_timeout(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
vmx = _setup_prepare_env(tmp_path)
|
|
monkeypatch.chdir(tmp_path)
|
|
backend = _make_backend([True])
|
|
_patch_prepare(monkeypatch, backend, MagicMock())
|
|
monkeypatch.setattr(tmpl_mod, "_wait_tcp", lambda *_: False)
|
|
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
[
|
|
"template", "prepare-win",
|
|
"--vmx", str(vmx),
|
|
"--ip", "192.168.1.100",
|
|
"--admin-password", "adminpass",
|
|
"--build-password", "buildpass",
|
|
],
|
|
)
|
|
assert result.exit_code != 0
|
|
assert "not reachable" in result.output
|
|
|
|
|
|
def test_prepare_win_toolchain_fails(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
vmx = _setup_prepare_env(tmp_path)
|
|
monkeypatch.chdir(tmp_path)
|
|
backend = _make_backend([True])
|
|
transport: MagicMock = MagicMock()
|
|
transport.is_ready.return_value = True
|
|
transport.run.return_value = WinRmResult(
|
|
stdout="CI_EXITCODE:1", stderr="err", returncode=0
|
|
)
|
|
_patch_prepare(monkeypatch, backend, transport)
|
|
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
[
|
|
"template", "prepare-win",
|
|
"--vmx", str(vmx),
|
|
"--ip", "192.168.1.100",
|
|
"--admin-password", "adminpass",
|
|
"--build-password", "buildpass",
|
|
],
|
|
)
|
|
assert result.exit_code != 0
|
|
assert "failed" in result.output
|
|
|
|
|
|
def test_prepare_win_snapshot_exists_skip(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
vmx = _setup_prepare_env(tmp_path)
|
|
monkeypatch.chdir(tmp_path)
|
|
backend = _make_backend([True, False])
|
|
backend.list_snapshots.return_value = ["BaseClean"]
|
|
transport = _make_transport()
|
|
_patch_prepare(monkeypatch, backend, transport)
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
[
|
|
"template", "prepare-win",
|
|
"--vmx", str(vmx),
|
|
"--ip", "192.168.1.100",
|
|
"--admin-password", "adminpass",
|
|
"--build-password", "buildpass",
|
|
],
|
|
input="N\n",
|
|
)
|
|
assert result.exit_code == 0, result.output
|
|
assert "Keeping existing snapshot" in result.output
|
|
|
|
|
|
def test_prepare_win_snapshot_exists_recreate(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
vmx = _setup_prepare_env(tmp_path)
|
|
monkeypatch.chdir(tmp_path)
|
|
backend = _make_backend([True, False])
|
|
backend.list_snapshots.return_value = ["BaseClean"]
|
|
transport = _make_transport()
|
|
sp_calls: list[list[str]] = []
|
|
monkeypatch.setattr(tmpl_mod, "load_config", lambda: _make_config())
|
|
monkeypatch.setattr(tmpl_mod, "WorkstationVmrunBackend", lambda **_: backend)
|
|
monkeypatch.setattr(tmpl_mod, "WinRmTransport", lambda **_: transport)
|
|
monkeypatch.setattr(tmpl_mod, "_wait_tcp", lambda *_: True)
|
|
monkeypatch.setattr(tmpl_mod, "_wait_winrm", lambda *_: None)
|
|
monkeypatch.setattr(tmpl_mod.time, "sleep", lambda _: None)
|
|
def _sp_run(cmd: list[str], **_: object) -> None:
|
|
sp_calls.append(cmd)
|
|
|
|
monkeypatch.setattr(tmpl_mod.subprocess, "run", _sp_run)
|
|
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
[
|
|
"template", "prepare-win",
|
|
"--vmx", str(vmx),
|
|
"--ip", "192.168.1.100",
|
|
"--admin-password", "adminpass",
|
|
"--build-password", "buildpass",
|
|
],
|
|
input="Y\n",
|
|
)
|
|
assert result.exit_code == 0, result.output
|
|
assert any("deleteSnapshot" in " ".join(c) for c in sp_calls)
|
|
assert any("snapshot" in " ".join(c) and "deleteSnapshot" not in " ".join(c) for c in sp_calls)
|
|
|
|
|
|
def test_prepare_win_recreate_snapshot_flag(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""--recreate-snapshot skips the interactive prompt and deletes+recreates."""
|
|
vmx = _setup_prepare_env(tmp_path)
|
|
monkeypatch.chdir(tmp_path)
|
|
backend = _make_backend([True, False])
|
|
backend.list_snapshots.return_value = ["BaseClean"]
|
|
transport = _make_transport()
|
|
sp_calls: list[list[str]] = []
|
|
_patch_prepare(monkeypatch, backend, transport)
|
|
|
|
def _sp_run(cmd: list[str], **_: object) -> None:
|
|
sp_calls.append(cmd)
|
|
|
|
monkeypatch.setattr(tmpl_mod.subprocess, "run", _sp_run)
|
|
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
[
|
|
"template", "prepare-win",
|
|
"--vmx", str(vmx),
|
|
"--ip", "192.168.1.100",
|
|
"--admin-password", "adminpass",
|
|
"--build-password", "buildpass",
|
|
"--recreate-snapshot",
|
|
],
|
|
)
|
|
assert result.exit_code == 0, result.output
|
|
assert any("deleteSnapshot" in " ".join(c) for c in sp_calls)
|
|
|
|
|
|
def test_prepare_win_skip_flags(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""-SkipCleanup:$true always in loop args; --skip-cleanup suppresses post-loop cleanup call."""
|
|
vmx = _setup_prepare_env(tmp_path, with_static_ip=False)
|
|
monkeypatch.chdir(tmp_path)
|
|
backend = _make_backend([True, False])
|
|
|
|
captured_scripts: list[str] = []
|
|
|
|
def _run(script: str = "", **kwargs: object) -> WinRmResult:
|
|
captured_scripts.append(script)
|
|
if "ConvertTo-Json" in script:
|
|
return WinRmResult(stdout="{}", stderr="", returncode=0)
|
|
return WinRmResult(stdout="CI_EXITCODE:0", stderr="", returncode=0)
|
|
|
|
transport: MagicMock = MagicMock()
|
|
transport.run.side_effect = _run
|
|
transport.is_ready.return_value = True
|
|
|
|
_patch_prepare(monkeypatch, backend, transport)
|
|
|
|
CliRunner().invoke(
|
|
cli,
|
|
[
|
|
"template", "prepare-win",
|
|
"--vmx", str(vmx),
|
|
"--ip", "192.168.1.100",
|
|
"--admin-password", "adminpass",
|
|
"--build-password", "buildpass",
|
|
"--skip-windows-update",
|
|
"--skip-tier2",
|
|
"--skip-cleanup",
|
|
],
|
|
)
|
|
invoke_script = next(
|
|
(s for s in captured_scripts if "SkipWindowsUpdate" in s or "CI_EXITCODE" in s),
|
|
"",
|
|
)
|
|
assert "SkipWindowsUpdate" in invoke_script
|
|
assert "SkipTier2" in invoke_script
|
|
assert "SkipCleanup" in invoke_script
|
|
|
|
|
|
def test_prepare_win_auto_detect_ip(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""No --ip supplied — IP auto-detected via backend.get_ip()."""
|
|
vmx = _setup_prepare_env(tmp_path, with_static_ip=False)
|
|
monkeypatch.chdir(tmp_path)
|
|
backend = _make_backend([True, False])
|
|
backend.get_ip.return_value = "10.0.0.5"
|
|
transport = _make_transport()
|
|
_patch_prepare(monkeypatch, backend, transport)
|
|
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
[
|
|
"template", "prepare-win",
|
|
"--vmx", str(vmx),
|
|
"--admin-password", "adminpass",
|
|
"--build-password", "buildpass",
|
|
],
|
|
)
|
|
assert result.exit_code == 0, result.output
|
|
assert "10.0.0.5" in result.output
|
|
|
|
|
|
def test_prepare_win_auto_detect_ip_fails(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""backend.get_ip() returns None → ClickException."""
|
|
vmx = _setup_prepare_env(tmp_path, with_static_ip=False)
|
|
monkeypatch.chdir(tmp_path)
|
|
backend = _make_backend([True])
|
|
backend.get_ip.return_value = None
|
|
_patch_prepare(monkeypatch, backend, MagicMock())
|
|
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
[
|
|
"template", "prepare-win",
|
|
"--vmx", str(vmx),
|
|
"--admin-password", "adminpass",
|
|
"--build-password", "buildpass",
|
|
],
|
|
)
|
|
assert result.exit_code != 0
|
|
assert "guest IP" in result.output
|
|
|
|
|
|
def test_prepare_win_3010_reboot_then_success(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""Toolchain returns 3010 once, then 0 — reboot loop executes."""
|
|
vmx = _setup_prepare_env(tmp_path, with_static_ip=False)
|
|
monkeypatch.chdir(tmp_path)
|
|
backend = _make_backend([True, False])
|
|
|
|
call_count = [0]
|
|
|
|
def _run(script: str = "", **kwargs: object) -> WinRmResult:
|
|
call_count[0] += 1
|
|
if "ConvertTo-Json" in script:
|
|
return WinRmResult(stdout="{}", stderr="", returncode=0)
|
|
# First invoke → 3010; subsequent → 0
|
|
code = 3010 if call_count[0] <= 2 else 0
|
|
return WinRmResult(stdout=f"CI_EXITCODE:{code}", stderr="", returncode=0)
|
|
|
|
transport: MagicMock = MagicMock()
|
|
transport.run.side_effect = _run
|
|
transport.is_ready.return_value = True
|
|
|
|
_patch_prepare(monkeypatch, backend, transport)
|
|
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
[
|
|
"template", "prepare-win",
|
|
"--vmx", str(vmx),
|
|
"--ip", "192.168.1.100",
|
|
"--admin-password", "adminpass",
|
|
"--build-password", "buildpass",
|
|
],
|
|
)
|
|
assert result.exit_code == 0, result.output
|
|
assert "reboot" in result.output.lower()
|
|
|
|
|
|
def test_prepare_win_transport_error_treated_as_3010(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""TransportError during toolchain run → treated as 3010, reboot, then success."""
|
|
from ci_orchestrator.transport.errors import TransportConnectError
|
|
|
|
vmx = _setup_prepare_env(tmp_path, with_static_ip=False)
|
|
monkeypatch.chdir(tmp_path)
|
|
backend = _make_backend([True, False])
|
|
|
|
raised = [False]
|
|
|
|
def _run(script: str = "", **kwargs: object) -> WinRmResult:
|
|
if not raised[0] and "CI_EXITCODE" in script:
|
|
raised[0] = True
|
|
raise TransportConnectError("VM rebooted")
|
|
if "ConvertTo-Json" in script:
|
|
return WinRmResult(stdout="{}", stderr="", returncode=0)
|
|
return WinRmResult(stdout="CI_EXITCODE:0", stderr="", returncode=0)
|
|
|
|
transport: MagicMock = MagicMock()
|
|
transport.run.side_effect = _run
|
|
transport.is_ready.return_value = True
|
|
|
|
_patch_prepare(monkeypatch, backend, transport)
|
|
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
[
|
|
"template", "prepare-win",
|
|
"--vmx", str(vmx),
|
|
"--ip", "192.168.1.100",
|
|
"--admin-password", "adminpass",
|
|
"--build-password", "buildpass",
|
|
],
|
|
)
|
|
assert result.exit_code == 0, result.output
|
|
assert "Transport error" in result.output
|
|
|
|
|
|
def test_prepare_win_validation_with_failures(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""Validation JSON contains False values → WARNING printed but exit 0."""
|
|
vmx = _setup_prepare_env(tmp_path, with_static_ip=False)
|
|
monkeypatch.chdir(tmp_path)
|
|
backend = _make_backend([True, False])
|
|
|
|
partial_json = json.dumps(
|
|
{k: (k != "MSBuildExists") for k in [
|
|
"WinRMRunning", "UserExists", "UserIsAdmin", "UACDisabled",
|
|
"FirewallOff", "DotNetExists", "PythonExists", "MSBuildExists",
|
|
"CIDirsPresent", "StaticIpTask", "StaticIpScript",
|
|
]}
|
|
)
|
|
|
|
def _run(script: str = "", **kwargs: object) -> WinRmResult:
|
|
if "ConvertTo-Json" in script:
|
|
return WinRmResult(stdout=partial_json, stderr="", returncode=0)
|
|
return WinRmResult(stdout="CI_EXITCODE:0", stderr="", returncode=0)
|
|
|
|
transport: MagicMock = MagicMock()
|
|
transport.run.side_effect = _run
|
|
transport.is_ready.return_value = True
|
|
|
|
_patch_prepare(monkeypatch, backend, transport)
|
|
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
[
|
|
"template", "prepare-win",
|
|
"--vmx", str(vmx),
|
|
"--ip", "192.168.1.100",
|
|
"--admin-password", "adminpass",
|
|
"--build-password", "buildpass",
|
|
],
|
|
)
|
|
assert result.exit_code == 0, result.output
|
|
assert "WARNING" in result.output
|
|
assert "MSBuildExists" in result.output
|
|
|
|
|
|
def test_prepare_win_vm_shutdown_timeout(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""VM never powers off within 120 s → ClickException."""
|
|
vmx = _setup_prepare_env(tmp_path, with_static_ip=False)
|
|
monkeypatch.chdir(tmp_path)
|
|
|
|
backend = _make_backend([True]) # always "running"
|
|
backend.is_running.side_effect = None # override side_effect
|
|
backend.is_running.return_value = True # always reports as running
|
|
|
|
transport = _make_transport()
|
|
_patch_prepare(monkeypatch, backend, transport)
|
|
|
|
# Patch monotonic so the 120 s deadline expires immediately
|
|
_orig = tmpl_mod.time.monotonic
|
|
call_n = [0]
|
|
|
|
def _fast_mono() -> float:
|
|
call_n[0] += 1
|
|
# After the 5th call (inside shutdown-wait loop) return a huge number
|
|
return _orig() if call_n[0] < 20 else _orig() + 200
|
|
|
|
monkeypatch.setattr(tmpl_mod.time, "monotonic", _fast_mono)
|
|
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
[
|
|
"template", "prepare-win",
|
|
"--vmx", str(vmx),
|
|
"--ip", "192.168.1.100",
|
|
"--admin-password", "adminpass",
|
|
"--build-password", "buildpass",
|
|
],
|
|
)
|
|
assert result.exit_code != 0
|
|
assert "power off" in result.output
|
|
|
|
|
|
def test_prepare_win_store_credential(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""--store-credential calls keyring.set_password with the two-entry scheme."""
|
|
vmx = _setup_prepare_env(tmp_path, with_static_ip=False)
|
|
monkeypatch.chdir(tmp_path)
|
|
backend = _make_backend([True, False])
|
|
transport = _make_transport()
|
|
_patch_prepare(monkeypatch, backend, transport)
|
|
|
|
stored: list[tuple[str, str, str]] = []
|
|
|
|
fake_keyring = types.SimpleNamespace(
|
|
set_password=lambda svc, user, pw: stored.append((svc, user, pw))
|
|
)
|
|
monkeypatch.setattr(tmpl_mod, "keyring", fake_keyring, raising=False)
|
|
|
|
# Patch the import inside the function so it returns our fake module.
|
|
import builtins
|
|
real_import = builtins.__import__
|
|
|
|
def _fake_import(name: str, *args: object, **kwargs: object) -> object:
|
|
if name == "keyring":
|
|
return fake_keyring
|
|
return real_import(name, *args, **kwargs) # type: ignore[arg-type]
|
|
|
|
monkeypatch.setattr(builtins, "__import__", _fake_import)
|
|
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
[
|
|
"template", "prepare-win",
|
|
"--vmx", str(vmx),
|
|
"--ip", "192.168.1.100",
|
|
"--admin-password", "adminpass",
|
|
"--build-password", "buildpass",
|
|
"--store-credential",
|
|
"--credential-target", "TestTarget",
|
|
],
|
|
)
|
|
assert result.exit_code == 0, result.output
|
|
assert any(s[0] == "TestTarget:meta" for s in stored), "username meta-entry not stored"
|
|
assert any(s[0] == "TestTarget" for s in stored), "password entry not stored"
|
|
assert "Credentials stored" in result.output
|
|
|
|
|
|
def test_prepare_win_store_credential_default_target(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""Default credential target is BuildVMGuest."""
|
|
vmx = _setup_prepare_env(tmp_path, with_static_ip=False)
|
|
monkeypatch.chdir(tmp_path)
|
|
backend = _make_backend([True, False])
|
|
transport = _make_transport()
|
|
_patch_prepare(monkeypatch, backend, transport)
|
|
|
|
stored: list[tuple[str, str, str]] = []
|
|
|
|
import builtins
|
|
real_import = builtins.__import__
|
|
|
|
def _fake_import(name: str, *args: object, **kwargs: object) -> object:
|
|
if name == "keyring":
|
|
return types.SimpleNamespace(
|
|
set_password=lambda svc, user, pw: stored.append((svc, user, pw))
|
|
)
|
|
return real_import(name, *args, **kwargs) # type: ignore[arg-type]
|
|
|
|
monkeypatch.setattr(builtins, "__import__", _fake_import)
|
|
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
[
|
|
"template", "prepare-win",
|
|
"--vmx", str(vmx),
|
|
"--ip", "192.168.1.100",
|
|
"--admin-password", "adminpass",
|
|
"--build-password", "buildpass",
|
|
"--store-credential",
|
|
],
|
|
)
|
|
assert result.exit_code == 0, result.output
|
|
assert any(s[0] == "BuildVMGuest:meta" for s in stored)
|
|
assert any(s[0] == "BuildVMGuest" for s in stored)
|
|
|
|
|
|
def test_prepare_win_cleanup_runs_after_loop(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""CleanupOnly invocation fires after WU loop exits 0 (default, no --skip-cleanup)."""
|
|
vmx = _setup_prepare_env(tmp_path, with_static_ip=False)
|
|
monkeypatch.chdir(tmp_path)
|
|
backend = _make_backend([True, False])
|
|
|
|
captured_scripts: list[str] = []
|
|
|
|
def _run(script: str = "", **kwargs: object) -> WinRmResult:
|
|
captured_scripts.append(script)
|
|
if "ConvertTo-Json" in script:
|
|
return WinRmResult(stdout="{}", stderr="", returncode=0)
|
|
return WinRmResult(stdout="CI_EXITCODE:0", stderr="", returncode=0)
|
|
|
|
transport: MagicMock = MagicMock()
|
|
transport.run.side_effect = _run
|
|
transport.is_ready.return_value = True
|
|
|
|
_patch_prepare(monkeypatch, backend, transport)
|
|
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
[
|
|
"template", "prepare-win",
|
|
"--vmx", str(vmx),
|
|
"--ip", "192.168.1.100",
|
|
"--admin-password", "adminpass",
|
|
"--build-password", "buildpass",
|
|
],
|
|
)
|
|
assert result.exit_code == 0, result.output
|
|
assert any("CleanupOnly:$true" in s for s in captured_scripts), (
|
|
"Expected a -CleanupOnly:$true script after the WU loop"
|
|
)
|
|
assert "Post-update cleanup complete" in result.output
|
|
|
|
|
|
def test_prepare_win_skip_cleanup_no_cleanup_call(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""--skip-cleanup suppresses the post-loop CleanupOnly invocation."""
|
|
vmx = _setup_prepare_env(tmp_path, with_static_ip=False)
|
|
monkeypatch.chdir(tmp_path)
|
|
backend = _make_backend([True, False])
|
|
|
|
captured_scripts: list[str] = []
|
|
|
|
def _run(script: str = "", **kwargs: object) -> WinRmResult:
|
|
captured_scripts.append(script)
|
|
if "ConvertTo-Json" in script:
|
|
return WinRmResult(stdout="{}", stderr="", returncode=0)
|
|
return WinRmResult(stdout="CI_EXITCODE:0", stderr="", returncode=0)
|
|
|
|
transport: MagicMock = MagicMock()
|
|
transport.run.side_effect = _run
|
|
transport.is_ready.return_value = True
|
|
|
|
_patch_prepare(monkeypatch, backend, transport)
|
|
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
[
|
|
"template", "prepare-win",
|
|
"--vmx", str(vmx),
|
|
"--ip", "192.168.1.100",
|
|
"--admin-password", "adminpass",
|
|
"--build-password", "buildpass",
|
|
"--skip-cleanup",
|
|
],
|
|
)
|
|
assert result.exit_code == 0, result.output
|
|
assert not any("CleanupOnly:$true" in s for s in captured_scripts), (
|
|
"--skip-cleanup must suppress the -CleanupOnly:$true call"
|
|
)
|
|
|
|
|
|
def test_prepare_win_cleanup_only_fails(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""Cleanup-only step returning nonzero exits the command with an error."""
|
|
vmx = _setup_prepare_env(tmp_path, with_static_ip=False)
|
|
monkeypatch.chdir(tmp_path)
|
|
backend = _make_backend([True])
|
|
|
|
def _run(script: str = "", **kwargs: object) -> WinRmResult:
|
|
if "CleanupOnly" in script:
|
|
return WinRmResult(stdout="CI_EXITCODE:1", stderr="cleanup error", returncode=0)
|
|
if "ConvertTo-Json" in script:
|
|
return WinRmResult(stdout="{}", stderr="", returncode=0)
|
|
return WinRmResult(stdout="CI_EXITCODE:0", stderr="", returncode=0)
|
|
|
|
transport: MagicMock = MagicMock()
|
|
transport.run.side_effect = _run
|
|
transport.is_ready.return_value = True
|
|
|
|
_patch_prepare(monkeypatch, backend, transport)
|
|
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
[
|
|
"template", "prepare-win",
|
|
"--vmx", str(vmx),
|
|
"--ip", "192.168.1.100",
|
|
"--admin-password", "adminpass",
|
|
"--build-password", "buildpass",
|
|
],
|
|
)
|
|
assert result.exit_code != 0
|
|
assert "Disk cleanup failed" in result.output
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
# template deploy-linux — helper unit tests
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
|
|
from ci_orchestrator.config import BackendConfig, Config, Paths # noqa: E402
|
|
|
|
|
|
def _deploy_config(tmp_path: Path) -> Config:
|
|
return Config(
|
|
paths=Paths(
|
|
root=tmp_path,
|
|
templates=tmp_path / "templates",
|
|
build_vms=tmp_path / "build-vms",
|
|
artifacts=tmp_path / "artifacts",
|
|
keys=tmp_path / "keys",
|
|
),
|
|
backend=BackendConfig(),
|
|
vmrun_path="/usr/bin/vmrun",
|
|
ssh_key_path=tmp_path / "keys" / "ci_linux",
|
|
)
|
|
|
|
|
|
def _make_keys(keys_dir: Path) -> Path:
|
|
keys_dir.mkdir(parents=True, exist_ok=True)
|
|
key = keys_dir / "ci_linux"
|
|
key.write_text("FAKE PRIVATE KEY")
|
|
(keys_dir / "ci_linux.pub").write_text("ssh-rsa AAAA test@host")
|
|
return key
|
|
|
|
|
|
def _make_vmdk_cache(root: Path) -> Path:
|
|
iso_dir = root / "iso"
|
|
iso_dir.mkdir(exist_ok=True)
|
|
vmdk = iso_dir / "noble-server-cloudimg-amd64.vmdk"
|
|
vmdk.write_bytes(b"\x00" * 100)
|
|
return vmdk
|
|
|
|
|
|
@pytest.fixture()
|
|
def deploy_env(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> dict:
|
|
ssh_key = _make_keys(tmp_path / "keys")
|
|
vmdk_cache = _make_vmdk_cache(tmp_path)
|
|
vm_dir = tmp_path / "templates" / "LinuxBuild2404"
|
|
vm_dir.mkdir(parents=True)
|
|
config = _deploy_config(tmp_path)
|
|
monkeypatch.setattr(tmpl_mod, "load_config", lambda: config)
|
|
monkeypatch.setattr(tmpl_mod, "_find_vdisk_manager", lambda _: "/usr/bin/vmware-vdiskmanager")
|
|
monkeypatch.setattr(
|
|
tmpl_mod.shutil, "which",
|
|
lambda n: f"/usr/bin/{n}" if n in ("vmrun", "vmware-vdiskmanager", "genisoimage") else None,
|
|
)
|
|
monkeypatch.setattr(tmpl_mod.time, "sleep", lambda _: None)
|
|
return {
|
|
"tmp_path": tmp_path,
|
|
"ssh_key": ssh_key,
|
|
"vmdk_cache": vmdk_cache,
|
|
"vm_dir": vm_dir,
|
|
"vmx": vm_dir / "LinuxBuild2404.vmx",
|
|
}
|
|
|
|
|
|
# ── _patch_vmx_key ────────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_patch_vmx_key_replaces_existing(tmp_path: Path) -> None:
|
|
vmx = tmp_path / "t.vmx"
|
|
vmx.write_text('ide1:0.present = "TRUE"\nother = "val"\n', encoding="ascii")
|
|
tmpl_mod._patch_vmx_key(vmx, "ide1:0.present", "FALSE")
|
|
content = vmx.read_text()
|
|
assert 'ide1:0.present = "FALSE"' in content
|
|
assert '"TRUE"' not in content
|
|
|
|
|
|
def test_patch_vmx_key_appends_when_absent(tmp_path: Path) -> None:
|
|
vmx = tmp_path / "t.vmx"
|
|
vmx.write_text('existing = "val"\n', encoding="ascii")
|
|
tmpl_mod._patch_vmx_key(vmx, "newkey", "nv")
|
|
content = vmx.read_text()
|
|
assert 'newkey = "nv"' in content
|
|
assert 'existing = "val"' in content
|
|
|
|
|
|
# ── _sha256_file ──────────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_sha256_file_correct(tmp_path: Path) -> None:
|
|
import hashlib
|
|
data = b"sha256 test data"
|
|
f = tmp_path / "data.bin"
|
|
f.write_bytes(data)
|
|
assert tmpl_mod._sha256_file(f) == hashlib.sha256(data).hexdigest().upper()
|
|
|
|
|
|
# ── _find_vdisk_manager ───────────────────────────────────────────────────────
|
|
|
|
|
|
def test_find_vdisk_manager_uses_which(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setattr(tmpl_mod.shutil, "which", lambda n: "/usr/bin/vmware-vdiskmanager")
|
|
assert tmpl_mod._find_vdisk_manager(None) == "/usr/bin/vmware-vdiskmanager"
|
|
|
|
|
|
def test_find_vdisk_manager_explicit_missing(tmp_path: Path) -> None:
|
|
with pytest.raises(click.ClickException):
|
|
tmpl_mod._find_vdisk_manager(str(tmp_path / "nonexistent"))
|
|
|
|
|
|
def test_find_vdisk_manager_not_in_path(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setattr(tmpl_mod.shutil, "which", lambda n: None)
|
|
with pytest.raises(click.ClickException, match="PATH"):
|
|
tmpl_mod._find_vdisk_manager(None)
|
|
|
|
|
|
# ── _build_seed_iso ───────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_build_seed_iso_calls_genisoimage(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
staging = tmp_path / "staging"
|
|
staging.mkdir()
|
|
iso = tmp_path / "seed.iso"
|
|
calls: list[list[str]] = []
|
|
|
|
import subprocess as _sub
|
|
|
|
def fake_run(cmd: list[str], **kw: object) -> _sub.CompletedProcess[bytes]:
|
|
calls.append(list(cmd))
|
|
iso.write_bytes(b"\x00" * 8192)
|
|
return _sub.CompletedProcess(cmd, 0)
|
|
|
|
monkeypatch.setattr(tmpl_mod.subprocess, "run", fake_run)
|
|
monkeypatch.setattr(
|
|
tmpl_mod.shutil, "which",
|
|
lambda n: "/usr/bin/genisoimage" if n == "genisoimage" else None,
|
|
)
|
|
tmpl_mod._build_seed_iso(staging, iso)
|
|
assert calls and "genisoimage" in calls[0][0]
|
|
assert "cidata" in calls[0]
|
|
|
|
|
|
def test_build_seed_iso_no_tool_raises(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
monkeypatch.setattr(tmpl_mod.shutil, "which", lambda n: None)
|
|
with pytest.raises(click.ClickException, match="genisoimage"):
|
|
tmpl_mod._build_seed_iso(tmp_path, tmp_path / "seed.iso")
|
|
|
|
|
|
# ── _read_guestinfo_ip ────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_read_guestinfo_ip_valid(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
import subprocess as _sub
|
|
|
|
monkeypatch.setattr(
|
|
tmpl_mod.subprocess, "run",
|
|
lambda cmd, **kw: _sub.CompletedProcess(cmd, 0, "10.0.0.5\n", ""),
|
|
)
|
|
assert tmpl_mod._read_guestinfo_ip("/vmrun", "/vm.vmx") == "10.0.0.5"
|
|
|
|
|
|
def test_read_guestinfo_ip_empty(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
import subprocess as _sub
|
|
|
|
monkeypatch.setattr(
|
|
tmpl_mod.subprocess, "run",
|
|
lambda cmd, **kw: _sub.CompletedProcess(cmd, 0, "", ""),
|
|
)
|
|
assert tmpl_mod._read_guestinfo_ip("/vmrun", "/vm.vmx") is None
|
|
|
|
|
|
def test_read_guestinfo_ip_timeout(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
import subprocess as _sub
|
|
|
|
def raise_timeout(cmd: list[str], **kw: object) -> _sub.CompletedProcess[str]:
|
|
raise _sub.TimeoutExpired(cmd, 10)
|
|
|
|
monkeypatch.setattr(tmpl_mod.subprocess, "run", raise_timeout)
|
|
assert tmpl_mod._read_guestinfo_ip("/vmrun", "/vm.vmx") is None
|
|
|
|
|
|
# ── _poll_guest_ip_deploy ─────────────────────────────────────────────────────
|
|
|
|
|
|
def test_poll_guest_ip_uses_guestinfo_first(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""If ci-report-ip has written guestinfo.ci-ip, return it without getGuestIPAddress."""
|
|
called: list[list[str]] = []
|
|
import subprocess as _sub
|
|
|
|
def fake_run(cmd: list[str], **kw: object) -> _sub.CompletedProcess[str]:
|
|
called.append(list(cmd))
|
|
if "readVariable" in cmd:
|
|
return _sub.CompletedProcess(cmd, 0, "10.0.1.50", "")
|
|
return _sub.CompletedProcess(cmd, 0, "", "")
|
|
|
|
monkeypatch.setattr(tmpl_mod.subprocess, "run", fake_run)
|
|
monkeypatch.setattr(tmpl_mod.time, "sleep", lambda _: None)
|
|
result = tmpl_mod._poll_guest_ip_deploy("/vmrun", "/vm.vmx", 60.0)
|
|
assert result == "10.0.1.50"
|
|
assert not any("getGuestIPAddress" in " ".join(c) for c in called)
|
|
|
|
|
|
def test_poll_guest_ip_returns_valid(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
import subprocess as _sub
|
|
|
|
monkeypatch.setattr(
|
|
tmpl_mod.subprocess, "run",
|
|
lambda cmd, **kw: _sub.CompletedProcess(cmd, 0, "192.168.1.100", ""),
|
|
)
|
|
monkeypatch.setattr(tmpl_mod.time, "sleep", lambda _: None)
|
|
assert tmpl_mod._poll_guest_ip_deploy("/vmrun", "/vm.vmx", 60.0) == "192.168.1.100"
|
|
|
|
|
|
def test_poll_guest_ip_timeout(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
import subprocess as _sub
|
|
|
|
monkeypatch.setattr(
|
|
tmpl_mod.subprocess, "run",
|
|
lambda cmd, **kw: _sub.CompletedProcess(cmd, 0, "Error: not ready", ""),
|
|
)
|
|
monkeypatch.setattr(tmpl_mod.time, "sleep", lambda _: None)
|
|
assert tmpl_mod._poll_guest_ip_deploy("/vmrun", "/vm.vmx", 0.0) is None
|
|
|
|
|
|
# ── _wait_ssh_port_deploy ─────────────────────────────────────────────────────
|
|
|
|
|
|
def test_wait_ssh_port_success(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
ctx = MagicMock()
|
|
ctx.__enter__ = lambda s: s
|
|
ctx.__exit__ = lambda s, *a: False
|
|
monkeypatch.setattr(tmpl_mod.socket, "create_connection", lambda *a, **kw: ctx)
|
|
monkeypatch.setattr(tmpl_mod.time, "sleep", lambda _: None)
|
|
assert tmpl_mod._wait_ssh_port_deploy("192.168.1.100", 22, 10.0, 1.0) is True
|
|
|
|
|
|
def test_wait_ssh_port_timeout(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
def _raise(*a: object, **kw: object) -> None:
|
|
raise OSError("refused")
|
|
|
|
monkeypatch.setattr(tmpl_mod.socket, "create_connection", _raise)
|
|
monkeypatch.setattr(tmpl_mod.time, "sleep", lambda _: None)
|
|
assert tmpl_mod._wait_ssh_port_deploy("192.168.1.100", 22, 0.0, 1.0) is False
|
|
|
|
|
|
# ── _verify_cloud_init_done ───────────────────────────────────────────────────
|
|
|
|
|
|
def test_verify_cloud_init_success(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
stdout_m = MagicMock()
|
|
stdout_m.read.return_value = b"OK"
|
|
client_m = MagicMock()
|
|
client_m.exec_command.return_value = (MagicMock(), stdout_m, MagicMock())
|
|
monkeypatch.setattr(tmpl_mod.paramiko, "SSHClient", lambda: client_m)
|
|
monkeypatch.setattr(tmpl_mod.paramiko, "AutoAddPolicy", MagicMock)
|
|
monkeypatch.setattr(tmpl_mod.time, "sleep", lambda _: None)
|
|
assert tmpl_mod._verify_cloud_init_done("host", "user", "/key", 30.0) is True
|
|
|
|
|
|
def test_verify_cloud_init_exception_retries(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
calls = [0]
|
|
stdout_m = MagicMock()
|
|
stdout_m.read.return_value = b"OK"
|
|
client_m = MagicMock()
|
|
|
|
def _connect(*a: object, **kw: object) -> None:
|
|
calls[0] += 1
|
|
if calls[0] == 1:
|
|
raise OSError("not yet")
|
|
|
|
client_m.connect.side_effect = _connect
|
|
client_m.exec_command.return_value = (MagicMock(), stdout_m, MagicMock())
|
|
monkeypatch.setattr(tmpl_mod.paramiko, "SSHClient", lambda: client_m)
|
|
monkeypatch.setattr(tmpl_mod.paramiko, "AutoAddPolicy", MagicMock)
|
|
monkeypatch.setattr(tmpl_mod.time, "sleep", lambda _: None)
|
|
# Give enough time budget for 2 tries
|
|
assert tmpl_mod._verify_cloud_init_done("host", "user", "/key", 60.0) is True
|
|
|
|
|
|
def test_verify_cloud_init_timeout(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
client_m = MagicMock()
|
|
client_m.connect.side_effect = OSError("refused")
|
|
monkeypatch.setattr(tmpl_mod.paramiko, "SSHClient", lambda: client_m)
|
|
monkeypatch.setattr(tmpl_mod.paramiko, "AutoAddPolicy", MagicMock)
|
|
monkeypatch.setattr(tmpl_mod.time, "sleep", lambda _: None)
|
|
assert tmpl_mod._verify_cloud_init_done("host", "user", "/key", 0.0) is False
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
# template deploy-linux — CLI tests
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
|
|
|
|
def test_deploy_linux_help() -> None:
|
|
result = CliRunner().invoke(cli, ["template", "deploy-linux", "--help"])
|
|
assert result.exit_code == 0
|
|
assert "Ubuntu" in result.output
|
|
|
|
|
|
def test_deploy_linux_preflight_missing_pub_key(
|
|
deploy_env: dict, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
(deploy_env["ssh_key"].parent / "ci_linux.pub").unlink()
|
|
result = CliRunner().invoke(
|
|
cli, ["template", "deploy-linux", "--vmx", str(deploy_env["vmx"])]
|
|
)
|
|
assert result.exit_code != 0
|
|
assert "ci_linux.pub" in result.output or "public key" in result.output.lower()
|
|
|
|
|
|
def test_deploy_linux_vm_dir_exists_no_force(
|
|
deploy_env: dict, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
result = CliRunner().invoke(
|
|
cli, ["template", "deploy-linux", "--vmx", str(deploy_env["vmx"])]
|
|
)
|
|
assert result.exit_code != 0
|
|
assert "force" in result.output.lower()
|
|
|
|
|
|
def test_deploy_linux_sha256_mismatch(
|
|
deploy_env: dict, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
[
|
|
"template", "deploy-linux",
|
|
"--vmx", str(deploy_env["vmx"]),
|
|
"--force",
|
|
"--vmdk-sha256", "DEADBEEF" + "0" * 56,
|
|
],
|
|
)
|
|
assert result.exit_code != 0
|
|
assert "mismatch" in result.output.lower()
|
|
|
|
|
|
def test_deploy_linux_step3_iso_fail(
|
|
deploy_env: dict, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
import subprocess as _sub
|
|
|
|
def fake_run(cmd: list[str], **kw: object) -> _sub.CompletedProcess[bytes]:
|
|
if "genisoimage" in str(cmd[0]):
|
|
raise _sub.CalledProcessError(1, cmd, b"", b"genisoimage error")
|
|
return _sub.CompletedProcess(cmd, 0)
|
|
|
|
monkeypatch.setattr(tmpl_mod.subprocess, "run", fake_run)
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
[
|
|
"template", "deploy-linux",
|
|
"--vmx", str(deploy_env["vmx"]),
|
|
"--force",
|
|
"--start-from-step", "2",
|
|
],
|
|
)
|
|
assert result.exit_code != 0
|
|
assert "ISO creation failed" in result.output
|
|
|
|
|
|
def test_deploy_linux_step4_convert_fail(
|
|
deploy_env: dict, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
import subprocess as _sub
|
|
|
|
def fake_run(cmd: list[str], **kw: object) -> _sub.CompletedProcess[bytes]:
|
|
if "vmware-vdiskmanager" in str(cmd[0]):
|
|
return _sub.CompletedProcess(cmd, 1, b"", b"convert error")
|
|
return _sub.CompletedProcess(cmd, 0)
|
|
|
|
monkeypatch.setattr(tmpl_mod.subprocess, "run", fake_run)
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
[
|
|
"template", "deploy-linux",
|
|
"--vmx", str(deploy_env["vmx"]),
|
|
"--force",
|
|
"--start-from-step", "4",
|
|
],
|
|
)
|
|
assert result.exit_code != 0
|
|
assert "convert failed" in result.output
|
|
|
|
|
|
def test_deploy_linux_step5_vmx_content(
|
|
deploy_env: dict, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
import subprocess as _sub
|
|
|
|
popen_m = MagicMock()
|
|
popen_m.pid = 1
|
|
monkeypatch.setattr(tmpl_mod.subprocess, "Popen", lambda *a, **kw: popen_m)
|
|
|
|
# vmrun list returns empty so step 6 fails — we can still check VMX written in step 5
|
|
monkeypatch.setattr(
|
|
tmpl_mod.subprocess, "run",
|
|
lambda cmd, **kw: _sub.CompletedProcess(list(cmd), 0, "Total running VMs: 0\n", ""),
|
|
)
|
|
|
|
CliRunner().invoke(
|
|
cli,
|
|
[
|
|
"template", "deploy-linux",
|
|
"--vmx", str(deploy_env["vmx"]),
|
|
"--force",
|
|
"--start-from-step", "5",
|
|
],
|
|
)
|
|
# Step 6 should fail (VM not in list), but VMX should have been written in step 5
|
|
assert deploy_env["vmx"].exists()
|
|
content = deploy_env["vmx"].read_text()
|
|
assert 'guestOS = "ubuntu-64"' in content
|
|
assert 'firmware = "efi"' in content
|
|
assert 'ethernet0.virtualDev = "vmxnet3"' in content
|
|
|
|
|
|
def test_deploy_linux_step5_guestinfo_skipped_when_step2_skipped(
|
|
deploy_env: dict, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
import subprocess as _sub
|
|
|
|
popen_m = MagicMock()
|
|
popen_m.pid = 1
|
|
monkeypatch.setattr(tmpl_mod.subprocess, "Popen", lambda *a, **kw: popen_m)
|
|
monkeypatch.setattr(
|
|
tmpl_mod.subprocess, "run",
|
|
lambda cmd, **kw: _sub.CompletedProcess(list(cmd), 0, "Total running VMs: 0\n", ""),
|
|
)
|
|
|
|
CliRunner().invoke(
|
|
cli,
|
|
[
|
|
"template", "deploy-linux",
|
|
"--vmx", str(deploy_env["vmx"]),
|
|
"--force",
|
|
"--start-from-step", "5", # step 2 was skipped → no b64 strings
|
|
],
|
|
)
|
|
if deploy_env["vmx"].exists():
|
|
content = deploy_env["vmx"].read_text()
|
|
assert "guestinfo.userdata" not in content
|
|
|
|
|
|
def test_deploy_linux_step7_ip_timeout(
|
|
deploy_env: dict, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
import subprocess as _sub
|
|
|
|
popen_m = MagicMock()
|
|
popen_m.pid = 1
|
|
monkeypatch.setattr(tmpl_mod.subprocess, "Popen", lambda *a, **kw: popen_m)
|
|
|
|
vmx_str = str(deploy_env["vmx"])
|
|
monkeypatch.setattr(
|
|
tmpl_mod.subprocess, "run",
|
|
lambda cmd, **kw: _sub.CompletedProcess(
|
|
list(cmd), 0, f"Total running VMs: 1\n{vmx_str}\n", ""
|
|
),
|
|
)
|
|
# patch the poller so it returns None immediately (no 300-second spin loop)
|
|
monkeypatch.setattr(tmpl_mod, "_poll_guest_ip_deploy", lambda *a, **kw: None)
|
|
|
|
deploy_env["vmx"].write_text('dummy = "vmx"', encoding="ascii")
|
|
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
[
|
|
"template", "deploy-linux",
|
|
"--vmx", str(deploy_env["vmx"]),
|
|
"--force",
|
|
"--start-from-step", "6",
|
|
],
|
|
)
|
|
assert result.exit_code != 0
|
|
assert "guest IP" in result.output
|
|
|
|
|
|
def test_deploy_linux_step8_no_guest_ip(
|
|
deploy_env: dict, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
deploy_env["vmx"].write_text('dummy = "vmx"', encoding="ascii")
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
[
|
|
"template", "deploy-linux",
|
|
"--vmx", str(deploy_env["vmx"]),
|
|
"--force",
|
|
"--start-from-step", "8",
|
|
],
|
|
)
|
|
assert result.exit_code != 0
|
|
assert "Guest IP not available" in result.output
|
|
|
|
|
|
def test_deploy_linux_step10_patches_vmx(
|
|
deploy_env: dict, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
vmx = deploy_env["vmx"]
|
|
vmx.write_text(
|
|
'ide1:0.present = "TRUE"\nide1:0.startConnected = "TRUE"\n',
|
|
encoding="ascii",
|
|
)
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
[
|
|
"template", "deploy-linux",
|
|
"--vmx", str(vmx),
|
|
"--force",
|
|
"--start-from-step", "10",
|
|
],
|
|
)
|
|
assert result.exit_code == 0, result.output
|
|
content = vmx.read_text()
|
|
assert 'ide1:0.present = "FALSE"' in content
|
|
assert 'ide1:0.startConnected = "FALSE"' in content
|
|
|
|
|
|
def test_deploy_linux_full_happy_path(
|
|
deploy_env: dict, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
import subprocess as _sub
|
|
|
|
vmx = deploy_env["vmx"]
|
|
|
|
def fake_run(cmd: list[str], **kw: object) -> _sub.CompletedProcess[str]:
|
|
cmd_str = " ".join(str(c) for c in cmd)
|
|
if "genisoimage" in cmd_str:
|
|
for i, arg in enumerate(cmd):
|
|
if i > 0 and cmd[i - 1] == "-o":
|
|
Path(arg).write_bytes(b"\x00" * 8192)
|
|
return _sub.CompletedProcess(cmd, 0, "", "")
|
|
if "vmware-vdiskmanager" in cmd_str and "-r" in cmd:
|
|
dest = str(cmd[-1])
|
|
Path(dest).write_bytes(b"\x00" * 1024)
|
|
return _sub.CompletedProcess(cmd, 0, "", "")
|
|
if "vmware-vdiskmanager" in cmd_str:
|
|
return _sub.CompletedProcess(cmd, 0, "", "")
|
|
if "list" in cmd:
|
|
return _sub.CompletedProcess(cmd, 0, f"Total running VMs: 1\n{vmx}\n", "")
|
|
if "getGuestIPAddress" in cmd:
|
|
return _sub.CompletedProcess(cmd, 0, "192.168.79.200", "")
|
|
return _sub.CompletedProcess(cmd, 0, "", "")
|
|
|
|
monkeypatch.setattr(tmpl_mod.subprocess, "run", fake_run)
|
|
|
|
popen_m = MagicMock()
|
|
popen_m.pid = 99
|
|
monkeypatch.setattr(tmpl_mod.subprocess, "Popen", lambda *a, **kw: popen_m)
|
|
|
|
sock_m = MagicMock()
|
|
sock_m.__enter__ = lambda s: s
|
|
sock_m.__exit__ = lambda s, *a: False
|
|
monkeypatch.setattr(tmpl_mod.socket, "create_connection", lambda *a, **kw: sock_m)
|
|
|
|
stdout_m = MagicMock()
|
|
stdout_m.read.return_value = b"OK"
|
|
client_m = MagicMock()
|
|
client_m.exec_command.return_value = (MagicMock(), stdout_m, MagicMock())
|
|
monkeypatch.setattr(tmpl_mod.paramiko, "SSHClient", lambda: client_m)
|
|
monkeypatch.setattr(tmpl_mod.paramiko, "AutoAddPolicy", MagicMock)
|
|
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
["template", "deploy-linux", "--vmx", str(vmx), "--force"],
|
|
)
|
|
assert result.exit_code == 0, result.output
|
|
|
|
assert vmx.exists()
|
|
vmx_content = vmx.read_text()
|
|
assert 'guestOS = "ubuntu-64"' in vmx_content
|
|
assert 'ide1:0.present = "FALSE"' in vmx_content
|
|
assert 'ide1:0.startConnected = "FALSE"' in vmx_content
|
|
assert "guestinfo.userdata" in vmx_content # b64 from step 2
|
|
assert "192.168.79.200" in result.output # guest IP in summary
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
# template prepare-linux — helper unit tests
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
|
|
|
|
def test_pl_run_returns_stdout_and_rc(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
stdout_m = MagicMock()
|
|
stdout_m.read.return_value = b"hello\n"
|
|
stdout_m.channel.recv_exit_status.return_value = 0
|
|
stderr_m = MagicMock()
|
|
stderr_m.read.return_value = b""
|
|
client_m = MagicMock()
|
|
client_m.exec_command.return_value = (MagicMock(), stdout_m, stderr_m)
|
|
|
|
rc, out, err = tmpl_mod._pl_run(client_m, "echo hello")
|
|
assert rc == 0
|
|
assert "hello" in out
|
|
assert err == ""
|
|
|
|
|
|
def test_pl_run_nonzero_rc(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
stdout_m = MagicMock()
|
|
stdout_m.read.return_value = b""
|
|
stdout_m.channel.recv_exit_status.return_value = 1
|
|
stderr_m = MagicMock()
|
|
stderr_m.read.return_value = b"error"
|
|
client_m = MagicMock()
|
|
client_m.exec_command.return_value = (MagicMock(), stdout_m, stderr_m)
|
|
|
|
rc, _, err = tmpl_mod._pl_run(client_m, "false")
|
|
assert rc == 1
|
|
assert "error" in err
|
|
|
|
|
|
def test_pl_stream_success(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
channel_m = MagicMock()
|
|
# First call: stdout data; second: empty → exit_status_ready True on second loop
|
|
channel_m.recv_ready.side_effect = [True, False, False]
|
|
channel_m.recv.return_value = b"output line\n"
|
|
channel_m.recv_stderr_ready.return_value = False
|
|
channel_m.exit_status_ready.side_effect = [False, True]
|
|
channel_m.recv_exit_status.return_value = 0
|
|
|
|
transport_m = MagicMock()
|
|
transport_m.open_session.return_value = channel_m
|
|
client_m = MagicMock()
|
|
client_m.get_transport.return_value = transport_m
|
|
|
|
monkeypatch.setattr(tmpl_mod.time, "sleep", lambda _: None)
|
|
rc = tmpl_mod._pl_stream(client_m, "cat file")
|
|
assert rc == 0
|
|
|
|
|
|
def test_pl_upload_calls_sftp_put(tmp_path: Path) -> None:
|
|
local = tmp_path / "script.sh"
|
|
local.write_text("#!/bin/bash\n")
|
|
sftp_m = MagicMock()
|
|
client_m = MagicMock()
|
|
client_m.open_sftp.return_value = sftp_m
|
|
|
|
tmpl_mod._pl_upload(client_m, local, "/tmp/script.sh")
|
|
sftp_m.put.assert_called_once_with(str(local), "/tmp/script.sh")
|
|
sftp_m.close.assert_called_once()
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
# template prepare-linux — CLI tests
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
|
|
|
|
@pytest.fixture()
|
|
def prepare_env(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> dict:
|
|
"""Environment with SSH keys, VMX, and toolchain script."""
|
|
ssh_key = _make_keys(tmp_path / "keys")
|
|
vm_dir = tmp_path / "templates" / "LinuxBuild2404"
|
|
vm_dir.mkdir(parents=True)
|
|
vmx = vm_dir / "LinuxBuild2404.vmx"
|
|
vmx.write_text('guestOS = "ubuntu-64"\n', encoding="ascii")
|
|
|
|
# toolchain script in repo-relative location
|
|
tmpl_dir = tmp_path / "template"
|
|
tmpl_dir.mkdir()
|
|
toolchain = tmpl_dir / "Install-CIToolchain-Linux2404.sh"
|
|
toolchain.write_text("#!/bin/bash\necho done\n")
|
|
|
|
config = _deploy_config(tmp_path)
|
|
monkeypatch.setattr(tmpl_mod, "load_config", lambda: config)
|
|
monkeypatch.setattr(tmpl_mod.time, "sleep", lambda _: None)
|
|
monkeypatch.chdir(tmp_path)
|
|
|
|
return {
|
|
"tmp_path": tmp_path,
|
|
"ssh_key": ssh_key,
|
|
"vmx": vmx,
|
|
"toolchain": toolchain,
|
|
}
|
|
|
|
|
|
def _stub_prepare(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
env: dict,
|
|
*,
|
|
toolchain_exit: int = 0,
|
|
val_output: str = (
|
|
"UserSudo:OK\nSudoNoPass:OK\nTool:gcc:OK\nTool:g++:OK\n"
|
|
"Tool:clang:OK\nTool:cmake:OK\nTool:python3:OK\nTool:git:OK\n"
|
|
"Tool:7z:OK\nDir:build:OK\nDir:output:OK\nDir:scripts:OK\n"
|
|
"Swap:off:OK\nSshd:active:OK\nSSHPwdAuth:disabled:OK\n"
|
|
"CIReportIP:enabled:OK\nCIReportIPScript:exists:OK\n"
|
|
),
|
|
machine_id_size: str = "0",
|
|
) -> MagicMock:
|
|
"""Stub all SSH / vmrun calls for prepare-linux happy-path testing."""
|
|
client_m = MagicMock()
|
|
raw = env["toolchain"].read_bytes()
|
|
if raw[:3] == b"\xef\xbb\xbf":
|
|
raw = raw[3:]
|
|
_toolchain_size = str(len(raw.replace(b"\r\n", b"\n").replace(b"\r", b"\n")))
|
|
|
|
def fake_pl_run(
|
|
client: object, cmd: str, timeout: float = 60.0
|
|
) -> tuple[int, str, str]:
|
|
if "stat -c %s /tmp/" in cmd:
|
|
return 0, _toolchain_size, ""
|
|
if "stat -c %s /etc/machine-id" in cmd:
|
|
return 0, machine_id_size, ""
|
|
if "truncate" in cmd:
|
|
return 0, "", ""
|
|
if "shutdown" in cmd:
|
|
return 0, "", ""
|
|
if "echo connected" in cmd:
|
|
return 0, "connected", ""
|
|
# validation script
|
|
return 0, val_output, ""
|
|
|
|
monkeypatch.setattr(tmpl_mod, "_pl_connect", lambda *a, **kw: client_m)
|
|
monkeypatch.setattr(tmpl_mod, "_pl_run", fake_pl_run)
|
|
monkeypatch.setattr(tmpl_mod, "_pl_stream", lambda *a, **kw: toolchain_exit)
|
|
monkeypatch.setattr(tmpl_mod, "_pl_upload", lambda *a, **kw: None)
|
|
monkeypatch.setattr(tmpl_mod, "_wait_ssh_port_deploy", lambda *a, **kw: True)
|
|
return client_m
|
|
|
|
|
|
def test_prepare_linux_missing_vmx_and_ip(
|
|
prepare_env: dict, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
result = CliRunner().invoke(cli, ["template", "prepare-linux"])
|
|
assert result.exit_code != 0
|
|
assert "--vmx" in result.output or "--ip" in result.output
|
|
|
|
|
|
def test_prepare_linux_toolchain_not_found(
|
|
prepare_env: dict, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
prepare_env["toolchain"].unlink()
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
["template", "prepare-linux", "--ip", "192.168.1.100"],
|
|
)
|
|
assert result.exit_code != 0
|
|
assert "not found" in result.output.lower()
|
|
|
|
|
|
def test_prepare_linux_ssh_key_missing(
|
|
prepare_env: dict, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
prepare_env["ssh_key"].unlink()
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
["template", "prepare-linux", "--ip", "192.168.1.100"],
|
|
)
|
|
assert result.exit_code != 0
|
|
assert "not found" in result.output.lower()
|
|
|
|
|
|
def test_prepare_linux_ssh_connect_fail(
|
|
prepare_env: dict, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
monkeypatch.setattr(
|
|
tmpl_mod, "_wait_ssh_port_deploy", lambda *a, **kw: True
|
|
)
|
|
monkeypatch.setattr(
|
|
tmpl_mod, "_pl_connect",
|
|
lambda *a, **kw: (_ for _ in ()).throw(OSError("refused")),
|
|
)
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
["template", "prepare-linux", "--ip", "192.168.1.100"],
|
|
)
|
|
assert result.exit_code != 0
|
|
assert "failed" in result.output.lower()
|
|
|
|
|
|
def test_prepare_linux_ssh_port_timeout(
|
|
prepare_env: dict, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
monkeypatch.setattr(tmpl_mod, "_wait_ssh_port_deploy", lambda *a, **kw: False)
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
["template", "prepare-linux", "--ip", "192.168.1.100"],
|
|
)
|
|
assert result.exit_code != 0
|
|
assert "not reachable" in result.output or "SSH" in result.output
|
|
|
|
|
|
def test_prepare_linux_toolchain_fail(
|
|
prepare_env: dict, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
_stub_prepare(monkeypatch, prepare_env, toolchain_exit=1)
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
["template", "prepare-linux", "--ip", "192.168.1.100"],
|
|
)
|
|
assert result.exit_code != 0
|
|
assert "exited 1" in result.output or "Toolchain" in result.output
|
|
|
|
|
|
def test_prepare_linux_validation_fail(
|
|
prepare_env: dict, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
_stub_prepare(
|
|
monkeypatch, prepare_env,
|
|
val_output="UserSudo:OK\nTool:gcc:FAIL\nDir:build:OK\n",
|
|
)
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
["template", "prepare-linux", "--ip", "192.168.1.100"],
|
|
)
|
|
assert result.exit_code != 0
|
|
assert "FAIL" in result.output
|
|
|
|
|
|
def test_prepare_linux_machine_id_not_cleared(
|
|
prepare_env: dict, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
_stub_prepare(monkeypatch, prepare_env, machine_id_size="36")
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
["template", "prepare-linux", "--ip", "192.168.1.100"],
|
|
)
|
|
assert result.exit_code != 0
|
|
assert "machine-id" in result.output.lower()
|
|
|
|
|
|
def test_prepare_linux_happy_path_ip_only(
|
|
prepare_env: dict, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""No --vmx: provisions but skips auto power-on, IP detect, and snapshot."""
|
|
_stub_prepare(monkeypatch, prepare_env)
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
["template", "prepare-linux", "--ip", "192.168.1.100"],
|
|
)
|
|
assert result.exit_code == 0, result.output
|
|
assert "Provisioning complete" in result.output
|
|
assert "not taken" in result.output # snapshot skipped
|
|
|
|
|
|
def test_prepare_linux_happy_path_with_vmx(
|
|
prepare_env: dict, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""--vmx: auto IP detect + shutdown + snapshot."""
|
|
_stub_prepare(monkeypatch, prepare_env)
|
|
monkeypatch.setattr(tmpl_mod, "_poll_guest_ip_deploy", lambda *a, **kw: "192.168.1.200")
|
|
|
|
backend_m = MagicMock()
|
|
backend_m.is_running.side_effect = [False, True, False] # start, check, powered off
|
|
backend_m.list_snapshots.return_value = []
|
|
monkeypatch.setattr(tmpl_mod, "WorkstationVmrunBackend", lambda **kw: backend_m)
|
|
|
|
import subprocess as _sub
|
|
monkeypatch.setattr(
|
|
tmpl_mod.subprocess, "run",
|
|
lambda cmd, **kw: _sub.CompletedProcess(list(cmd), 0, "", ""),
|
|
)
|
|
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
["template", "prepare-linux", "--vmx", str(prepare_env["vmx"])],
|
|
)
|
|
assert result.exit_code == 0, result.output
|
|
assert "Snapshot" in result.output
|
|
assert "192.168.1.200" in result.output
|
|
|
|
|
|
def test_prepare_linux_zombie_vm_restarts(
|
|
prepare_env: dict, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""VM process alive but guest powered off → quick probe fails → stop+restart."""
|
|
_stub_prepare(monkeypatch, prepare_env)
|
|
|
|
# First _poll call (quick probe, 30s) returns None; second (full 300s) returns IP.
|
|
poll_calls: list[float] = []
|
|
|
|
def fake_poll(vmrun: str, vmx: str, timeout: float) -> str | None:
|
|
poll_calls.append(timeout)
|
|
return None if timeout == 30.0 else "192.168.1.200"
|
|
|
|
monkeypatch.setattr(tmpl_mod, "_poll_guest_ip_deploy", fake_poll)
|
|
|
|
backend_m = MagicMock()
|
|
backend_m.is_running.side_effect = [True, True, False] # already running; post-stop checks
|
|
backend_m.list_snapshots.return_value = []
|
|
monkeypatch.setattr(tmpl_mod, "WorkstationVmrunBackend", lambda **kw: backend_m)
|
|
|
|
import subprocess as _sub
|
|
monkeypatch.setattr(
|
|
tmpl_mod.subprocess, "run",
|
|
lambda cmd, **kw: _sub.CompletedProcess(list(cmd), 0, "", ""),
|
|
)
|
|
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
["template", "prepare-linux", "--vmx", str(prepare_env["vmx"])],
|
|
)
|
|
assert result.exit_code == 0, result.output
|
|
assert "restarting" in result.output
|
|
backend_m.stop.assert_called_once()
|
|
backend_m.start.assert_called()
|
|
assert 30.0 in poll_calls
|
|
|
|
|
|
def test_prepare_linux_happy_path_skip_flags(
|
|
prepare_env: dict, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""--skip-update, --install-dotnet, --no-mingw reach the toolchain command."""
|
|
streamed: list[str] = []
|
|
client_m = MagicMock()
|
|
|
|
_raw = prepare_env["toolchain"].read_bytes()
|
|
_tsize = str(len(_raw.replace(b"\r\n", b"\n").replace(b"\r", b"\n")))
|
|
|
|
def fake_pl_run(client: object, cmd: str, timeout: float = 60.0) -> tuple[int, str, str]:
|
|
if "stat -c %s /tmp/" in cmd:
|
|
return 0, _tsize, ""
|
|
if "stat -c %s /etc/machine-id" in cmd:
|
|
return 0, "0", ""
|
|
if "truncate" in cmd:
|
|
return 0, "", ""
|
|
if "echo connected" in cmd:
|
|
return 0, "connected", ""
|
|
return 0, "UserSudo:OK\nSudoNoPass:OK\n", ""
|
|
|
|
def fake_pl_stream(client: object, cmd: str) -> int:
|
|
streamed.append(cmd)
|
|
return 0
|
|
|
|
monkeypatch.setattr(tmpl_mod, "_pl_connect", lambda *a, **kw: client_m)
|
|
monkeypatch.setattr(tmpl_mod, "_pl_run", fake_pl_run)
|
|
monkeypatch.setattr(tmpl_mod, "_pl_stream", fake_pl_stream)
|
|
monkeypatch.setattr(tmpl_mod, "_pl_upload", lambda *a, **kw: None)
|
|
monkeypatch.setattr(tmpl_mod, "_wait_ssh_port_deploy", lambda *a, **kw: True)
|
|
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
[
|
|
"template", "prepare-linux",
|
|
"--ip", "192.168.1.100",
|
|
"--skip-update", "--install-dotnet", "--no-mingw",
|
|
],
|
|
)
|
|
assert result.exit_code == 0, result.output
|
|
assert streamed
|
|
assert "--skip-update" in streamed[0]
|
|
assert "--dotnet" in streamed[0]
|
|
assert "--no-mingw" in streamed[0]
|
|
|
|
|
|
def test_prepare_linux_existing_snapshot_deleted(
|
|
prepare_env: dict, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""If snapshot already exists, delete before recreating."""
|
|
_stub_prepare(monkeypatch, prepare_env)
|
|
monkeypatch.setattr(tmpl_mod, "_poll_guest_ip_deploy", lambda *a, **kw: "192.168.1.200")
|
|
|
|
backend_m = MagicMock()
|
|
backend_m.is_running.side_effect = [True, False]
|
|
backend_m.list_snapshots.return_value = ["BaseClean-Linux"]
|
|
monkeypatch.setattr(tmpl_mod, "WorkstationVmrunBackend", lambda **kw: backend_m)
|
|
|
|
run_calls: list[list[str]] = []
|
|
import subprocess as _sub
|
|
|
|
def fake_run(cmd: list[str], **kw: object) -> _sub.CompletedProcess[str]:
|
|
run_calls.append(list(cmd))
|
|
return _sub.CompletedProcess(cmd, 0, "", "")
|
|
|
|
monkeypatch.setattr(tmpl_mod.subprocess, "run", fake_run)
|
|
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
["template", "prepare-linux", "--vmx", str(prepare_env["vmx"])],
|
|
)
|
|
assert result.exit_code == 0, result.output
|
|
cmds = [" ".join(c) for c in run_calls]
|
|
assert any("deleteSnapshot" in c for c in cmds)
|
|
assert any("snapshot" in c and "deleteSnapshot" not in c for c in cmds)
|