Skip to content

QEMU

QEMU guest agent and virtual device management — for tests that run inside QEMU/KVM VMs and need to interact with virtual disks, hotplug, or the guest agent socket.

sts.qemu

QEMU disk image creation and deletion via qemu-img.

create_image(name, *, size='1024', fmt=None, path='/var/tmp', **options)

Create disk image.

Parameters:

Name Type Description Default
name str

Image name (without extension)

required
size str

Image size with optional suffix K/M/G/T (default: 1024 bytes)

'1024'
fmt str | None

Image format (default: raw)

None
path str | Path

Output directory (default: /var/tmp)

'/var/tmp'
**options str

Format-specific options (see QCOW_OPTIONS for qcow2)

{}
Example
create_image('test', size='1G', fmt='qcow2', compat='1.1').assert_ok()
Source code in sts_libs/src/sts/qemu.py
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
def create_image(
    name: str,
    *,
    size: str = '1024',
    fmt: str | None = None,
    path: str | Path = '/var/tmp',
    **options: str,
) -> CommandResult:
    """Create disk image.

    Args:
        name: Image name (without extension)
        size: Image size with optional suffix K/M/G/T (default: 1024 bytes)
        fmt: Image format (default: raw)
        path: Output directory (default: /var/tmp)
        **options: Format-specific options (see QCOW_OPTIONS for qcow2)

    Example:
        ```python
        create_image('test', size='1G', fmt='qcow2', compat='1.1').assert_ok()
        ```
    """
    # Ensure qemu-img is installed
    pm = Dnf()
    if not pm.install('qemu-img'):
        return CommandResult(command='dnf install -y qemu-img', rc=1, stderr='Failed to install qemu-img')

    # Build image path (always .img extension)
    img_path = Path(path) / f'{name}.img'

    # Build qemu-img command
    cmd = ['qemu-img', 'create']
    if fmt:
        cmd.extend(['-f', fmt])
        # Handle qcow2-specific options
        if fmt == 'qcow2' and options:
            opts = [f'{k}={v}' for k, v in options.items() if k in QCOW_OPTIONS]
            if opts:
                cmd.extend(['-o', ','.join(opts)])
    cmd.extend([str(img_path), size])

    # Create image using qemu-img
    result = run(' '.join(cmd))
    if result.failed:
        logger.error('Failed to create disk image')

    return result

delete_image(name, path='/var/tmp')

Delete disk image.

Parameters:

Name Type Description Default
name str

Image name (without extension)

required
path str | Path

Image directory (default: /var/tmp)

'/var/tmp'
Source code in sts_libs/src/sts/qemu.py
76
77
78
79
80
81
82
83
84
85
86
87
88
89
def delete_image(name: str, path: str | Path = '/var/tmp') -> bool:
    """Delete disk image.

    Args:
        name: Image name (without extension)
        path: Image directory (default: /var/tmp)
    """
    img_path = Path(path) / f'{name}.img'
    try:
        img_path.unlink(missing_ok=True)
    except OSError:
        logger.exception(f'Failed to delete image: {img_path}')
        return False
    return True