88 lines
3.5 KiB
Python
88 lines
3.5 KiB
Python
"""Image tools exercised against the real qemu-img (installed, fast, no VM)."""
|
|
|
|
import pytest
|
|
from fastmcp.exceptions import ToolError
|
|
|
|
from conftest import result_data
|
|
|
|
|
|
async def test_create_and_info(client, tmp_path):
|
|
img = tmp_path / "disk.qcow2"
|
|
result = await client.call_tool("image_create", {"path": str(img), "size": "1G"})
|
|
data = result_data(result)
|
|
assert data["format"] == "qcow2"
|
|
assert data["virtual_size_bytes"] == 1024**3
|
|
assert img.exists()
|
|
|
|
info = result_data(await client.call_tool("image_info", {"path": str(img)}))
|
|
assert info["format"] == "qcow2"
|
|
|
|
|
|
async def test_create_refuses_overwrite(client, tmp_path):
|
|
img = tmp_path / "disk.qcow2"
|
|
await client.call_tool("image_create", {"path": str(img), "size": "1G"})
|
|
with pytest.raises(ToolError, match="overwrite=True"):
|
|
await client.call_tool("image_create", {"path": str(img), "size": "2G"})
|
|
# ... and succeeds when explicitly allowed
|
|
data = result_data(
|
|
await client.call_tool("image_create", {"path": str(img), "size": "2G", "overwrite": True})
|
|
)
|
|
assert data["virtual_size_bytes"] == 2 * 1024**3
|
|
|
|
|
|
async def test_backing_file_overlay(client, tmp_path):
|
|
base = tmp_path / "base.qcow2"
|
|
overlay = tmp_path / "overlay.qcow2"
|
|
await client.call_tool("image_create", {"path": str(base), "size": "1G"})
|
|
data = result_data(
|
|
await client.call_tool(
|
|
"image_create",
|
|
{"path": str(overlay), "size": "1G", "backing_file": str(base)},
|
|
)
|
|
)
|
|
assert data["backing_file"] == str(base)
|
|
info = result_data(await client.call_tool("image_info", {"path": str(overlay)}))
|
|
assert info["backing-filename"] == str(base)
|
|
|
|
|
|
async def test_convert_to_raw(client, tmp_path):
|
|
src = tmp_path / "src.qcow2"
|
|
dst = tmp_path / "dst.raw"
|
|
await client.call_tool("image_create", {"path": str(src), "size": "64M"})
|
|
data = result_data(
|
|
await client.call_tool(
|
|
"image_convert", {"source": str(src), "dest": str(dst), "format": "raw"}
|
|
)
|
|
)
|
|
assert data["format"] == "raw"
|
|
|
|
|
|
async def test_resize_grow_and_shrink_guard(client, tmp_path):
|
|
img = tmp_path / "disk.qcow2"
|
|
await client.call_tool("image_create", {"path": str(img), "size": "1G"})
|
|
data = result_data(await client.call_tool("image_resize", {"path": str(img), "size": "2G"}))
|
|
assert data["virtual_size_bytes"] == 2 * 1024**3
|
|
with pytest.raises(ToolError):
|
|
await client.call_tool("image_resize", {"path": str(img), "size": "1G"})
|
|
data = result_data(
|
|
await client.call_tool("image_resize", {"path": str(img), "size": "1G", "shrink": True})
|
|
)
|
|
assert data["virtual_size_bytes"] == 1024**3
|
|
|
|
|
|
async def test_snapshot_lifecycle(client, tmp_path):
|
|
img = tmp_path / "disk.qcow2"
|
|
await client.call_tool("image_create", {"path": str(img), "size": "1G"})
|
|
await client.call_tool("image_snapshot_create", {"path": str(img), "tag": "clean"})
|
|
snaps = result_data(await client.call_tool("image_snapshot_list", {"path": str(img)}))
|
|
assert [s["tag"] for s in snaps] == ["clean"]
|
|
await client.call_tool("image_snapshot_apply", {"path": str(img), "tag": "clean"})
|
|
await client.call_tool("image_snapshot_delete", {"path": str(img), "tag": "clean"})
|
|
snaps = result_data(await client.call_tool("image_snapshot_list", {"path": str(img)}))
|
|
assert snaps == []
|
|
|
|
|
|
async def test_info_missing_file(client, tmp_path):
|
|
with pytest.raises(ToolError, match="not found"):
|
|
await client.call_tool("image_info", {"path": str(tmp_path / "nope.qcow2")})
|