"""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"})