65 lines
2.6 KiB
Python
65 lines
2.6 KiB
Python
"""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"]
|