from __future__ import annotations
import importlib
import os
import sys
from pathlib import Path
from typing import Any
import pytest
RUN_INTEGRATION = os.environ.get("MCP_QEMU_LAB_RUN_INTEGRATION") == "1"
INTEGRATION_TEST_TIMEOUT_SEC = int(os.environ.get("MCP_QEMU_LAB_INTEGRATION_TIMEOUT_SEC", "1800"))
pytestmark = [
pytest.mark.integration,
pytest.mark.skipif(not RUN_INTEGRATION, reason="Set MCP_QEMU_LAB_RUN_INTEGRATION=1 to run QEMU integration tests."),
]
@pytest.mark.timeout(INTEGRATION_TEST_TIMEOUT_SEC)
def test_boot_exec_attach_core_dump(tmp_path: Path) -> None:
workspace = tmp_path / "workspace"
server = _load_server_module(workspace)
vm_id: str | None = None
sleep_pid: int | None = None
session_id: str | None = None
try:
deps = server.ensure_dependencies()
assert deps["dependencies"]
vm = server.vm_create(
name="itest",
net_mode="user",
base_image_download_timeout_sec=1800,
)
vm_id = vm["vm_id"]
start = server.vm_start(vm_id=vm_id, qmp_wait_timeout_sec=240, qmp_command_timeout_sec=20)
assert start["status"] in {"running", "prelaunch", "inmigrate", "unknown"}
ready = server.guest_wait_ready(
vm_id=vm_id,
timeout_sec=900,
poll_interval_sec=5,
wait_for_cloud_init=True,
require_gdb=True,
)
assert ready["ssh_ready"] is True
assert ready["checks"]["gdb"] is True
true_result = server.guest_exec(vm_id=vm_id, allowed_command="true", timeout_sec=30)
assert true_result["exit_code"] == 0
pid_result = server.guest_exec(
vm_id=vm_id,
allowed_command="ps",
unsafe_allow_arbitrary_commands=True,
unsafe_command="nohup sleep 300 >/tmp/mcpqemu-sleep.log 2>&1 & echo $!",
timeout_sec=30,
)
assert pid_result["exit_code"] == 0
sleep_pid = int(pid_result["stdout"].strip().splitlines()[-1])
assert sleep_pid > 0
session = server.debugger_attach(
vm_id=vm_id,
pid=sleep_pid,
ready_timeout_sec=300,
pid_check_timeout_sec=30,
)
session_id = session["session_id"]
registers = server.debugger_read_registers(session_id=session_id, timeout_sec=90)
assert "registers" in registers
core = server.process_dump_core(
vm_id=vm_id,
pid=sleep_pid,
reason_label="itest",
gdb_timeout_sec=600,
copy_timeout_sec=300,
)
artifact = core["artifact"]
artifact_path = Path(artifact["path"])
assert artifact_path.exists()
assert artifact_path.stat().st_size > 0
assert len(artifact["sha256"]) == 64
finally:
if session_id:
try:
server.debugger_detach(session_id=session_id)
except Exception:
pass
if vm_id:
try:
server.vm_stop(
vm_id=vm_id,
mode="force",
graceful_timeout_sec=10,
force_timeout_sec=30,
qmp_command_timeout_sec=15,
)
except Exception:
pass
def _load_server_module(workspace: Path) -> Any:
os.environ["MCP_QEMU_LAB_WORKSPACE"] = str(workspace)
repo_root = Path(__file__).resolve().parents[1]
qemu_dir = repo_root / "tools" / "qemu"
if qemu_dir.exists():
os.environ["PATH"] = f"{qemu_dir}{os.pathsep}{os.environ.get('PATH', '')}"
if "mcp_qemu_lab.server" in sys.modules:
return importlib.reload(sys.modules["mcp_qemu_lab.server"])
import mcp_qemu_lab.server as server_module
return server_module
