From 7377ad5dd8020dc2526b3e800a35a1addab1092c Mon Sep 17 00:00:00 2001 From: Ryan Malloy Date: Sun, 16 Aug 2026 21:05:02 -0600 Subject: [PATCH] Add test suite: fake QMP client, launcher table tests, real qemu-img and boot integration --- pyproject.toml | 4 + src/mcqemu/config.py | 2 +- src/mcqemu/launcher.py | 34 +++++--- src/mcqemu/models.py | 2 +- src/mcqemu/qga.py | 4 +- src/mcqemu/qmp.py | 7 +- src/mcqemu/registry.py | 5 +- src/mcqemu/tools/guest.py | 4 +- src/mcqemu/tools/lifecycle.py | 6 +- tests/conftest.py | 109 ++++++++++++++++++++++++ tests/test_guest.py | 120 +++++++++++++++++++++++++++ tests/test_images.py | 87 +++++++++++++++++++ tests/test_integration.py | 57 +++++++++++++ tests/test_launcher.py | 128 ++++++++++++++++++++++++++++ tests/test_lifecycle.py | 152 ++++++++++++++++++++++++++++++++++ tests/test_query.py | 58 +++++++++++++ tests/test_registry.py | 69 +++++++++++++++ tests/test_snapshots.py | 64 ++++++++++++++ 18 files changed, 886 insertions(+), 26 deletions(-) create mode 100644 tests/conftest.py create mode 100644 tests/test_guest.py create mode 100644 tests/test_images.py create mode 100644 tests/test_integration.py create mode 100644 tests/test_launcher.py create mode 100644 tests/test_lifecycle.py create mode 100644 tests/test_query.py create mode 100644 tests/test_registry.py create mode 100644 tests/test_snapshots.py diff --git a/pyproject.toml b/pyproject.toml index be330a2..ae77aed 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -46,6 +46,10 @@ src = ["src", "tests"] [tool.ruff.lint] select = ["E", "F", "I", "UP", "B", "SIM", "ASYNC"] +# ASYNC109: `timeout` params here are deliberate LLM-facing tool parameters. +# ASYNC110: polling an external process's death has no event to await. +# ASYNC240: pathlib use is microsecond stat/exists checks, not bulk I/O. +ignore = ["ASYNC109", "ASYNC110", "ASYNC240"] [tool.pytest.ini_options] asyncio_mode = "auto" diff --git a/src/mcqemu/config.py b/src/mcqemu/config.py index bf3508e..3df69a3 100644 --- a/src/mcqemu/config.py +++ b/src/mcqemu/config.py @@ -3,8 +3,8 @@ from __future__ import annotations import os -import re import platform +import re from dataclasses import dataclass from pathlib import Path diff --git a/src/mcqemu/launcher.py b/src/mcqemu/launcher.py index 4570def..7c0356b 100644 --- a/src/mcqemu/launcher.py +++ b/src/mcqemu/launcher.py @@ -6,7 +6,7 @@ import asyncio import re import shutil from dataclasses import dataclass -from datetime import datetime, timezone +from datetime import UTC, datetime from pathlib import Path from fastmcp.exceptions import ToolError @@ -78,8 +78,10 @@ def build_cmdline( ) if fw["mode"] == "pflash": args += [ - "-drive", f"if=pflash,format=raw,readonly=on,file={fw['code']}", - "-drive", f"if=pflash,format=raw,file={paths.uefi_vars}", + "-drive", + f"if=pflash,format=raw,readonly=on,file={fw['code']}", + "-drive", + f"if=pflash,format=raw,file={paths.uefi_vars}", ] else: args += ["-bios", fw["code"]] @@ -108,15 +110,23 @@ def build_cmdline( args += ["-netdev", netdev, "-device", "virtio-net-pci,netdev=net0"] args += [ - "-display", "none", - "-serial", f"file:{paths.serial_log}", - "-qmp", f"unix:{paths.qmp_socket},server=on,wait=off", - "-chardev", f"socket,id=qga0,path={paths.qga_socket},server=on,wait=off", - "-device", "virtio-serial", - "-device", "virtserialport,chardev=qga0,name=org.qemu.guest_agent.0", + "-display", + "none", + "-serial", + f"file:{paths.serial_log}", + "-qmp", + f"unix:{paths.qmp_socket},server=on,wait=off", + "-chardev", + f"socket,id=qga0,path={paths.qga_socket},server=on,wait=off", + "-device", + "virtio-serial", + "-device", + "virtserialport,chardev=qga0,name=org.qemu.guest_agent.0", "-daemonize", - "-pidfile", str(paths.pidfile), - "-D", str(paths.qemu_log), + "-pidfile", + str(paths.pidfile), + "-D", + str(paths.qemu_log), ] args += cfg.extra_args return args @@ -189,5 +199,5 @@ async def spawn_vm(cfg: VMConfig, config: Config) -> VMRecord: config=cfg.__dict__.copy(), qemu_log=str(paths.qemu_log), serial_log=str(paths.serial_log), - created_at=datetime.now(timezone.utc).isoformat(timespec="seconds"), + created_at=datetime.now(UTC).isoformat(timespec="seconds"), ) diff --git a/src/mcqemu/models.py b/src/mcqemu/models.py index afff675..de712af 100644 --- a/src/mcqemu/models.py +++ b/src/mcqemu/models.py @@ -2,7 +2,7 @@ from __future__ import annotations -from dataclasses import dataclass, field, asdict +from dataclasses import asdict, dataclass, field from typing import Any, Literal diff --git a/src/mcqemu/qga.py b/src/mcqemu/qga.py index 2d0c445..52452a1 100644 --- a/src/mcqemu/qga.py +++ b/src/mcqemu/qga.py @@ -12,8 +12,8 @@ from __future__ import annotations import asyncio import contextlib import secrets +from collections.abc import AsyncIterator from contextlib import asynccontextmanager -from typing import AsyncIterator from fastmcp.exceptions import ToolError from qemu.qmp import QMPClient @@ -44,7 +44,7 @@ async def qga_session(record: VMRecord) -> AsyncIterator[QMPClient]: ) if reply != token: raise guest_agent_unavailable(record.name) - except (asyncio.TimeoutError, OSError): + except (TimeoutError, OSError): raise guest_agent_unavailable(record.name) from None except ToolError: raise diff --git a/src/mcqemu/qmp.py b/src/mcqemu/qmp.py index 5fee163..9795407 100644 --- a/src/mcqemu/qmp.py +++ b/src/mcqemu/qmp.py @@ -10,8 +10,9 @@ from __future__ import annotations import asyncio import contextlib +from collections.abc import AsyncIterator from contextlib import asynccontextmanager -from typing import Any, AsyncIterator +from typing import Any from qemu.qmp import ExecuteError, QMPClient @@ -27,7 +28,7 @@ async def qmp_session(record: VMRecord) -> AsyncIterator[QMPClient]: try: try: await asyncio.wait_for(client.connect(record.qmp_socket), timeout=CONNECT_TIMEOUT) - except asyncio.TimeoutError: + except TimeoutError: raise qmp_unreachable( record.name, record.qmp_socket, "connect timed out", record.qemu_log ) from None @@ -63,5 +64,5 @@ async def wait_for_event(client: QMPClient, event_name: str, timeout: float) -> try: await asyncio.wait_for(_drain(), timeout=timeout) return True - except asyncio.TimeoutError: + except TimeoutError: return False diff --git a/src/mcqemu/registry.py b/src/mcqemu/registry.py index a0cb5c8..9b32d29 100644 --- a/src/mcqemu/registry.py +++ b/src/mcqemu/registry.py @@ -2,6 +2,7 @@ from __future__ import annotations +import contextlib import json import os from pathlib import Path @@ -66,10 +67,8 @@ class VMRegistry: def refresh_pid(self, record: VMRecord) -> int | None: """Re-read the pidfile for a spawned VM; returns the PID or None.""" if record.pidfile: - try: + with contextlib.suppress(OSError, ValueError): record.pid = int(Path(record.pidfile).read_text().strip()) - except (OSError, ValueError): - pass return record.pid def is_process_alive(self, record: VMRecord) -> bool: diff --git a/src/mcqemu/tools/guest.py b/src/mcqemu/tools/guest.py index f7ab02d..b1dfa22 100644 --- a/src/mcqemu/tools/guest.py +++ b/src/mcqemu/tools/guest.py @@ -32,9 +32,7 @@ async def guest_info(name: str, ctx: Context = None) -> dict: osinfo = await client.execute("guest-get-osinfo") except Exception: osinfo = None - supported = sorted( - c["name"] for c in agent.get("supported_commands", []) if c.get("enabled") - ) + supported = sorted(c["name"] for c in agent.get("supported_commands", []) if c.get("enabled")) return { "name": name, "agent_version": agent.get("version"), diff --git a/src/mcqemu/tools/lifecycle.py b/src/mcqemu/tools/lifecycle.py index 0ce0eff..cfe5301 100644 --- a/src/mcqemu/tools/lifecycle.py +++ b/src/mcqemu/tools/lifecycle.py @@ -193,7 +193,11 @@ async def forget_vm(name: str, force: bool = False, ctx: Context = None) -> dict "or pass force=True to deliberately orphan the process." ) state.registry.remove(name) - return {"name": name, "forgotten": True, "process_left_running": bool(record.pid and pid_alive(record.pid))} + return { + "name": name, + "forgotten": True, + "process_left_running": bool(record.pid and pid_alive(record.pid)), + } def register(mcp: FastMCP) -> None: diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..ba1b9e2 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,109 @@ +"""Shared fixtures: temp XDG dirs, scriptable fake QMP client, in-memory MCP client.""" + +import asyncio +import json +from types import SimpleNamespace + +import pytest +from fastmcp import Client + +from mcqemu.server import mcp + + +@pytest.fixture +def dirs(tmp_path, monkeypatch): + state = tmp_path / "state" + run = tmp_path / "run" + state.mkdir() + run.mkdir() + monkeypatch.setenv("MCQEMU_STATE_DIR", str(state)) + monkeypatch.setenv("MCQEMU_RUNTIME_DIR", str(run)) + return SimpleNamespace(state=state, run=run) + + +def write_registry(dirs, *records) -> None: + """Pre-seed the registry file before a Client connects (lifespan loads it).""" + (dirs.state / "vms.json").write_text(json.dumps({r.name: r.to_dict() for r in records})) + + +@pytest.fixture +async def client(dirs): + async with Client(mcp) as c: + yield c + + +def result_data(result): + # Empty-list results carry no text content block; .data covers both shapes. + if result.data is not None: + return result.data + return json.loads(result.content[0].text) + + +class FakeQMPClient: + """Stands in for qemu.qmp.QMPClient. Script it via class attributes: + + responses: dict command -> canned reply, Exception to raise, or callable(args) + events_to_emit: list of event dicts fed to the events queue + connect_error: exception raised from connect() + calls: recorded (command, arguments) tuples + """ + + responses: dict = {} + events_to_emit: list = [] + connect_error: Exception | None = None + calls: list = [] + + @classmethod + def reset(cls): + cls.responses = {} + cls.events_to_emit = [] + cls.connect_error = None + cls.calls = [] + + def __init__(self, name: str): + self.name = name + self.await_greeting = True + self.negotiate = True + self._events: asyncio.Queue = asyncio.Queue() + + async def connect(self, address): + if FakeQMPClient.connect_error is not None: + raise FakeQMPClient.connect_error + for event in FakeQMPClient.events_to_emit: + self._events.put_nowait(event) + + async def disconnect(self): + pass + + async def execute(self, command, arguments=None): + FakeQMPClient.calls.append((command, arguments)) + reply = FakeQMPClient.responses.get(command, {}) + if isinstance(reply, Exception): + raise reply + if callable(reply): + return reply(arguments) + return reply + + @property + def events(self): + return self._events + + +@pytest.fixture +def fake_qmp(monkeypatch): + FakeQMPClient.reset() + monkeypatch.setattr("mcqemu.qmp.QMPClient", FakeQMPClient) + monkeypatch.setattr("mcqemu.qga.QMPClient", FakeQMPClient) + return FakeQMPClient + + +@pytest.fixture +def all_pids_dead(monkeypatch): + monkeypatch.setattr("mcqemu.registry.pid_alive", lambda pid: False) + monkeypatch.setattr("mcqemu.tools.lifecycle.pid_alive", lambda pid: False) + + +@pytest.fixture +def all_pids_alive(monkeypatch): + monkeypatch.setattr("mcqemu.registry.pid_alive", lambda pid: bool(pid)) + monkeypatch.setattr("mcqemu.tools.lifecycle.pid_alive", lambda pid: bool(pid)) diff --git a/tests/test_guest.py b/tests/test_guest.py new file mode 100644 index 0000000..968e5f1 --- /dev/null +++ b/tests/test_guest.py @@ -0,0 +1,120 @@ +"""Guest-agent tools with a scripted fake QGA.""" + +import base64 + +import pytest +from fastmcp import Client +from fastmcp.exceptions import ToolError + +from conftest import FakeQMPClient, result_data, write_registry +from mcqemu.server import mcp +from test_lifecycle import seeded_record + + +def b64(s: str) -> str: + return base64.b64encode(s.encode()).decode() + + +def arm_sync(): + FakeQMPClient.responses["guest-sync"] = lambda args: args["id"] + + +async def test_guest_ping(dirs, fake_qmp): + arm_sync() + FakeQMPClient.responses["guest-ping"] = {} + write_registry(dirs, seeded_record(dirs, "vm1")) + async with Client(mcp) as client: + data = result_data(await client.call_tool("guest_ping", {"name": "vm1"})) + assert data["guest_agent"] == "responding" + + +async def test_guest_agent_absent_gives_actionable_error(dirs, fake_qmp): + FakeQMPClient.connect_error = ConnectionRefusedError("nobody listening") + write_registry(dirs, seeded_record(dirs, "vm1")) + async with Client(mcp) as client: + with pytest.raises(ToolError, match="qemu-guest-agent"): + await client.call_tool("guest_ping", {"name": "vm1"}) + + +async def test_guest_exec_collects_output(dirs, fake_qmp): + arm_sync() + FakeQMPClient.responses["guest-exec"] = {"pid": 77} + FakeQMPClient.responses["guest-exec-status"] = { + "exited": True, + "exitcode": 0, + "out-data": b64("Linux vm1 6.1.0\n"), + } + write_registry(dirs, seeded_record(dirs, "vm1")) + async with Client(mcp) as client: + data = result_data( + await client.call_tool( + "guest_exec", {"name": "vm1", "command": "uname", "args": ["-a"]} + ) + ) + assert data["exitcode"] == 0 + assert "Linux vm1" in data["stdout"] + exec_call = next(a for c, a in FakeQMPClient.calls if c == "guest-exec") + assert exec_call == {"path": "uname", "capture-output": True, "arg": ["-a"]} + + +async def test_guest_exec_timeout(dirs, fake_qmp): + arm_sync() + FakeQMPClient.responses["guest-exec"] = {"pid": 77} + FakeQMPClient.responses["guest-exec-status"] = {"exited": False} + write_registry(dirs, seeded_record(dirs, "vm1")) + async with Client(mcp) as client: + with pytest.raises(ToolError, match="still running"): + await client.call_tool( + "guest_exec", {"name": "vm1", "command": "sleep", "args": ["999"], "timeout": 1} + ) + + +async def test_guest_file_round_trip(dirs, fake_qmp): + arm_sync() + FakeQMPClient.responses["guest-file-open"] = 5 + FakeQMPClient.responses["guest-file-write"] = {"count": 12} + FakeQMPClient.responses["guest-file-read"] = { + "count": 12, + "buf-b64": b64("hello guest\n"), + "eof": True, + } + FakeQMPClient.responses["guest-file-close"] = {} + write_registry(dirs, seeded_record(dirs, "vm1")) + async with Client(mcp) as client: + data = result_data( + await client.call_tool( + "guest_file_write", + {"name": "vm1", "path": "/tmp/x", "content": "hello guest\n"}, + ) + ) + assert data["bytes_written"] == 12 + data = result_data( + await client.call_tool("guest_file_read", {"name": "vm1", "path": "/tmp/x"}) + ) + assert data["content"] == "hello guest\n" + assert data["truncated"] is False + open_calls = [a for c, a in FakeQMPClient.calls if c == "guest-file-open"] + assert {"path": "/tmp/x", "mode": "w"} in open_calls + assert {"path": "/tmp/x", "mode": "r"} in open_calls + assert len([c for c, _ in FakeQMPClient.calls if c == "guest-file-close"]) == 2 + + +async def test_guest_session_disables_negotiation(dirs, fake_qmp): + arm_sync() + FakeQMPClient.responses["guest-ping"] = {} + seen = {} + original_connect = FakeQMPClient.connect + + async def spy_connect(self, address): + seen["negotiate"] = self.negotiate + seen["await_greeting"] = self.await_greeting + await original_connect(self, address) + + FakeQMPClient.connect = spy_connect + try: + write_registry(dirs, seeded_record(dirs, "vm1")) + async with Client(mcp) as client: + await client.call_tool("guest_ping", {"name": "vm1"}) + finally: + FakeQMPClient.connect = original_connect + assert seen == {"negotiate": False, "await_greeting": False} diff --git a/tests/test_images.py b/tests/test_images.py new file mode 100644 index 0000000..1d517b9 --- /dev/null +++ b/tests/test_images.py @@ -0,0 +1,87 @@ +"""Image tools exercised against the real qemu-img (installed, fast, no VM).""" + +import pytest +from fastmcp.exceptions import ToolError + +from conftest import result_data + + +async def test_create_and_info(client, tmp_path): + img = tmp_path / "disk.qcow2" + result = await client.call_tool("image_create", {"path": str(img), "size": "1G"}) + data = result_data(result) + assert data["format"] == "qcow2" + assert data["virtual_size_bytes"] == 1024**3 + assert img.exists() + + info = result_data(await client.call_tool("image_info", {"path": str(img)})) + assert info["format"] == "qcow2" + + +async def test_create_refuses_overwrite(client, tmp_path): + img = tmp_path / "disk.qcow2" + await client.call_tool("image_create", {"path": str(img), "size": "1G"}) + with pytest.raises(ToolError, match="overwrite=True"): + await client.call_tool("image_create", {"path": str(img), "size": "2G"}) + # ... and succeeds when explicitly allowed + data = result_data( + await client.call_tool("image_create", {"path": str(img), "size": "2G", "overwrite": True}) + ) + assert data["virtual_size_bytes"] == 2 * 1024**3 + + +async def test_backing_file_overlay(client, tmp_path): + base = tmp_path / "base.qcow2" + overlay = tmp_path / "overlay.qcow2" + await client.call_tool("image_create", {"path": str(base), "size": "1G"}) + data = result_data( + await client.call_tool( + "image_create", + {"path": str(overlay), "size": "1G", "backing_file": str(base)}, + ) + ) + assert data["backing_file"] == str(base) + info = result_data(await client.call_tool("image_info", {"path": str(overlay)})) + assert info["backing-filename"] == str(base) + + +async def test_convert_to_raw(client, tmp_path): + src = tmp_path / "src.qcow2" + dst = tmp_path / "dst.raw" + await client.call_tool("image_create", {"path": str(src), "size": "64M"}) + data = result_data( + await client.call_tool( + "image_convert", {"source": str(src), "dest": str(dst), "format": "raw"} + ) + ) + assert data["format"] == "raw" + + +async def test_resize_grow_and_shrink_guard(client, tmp_path): + img = tmp_path / "disk.qcow2" + await client.call_tool("image_create", {"path": str(img), "size": "1G"}) + data = result_data(await client.call_tool("image_resize", {"path": str(img), "size": "2G"})) + assert data["virtual_size_bytes"] == 2 * 1024**3 + with pytest.raises(ToolError): + await client.call_tool("image_resize", {"path": str(img), "size": "1G"}) + data = result_data( + await client.call_tool("image_resize", {"path": str(img), "size": "1G", "shrink": True}) + ) + assert data["virtual_size_bytes"] == 1024**3 + + +async def test_snapshot_lifecycle(client, tmp_path): + img = tmp_path / "disk.qcow2" + await client.call_tool("image_create", {"path": str(img), "size": "1G"}) + await client.call_tool("image_snapshot_create", {"path": str(img), "tag": "clean"}) + snaps = result_data(await client.call_tool("image_snapshot_list", {"path": str(img)})) + assert [s["tag"] for s in snaps] == ["clean"] + await client.call_tool("image_snapshot_apply", {"path": str(img), "tag": "clean"}) + await client.call_tool("image_snapshot_delete", {"path": str(img), "tag": "clean"}) + snaps = result_data(await client.call_tool("image_snapshot_list", {"path": str(img)})) + assert snaps == [] + + +async def test_info_missing_file(client, tmp_path): + with pytest.raises(ToolError, match="not found"): + await client.call_tool("image_info", {"path": str(tmp_path / "nope.qcow2")}) diff --git a/tests/test_integration.py b/tests/test_integration.py new file mode 100644 index 0000000..a4d6c40 --- /dev/null +++ b/tests/test_integration.py @@ -0,0 +1,57 @@ +"""Integration: boots a real (diskless) QEMU VM through the MCP tools. + +Run with: uv run pytest -m integration +""" + +import shutil + +import pytest +from fastmcp import Client + +from conftest import result_data +from mcqemu.registry import pid_alive +from mcqemu.server import mcp + +pytestmark = [ + pytest.mark.integration, + pytest.mark.skipif( + shutil.which("qemu-system-x86_64") is None, reason="qemu-system-x86_64 not installed" + ), +] + + +async def test_real_vm_lifecycle(dirs): + async with Client(mcp) as client: + data = result_data( + await client.call_tool( + "launch_vm", + {"name": "itest", "memory_mb": 128, "cpus": 1, "no_net": True}, + ) + ) + pid = data["pid"] + try: + assert data["status"] == "running" + assert pid_alive(pid) + + vms = result_data(await client.call_tool("list_vms", {})) + assert [(v["name"], v["status"]) for v in vms] == [("itest", "running")] + + info = result_data(await client.call_tool("vm_info", {"name": "itest"})) + assert info["vcpus"] == 1 + + data = result_data(await client.call_tool("pause_vm", {"name": "itest"})) + assert data["status"] == "paused" + vms = result_data(await client.call_tool("list_vms", {})) + assert vms[0]["status"] == "paused" + + await client.call_tool("resume_vm", {"name": "itest"}) + + data = result_data(await client.call_tool("stop_vm", {"name": "itest", "force": True})) + assert data["status"] == "stopped" + assert not pid_alive(pid) + finally: + if pid_alive(pid): + import os + import signal + + os.kill(pid, signal.SIGKILL) diff --git a/tests/test_launcher.py b/tests/test_launcher.py new file mode 100644 index 0000000..f4db9d7 --- /dev/null +++ b/tests/test_launcher.py @@ -0,0 +1,128 @@ +"""build_cmdline is pure — table-test it hard.""" + +from pathlib import Path + +import pytest +from fastmcp.exceptions import ToolError + +from mcqemu.launcher import VMPaths, build_cmdline +from mcqemu.models import VMConfig + + +def paths() -> VMPaths: + return VMPaths( + qmp_socket=Path("/run/t/qmp.sock"), + qga_socket=Path("/run/t/qga.sock"), + pidfile=Path("/run/t/pid"), + qemu_log=Path("/state/t/qemu.log"), + serial_log=Path("/state/t/serial.log"), + uefi_vars=Path("/state/t/uefi-vars.fd"), + ) + + +def argstr(args: list[str]) -> str: + return " ".join(args) + + +def test_defaults_kvm(): + cfg = VMConfig(name="t") + args = build_cmdline(cfg, paths(), accel="kvm", disk_formats={}) + s = argstr(args) + assert args[0] == "qemu-system-x86_64" + assert "-machine q35,accel=kvm" in s + assert "-cpu host" in s + assert "-m 2048" in s and "-smp 2" in s + assert "-display none" in s + assert "-daemonize" in s + assert "-qmp unix:/run/t/qmp.sock,server=on,wait=off" in s + assert "org.qemu.guest_agent.0" in s + assert "-netdev user,id=net0 -device virtio-net-pci,netdev=net0" in s + + +def test_tcg_uses_cpu_max(): + cfg = VMConfig(name="t", arch="riscv64") + args = build_cmdline(cfg, paths(), accel="tcg", disk_formats={}) + s = argstr(args) + assert args[0] == "qemu-system-riscv64" + assert "-machine virt,accel=tcg" in s + assert "-cpu max" in s + + +def test_unknown_arch_machine_falls_back_to_accel_flag(): + cfg = VMConfig(name="t", arch="m68k") + s = argstr(build_cmdline(cfg, paths(), accel="tcg", disk_formats={})) + assert "-accel tcg" in s + assert "-machine" not in s + + +def test_disks_get_probed_format(): + cfg = VMConfig(name="t", disks=["/vms/a.qcow2", "/vms/b.raw"]) + s = argstr( + build_cmdline( + cfg, + paths(), + accel="kvm", + disk_formats={"/vms/a.qcow2": "qcow2", "/vms/b.raw": "raw"}, + ) + ) + assert "-drive file=/vms/a.qcow2,if=virtio,format=qcow2" in s + assert "-drive file=/vms/b.raw,if=virtio,format=raw" in s + + +def test_iso_with_disk_boots_once_from_cdrom(): + cfg = VMConfig(name="t", disks=["/vms/a.qcow2"], iso="/isos/x.iso") + s = argstr(build_cmdline(cfg, paths(), accel="kvm", disk_formats={"/vms/a.qcow2": "qcow2"})) + assert "media=cdrom,readonly=on" in s + assert "-boot once=d" in s + + +def test_iso_alone_no_boot_flag(): + cfg = VMConfig(name="t", iso="/isos/x.iso") + s = argstr(build_cmdline(cfg, paths(), accel="kvm", disk_formats={})) + assert "-boot" not in s + + +def test_port_forwards(): + cfg = VMConfig(name="t", port_forwards=["2222:22", "8080:80"]) + s = argstr(build_cmdline(cfg, paths(), accel="kvm", disk_formats={})) + assert "hostfwd=tcp::2222-:22" in s + assert "hostfwd=tcp::8080-:80" in s + + +def test_bad_port_forward_raises(): + cfg = VMConfig(name="t", port_forwards=["22->2222"]) + with pytest.raises(ToolError, match="port forward"): + build_cmdline(cfg, paths(), accel="kvm", disk_formats={}) + + +def test_no_net(): + cfg = VMConfig(name="t", no_net=True) + s = argstr(build_cmdline(cfg, paths(), accel="kvm", disk_formats={})) + assert "-nic none" in s + assert "-netdev" not in s + + +def test_uefi_pflash_x86(): + cfg = VMConfig(name="t", firmware="uefi") + s = argstr(build_cmdline(cfg, paths(), accel="kvm", disk_formats={})) + assert "if=pflash,format=raw,readonly=on,file=/usr/share/edk2/x64/OVMF_CODE.4m.fd" in s + assert "if=pflash,format=raw,file=/state/t/uefi-vars.fd" in s + + +def test_uefi_bios_mode_aarch64(): + cfg = VMConfig(name="t", arch="aarch64", firmware="uefi") + s = argstr(build_cmdline(cfg, paths(), accel="tcg", disk_formats={})) + assert "-bios /usr/share/edk2/aarch64/QEMU_EFI.fd" in s + assert "pflash" not in s + + +def test_uefi_unsupported_arch_raises(): + cfg = VMConfig(name="t", arch="m68k", firmware="uefi") + with pytest.raises(ToolError, match="UEFI"): + build_cmdline(cfg, paths(), accel="tcg", disk_formats={}) + + +def test_extra_args_appended_last(): + cfg = VMConfig(name="t", extra_args=["-vga", "std"]) + args = build_cmdline(cfg, paths(), accel="kvm", disk_formats={}) + assert args[-2:] == ["-vga", "std"] diff --git a/tests/test_lifecycle.py b/tests/test_lifecycle.py new file mode 100644 index 0000000..90d470a --- /dev/null +++ b/tests/test_lifecycle.py @@ -0,0 +1,152 @@ +"""Lifecycle tools with mocked spawn / fake QMP.""" + +import pytest +from fastmcp import Client +from fastmcp.exceptions import ToolError + +from conftest import FakeQMPClient, result_data, write_registry +from mcqemu.models import VMRecord +from mcqemu.server import mcp + + +def seeded_record(dirs, name="vm1", **kw) -> VMRecord: + defaults = dict( + source="spawned", + qmp_socket=str(dirs.run / name / "qmp.sock"), + qga_socket=str(dirs.run / name / "qga.sock"), + pid=4242, + arch="x86_64", + config={"disks": []}, + ) + defaults.update(kw) + return VMRecord(name=name, **defaults) + + +async def test_launch_rejects_bad_name(client): + with pytest.raises(ToolError, match="Invalid VM name"): + await client.call_tool("launch_vm", {"name": "bad name!"}) + + +async def test_launch_rejects_missing_disk(client, tmp_path): + with pytest.raises(ToolError, match="image not found"): + await client.call_tool( + "launch_vm", {"name": "vm1", "disks": [str(tmp_path / "nope.qcow2")]} + ) + + +async def test_launch_rejects_disk_of_running_vm(dirs, all_pids_alive, tmp_path, monkeypatch): + disk = tmp_path / "busy.qcow2" + disk.touch() + write_registry(dirs, seeded_record(dirs, "other", config={"disks": [str(disk)]})) + async with Client(mcp) as client: + with pytest.raises(ToolError, match="already attached to running VM 'other'"): + await client.call_tool("launch_vm", {"name": "vm2", "disks": [str(disk)]}) + + +async def test_launch_rejects_duplicate_running_name(dirs, all_pids_alive): + write_registry(dirs, seeded_record(dirs, "vm1")) + async with Client(mcp) as client: + with pytest.raises(ToolError, match="already running"): + await client.call_tool("launch_vm", {"name": "vm1"}) + + +async def test_launch_happy_path_registers(dirs, monkeypatch): + async def fake_spawn(cfg, config): + return seeded_record(dirs, cfg.name, accel="kvm", config=cfg.__dict__.copy()) + + monkeypatch.setattr("mcqemu.tools.lifecycle.spawn_vm", fake_spawn) + async with Client(mcp) as client: + data = result_data(await client.call_tool("launch_vm", {"name": "fresh", "no_net": True})) + assert data["status"] == "running" + assert data["accel"] == "kvm" + vms = result_data(await client.call_tool("list_vms", {})) + assert [v["name"] for v in vms] == ["fresh"] + + +async def test_stop_force_sends_quit(dirs, fake_qmp, all_pids_dead, monkeypatch): + monkeypatch.setattr("mcqemu.tools.lifecycle.Path.exists", lambda self: True) + write_registry(dirs, seeded_record(dirs, "vm1")) + async with Client(mcp) as client: + data = result_data(await client.call_tool("stop_vm", {"name": "vm1", "force": True})) + assert data["status"] == "stopped" + assert ("quit", {}) in FakeQMPClient.calls + + +async def test_stop_graceful_waits_for_shutdown_event(dirs, fake_qmp, all_pids_dead, monkeypatch): + # is_process_alive is False (all_pids_dead), so make the socket "exist" + # to get past the already-stopped short-circuit. + monkeypatch.setattr("mcqemu.tools.lifecycle.Path.exists", lambda self: True) + FakeQMPClient.events_to_emit = [{"event": "SHUTDOWN"}] + write_registry(dirs, seeded_record(dirs, "vm1")) + async with Client(mcp) as client: + data = result_data(await client.call_tool("stop_vm", {"name": "vm1", "timeout": 2})) + assert data["method"] == "graceful ACPI shutdown" + assert ("system_powerdown", {}) in FakeQMPClient.calls + + +async def test_stop_graceful_timeout_advises_force(dirs, fake_qmp, all_pids_dead, monkeypatch): + monkeypatch.setattr("mcqemu.tools.lifecycle.Path.exists", lambda self: True) + FakeQMPClient.events_to_emit = [] # guest ignores ACPI + write_registry(dirs, seeded_record(dirs, "vm1")) + async with Client(mcp) as client: + with pytest.raises(ToolError, match="force=True"): + await client.call_tool("stop_vm", {"name": "vm1", "timeout": 1}) + + +async def test_stop_already_stopped(dirs, all_pids_dead): + write_registry(dirs, seeded_record(dirs, "vm1")) + async with Client(mcp) as client: + data = result_data(await client.call_tool("stop_vm", {"name": "vm1"})) + assert data["status"] == "stopped" + assert "already" in data["note"] + + +async def test_pause_resume(dirs, fake_qmp, all_pids_alive): + write_registry(dirs, seeded_record(dirs, "vm1")) + async with Client(mcp) as client: + assert ( + result_data(await client.call_tool("pause_vm", {"name": "vm1"}))["status"] == "paused" + ) + assert ( + result_data(await client.call_tool("resume_vm", {"name": "vm1"}))["status"] == "running" + ) + assert ("stop", {}) in FakeQMPClient.calls + assert ("cont", {}) in FakeQMPClient.calls + + +async def test_unknown_vm_lists_known(dirs, fake_qmp): + write_registry(dirs, seeded_record(dirs, "alpha")) + async with Client(mcp) as client: + with pytest.raises(ToolError, match="alpha"): + await client.call_tool("pause_vm", {"name": "nope"}) + + +async def test_attach_requires_existing_socket(client, tmp_path): + with pytest.raises(ToolError, match="No socket"): + await client.call_tool( + "attach_vm", {"name": "ext", "qmp_socket": str(tmp_path / "no.sock")} + ) + + +async def test_attach_and_forget(dirs, fake_qmp, tmp_path): + sock = tmp_path / "ext.sock" + sock.touch() + FakeQMPClient.responses["query-status"] = {"status": "running"} + async with Client(mcp) as client: + data = result_data( + await client.call_tool("attach_vm", {"name": "ext", "qmp_socket": str(sock)}) + ) + assert data["status"] == "running" + data = result_data(await client.call_tool("forget_vm", {"name": "ext"})) + assert data["forgotten"] is True + vms = result_data(await client.call_tool("list_vms", {})) + assert vms == [] + + +async def test_forget_refuses_running_spawned_vm(dirs, all_pids_alive): + write_registry(dirs, seeded_record(dirs, "vm1")) + async with Client(mcp) as client: + with pytest.raises(ToolError, match="still running"): + await client.call_tool("forget_vm", {"name": "vm1"}) + data = result_data(await client.call_tool("forget_vm", {"name": "vm1", "force": True})) + assert data["forgotten"] is True diff --git a/tests/test_query.py b/tests/test_query.py new file mode 100644 index 0000000..845891c --- /dev/null +++ b/tests/test_query.py @@ -0,0 +1,58 @@ +"""list_vms / vm_info status reporting.""" + +from fastmcp import Client + +from conftest import FakeQMPClient, result_data, write_registry +from mcqemu.server import mcp +from test_lifecycle import seeded_record + + +async def test_list_vms_stopped(dirs, all_pids_dead): + write_registry(dirs, seeded_record(dirs, "vm1")) + async with Client(mcp) as client: + vms = result_data(await client.call_tool("list_vms", {})) + assert vms[0]["status"] == "stopped" + + +async def test_list_vms_running(dirs, fake_qmp, all_pids_alive): + FakeQMPClient.responses["query-status"] = {"status": "running"} + write_registry(dirs, seeded_record(dirs, "vm1")) + async with Client(mcp) as client: + vms = result_data(await client.call_tool("list_vms", {})) + assert vms[0]["status"] == "running" + + +async def test_list_vms_unreachable_when_qmp_fails(dirs, fake_qmp, all_pids_alive): + FakeQMPClient.connect_error = OSError("connection refused") + write_registry(dirs, seeded_record(dirs, "vm1")) + async with Client(mcp) as client: + vms = result_data(await client.call_tool("list_vms", {})) + assert vms[0]["status"] == "unreachable" + + +async def test_vm_info_running_includes_block_devices(dirs, fake_qmp, all_pids_alive): + FakeQMPClient.responses["query-status"] = {"status": "running"} + FakeQMPClient.responses["query-cpus-fast"] = [{"cpu-index": 0}, {"cpu-index": 1}] + FakeQMPClient.responses["query-block"] = [ + { + "device": "virtio0", + "inserted": {"file": "/vms/a.qcow2", "drv": "qcow2", "ro": False}, + }, + {"device": "empty-cd"}, + ] + write_registry(dirs, seeded_record(dirs, "vm1")) + async with Client(mcp) as client: + info = result_data(await client.call_tool("vm_info", {"name": "vm1"})) + assert info["status"] == "running" + assert info["vcpus"] == 2 + assert info["block_devices"] == [ + {"device": "virtio0", "file": "/vms/a.qcow2", "format": "qcow2", "read_only": False} + ] + + +async def test_vm_info_stopped_skips_qmp(dirs, all_pids_dead): + write_registry(dirs, seeded_record(dirs, "vm1")) + async with Client(mcp) as client: + info = result_data(await client.call_tool("vm_info", {"name": "vm1"})) + assert info["status"] == "stopped" + assert "vcpus" not in info diff --git a/tests/test_registry.py b/tests/test_registry.py new file mode 100644 index 0000000..f7d4f7a --- /dev/null +++ b/tests/test_registry.py @@ -0,0 +1,69 @@ +"""Registry persistence and liveness logic.""" + +from mcqemu.config import Config +from mcqemu.models import VMRecord +from mcqemu.registry import VMRegistry, pid_alive + + +def make_config(dirs) -> Config: + return Config(state_dir=dirs.state, runtime_dir=dirs.run) + + +def record(name="vm1", **kw) -> VMRecord: + defaults = dict(source="spawned", qmp_socket=f"/run/{name}/qmp.sock", pid=12345) + defaults.update(kw) + return VMRecord(name=name, **defaults) + + +def test_round_trip(dirs): + reg = VMRegistry(make_config(dirs)) + reg.load() + reg.add(record("alpha", arch="x86_64", config={"disks": ["/vms/a.qcow2"]})) + reg.add(record("beta", source="attached", pid=None)) + + reg2 = VMRegistry(make_config(dirs)) + reg2.load() + assert sorted(reg2.names()) == ["alpha", "beta"] + alpha = reg2.get("alpha") + assert alpha.arch == "x86_64" + assert alpha.config["disks"] == ["/vms/a.qcow2"] + assert reg2.get("beta").source == "attached" + + +def test_load_missing_file_is_empty(dirs): + reg = VMRegistry(make_config(dirs)) + reg.load() + assert reg.all() == [] + + +def test_from_dict_ignores_unknown_fields(): + rec = VMRecord.from_dict( + {"name": "x", "source": "spawned", "qmp_socket": "/s", "future_field": 1} + ) + assert rec.name == "x" + + +def test_pid_alive_rejects_non_qemu_process(): + import os + + # Our own PID exists but comm is python/pytest, not qemu-system — must be False. + assert pid_alive(os.getpid()) is False + + +def test_pid_alive_rejects_dead_and_bogus_pids(): + assert pid_alive(2**22 + 12345) is False + assert pid_alive(None) is False + assert pid_alive(0) is False + + +def test_disks_in_use_only_counts_live_vms(dirs, monkeypatch): + reg = VMRegistry(make_config(dirs)) + reg.load() + reg.add(record("live", pid=111, config={"disks": ["/vms/live.qcow2"]})) + reg.add(record("dead", pid=222, config={"disks": ["/vms/dead.qcow2"]})) + monkeypatch.setattr("mcqemu.registry.pid_alive", lambda pid: pid == 111) + # refresh_pid would overwrite from pidfile; none set, so pids stay put. + used = reg.disks_in_use() + assert "/vms/live.qcow2" in used + assert used["/vms/live.qcow2"] == "live" + assert "/vms/dead.qcow2" not in used diff --git a/tests/test_snapshots.py b/tests/test_snapshots.py new file mode 100644 index 0000000..e4e1c74 --- /dev/null +++ b/tests/test_snapshots.py @@ -0,0 +1,64 @@ +"""Live snapshot tools (HMP passthrough).""" + +import pytest +from fastmcp import Client +from fastmcp.exceptions import ToolError + +from conftest import FakeQMPClient, result_data, write_registry +from mcqemu.server import mcp +from test_lifecycle import seeded_record + + +async def test_snapshot_create_success_on_silence(dirs, fake_qmp): + FakeQMPClient.responses["human-monitor-command"] = "" + write_registry(dirs, seeded_record(dirs, "vm1")) + async with Client(mcp) as client: + data = result_data( + await client.call_tool("vm_snapshot_create", {"name": "vm1", "tag": "clean"}) + ) + assert data["created"] is True + assert ("human-monitor-command", {"command-line": "savevm clean"}) in FakeQMPClient.calls + + +async def test_snapshot_create_surfaces_hmp_error(dirs, fake_qmp): + FakeQMPClient.responses["human-monitor-command"] = ( + "Error: Device 'virtio0' is writable but does not support snapshots" + ) + write_registry(dirs, seeded_record(dirs, "vm1")) + async with Client(mcp) as client: + with pytest.raises(ToolError, match="does not support snapshots"): + await client.call_tool("vm_snapshot_create", {"name": "vm1", "tag": "clean"}) + + +async def test_snapshot_restore_and_delete(dirs, fake_qmp): + FakeQMPClient.responses["human-monitor-command"] = "" + write_registry(dirs, seeded_record(dirs, "vm1")) + async with Client(mcp) as client: + assert result_data( + await client.call_tool("vm_snapshot_restore", {"name": "vm1", "tag": "clean"}) + )["restored"] + assert result_data( + await client.call_tool("vm_snapshot_delete", {"name": "vm1", "tag": "clean"}) + )["deleted"] + cmdlines = [a["command-line"] for _, a in FakeQMPClient.calls] + assert "loadvm clean" in cmdlines + assert "delvm clean" in cmdlines + + +async def test_snapshot_bad_tag_rejected(dirs, fake_qmp): + write_registry(dirs, seeded_record(dirs, "vm1")) + async with Client(mcp) as client: + for tag in ["has space", "-leading-dash", ""]: + with pytest.raises(ToolError): + await client.call_tool("vm_snapshot_create", {"name": "vm1", "tag": tag}) + + +async def test_snapshot_list_raw(dirs, fake_qmp): + FakeQMPClient.responses["human-monitor-command"] = ( + "List of snapshots present on all disks:\n" + "ID TAG VM SIZE DATE\n-- clean 250M 2026-08-16" + ) + write_registry(dirs, seeded_record(dirs, "vm1")) + async with Client(mcp) as client: + data = result_data(await client.call_tool("vm_snapshot_list", {"name": "vm1"})) + assert "clean" in data["snapshots_raw"]