diff --git a/tests/python/test_backends_protocol.py b/tests/python/test_backends_protocol.py new file mode 100644 index 0000000..03b05f7 --- /dev/null +++ b/tests/python/test_backends_protocol.py @@ -0,0 +1,54 @@ +"""Tests for backends/protocol.py. + +Covers VmState, VmHandle, and the Protocol stub bodies (the ``...`` +ellipsis in each method), which exist as documentation-level defaults +and are callable on the Protocol class itself. +""" + +from __future__ import annotations + +import pytest + +from ci_orchestrator.backends.protocol import VmBackend, VmHandle, VmState + + +def test_vm_state_values() -> None: + assert VmState.POWERED_OFF == "powered_off" + assert VmState.POWERED_ON == "powered_on" + assert VmState.SUSPENDED == "suspended" + assert VmState.UNKNOWN == "unknown" + + +def test_vm_handle_fields() -> None: + h = VmHandle(identifier="x.vmx", name="job-1") + assert h.identifier == "x.vmx" + assert h.name == "job-1" + + +def test_vm_handle_name_defaults_to_none() -> None: + h = VmHandle(identifier="x.vmx") + assert h.name is None + + +def test_vm_handle_is_frozen() -> None: + h = VmHandle("x.vmx") + with pytest.raises((AttributeError, TypeError)): + h.identifier = "y.vmx" # type: ignore[misc] + + +# ── Protocol stub bodies (lines 53, 57, 61, 65, 69, 73, 81) ────────────────── +# The Protocol stubs are callable as default implementations that return None. +# Exercising them satisfies coverage without asserting behaviour (they are +# documentation stubs, not functional implementations). + +def test_protocol_stubs_are_callable() -> None: + # Protocol.__init__ blocks VmBackend() in Python 3.12; bypass via __new__. + b: VmBackend = object.__new__(VmBackend) + h = VmHandle("x.vmx") + assert b.clone_linked("tpl.vmx", "snap", "name") is None # line 53 + assert b.start(h) is None # line 57 + assert b.stop(h) is None # line 61 + assert b.delete(h) is None # line 65 + assert b.get_ip(h) is None # line 69 + assert b.list_snapshots(h) is None # line 73 + assert b.is_running(h) is None # line 81 diff --git a/tests/python/test_workstation_backend.py b/tests/python/test_workstation_backend.py index f4bc977..b6c7359 100644 --- a/tests/python/test_workstation_backend.py +++ b/tests/python/test_workstation_backend.py @@ -160,3 +160,205 @@ def test_is_running_returns_false_when_absent( ) backend = WorkstationVmrunBackend(vmrun_path=fake_vmrun) assert backend.is_running(VmHandle(str(vmx))) is False + + +# ── _resolve_vmrun(explicit=None) paths ─────────────────────────────────────── + +def test_resolve_vmrun_uses_which(monkeypatch: pytest.MonkeyPatch, fake_vmrun: str) -> None: + import shutil as _shutil + monkeypatch.setattr(_shutil, "which", lambda _name: fake_vmrun) + backend = WorkstationVmrunBackend() # no explicit path + assert backend.vmrun_path == fake_vmrun + + +def test_resolve_vmrun_uses_default_win_path( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + import shutil as _shutil + from ci_orchestrator.backends import workstation as ws_mod + + fake_default = tmp_path / "vmrun.exe" + fake_default.write_bytes(b"") + monkeypatch.setattr(_shutil, "which", lambda _name: None) + monkeypatch.setattr(ws_mod, "_DEFAULT_WIN_VMRUN", str(fake_default)) + backend = WorkstationVmrunBackend() + assert backend.vmrun_path == str(fake_default) + + +def test_resolve_vmrun_raises_when_nothing_found(monkeypatch: pytest.MonkeyPatch) -> None: + import shutil as _shutil + from ci_orchestrator.backends import workstation as ws_mod + + monkeypatch.setattr(_shutil, "which", lambda _name: None) + monkeypatch.setattr(ws_mod, "_DEFAULT_WIN_VMRUN", "/nonexistent/vmrun.exe") + with pytest.raises(BackendNotAvailable): + WorkstationVmrunBackend() + + +# ── clone_linked destination=None ──────────────────────────────────────────── + +def test_clone_linked_auto_destination(monkeypatch: pytest.MonkeyPatch, fake_vmrun: str) -> None: + captured: dict[str, list[str]] = {} + + def fake_run(cmd: list[str], **_kw: object) -> Any: + captured["cmd"] = cmd + return _make_completed(0) + + monkeypatch.setattr(subprocess, "run", fake_run) + backend = WorkstationVmrunBackend(vmrun_path=fake_vmrun) + handle = backend.clone_linked( + template=r"/ci/tpl/WinBuild2025/WinBuild2025.vmx", + snapshot="BaseClean", + name="job-auto", + ) + # destination auto-computed as //.vmx + assert handle.name == "job-auto" + assert handle.identifier.endswith("job-auto.vmx") + + +# ── start headless=False ───────────────────────────────────────────────────── + +def test_start_gui_mode(monkeypatch: pytest.MonkeyPatch, fake_vmrun: str) -> None: + captured: list[str] = [] + monkeypatch.setattr( + subprocess, "run", + lambda cmd, **_kw: (captured.append(cmd[-1]) or _make_completed(0)), + ) + backend = WorkstationVmrunBackend(vmrun_path=fake_vmrun) + backend.start(VmHandle("x.vmx"), headless=False) + assert captured[-1] == "gui" + + +# ── stop edge cases ─────────────────────────────────────────────────────────── + +def test_stop_succeeds_on_rc_zero(monkeypatch: pytest.MonkeyPatch, fake_vmrun: str) -> None: + monkeypatch.setattr(subprocess, "run", lambda *a, **kw: _make_completed(0)) + backend = WorkstationVmrunBackend(vmrun_path=fake_vmrun) + backend.stop(VmHandle("x.vmx")) # must not raise + + +def test_stop_raises_on_unrecognized_error(monkeypatch: pytest.MonkeyPatch, fake_vmrun: str) -> None: + monkeypatch.setattr( + subprocess, "run", + lambda *a, **kw: _make_completed(1, stderr="Error: something unexpected"), + ) + backend = WorkstationVmrunBackend(vmrun_path=fake_vmrun) + with pytest.raises(BackendOperationFailed): + backend.stop(VmHandle("x.vmx")) + + +def test_stop_tolerates_is_not_running(monkeypatch: pytest.MonkeyPatch, fake_vmrun: str) -> None: + monkeypatch.setattr( + subprocess, "run", + lambda *a, **kw: _make_completed(1, stderr="Error: vm is not running"), + ) + backend = WorkstationVmrunBackend(vmrun_path=fake_vmrun) + backend.stop(VmHandle("x.vmx")) # must not raise + + +# ── delete ──────────────────────────────────────────────────────────────────── + +def test_delete_invokes_delete_vm(monkeypatch: pytest.MonkeyPatch, fake_vmrun: str) -> None: + ops: list[str] = [] + monkeypatch.setattr( + subprocess, "run", + lambda cmd, **_kw: (ops.append(cmd[3]) or _make_completed(0)), + ) + backend = WorkstationVmrunBackend(vmrun_path=fake_vmrun) + backend.delete(VmHandle("x.vmx")) + assert ops == ["deleteVM"] + + +# ── get_ip timeout / TimeoutExpired / empty stdout ─────────────────────────── + +def test_get_ip_passes_wait_when_timeout_positive( + monkeypatch: pytest.MonkeyPatch, fake_vmrun: str +) -> None: + captured: list[list[str]] = [] + monkeypatch.setattr( + subprocess, "run", + lambda cmd, **_kw: (captured.append(cmd) or _make_completed(0, stdout="10.0.0.1\n")), + ) + backend = WorkstationVmrunBackend(vmrun_path=fake_vmrun) + ip = backend.get_ip(VmHandle("x.vmx"), timeout=5.0) + assert ip == "10.0.0.1" + assert "-wait" in captured[0] + + +def test_get_ip_returns_none_on_timeout_expired( + monkeypatch: pytest.MonkeyPatch, fake_vmrun: str +) -> None: + def boom(*_a: object, **_kw: object) -> Any: + raise subprocess.TimeoutExpired(cmd="vmrun", timeout=1) + + monkeypatch.setattr(subprocess, "run", boom) + backend = WorkstationVmrunBackend(vmrun_path=fake_vmrun) + assert backend.get_ip(VmHandle("x.vmx"), timeout=1.0) is None + + +def test_get_ip_returns_none_on_empty_stdout( + monkeypatch: pytest.MonkeyPatch, fake_vmrun: str +) -> None: + monkeypatch.setattr(subprocess, "run", lambda *a, **kw: _make_completed(0, stdout=" \n")) + backend = WorkstationVmrunBackend(vmrun_path=fake_vmrun) + assert backend.get_ip(VmHandle("x.vmx")) is None + + +# ── list_snapshots without header ───────────────────────────────────────────── + +def test_list_snapshots_without_header(monkeypatch: pytest.MonkeyPatch, fake_vmrun: str) -> None: + monkeypatch.setattr( + subprocess, "run", + lambda *a, **kw: _make_completed(0, stdout="BaseClean\n"), + ) + backend = WorkstationVmrunBackend(vmrun_path=fake_vmrun) + assert backend.list_snapshots(VmHandle("x.vmx")) == ["BaseClean"] + + +# ── is_running edge cases ───────────────────────────────────────────────────── + +def test_is_running_returns_false_on_list_failure( + monkeypatch: pytest.MonkeyPatch, fake_vmrun: str +) -> None: + monkeypatch.setattr(subprocess, "run", lambda *a, **kw: _make_completed(1, stderr="err")) + backend = WorkstationVmrunBackend(vmrun_path=fake_vmrun) + assert backend.is_running(VmHandle("x.vmx")) is False + + +def test_is_running_skips_non_matching_line( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path, fake_vmrun: str +) -> None: + target = tmp_path / "job1.vmx" + target.write_text("") + other = tmp_path / "other.vmx" + other.write_text("") + monkeypatch.setattr( + subprocess, "run", + lambda *a, **kw: _make_completed(0, stdout=f"Total running VMs: 1\n{other}\n"), + ) + backend = WorkstationVmrunBackend(vmrun_path=fake_vmrun) + assert backend.is_running(VmHandle(str(target))) is False + + +def test_is_running_handles_oserror_on_bad_path( + monkeypatch: pytest.MonkeyPatch, fake_vmrun: str +) -> None: + import pathlib + + _real_resolve = pathlib.Path.resolve + + def _patched_resolve(self: Path, *args: object, **kw: object) -> Path: + if "trigger_oserror" in str(self): + raise OSError("EACCES: permission denied") + return _real_resolve(self, *args, **kw) # type: ignore[return-value] + + monkeypatch.setattr(pathlib.Path, "resolve", _patched_resolve) + monkeypatch.setattr( + subprocess, "run", + lambda *a, **kw: _make_completed( + 0, stdout="Total running VMs: 1\n/some/trigger_oserror.vmx\n" + ), + ) + backend = WorkstationVmrunBackend(vmrun_path=fake_vmrun) + # OSError from Path.resolve() must be swallowed; VM not found → False. + assert backend.is_running(VmHandle("x.vmx")) is False