Simone
6bce846953
Four defects, each of which could waste a full template rebuild: 1. getGuestIPAddress was called without -wait, so vmrun answered "The VMware Tools are not running" unless the Tools happened to have registered the IP at that exact instant. A guest that was booted, on the network and answering SSH could still burn the entire 10-minute Step 7 budget. This race is the likely cause of every intermittent "no guest IP" incident so far, including ones previously attributed to wedged vmnet DHCP. AGENTS.md #10 already prescribed -wait; the code did not follow it. 2. --force did not check whether the target VM was powered on. It deleted the live .vmdk.lck as a stale artifact and Step 4 rewrote the base VMDK under the running guest, killing it; the damage surfaced ten minutes later as an IP timeout because `vmrun list` still showed the OLD instance. 3. Step 6 accepted "VM appears in vmrun list" as proof of power-on, which a pre-existing instance satisfies. Check the exit code of `vmrun start`. 4. The cloud-init seed ISO was detached while the VM was still running, and VMware rewrote the VMX from its in-memory copy at power-off, restoring ide1:0.present = "TRUE". Every clone would boot with the seed attached (AGENTS.md #6). Detach it in prepare, after power-off and before the snapshot. Also in this change: - --then-prepare chains deploy into the matching prepare in one run, reusing the VM deploy leaves running. The two commands stay separately invocable. The guest toolchain script is resolved and verified BEFORE any VM work, so a wrong working directory costs two seconds instead of failing the hand-off after a 45-minute deploy; --toolchain-script allows running from anywhere. - Corrected the final messages: deploy-rust told the user to run prepare-linux, and both deploy commands reported a snapshot name (PostInstall-Linux) that no command ever creates. The prepare hand-off hint passed a bare filename to --template and omitted --snapshot, so copy-pasting it cloned BaseClean, the Windows snapshot. - template backup --exclude, repeatable, for retired templates. An unknown name or --exclude without --all-templates is an error rather than a silent no-op. test_deploy_linux_full_happy_path declared the VM already running before Step 6, an impossible state that masked the new --force guard; its vmrun mock is now stateful.
3061 lines
108 KiB
Python
3061 lines
108 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_reboot_and_wait_boot_change(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
calls: list[str] = []
|
|
transport = MagicMock()
|
|
monkeypatch.setattr(tmpl_mod, "_tcp_open", lambda *a, **k: True)
|
|
# boot ticks: before reboot, then a newer value once it's back.
|
|
ticks = iter([100, 200])
|
|
monkeypatch.setattr(tmpl_mod, "_guest_boot_ticks", lambda _t: next(ticks))
|
|
monkeypatch.setattr(tmpl_mod, "_wait_winrm", lambda *a, **k: calls.append("winrm"))
|
|
monkeypatch.setattr(tmpl_mod.time, "sleep", lambda _s: None)
|
|
tmpl_mod._reboot_and_wait(transport, "10.0.0.5", 5986, 30)
|
|
assert "winrm" in calls
|
|
transport.close.assert_called_once()
|
|
|
|
|
|
def test_reboot_and_wait_timeout(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
transport = MagicMock()
|
|
monkeypatch.setattr(tmpl_mod, "_tcp_open", lambda *a, **k: False) # never comes back
|
|
monkeypatch.setattr(tmpl_mod, "_guest_boot_ticks", lambda _t: 100)
|
|
monkeypatch.setattr(tmpl_mod.time, "sleep", lambda _s: None)
|
|
clock = iter([0.0, 1.0, 1e9]) # deadline calc, one loop check, then past deadline
|
|
monkeypatch.setattr(tmpl_mod.time, "monotonic", lambda: next(clock))
|
|
with pytest.raises(click.ClickException, match="did not reboot"):
|
|
tmpl_mod._reboot_and_wait(transport, "10.0.0.5", 5986, 1)
|
|
|
|
|
|
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 "PercentProcessorTime" in script: # pre-snapshot idle check
|
|
return WinRmResult(stdout='{"cpu":2.0,"ram":21.0,"top":"idle"}', stderr="", returncode=0)
|
|
if ").Shell" in script: # up-front shell check — already cmd, skip reboot
|
|
return WinRmResult(stdout="cmd.exe", stderr="", returncode=0)
|
|
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:
|
|
# _run_with_heartbeat prefers transport.run_streaming; route it through the
|
|
# test's transport.run side_effect and feed stdout to the on_line callback.
|
|
if not hasattr(transport.run_streaming, "side_effect") or transport.run_streaming.side_effect is None:
|
|
def _stream(script: str = "", on_line: object = None, **kwargs: object) -> WinRmResult:
|
|
res = transport.run(script, **kwargs)
|
|
if callable(on_line) and isinstance(res, WinRmResult):
|
|
on_line(res.stdout)
|
|
return res
|
|
transport.run_streaming.side_effect = _stream
|
|
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, "_reboot_and_wait", lambda *_a, **_k: None)
|
|
monkeypatch.setattr(tmpl_mod, "_wait_winrm", lambda *_: None)
|
|
monkeypatch.setattr(tmpl_mod, "_check_guest_idle", lambda *_a, **_k: 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, "_reboot_and_wait", lambda *_a, **_k: None)
|
|
monkeypatch.setattr(tmpl_mod, "_check_guest_idle", lambda *_a, **_k: None)
|
|
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"]
|
|
|
|
# `vmrun list` must reflect reality: the VM is NOT running before Step 6
|
|
# powers it on. A mock that always reports it running hides the --force
|
|
# pre-flight guard (deploy must refuse to rewrite a live VM's disk).
|
|
started = {"value": False}
|
|
|
|
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 "start" in cmd:
|
|
started["value"] = True
|
|
return _sub.CompletedProcess(cmd, 0, "", "")
|
|
if "list" in cmd:
|
|
body = (
|
|
f"Total running VMs: 1\n{vmx}\n"
|
|
if started["value"]
|
|
else "Total running VMs: 0\n"
|
|
)
|
|
return _sub.CompletedProcess(cmd, 0, body, "")
|
|
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", ""
|
|
# Step 7d cloud-init seal verification.
|
|
if "cloud-init.disabled" in cmd and "echo DISABLED" in cmd:
|
|
return 0, "DISABLED\nNO50\n", ""
|
|
# 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:
|
|
"""Without --vmx/--ip AND no default template VMX on disk → error."""
|
|
prepare_env["vmx"].unlink() # default LinuxBuild2404.vmx absent
|
|
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_default_vmx_used(
|
|
prepare_env: dict, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""Without --vmx/--ip, the default LinuxBuild2404 VMX is picked up."""
|
|
_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, False]
|
|
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"])
|
|
assert result.exit_code == 0, result.output
|
|
assert "LinuxBuild2404.vmx" in result.output
|
|
assert "192.168.1.200" 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", ""
|
|
if "cloud-init.disabled" in cmd and "echo DISABLED" in cmd:
|
|
return 0, "DISABLED\nNO50\n", ""
|
|
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)
|
|
|
|
|
|
# ── _clean_force_artifacts (deploy-linux --force snapshot-chain cleanup) ────────
|
|
|
|
|
|
def test_clean_force_artifacts_removes_snapshot_chain(tmp_path: Path) -> None:
|
|
"""Snapshot delta/.vmsn/.vmsd/.lck are removed; base disk + vmx kept."""
|
|
(tmp_path / "LinuxBuild2404.vmdk").write_text("base")
|
|
(tmp_path / "LinuxBuild2404.vmx").write_text("vmx")
|
|
(tmp_path / "LinuxBuild2404-000001.vmdk").write_text("delta")
|
|
(tmp_path / "LinuxBuild2404-000002.vmdk").write_text("delta2")
|
|
(tmp_path / "LinuxBuild2404-Snapshot7.vmsn").write_text("snap")
|
|
(tmp_path / "LinuxBuild2404.vmsd").write_text("snapdb")
|
|
(tmp_path / "LinuxBuild2404.vmdk.lck").write_text("lock")
|
|
|
|
removed = tmpl_mod._clean_force_artifacts(tmp_path, "LinuxBuild2404")
|
|
|
|
assert set(removed) == {
|
|
"LinuxBuild2404-000001.vmdk",
|
|
"LinuxBuild2404-000002.vmdk",
|
|
"LinuxBuild2404-Snapshot7.vmsn",
|
|
"LinuxBuild2404.vmsd",
|
|
"LinuxBuild2404.vmdk.lck",
|
|
}
|
|
# Base disk and vmx survive (Step 4/5 overwrite them).
|
|
assert (tmp_path / "LinuxBuild2404.vmdk").is_file()
|
|
assert (tmp_path / "LinuxBuild2404.vmx").is_file()
|
|
# Chain artifacts gone.
|
|
assert not (tmp_path / "LinuxBuild2404-000001.vmdk").exists()
|
|
assert not (tmp_path / "LinuxBuild2404.vmsd").exists()
|
|
|
|
|
|
def test_clean_force_artifacts_removes_lock_directory(tmp_path: Path) -> None:
|
|
"""Lock entries may be directories (VMware writes .lck dirs)."""
|
|
lck = tmp_path / "LinuxBuild2404.vmx.lck"
|
|
lck.mkdir()
|
|
(lck / "M12345.lck").write_text("host")
|
|
|
|
removed = tmpl_mod._clean_force_artifacts(tmp_path, "LinuxBuild2404")
|
|
|
|
assert removed == ["LinuxBuild2404.vmx.lck"]
|
|
assert not lck.exists()
|
|
|
|
|
|
def test_clean_force_artifacts_noop_on_clean_dir(tmp_path: Path) -> None:
|
|
"""A clean directory yields no removals and leaves the base disk."""
|
|
(tmp_path / "LinuxBuild2404.vmdk").write_text("base")
|
|
|
|
removed = tmpl_mod._clean_force_artifacts(tmp_path, "LinuxBuild2404")
|
|
|
|
assert removed == []
|
|
assert (tmp_path / "LinuxBuild2404.vmdk").is_file()
|
|
|
|
|
|
def test_check_guest_idle_ok() -> None:
|
|
transport = MagicMock()
|
|
transport.run.return_value = WinRmResult(
|
|
stdout='{"cpu":3.0,"ram":22.5,"top":"idle"}', stderr="", returncode=0
|
|
)
|
|
tmpl_mod._check_guest_idle(transport, max_cpu=10.0, max_ram=50.0, settle_s=0, timeout_s=5)
|
|
|
|
|
|
def test_check_guest_idle_cpu_over() -> None:
|
|
transport = MagicMock()
|
|
transport.run.return_value = WinRmResult(
|
|
stdout='{"cpu":42.0,"ram":20.0,"top":"TiWorker"}', stderr="", returncode=0
|
|
)
|
|
with pytest.raises(click.ClickException, match=r"CPU 42\.0% > 10\.0%"):
|
|
tmpl_mod._check_guest_idle(transport, max_cpu=10.0, max_ram=50.0, settle_s=0, timeout_s=0)
|
|
|
|
|
|
def test_check_guest_idle_ram_over() -> None:
|
|
transport = MagicMock()
|
|
transport.run.return_value = WinRmResult(
|
|
stdout='{"cpu":1.0,"ram":80.0,"top":"x"}', stderr="", returncode=0
|
|
)
|
|
with pytest.raises(click.ClickException, match=r"RAM 80\.0% > 50\.0%"):
|
|
tmpl_mod._check_guest_idle(transport, max_cpu=10.0, max_ram=50.0, settle_s=0, timeout_s=0)
|
|
|
|
|
|
def test_check_guest_idle_unparseable() -> None:
|
|
transport = MagicMock()
|
|
transport.run.return_value = WinRmResult(stdout="not json", stderr="", returncode=0)
|
|
with pytest.raises(click.ClickException, match="could not parse"):
|
|
tmpl_mod._sample_guest_idle(transport)
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
# template deploy-rust / prepare-rust — CLI tests
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
|
|
|
|
def test_deploy_rust_help() -> None:
|
|
result = CliRunner().invoke(cli, ["template", "deploy-rust", "--help"])
|
|
assert result.exit_code == 0
|
|
assert "LinuxRust2404" in result.output
|
|
|
|
|
|
def test_deploy_rust_forwards_to_deploy_linux(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""deploy-rust calls the deploy-linux callback with the Rust defaults."""
|
|
captured: dict = {}
|
|
|
|
def fake_cb(**kwargs: object) -> None:
|
|
captured.update(kwargs)
|
|
|
|
monkeypatch.setattr(tmpl_mod.template_deploy_linux, "callback", fake_cb)
|
|
|
|
result = CliRunner().invoke(cli, ["template", "deploy-rust", "--force"])
|
|
assert result.exit_code == 0, result.output
|
|
assert captured["vm_name"] == "LinuxRust2404"
|
|
assert captured["guest_hostname"] == "ci-rust-template"
|
|
assert captured["disk_size_gb"] == 60
|
|
assert captured["force"] is True
|
|
# Default VMX points at the LinuxRust2404 template dir.
|
|
vmx = str(captured["vmx_path_str"])
|
|
assert vmx.endswith("LinuxRust2404/LinuxRust2404.vmx")
|
|
|
|
|
|
def test_deploy_rust_vmx_override(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""An explicit --vmx is forwarded untouched."""
|
|
captured: dict = {}
|
|
monkeypatch.setattr(
|
|
tmpl_mod.template_deploy_linux, "callback",
|
|
lambda **kw: captured.update(kw),
|
|
)
|
|
result = CliRunner().invoke(
|
|
cli, ["template", "deploy-rust", "--vmx", "/tmp/custom/Custom.vmx"]
|
|
)
|
|
assert result.exit_code == 0, result.output
|
|
assert captured["vmx_path_str"] == "/tmp/custom/Custom.vmx"
|
|
|
|
|
|
@pytest.fixture()
|
|
def prepare_rust_env(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> dict:
|
|
"""Environment with SSH keys, VMX, and the Rust toolchain script."""
|
|
ssh_key = _make_keys(tmp_path / "keys")
|
|
vm_dir = tmp_path / "templates" / "LinuxRust2404"
|
|
vm_dir.mkdir(parents=True)
|
|
vmx = vm_dir / "LinuxRust2404.vmx"
|
|
vmx.write_text('guestOS = "ubuntu-64"\n', encoding="ascii")
|
|
|
|
tmpl_dir = tmp_path / "template"
|
|
tmpl_dir.mkdir()
|
|
toolchain = tmpl_dir / "Install-CIToolchain-LinuxRust2404.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_rust(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
env: dict,
|
|
*,
|
|
streamed: list[str] | None = None,
|
|
) -> MagicMock:
|
|
"""Stub SSH/vmrun for prepare-rust; records the streamed toolchain command."""
|
|
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, "0", ""
|
|
if "truncate" in cmd:
|
|
return 0, "", ""
|
|
if "shutdown" in cmd:
|
|
return 0, "", ""
|
|
if "echo connected" in cmd:
|
|
return 0, "connected", ""
|
|
if "cloud-init.disabled" in cmd and "echo DISABLED" in cmd:
|
|
return 0, "DISABLED\nNO50\n", ""
|
|
return 0, "UserSudo:OK\nSudoNoPass:OK\n", ""
|
|
|
|
def fake_pl_stream(client: object, cmd: str) -> int:
|
|
if streamed is not None:
|
|
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)
|
|
return client_m
|
|
|
|
|
|
def test_prepare_rust_help() -> None:
|
|
result = CliRunner().invoke(cli, ["template", "prepare-rust", "--help"])
|
|
assert result.exit_code == 0
|
|
assert "--rust-channel" in result.output
|
|
assert "--no-cargo-audit" in result.output
|
|
|
|
|
|
def test_prepare_rust_missing_vmx_and_ip(prepare_rust_env: dict) -> None:
|
|
"""Without --vmx/--ip AND no default template VMX on disk → error."""
|
|
prepare_rust_env["vmx"].unlink() # default LinuxRust2404.vmx absent
|
|
result = CliRunner().invoke(cli, ["template", "prepare-rust"])
|
|
assert result.exit_code != 0
|
|
assert "--vmx" in result.output or "--ip" in result.output
|
|
|
|
|
|
def test_prepare_rust_default_vmx_used(
|
|
prepare_rust_env: dict, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""Without --vmx/--ip, the default LinuxRust2404 VMX is picked up."""
|
|
_stub_prepare_rust(monkeypatch, prepare_rust_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, False]
|
|
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-rust"])
|
|
assert result.exit_code == 0, result.output
|
|
assert "LinuxRust2404.vmx" in result.output
|
|
assert "BaseClean-LinuxRust" in result.output
|
|
|
|
|
|
def test_prepare_rust_toolchain_not_found(prepare_rust_env: dict) -> None:
|
|
prepare_rust_env["toolchain"].unlink()
|
|
result = CliRunner().invoke(
|
|
cli, ["template", "prepare-rust", "--ip", "192.168.1.100"]
|
|
)
|
|
assert result.exit_code != 0
|
|
assert "not found" in result.output.lower()
|
|
assert "LinuxRust2404" in result.output
|
|
|
|
|
|
def test_prepare_rust_happy_path_defaults(
|
|
prepare_rust_env: dict, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""Default invocation: Rust toolchain script, BaseClean-LinuxRust, stable channel."""
|
|
streamed: list[str] = []
|
|
_stub_prepare_rust(monkeypatch, prepare_rust_env, streamed=streamed)
|
|
|
|
result = CliRunner().invoke(
|
|
cli, ["template", "prepare-rust", "--ip", "192.168.1.100"]
|
|
)
|
|
assert result.exit_code == 0, result.output
|
|
assert "[prepare-rust]" in result.output
|
|
assert "BaseClean-LinuxRust" in result.output
|
|
assert streamed
|
|
assert "Install-CIToolchain-LinuxRust2404.sh" in streamed[0]
|
|
assert "--rust-channel stable" in streamed[0]
|
|
assert "--no-cargo-audit" not in streamed[0]
|
|
|
|
|
|
def test_prepare_rust_forwards_rust_flags(
|
|
prepare_rust_env: dict, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""--rust-channel and --no-cargo-audit reach the toolchain command."""
|
|
streamed: list[str] = []
|
|
_stub_prepare_rust(monkeypatch, prepare_rust_env, streamed=streamed)
|
|
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
[
|
|
"template", "prepare-rust",
|
|
"--ip", "192.168.1.100",
|
|
"--rust-channel", "1.97.1",
|
|
"--no-cargo-audit",
|
|
],
|
|
)
|
|
assert result.exit_code == 0, result.output
|
|
assert streamed
|
|
assert "--rust-channel 1.97.1" in streamed[0]
|
|
assert "--no-cargo-audit" in streamed[0]
|
|
|
|
|
|
def test_prepare_rust_snapshot_override(
|
|
prepare_rust_env: dict, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""--snapshot overrides the BaseClean-LinuxRust default."""
|
|
_stub_prepare_rust(monkeypatch, prepare_rust_env)
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
[
|
|
"template", "prepare-rust",
|
|
"--ip", "192.168.1.100",
|
|
"--snapshot", "CustomSnap",
|
|
],
|
|
)
|
|
assert result.exit_code == 0, result.output
|
|
assert "CustomSnap" in result.output
|
|
|
|
|
|
def test_prepare_linux_unchanged_defaults(
|
|
prepare_env: dict, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""Regression: prepare-linux still uses the Linux toolchain + BaseClean-Linux
|
|
and never forwards rust flags."""
|
|
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 or "shutdown" in cmd:
|
|
return 0, "", ""
|
|
if "echo connected" in cmd:
|
|
return 0, "connected", ""
|
|
if "cloud-init.disabled" in cmd and "echo DISABLED" in cmd:
|
|
return 0, "DISABLED\nNO50\n", ""
|
|
return 0, "UserSudo:OK\nSudoNoPass:OK\n", ""
|
|
|
|
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 c, cmd: streamed.append(cmd) or 0,
|
|
)
|
|
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"]
|
|
)
|
|
assert result.exit_code == 0, result.output
|
|
assert "[prepare-linux]" in result.output
|
|
assert "BaseClean-Linux" in result.output
|
|
assert streamed
|
|
assert "Install-CIToolchain-Linux2404.sh" in streamed[0]
|
|
assert "--rust-channel" not in streamed[0]
|
|
assert "--no-cargo-audit" not in streamed[0]
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
# Final messages + deploy -> prepare hand-off
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
|
|
|
|
def _deploy_linux_stub_env(tmp_path: Path, name: str) -> dict:
|
|
"""Minimal on-disk state for a deploy-* run resumed at the last step."""
|
|
vm_dir = tmp_path / "templates" / name
|
|
vm_dir.mkdir(parents=True)
|
|
vmx = vm_dir / f"{name}.vmx"
|
|
vmx.write_text('ide1:0.present = "TRUE"\n', encoding="ascii")
|
|
key = tmp_path / "ci_linux"
|
|
key.write_text("PRIVATE KEY", encoding="ascii")
|
|
return {"vmx": vmx, "key": key}
|
|
|
|
|
|
def test_deploy_rust_summary_points_at_prepare_rust(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""Regression: deploy-rust used to tell the user to run prepare-linux."""
|
|
env = _deploy_linux_stub_env(tmp_path, "LinuxRust2404")
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
[
|
|
"template", "deploy-rust",
|
|
"--vmx", str(env["vmx"]),
|
|
"--ssh-key", str(env["key"]),
|
|
"--start-from-step", "10",
|
|
],
|
|
)
|
|
assert result.exit_code == 0, result.output
|
|
assert "deploy-rust complete" in result.output
|
|
assert "prepare-rust" in result.output
|
|
assert "prepare-linux" not in result.output
|
|
# The snapshot named in the summary is the one prepare-rust creates.
|
|
assert "BaseClean-LinuxRust" in result.output
|
|
assert "PostInstall-Linux" not in result.output
|
|
|
|
|
|
def test_deploy_linux_summary_points_at_prepare_linux(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
env = _deploy_linux_stub_env(tmp_path, "LinuxBuild2404")
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
[
|
|
"template", "deploy-linux",
|
|
"--vmx", str(env["vmx"]),
|
|
"--ssh-key", str(env["key"]),
|
|
"--start-from-step", "10",
|
|
],
|
|
)
|
|
assert result.exit_code == 0, result.output
|
|
assert "deploy-linux complete" in result.output
|
|
assert "template prepare-linux --vmx" in result.output
|
|
assert "BaseClean-Linux" in result.output
|
|
|
|
|
|
def test_deploy_rust_then_prepare_hands_off(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""--then-prepare chains into prepare-rust with the Rust defaults."""
|
|
env = _deploy_linux_stub_env(tmp_path, "LinuxRust2404")
|
|
captured: dict = {}
|
|
monkeypatch.setattr(
|
|
tmpl_mod, "_prepare_linux_impl", lambda **kw: captured.update(kw)
|
|
)
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
[
|
|
"template", "deploy-rust",
|
|
"--vmx", str(env["vmx"]),
|
|
"--ssh-key", str(env["key"]),
|
|
"--start-from-step", "10",
|
|
"--then-prepare",
|
|
"--rust-channel", "1.97.1",
|
|
],
|
|
)
|
|
assert result.exit_code == 0, result.output
|
|
assert captured["prog_name"] == "prepare-rust"
|
|
assert captured["default_vmx_name"] == "LinuxRust2404"
|
|
assert captured["default_toolchain_name"] == "Install-CIToolchain-LinuxRust2404.sh"
|
|
assert captured["snapshot"] == "BaseClean-LinuxRust"
|
|
assert captured["rust_channel"] == "1.97.1"
|
|
assert captured["take_snapshot"] is True
|
|
assert captured["vmx_path_str"] == str(env["vmx"])
|
|
|
|
|
|
def test_deploy_linux_then_prepare_hands_off(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
env = _deploy_linux_stub_env(tmp_path, "LinuxBuild2404")
|
|
captured: dict = {}
|
|
monkeypatch.setattr(
|
|
tmpl_mod, "_prepare_linux_impl", lambda **kw: captured.update(kw)
|
|
)
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
[
|
|
"template", "deploy-linux",
|
|
"--vmx", str(env["vmx"]),
|
|
"--ssh-key", str(env["key"]),
|
|
"--start-from-step", "10",
|
|
"--then-prepare",
|
|
],
|
|
)
|
|
assert result.exit_code == 0, result.output
|
|
assert captured["prog_name"] == "prepare-linux"
|
|
assert captured["default_toolchain_name"] == "Install-CIToolchain-Linux2404.sh"
|
|
assert captured["snapshot"] == "BaseClean-Linux"
|
|
assert captured["rust_channel"] is None
|
|
|
|
|
|
def test_deploy_without_then_prepare_does_not_chain(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""The two commands stay separate: no hand-off unless asked for."""
|
|
env = _deploy_linux_stub_env(tmp_path, "LinuxBuild2404")
|
|
calls: list[dict] = []
|
|
monkeypatch.setattr(
|
|
tmpl_mod, "_prepare_linux_impl", lambda **kw: calls.append(kw)
|
|
)
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
[
|
|
"template", "deploy-linux",
|
|
"--vmx", str(env["vmx"]),
|
|
"--ssh-key", str(env["key"]),
|
|
"--start-from-step", "10",
|
|
],
|
|
)
|
|
assert result.exit_code == 0, result.output
|
|
assert calls == []
|
|
|
|
|
|
def test_prepare_rust_next_hint_is_copy_pasteable(
|
|
prepare_rust_env: dict, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""The `vm new` hint needs the full VMX path and an explicit --snapshot.
|
|
|
|
`vm new` defaults --snapshot to "BaseClean" (the Windows snapshot), so a
|
|
hint that omits it clones the wrong snapshot.
|
|
"""
|
|
_stub_prepare_rust(monkeypatch, prepare_rust_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, False]
|
|
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-rust"])
|
|
assert result.exit_code == 0, result.output
|
|
vmx = str(prepare_rust_env["vmx"])
|
|
assert f"--template {vmx}" in result.output
|
|
assert "--snapshot BaseClean-LinuxRust" in result.output
|
|
assert "--guest-os linux" in result.output
|
|
|
|
|
|
# ── --exclude ─────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def _tmpl_root_with(tmp_path: Path, names: tuple[str, ...]) -> Path:
|
|
tmpl_root = tmp_path / "templates"
|
|
for name in names:
|
|
d = tmpl_root / name
|
|
d.mkdir(parents=True)
|
|
(d / f"{name}.vmx").write_text("vmx")
|
|
return tmpl_root
|
|
|
|
|
|
def test_exclude_skips_named_template(
|
|
tmp_path: Path, backup_dir: Path, monkeypatch: pytest.MonkeyPatch, _no_systemctl: None
|
|
) -> None:
|
|
"""A retired template must not keep costing a backup generation."""
|
|
tmpl_root = _tmpl_root_with(
|
|
tmp_path, ("WinBuild2025", "WinBuildTest", "LinuxBuild2404")
|
|
)
|
|
monkeypatch.setenv("CI_TEMPLATES", str(tmpl_root))
|
|
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
[
|
|
"template", "backup", "--all-templates",
|
|
"--exclude", "WinBuildTest",
|
|
"--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("LinuxBuild2404" in n for n in names)
|
|
assert not any("WinBuildTest" in n for n in names)
|
|
assert "Skipping 'WinBuildTest'" in result.output
|
|
|
|
|
|
def test_exclude_is_repeatable(
|
|
tmp_path: Path, backup_dir: Path, monkeypatch: pytest.MonkeyPatch, _no_systemctl: None
|
|
) -> None:
|
|
tmpl_root = _tmpl_root_with(
|
|
tmp_path, ("WinBuild2025", "WinBuildTest", "LinuxBuildTest")
|
|
)
|
|
monkeypatch.setenv("CI_TEMPLATES", str(tmpl_root))
|
|
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
[
|
|
"template", "backup", "--all-templates",
|
|
"--exclude", "WinBuildTest",
|
|
"--exclude", "LinuxBuildTest",
|
|
"--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 names and all("Test" not in n for n in names)
|
|
|
|
|
|
def test_exclude_unknown_name_raises(
|
|
tmp_path: Path, backup_dir: Path, monkeypatch: pytest.MonkeyPatch, _no_systemctl: None
|
|
) -> None:
|
|
"""A typo must fail loudly, not silently back up everything."""
|
|
tmpl_root = _tmpl_root_with(tmp_path, ("WinBuild2025",))
|
|
monkeypatch.setenv("CI_TEMPLATES", str(tmpl_root))
|
|
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
[
|
|
"template", "backup", "--all-templates",
|
|
"--exclude", "WinBuildTst",
|
|
"--backup-base-dir", str(backup_dir),
|
|
"--skip-runner-stop",
|
|
],
|
|
)
|
|
assert result.exit_code != 0
|
|
assert "no such template" in result.output
|
|
|
|
|
|
def test_exclude_everything_raises(
|
|
tmp_path: Path, backup_dir: Path, monkeypatch: pytest.MonkeyPatch, _no_systemctl: None
|
|
) -> None:
|
|
tmpl_root = _tmpl_root_with(tmp_path, ("WinBuild2025",))
|
|
monkeypatch.setenv("CI_TEMPLATES", str(tmpl_root))
|
|
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
[
|
|
"template", "backup", "--all-templates",
|
|
"--exclude", "WinBuild2025",
|
|
"--backup-base-dir", str(backup_dir),
|
|
"--skip-runner-stop",
|
|
],
|
|
)
|
|
assert result.exit_code != 0
|
|
assert "nothing to back up" in result.output
|
|
|
|
|
|
def test_exclude_without_all_templates_raises(
|
|
tmp_path: Path, backup_dir: Path, monkeypatch: pytest.MonkeyPatch, _no_systemctl: None
|
|
) -> None:
|
|
"""--exclude alone would be a silent no-op; it must error instead."""
|
|
tmpl_root = _tmpl_root_with(tmp_path, ("WinBuild2025",))
|
|
monkeypatch.setenv("CI_TEMPLATES", str(tmpl_root))
|
|
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
[
|
|
"template", "backup",
|
|
"--template-path", str(tmpl_root / "WinBuild2025"),
|
|
"--exclude", "WinBuildTest",
|
|
"--backup-base-dir", str(backup_dir),
|
|
"--skip-runner-stop",
|
|
],
|
|
)
|
|
assert result.exit_code != 0
|
|
assert "only applies to --all-templates" in result.output
|
|
|
|
|
|
def test_then_prepare_fails_before_deploy_when_toolchain_missing(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""Regression: the toolchain was resolved only after the deploy finished.
|
|
|
|
Running deploy-rust --then-prepare from outside the repo root completed a
|
|
~45-minute deploy and only then died on the missing script. The check must
|
|
happen before any VM work starts.
|
|
"""
|
|
env = _deploy_linux_stub_env(tmp_path, "LinuxRust2404")
|
|
monkeypatch.chdir(tmp_path) # no template/ dir here
|
|
|
|
steps: list[int] = []
|
|
monkeypatch.setattr(
|
|
tmpl_mod, "_deploy_step", lambda *a, **kw: steps.append(1)
|
|
)
|
|
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
[
|
|
"template", "deploy-rust",
|
|
"--vmx", str(env["vmx"]),
|
|
"--ssh-key", str(env["key"]),
|
|
"--then-prepare",
|
|
],
|
|
)
|
|
assert result.exit_code != 0
|
|
assert "Install-CIToolchain-LinuxRust2404.sh" in result.output
|
|
assert "--toolchain-script" in result.output
|
|
# Nothing was deployed: the failure landed before the first step ran.
|
|
assert steps == []
|
|
|
|
|
|
def test_then_prepare_accepts_explicit_toolchain_script(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""--toolchain-script makes the chain independent of the current directory."""
|
|
env = _deploy_linux_stub_env(tmp_path, "LinuxRust2404")
|
|
script = tmp_path / "custom-toolchain.sh"
|
|
script.write_text("#!/bin/sh\n", encoding="ascii")
|
|
monkeypatch.chdir(tmp_path)
|
|
|
|
captured: dict = {}
|
|
monkeypatch.setattr(
|
|
tmpl_mod, "_prepare_linux_impl", lambda **kw: captured.update(kw)
|
|
)
|
|
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
[
|
|
"template", "deploy-rust",
|
|
"--vmx", str(env["vmx"]),
|
|
"--ssh-key", str(env["key"]),
|
|
"--start-from-step", "10",
|
|
"--then-prepare",
|
|
"--toolchain-script", str(script),
|
|
],
|
|
)
|
|
assert result.exit_code == 0, result.output
|
|
assert captured["toolchain_script_str"] == str(script)
|
|
|
|
|
|
def test_resolve_toolchain_script_prefers_cwd(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
tdir = tmp_path / "template"
|
|
tdir.mkdir()
|
|
script = tdir / "Install-CIToolchain-Linux2404.sh"
|
|
script.write_text("#!/bin/sh\n", encoding="ascii")
|
|
monkeypatch.chdir(tmp_path)
|
|
|
|
got = tmpl_mod._resolve_toolchain_script("Install-CIToolchain-Linux2404.sh", None)
|
|
assert got == script
|
|
|
|
|
|
def test_resolve_toolchain_script_explicit_missing_raises(tmp_path: Path) -> None:
|
|
import click as _click
|
|
|
|
with pytest.raises(_click.ClickException, match="not found"):
|
|
tmpl_mod._resolve_toolchain_script("x.sh", str(tmp_path / "nope.sh"))
|
|
|
|
|
|
def test_poll_guest_ip_passes_wait_flag(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Regression: without -wait vmrun reports "Tools are not running".
|
|
|
|
A guest that was booted, on the network and answering SSH still burned the
|
|
full 10-minute Step 7 budget because the poll omitted -wait, so vmrun only
|
|
answered when the Tools happened to have registered the IP at that instant.
|
|
"""
|
|
import subprocess as _sub
|
|
|
|
seen: list[list[str]] = []
|
|
|
|
def fake_run(cmd, **kw): # type: ignore[no-untyped-def]
|
|
seen.append(list(cmd))
|
|
return _sub.CompletedProcess(cmd, 0, "192.168.1.100", "")
|
|
|
|
monkeypatch.setattr(tmpl_mod, "_read_guestinfo_ip", lambda *a, **kw: None)
|
|
monkeypatch.setattr(tmpl_mod.subprocess, "run", fake_run)
|
|
monkeypatch.setattr(tmpl_mod.time, "sleep", lambda _: None)
|
|
|
|
assert tmpl_mod._poll_guest_ip_deploy("/vmrun", "/vm.vmx", 60.0) == "192.168.1.100"
|
|
assert seen, "vmrun was never invoked"
|
|
assert "-wait" in seen[0], f"-wait missing from {seen[0]}"
|
|
|
|
|
|
def test_poll_guest_ip_ignores_error_line_on_stdout(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""vmrun can exit 0 with "Error: ..." on stdout (AGENTS.md #10)."""
|
|
import subprocess as _sub
|
|
|
|
outputs = [
|
|
"Error: The VMware Tools are not running in the virtual machine\n",
|
|
"192.168.1.101\n",
|
|
]
|
|
|
|
def fake_run(cmd, **kw): # type: ignore[no-untyped-def]
|
|
return _sub.CompletedProcess(cmd, 0, outputs.pop(0), "")
|
|
|
|
monkeypatch.setattr(tmpl_mod, "_read_guestinfo_ip", lambda *a, **kw: None)
|
|
monkeypatch.setattr(tmpl_mod.subprocess, "run", fake_run)
|
|
monkeypatch.setattr(tmpl_mod.time, "sleep", lambda _: None)
|
|
|
|
# The Error line must not be mistaken for an address; the poll retries.
|
|
assert tmpl_mod._poll_guest_ip_deploy("/vmrun", "/vm.vmx", 60.0) == "192.168.1.101"
|
|
|
|
|
|
def test_deploy_force_refuses_when_vm_is_powered_on(
|
|
deploy_env: dict, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""Regression: --force against a running VM destroyed the live guest.
|
|
|
|
It deleted the live .vmdk.lck as "stale" and Step 4 rewrote the base VMDK
|
|
underneath it; the damage only surfaced ~10 minutes later as a Step 7 IP
|
|
timeout, because `vmrun list` still showed the OLD instance.
|
|
"""
|
|
import subprocess as _sub
|
|
|
|
vmx = deploy_env["vmx"]
|
|
|
|
def fake_run(cmd: list[str], **kw: object) -> _sub.CompletedProcess[str]:
|
|
if "list" in cmd:
|
|
return _sub.CompletedProcess(cmd, 0, f"Total running VMs: 1\n{vmx}\n", "")
|
|
return _sub.CompletedProcess(cmd, 0, "", "")
|
|
|
|
monkeypatch.setattr(tmpl_mod.subprocess, "run", fake_run)
|
|
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
["template", "deploy-linux", "--vmx", str(vmx), "--force"],
|
|
)
|
|
assert result.exit_code != 0
|
|
assert "powered on" in result.output
|
|
assert "vmrun -T ws stop" in result.output
|
|
|
|
|
|
def test_prepare_detaches_seed_iso_before_snapshot(
|
|
prepare_rust_env: dict, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""Regression: the seed ISO ended up attached inside the snapshot.
|
|
|
|
deploy Step 10 sets ide1:0.present = FALSE while the VM is still powered
|
|
on; VMware rewrites the VMX from its in-memory copy at power-off and
|
|
restores TRUE. Clones then boot with cloud-init's seed still attached
|
|
(AGENTS.md #6), so prepare must detach it after power-off.
|
|
"""
|
|
vmx: Path = prepare_rust_env["vmx"]
|
|
vmx.write_text(
|
|
'displayName = "LinuxRust2404"\n'
|
|
'ide1:0.present = "TRUE"\n'
|
|
'ide1:0.startConnected = "TRUE"\n'
|
|
'ide1:0.fileName = "cloud-init-seed.iso"\n',
|
|
encoding="ascii",
|
|
)
|
|
|
|
_stub_prepare_rust(monkeypatch, prepare_rust_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, False]
|
|
backend_m.list_snapshots.return_value = []
|
|
monkeypatch.setattr(tmpl_mod, "WorkstationVmrunBackend", lambda **kw: backend_m)
|
|
|
|
snapshot_calls: list[list[str]] = []
|
|
|
|
import subprocess as _sub
|
|
|
|
def fake_run(cmd, **kw): # type: ignore[no-untyped-def]
|
|
if "snapshot" in cmd:
|
|
# The VMX must already be patched by the time the snapshot is taken.
|
|
snapshot_calls.append(vmx.read_text(encoding="ascii").splitlines())
|
|
return _sub.CompletedProcess(list(cmd), 0, "", "")
|
|
|
|
monkeypatch.setattr(tmpl_mod.subprocess, "run", fake_run)
|
|
|
|
result = CliRunner().invoke(cli, ["template", "prepare-rust"])
|
|
assert result.exit_code == 0, result.output
|
|
|
|
text = vmx.read_text(encoding="ascii")
|
|
assert 'ide1:0.present = "FALSE"' in text
|
|
assert 'ide1:0.startConnected = "FALSE"' in text
|
|
assert snapshot_calls, "snapshot was never taken"
|
|
assert any('ide1:0.present = "FALSE"' in ln for ln in snapshot_calls[0]), (
|
|
"snapshot captured the VMX with the seed ISO still attached"
|
|
)
|