refactor(backend): add list_running() to VmBackend seam (W1)

operations.vms.list_live_vms needs a backend-agnostic running-VM list for the web dashboard; is_running now shares the single vmrun-list parse.
This commit is contained in:
2026-07-23 15:22:09 +02:00
parent 11a80d26f0
commit b9f6db6bd5
4 changed files with 66 additions and 9 deletions
+11
View File
@@ -79,3 +79,14 @@ class VmBackend(Protocol):
getGuestIPAddress``) for this check — see ``AGENTS.md`` error #10.
"""
...
def list_running(self) -> list[VmHandle]:
"""Return a handle for every VM the backend reports as running.
Powers the read-only inventory in the web UI and orphan
reconciliation. Like :meth:`is_running`, implementations MUST NOT
rely on guest tools (see ``AGENTS.md`` error #10) — a plain
power-state listing is enough. Handle ``identifier`` is the
backend-native reference (a VMX path on Workstation).
"""
...
+24 -9
View File
@@ -140,22 +140,37 @@ class WorkstationVmrunBackend:
lines = lines[1:]
return lines
def list_running(self) -> list[VmHandle]:
"""``vmrun list`` → one :class:`VmHandle` per running VMX path.
Skips the ``Total running VMs: N`` header and any ``Error:`` lines
(``AGENTS.md`` error #9). Does NOT call ``getGuestIPAddress`` (error
#10). The handle ``name`` is the VMX file stem. Returns ``[]`` when
``vmrun list`` fails.
"""
result = self._run("list", check=False)
if result.returncode != 0:
return []
handles: list[VmHandle] = []
for raw in (result.stdout or "").splitlines():
line = raw.strip()
low = line.lower()
if not line or low.startswith("total running") or low.startswith("error:"):
continue
handles.append(VmHandle(identifier=line, name=Path(line).stem))
return handles
def is_running(self, handle: VmHandle) -> bool:
"""``vmrun list`` lookup — does NOT call ``getGuestIPAddress``.
See ``AGENTS.md`` error #10: ``getGuestIPAddress`` blocks for 30-60s
without VMware Tools. ``vmrun list`` returns immediately.
without VMware Tools. ``vmrun list`` returns immediately. Shares the
single parse in :meth:`list_running`.
"""
result = self._run("list", check=False)
if result.returncode != 0:
return False
target = Path(handle.identifier).resolve()
for line in (result.stdout or "").splitlines():
line = line.strip()
if not line or line.lower().startswith("total running"):
continue
for running in self.list_running():
try:
if Path(line).resolve() == target:
if Path(running.identifier).resolve() == target:
return True
except OSError:
continue