Add tests for SSH transport and template deployment functionality
- Introduced a new test to verify behavior when SSH transport connects but is not ready, ensuring it returns False after the deadline. - Added comprehensive tests for the `deploy-linux` template, covering various scenarios including missing public keys, VM directory existence, SHA256 mismatches, and failures during ISO creation and VM conversion. - Implemented tests for SSH-related functions, including connection retries and streaming command execution, ensuring proper handling of exit statuses and progress indications. - Enhanced existing tests to cover edge cases and ensure robustness in the deployment process.
This commit is contained in:
@@ -1142,3 +1142,892 @@ def test_prepare_win_cleanup_only_fails(
|
||||
)
|
||||
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, IpPoolConfig, 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")
|
||||
|
||||
|
||||
# ── _poll_guest_ip_deploy ─────────────────────────────────────────────────────
|
||||
|
||||
|
||||
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", ""),
|
||||
)
|
||||
|
||||
result = 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", ""),
|
||||
)
|
||||
|
||||
result = CliRunner().invoke(
|
||||
cli,
|
||||
[
|
||||
"template", "deploy-linux",
|
||||
"--vmx", str(deploy_env["vmx"]),
|
||||
"--force",
|
||||
"--start-from-step", "5", # step 2 was skipped → no b64 strings
|
||||
],
|
||||
)
|
||||
if deploy_env["vmx"].exists():
|
||||
content = deploy_env["vmx"].read_text()
|
||||
assert "guestinfo.userdata" not in content
|
||||
|
||||
|
||||
def test_deploy_linux_step7_ip_timeout(
|
||||
deploy_env: dict, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
import subprocess as _sub
|
||||
|
||||
popen_m = MagicMock()
|
||||
popen_m.pid = 1
|
||||
monkeypatch.setattr(tmpl_mod.subprocess, "Popen", lambda *a, **kw: popen_m)
|
||||
|
||||
vmx_str = str(deploy_env["vmx"])
|
||||
monkeypatch.setattr(
|
||||
tmpl_mod.subprocess, "run",
|
||||
lambda cmd, **kw: _sub.CompletedProcess(
|
||||
list(cmd), 0, f"Total running VMs: 1\n{vmx_str}\n", ""
|
||||
),
|
||||
)
|
||||
# patch the poller so it returns None immediately (no 300-second spin loop)
|
||||
monkeypatch.setattr(tmpl_mod, "_poll_guest_ip_deploy", lambda *a, **kw: None)
|
||||
|
||||
deploy_env["vmx"].write_text('dummy = "vmx"', encoding="ascii")
|
||||
|
||||
result = CliRunner().invoke(
|
||||
cli,
|
||||
[
|
||||
"template", "deploy-linux",
|
||||
"--vmx", str(deploy_env["vmx"]),
|
||||
"--force",
|
||||
"--start-from-step", "6",
|
||||
],
|
||||
)
|
||||
assert result.exit_code != 0
|
||||
assert "guest IP" in result.output
|
||||
|
||||
|
||||
def test_deploy_linux_step8_no_guest_ip(
|
||||
deploy_env: dict, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
deploy_env["vmx"].write_text('dummy = "vmx"', encoding="ascii")
|
||||
result = CliRunner().invoke(
|
||||
cli,
|
||||
[
|
||||
"template", "deploy-linux",
|
||||
"--vmx", str(deploy_env["vmx"]),
|
||||
"--force",
|
||||
"--start-from-step", "8",
|
||||
],
|
||||
)
|
||||
assert result.exit_code != 0
|
||||
assert "Guest IP not available" in result.output
|
||||
|
||||
|
||||
def test_deploy_linux_step10_patches_vmx(
|
||||
deploy_env: dict, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
vmx = deploy_env["vmx"]
|
||||
vmx.write_text(
|
||||
'ide1:0.present = "TRUE"\nide1:0.startConnected = "TRUE"\n',
|
||||
encoding="ascii",
|
||||
)
|
||||
result = CliRunner().invoke(
|
||||
cli,
|
||||
[
|
||||
"template", "deploy-linux",
|
||||
"--vmx", str(vmx),
|
||||
"--force",
|
||||
"--start-from-step", "10",
|
||||
],
|
||||
)
|
||||
assert result.exit_code == 0, result.output
|
||||
content = vmx.read_text()
|
||||
assert 'ide1:0.present = "FALSE"' in content
|
||||
assert 'ide1:0.startConnected = "FALSE"' in content
|
||||
|
||||
|
||||
def test_deploy_linux_full_happy_path(
|
||||
deploy_env: dict, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
import subprocess as _sub
|
||||
|
||||
vmx = deploy_env["vmx"]
|
||||
|
||||
def fake_run(cmd: list[str], **kw: object) -> _sub.CompletedProcess[str]:
|
||||
cmd_str = " ".join(str(c) for c in cmd)
|
||||
if "genisoimage" in cmd_str:
|
||||
for i, arg in enumerate(cmd):
|
||||
if i > 0 and cmd[i - 1] == "-o":
|
||||
Path(arg).write_bytes(b"\x00" * 8192)
|
||||
return _sub.CompletedProcess(cmd, 0, "", "")
|
||||
if "vmware-vdiskmanager" in cmd_str and "-r" in cmd:
|
||||
dest = str(cmd[-1])
|
||||
Path(dest).write_bytes(b"\x00" * 1024)
|
||||
return _sub.CompletedProcess(cmd, 0, "", "")
|
||||
if "vmware-vdiskmanager" in cmd_str:
|
||||
return _sub.CompletedProcess(cmd, 0, "", "")
|
||||
if "list" in cmd:
|
||||
return _sub.CompletedProcess(cmd, 0, f"Total running VMs: 1\n{vmx}\n", "")
|
||||
if "getGuestIPAddress" in cmd:
|
||||
return _sub.CompletedProcess(cmd, 0, "192.168.79.200", "")
|
||||
return _sub.CompletedProcess(cmd, 0, "", "")
|
||||
|
||||
monkeypatch.setattr(tmpl_mod.subprocess, "run", fake_run)
|
||||
|
||||
popen_m = MagicMock()
|
||||
popen_m.pid = 99
|
||||
monkeypatch.setattr(tmpl_mod.subprocess, "Popen", lambda *a, **kw: popen_m)
|
||||
|
||||
sock_m = MagicMock()
|
||||
sock_m.__enter__ = lambda s: s
|
||||
sock_m.__exit__ = lambda s, *a: False
|
||||
monkeypatch.setattr(tmpl_mod.socket, "create_connection", lambda *a, **kw: sock_m)
|
||||
|
||||
stdout_m = MagicMock()
|
||||
stdout_m.read.return_value = b"OK"
|
||||
client_m = MagicMock()
|
||||
client_m.exec_command.return_value = (MagicMock(), stdout_m, MagicMock())
|
||||
monkeypatch.setattr(tmpl_mod.paramiko, "SSHClient", lambda: client_m)
|
||||
monkeypatch.setattr(tmpl_mod.paramiko, "AutoAddPolicy", MagicMock)
|
||||
|
||||
result = CliRunner().invoke(
|
||||
cli,
|
||||
["template", "deploy-linux", "--vmx", str(vmx), "--force"],
|
||||
)
|
||||
assert result.exit_code == 0, result.output
|
||||
|
||||
assert vmx.exists()
|
||||
vmx_content = vmx.read_text()
|
||||
assert 'guestOS = "ubuntu-64"' in vmx_content
|
||||
assert 'ide1:0.present = "FALSE"' in vmx_content
|
||||
assert 'ide1:0.startConnected = "FALSE"' in vmx_content
|
||||
assert "guestinfo.userdata" in vmx_content # b64 from step 2
|
||||
assert "192.168.79.200" in result.output # guest IP in summary
|
||||
|
||||
|
||||
# ═══════════════════════════════════════════════════════════════════════════
|
||||
# template prepare-linux — helper unit tests
|
||||
# ═══════════════════════════════════════════════════════════════════════════
|
||||
|
||||
|
||||
def test_pl_run_returns_stdout_and_rc(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
stdout_m = MagicMock()
|
||||
stdout_m.read.return_value = b"hello\n"
|
||||
stdout_m.channel.recv_exit_status.return_value = 0
|
||||
stderr_m = MagicMock()
|
||||
stderr_m.read.return_value = b""
|
||||
client_m = MagicMock()
|
||||
client_m.exec_command.return_value = (MagicMock(), stdout_m, stderr_m)
|
||||
|
||||
rc, out, err = tmpl_mod._pl_run(client_m, "echo hello")
|
||||
assert rc == 0
|
||||
assert "hello" in out
|
||||
assert err == ""
|
||||
|
||||
|
||||
def test_pl_run_nonzero_rc(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
stdout_m = MagicMock()
|
||||
stdout_m.read.return_value = b""
|
||||
stdout_m.channel.recv_exit_status.return_value = 1
|
||||
stderr_m = MagicMock()
|
||||
stderr_m.read.return_value = b"error"
|
||||
client_m = MagicMock()
|
||||
client_m.exec_command.return_value = (MagicMock(), stdout_m, stderr_m)
|
||||
|
||||
rc, _, err = tmpl_mod._pl_run(client_m, "false")
|
||||
assert rc == 1
|
||||
assert "error" in err
|
||||
|
||||
|
||||
def test_pl_stream_success(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
channel_m = MagicMock()
|
||||
# First call: stdout data; second: empty → exit_status_ready True on second loop
|
||||
channel_m.recv_ready.side_effect = [True, False, False]
|
||||
channel_m.recv.return_value = b"output line\n"
|
||||
channel_m.recv_stderr_ready.return_value = False
|
||||
channel_m.exit_status_ready.side_effect = [False, True]
|
||||
channel_m.recv_exit_status.return_value = 0
|
||||
|
||||
transport_m = MagicMock()
|
||||
transport_m.open_session.return_value = channel_m
|
||||
client_m = MagicMock()
|
||||
client_m.get_transport.return_value = transport_m
|
||||
|
||||
monkeypatch.setattr(tmpl_mod.time, "sleep", lambda _: None)
|
||||
rc = tmpl_mod._pl_stream(client_m, "cat file")
|
||||
assert rc == 0
|
||||
|
||||
|
||||
def test_pl_upload_calls_sftp_put(tmp_path: Path) -> None:
|
||||
local = tmp_path / "script.sh"
|
||||
local.write_text("#!/bin/bash\n")
|
||||
sftp_m = MagicMock()
|
||||
client_m = MagicMock()
|
||||
client_m.open_sftp.return_value = sftp_m
|
||||
|
||||
tmpl_mod._pl_upload(client_m, local, "/tmp/script.sh")
|
||||
sftp_m.put.assert_called_once_with(str(local), "/tmp/script.sh")
|
||||
sftp_m.close.assert_called_once()
|
||||
|
||||
|
||||
# ═══════════════════════════════════════════════════════════════════════════
|
||||
# template prepare-linux — CLI tests
|
||||
# ═══════════════════════════════════════════════════════════════════════════
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def prepare_env(
|
||||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||||
) -> dict:
|
||||
"""Environment with SSH keys, VMX, and toolchain script."""
|
||||
ssh_key = _make_keys(tmp_path / "keys")
|
||||
vm_dir = tmp_path / "templates" / "LinuxBuild2404"
|
||||
vm_dir.mkdir(parents=True)
|
||||
vmx = vm_dir / "LinuxBuild2404.vmx"
|
||||
vmx.write_text('guestOS = "ubuntu-64"\n', encoding="ascii")
|
||||
|
||||
# toolchain script in repo-relative location
|
||||
tmpl_dir = tmp_path / "template"
|
||||
tmpl_dir.mkdir()
|
||||
toolchain = tmpl_dir / "Install-CIToolchain-Linux2404.sh"
|
||||
toolchain.write_text("#!/bin/bash\necho done\n")
|
||||
|
||||
config = _deploy_config(tmp_path)
|
||||
monkeypatch.setattr(tmpl_mod, "load_config", lambda: config)
|
||||
monkeypatch.setattr(tmpl_mod.time, "sleep", lambda _: None)
|
||||
monkeypatch.chdir(tmp_path)
|
||||
|
||||
return {
|
||||
"tmp_path": tmp_path,
|
||||
"ssh_key": ssh_key,
|
||||
"vmx": vmx,
|
||||
"toolchain": toolchain,
|
||||
}
|
||||
|
||||
|
||||
def _stub_prepare(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
env: dict,
|
||||
*,
|
||||
toolchain_exit: int = 0,
|
||||
val_output: str = (
|
||||
"UserSudo:OK\nSudoNoPass:OK\nTool:gcc:OK\nTool:g++:OK\n"
|
||||
"Tool:clang:OK\nTool:cmake:OK\nTool:python3:OK\nTool:git:OK\n"
|
||||
"Tool:7z:OK\nDir:build:OK\nDir:output:OK\nDir:scripts:OK\n"
|
||||
"Swap:off:OK\nSshd:active:OK\nSSHPwdAuth:disabled:OK\n"
|
||||
"CIReportIP:enabled:OK\nCIReportIPScript:exists:OK\n"
|
||||
),
|
||||
machine_id_size: str = "0",
|
||||
) -> MagicMock:
|
||||
"""Stub all SSH / vmrun calls for prepare-linux happy-path testing."""
|
||||
client_m = MagicMock()
|
||||
raw = env["toolchain"].read_bytes()
|
||||
if raw[:3] == b"\xef\xbb\xbf":
|
||||
raw = raw[3:]
|
||||
_toolchain_size = str(len(raw.replace(b"\r\n", b"\n").replace(b"\r", b"\n")))
|
||||
|
||||
def fake_pl_run(
|
||||
client: object, cmd: str, timeout: float = 60.0
|
||||
) -> tuple[int, str, str]:
|
||||
if "stat -c %s /tmp/" in cmd:
|
||||
return 0, _toolchain_size, ""
|
||||
if "stat -c %s /etc/machine-id" in cmd:
|
||||
return 0, machine_id_size, ""
|
||||
if "truncate" in cmd:
|
||||
return 0, "", ""
|
||||
if "shutdown" in cmd:
|
||||
return 0, "", ""
|
||||
if "echo connected" in cmd:
|
||||
return 0, "connected", ""
|
||||
# validation script
|
||||
return 0, val_output, ""
|
||||
|
||||
monkeypatch.setattr(tmpl_mod, "_pl_connect", lambda *a, **kw: client_m)
|
||||
monkeypatch.setattr(tmpl_mod, "_pl_run", fake_pl_run)
|
||||
monkeypatch.setattr(tmpl_mod, "_pl_stream", lambda *a, **kw: toolchain_exit)
|
||||
monkeypatch.setattr(tmpl_mod, "_pl_upload", lambda *a, **kw: None)
|
||||
monkeypatch.setattr(tmpl_mod, "_wait_ssh_port_deploy", lambda *a, **kw: True)
|
||||
return client_m
|
||||
|
||||
|
||||
def test_prepare_linux_missing_vmx_and_ip(
|
||||
prepare_env: dict, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
result = CliRunner().invoke(cli, ["template", "prepare-linux"])
|
||||
assert result.exit_code != 0
|
||||
assert "--vmx" in result.output or "--ip" in result.output
|
||||
|
||||
|
||||
def test_prepare_linux_toolchain_not_found(
|
||||
prepare_env: dict, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
prepare_env["toolchain"].unlink()
|
||||
result = CliRunner().invoke(
|
||||
cli,
|
||||
["template", "prepare-linux", "--ip", "192.168.1.100"],
|
||||
)
|
||||
assert result.exit_code != 0
|
||||
assert "not found" in result.output.lower()
|
||||
|
||||
|
||||
def test_prepare_linux_ssh_key_missing(
|
||||
prepare_env: dict, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
prepare_env["ssh_key"].unlink()
|
||||
result = CliRunner().invoke(
|
||||
cli,
|
||||
["template", "prepare-linux", "--ip", "192.168.1.100"],
|
||||
)
|
||||
assert result.exit_code != 0
|
||||
assert "not found" in result.output.lower()
|
||||
|
||||
|
||||
def test_prepare_linux_ssh_connect_fail(
|
||||
prepare_env: dict, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
monkeypatch.setattr(
|
||||
tmpl_mod, "_wait_ssh_port_deploy", lambda *a, **kw: True
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
tmpl_mod, "_pl_connect",
|
||||
lambda *a, **kw: (_ for _ in ()).throw(OSError("refused")),
|
||||
)
|
||||
result = CliRunner().invoke(
|
||||
cli,
|
||||
["template", "prepare-linux", "--ip", "192.168.1.100"],
|
||||
)
|
||||
assert result.exit_code != 0
|
||||
assert "failed" in result.output.lower()
|
||||
|
||||
|
||||
def test_prepare_linux_ssh_port_timeout(
|
||||
prepare_env: dict, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
monkeypatch.setattr(tmpl_mod, "_wait_ssh_port_deploy", lambda *a, **kw: False)
|
||||
result = CliRunner().invoke(
|
||||
cli,
|
||||
["template", "prepare-linux", "--ip", "192.168.1.100"],
|
||||
)
|
||||
assert result.exit_code != 0
|
||||
assert "not reachable" in result.output or "SSH" in result.output
|
||||
|
||||
|
||||
def test_prepare_linux_toolchain_fail(
|
||||
prepare_env: dict, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
_stub_prepare(monkeypatch, prepare_env, toolchain_exit=1)
|
||||
result = CliRunner().invoke(
|
||||
cli,
|
||||
["template", "prepare-linux", "--ip", "192.168.1.100"],
|
||||
)
|
||||
assert result.exit_code != 0
|
||||
assert "exited 1" in result.output or "Toolchain" in result.output
|
||||
|
||||
|
||||
def test_prepare_linux_validation_fail(
|
||||
prepare_env: dict, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
_stub_prepare(
|
||||
monkeypatch, prepare_env,
|
||||
val_output="UserSudo:OK\nTool:gcc:FAIL\nDir:build:OK\n",
|
||||
)
|
||||
result = CliRunner().invoke(
|
||||
cli,
|
||||
["template", "prepare-linux", "--ip", "192.168.1.100"],
|
||||
)
|
||||
assert result.exit_code != 0
|
||||
assert "FAIL" in result.output
|
||||
|
||||
|
||||
def test_prepare_linux_machine_id_not_cleared(
|
||||
prepare_env: dict, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
_stub_prepare(monkeypatch, prepare_env, machine_id_size="36")
|
||||
result = CliRunner().invoke(
|
||||
cli,
|
||||
["template", "prepare-linux", "--ip", "192.168.1.100"],
|
||||
)
|
||||
assert result.exit_code != 0
|
||||
assert "machine-id" in result.output.lower()
|
||||
|
||||
|
||||
def test_prepare_linux_happy_path_ip_only(
|
||||
prepare_env: dict, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
"""No --vmx: provisions but skips auto power-on, IP detect, and snapshot."""
|
||||
_stub_prepare(monkeypatch, prepare_env)
|
||||
result = CliRunner().invoke(
|
||||
cli,
|
||||
["template", "prepare-linux", "--ip", "192.168.1.100"],
|
||||
)
|
||||
assert result.exit_code == 0, result.output
|
||||
assert "Provisioning complete" in result.output
|
||||
assert "not taken" in result.output # snapshot skipped
|
||||
|
||||
|
||||
def test_prepare_linux_happy_path_with_vmx(
|
||||
prepare_env: dict, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
"""--vmx: auto IP detect + shutdown + snapshot."""
|
||||
_stub_prepare(monkeypatch, prepare_env)
|
||||
monkeypatch.setattr(tmpl_mod, "_poll_guest_ip_deploy", lambda *a, **kw: "192.168.1.200")
|
||||
|
||||
backend_m = MagicMock()
|
||||
backend_m.is_running.side_effect = [False, True, False] # start, check, powered off
|
||||
backend_m.list_snapshots.return_value = []
|
||||
monkeypatch.setattr(tmpl_mod, "WorkstationVmrunBackend", lambda **kw: backend_m)
|
||||
|
||||
import subprocess as _sub
|
||||
monkeypatch.setattr(
|
||||
tmpl_mod.subprocess, "run",
|
||||
lambda cmd, **kw: _sub.CompletedProcess(list(cmd), 0, "", ""),
|
||||
)
|
||||
|
||||
result = CliRunner().invoke(
|
||||
cli,
|
||||
["template", "prepare-linux", "--vmx", str(prepare_env["vmx"])],
|
||||
)
|
||||
assert result.exit_code == 0, result.output
|
||||
assert "Snapshot" in result.output
|
||||
assert "192.168.1.200" in result.output
|
||||
|
||||
|
||||
def test_prepare_linux_happy_path_skip_flags(
|
||||
prepare_env: dict, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
"""--skip-update, --install-dotnet, --no-mingw reach the toolchain command."""
|
||||
streamed: list[str] = []
|
||||
client_m = MagicMock()
|
||||
|
||||
_raw = prepare_env["toolchain"].read_bytes()
|
||||
_tsize = str(len(_raw.replace(b"\r\n", b"\n").replace(b"\r", b"\n")))
|
||||
|
||||
def fake_pl_run(client: object, cmd: str, timeout: float = 60.0) -> tuple[int, str, str]:
|
||||
if "stat -c %s /tmp/" in cmd:
|
||||
return 0, _tsize, ""
|
||||
if "stat -c %s /etc/machine-id" in cmd:
|
||||
return 0, "0", ""
|
||||
if "truncate" in cmd:
|
||||
return 0, "", ""
|
||||
if "echo connected" in cmd:
|
||||
return 0, "connected", ""
|
||||
return 0, "UserSudo:OK\nSudoNoPass:OK\n", ""
|
||||
|
||||
def fake_pl_stream(client: object, cmd: str) -> int:
|
||||
streamed.append(cmd)
|
||||
return 0
|
||||
|
||||
monkeypatch.setattr(tmpl_mod, "_pl_connect", lambda *a, **kw: client_m)
|
||||
monkeypatch.setattr(tmpl_mod, "_pl_run", fake_pl_run)
|
||||
monkeypatch.setattr(tmpl_mod, "_pl_stream", fake_pl_stream)
|
||||
monkeypatch.setattr(tmpl_mod, "_pl_upload", lambda *a, **kw: None)
|
||||
monkeypatch.setattr(tmpl_mod, "_wait_ssh_port_deploy", lambda *a, **kw: True)
|
||||
|
||||
result = CliRunner().invoke(
|
||||
cli,
|
||||
[
|
||||
"template", "prepare-linux",
|
||||
"--ip", "192.168.1.100",
|
||||
"--skip-update", "--install-dotnet", "--no-mingw",
|
||||
],
|
||||
)
|
||||
assert result.exit_code == 0, result.output
|
||||
assert streamed
|
||||
assert "--skip-update" in streamed[0]
|
||||
assert "--dotnet" in streamed[0]
|
||||
assert "--no-mingw" in streamed[0]
|
||||
|
||||
|
||||
def test_prepare_linux_existing_snapshot_deleted(
|
||||
prepare_env: dict, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
"""If snapshot already exists, delete before recreating."""
|
||||
_stub_prepare(monkeypatch, prepare_env)
|
||||
monkeypatch.setattr(tmpl_mod, "_poll_guest_ip_deploy", lambda *a, **kw: "192.168.1.200")
|
||||
|
||||
backend_m = MagicMock()
|
||||
backend_m.is_running.side_effect = [True, False]
|
||||
backend_m.list_snapshots.return_value = ["BaseClean-Linux"]
|
||||
monkeypatch.setattr(tmpl_mod, "WorkstationVmrunBackend", lambda **kw: backend_m)
|
||||
|
||||
run_calls: list[list[str]] = []
|
||||
import subprocess as _sub
|
||||
|
||||
def fake_run(cmd: list[str], **kw: object) -> _sub.CompletedProcess[str]:
|
||||
run_calls.append(list(cmd))
|
||||
return _sub.CompletedProcess(cmd, 0, "", "")
|
||||
|
||||
monkeypatch.setattr(tmpl_mod.subprocess, "run", fake_run)
|
||||
|
||||
result = CliRunner().invoke(
|
||||
cli,
|
||||
["template", "prepare-linux", "--vmx", str(prepare_env["vmx"])],
|
||||
)
|
||||
assert result.exit_code == 0, result.output
|
||||
cmds = [" ".join(c) for c in run_calls]
|
||||
assert any("deleteSnapshot" in c for c in cmds)
|
||||
assert any("snapshot" in c and "deleteSnapshot" not in c for c in cmds)
|
||||
|
||||
Reference in New Issue
Block a user