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