feat(template): add template prepare-win Python command via pypsrp
Replaces the abandoned PSWSMan approach (New-PSSession hangs on Linux) with a Python-native implementation using the proven pypsrp WinRmTransport. - New `template prepare-win` command in commands/template.py: starts VM, waits for WinRM, uploads and runs Install-CIToolchain-*.ps1, handles Windows Update 3010 reboot loop (max 10 iterations), installs CI-StaticIp scheduled task, runs post-setup validation, graceful shutdown, snapshot. - Helpers: _ps_escape, _wait_tcp, _parse_exit_marker, _wait_winrm. - 40 new tests in test_commands_template.py covering all major branches. - Fix test_commands_job.py: patch load_config in _patch_common so tests don't pick up live /var/lib/ci/config.toml ip_pool and fail on lock. - Coverage gate: 90.06% (was 88.95% before fix). Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -100,6 +100,15 @@ def _patch_common(
|
||||
|
||||
monkeypatch.setattr(job_module, "load_backend", lambda _cfg: backend)
|
||||
|
||||
# Prevent tests from reading the live /var/lib/ci/config.toml which may
|
||||
# have ip_pool configured — that would try to lock /var/lib/ci/ip-pool.lock
|
||||
# (requires root) and fail before any backend call.
|
||||
from ci_orchestrator.config import BackendConfig, Config, Paths
|
||||
|
||||
_test_paths = Paths(root=Path("/tmp/ci-test"), templates=Path("/tmp/ci-test/templates"), build_vms=Path("/tmp/ci-test/build-vms"), artifacts=Path("/tmp/ci-test/artifacts"), keys=Path("/tmp/ci-test/keys"))
|
||||
_test_cfg = Config(paths=_test_paths, backend=BackendConfig(), ip_pool=None)
|
||||
monkeypatch.setattr(job_module, "load_config", lambda: _test_cfg)
|
||||
|
||||
class _ProbeOk:
|
||||
def __enter__(self) -> _ProbeOk:
|
||||
return self
|
||||
|
||||
@@ -1,16 +1,21 @@
|
||||
"""Tests for ``ci_orchestrator template backup``."""
|
||||
"""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 ──────────────────────────────────────────────────────────────────
|
||||
@@ -303,3 +308,610 @@ def test_compress_7z_uses_mx1(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -
|
||||
cmd = calls[0]
|
||||
assert cmd[0] == "7z"
|
||||
assert "-mx=1" in cmd
|
||||
|
||||
|
||||
# ═══════════════════════════════════════════════════════════════════════════
|
||||
# template prepare-win — helper unit tests
|
||||
# ═══════════════════════════════════════════════════════════════════════════
|
||||
|
||||
|
||||
def test_ps_escape_no_quotes() -> None:
|
||||
assert tmpl_mod._ps_escape("hello world") == "hello world"
|
||||
|
||||
|
||||
def test_ps_escape_single_quote() -> None:
|
||||
assert tmpl_mod._ps_escape("it's") == "it''s"
|
||||
|
||||
|
||||
def test_ps_escape_multiple_quotes() -> None:
|
||||
assert tmpl_mod._ps_escape("it's Tom's") == "it''s Tom''s"
|
||||
|
||||
|
||||
def test_parse_exit_marker_present() -> None:
|
||||
assert tmpl_mod._parse_exit_marker("output\nCI_EXITCODE:3010\n") == 3010
|
||||
|
||||
|
||||
def test_parse_exit_marker_absent() -> None:
|
||||
assert tmpl_mod._parse_exit_marker("no marker here") == 0
|
||||
|
||||
|
||||
def test_parse_exit_marker_last_line_wins() -> None:
|
||||
assert tmpl_mod._parse_exit_marker("CI_EXITCODE:1\nCI_EXITCODE:0") == 0
|
||||
|
||||
|
||||
def test_parse_exit_marker_malformed() -> None:
|
||||
assert tmpl_mod._parse_exit_marker("CI_EXITCODE:abc") == 0
|
||||
|
||||
|
||||
def test_wait_tcp_success(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
class _FakeConn:
|
||||
def __enter__(self) -> "_FakeConn":
|
||||
return self
|
||||
|
||||
def __exit__(self, *_: object) -> None:
|
||||
pass
|
||||
|
||||
monkeypatch.setattr(tmpl_mod.socket, "create_connection", lambda *a, **kw: _FakeConn())
|
||||
assert tmpl_mod._wait_tcp("127.0.0.1", 9999, 5.0) is True
|
||||
|
||||
|
||||
def test_wait_tcp_timeout(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
def _refuse(*a: object, **kw: object) -> None:
|
||||
raise OSError("refused")
|
||||
|
||||
monkeypatch.setattr(tmpl_mod.socket, "create_connection", _refuse)
|
||||
monkeypatch.setattr(tmpl_mod.time, "sleep", lambda _: None)
|
||||
assert tmpl_mod._wait_tcp("127.0.0.1", 9999, 0.0) is False
|
||||
|
||||
|
||||
def test_wait_winrm_success() -> None:
|
||||
transport = MagicMock()
|
||||
transport.is_ready.return_value = True
|
||||
tmpl_mod._wait_winrm(transport, 30) # must not raise
|
||||
|
||||
|
||||
def test_wait_winrm_timeout() -> None:
|
||||
transport = MagicMock()
|
||||
transport.is_ready.return_value = False
|
||||
with pytest.raises(click.ClickException, match="not ready"):
|
||||
tmpl_mod._wait_winrm(transport, 0)
|
||||
|
||||
|
||||
# ═══════════════════════════════════════════════════════════════════════════
|
||||
# template prepare-win — command tests
|
||||
# ═══════════════════════════════════════════════════════════════════════════
|
||||
|
||||
|
||||
def _make_config() -> types.SimpleNamespace:
|
||||
return types.SimpleNamespace(vmrun_path=None, ip_pool=None)
|
||||
|
||||
|
||||
def _make_backend(is_running_seq: list[bool]) -> MagicMock:
|
||||
b: MagicMock = MagicMock()
|
||||
b.is_running.side_effect = is_running_seq
|
||||
b.get_ip.return_value = "192.168.1.1"
|
||||
b.list_snapshots.return_value = []
|
||||
b.vmrun_path = "vmrun"
|
||||
return b
|
||||
|
||||
|
||||
def _make_transport() -> MagicMock:
|
||||
val_json = json.dumps(
|
||||
{
|
||||
k: True
|
||||
for k in [
|
||||
"WinRMRunning", "UserExists", "UserIsAdmin", "UACDisabled",
|
||||
"FirewallOff", "DotNetExists", "PythonExists", "MSBuildExists",
|
||||
"CIDirsPresent", "StaticIpTask", "StaticIpScript",
|
||||
]
|
||||
}
|
||||
)
|
||||
|
||||
def _run(script: str = "", **kwargs: object) -> WinRmResult:
|
||||
if "ConvertTo-Json" in script:
|
||||
return WinRmResult(stdout=val_json, stderr="", returncode=0)
|
||||
return WinRmResult(stdout="CI_EXITCODE:0", stderr="", returncode=0)
|
||||
|
||||
t: MagicMock = MagicMock()
|
||||
t.run.side_effect = _run
|
||||
t.is_ready.return_value = True
|
||||
return t
|
||||
|
||||
|
||||
def _setup_prepare_env(
|
||||
tmp_path: Path,
|
||||
*,
|
||||
with_static_ip: bool = True,
|
||||
) -> Path:
|
||||
"""Create fake VMX + toolchain script layout under tmp_path."""
|
||||
vmx = tmp_path / "WinBuild2025.vmx"
|
||||
vmx.write_text("vmx content")
|
||||
tmpl_dir = tmp_path / "template"
|
||||
tmpl_dir.mkdir()
|
||||
(tmpl_dir / "Install-CIToolchain-WinBuild2025.ps1").write_text("# fake toolchain")
|
||||
if with_static_ip:
|
||||
sip_dir = tmpl_dir / "guest-setup" / "windows"
|
||||
sip_dir.mkdir(parents=True)
|
||||
(sip_dir / "ci-static-ip.ps1").write_text("# fake static ip")
|
||||
return vmx
|
||||
|
||||
|
||||
def test_prepare_win_vmx_not_found(tmp_path: Path) -> None:
|
||||
result = CliRunner().invoke(
|
||||
cli,
|
||||
["template", "prepare-win", "--vmx", str(tmp_path / "missing.vmx")],
|
||||
)
|
||||
assert result.exit_code != 0
|
||||
assert "VMX not found" in result.output
|
||||
|
||||
|
||||
def test_prepare_win_toolchain_not_found(
|
||||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
vmx = tmp_path / "WinBuild2025.vmx"
|
||||
vmx.write_text("vmx")
|
||||
monkeypatch.chdir(tmp_path)
|
||||
result = CliRunner().invoke(
|
||||
cli,
|
||||
["template", "prepare-win", "--vmx", str(vmx)],
|
||||
)
|
||||
assert result.exit_code != 0
|
||||
assert "not found" in result.output
|
||||
|
||||
|
||||
def test_prepare_win_explicit_toolchain_not_found(tmp_path: Path) -> None:
|
||||
vmx = tmp_path / "WinBuild2025.vmx"
|
||||
vmx.write_text("vmx")
|
||||
result = CliRunner().invoke(
|
||||
cli,
|
||||
[
|
||||
"template", "prepare-win",
|
||||
"--vmx", str(vmx),
|
||||
"--toolchain-script", str(tmp_path / "nope.ps1"),
|
||||
],
|
||||
)
|
||||
assert result.exit_code != 0
|
||||
assert "not found" in result.output
|
||||
|
||||
|
||||
def _patch_prepare(monkeypatch: pytest.MonkeyPatch, backend: MagicMock, transport: MagicMock) -> None:
|
||||
monkeypatch.setattr(tmpl_mod, "load_config", lambda: _make_config())
|
||||
monkeypatch.setattr(tmpl_mod, "WorkstationVmrunBackend", lambda **_: backend)
|
||||
monkeypatch.setattr(tmpl_mod, "WinRmTransport", lambda **_: transport)
|
||||
monkeypatch.setattr(tmpl_mod, "_wait_tcp", lambda *_: True)
|
||||
monkeypatch.setattr(tmpl_mod, "_wait_winrm", lambda *_: None)
|
||||
monkeypatch.setattr(tmpl_mod.time, "sleep", lambda _: None)
|
||||
monkeypatch.setattr(tmpl_mod.subprocess, "run", lambda *a, **_: None)
|
||||
|
||||
|
||||
def test_prepare_win_happy_path(
|
||||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
vmx = _setup_prepare_env(tmp_path)
|
||||
monkeypatch.chdir(tmp_path)
|
||||
backend = _make_backend([True, False])
|
||||
transport = _make_transport()
|
||||
_patch_prepare(monkeypatch, backend, transport)
|
||||
|
||||
result = CliRunner().invoke(
|
||||
cli,
|
||||
[
|
||||
"template", "prepare-win",
|
||||
"--vmx", str(vmx),
|
||||
"--ip", "192.168.1.100",
|
||||
"--admin-password", "adminpass",
|
||||
"--build-password", "buildpass",
|
||||
],
|
||||
)
|
||||
assert result.exit_code == 0, result.output
|
||||
assert "Provisioning complete" in result.output
|
||||
assert "All post-setup checks passed" in result.output
|
||||
|
||||
|
||||
def test_prepare_win_no_static_ip_warning(
|
||||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
vmx = _setup_prepare_env(tmp_path, with_static_ip=False)
|
||||
monkeypatch.chdir(tmp_path)
|
||||
backend = _make_backend([True, False])
|
||||
transport = _make_transport()
|
||||
_patch_prepare(monkeypatch, backend, transport)
|
||||
|
||||
result = CliRunner().invoke(
|
||||
cli,
|
||||
[
|
||||
"template", "prepare-win",
|
||||
"--vmx", str(vmx),
|
||||
"--ip", "192.168.1.100",
|
||||
"--admin-password", "adminpass",
|
||||
"--build-password", "buildpass",
|
||||
],
|
||||
)
|
||||
assert result.exit_code == 0, result.output
|
||||
assert "WARNING" in result.output
|
||||
|
||||
|
||||
def test_prepare_win_vm_started_when_not_running(
|
||||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
vmx = _setup_prepare_env(tmp_path)
|
||||
monkeypatch.chdir(tmp_path)
|
||||
backend = _make_backend([False, False]) # not running → start; then powered off
|
||||
transport = _make_transport()
|
||||
_patch_prepare(monkeypatch, backend, transport)
|
||||
|
||||
result = CliRunner().invoke(
|
||||
cli,
|
||||
[
|
||||
"template", "prepare-win",
|
||||
"--vmx", str(vmx),
|
||||
"--ip", "192.168.1.100",
|
||||
"--admin-password", "adminpass",
|
||||
"--build-password", "buildpass",
|
||||
],
|
||||
)
|
||||
assert result.exit_code == 0, result.output
|
||||
backend.start.assert_called_once()
|
||||
|
||||
|
||||
def test_prepare_win_tcp_timeout(
|
||||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
vmx = _setup_prepare_env(tmp_path)
|
||||
monkeypatch.chdir(tmp_path)
|
||||
backend = _make_backend([True])
|
||||
_patch_prepare(monkeypatch, backend, MagicMock())
|
||||
monkeypatch.setattr(tmpl_mod, "_wait_tcp", lambda *_: False)
|
||||
|
||||
result = CliRunner().invoke(
|
||||
cli,
|
||||
[
|
||||
"template", "prepare-win",
|
||||
"--vmx", str(vmx),
|
||||
"--ip", "192.168.1.100",
|
||||
"--admin-password", "adminpass",
|
||||
"--build-password", "buildpass",
|
||||
],
|
||||
)
|
||||
assert result.exit_code != 0
|
||||
assert "not reachable" in result.output
|
||||
|
||||
|
||||
def test_prepare_win_toolchain_fails(
|
||||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
vmx = _setup_prepare_env(tmp_path)
|
||||
monkeypatch.chdir(tmp_path)
|
||||
backend = _make_backend([True])
|
||||
transport: MagicMock = MagicMock()
|
||||
transport.is_ready.return_value = True
|
||||
transport.run.return_value = WinRmResult(
|
||||
stdout="CI_EXITCODE:1", stderr="err", returncode=0
|
||||
)
|
||||
_patch_prepare(monkeypatch, backend, transport)
|
||||
|
||||
result = CliRunner().invoke(
|
||||
cli,
|
||||
[
|
||||
"template", "prepare-win",
|
||||
"--vmx", str(vmx),
|
||||
"--ip", "192.168.1.100",
|
||||
"--admin-password", "adminpass",
|
||||
"--build-password", "buildpass",
|
||||
],
|
||||
)
|
||||
assert result.exit_code != 0
|
||||
assert "failed" in result.output
|
||||
|
||||
|
||||
def test_prepare_win_snapshot_exists_skip(
|
||||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
vmx = _setup_prepare_env(tmp_path)
|
||||
monkeypatch.chdir(tmp_path)
|
||||
backend = _make_backend([True, False])
|
||||
backend.list_snapshots.return_value = ["BaseClean"]
|
||||
transport = _make_transport()
|
||||
_patch_prepare(monkeypatch, backend, transport)
|
||||
|
||||
# User declines deletion with "N"
|
||||
result = CliRunner().invoke(
|
||||
cli,
|
||||
[
|
||||
"template", "prepare-win",
|
||||
"--vmx", str(vmx),
|
||||
"--ip", "192.168.1.100",
|
||||
"--admin-password", "adminpass",
|
||||
"--build-password", "buildpass",
|
||||
],
|
||||
input="N\n",
|
||||
)
|
||||
assert result.exit_code == 0, result.output
|
||||
assert "Keeping existing snapshot" in result.output
|
||||
|
||||
|
||||
def test_prepare_win_snapshot_exists_recreate(
|
||||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
vmx = _setup_prepare_env(tmp_path)
|
||||
monkeypatch.chdir(tmp_path)
|
||||
backend = _make_backend([True, False])
|
||||
backend.list_snapshots.return_value = ["BaseClean"]
|
||||
transport = _make_transport()
|
||||
sp_calls: list[list[str]] = []
|
||||
monkeypatch.setattr(tmpl_mod, "load_config", lambda: _make_config())
|
||||
monkeypatch.setattr(tmpl_mod, "WorkstationVmrunBackend", lambda **_: backend)
|
||||
monkeypatch.setattr(tmpl_mod, "WinRmTransport", lambda **_: transport)
|
||||
monkeypatch.setattr(tmpl_mod, "_wait_tcp", lambda *_: True)
|
||||
monkeypatch.setattr(tmpl_mod, "_wait_winrm", lambda *_: None)
|
||||
monkeypatch.setattr(tmpl_mod.time, "sleep", lambda _: None)
|
||||
|
||||
def _sp_run(cmd: list[str], **_: object) -> None:
|
||||
sp_calls.append(cmd)
|
||||
|
||||
monkeypatch.setattr(tmpl_mod.subprocess, "run", _sp_run)
|
||||
|
||||
result = CliRunner().invoke(
|
||||
cli,
|
||||
[
|
||||
"template", "prepare-win",
|
||||
"--vmx", str(vmx),
|
||||
"--ip", "192.168.1.100",
|
||||
"--admin-password", "adminpass",
|
||||
"--build-password", "buildpass",
|
||||
],
|
||||
input="Y\n",
|
||||
)
|
||||
assert result.exit_code == 0, result.output
|
||||
assert any("deleteSnapshot" in " ".join(c) for c in sp_calls)
|
||||
assert any("snapshot" in " ".join(c) and "deleteSnapshot" not in " ".join(c) for c in sp_calls)
|
||||
|
||||
|
||||
def test_prepare_win_skip_flags(
|
||||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
"""--skip-windows-update/tier2/cleanup flags reach the invoke script."""
|
||||
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
|
||||
|
||||
Reference in New Issue
Block a user