58 lines
1.7 KiB
Python
58 lines
1.7 KiB
Python
"""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)
|