Ryan Malloy
8c619a1674
The unit tests mock QMP and the guest agent, so they cannot catch a wrong argument name, a QEMU option that stopped parsing, or a reply shape that differs from the fake. This suite drives the real thing: the qemu-img toolchain, lifecycle and see-and-drive on a diskless BIOS VM, adopting a forgotten VM through attach_vm, the refusal to reuse a live socket, and the full sandbox journey (guest exec, file round trip, live snapshot create / restore / delete verified by guest state, screenshot) with a check that the base image is never modified. It immediately earned its keep: pid_matches_vm was rejecting attached VMs, because an externally launched QEMU carries whatever -name its launcher chose, so adopting one under a different name made it read as stopped. The identity check now applies only to VMs we spawned.
293 lines
12 KiB
Python
293 lines
12 KiB
Python
"""End-to-end acceptance: every tool group against a real QEMU.
|
|
|
|
Unit tests mock QMP and the guest agent, so they cannot catch a wrong QMP
|
|
argument name, a QEMU option that no longer parses, or a guest agent reply
|
|
shape that differs from the fake. This suite drives the real thing.
|
|
|
|
Run with: uv run pytest -m integration -q
|
|
|
|
The guest-agent and display journeys need a base image with qemu-guest-agent
|
|
installed; set MCQEMU_TEST_BASE_IMAGE or drop one at ~/vms/ubuntu-agent.qcow2.
|
|
"""
|
|
|
|
import base64
|
|
import os
|
|
import shutil
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from fastmcp import Client
|
|
|
|
from conftest import result_data
|
|
from mcqemu.server import mcp
|
|
|
|
pytestmark = pytest.mark.integration
|
|
|
|
BASE_IMAGE = Path(
|
|
os.environ.get("MCQEMU_TEST_BASE_IMAGE", str(Path.home() / "vms" / "ubuntu-agent.qcow2"))
|
|
)
|
|
needs_qemu = pytest.mark.skipif(
|
|
shutil.which("qemu-system-x86_64") is None, reason="qemu-system-x86_64 not installed"
|
|
)
|
|
needs_agent_image = pytest.mark.skipif(
|
|
not BASE_IMAGE.exists(),
|
|
reason=f"no guest-agent base image at {BASE_IMAGE} (set MCQEMU_TEST_BASE_IMAGE)",
|
|
)
|
|
|
|
|
|
async def guest_sh(client, name: str, script: str, timeout: int = 60) -> str:
|
|
result = await client.call_tool(
|
|
"guest_exec",
|
|
{"name": name, "command": "/bin/sh", "args": ["-c", script], "timeout": timeout},
|
|
)
|
|
assert result.data["exitcode"] == 0, result.data
|
|
return result.data["stdout"]
|
|
|
|
|
|
@needs_qemu
|
|
async def test_image_toolchain(dirs, tmp_path):
|
|
"""All eight qemu-img tools, including the overlay chain sandboxes rely on."""
|
|
base = tmp_path / "base.qcow2"
|
|
overlay = tmp_path / "overlay.qcow2"
|
|
raw = tmp_path / "flat.raw"
|
|
|
|
async with Client(mcp) as client:
|
|
created = result_data(
|
|
await client.call_tool("image_create", {"path": str(base), "size": "64M"})
|
|
)
|
|
assert created["virtual_size_bytes"] == 64 * 1024**2
|
|
|
|
result_data(
|
|
await client.call_tool(
|
|
"image_create",
|
|
{"path": str(overlay), "size": "64M", "backing_file": str(base)},
|
|
)
|
|
)
|
|
chain = result_data(
|
|
await client.call_tool("image_info", {"path": str(overlay), "backing_chain": True})
|
|
)
|
|
assert len(chain) == 2, "overlay must report its backing file"
|
|
|
|
result_data(await client.call_tool("image_resize", {"path": str(base), "size": "128M"}))
|
|
assert result_data(await client.call_tool("image_info", {"path": str(base)}))[
|
|
"virtual-size"
|
|
] == 128 * 1024**2
|
|
|
|
converted = result_data(
|
|
await client.call_tool(
|
|
"image_convert", {"source": str(base), "dest": str(raw), "format": "raw"}
|
|
)
|
|
)
|
|
assert converted["format"] == "raw"
|
|
|
|
await client.call_tool("image_snapshot_create", {"path": str(base), "tag": "clean"})
|
|
snaps = result_data(await client.call_tool("image_snapshot_list", {"path": str(base)}))
|
|
assert [s["tag"] for s in snaps] == ["clean"]
|
|
await client.call_tool("image_snapshot_apply", {"path": str(base), "tag": "clean"})
|
|
await client.call_tool("image_snapshot_delete", {"path": str(base), "tag": "clean"})
|
|
assert result_data(await client.call_tool("image_snapshot_list", {"path": str(base)})) == []
|
|
|
|
|
|
@needs_qemu
|
|
async def test_lifecycle_and_display_on_a_diskless_vm(dirs):
|
|
"""Lifecycle plus see-and-drive against SeaBIOS — no guest OS required."""
|
|
async with Client(mcp) as client:
|
|
launched = result_data(
|
|
await client.call_tool(
|
|
"launch_vm",
|
|
{"name": "acc-bios", "memory_mb": 128, "cpus": 1, "no_net": True},
|
|
)
|
|
)
|
|
assert launched["status"] == "running"
|
|
try:
|
|
listing = result_data(await client.call_tool("list_vms", {}))
|
|
assert listing["registry_warnings"] == []
|
|
assert [v["name"] for v in listing["vms"]] == ["acc-bios"]
|
|
|
|
info = result_data(await client.call_tool("vm_info", {"name": "acc-bios"}))
|
|
assert info["vcpus"] == 1 and info["status"] == "running"
|
|
|
|
shot = await client.call_tool("vm_screenshot", {"name": "acc-bios"})
|
|
assert base64.b64decode(shot.content[0].data)[:8] == b"\x89PNG\r\n\x1a\n"
|
|
|
|
await client.call_tool("vm_send_keys", {"name": "acc-bios", "keys": ["f2", "esc"]})
|
|
await client.call_tool("vm_type_text", {"name": "acc-bios", "text": "hello"})
|
|
await client.call_tool("vm_click", {"name": "acc-bios", "x": 5, "y": 5})
|
|
moved = result_data(
|
|
await client.call_tool(
|
|
"vm_mouse_move",
|
|
{"name": "acc-bios", "home": "bottom-right", "dx": -20, "dy": -20},
|
|
)
|
|
)
|
|
assert moved["homed"] == "bottom-right"
|
|
serial = result_data(await client.call_tool("vm_serial_read", {"name": "acc-bios"}))
|
|
assert "log_bytes" in serial
|
|
|
|
assert (
|
|
result_data(await client.call_tool("pause_vm", {"name": "acc-bios"}))["status"]
|
|
== "paused"
|
|
)
|
|
assert (
|
|
result_data(await client.call_tool("resume_vm", {"name": "acc-bios"}))["status"]
|
|
== "running"
|
|
)
|
|
finally:
|
|
result_data(await client.call_tool("stop_vm", {"name": "acc-bios", "force": True}))
|
|
assert result_data(await client.call_tool("list_vms", {}))["vms"][0]["status"] == "stopped"
|
|
await client.call_tool("forget_vm", {"name": "acc-bios"})
|
|
|
|
|
|
@needs_qemu
|
|
async def test_attach_to_an_externally_managed_vm(dirs):
|
|
"""A VM we forget (leaving it running) can be re-adopted through attach_vm."""
|
|
async with Client(mcp) as client:
|
|
launched = result_data(
|
|
await client.call_tool(
|
|
"launch_vm",
|
|
{"name": "acc-attach", "memory_mb": 128, "cpus": 1, "no_net": True},
|
|
)
|
|
)
|
|
socket_path, pid = launched["qmp_socket"], launched["pid"]
|
|
# Deliberately orphan it, then adopt it under a new name.
|
|
result_data(await client.call_tool("forget_vm", {"name": "acc-attach", "force": True}))
|
|
assert result_data(await client.call_tool("list_vms", {}))["vms"] == []
|
|
|
|
adopted = result_data(
|
|
await client.call_tool(
|
|
"attach_vm",
|
|
{"name": "acc-adopted", "qmp_socket": socket_path, "pid": pid},
|
|
)
|
|
)
|
|
assert adopted["status"] == "running"
|
|
try:
|
|
info = result_data(await client.call_tool("vm_info", {"name": "acc-adopted"}))
|
|
assert info["source"] == "attached" and info["status"] == "running"
|
|
finally:
|
|
result_data(await client.call_tool("stop_vm", {"name": "acc-adopted", "force": True}))
|
|
await client.call_tool("forget_vm", {"name": "acc-adopted"})
|
|
|
|
|
|
@needs_qemu
|
|
async def test_launch_refuses_to_trample_a_running_vm(dirs):
|
|
"""The name of a running-but-forgotten VM must not be silently reused."""
|
|
async with Client(mcp) as client:
|
|
launched = result_data(
|
|
await client.call_tool(
|
|
"launch_vm", {"name": "acc-ghost", "memory_mb": 128, "cpus": 1, "no_net": True}
|
|
)
|
|
)
|
|
await client.call_tool("forget_vm", {"name": "acc-ghost", "force": True})
|
|
try:
|
|
with pytest.raises(Exception, match="LIVE QMP socket"):
|
|
await client.call_tool(
|
|
"launch_vm",
|
|
{"name": "acc-ghost", "memory_mb": 128, "cpus": 1, "no_net": True},
|
|
)
|
|
finally:
|
|
await client.call_tool(
|
|
"attach_vm",
|
|
{
|
|
"name": "acc-ghost-adopted",
|
|
"qmp_socket": launched["qmp_socket"],
|
|
"pid": launched["pid"],
|
|
},
|
|
)
|
|
await client.call_tool("stop_vm", {"name": "acc-ghost-adopted", "force": True})
|
|
await client.call_tool("forget_vm", {"name": "acc-ghost-adopted"})
|
|
|
|
|
|
@needs_qemu
|
|
@needs_agent_image
|
|
async def test_sandbox_guest_and_snapshot_journey(dirs):
|
|
"""The full disposable-sandbox workflow against a real guest OS."""
|
|
base_before = (BASE_IMAGE.stat().st_size, BASE_IMAGE.stat().st_mtime)
|
|
|
|
async with Client(mcp) as client:
|
|
sandbox = result_data(
|
|
await client.call_tool(
|
|
"sandbox_vm", {"base_image": str(BASE_IMAGE), "memory_mb": 2048}
|
|
)
|
|
)
|
|
assert sandbox["guest_agent"] == "responding", sandbox
|
|
assert sandbox["network"] == "outbound blocked"
|
|
assert sandbox["port_forwards"][0].endswith(":22")
|
|
name = sandbox["name"]
|
|
try:
|
|
info = result_data(await client.call_tool("guest_info", {"name": name}))
|
|
assert info["os"]["id"], "guest should report its OS"
|
|
assert "guest-exec" in info["supported_commands"]
|
|
|
|
assert "acceptance" in await guest_sh(client, name, "echo acceptance")
|
|
|
|
marker = "/tmp/mcqemu-acceptance.txt"
|
|
written = result_data(
|
|
await client.call_tool(
|
|
"guest_file_write",
|
|
{"name": name, "path": marker, "content": "written by the host\n"},
|
|
)
|
|
)
|
|
assert written["bytes_written"] == len("written by the host\n")
|
|
read = result_data(
|
|
await client.call_tool("guest_file_read", {"name": name, "path": marker})
|
|
)
|
|
assert read["content"] == "written by the host\n"
|
|
# The guest must agree the file exists — proves we wrote to the guest,
|
|
# not to some host path.
|
|
assert "written by the host" in await guest_sh(client, name, f"cat {marker}")
|
|
|
|
# Live snapshot round trip, with a marker that must survive the
|
|
# restore because it predates the snapshot.
|
|
await guest_sh(client, name, "echo pre-snapshot > /tmp/marker")
|
|
result_data(
|
|
await client.call_tool("vm_snapshot_create", {"name": name, "tag": "acc-point"})
|
|
)
|
|
snaps = result_data(await client.call_tool("vm_snapshot_list", {"name": name}))
|
|
assert "acc-point" in snaps["snapshots_raw"]
|
|
|
|
await guest_sh(client, name, "echo post-snapshot >> /tmp/marker")
|
|
assert "post-snapshot" in await guest_sh(client, name, "cat /tmp/marker")
|
|
|
|
result_data(
|
|
await client.call_tool("vm_snapshot_restore", {"name": name, "tag": "acc-point"})
|
|
)
|
|
restored = await guest_sh(client, name, "cat /tmp/marker")
|
|
assert "pre-snapshot" in restored
|
|
assert "post-snapshot" not in restored, "restore must roll back guest state"
|
|
|
|
result_data(
|
|
await client.call_tool("vm_snapshot_delete", {"name": name, "tag": "acc-point"})
|
|
)
|
|
|
|
shot = await client.call_tool("vm_screenshot", {"name": name})
|
|
assert base64.b64decode(shot.content[0].data)[:8] == b"\x89PNG\r\n\x1a\n"
|
|
finally:
|
|
destroyed = result_data(await client.call_tool("sandbox_destroy", {"name": name}))
|
|
assert destroyed["destroyed"] is True
|
|
assert destroyed["overlay_deleted"] is True
|
|
assert result_data(await client.call_tool("list_vms", {}))["vms"] == []
|
|
|
|
assert (BASE_IMAGE.stat().st_size, BASE_IMAGE.stat().st_mtime) == base_before, (
|
|
"the base image must never be modified"
|
|
)
|
|
|
|
|
|
@needs_qemu
|
|
async def test_resources_and_prompt_are_reachable(dirs):
|
|
async with Client(mcp) as client:
|
|
launched = result_data(
|
|
await client.call_tool(
|
|
"launch_vm", {"name": "acc-res", "memory_mb": 128, "cpus": 1, "no_net": True}
|
|
)
|
|
)
|
|
assert launched["status"] == "running"
|
|
try:
|
|
contents = await client.read_resource("mcqemu://vms")
|
|
assert "acc-res" in contents[0].text
|
|
one = await client.read_resource("mcqemu://vm/acc-res")
|
|
assert "acc-res" in one[0].text
|
|
prompt = await client.get_prompt("provision_test_vm", {"os_hint": "Alpine"})
|
|
assert "image_create" in prompt.messages[0].content.text
|
|
finally:
|
|
await client.call_tool("stop_vm", {"name": "acc-res", "force": True})
|
|
await client.call_tool("forget_vm", {"name": "acc-res"})
|