clan_cli: Made qmp implementation lazy
This commit is contained in:
@@ -1,11 +1,11 @@
|
||||
import json
|
||||
import logging
|
||||
from collections.abc import Generator
|
||||
from contextlib import contextmanager
|
||||
from os import path
|
||||
from pathlib import Path
|
||||
from tempfile import NamedTemporaryFile
|
||||
from time import sleep
|
||||
from tempfile import NamedTemporaryFile, TemporaryDirectory
|
||||
|
||||
from clan_cli.dirs import vm_state_dir
|
||||
from qemu.qmp import QEMUMonitorProtocol
|
||||
|
||||
from ..cmd import run
|
||||
@@ -16,6 +16,39 @@ from ..ssh import Host, parse_deployment_address
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class VMAttr:
|
||||
def __init__(self, machine_name: str) -> None:
|
||||
self.temp_dir = TemporaryDirectory(prefix="clan_vm-", suffix=f"-{machine_name}")
|
||||
self._qmp_socket: Path = Path(self.temp_dir.name) / "qmp.sock"
|
||||
self._qga_socket: Path = Path(self.temp_dir.name) / "qga.sock"
|
||||
self._qmp: QEMUMonitorProtocol | None = None
|
||||
|
||||
@contextmanager
|
||||
def qmp(self) -> Generator[QEMUMonitorProtocol, None, None]:
|
||||
if self._qmp is None:
|
||||
log.debug(f"qmp_socket: {self._qmp_socket}")
|
||||
self._qmp = QEMUMonitorProtocol(path.realpath(self._qmp_socket))
|
||||
self._qmp.connect()
|
||||
try:
|
||||
yield self._qmp
|
||||
finally:
|
||||
self._qmp.close()
|
||||
|
||||
@property
|
||||
def qmp_socket(self) -> Path:
|
||||
if self._qmp is None:
|
||||
log.debug(f"qmp_socket: {self._qmp_socket}")
|
||||
self._qmp = QEMUMonitorProtocol(path.realpath(self._qmp_socket))
|
||||
return self._qmp_socket
|
||||
|
||||
@property
|
||||
def qga_socket(self) -> Path:
|
||||
if self._qmp is None:
|
||||
log.debug(f"qmp_socket: {self.qga_socket}")
|
||||
self._qmp = QEMUMonitorProtocol(path.realpath(self._qmp_socket))
|
||||
return self._qga_socket
|
||||
|
||||
|
||||
class Machine:
|
||||
def __init__(
|
||||
self,
|
||||
@@ -36,14 +69,8 @@ class Machine:
|
||||
self.build_cache: dict[str, Path] = {}
|
||||
|
||||
self._deployment_info: None | dict[str, str] = deployment_info
|
||||
state_dir = vm_state_dir(flake_url=str(self.flake), vm_name=self.name)
|
||||
|
||||
self.qmp_socket: Path = state_dir / "qmp.sock"
|
||||
self.qga_socket: Path = state_dir / "qga.sock"
|
||||
|
||||
log.debug(f"qmp_socket: {self.qmp_socket}")
|
||||
self._qmp = QEMUMonitorProtocol(path.realpath(self.qmp_socket))
|
||||
self._qmp_connected = False
|
||||
self.vm: VMAttr = VMAttr(name)
|
||||
|
||||
def __str__(self) -> str:
|
||||
return f"Machine(name={self.name}, flake={self.flake})"
|
||||
@@ -60,28 +87,6 @@ class Machine:
|
||||
)
|
||||
return self._deployment_info
|
||||
|
||||
def qmp_connect(self) -> None:
|
||||
if not self._qmp_connected:
|
||||
tries = 100
|
||||
for num in range(tries):
|
||||
try:
|
||||
# the socket file link might be outdated, therefore re-init the qmp object
|
||||
self._qmp = QEMUMonitorProtocol(path.realpath(self.qmp_socket))
|
||||
self._qmp.connect()
|
||||
self._qmp_connected = True
|
||||
log.debug("QMP Connected")
|
||||
return
|
||||
except FileNotFoundError:
|
||||
if num < 99:
|
||||
sleep(0.1)
|
||||
continue
|
||||
else:
|
||||
raise
|
||||
|
||||
def qmp_command(self, command: str) -> dict:
|
||||
self.qmp_connect()
|
||||
return self._qmp.command(command)
|
||||
|
||||
@property
|
||||
def target_host_address(self) -> str:
|
||||
# deploymentAddress is deprecated.
|
||||
|
||||
@@ -146,7 +146,9 @@ def qemu_command(
|
||||
|
||||
|
||||
# TODO move this to the Machines class
|
||||
def build_vm(machine: Machine, vm: VmConfig, nix_options: list[str]) -> dict[str, str]:
|
||||
def build_vm(
|
||||
machine: Machine, vm: VmConfig, nix_options: list[str] = []
|
||||
) -> dict[str, str]:
|
||||
config = nix_config()
|
||||
system = config["system"]
|
||||
|
||||
|
||||
@@ -1,7 +1,10 @@
|
||||
import os
|
||||
import sys
|
||||
import tempfile
|
||||
import threading
|
||||
import traceback
|
||||
from collections.abc import Generator
|
||||
from contextlib import contextmanager
|
||||
from pathlib import Path
|
||||
from time import sleep
|
||||
from typing import TYPE_CHECKING
|
||||
@@ -21,6 +24,34 @@ if TYPE_CHECKING:
|
||||
no_kvm = not os.path.exists("/dev/kvm")
|
||||
|
||||
|
||||
@contextmanager
|
||||
def monkeypatch_tempdir_with_custom_path(
|
||||
*, monkeypatch: pytest.MonkeyPatch, custom_path: str, prefix_condition: str
|
||||
) -> Generator[None, None, None]:
|
||||
# Custom wrapper function that checks the prefix and either modifies the behavior or falls back to the original
|
||||
class CustomTemporaryDirectory(tempfile.TemporaryDirectory):
|
||||
def __init__(
|
||||
self,
|
||||
suffix: str | None = None,
|
||||
prefix: str | None = None,
|
||||
dir: str | None = None, # noqa: A002
|
||||
) -> None:
|
||||
if prefix == prefix_condition:
|
||||
self.name = custom_path # Use the custom path
|
||||
self._finalizer = None # Prevent cleanup attempts on the custom path by the original finalizer
|
||||
else:
|
||||
super().__init__(suffix=suffix, prefix=prefix, dir=dir)
|
||||
|
||||
# Use ExitStack to ensure unpatching
|
||||
try:
|
||||
# Patch the TemporaryDirectory with our custom class
|
||||
monkeypatch.setattr(tempfile, "TemporaryDirectory", CustomTemporaryDirectory)
|
||||
yield # This allows the code within the 'with' block of this context manager to run
|
||||
finally:
|
||||
# Unpatch the TemporaryDirectory
|
||||
monkeypatch.undo()
|
||||
|
||||
|
||||
def run_vm_in_thread(machine_name: str) -> None:
|
||||
# runs machine and prints exceptions
|
||||
def run() -> None:
|
||||
@@ -125,22 +156,26 @@ def test_vm_qmp(
|
||||
# 'clan vms run' must be executed from within the flake
|
||||
monkeypatch.chdir(flake.path)
|
||||
|
||||
# the state dir is a point of reference for qemu interactions as it links to the qga/qmp sockets
|
||||
state_dir = vm_state_dir(str(flake.path), "my_machine")
|
||||
with monkeypatch_tempdir_with_custom_path(
|
||||
monkeypatch=monkeypatch,
|
||||
custom_path=str(temporary_home / "vm-tmp"),
|
||||
prefix_condition="clan_vm-",
|
||||
):
|
||||
# the state dir is a point of reference for qemu interactions as it links to the qga/qmp sockets
|
||||
state_dir = vm_state_dir(str(flake.path), "my_machine")
|
||||
# start the VM
|
||||
run_vm_in_thread("my_machine")
|
||||
|
||||
# start the VM
|
||||
run_vm_in_thread("my_machine")
|
||||
# connect with qmp
|
||||
qmp = qmp_connect(state_dir)
|
||||
|
||||
# connect with qmp
|
||||
qmp = qmp_connect(state_dir)
|
||||
# verify that issuing a command works
|
||||
# result = qmp.cmd_obj({"execute": "query-status"})
|
||||
result = qmp.command("query-status")
|
||||
assert result["status"] == "running", result
|
||||
|
||||
# verify that issuing a command works
|
||||
# result = qmp.cmd_obj({"execute": "query-status"})
|
||||
result = qmp.command("query-status")
|
||||
assert result["status"] == "running", result
|
||||
|
||||
# shutdown machine (prevent zombie qemu processes)
|
||||
qmp.command("system_powerdown")
|
||||
# shutdown machine (prevent zombie qemu processes)
|
||||
qmp.command("system_powerdown")
|
||||
|
||||
|
||||
@pytest.mark.skipif(no_kvm, reason="Requires KVM")
|
||||
|
||||
Reference in New Issue
Block a user