Skip to content

SCSI Debug

The scsi_debug kernel module creates virtual SCSI devices for testing — configurable size, sector size, number of devices, and error injection, without requiring real hardware.

sts.scsi_debug

Virtual SCSI devices via the scsi_debug kernel module for testing.

ScsiDebugDevice pydantic-model

Bases: StorageDevice

Virtual SCSI device backed by the scsi_debug kernel module.

Example 
device = ScsiDebugDevice.create(size=1024 * 1024 * 1024)
Show JSON schema: 
{
  "additionalProperties": false,
  "description": "Virtual SCSI device backed by the scsi_debug kernel module.\n\nExample:\n    ```python\n    device = ScsiDebugDevice.create(size=1024 * 1024 * 1024)\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": "SCSI Debug",
      "title": "Model"
    }
  },
  "title": "ScsiDebugDevice",
  "type": "object"
}

Fields:

  • name (str | None)
  • path (PathOrStr | None)
  • size (int | None)
  • model (str | None)

Validators:

  • _derive_fields
Source code in sts_libs/src/sts/scsi_debug.py 
 23
 24
 25
 26
 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
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
class ScsiDebugDevice(StorageDevice):
    """Virtual SCSI device backed by the scsi_debug kernel module.

    Example:
        ```python
        device = ScsiDebugDevice.create(size=1024 * 1024 * 1024)
        ```
    """

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

    _module: ModuleManager = PrivateAttr(default_factory=ModuleManager)
    _multipath: MultipathService = PrivateAttr(default_factory=MultipathService)

    # Sysfs path for module parameters
    SYSFS_PATH: ClassVar[Path] = Path('/sys/bus/pseudo/drivers/scsi_debug')

    @staticmethod
    def get_scsi_name_by_vendor(vendor: str) -> list[str] | None:
        """Get SCSI device names matching a vendor string via lsscsi."""
        result = run('lsscsi -s')
        if result.failed:
            return None

        devices: list[str] = []
        for line in result.stdout.splitlines():
            if vendor in line:
                # Parse line like: [0:0:0:0] disk Linux SCSI disk 1.0 /dev/sda 1024M
                parts = line.split()
                if len(parts) >= 6:
                    devices.append(parts[5].split('/')[-1])

        return devices or None

    @classmethod
    def create(cls, *, size: int | None = None, options: str | None = None) -> ScsiDebugDevice | None:
        """Create a virtual SCSI device by loading the scsi_debug module.

        Args:
            size: Device size in bytes (minimum 1MB, default 8MB)
            options: Additional module options (e.g. 'num_tgts=2 max_luns=4')
        """
        # Convert size to megabytes for module parameter
        if size:
            size_mb = size // (1024 * 1024)
            size_mb = max(size_mb, 1)  # Minimum 1MB
        else:
            size_mb = 8  # Default 8MB

        # Build module options
        module_options = f'dev_size_mb={size_mb}'
        if options:
            module_options = f'{module_options} {options}'

        # Load scsi_debug module with options
        module = ModuleManager()
        if not module.load('scsi_debug', module_options):
            return None

        # Get created device names
        devices = cls.get_devices()
        if not devices:
            return None

        # Return first device
        return cls(name=devices[0], size=size)

    def remove(self) -> CommandResult:
        """Flush multipath devices and unload the scsi_debug module."""
        command = 'modprobe -r scsi_debug'

        # Remove multipath devices if active
        if self._multipath.is_running():
            for _mpath in MultipathDevice.get_by_vendor('Linux'):
                if not self._multipath.flush():
                    return CommandResult(command='multipath -F', rc=1, stderr='Failed to flush multipath devices')

        # Unload scsi_debug module
        if self._module.unload('scsi_debug'):
            return CommandResult(command=command, rc=0)
        return CommandResult(command=command, rc=1, stderr='Failed to unload scsi_debug module')

    def set_param(self, param_name: str, value: str | int) -> bool:
        """Set a scsi_debug module parameter via sysfs."""
        param = self.SYSFS_PATH / param_name
        try:
            param.write_text(str(value))
        except OSError:
            logger.exception(f'Failed to set parameter: {param_name}={value}')
            return False

        return True

    def inject_failure(self, every_nth: int = 0, opts: int = 0) -> bool:
        """Inject device failures.

        Args:
            every_nth: How often to inject failure (0 = disabled)
            opts: Failure options bitmask:
                1 = noisy (log details), 2 = medium error,
                4 = ignore nth (always fail), 8 = RECOVERED_ERROR,
                16 = ABORTED_COMMAND

        Example:
            ```python
            device.inject_failure(every_nth=1, opts=2)  # media errors on every op
            ```
        """
        if not self.set_param('every_nth', every_nth):
            return False
        return self.set_param('opts', opts)

    @classmethod
    def get_devices(cls) -> list[str] | None:
        """Get device names from the loaded scsi_debug module, or None if not loaded."""
        # Check if module is loaded
        if not ModuleInfo.from_name('scsi_debug'):
            return None

        # Fall back to direct SCSI devices
        return cls.get_scsi_name_by_vendor('Linux')

create(*, size=None, options=None) classmethod

Create a virtual SCSI device by loading the scsi_debug module.

Parameters:

Name Type Description Default
size int | None

Device size in bytes (minimum 1MB, default 8MB)

 None
options str | None

Additional module options (e.g. 'num_tgts=2 max_luns=4')

 None
Source code in sts_libs/src/sts/scsi_debug.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
@classmethod
def create(cls, *, size: int | None = None, options: str | None = None) -> ScsiDebugDevice | None:
    """Create a virtual SCSI device by loading the scsi_debug module.

    Args:
        size: Device size in bytes (minimum 1MB, default 8MB)
        options: Additional module options (e.g. 'num_tgts=2 max_luns=4')
    """
    # Convert size to megabytes for module parameter
    if size:
        size_mb = size // (1024 * 1024)
        size_mb = max(size_mb, 1)  # Minimum 1MB
    else:
        size_mb = 8  # Default 8MB

    # Build module options
    module_options = f'dev_size_mb={size_mb}'
    if options:
        module_options = f'{module_options} {options}'

    # Load scsi_debug module with options
    module = ModuleManager()
    if not module.load('scsi_debug', module_options):
        return None

    # Get created device names
    devices = cls.get_devices()
    if not devices:
        return None

    # Return first device
    return cls(name=devices[0], size=size)

get_devices() classmethod

Get device names from the loaded scsi_debug module, or None if not loaded.

Source code in sts_libs/src/sts/scsi_debug.py 
138
139
140
141
142
143
144
145
146
@classmethod
def get_devices(cls) -> list[str] | None:
    """Get device names from the loaded scsi_debug module, or None if not loaded."""
    # Check if module is loaded
    if not ModuleInfo.from_name('scsi_debug'):
        return None

    # Fall back to direct SCSI devices
    return cls.get_scsi_name_by_vendor('Linux')

get_scsi_name_by_vendor(vendor) staticmethod

Get SCSI device names matching a vendor string via lsscsi.

Source code in sts_libs/src/sts/scsi_debug.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
@staticmethod
def get_scsi_name_by_vendor(vendor: str) -> list[str] | None:
    """Get SCSI device names matching a vendor string via lsscsi."""
    result = run('lsscsi -s')
    if result.failed:
        return None

    devices: list[str] = []
    for line in result.stdout.splitlines():
        if vendor in line:
            # Parse line like: [0:0:0:0] disk Linux SCSI disk 1.0 /dev/sda 1024M
            parts = line.split()
            if len(parts) >= 6:
                devices.append(parts[5].split('/')[-1])

    return devices or None

inject_failure(every_nth=0, opts=0)

Inject device failures.

Parameters:

Name Type Description Default
every_nth int

How often to inject failure (0 = disabled)

 0
opts int

Failure options bitmask: 1 = noisy (log details), 2 = medium error, 4 = ignore nth (always fail), 8 = RECOVERED_ERROR, 16 = ABORTED_COMMAND

 0
Example 
device.inject_failure(every_nth=1, opts=2)  # media errors on every op
Source code in sts_libs/src/sts/scsi_debug.py 
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
def inject_failure(self, every_nth: int = 0, opts: int = 0) -> bool:
    """Inject device failures.

    Args:
        every_nth: How often to inject failure (0 = disabled)
        opts: Failure options bitmask:
            1 = noisy (log details), 2 = medium error,
            4 = ignore nth (always fail), 8 = RECOVERED_ERROR,
            16 = ABORTED_COMMAND

    Example:
        ```python
        device.inject_failure(every_nth=1, opts=2)  # media errors on every op
        ```
    """
    if not self.set_param('every_nth', every_nth):
        return False
    return self.set_param('opts', opts)

remove()

Flush multipath devices and unload the scsi_debug module.

Source code in sts_libs/src/sts/scsi_debug.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
def remove(self) -> CommandResult:
    """Flush multipath devices and unload the scsi_debug module."""
    command = 'modprobe -r scsi_debug'

    # Remove multipath devices if active
    if self._multipath.is_running():
        for _mpath in MultipathDevice.get_by_vendor('Linux'):
            if not self._multipath.flush():
                return CommandResult(command='multipath -F', rc=1, stderr='Failed to flush multipath devices')

    # Unload scsi_debug module
    if self._module.unload('scsi_debug'):
        return CommandResult(command=command, rc=0)
    return CommandResult(command=command, rc=1, stderr='Failed to unload scsi_debug module')

set_param(param_name, value)

Set a scsi_debug module parameter via sysfs.

Source code in sts_libs/src/sts/scsi_debug.py 
108
109
110
111
112
113
114
115
116
117
def set_param(self, param_name: str, value: str | int) -> bool:
    """Set a scsi_debug module parameter via sysfs."""
    param = self.SYSFS_PATH / param_name
    try:
        param.write_text(str(value))
    except OSError:
        logger.exception(f'Failed to set parameter: {param_name}={value}')
        return False

    return True