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: ScsiDevice

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"
    },
    "scsi_id": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Scsi Id"
    },
    "host_id": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Host Id"
    }
  },
  "title": "ScsiDebugDevice",
  "type": "object"
}

Fields:

  • path (PathOrStr | None)
  • name (str | None)
  • size (int | None)
  • scsi_id (str | None)
  • host_id (str | None)
  • model (str | None)
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
 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
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
class ScsiDebugDevice(ScsiDevice):
    """Virtual SCSI device backed by the scsi_debug kernel module.

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

    model: str | None = 'SCSI Debug'

    DRIVER_PATH: ClassVar[Path] = Path('/sys/bus/pseudo/drivers/scsi_debug')
    SYSFS_PATH: ClassVar[Path] = DRIVER_PATH
    TRIGGER_PARAMETERS: ClassVar[frozenset[str]] = frozenset({'add_host'})

    _lifecycle: ManagedModule | None = PrivateAttr(default=None)

    @staticmethod
    def get_scsi_name_by_vendor(vendor: str) -> list[str] | None:
        """Get SCSI device names matching a vendor string via lsscsi.

        This legacy helper is retained for callers that explicitly need a
        vendor query. scsi_debug discovery uses :attr:`DRIVER_PATH` instead.
        """
        result = run('lsscsi -s')
        if result.failed:
            return None

        devices: list[str] = []
        for line in result.stdout.splitlines():
            if vendor in line:
                parts = line.split()
                if len(parts) >= 6:
                    devices.append(parts[5].split('/')[-1])
        return devices or None

    @staticmethod
    def option_mapping(options: object) -> dict[str, str | int | bool] | None:
        """Convert supported module options into a validated mapping."""
        if options is None:
            return {}
        if isinstance(options, Mapping):
            raw_options = cast('Mapping[object, object]', options)
            if not all(
                isinstance(key, str) and isinstance(value, (str, int, bool)) for key, value in raw_options.items()
            ):
                return None
            return cast('dict[str, str | int | bool]', dict(raw_options))
        if not isinstance(options, str):
            return None

        parsed: dict[str, str | int | bool] = {}
        try:
            tokens = shlex.split(options)
        except ValueError:
            return None
        for option in tokens:
            key, separator, value = option.partition('=')
            if not separator or not key or not value or key in parsed:
                return None
            parsed[key] = value
        return parsed

    @staticmethod
    def parameters_match(actual: Mapping[str, object], expected: Mapping[str, str | int | bool]) -> bool:
        """Return whether readable module parameters prove the requested configuration."""
        return parameters_match(actual, expected)

    @classmethod
    def create(
        cls,
        *,
        size: int | None = None,
        options: str | Mapping[str, str | int | bool] | None = None,
    ) -> ScsiDebugDevice | None:
        """Create a virtual SCSI device with verifiable module options.

        Args:
            size: Device size in bytes (minimum 1 MiB, default 8 MiB).
            options: Additional module options as a shell-like string or mapping.
        """
        if size is not None and (type(size) is not int or size < 1):
            raise DeviceError('scsi_debug size must be a positive integer')
        size_mb = max(size // MEBIBYTE, 1) if size is not None else 8
        module_options = cls.option_mapping(options)
        if module_options is None or 'dev_size_mb' in module_options:
            raise DeviceError('scsi_debug options must be valid and must not override dev_size_mb')
        all_params = {'dev_size_mb': size_mb, **module_options}
        parameters = {k: v for k, v in all_params.items() if k not in cls.TRIGGER_PARAMETERS}
        trigger = {k: v for k, v in all_params.items() if k in cls.TRIGGER_PARAMETERS}

        lifecycle = ManagedModule('scsi_debug', parameters=parameters, load_only_parameters=trigger)
        try:
            try:
                module_info = lifecycle.acquire()
            except ModuleUnavailableError:
                return None
            device = cls._from_acquired_module(module_info, lifecycle, size=size, parameters=parameters)
        except BaseException as error:
            if lifecycle.owns_load and not lifecycle.release():
                cleanup_error = ModuleUnloadError('Failed to unload owned scsi_debug module after setup failure')
                raise BaseExceptionGroup(
                    'scsi_debug setup and cleanup failed',
                    [error, cleanup_error],
                ) from None
            raise
        else:
            return device

    @classmethod
    def _from_acquired_module(
        cls,
        module_info: ModuleInfo,
        lifecycle: ManagedModule,
        *,
        size: int | None,
        parameters: Mapping[str, str | int | bool],
    ) -> ScsiDebugDevice | None:
        """Validate post-load state and build the exact discovered device."""
        if not cls.parameters_match(module_info.parameters, parameters):
            if lifecycle.owns_load:
                raise DeviceError('Owned scsi_debug load has unverified parameters')
            return None

        devices = cls.get_devices() or []
        if len(devices) != 1:
            raise DeviceError(f'scsi_debug created {len(devices)} devices; expected one device')

        device = cls(name=devices[0], size=size)
        if not device.wait_for_path():
            raise DeviceError(f'Timed out waiting for scsi_debug device {device.path}')
        device._lifecycle = lifecycle
        return device

    def remove(self) -> CommandResult:
        """Release the scsi_debug module only when this device loaded it."""
        command = 'modprobe -r scsi_debug'
        if self._lifecycle is None or not self._lifecycle.owns_load:
            return CommandResult(command=command, rc=1, stderr='scsi_debug is not owned by this device')
        if self._lifecycle.release():
            if self.wait_for_path(present=False):
                return CommandResult(command=command, rc=0)
            return CommandResult(
                command=command,
                rc=1,
                stderr=f'Timed out waiting for scsi_debug device {self.path} to disappear',
            )
        return CommandResult(command=command, rc=1, stderr='Failed to unload owned 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.
        """
        return self.set_param('every_nth', every_nth) and self.set_param('opts', opts)

    @classmethod
    def get_devices(cls) -> list[str] | None:
        """Get block device names attached to hosts owned by scsi_debug."""
        host_ids: set[str] = set()
        for adapter in _directory_entries(cls.DRIVER_PATH):
            if not adapter.name.startswith('adapter'):
                continue

            host_ids.update(
                host_id
                for host in _directory_entries(adapter / 'scsi_host')
                if host.is_dir() and (host_id := _numeric_suffix(host.name, 'host')) is not None
            )
            for host_path in _directory_entries(adapter):
                host_id = _numeric_suffix(host_path.name, 'host')
                if host_id is not None and (host_path / 'scsi_host' / host_path.name).is_dir():
                    host_ids.add(host_id)

        if not host_ids:
            return None

        devices = sorted(
            {device.name for device in cls.get_all_scsi_devices() if device.host_id in host_ids and device.name}
        )
        return devices or None

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

Create a virtual SCSI device with verifiable module options.

Parameters:

Name Type Description Default
size int | None

Device size in bytes (minimum 1 MiB, default 8 MiB).

 None
options str | Mapping[str, str | int | bool] | None

Additional module options as a shell-like string or mapping.

 None
Source code in sts_libs/src/sts/scsi_debug.py 
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
147
148
149
150
@classmethod
def create(
    cls,
    *,
    size: int | None = None,
    options: str | Mapping[str, str | int | bool] | None = None,
) -> ScsiDebugDevice | None:
    """Create a virtual SCSI device with verifiable module options.

    Args:
        size: Device size in bytes (minimum 1 MiB, default 8 MiB).
        options: Additional module options as a shell-like string or mapping.
    """
    if size is not None and (type(size) is not int or size < 1):
        raise DeviceError('scsi_debug size must be a positive integer')
    size_mb = max(size // MEBIBYTE, 1) if size is not None else 8
    module_options = cls.option_mapping(options)
    if module_options is None or 'dev_size_mb' in module_options:
        raise DeviceError('scsi_debug options must be valid and must not override dev_size_mb')
    all_params = {'dev_size_mb': size_mb, **module_options}
    parameters = {k: v for k, v in all_params.items() if k not in cls.TRIGGER_PARAMETERS}
    trigger = {k: v for k, v in all_params.items() if k in cls.TRIGGER_PARAMETERS}

    lifecycle = ManagedModule('scsi_debug', parameters=parameters, load_only_parameters=trigger)
    try:
        try:
            module_info = lifecycle.acquire()
        except ModuleUnavailableError:
            return None
        device = cls._from_acquired_module(module_info, lifecycle, size=size, parameters=parameters)
    except BaseException as error:
        if lifecycle.owns_load and not lifecycle.release():
            cleanup_error = ModuleUnloadError('Failed to unload owned scsi_debug module after setup failure')
            raise BaseExceptionGroup(
                'scsi_debug setup and cleanup failed',
                [error, cleanup_error],
            ) from None
        raise
    else:
        return device

get_devices() classmethod

Get block device names attached to hosts owned by scsi_debug.

Source code in sts_libs/src/sts/scsi_debug.py 
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
@classmethod
def get_devices(cls) -> list[str] | None:
    """Get block device names attached to hosts owned by scsi_debug."""
    host_ids: set[str] = set()
    for adapter in _directory_entries(cls.DRIVER_PATH):
        if not adapter.name.startswith('adapter'):
            continue

        host_ids.update(
            host_id
            for host in _directory_entries(adapter / 'scsi_host')
            if host.is_dir() and (host_id := _numeric_suffix(host.name, 'host')) is not None
        )
        for host_path in _directory_entries(adapter):
            host_id = _numeric_suffix(host_path.name, 'host')
            if host_id is not None and (host_path / 'scsi_host' / host_path.name).is_dir():
                host_ids.add(host_id)

    if not host_ids:
        return None

    devices = sorted(
        {device.name for device in cls.get_all_scsi_devices() if device.host_id in host_ids and device.name}
    )
    return devices or None

get_scsi_name_by_vendor(vendor) staticmethod

Get SCSI device names matching a vendor string via lsscsi.

This legacy helper is retained for callers that explicitly need a vendor query. scsi_debug discovery uses :attr:DRIVER_PATH instead.

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
@staticmethod
def get_scsi_name_by_vendor(vendor: str) -> list[str] | None:
    """Get SCSI device names matching a vendor string via lsscsi.

    This legacy helper is retained for callers that explicitly need a
    vendor query. scsi_debug discovery uses :attr:`DRIVER_PATH` instead.
    """
    result = run('lsscsi -s')
    if result.failed:
        return None

    devices: list[str] = []
    for line in result.stdout.splitlines():
        if vendor in line:
            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.

 0
Source code in sts_libs/src/sts/scsi_debug.py 
202
203
204
205
206
207
208
209
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.
    """
    return self.set_param('every_nth', every_nth) and self.set_param('opts', opts)

option_mapping(options) staticmethod

Convert supported module options into a validated mapping.

Source code in sts_libs/src/sts/scsi_debug.py 
 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
@staticmethod
def option_mapping(options: object) -> dict[str, str | int | bool] | None:
    """Convert supported module options into a validated mapping."""
    if options is None:
        return {}
    if isinstance(options, Mapping):
        raw_options = cast('Mapping[object, object]', options)
        if not all(
            isinstance(key, str) and isinstance(value, (str, int, bool)) for key, value in raw_options.items()
        ):
            return None
        return cast('dict[str, str | int | bool]', dict(raw_options))
    if not isinstance(options, str):
        return None

    parsed: dict[str, str | int | bool] = {}
    try:
        tokens = shlex.split(options)
    except ValueError:
        return None
    for option in tokens:
        key, separator, value = option.partition('=')
        if not separator or not key or not value or key in parsed:
            return None
        parsed[key] = value
    return parsed

parameters_match(actual, expected) staticmethod

Return whether readable module parameters prove the requested configuration.

Source code in sts_libs/src/sts/scsi_debug.py 
106
107
108
109
@staticmethod
def parameters_match(actual: Mapping[str, object], expected: Mapping[str, str | int | bool]) -> bool:
    """Return whether readable module parameters prove the requested configuration."""
    return parameters_match(actual, expected)

remove()

Release the scsi_debug module only when this device loaded it.

Source code in sts_libs/src/sts/scsi_debug.py 
177
178
179
180
181
182
183
184
185
186
187
188
189
190
def remove(self) -> CommandResult:
    """Release the scsi_debug module only when this device loaded it."""
    command = 'modprobe -r scsi_debug'
    if self._lifecycle is None or not self._lifecycle.owns_load:
        return CommandResult(command=command, rc=1, stderr='scsi_debug is not owned by this device')
    if self._lifecycle.release():
        if self.wait_for_path(present=False):
            return CommandResult(command=command, rc=0)
        return CommandResult(
            command=command,
            rc=1,
            stderr=f'Timed out waiting for scsi_debug device {self.path} to disappear',
        )
    return CommandResult(command=command, rc=1, stderr='Failed to unload owned 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 
192
193
194
195
196
197
198
199
200
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