Skip to content

SCSI Target

Linux SCSI target subsystem (targetcli / LIO) — configures the local machine as an iSCSI, FC, or other SCSI target, exporting backstores as LUNs to remote initiators.

sts.target

LIO target management via targetcli (backstores, iSCSI, loopback, ACLs, portals).

ACL pydantic-model

Bases: Targetcli

ACL (Access Control List) operations for an iSCSI target.

Show JSON schema: 
{
  "additionalProperties": false,
  "description": "ACL (Access Control List) operations for an iSCSI target.",
  "properties": {
    "path": {
      "default": "",
      "title": "Path",
      "type": "string"
    },
    "target_wwn": {
      "title": "Target Wwn",
      "type": "string"
    },
    "initiator_wwn": {
      "title": "Initiator Wwn",
      "type": "string"
    },
    "tpg": {
      "default": 1,
      "title": "Tpg",
      "type": "integer"
    },
    "acls_path": {
      "default": "",
      "title": "Acls Path",
      "type": "string"
    }
  },
  "required": [
    "target_wwn",
    "initiator_wwn"
  ],
  "title": "ACL",
  "type": "object"
}

Fields:

  • path (str)
  • target_wwn (str)
  • initiator_wwn (str)
  • tpg (int)
  • acls_path (str)

Validators:

  • _set_paths
Source code in sts_libs/src/sts/target.py 
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
class ACL(Targetcli):
    """ACL (Access Control List) operations for an iSCSI target."""

    path: str = ''
    target_wwn: str
    initiator_wwn: str
    tpg: int = 1
    acls_path: str = ''

    @model_validator(mode='after')
    def _set_paths(self) -> Self:
        """Derive acls_path and path from target_wwn, tpg and initiator_wwn."""
        if not self.acls_path:
            self.acls_path = f'/iscsi/{self.target_wwn}/tpg{self.tpg}/acls/'
        if not self.path:
            self.path = f'{self.acls_path}{self.initiator_wwn}'
        return self

    def create_acl(self) -> CommandResult:
        """Create an ACL."""
        with self.temporary_path(self.acls_path):
            return self.create(wwn=self.initiator_wwn)

    def delete_acl(self) -> CommandResult:
        """Delete the ACL."""
        with self.temporary_path(self.acls_path):
            return self.delete(wwn=self.initiator_wwn)

    def set_auth(
        self,
        userid: str | None = None,
        password: str | None = None,
        mutual_userid: str | None = None,
        mutual_password: str | None = None,
    ) -> CommandResult:
        """Set authentication for the ACL (None values become empty strings)."""
        self.set_('auth', userid='' if userid is None else userid)
        self.set_('auth', password='' if password is None else password)
        self.set_('auth', mutual_userid='' if mutual_userid is None else mutual_userid)
        self.set_('auth', mutual_password='' if mutual_password is None else mutual_password)
        return self.set_('attribute', authentication='1')

    def disable_auth(self) -> CommandResult:
        """Disable authentication for the ACL."""
        return self.set_('attribute', authentication='0')

    def map_lun(
        self,
        mapped_lun: int,
        tpg_lun_or_backstore: str,
        *,
        write_protect: bool = False,
    ) -> CommandResult:
        """Map a LUN to the ACL."""
        return self.create(
            mapped_lun=str(mapped_lun),
            tpg_lun_or_backstore=tpg_lun_or_backstore,
            write_protect=str(write_protect),
        )

create_acl()

Create an ACL.

Source code in sts_libs/src/sts/target.py 
459
460
461
462
def create_acl(self) -> CommandResult:
    """Create an ACL."""
    with self.temporary_path(self.acls_path):
        return self.create(wwn=self.initiator_wwn)

delete_acl()

Delete the ACL.

Source code in sts_libs/src/sts/target.py 
464
465
466
467
def delete_acl(self) -> CommandResult:
    """Delete the ACL."""
    with self.temporary_path(self.acls_path):
        return self.delete(wwn=self.initiator_wwn)

disable_auth()

Disable authentication for the ACL.

Source code in sts_libs/src/sts/target.py 
483
484
485
def disable_auth(self) -> CommandResult:
    """Disable authentication for the ACL."""
    return self.set_('attribute', authentication='0')

map_lun(mapped_lun, tpg_lun_or_backstore, *, write_protect=False)

Map a LUN to the ACL.

Source code in sts_libs/src/sts/target.py 
487
488
489
490
491
492
493
494
495
496
497
498
499
def map_lun(
    self,
    mapped_lun: int,
    tpg_lun_or_backstore: str,
    *,
    write_protect: bool = False,
) -> CommandResult:
    """Map a LUN to the ACL."""
    return self.create(
        mapped_lun=str(mapped_lun),
        tpg_lun_or_backstore=tpg_lun_or_backstore,
        write_protect=str(write_protect),
    )

set_auth(userid=None, password=None, mutual_userid=None, mutual_password=None)

Set authentication for the ACL (None values become empty strings).

Source code in sts_libs/src/sts/target.py 
469
470
471
472
473
474
475
476
477
478
479
480
481
def set_auth(
    self,
    userid: str | None = None,
    password: str | None = None,
    mutual_userid: str | None = None,
    mutual_password: str | None = None,
) -> CommandResult:
    """Set authentication for the ACL (None values become empty strings)."""
    self.set_('auth', userid='' if userid is None else userid)
    self.set_('auth', password='' if password is None else password)
    self.set_('auth', mutual_userid='' if mutual_userid is None else mutual_userid)
    self.set_('auth', mutual_password='' if mutual_password is None else mutual_password)
    return self.set_('attribute', authentication='1')

Backstore pydantic-model

Bases: Targetcli

Base class for backstore operations.

Show JSON schema: 
{
  "additionalProperties": false,
  "description": "Base class for backstore operations.",
  "properties": {
    "path": {
      "default": "",
      "title": "Path",
      "type": "string"
    },
    "backstore_type": {
      "enum": [
        "block",
        "fileio",
        "pscsi",
        "ramdisk"
      ],
      "title": "Backstore Type",
      "type": "string"
    }
  },
  "required": [
    "backstore_type"
  ],
  "title": "Backstore",
  "type": "object"
}

Fields:

  • path (str)
  • backstore_type (Literal['block', 'fileio', 'pscsi', 'ramdisk'])

Validators:

  • _set_path
Source code in sts_libs/src/sts/target.py 
125
126
127
128
129
130
131
132
133
134
135
136
class Backstore(Targetcli):
    """Base class for backstore operations."""

    path: str = ''
    backstore_type: Literal['block', 'fileio', 'pscsi', 'ramdisk']

    @model_validator(mode='after')
    def _set_path(self) -> Self:
        """Derive path from backstore_type when not explicitly provided."""
        if not self.path:
            self.path = f'/backstores/{self.backstore_type}/'
        return self

BackstoreBlock pydantic-model

Bases: Backstore

Block backstore operations.

Show JSON schema: 
{
  "additionalProperties": false,
  "description": "Block backstore operations.",
  "properties": {
    "path": {
      "default": "",
      "title": "Path",
      "type": "string"
    },
    "backstore_type": {
      "default": "block",
      "enum": [
        "block",
        "fileio",
        "pscsi",
        "ramdisk"
      ],
      "title": "Backstore Type",
      "type": "string"
    },
    "name": {
      "title": "Name",
      "type": "string"
    },
    "backstores_path": {
      "default": "",
      "title": "Backstores Path",
      "type": "string"
    }
  },
  "required": [
    "name"
  ],
  "title": "BackstoreBlock",
  "type": "object"
}

Fields:

  • path (str)
  • backstore_type (Literal['block', 'fileio', 'pscsi', 'ramdisk'])
  • name (str)
  • backstores_path (str)

Validators:

  • _set_path
  • _set_backstore_path
Source code in sts_libs/src/sts/target.py 
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
class BackstoreBlock(Backstore):
    """Block backstore operations."""

    backstore_type: Literal['block', 'fileio', 'pscsi', 'ramdisk'] = 'block'
    name: str
    backstores_path: str = ''

    @model_validator(mode='after')
    def _set_backstore_path(self) -> Self:
        """Save the backstores base path, then append name to path."""
        if not self.backstores_path:
            self.backstores_path = self.path
            self.path = f'{self.path}{self.name}'
        return self

    def create_backstore(self, dev: str) -> CommandResult:
        """Create a block backstore."""
        arguments = {
            'name': self.name,
            'dev': dev,
        }
        with self.temporary_path(self.backstores_path):
            return self.create(**arguments)

    def delete_backstore(self) -> CommandResult:
        """Delete the block backstore."""
        with self.temporary_path(self.backstores_path):
            return self.delete(self.name)

create_backstore(dev)

Create a block backstore.

Source code in sts_libs/src/sts/target.py 
185
186
187
188
189
190
191
192
def create_backstore(self, dev: str) -> CommandResult:
    """Create a block backstore."""
    arguments = {
        'name': self.name,
        'dev': dev,
    }
    with self.temporary_path(self.backstores_path):
        return self.create(**arguments)

delete_backstore()

Delete the block backstore.

Source code in sts_libs/src/sts/target.py 
194
195
196
197
def delete_backstore(self) -> CommandResult:
    """Delete the block backstore."""
    with self.temporary_path(self.backstores_path):
        return self.delete(self.name)

BackstoreFileio pydantic-model

Bases: Backstore

Fileio backstore operations.

Show JSON schema: 
{
  "additionalProperties": false,
  "description": "Fileio backstore operations.",
  "properties": {
    "path": {
      "default": "",
      "title": "Path",
      "type": "string"
    },
    "backstore_type": {
      "default": "fileio",
      "enum": [
        "block",
        "fileio",
        "pscsi",
        "ramdisk"
      ],
      "title": "Backstore Type",
      "type": "string"
    },
    "name": {
      "title": "Name",
      "type": "string"
    },
    "backstores_path": {
      "default": "",
      "title": "Backstores Path",
      "type": "string"
    }
  },
  "required": [
    "name"
  ],
  "title": "BackstoreFileio",
  "type": "object"
}

Fields:

  • path (str)
  • backstore_type (Literal['block', 'fileio', 'pscsi', 'ramdisk'])
  • name (str)
  • backstores_path (str)

Validators:

  • _set_path
  • _set_backstore_path
Source code in sts_libs/src/sts/target.py 
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
class BackstoreFileio(Backstore):
    """Fileio backstore operations."""

    backstore_type: Literal['block', 'fileio', 'pscsi', 'ramdisk'] = 'fileio'
    name: str
    backstores_path: str = ''

    @model_validator(mode='after')
    def _set_backstore_path(self) -> Self:
        """Save the backstores base path, then append name to path."""
        if not self.backstores_path:
            self.backstores_path = self.path
            self.path = f'{self.path}{self.name}'
        return self

    def create_backstore(self, size: str, file_or_dev: str) -> CommandResult:
        """Create a fileio backstore."""
        arguments = {
            'name': self.name,
            'size': size,
            'file_or_dev': file_or_dev,
        }
        with self.temporary_path(self.backstores_path):
            return self.create(**arguments)

    def delete_backstore(self) -> CommandResult:
        """Delete the fileio backstore."""
        with self.temporary_path(self.backstores_path):
            return self.delete(self.name)

create_backstore(size, file_or_dev)

Create a fileio backstore.

Source code in sts_libs/src/sts/target.py 
154
155
156
157
158
159
160
161
162
def create_backstore(self, size: str, file_or_dev: str) -> CommandResult:
    """Create a fileio backstore."""
    arguments = {
        'name': self.name,
        'size': size,
        'file_or_dev': file_or_dev,
    }
    with self.temporary_path(self.backstores_path):
        return self.create(**arguments)

delete_backstore()

Delete the fileio backstore.

Source code in sts_libs/src/sts/target.py 
164
165
166
167
def delete_backstore(self) -> CommandResult:
    """Delete the fileio backstore."""
    with self.temporary_path(self.backstores_path):
        return self.delete(self.name)

BackstoreRamdisk pydantic-model

Bases: Backstore

Ramdisk backstore operations.

Show JSON schema: 
{
  "additionalProperties": false,
  "description": "Ramdisk backstore operations.",
  "properties": {
    "path": {
      "default": "",
      "title": "Path",
      "type": "string"
    },
    "backstore_type": {
      "default": "ramdisk",
      "enum": [
        "block",
        "fileio",
        "pscsi",
        "ramdisk"
      ],
      "title": "Backstore Type",
      "type": "string"
    },
    "name": {
      "title": "Name",
      "type": "string"
    },
    "backstores_path": {
      "default": "",
      "title": "Backstores Path",
      "type": "string"
    }
  },
  "required": [
    "name"
  ],
  "title": "BackstoreRamdisk",
  "type": "object"
}

Fields:

  • path (str)
  • backstore_type (Literal['block', 'fileio', 'pscsi', 'ramdisk'])
  • name (str)
  • backstores_path (str)

Validators:

  • _set_path
  • _set_backstore_path
Source code in sts_libs/src/sts/target.py 
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
class BackstoreRamdisk(Backstore):
    """Ramdisk backstore operations."""

    backstore_type: Literal['block', 'fileio', 'pscsi', 'ramdisk'] = 'ramdisk'
    name: str
    backstores_path: str = ''

    @model_validator(mode='after')
    def _set_backstore_path(self) -> Self:
        """Save the backstores base path, then append name to path."""
        if not self.backstores_path:
            self.backstores_path = self.path
            self.path = f'{self.path}{self.name}'
        return self

    def create_backstore(self, size: str) -> CommandResult:
        """Create a ramdisk backstore."""
        arguments = {
            'name': self.name,
            'size': size,
        }
        with self.temporary_path(self.backstores_path):
            return self.create(**arguments)

    def delete_backstore(self) -> CommandResult:
        """Delete the ramdisk backstore."""
        with self.temporary_path(self.backstores_path):
            return self.delete(self.name)

create_backstore(size)

Create a ramdisk backstore.

Source code in sts_libs/src/sts/target.py 
215
216
217
218
219
220
221
222
def create_backstore(self, size: str) -> CommandResult:
    """Create a ramdisk backstore."""
    arguments = {
        'name': self.name,
        'size': size,
    }
    with self.temporary_path(self.backstores_path):
        return self.create(**arguments)

delete_backstore()

Delete the ramdisk backstore.

Source code in sts_libs/src/sts/target.py 
224
225
226
227
def delete_backstore(self) -> CommandResult:
    """Delete the ramdisk backstore."""
    with self.temporary_path(self.backstores_path):
        return self.delete(self.name)

Iscsi pydantic-model

Bases: Targetcli

iSCSI target operations.

Show JSON schema: 
{
  "additionalProperties": false,
  "description": "iSCSI target operations.",
  "properties": {
    "path": {
      "default": "",
      "title": "Path",
      "type": "string"
    },
    "target_wwn": {
      "title": "Target Wwn",
      "type": "string"
    },
    "tpg": {
      "default": 1,
      "title": "Tpg",
      "type": "integer"
    },
    "target_path": {
      "default": "",
      "title": "Target Path",
      "type": "string"
    }
  },
  "required": [
    "target_wwn"
  ],
  "title": "Iscsi",
  "type": "object"
}

Fields:

  • path (str)
  • target_wwn (str)
  • tpg (int)
  • target_path (str)

Validators:

  • _set_paths
Source code in sts_libs/src/sts/target.py 
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
class Iscsi(Targetcli):
    """iSCSI target operations."""

    path: str = ''
    target_wwn: str
    tpg: int = 1
    target_path: str = ''

    iscsi_path: ClassVar[str] = '/iscsi/'

    @model_validator(mode='after')
    def _set_paths(self) -> Self:
        """Derive target_path and path from target_wwn and tpg."""
        if not self.target_path:
            self.target_path = f'{self.iscsi_path}{self.target_wwn}/tpg{self.tpg}/'
        if not self.path:
            self.path = self.target_path
        return self

    def create_target(self) -> CommandResult:
        """Create an iSCSI target."""
        with self.temporary_path(self.iscsi_path):
            return self.create(wwn=self.target_wwn)

    def delete_target(self) -> CommandResult:
        """Delete the iSCSI target."""
        with self.temporary_path(self.iscsi_path):
            return self.delete(wwn=self.target_wwn)

    def set_discovery_auth(
        self,
        userid: str | None = None,
        password: str | None = None,
        mutual_userid: str | None = None,
        mutual_password: str | None = None,
    ) -> CommandResult:
        """Set discovery authentication (None values become empty strings)."""
        with self.temporary_path(self.iscsi_path):
            # Passing empty strings in one command does not work
            self.set_('discovery_auth', userid='' if userid is None else userid)
            self.set_('discovery_auth', password='' if password is None else password)
            self.set_('discovery_auth', mutual_userid='' if mutual_userid is None else mutual_userid)
            self.set_('discovery_auth', mutual_password='' if mutual_password is None else mutual_password)
            return self.set_('discovery_auth', enable='1')

    def disable_discovery_auth(self) -> CommandResult:
        """Disable discovery authentication."""
        with self.temporary_path(self.iscsi_path):
            return self.set_('discovery_auth', enable='0')

create_target()

Create an iSCSI target.

Source code in sts_libs/src/sts/target.py 
249
250
251
252
def create_target(self) -> CommandResult:
    """Create an iSCSI target."""
    with self.temporary_path(self.iscsi_path):
        return self.create(wwn=self.target_wwn)

delete_target()

Delete the iSCSI target.

Source code in sts_libs/src/sts/target.py 
254
255
256
257
def delete_target(self) -> CommandResult:
    """Delete the iSCSI target."""
    with self.temporary_path(self.iscsi_path):
        return self.delete(wwn=self.target_wwn)

disable_discovery_auth()

Disable discovery authentication.

Source code in sts_libs/src/sts/target.py 
275
276
277
278
def disable_discovery_auth(self) -> CommandResult:
    """Disable discovery authentication."""
    with self.temporary_path(self.iscsi_path):
        return self.set_('discovery_auth', enable='0')

set_discovery_auth(userid=None, password=None, mutual_userid=None, mutual_password=None)

Set discovery authentication (None values become empty strings).

Source code in sts_libs/src/sts/target.py 
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
def set_discovery_auth(
    self,
    userid: str | None = None,
    password: str | None = None,
    mutual_userid: str | None = None,
    mutual_password: str | None = None,
) -> CommandResult:
    """Set discovery authentication (None values become empty strings)."""
    with self.temporary_path(self.iscsi_path):
        # Passing empty strings in one command does not work
        self.set_('discovery_auth', userid='' if userid is None else userid)
        self.set_('discovery_auth', password='' if password is None else password)
        self.set_('discovery_auth', mutual_userid='' if mutual_userid is None else mutual_userid)
        self.set_('discovery_auth', mutual_password='' if mutual_password is None else mutual_password)
        return self.set_('discovery_auth', enable='1')

IscsiLUN pydantic-model

Bases: LUN

LUN operations scoped to an iSCSI target/TPG.

Show JSON schema: 
{
  "additionalProperties": false,
  "description": "LUN operations scoped to an iSCSI target/TPG.",
  "properties": {
    "path": {
      "default": "",
      "title": "Path",
      "type": "string"
    },
    "target_wwn": {
      "title": "Target Wwn",
      "type": "string"
    },
    "tpg": {
      "default": 1,
      "title": "Tpg",
      "type": "integer"
    }
  },
  "required": [
    "target_wwn"
  ],
  "title": "IscsiLUN",
  "type": "object"
}

Fields:

  • path (str)
  • target_wwn (str)
  • tpg (int)

Validators:

  • _set_path
Source code in sts_libs/src/sts/target.py 
354
355
356
357
358
359
360
361
362
363
364
365
366
class IscsiLUN(LUN):
    """LUN operations scoped to an iSCSI target/TPG."""

    path: str = ''
    target_wwn: str
    tpg: int = 1

    @model_validator(mode='after')
    def _set_path(self) -> Self:
        """Derive path from target_wwn and tpg."""
        if not self.path:
            self.path = f'/iscsi/{self.target_wwn}/tpg{self.tpg}/luns/'
        return self

LUN pydantic-model

Bases: Targetcli

LUN operations.

Show JSON schema: 
{
  "additionalProperties": false,
  "description": "LUN operations.",
  "properties": {
    "path": {
      "title": "Path",
      "type": "string"
    }
  },
  "required": [
    "path"
  ],
  "title": "LUN",
  "type": "object"
}

Fields:

  • path (str)
Source code in sts_libs/src/sts/target.py 
342
343
344
345
346
347
348
349
350
351
class LUN(Targetcli):
    """LUN operations."""

    def create_lun(self, storage_object: str) -> CommandResult:
        """Create a LUN backed by the given storage object."""
        return self.create(storage_object)

    def delete_lun(self, lun_number: int) -> CommandResult:
        """Delete a LUN by number."""
        return self.delete(str(lun_number))

create_lun(storage_object)

Create a LUN backed by the given storage object.

Source code in sts_libs/src/sts/target.py 
345
346
347
def create_lun(self, storage_object: str) -> CommandResult:
    """Create a LUN backed by the given storage object."""
    return self.create(storage_object)

delete_lun(lun_number)

Delete a LUN by number.

Source code in sts_libs/src/sts/target.py 
349
350
351
def delete_lun(self, lun_number: int) -> CommandResult:
    """Delete a LUN by number."""
    return self.delete(str(lun_number))

Loopback pydantic-model

Bases: Targetcli

Loopback target device in targetcli.

Attributes:

Name Type Description
target_wwn str | None

WWN of the target (auto-generated on create if None)
Show JSON schema: 
{
  "additionalProperties": false,
  "description": "Loopback target device in targetcli.\n\nAttributes:\n    target_wwn: WWN of the target (auto-generated on create if None)",
  "properties": {
    "path": {
      "default": "",
      "title": "Path",
      "type": "string"
    },
    "target_wwn": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Target Wwn"
    },
    "target_path": {
      "default": "",
      "title": "Target Path",
      "type": "string"
    }
  },
  "title": "Loopback",
  "type": "object"
}

Fields:

  • path (str)
  • target_wwn (str | None)
  • target_path (str)

Validators:

  • _set_paths
Source code in sts_libs/src/sts/target.py 
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
class Loopback(Targetcli):
    """Loopback target device in targetcli.

    Attributes:
        target_wwn: WWN of the target (auto-generated on create if None)
    """

    path: str = ''
    target_wwn: str | None = None
    target_path: str = ''

    loopback_path: ClassVar[str] = '/loopback/'

    @model_validator(mode='after')
    def _set_paths(self) -> Self:
        """Derive target_path and path from target_wwn."""
        if not self.target_path:
            self.target_path = f'{self.loopback_path}{self.target_wwn}/' if self.target_wwn else self.loopback_path
        if not self.path:
            self.path = self.target_path
        return self

    def create_target(self) -> CommandResult:
        """Create a new loopback target.

        If target_wwn is None, a WWN is auto-generated and stored back
        into target_wwn from the command output.
        """
        with self.temporary_path(self.loopback_path):
            if not self.target_wwn:
                ret = self.create()
                if ret.succeeded:
                    match = re.search(r'naa\.\w+', ret.stdout)
                    if match:
                        self.target_wwn = match.group(0)
                        self.target_path = f'{self.loopback_path}{self.target_wwn}/'
                return ret
            return self.create(wwn=self.target_wwn)

    def delete_target(self) -> CommandResult:
        """Delete the loopback target.

        Raises:
            ValueError: If target_wwn is None.
        """
        if not self.target_wwn:
            raise ValueError('Cannot delete target: target_wwn is not set')

        with self.temporary_path(self.loopback_path):
            return self.delete(wwn=self.target_wwn)

create_target()

Create a new loopback target.

If target_wwn is None, a WWN is auto-generated and stored back into target_wwn from the command output.

Source code in sts_libs/src/sts/target.py 
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
def create_target(self) -> CommandResult:
    """Create a new loopback target.

    If target_wwn is None, a WWN is auto-generated and stored back
    into target_wwn from the command output.
    """
    with self.temporary_path(self.loopback_path):
        if not self.target_wwn:
            ret = self.create()
            if ret.succeeded:
                match = re.search(r'naa\.\w+', ret.stdout)
                if match:
                    self.target_wwn = match.group(0)
                    self.target_path = f'{self.loopback_path}{self.target_wwn}/'
            return ret
        return self.create(wwn=self.target_wwn)

delete_target()

Delete the loopback target.

Raises:

Type Description
ValueError

If target_wwn is None.
Source code in sts_libs/src/sts/target.py 
408
409
410
411
412
413
414
415
416
417
418
def delete_target(self) -> CommandResult:
    """Delete the loopback target.

    Raises:
        ValueError: If target_wwn is None.
    """
    if not self.target_wwn:
        raise ValueError('Cannot delete target: target_wwn is not set')

    with self.temporary_path(self.loopback_path):
        return self.delete(wwn=self.target_wwn)

LoopbackLUN pydantic-model

Bases: LUN

LUN operations scoped to a loopback target.

Show JSON schema: 
{
  "additionalProperties": false,
  "description": "LUN operations scoped to a loopback target.",
  "properties": {
    "path": {
      "default": "",
      "title": "Path",
      "type": "string"
    },
    "target_wwn": {
      "title": "Target Wwn",
      "type": "string"
    }
  },
  "required": [
    "target_wwn"
  ],
  "title": "LoopbackLUN",
  "type": "object"
}

Fields:

  • path (str)
  • target_wwn (str)

Validators:

  • _set_path
Source code in sts_libs/src/sts/target.py 
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
class LoopbackLUN(LUN):
    """LUN operations scoped to a loopback target."""

    path: str = ''
    target_wwn: str

    @model_validator(mode='after')
    def _set_path(self) -> Self:
        """Validate target_wwn and derive path from it.

        Raises:
            ValueError: If target_wwn is empty or invalid.
        """
        if not self.target_wwn:
            raise ValueError('target_wwn cannot be empty')
        if not self.path:
            self.path = f'/loopback/{self.target_wwn}/luns/'
        return self

Portal pydantic-model

Bases: Targetcli

iSCSI portal (network endpoint) operations.

Show JSON schema: 
{
  "additionalProperties": false,
  "description": "iSCSI portal (network endpoint) operations.",
  "properties": {
    "path": {
      "default": "",
      "title": "Path",
      "type": "string"
    },
    "target_wwn": {
      "title": "Target Wwn",
      "type": "string"
    },
    "portal": {
      "title": "Portal",
      "type": "string"
    },
    "tpg": {
      "default": 1,
      "title": "Tpg",
      "type": "integer"
    },
    "ip_port": {
      "default": 3260,
      "title": "Ip Port",
      "type": "integer"
    },
    "portals_path": {
      "default": "",
      "title": "Portals Path",
      "type": "string"
    },
    "portal_path": {
      "default": "",
      "title": "Portal Path",
      "type": "string"
    }
  },
  "required": [
    "target_wwn",
    "portal"
  ],
  "title": "Portal",
  "type": "object"
}

Fields:

  • path (str)
  • target_wwn (str)
  • portal (str)
  • tpg (int)
  • ip_port (int)
  • portals_path (str)
  • portal_path (str)

Validators:

  • _set_paths
Source code in sts_libs/src/sts/target.py 
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
class Portal(Targetcli):
    """iSCSI portal (network endpoint) operations."""

    path: str = ''
    target_wwn: str
    portal: str
    tpg: int = 1
    ip_port: int = 3260
    portals_path: str = ''
    portal_path: str = ''

    @model_validator(mode='after')
    def _set_paths(self) -> Self:
        """Derive portals_path, portal_path and path from target_wwn, tpg, portal and ip_port."""
        if not self.portals_path:
            self.portals_path = f'/iscsi/{self.target_wwn}/tpg{self.tpg}/portals/'
        if not self.portal_path:
            self.portal_path = f'{self.portals_path}{self.portal}:{self.ip_port}'
        if not self.path:
            self.path = self.portal_path
        return self

    def create_portal(self) -> CommandResult:
        """Create a portal."""
        with self.temporary_path(self.portals_path):
            return self.create(ip_address=self.portal, ip_port=str(self.ip_port))

    def delete_portal(self) -> CommandResult:
        """Delete the portal."""
        with self.temporary_path(self.portals_path):
            return self.delete(ip_address=self.portal, ip_port=str(self.ip_port))

    def enable_offload(self) -> CommandResult:
        """Enable offload for the portal."""
        return self._run('enable_offload=True')

    def disable_offload(self) -> CommandResult:
        """Disable offload for the portal."""
        return self._run('enable_offload=False')

create_portal()

Create a portal.

Source code in sts_libs/src/sts/target.py 
524
525
526
527
def create_portal(self) -> CommandResult:
    """Create a portal."""
    with self.temporary_path(self.portals_path):
        return self.create(ip_address=self.portal, ip_port=str(self.ip_port))

delete_portal()

Delete the portal.

Source code in sts_libs/src/sts/target.py 
529
530
531
532
def delete_portal(self) -> CommandResult:
    """Delete the portal."""
    with self.temporary_path(self.portals_path):
        return self.delete(ip_address=self.portal, ip_port=str(self.ip_port))

disable_offload()

Disable offload for the portal.

Source code in sts_libs/src/sts/target.py 
538
539
540
def disable_offload(self) -> CommandResult:
    """Disable offload for the portal."""
    return self._run('enable_offload=False')

enable_offload()

Enable offload for the portal.

Source code in sts_libs/src/sts/target.py 
534
535
536
def enable_offload(self) -> CommandResult:
    """Enable offload for the portal."""
    return self._run('enable_offload=True')

TPG pydantic-model

Bases: Targetcli

Target Portal Group operations.

Show JSON schema: 
{
  "additionalProperties": false,
  "description": "Target Portal Group operations.",
  "properties": {
    "path": {
      "default": "",
      "title": "Path",
      "type": "string"
    },
    "target_wwn": {
      "title": "Target Wwn",
      "type": "string"
    },
    "tpg": {
      "default": 1,
      "title": "Tpg",
      "type": "integer"
    },
    "target_path": {
      "default": "",
      "title": "Target Path",
      "type": "string"
    },
    "tpg_path": {
      "default": "",
      "title": "Tpg Path",
      "type": "string"
    }
  },
  "required": [
    "target_wwn"
  ],
  "title": "TPG",
  "type": "object"
}

Fields:

  • path (str)
  • target_wwn (str)
  • tpg (int)
  • target_path (str)
  • tpg_path (str)

Validators:

  • _set_paths
Source code in sts_libs/src/sts/target.py 
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
class TPG(Targetcli):
    """Target Portal Group operations."""

    path: str = ''
    target_wwn: str
    tpg: int = 1
    target_path: str = ''
    tpg_path: str = ''

    @model_validator(mode='after')
    def _set_paths(self) -> Self:
        """Derive target_path, tpg_path and path from target_wwn and tpg."""
        if not self.target_path:
            self.target_path = f'/iscsi/{self.target_wwn}/'
        if not self.tpg_path:
            self.tpg_path = f'{self.target_path}tpg{self.tpg}/'
        if not self.path:
            self.path = self.tpg_path
        return self

    def create_tpg(self) -> CommandResult:
        """Create a Target Portal Group."""
        with self.temporary_path(self.target_path):
            return self.create(tag=str(self.tpg))

    def delete_tpg(self) -> CommandResult:
        """Delete the Target Portal Group."""
        with self.temporary_path(self.target_path):
            return self.delete(tag=str(self.tpg))

    def enable_tpg(self) -> CommandResult:
        """Enable the Target Portal Group."""
        return self._run('enable')

    def disable_tpg(self) -> CommandResult:
        """Disable the Target Portal Group."""
        return self._run('disable')

    def set_auth(
        self,
        userid: str | None = None,
        password: str | None = None,
        mutual_userid: str | None = None,
        mutual_password: str | None = None,
    ) -> CommandResult:
        """Set authentication for the TPG (None values become empty strings)."""
        self.set_('auth', userid='' if userid is None else userid)
        self.set_('auth', password='' if password is None else password)
        self.set_('auth', mutual_userid='' if mutual_userid is None else mutual_userid)
        self.set_('auth', mutual_password='' if mutual_password is None else mutual_password)
        return self.set_('attribute', authentication='1', generate_node_acls='1')

    def disable_auth_per_tpg(self) -> CommandResult:
        """Disable authentication for the Target Portal Group."""
        return self.set_('attribute', authentication='0')

    def disable_generate_node_acls(self) -> CommandResult:
        """Disable generate_node_acls for the Target Portal Group."""
        return self.set_('attribute', generate_node_acls='0')

create_tpg()

Create a Target Portal Group.

Source code in sts_libs/src/sts/target.py 
301
302
303
304
def create_tpg(self) -> CommandResult:
    """Create a Target Portal Group."""
    with self.temporary_path(self.target_path):
        return self.create(tag=str(self.tpg))

delete_tpg()

Delete the Target Portal Group.

Source code in sts_libs/src/sts/target.py 
306
307
308
309
def delete_tpg(self) -> CommandResult:
    """Delete the Target Portal Group."""
    with self.temporary_path(self.target_path):
        return self.delete(tag=str(self.tpg))

disable_auth_per_tpg()

Disable authentication for the Target Portal Group.

Source code in sts_libs/src/sts/target.py 
333
334
335
def disable_auth_per_tpg(self) -> CommandResult:
    """Disable authentication for the Target Portal Group."""
    return self.set_('attribute', authentication='0')

disable_generate_node_acls()

Disable generate_node_acls for the Target Portal Group.

Source code in sts_libs/src/sts/target.py 
337
338
339
def disable_generate_node_acls(self) -> CommandResult:
    """Disable generate_node_acls for the Target Portal Group."""
    return self.set_('attribute', generate_node_acls='0')

disable_tpg()

Disable the Target Portal Group.

Source code in sts_libs/src/sts/target.py 
315
316
317
def disable_tpg(self) -> CommandResult:
    """Disable the Target Portal Group."""
    return self._run('disable')

enable_tpg()

Enable the Target Portal Group.

Source code in sts_libs/src/sts/target.py 
311
312
313
def enable_tpg(self) -> CommandResult:
    """Enable the Target Portal Group."""
    return self._run('enable')

set_auth(userid=None, password=None, mutual_userid=None, mutual_password=None)

Set authentication for the TPG (None values become empty strings).

Source code in sts_libs/src/sts/target.py 
319
320
321
322
323
324
325
326
327
328
329
330
331
def set_auth(
    self,
    userid: str | None = None,
    password: str | None = None,
    mutual_userid: str | None = None,
    mutual_password: str | None = None,
) -> CommandResult:
    """Set authentication for the TPG (None values become empty strings)."""
    self.set_('auth', userid='' if userid is None else userid)
    self.set_('auth', password='' if password is None else password)
    self.set_('auth', mutual_userid='' if mutual_userid is None else mutual_userid)
    self.set_('auth', mutual_password='' if mutual_password is None else mutual_password)
    return self.set_('attribute', authentication='1', generate_node_acls='1')

Targetcli pydantic-model

Bases: StsBaseModel

Use to run targetcli commands.

rtslib-fb API would normally be used in Python, however we want to test targetcli commands.

Attributes:

Name Type Description
path str

The path within targetcli shell structure
Show JSON schema: 
{
  "additionalProperties": false,
  "description": "Use to run targetcli commands.\n\nrtslib-fb API would normally be used in Python, however we want to test targetcli commands.\n\nAttributes:\n    path: The path within targetcli shell structure",
  "properties": {
    "path": {
      "title": "Path",
      "type": "string"
    }
  },
  "required": [
    "path"
  ],
  "title": "Targetcli",
  "type": "object"
}

Config:

  • extra: forbid

Fields:

  • path (str)
Source code in sts_libs/src/sts/target.py 
 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
class Targetcli(StsBaseModel):
    """Use to run targetcli commands.

    rtslib-fb API would normally be used in Python, however we want to test targetcli commands.

    Attributes:
        path: The path within targetcli shell structure
    """

    TARGETCLI: ClassVar[str] = TARGETCLI

    path: str

    def _run(self, *args: str, **kwargs: str) -> CommandResult:
        """Execute a targetcli command at the current path."""
        cmd = f'{self.TARGETCLI} {self.path} {" ".join(args)}'
        arguments = {**kwargs}
        if arguments:
            arguments_unpacked = ' '.join([f'{key}={value}' for key, value in arguments.items()])
            cmd = f'{cmd} {arguments_unpacked}'
        return run(cmd)

    def set_(self, *args: str, **kwargs: str) -> CommandResult:
        """Set parameters or attributes."""
        return self._run('set', *args, **kwargs)

    def set_parameter(self, parameter: str, value: str) -> CommandResult:
        """Set a specific parameter."""
        return self.set_('parameter', **{parameter: value})

    def set_attribute(self, attribute: str, value: str) -> CommandResult:
        """Set a specific attribute."""
        return self.set_('attribute', **{attribute: value})

    def set_attributes(self, **kwargs: str) -> CommandResult:
        """Set multiple attributes at once."""
        return self.set_('attribute', **kwargs)

    def get(self, *args: str, **kwargs: str) -> CommandResult:
        """Get parameters or attributes."""
        return self._run('get', *args, **kwargs)

    def get_parameter(self, parameter: str) -> CommandResult:
        """Get a specific parameter."""
        return self.get('parameter', parameter)

    def get_attribute(self, parameter: str) -> CommandResult:
        """Get a specific attribute."""
        return self.get('attribute', parameter)

    def get_attributes(self) -> dict[str, str]:
        """Get all attributes as a dictionary."""
        output = self.get('attribute').stdout.removeprefix('ATTRIBUTE CONFIG GROUP\n======================\n')
        return dict(_.split('=', 1) for _ in output.splitlines() if '=' in _)

    def create(self, *args: str, **kwargs: str) -> CommandResult:
        """Create an object."""
        return self._run('create', *args, **kwargs)

    def delete(self, *args: str, **kwargs: str) -> CommandResult:
        """Delete an object."""
        return self._run('delete', *args, **kwargs)

    def ls(self) -> CommandResult:
        """List contents."""
        return self._run('ls')

    def get_path(self) -> str:
        """Get the current path."""
        return self.path

    def clearconfig(self) -> CommandResult:
        """Clear the target configuration."""
        return self._run('clearconfig confirm=True')

    @contextmanager
    def temporary_path(self, temp_path: str) -> Generator[None, None, None]:
        """Temporarily change the path for command execution.

        Args:
            temp_path: The temporary path to use
        """
        old_path = self.path
        self.path = temp_path
        try:
            yield
        finally:
            self.path = old_path

clearconfig()

Clear the target configuration.

Source code in sts_libs/src/sts/target.py 
106
107
108
def clearconfig(self) -> CommandResult:
    """Clear the target configuration."""
    return self._run('clearconfig confirm=True')

create(*args, **kwargs)

Create an object.

Source code in sts_libs/src/sts/target.py 
90
91
92
def create(self, *args: str, **kwargs: str) -> CommandResult:
    """Create an object."""
    return self._run('create', *args, **kwargs)

delete(*args, **kwargs)

Delete an object.

Source code in sts_libs/src/sts/target.py 
94
95
96
def delete(self, *args: str, **kwargs: str) -> CommandResult:
    """Delete an object."""
    return self._run('delete', *args, **kwargs)

get(*args, **kwargs)

Get parameters or attributes.

Source code in sts_libs/src/sts/target.py 
73
74
75
def get(self, *args: str, **kwargs: str) -> CommandResult:
    """Get parameters or attributes."""
    return self._run('get', *args, **kwargs)

get_attribute(parameter)

Get a specific attribute.

Source code in sts_libs/src/sts/target.py 
81
82
83
def get_attribute(self, parameter: str) -> CommandResult:
    """Get a specific attribute."""
    return self.get('attribute', parameter)

get_attributes()

Get all attributes as a dictionary.

Source code in sts_libs/src/sts/target.py 
85
86
87
88
def get_attributes(self) -> dict[str, str]:
    """Get all attributes as a dictionary."""
    output = self.get('attribute').stdout.removeprefix('ATTRIBUTE CONFIG GROUP\n======================\n')
    return dict(_.split('=', 1) for _ in output.splitlines() if '=' in _)

get_parameter(parameter)

Get a specific parameter.

Source code in sts_libs/src/sts/target.py 
77
78
79
def get_parameter(self, parameter: str) -> CommandResult:
    """Get a specific parameter."""
    return self.get('parameter', parameter)

get_path()

Get the current path.

Source code in sts_libs/src/sts/target.py 
102
103
104
def get_path(self) -> str:
    """Get the current path."""
    return self.path

ls()

List contents.

Source code in sts_libs/src/sts/target.py 
 98
 99
100
def ls(self) -> CommandResult:
    """List contents."""
    return self._run('ls')

set_(*args, **kwargs)

Set parameters or attributes.

Source code in sts_libs/src/sts/target.py 
57
58
59
def set_(self, *args: str, **kwargs: str) -> CommandResult:
    """Set parameters or attributes."""
    return self._run('set', *args, **kwargs)

set_attribute(attribute, value)

Set a specific attribute.

Source code in sts_libs/src/sts/target.py 
65
66
67
def set_attribute(self, attribute: str, value: str) -> CommandResult:
    """Set a specific attribute."""
    return self.set_('attribute', **{attribute: value})

set_attributes(**kwargs)

Set multiple attributes at once.

Source code in sts_libs/src/sts/target.py 
69
70
71
def set_attributes(self, **kwargs: str) -> CommandResult:
    """Set multiple attributes at once."""
    return self.set_('attribute', **kwargs)

set_parameter(parameter, value)

Set a specific parameter.

Source code in sts_libs/src/sts/target.py 
61
62
63
def set_parameter(self, parameter: str, value: str) -> CommandResult:
    """Set a specific parameter."""
    return self.set_('parameter', **{parameter: value})

temporary_path(temp_path)

Temporarily change the path for command execution.

Parameters:

Name Type Description Default
temp_path str

The temporary path to use

 required
Source code in sts_libs/src/sts/target.py 
110
111
112
113
114
115
116
117
118
119
120
121
122
@contextmanager
def temporary_path(self, temp_path: str) -> Generator[None, None, None]:
    """Temporarily change the path for command execution.

    Args:
        temp_path: The temporary path to use
    """
    old_path = self.path
    self.path = temp_path
    try:
        yield
    finally:
        self.path = old_path

cleanup_loopback_devices(devices)

Clean up loopback devices (LUNs, backstores, and target).

Environment Variables

TARGET_WWN: World Wide Name for the target (default: 'naa.50014054c1441891') LUN_PREFIX: Prefix for LUN names (default: 'common-lun-')

Raises:

Type Description
ValueError

If devices list is empty
Source code in sts_libs/src/sts/target.py 
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
def cleanup_loopback_devices(devices: list[BlockDevice]) -> None:
    """Clean up loopback devices (LUNs, backstores, and target).

    Environment Variables:
        TARGET_WWN: World Wide Name for the target (default: 'naa.50014054c1441891')
        LUN_PREFIX: Prefix for LUN names (default: 'common-lun-')

    Raises:
        ValueError: If devices list is empty
    """
    if not devices:
        raise ValueError('No devices provided for cleanup')

    # Get configuration from environment variables
    wwn = getenv('TARGET_WWN', 'naa.50014054c1441891')
    lun_prefix = getenv('LUN_PREFIX', 'common-lun-')

    # Initialize loopback target and LUN
    loopback = Loopback(target_wwn=wwn)
    lun = LoopbackLUN(target_wwn=wwn)

    # Clean up each device
    for n in range(len(devices)):
        backstore = BackstoreFileio(name=f'{lun_prefix}{n}')

        # Delete LUN
        result = lun.delete_lun(n)
        assert result.succeeded, f'Failed to delete LUN {n}: {result.stderr}'

        # Delete backstore
        result = backstore.delete_backstore()
        assert result.succeeded, f'Failed to delete backstore {n}: {result.stderr}'

    # Delete target
    result = loopback.delete_target()
    assert result.succeeded, f'Failed to delete target: {result.stderr}'

create_basic_iscsi_target(target_wwn='', initiator_wwn='', size='1G', userid=None, password=None, mutual_userid=None, mutual_password=None)

Create simple iSCSI target using fileio backstore.

Parameters:

Name Type Description Default
target_wwn str

Target WWN (auto-generated if empty)

 ''
initiator_wwn str

Initiator WWN (read from /etc/iscsi/initiatorname.iscsi if empty)

 ''
size str

Size of the fileio backstore

 '1G'
userid str | None

CHAP user ID (None to disable auth)

 None
password str | None

CHAP password

 None
mutual_userid str | None

Mutual CHAP user ID

 None
mutual_password str | None

Mutual CHAP password

 None
Source code in sts_libs/src/sts/target.py 
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
def create_basic_iscsi_target(
    target_wwn: str = '',
    initiator_wwn: str = '',
    size: str = '1G',
    userid: str | None = None,
    password: str | None = None,
    mutual_userid: str | None = None,
    mutual_password: str | None = None,
) -> bool:
    """Create simple iSCSI target using fileio backstore.

    Args:
        target_wwn: Target WWN (auto-generated if empty)
        initiator_wwn: Initiator WWN (read from /etc/iscsi/initiatorname.iscsi if empty)
        size: Size of the fileio backstore
        userid: CHAP user ID (None to disable auth)
        password: CHAP password
        mutual_userid: Mutual CHAP user ID
        mutual_password: Mutual CHAP password
    """
    if not target_wwn:
        target_wwn = f'iqn.2023-01.com.sts:target:{uuid4().hex[-9:]}'
    if not initiator_wwn:
        try:
            # Try to set localhost initiatorname
            initiator_wwn = Path('/etc/iscsi/initiatorname.iscsi').read_text().split('=')[1]
        except FileNotFoundError:
            initiator_wwn = f'iqn.1994-05.com.redhat:{uuid4().hex[-9:]}'
        logger.info(f'Initiator iqn: "{initiator_wwn}"')
    backstore_name = initiator_wwn.split(':')[1]

    backstore = BackstoreFileio(name=backstore_name)
    backstore.create_backstore(size=size, file_or_dev=f'{backstore_name}_backstore_file')
    Iscsi(target_wwn=target_wwn).create_target()
    IscsiLUN(target_wwn=target_wwn).create_lun(storage_object=backstore.path)
    acl = ACL(target_wwn=target_wwn, initiator_wwn=initiator_wwn)
    acl.create_acl()
    if userid and password:
        acl.set_auth(
            userid=userid,
            password=password,
            mutual_userid=mutual_userid,
            mutual_password=mutual_password,
        )
    else:
        acl.disable_auth()
    return True

create_loopback_devices(count, block_size=4096)

Create loopback devices backed by fileio backstores.

Environment Variables

TARGET_WWN: World Wide Name for the target (default: 'naa.50014054c1441891') LOOPBACK_DEVICE_SIZE: Size of each device (default: '2G') LUN_PREFIX: Prefix for LUN names (default: 'common-lun-') IMAGE_PATH: Path where backing files are created (default: '/var/tmp/')

Raises:

Type Description
ValueError

If count is less than 1
Source code in sts_libs/src/sts/target.py 
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
def create_loopback_devices(count: int, block_size: int = 4096) -> list[BlockDevice]:
    """Create loopback devices backed by fileio backstores.

    Environment Variables:
        TARGET_WWN: World Wide Name for the target (default: 'naa.50014054c1441891')
        LOOPBACK_DEVICE_SIZE: Size of each device (default: '2G')
        LUN_PREFIX: Prefix for LUN names (default: 'common-lun-')
        IMAGE_PATH: Path where backing files are created (default: '/var/tmp/')

    Raises:
        ValueError: If count is less than 1
    """
    if count < 1:
        raise ValueError('Device count must be at least 1')

    # Get configuration from environment variables
    wwn = getenv('TARGET_WWN', 'naa.50014054c1441891')
    device_size = getenv('LOOPBACK_DEVICE_SIZE', '2G')
    lun_prefix = getenv('LUN_PREFIX', 'common-lun-')
    image_path = getenv('IMAGE_PATH', '/var/tmp/')
    suffix = '.image'

    # Initialize loopback target and LUN
    loopback = Loopback(target_wwn=wwn)
    lun = LoopbackLUN(target_wwn=wwn)

    # Create target
    result = loopback.create_target()
    assert result.succeeded, f'Failed to create target: {result.stderr}'

    # Create each device
    for n in range(count):
        backstore = BackstoreFileio(name=f'{lun_prefix}{n}')

        # Create backstore
        result = backstore.create_backstore(size=device_size, file_or_dev=f'{image_path}{lun_prefix}{n}{suffix}')
        assert result.succeeded, f'Failed to create backstore {n}: {result.stderr}'

        # Set block size
        result = backstore.set_attribute(attribute='block_size', value=str(block_size))
        assert result.succeeded, f'Failed to set block size for device {n}: {result.stderr}'

        # Create LUN
        result = lun.create(storage_object=backstore.path)
        assert result.succeeded, f'Failed to create LUN {n}: {result.stderr}'

    return [device for device in get_free_disks() if device.model and lun_prefix in device.model]