Simone
567f9da619
template backup ran a full multi-GB, multi-minute 7z pass over every template every night regardless of whether anything under it had actually changed -- the systemd unit's own comment already noted the resulting archives were byte-identical run to run. Now compares the template dir's newest file mtime against its most recent existing backup and skips the compress step (still honoring --keep-count via prune) when nothing is newer. --force opts back into an unconditional backup.
3257 lines
115 KiB
Python
3257 lines
115 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
|
|
|
|
|
|
# ── skip-if-unchanged ────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_backup_skips_when_unchanged_since_last_backup(
|
|
template_dir: Path, backup_dir: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""The nightly timer runs every night; most nights nothing in a template
|
|
changed, so the (very expensive) 7z pass over tens of GB is pure waste."""
|
|
past = time.time() - 7200
|
|
for f in template_dir.iterdir():
|
|
os.utime(f, (past, past))
|
|
# This backup's own mtime is "now" (just written) -- newer than every
|
|
# file the template directory holds.
|
|
existing = backup_dir / "Template_20250101_000000.7z"
|
|
existing.write_bytes(b"old-archive")
|
|
|
|
compress_calls: list[Path] = []
|
|
monkeypatch.setattr(
|
|
tmpl_mod, "_compress_7z", lambda src, dst: compress_calls.append(src)
|
|
)
|
|
monkeypatch.setattr(tmpl_mod, "_stop_runner", lambda **_kw: False)
|
|
monkeypatch.setattr(tmpl_mod, "_start_runner", lambda **_kw: 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
|
|
assert compress_calls == []
|
|
assert "unchanged" in result.output.lower()
|
|
archives = [f for f in backup_dir.iterdir() if f.suffix == ".7z"]
|
|
assert archives == [existing]
|
|
|
|
|
|
def test_backup_runs_when_changed_since_last_backup(
|
|
template_dir: Path, backup_dir: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
old_backup = backup_dir / "Template_20250101_000000.7z"
|
|
old_backup.write_bytes(b"old-archive")
|
|
old_time = time.time() - 7200
|
|
os.utime(old_backup, (old_time, old_time))
|
|
# template_dir's files keep their fixture-default "now" mtime -- newer
|
|
# than the existing backup.
|
|
|
|
compress_calls: list[Path] = []
|
|
monkeypatch.setattr(
|
|
tmpl_mod,
|
|
"_compress_7z",
|
|
lambda src, dst: (compress_calls.append(src), dst.write_bytes(b"fake"))[-1],
|
|
)
|
|
monkeypatch.setattr(tmpl_mod, "_stop_runner", lambda **_kw: False)
|
|
monkeypatch.setattr(tmpl_mod, "_start_runner", lambda **_kw: 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
|
|
assert compress_calls == [template_dir]
|
|
|
|
|
|
def test_force_flag_backs_up_even_when_unchanged(
|
|
template_dir: Path, backup_dir: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
past = time.time() - 7200
|
|
for f in template_dir.iterdir():
|
|
os.utime(f, (past, past))
|
|
existing = backup_dir / "Template_20250101_000000.7z"
|
|
existing.write_bytes(b"old-archive")
|
|
|
|
compress_calls: list[Path] = []
|
|
monkeypatch.setattr(
|
|
tmpl_mod,
|
|
"_compress_7z",
|
|
lambda src, dst: (compress_calls.append(src), dst.write_bytes(b"fake"))[-1],
|
|
)
|
|
monkeypatch.setattr(tmpl_mod, "_stop_runner", lambda **_kw: False)
|
|
monkeypatch.setattr(tmpl_mod, "_start_runner", lambda **_kw: None)
|
|
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
[
|
|
"template",
|
|
"backup",
|
|
"--template-path",
|
|
str(template_dir),
|
|
"--backup-base-dir",
|
|
str(backup_dir),
|
|
"--skip-runner-stop",
|
|
"--force",
|
|
],
|
|
)
|
|
|
|
assert result.exit_code == 0, result.output
|
|
assert compress_calls == [template_dir]
|
|
|
|
|
|
def test_what_if_reports_skip_when_unchanged(
|
|
template_dir: Path, backup_dir: Path, _no_systemctl: None
|
|
) -> None:
|
|
past = time.time() - 7200
|
|
for f in template_dir.iterdir():
|
|
os.utime(f, (past, past))
|
|
existing = backup_dir / "Template_20250101_000000.7z"
|
|
existing.write_bytes(b"old-archive")
|
|
|
|
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 "WhatIf" in result.output
|
|
assert "unchanged" in result.output.lower()
|
|
archives = [f for f in backup_dir.iterdir() if f.suffix == ".7z"]
|
|
assert archives == [existing]
|
|
|
|
|
|
def test_all_templates_skips_unchanged_and_backs_up_changed(
|
|
tmp_path: Path, backup_dir: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""Skip/run is decided per template, not for the whole ``--all-templates``
|
|
pass at once."""
|
|
tmpl_root = tmp_path / "templates"
|
|
unchanged = tmpl_root / "LinuxBuild2404"
|
|
changed = tmpl_root / "WinBuild2025"
|
|
for d in (unchanged, changed):
|
|
d.mkdir(parents=True)
|
|
(d / f"{d.name}.vmx").write_text("vmx")
|
|
|
|
past = time.time() - 7200
|
|
os.utime(unchanged / "LinuxBuild2404.vmx", (past, past))
|
|
# `changed`'s file keeps its just-created "now" mtime.
|
|
|
|
existing_unchanged = backup_dir / "Template_LinuxBuild2404_20250101_000000.7z"
|
|
existing_unchanged.write_bytes(b"old")
|
|
existing_changed = backup_dir / "Template_WinBuild2025_20250101_000000.7z"
|
|
existing_changed.write_bytes(b"old")
|
|
# Newer than `unchanged`'s file, older than `changed`'s.
|
|
mid_time = time.time() - 3600
|
|
os.utime(existing_unchanged, (mid_time, mid_time))
|
|
os.utime(existing_changed, (mid_time, mid_time))
|
|
|
|
monkeypatch.setenv("CI_TEMPLATES", str(tmpl_root))
|
|
compress_calls: list[Path] = []
|
|
monkeypatch.setattr(
|
|
tmpl_mod,
|
|
"_compress_7z",
|
|
lambda src, dst: (compress_calls.append(src), dst.write_bytes(b"fake"))[-1],
|
|
)
|
|
monkeypatch.setattr(tmpl_mod, "_stop_runner", lambda **_kw: False)
|
|
monkeypatch.setattr(tmpl_mod, "_start_runner", lambda **_kw: None)
|
|
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
[
|
|
"template",
|
|
"backup",
|
|
"--all-templates",
|
|
"--backup-base-dir",
|
|
str(backup_dir),
|
|
"--skip-runner-stop",
|
|
],
|
|
)
|
|
|
|
assert result.exit_code == 0, result.output
|
|
assert compress_calls == [changed]
|
|
|
|
|
|
# ── 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"
|
|
)
|