Loop Devices

Loop devices (/dev/loop*) expose regular files as block devices — commonly used in tests as backing storage for LVM, DM, and filesystem operations without requiring real disks.

sts.loop

Loop device management -- create, discover, attach/detach backing files.

LoopDevice pydantic-model

Bases: StorageDevice

Loop device backed by a regular file.

Construction performs no I/O. Call discover() to populate attributes from losetup/blockdev, or use the create() / using() class methods. Supports context-manager cleanup.

Example

device = LoopDevice(name='loop0').discover()
device = LoopDevice.create(size_mb=1024)
with LoopDevice.using(size_mb=512) as dev:
    ...

Show JSON schema:

{
  "$defs": {
    "LoopDeviceInfo": {
      "description": "Parsed ``losetup -J`` output for a single loop device.\n\nAttributes:\n    sizelimit: Size limit in bytes (0 = unlimited).\n    offset: Offset in bytes (0 = start of file).\n    autoclear: Whether device is automatically detached on last close.\n    dio: Whether direct I/O is enabled (bypass page cache).",
      "properties": {
        "name": {
          "title": "Name",
          "type": "string"
        },
        "back-file": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Back-File"
        },
        "sizelimit": {
          "anyOf": [
            {
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Sizelimit"
        },
        "offset": {
          "anyOf": [
            {
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Offset"
        },
        "autoclear": {
          "default": false,
          "title": "Autoclear",
          "type": "boolean"
        },
        "ro": {
          "default": false,
          "title": "Ro",
          "type": "boolean"
        },
        "dio": {
          "default": false,
          "title": "Dio",
          "type": "boolean"
        },
        "log-sec": {
          "anyOf": [
            {
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Log-Sec"
        }
      },
      "required": [
        "name"
      ],
      "title": "LoopDeviceInfo",
      "type": "object"
    }
  },
  "additionalProperties": false,
  "description": "Loop device backed by a regular file.\n\nConstruction performs no I/O. Call ``discover()`` to populate\nattributes from losetup/blockdev, or use the ``create()`` /\n``using()`` class methods. Supports context-manager cleanup.\n\nExample:\n    ```python\n    device = LoopDevice(name='loop0').discover()\n    device = LoopDevice.create(size_mb=1024)\n    with LoopDevice.using(size_mb=512) as dev:\n        ...\n    ```",
  "properties": {
    "path": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "format": "path",
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Path"
    },
    "name": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Name"
    },
    "size": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Size"
    },
    "model": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Model"
    },
    "image_path": {
      "anyOf": [
        {
          "format": "path",
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Image Path"
    },
    "info": {
      "anyOf": [
        {
          "$ref": "#/$defs/LoopDeviceInfo"
        },
        {
          "type": "null"
        }
      ],
      "default": null
    },
    "backing_file_exists": {
      "default": false,
      "title": "Backing File Exists",
      "type": "boolean"
    }
  },
  "title": "LoopDevice",
  "type": "object"
}

Fields:

• name (str | None)
• path (PathOrStr | None)
• size (int | None)
• model (str | None)
• image_path (Path | None)
• info (LoopDeviceInfo | None)
• backing_file_exists (bool)

Validators:

• _derive_fields

Source code in sts_libs/src/sts/loop.py

class LoopDevice(StorageDevice):
    """Loop device backed by a regular file.

    Construction performs no I/O. Call ``discover()`` to populate
    attributes from losetup/blockdev, or use the ``create()`` /
    ``using()`` class methods. Supports context-manager cleanup.

    Example:
        ```python
        device = LoopDevice(name='loop0').discover()
        device = LoopDevice.create(size_mb=1024)
        with LoopDevice.using(size_mb=512) as dev:
            ...
        ```
    """

    # Optional parameters from parent classes
    name: str | None = None
    path: PathOrStr | None = None
    size: int | None = None
    model: str | None = None

    # Optional parameters for this class
    image_path: Path | None = None

    # Discovered state (populated by discover(), not settable at construction)
    info: LoopDeviceInfo | None = Field(default=None, init=False, repr=False)
    backing_file_exists: bool = Field(default=False, init=False, repr=False)

    # Class-level paths
    LOOP_PATH: ClassVar[Path] = Path('/dev')  # Device nodes
    DEFAULT_IMAGE_PATH: ClassVar[Path] = Path('/var/tmp')  # Default image location

    @model_validator(mode='after')
    def _derive_fields(self) -> Self:
        """Derive path/name fields only — no I/O.

        Call discover() after construction to load state for existing devices.
        """
        if not self.path and self.name:
            self.path = Path(f'/dev/{self.name}')
        if self.path and not self.name:
            self.name = Path(self.path).name
        return self

    def discover(self) -> Self:
        """Load loop device state (losetup, blockdev) from the system; returns self for chaining."""
        info = self._get_device_info()
        if info:
            self.info = info
            if info.back_file:
                self.image_path = Path(info.back_file)
            self.backing_file_exists = self.image_path.exists() if self.image_path else False

        if self.path:
            result = run(f'blockdev --getsize64 {self.path}')
            if result.succeeded:
                self.size = int(result.stdout.strip())

        return self

    def _get_device_info(self) -> LoopDeviceInfo | None:
        """Parse ``losetup -lJ`` output for this device."""
        result = run(f'losetup -lJ {self.path}')
        if result.failed or not result.stdout:
            return None

        try:
            data = json.loads(result.stdout)
            devices = data.get('loopdevices', [])
            if not devices:
                return None
            return LoopDeviceInfo.model_validate(devices[0])
        except (json.JSONDecodeError, KeyError, IndexError):
            return None

    def __enter__(self) -> Self:
        return self

    def __exit__(
        self,
        exc_type: type[BaseException] | None,
        exc_val: BaseException | None,
        exc_tb: TracebackType | None,
    ) -> None:
        """Remove device and backing file on context exit."""
        self.remove()

    @property
    def device_path(self) -> Path:
        """Resolved path to the device node.

        Raises:
            DeviceNotFoundError: If device name is unset or node does not exist.
        """
        if not self.name:
            msg = 'Device name not available'
            raise DeviceNotFoundError(msg)

        path = self.LOOP_PATH / self.name
        if not path.exists():
            msg = f'Device {self.name} not found'
            raise DeviceNotFoundError(msg)
        return path

    @property
    def backing_file(self) -> Path | None:
        """Path to backing file, or None if not attached or file missing."""
        if not self.path:
            return None

        info = self._get_device_info()
        if not info or not info.back_file:
            return None

        path = Path(info.back_file)
        return path if path.exists() else None

    def detach(self) -> bool:
        """Detach device from backing file (file is not deleted)."""
        if not self.path:
            logger.error('Device path not available')
            return False

        result = run(f'losetup -d {self.path}')
        if result.failed:
            logger.error('Failed to detach device')
            return False
        return True

    def remove(self) -> bool:
        """Detach device and delete backing file."""
        # Save image path before detaching
        image_path = self.image_path

        # Detach device
        if not self.detach():
            return False

        # Remove backing file if we have a valid path
        if image_path and image_path.exists() and image_path.is_file():
            try:
                image_path.unlink()
            except OSError:
                logger.exception('Failed to remove backing file')
                return False

        return True

    @classmethod
    def _prepare_image_file(
        cls,
        name: str,
        image_path: Path,
        size_mb: int,
        *,
        reuse_file: bool = False,
    ) -> Path | None:
        """Create (or reuse) a sparse backing file for a loop device."""
        image_file = image_path / f'{name}{DEFAULT_IMAGE_SUFFIX}'

        # Handle existing file
        if image_file.exists():
            if reuse_file:
                return image_file

            try:
                image_file.unlink()
            except OSError:
                logger.exception('Failed to remove existing file')
                return None

        # Create parent directory
        try:
            image_path.mkdir(parents=True, exist_ok=True)
        except OSError:
            logger.exception('Failed to create image directory')
            return None

        # Create sparse file (allocates blocks only when written)
        result = run(f'fallocate -l {size_mb}M {image_file}')
        if result.failed:
            logger.error('Failed to create image file')
            return None

        return image_file

    @classmethod
    def _attach_device(cls, name: str, image_file: Path) -> bool:
        """Attach loop device to image file; cleans up on failure."""
        result = run(f'losetup /dev/{name} {image_file}')
        if result.failed:
            logger.error('Failed to attach device')
            try:
                image_file.unlink()
            except OSError:
                logger.exception('Failed to clean up image file')
            return False
        return True

    @classmethod
    def create(
        cls,
        name: str | None = None,
        *,
        size_mb: int = DEFAULT_SIZE_MB,
        image_path: str | Path = DEFAULT_IMAGE_PATH,
        reuse_file: bool = False,
    ) -> Self | None:
        """Create a loop device with a backing file, returning None on failure.

        Auto-detects the next free device if *name* is not provided.
        """
        # Get next available device if name not provided
        if not name:
            result = run('losetup -f')
            if result.failed:
                logger.error('Failed to find free device')
                return None
            name = Path(result.stdout.strip()).name

        # Ensure name is standardized (remove /dev/ prefix)
        device_name = name.replace('/dev/', '')

        # Create image file
        image_file = cls._prepare_image_file(
            device_name,
            Path(image_path),
            size_mb,
            reuse_file=reuse_file,
        )
        if not image_file:
            return None

        # Attach device
        if not cls._attach_device(device_name, image_file):
            return None

        return cls(name=device_name, image_path=image_file).discover()

    @classmethod
    def using(
        cls,
        name: str | None = None,
        *,
        size_mb: int = DEFAULT_SIZE_MB,
        image_path: str | Path = DEFAULT_IMAGE_PATH,
        reuse_file: bool = False,
    ) -> Self:
        """Like ``create()``, but raises on failure (for use as a context manager)."""
        device = cls.create(
            name,
            size_mb=size_mb,
            image_path=image_path,
            reuse_file=reuse_file,
        )
        if not device:
            msg = 'Failed to create loop device'
            raise DeviceError(msg)
        return device

    @classmethod
    def create_multiple(
        cls,
        count: int,
        *,
        size_mb: int = DEFAULT_SIZE_MB,
        image_path: str | Path = DEFAULT_IMAGE_PATH,
    ) -> list[Self]:
        """Create *count* loop devices, cleaning up all on failure."""
        devices: list[Self] = []
        for i in range(count):
            device = cls.create(size_mb=size_mb, image_path=image_path)
            if not device:
                for dev in devices:
                    dev.remove()
                msg = f'Failed to create loop device {i + 1} of {count}'
                raise DeviceError(msg)
            devices.append(device)
        return devices

    @classmethod
    def get_all(cls) -> Sequence[Self]:
        """Get all active loop devices."""
        result = run('losetup -lJ')
        if result.failed:
            logger.warning('No loop devices found')
            return []

        try:
            data = json.loads(result.stdout)
            devices = data.get('loopdevices', [])
            return [cls(name=Path(dev['name']).name).discover() for dev in devices]
        except (json.JSONDecodeError, KeyError):
            logger.warning('Failed to parse loop devices')
            return []

    @classmethod
    def get_by_name(cls, name: str) -> Self | None:
        """Find an active loop device by name (e.g. 'loop0')."""
        if not name:
            msg = 'Device name required'
            raise ValueError(msg)

        # Ensure name is standardized (remove /dev/ prefix)
        name = name.replace('/dev/', '')

        for device in cls.get_all():
            if device.name == name:
                return device

        return None

backing_file property

Path to backing file, or None if not attached or file missing.

device_path property

Resolved path to the device node.

Raises:

Type	Description
DeviceNotFoundError	If device name is unset or node does not exist.

__exit__(exc_type, exc_val, exc_tb)

Remove device and backing file on context exit.

Source code in sts_libs/src/sts/loop.py

def __exit__(
    self,
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: TracebackType | None,
) -> None:
    """Remove device and backing file on context exit."""
    self.remove()

create(name=None, *, size_mb=DEFAULT_SIZE_MB, image_path=DEFAULT_IMAGE_PATH, reuse_file=False) classmethod

Create a loop device with a backing file, returning None on failure.

Auto-detects the next free device if name is not provided.

Source code in sts_libs/src/sts/loop.py

@classmethod
def create(
    cls,
    name: str | None = None,
    *,
    size_mb: int = DEFAULT_SIZE_MB,
    image_path: str | Path = DEFAULT_IMAGE_PATH,
    reuse_file: bool = False,
) -> Self | None:
    """Create a loop device with a backing file, returning None on failure.

    Auto-detects the next free device if *name* is not provided.
    """
    # Get next available device if name not provided
    if not name:
        result = run('losetup -f')
        if result.failed:
            logger.error('Failed to find free device')
            return None
        name = Path(result.stdout.strip()).name

    # Ensure name is standardized (remove /dev/ prefix)
    device_name = name.replace('/dev/', '')

    # Create image file
    image_file = cls._prepare_image_file(
        device_name,
        Path(image_path),
        size_mb,
        reuse_file=reuse_file,
    )
    if not image_file:
        return None

    # Attach device
    if not cls._attach_device(device_name, image_file):
        return None

    return cls(name=device_name, image_path=image_file).discover()

create_multiple(count, *, size_mb=DEFAULT_SIZE_MB, image_path=DEFAULT_IMAGE_PATH) classmethod

Create count loop devices, cleaning up all on failure.

Source code in sts_libs/src/sts/loop.py

@classmethod
def create_multiple(
    cls,
    count: int,
    *,
    size_mb: int = DEFAULT_SIZE_MB,
    image_path: str | Path = DEFAULT_IMAGE_PATH,
) -> list[Self]:
    """Create *count* loop devices, cleaning up all on failure."""
    devices: list[Self] = []
    for i in range(count):
        device = cls.create(size_mb=size_mb, image_path=image_path)
        if not device:
            for dev in devices:
                dev.remove()
            msg = f'Failed to create loop device {i + 1} of {count}'
            raise DeviceError(msg)
        devices.append(device)
    return devices

detach()

Detach device from backing file (file is not deleted).

Source code in sts_libs/src/sts/loop.py

def detach(self) -> bool:
    """Detach device from backing file (file is not deleted)."""
    if not self.path:
        logger.error('Device path not available')
        return False

    result = run(f'losetup -d {self.path}')
    if result.failed:
        logger.error('Failed to detach device')
        return False
    return True

discover()

Load loop device state (losetup, blockdev) from the system; returns self for chaining.

Source code in sts_libs/src/sts/loop.py

def discover(self) -> Self:
    """Load loop device state (losetup, blockdev) from the system; returns self for chaining."""
    info = self._get_device_info()
    if info:
        self.info = info
        if info.back_file:
            self.image_path = Path(info.back_file)
        self.backing_file_exists = self.image_path.exists() if self.image_path else False

    if self.path:
        result = run(f'blockdev --getsize64 {self.path}')
        if result.succeeded:
            self.size = int(result.stdout.strip())

    return self

get_all() classmethod

Get all active loop devices.

Source code in sts_libs/src/sts/loop.py

@classmethod
def get_all(cls) -> Sequence[Self]:
    """Get all active loop devices."""
    result = run('losetup -lJ')
    if result.failed:
        logger.warning('No loop devices found')
        return []

    try:
        data = json.loads(result.stdout)
        devices = data.get('loopdevices', [])
        return [cls(name=Path(dev['name']).name).discover() for dev in devices]
    except (json.JSONDecodeError, KeyError):
        logger.warning('Failed to parse loop devices')
        return []

get_by_name(name) classmethod

Find an active loop device by name (e.g. 'loop0').

Source code in sts_libs/src/sts/loop.py

@classmethod
def get_by_name(cls, name: str) -> Self | None:
    """Find an active loop device by name (e.g. 'loop0')."""
    if not name:
        msg = 'Device name required'
        raise ValueError(msg)

    # Ensure name is standardized (remove /dev/ prefix)
    name = name.replace('/dev/', '')

    for device in cls.get_all():
        if device.name == name:
            return device

    return None

remove()

Detach device and delete backing file.

Source code in sts_libs/src/sts/loop.py

def remove(self) -> bool:
    """Detach device and delete backing file."""
    # Save image path before detaching
    image_path = self.image_path

    # Detach device
    if not self.detach():
        return False

    # Remove backing file if we have a valid path
    if image_path and image_path.exists() and image_path.is_file():
        try:
            image_path.unlink()
        except OSError:
            logger.exception('Failed to remove backing file')
            return False

    return True

using(name=None, *, size_mb=DEFAULT_SIZE_MB, image_path=DEFAULT_IMAGE_PATH, reuse_file=False) classmethod

Like create(), but raises on failure (for use as a context manager).

Source code in sts_libs/src/sts/loop.py

@classmethod
def using(
    cls,
    name: str | None = None,
    *,
    size_mb: int = DEFAULT_SIZE_MB,
    image_path: str | Path = DEFAULT_IMAGE_PATH,
    reuse_file: bool = False,
) -> Self:
    """Like ``create()``, but raises on failure (for use as a context manager)."""
    device = cls.create(
        name,
        size_mb=size_mb,
        image_path=image_path,
        reuse_file=reuse_file,
    )
    if not device:
        msg = 'Failed to create loop device'
        raise DeviceError(msg)
    return device

LoopDeviceInfo pydantic-model

Bases: ReportModel

Parsed losetup -J output for a single loop device.

Attributes:

Name	Type	Description
sizelimit	int | None	Size limit in bytes (0 = unlimited).
offset	int | None	Offset in bytes (0 = start of file).
autoclear	CoerceBool	Whether device is automatically detached on last close.
dio	CoerceBool	Whether direct I/O is enabled (bypass page cache).

Show JSON schema:

{
  "description": "Parsed ``losetup -J`` output for a single loop device.\n\nAttributes:\n    sizelimit: Size limit in bytes (0 = unlimited).\n    offset: Offset in bytes (0 = start of file).\n    autoclear: Whether device is automatically detached on last close.\n    dio: Whether direct I/O is enabled (bypass page cache).",
  "properties": {
    "name": {
      "title": "Name",
      "type": "string"
    },
    "back-file": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Back-File"
    },
    "sizelimit": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Sizelimit"
    },
    "offset": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Offset"
    },
    "autoclear": {
      "default": false,
      "title": "Autoclear",
      "type": "boolean"
    },
    "ro": {
      "default": false,
      "title": "Ro",
      "type": "boolean"
    },
    "dio": {
      "default": false,
      "title": "Dio",
      "type": "boolean"
    },
    "log-sec": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Log-Sec"
    }
  },
  "required": [
    "name"
  ],
  "title": "LoopDeviceInfo",
  "type": "object"
}

Config:

• populate_by_name: True

Fields:

• name (str)
• back_file (str | None)
• sizelimit (int | None)
• offset (int | None)
• autoclear (CoerceBool)
• ro (CoerceBool)
• dio (CoerceBool)
• log_sec (int | None)

Source code in sts_libs/src/sts/loop.py

class LoopDeviceInfo(ReportModel):
    """Parsed ``losetup -J`` output for a single loop device.

    Attributes:
        sizelimit: Size limit in bytes (0 = unlimited).
        offset: Offset in bytes (0 = start of file).
        autoclear: Whether device is automatically detached on last close.
        dio: Whether direct I/O is enabled (bypass page cache).
    """

    model_config = ConfigDict(populate_by_name=True)

    name: str
    back_file: str | None = Field(default=None, alias='back-file')
    sizelimit: int | None = None
    offset: int | None = None
    autoclear: CoerceBool = False
    ro: CoerceBool = False
    dio: CoerceBool = False
    log_sec: int | None = Field(default=None, alias='log-sec')