Test Fixtures

This section documents the test fixtures provided by sts-libs. These fixtures help set up test environments for various storage technologies.

Common Fixtures

sts.fixtures.common_fixtures

Common test fixtures.

This module provides fixtures that can be used across different test suites: - Virtual block devices (loop, scsi_debug) - System resources - Common utilities

Fixture Dependencies: 1. loop_devices - Independent fixture - Creates temporary loop devices - Handles cleanup automatically

1. scsi_debug_devices
2. Independent fixture
3. Creates SCSI debug devices
4. Handles cleanup automatically

Common Usage:

1. Basic device testing:

   def test_single_device(loop_devices):
       device = loop_devices[0]
       # Test with single device
   

2. Multi-device testing:

   @pytest.mark.parametrize('loop_devices', [2], indirect=True)
   def test_multiple_devices(loop_devices):
       dev1, dev2 = loop_devices
       # Test with multiple devices
   

3. SCSI debug testing:

   @pytest.mark.parametrize('scsi_debug_devices', [2], indirect=True)
   def test_scsi_devices(scsi_debug_devices):
       dev1, dev2 = scsi_debug_devices
       # Test with SCSI debug devices
   

Error Handling: - Device creation failures skip the test - Cleanup runs even if test fails - Resource limits are checked

ensure_minimum_devices()

Fixture that ensures minimum number of devices without block size filtering.

Source code in sts_libs/src/sts/fixtures/common_fixtures.py

@pytest.fixture
def ensure_minimum_devices() -> Generator:
    """Fixture that ensures minimum number of devices without block size filtering."""
    yield from _ensure_minimum_devices_base(filter_by_block_size=False)

ensure_minimum_devices_with_same_block_sizes()

Fixture that ensures minimum number of devices with same block sizes.

Source code in sts_libs/src/sts/fixtures/common_fixtures.py

@pytest.fixture
def ensure_minimum_devices_with_same_block_sizes() -> Generator:
    """Fixture that ensures minimum number of devices with same block sizes."""
    yield from _ensure_minimum_devices_base(filter_by_block_size=True)

loop_devices(request)

Create loop devices for testing.

Creates virtual block devices using the loop driver: - Each device is 1GB in size - Devices are sparse (only allocate used space) - Devices are automatically cleaned up - Supports multiple devicesce(loop_devices): assert len(loop_devices) == 1 assert loop_d per test

Configuration: - count: Number of devices to create (default: 1) Set via parametrize: @pytest.mark.parametrize('loop_devices', [2])

Error Handling: - Skips test if device creation fails - Cleans up any created devices on failure - Validates device paths before yielding

Parameters:

Name	Type	Description	Default
request	FixtureRequest	Pytest fixture request with 'count' parameter	required

Yields:

Type	Description
list[str]	List of loop device paths (e.g. ['/dev/loop0', '/dev/loop1'])

Example

Single device

def test_device(loop_devices):
    assert len(loop_devices) == 1
    assert loop_devices[0].startswith('/dev/loop')

Multiple devices

@pytest.mark.parametrize('loop_devices', [2], indirect=True)
    def test_devices(loop_devices):
        assert len(loop_devices) == 2
        assert all(d.startswith('/dev/loop') for d in loop_devices)

Source code in sts_libs/src/sts/fixtures/common_fixtures.py

@pytest.fixture
def loop_devices(request: pytest.FixtureRequest) -> Generator[list[str], None, None]:
    """Create loop devices for testing.

    Creates virtual block devices using the loop driver:
    - Each device is 1GB in size
    - Devices are sparse (only allocate used space)
    - Devices are automatically cleaned up
    - Supports multiple devicesce(loop_devices): assert len(loop_devices) == 1 assert loop_d per test

    Configuration:
    - count: Number of devices to create (default: 1)
      Set via parametrize: @pytest.mark.parametrize('loop_devices', [2])

    Error Handling:
    - Skips test if device creation fails
    - Cleans up any created devices on failure
    - Validates device paths before yielding

    Args:
        request: Pytest fixture request with 'count' parameter

    Yields:
        List of loop device paths (e.g. ['/dev/loop0', '/dev/loop1'])

    Example:
        # Single device
        ```python
        def test_device(loop_devices):
            assert len(loop_devices) == 1
            assert loop_devices[0].startswith('/dev/loop')
        ```
        # Multiple devices
        ```
        @pytest.mark.parametrize('loop_devices', [2], indirect=True)
            def test_devices(loop_devices):
                assert len(loop_devices) == 2
                assert all(d.startswith('/dev/loop') for d in loop_devices)
        ```
    """
    count = getattr(request, 'param', 1)  # Default to 1 device if not specified
    devices = []

    # Create devices one by one
    for _ in range(count):
        device = LoopDevice.create(size_mb=1024)
        if not device:
            # Clean up any created devices
            for dev in devices:
                dev.remove()
            pytest.skip(f'Failed to create loop device {len(devices) + 1} of {count}')
        devices.append(device)

    # Yield device paths
    yield [str(dev.device_path) for dev in devices]

    # Clean up
    for device in devices:
        device.remove()

multipath_setup()

Fixture to set up and tear down multipath devices for testing.

This fixture performs the following steps: 1. Checks if the multipath service is running. 2. If not running, starts the multipath service. 3. Retrieves all multipath devices. 4. Yields the list of multipath devices for use in tests. 5. Stops the multipath service if it was started by this fixture.

Yields:

Type	Description
Sequence[DmDevice]	list[MultipathDevice]: A list of multipath devices available for testing.

Source code in sts_libs/src/sts/fixtures/common_fixtures.py

@pytest.fixture(scope='class')
def multipath_setup() -> Generator[Sequence[DmDevice], None, None]:
    """Fixture to set up and tear down multipath devices for testing.

    This fixture performs the following steps:
    1. Checks if the multipath service is running.
    2. If not running, starts the multipath service.
    3. Retrieves all multipath devices.
    4. Yields the list of multipath devices for use in tests.
    5. Stops the multipath service if it was started by this fixture.

    Yields:
        list[MultipathDevice]: A list of multipath devices available for testing.
    """
    mp_service = MultipathService()
    if not mp_service.is_running():
        mp_service.start()
        devices = MultipathDevice.get_all()
        yield devices
        mp_service.stop()
    else:
        yield MultipathDevice.get_all()

scsi_debug_devices(request)

Create SCSI debug devices for testing.

Creates virtual SCSI devices using the scsi_debug module: - Each device is 1GB in size - Devices share a single scsi_debug instance - Devices are automatically cleaned up - Supports multiple devices per test

Configuration: - count: Number of devices to create (default: 1) Set via parametrize: @pytest.mark.parametrize('scsi_debug_devices', [2])

Error Handling: - Skips test if module loading fails - Skips test if device creation fails - Cleans up module and devices on failure - Validates device count before yielding

Parameters:

Name	Type	Description	Default
request	FixtureRequest	Pytest fixture request with 'count' parameter	required

Yields:

Type	Description
list[str]	List of SCSI device paths (e.g. ['/dev/sda', '/dev/sdb'])

Example

# Single device
def test_device(scsi_debug_devices):
    assert len(scsi_debug_devices) == 1
    assert scsi_debug_devices[0].startswith('/dev/sd')


# Multiple devices
@pytest.mark.parametrize('scsi_debug_devices', [2], indirect=True)
def test_devices(scsi_debug_devices):
    assert len(scsi_debug_devices) == 2
    assert all(d.startswith('/dev/sd') for d in scsi_debug_devices)

Source code in sts_libs/src/sts/fixtures/common_fixtures.py

@pytest.fixture
def scsi_debug_devices(request: pytest.FixtureRequest) -> Generator[list[str], None, None]:
    """Create SCSI debug devices for testing.

    Creates virtual SCSI devices using the scsi_debug module:
    - Each device is 1GB in size
    - Devices share a single scsi_debug instance
    - Devices are automatically cleaned up
    - Supports multiple devices per test

    Configuration:
    - count: Number of devices to create (default: 1)
      Set via parametrize: @pytest.mark.parametrize('scsi_debug_devices', [2])

    Error Handling:
    - Skips test if module loading fails
    - Skips test if device creation fails
    - Cleans up module and devices on failure
    - Validates device count before yielding

    Args:
        request: Pytest fixture request with 'count' parameter

    Yields:
        List of SCSI device paths (e.g. ['/dev/sda', '/dev/sdb'])

    Example:
        ```python
        # Single device
        def test_device(scsi_debug_devices):
            assert len(scsi_debug_devices) == 1
            assert scsi_debug_devices[0].startswith('/dev/sd')


        # Multiple devices
        @pytest.mark.parametrize('scsi_debug_devices', [2], indirect=True)
        def test_devices(scsi_debug_devices):
            assert len(scsi_debug_devices) == 2
            assert all(d.startswith('/dev/sd') for d in scsi_debug_devices)
        ```
    """
    count = getattr(request, 'param', 1)  # Default to 1 device if not specified
    logging.info(f'Creating {count} scsi_debug devices')

    # Create SCSI debug device with specified number of targets
    device = ScsiDebugDevice.create(
        size=1024 * 1024 * 1024,  # 1GB
        options=f'num_tgts={count} add_host={count}',
    )
    if not device:
        pytest.skip('Failed to create SCSI debug device')

    # Get all SCSI debug devices
    devices = ScsiDebugDevice.get_devices()
    if not devices or len(devices) < count:
        device.remove()
        pytest.skip(f'Expected {count} SCSI debug devices, got {len(devices or [])}')

    # Yield device paths
    yield [f'/dev/{dev}' for dev in devices[:count]]

    # Clean up
    device.remove()

iSCSI Fixtures

sts.fixtures.iscsi_fixtures

iSCSI test fixtures.

This module provides fixtures for testing iSCSI: - Package installation - Service management - Device configuration - Parameter verification - Session management

Fixture Dependencies: 1. _iscsi_test (base fixture) - Installs iSCSI utilities - Manages sessions 2. iscsi_localhost_test (depends on _iscsi_test) - Sets up target environment 3. iscsi_target (depends on iscsi_localhost_test) - Creates target and initiator - Manages connections

IscsiTestConfig dataclass

Configuration for iSCSI test environment.

Attributes:

Name	Type	Description
base_iqn	str	Base IQN for test
target_iqn	str	Target IQN
initiator_iqn	str	Initiator IQN
size	str	Size of LUNs
n_luns	int	Number of LUNs

Source code in sts_libs/src/sts/fixtures/iscsi_fixtures.py

@dataclass
class IscsiTestConfig:
    """Configuration for iSCSI test environment.

    Attributes:
        base_iqn: Base IQN for test
        target_iqn: Target IQN
        initiator_iqn: Initiator IQN
        size: Size of LUNs
        n_luns: Number of LUNs
    """

    base_iqn: str
    target_iqn: str
    initiator_iqn: str
    size: str = '1G'
    n_luns: int = 1

generate_test_iqns(test_name)

Generate IQNs for test environment.

Parameters:

Name	Type	Description	Default
test_name	str	Name of the test	required

Returns:

Type	Description
tuple[str, str, str]	Tuple of (base_iqn, target_iqn, initiator_iqn)

Source code in sts_libs/src/sts/fixtures/iscsi_fixtures.py

def generate_test_iqns(test_name: str) -> tuple[str, str, str]:
    """Generate IQNs for test environment.

    Args:
        test_name: Name of the test

    Returns:
        Tuple of (base_iqn, target_iqn, initiator_iqn)
    """
    test_name = test_name.split('[')[0]  # Remove parametrize part
    # Replace underscores with dashes for IQN compatibility
    test_name = test_name.replace('_', '-')
    base_iqn = f'iqn.2024-01.sts.{test_name}'
    target_iqn = f'{base_iqn}:target'
    initiator_iqn = f'{base_iqn}:initiator'
    return base_iqn, target_iqn, initiator_iqn

get_test_device()

Get test device path.

Returns:

Type	Description
Callable[[], Path]	Function to get device path

Example

def test_device(get_test_device):
    device = get_test_device()
    assert device.exists()

Source code in sts_libs/src/sts/fixtures/iscsi_fixtures.py

@pytest.fixture
def get_test_device() -> Callable[[], Path]:
    """Get test device path.

    Returns:
        Function to get device path

    Example:
        ```python
        def test_device(get_test_device):
            device = get_test_device()
            assert device.exists()
        ```
    """

    def _get_test_device() -> Path:
        """Get test device path.

        Returns:
            Device path

        Raises:
            AssertionError: If no device found
        """
        mp_service = MultipathService()
        if mp_service.is_running():
            devices = MultipathDevice.get_all()
            if devices and devices[0].path:
                return Path(str(devices[0].path))

        devices = ScsiDevice.get_by_vendor('LIO')
        # Break down complex assertion
        assert devices, 'No LIO devices found'
        assert devices[0].path, 'Device path not available'
        return Path(str(devices[0].path))

    return _get_test_device

iscsi_localhost_test(request, _iscsi_test)

Set up iSCSI target environment.

This fixture: - Installs target utilities - Creates target configuration - Cleans up environment

Parameters:

Name	Type	Description	Default
request	FixtureRequest	Fixture request	required
_iscsi_test	None	Parent fixture providing base setup	required

Yields:

Type	Description
str	Target IQN

Example

def test_target(iscsi_localhost_test):
    target_iqn = iscsi_localhost_test
    assert Iscsi(target_iqn).exists()

Source code in sts_libs/src/sts/fixtures/iscsi_fixtures.py

@pytest.fixture(scope='class')
def iscsi_localhost_test(request: FixtureRequest, _iscsi_test: None) -> Generator[str, None, None]:
    """Set up iSCSI target environment.

    This fixture:
    - Installs target utilities
    - Creates target configuration
    - Cleans up environment

    Args:
        request: Fixture request
        _iscsi_test: Parent fixture providing base setup

    Yields:
        Target IQN

    Example:
        ```python
        def test_target(iscsi_localhost_test):
            target_iqn = iscsi_localhost_test
            assert Iscsi(target_iqn).exists()
        ```
    """
    system = SystemManager()
    assert system.package_manager.install('targetcli')

    # Generate IQNs
    _, target_iqn, _ = generate_test_iqns(request.node.name)

    # Clean up target config
    target = Iscsi(target_wwn=target_iqn)
    target.delete_target()

    yield target_iqn

    # Clean up target config
    target.delete_target()

iscsi_target(request, iscsi_localhost_test)

Create iSCSI target and connect initiator.

This fixture: - Creates target with specified size and number of LUNs - Sets up initiator - Logs in to target - Yields connected node - Cleans up on exit

Parameters:

Name	Type	Description	Default
request	FixtureRequest	Fixture request with parameters: - size: Size of each LUN (default: '1G') - n_luns: Number of LUNs (default: 1)	required
iscsi_localhost_test	None	Parent fixture providing target IQN	required

Example

@pytest.mark.parametrize('iscsi_target', [{'size': '2G', 'n_luns': 2}], indirect=True)
def test_something(iscsi_target):
    assert iscsi_target.exists()

Source code in sts_libs/src/sts/fixtures/iscsi_fixtures.py

@pytest.fixture
def iscsi_target(request: FixtureRequest, iscsi_localhost_test: None) -> Generator[IscsiNode, None, None]:  # noqa: ARG001
    """Create iSCSI target and connect initiator.

    This fixture:
    - Creates target with specified size and number of LUNs
    - Sets up initiator
    - Logs in to target
    - Yields connected node
    - Cleans up on exit

    Args:
        request: Fixture request with parameters:
            - size: Size of each LUN (default: '1G')
            - n_luns: Number of LUNs (default: 1)
        iscsi_localhost_test: Parent fixture providing target IQN

    Example:
        ```python
        @pytest.mark.parametrize('iscsi_target', [{'size': '2G', 'n_luns': 2}], indirect=True)
        def test_something(iscsi_target):
            assert iscsi_target.exists()
        ```
    """
    # Generate IQNs
    _, target_iqn, initiator_iqn = generate_test_iqns(request.node.name)

    # Get parameters
    params = request.param if hasattr(request, 'param') else {}
    config = IscsiTestConfig(
        base_iqn=target_iqn.rsplit(':', 1)[0],
        target_iqn=target_iqn,
        initiator_iqn=initiator_iqn,
        size=params.get('size', '1G'),
        n_luns=params.get('n_luns', 1),
    )

    # Set initiator name
    assert set_initiatorname(config.initiator_iqn), 'Failed to set initiator name'

    # Create target
    assert create_basic_iscsi_target(
        target_wwn=config.target_iqn,
        initiator_wwn=config.initiator_iqn,
        size=config.size,
    ), 'Failed to create target'

    # Create additional LUNs if needed
    if config.n_luns > 1:
        test_name = request.node.name.split('[')[0]
        for i in range(1, config.n_luns):
            backstore_name = f'{test_name}_lun{i}'
            backstore = BackstoreFileio(name=backstore_name)
            backstore.create_backstore(size=config.size, file_or_dev=f'{backstore_name}_file')
            IscsiLUN(target_wwn=config.target_iqn).create_lun(storage_object=backstore.path)

    # Set up initiator and login
    node = IscsiNode.setup_and_login(
        portal='127.0.0.1:3260',
        initiator_iqn=config.initiator_iqn,
        target_iqn=config.target_iqn,
    )

    with manage_iscsi_session(node):
        yield node

manage_iscsi_session(node)

Context manager for iSCSI session management.

Parameters:

Name	Type	Description	Default
node	IscsiNode	IscsiNode instance to manage	required

Yields:

Type	Description
None	None

Source code in sts_libs/src/sts/fixtures/iscsi_fixtures.py

@contextmanager
def manage_iscsi_session(node: IscsiNode) -> Generator[None, None, None]:
    """Context manager for iSCSI session management.

    Args:
        node: IscsiNode instance to manage

    Yields:
        None
    """
    try:
        yield
    finally:
        node.logout()

LVM Fixtures

sts.fixtures.lvm_fixtures

LVM test fixtures.

This module provides fixtures for testing LVM (Logical Volume Management): - Package installation and cleanup - Service management - Device configuration - VDO (Virtual Data Optimizer) support

Fixture Dependencies: 1. _lvm_test (base fixture) - Installs LVM packages - Manages volume cleanup - Logs system information

1. _vdo_test (depends on _lvm_test)
2. Installs VDO packages
3. Manages kernel module
4. Provides data reduction features

Common Usage: 1. Basic LVM testing: @pytest.mark.usefixtures('_lvm_test') def test_lvm(): # LVM utilities are installed # Volumes are cleaned up after test

1. VDO-enabled testing: @pytest.mark.usefixtures('_vdo_test') def test_vdo(): # VDO module is loaded # Data reduction is available

Error Handling: - Package installation failures fail the test - Module loading failures fail the test - Volume cleanup runs even if test fails - Service issues are logged

setup_vg(ensure_minimum_devices_with_same_block_sizes)

Set up an LVM Volume Group (VG) with Physical Volumes (PVs) for testing.

This fixture creates a Volume Group using the provided block devices. It handles the creation of Physical Volumes from the block devices and ensures proper cleanup after tests, even if they fail.

Parameters:

Name	Type	Description	Default
ensure_minimum_devices_with_same_block_sizes	list[BlockDevice]	List of BlockDevice objects with matching block sizes to be used for creating Physical Volumes.	required

Yields:

Name	Type	Description
str	str	Name of the created Volume Group.

Raises:

Type	Description
AssertionError	If PV creation fails for any device.

Example

def test_volume_group(setup_vg): vg_name = setup_vg # Use vg_name in your test...

Source code in sts_libs/src/sts/fixtures/lvm_fixtures.py

@pytest.fixture
def setup_vg(ensure_minimum_devices_with_same_block_sizes: list[BlockDevice]) -> Generator[str, None, None]:
    """Set up an LVM Volume Group (VG) with Physical Volumes (PVs) for testing.

    This fixture creates a Volume Group using the provided block devices. It handles the creation
    of Physical Volumes from the block devices and ensures proper cleanup after tests, even if
    they fail.

    Args:
        ensure_minimum_devices_with_same_block_sizes: List of BlockDevice objects with matching
            block sizes to be used for creating Physical Volumes.

    Yields:
        str: Name of the created Volume Group.

    Raises:
        AssertionError: If PV creation fails for any device.

    Example:
        def test_volume_group(setup_vg):
            vg_name = setup_vg
            # Use vg_name in your test...
    """
    vg_name = getenv('STS_VG_NAME', 'stsvg0')
    pvs = []

    try:
        # Create PVs
        for device in ensure_minimum_devices_with_same_block_sizes:
            device_name = str(device.path).replace('/dev/', '')
            device_path = str(device.path)

            pv = lvm.PhysicalVolume(name=device_name, path=device_path)
            assert pv.create(), f'Failed to create PV on device {device_path}'
            pvs.append(pv)

        # Create VG
        vg = lvm.VolumeGroup(name=vg_name, pvs=[pv.path for pv in pvs])
        assert vg.create(), f'Failed to create VG {vg_name}'

        yield vg_name

    finally:
        # Cleanup in reverse order
        vg = lvm.VolumeGroup(name=vg_name)
        if not vg.remove():
            logging.warning(f'Failed to remove VG {vg_name}')

        for pv in pvs:
            if not pv.remove():
                logging.warning(f'Failed to remove PV {pv.path}')

RDMA Fixtures

sts.fixtures.rdma_fixtures

RDMA test fixtures.

This module provides fixtures for testing RDMA (Remote Direct Memory Access): - Device discovery and validation - Device configuration and management - Port and interface handling - SR-IOV configuration

Fixture Dependencies: 1. _exists_rdma (base fixture) - Validates RDMA device presence - Skips tests if no devices found

1. rdma_device (independent fixture)
2. Creates device factory function
3. Validates device existence
4. Provides device management

Common Usage: 1. Basic device validation: @pytest.mark.usefixtures('_exists_rdma') def test_rdma(): # Test runs only if RDMA device exists

1. Specific device testing: def test_device(rdma_device): device = rdma_device('mlx5_0') # Test specific RDMA device

2. Port configuration: def test_ports(rdma_device): device = rdma_device('mlx5_0') ports = device.get_ports() # Test port configuration

3. SR-IOV setup: def test_sriov(rdma_device): device = rdma_device('mlx5_0') sriov = device.get_sriov() # Test SR-IOV configuration

Error Handling: - Missing devices skip tests - Invalid device IDs raise assertion errors - Device access issues are logged - Configuration failures are reported

rdma_device()

Create RDMA device factory.

This fixture provides a factory function for RDMA devices: - Creates device instances on demand - Validates device existence - Provides device management interface - Supports multiple device types

Device Management: - Port configuration - Interface binding - SR-IOV setup - Power management

Returns:

Type	Description
Callable[[str], RdmaDevice]	Factory function that takes HCA ID and returns RdmaDevice

Example

def test_rdma(rdma_device):
    # Create device instance
    device = rdma_device('mlx5_0')
...
    # Access device information
    assert device.exists
...
    # Configure ports
    ports = device.get_ports()
    for port in ports:
        print(f'Port {port.name}: {port.state}')
...
    # Set up SR-IOV if supported
    if device.is_sriov_capable:
        sriov = device.get_sriov()
        sriov.set_numvfs('4')

Source code in sts_libs/src/sts/fixtures/rdma_fixtures.py

@pytest.fixture(scope='class')
def rdma_device() -> Callable[[str], RdmaDeviceType]:
    """Create RDMA device factory.

    This fixture provides a factory function for RDMA devices:
    - Creates device instances on demand
    - Validates device existence
    - Provides device management interface
    - Supports multiple device types

    Device Management:
    - Port configuration
    - Interface binding
    - SR-IOV setup
    - Power management

    Returns:
        Factory function that takes HCA ID and returns RdmaDevice

    Example:
        ```python
        def test_rdma(rdma_device):
            # Create device instance
            device = rdma_device('mlx5_0')
        ...
            # Access device information
            assert device.exists
        ...
            # Configure ports
            ports = device.get_ports()
            for port in ports:
                print(f'Port {port.name}: {port.state}')
        ...
            # Set up SR-IOV if supported
            if device.is_sriov_capable:
                sriov = device.get_sriov()
                sriov.set_numvfs('4')
        ```
    """

    def _device_factory(hca_id: str) -> RdmaDeviceType:
        """Create RDMA device.

        Creates and validates RDMA device instance:
        - Checks device existence
        - Initializes device paths
        - Sets up device attributes
        - Validates configuration

        Args:
            hca_id: HCA ID (e.g. 'mlx5_0', 'mlx4_1')

        Returns:
            RDMA device instance

        Raises:
            AssertionError: If device not found or invalid

        Example:
            ```python
            device = _device_factory('mlx5_0')
            assert device.exists
            ```
        """
        assert exists_device(hca_id), f'No RDMA device found: {hca_id}'
        return RdmaDevice(ibdev=hca_id)

    return _device_factory

Stratis Fixtures

sts.fixtures.stratis_fixtures

Stratis test fixtures.

This module provides fixtures for testing Stratis storage: - Pool creation and management - Filesystem operations - Encryption configuration - Error injection and recovery

Fixture Dependencies: 1. _stratis_test (base fixture) - Installs Stratis packages - Manages pool cleanup - Logs system information

1. setup_stratis_key (independent fixture)
2. Creates encryption key
3. Manages key registration

4. Handles key cleanup

5. stratis_test_pool (depends on loop_devices)

6. Creates test pool
7. Manages devices

8. Handles cleanup

9. stratis_encrypted_pool (depends on loop_devices, setup_stratis_key)

10. Creates encrypted pool
11. Manages key and devices

12. Handles secure cleanup

13. stratis_failing_pool (depends on scsi_debug_devices)

14. Creates pool with failing device
15. Injects failures
16. Tests error handling

Common Usage: 1. Basic pool testing: @pytest.mark.usefixtures('_stratis_test') def test_stratis(): # Create and test pools # Pools are cleaned up after test

1. Encrypted pool testing: def test_encryption(stratis_encrypted_pool): assert stratis_encrypted_pool.is_encrypted # Test encrypted operations

2. Error handling testing: def test_failures(stratis_failing_pool): assert not stratis_failing_pool.stop() # Test failure handling

Error Handling: - Package installation failures fail test - Pool creation failures skip test - Device failures are handled gracefully - Resources are cleaned up on failure

setup_stratis_key()

Set up Stratis encryption key.

Creates and manages encryption key: - Creates temporary key file - Registers key with Stratis - Handles key cleanup - Supports custom key configuration

Configuration (via environment): - STRATIS_KEY_DESC: Key description (default: 'sts-stratis-test-key') - STRATIS_KEY_PATH: Key file path (default: '/tmp/sts-stratis-test-key') - STRATIS_KEY: Key content (default: 'Stra123tisKey45')

Key Management: 1. Creates key file with specified content 2. Registers key with Stratis daemon 3. Yields key description for use 4. Unregisters key and removes file

Example

def test_encryption(setup_stratis_key):
    # Create encrypted pool
    pool = StratisPool()
    pool.create(key_desc=setup_stratis_key)
    assert pool.is_encrypted

Source code in sts_libs/src/sts/fixtures/stratis_fixtures.py

@pytest.fixture
def setup_stratis_key() -> Generator[str, None, None]:
    """Set up Stratis encryption key.

    Creates and manages encryption key:
    - Creates temporary key file
    - Registers key with Stratis
    - Handles key cleanup
    - Supports custom key configuration

    Configuration (via environment):
    - STRATIS_KEY_DESC: Key description (default: 'sts-stratis-test-key')
    - STRATIS_KEY_PATH: Key file path (default: '/tmp/sts-stratis-test-key')
    - STRATIS_KEY: Key content (default: 'Stra123tisKey45')

    Key Management:
    1. Creates key file with specified content
    2. Registers key with Stratis daemon
    3. Yields key description for use
    4. Unregisters key and removes file

    Example:
        ```python
        def test_encryption(setup_stratis_key):
            # Create encrypted pool
            pool = StratisPool()
            pool.create(key_desc=setup_stratis_key)
            assert pool.is_encrypted
        ```
    """
    stratis_key = Key()
    keydesc = getenv('STRATIS_KEY_DESC', 'sts-stratis-test-key')
    keypath = getenv('STRATIS_KEY_PATH', '/tmp/sts-stratis-test-key')
    key = getenv('STRATIS_KEY', 'Stra123tisKey45')

    # Create key file
    keyp = Path(keypath)
    keyp.write_text(key)
    assert keyp.is_file()

    # Register key with Stratis
    assert stratis_key.set(keydesc=keydesc, keyfile_path=keypath).succeeded

    yield keydesc

    # Clean up
    assert stratis_key.unset(keydesc).succeeded
    keyp.unlink()
    assert not keyp.is_file()

stratis_encrypted_pool(loop_devices, setup_stratis_key)

Create encrypted test pool with loop devices.

Creates and manages encrypted pool: - Uses loop devices as storage - Creates encrypted pool - Manages encryption key - Handles secure cleanup

Parameters:

Name	Type	Description	Default
loop_devices	list[str]	Loop device fixture (requires 2 devices)	required
setup_stratis_key	str	Stratis key fixture	required

Pool Configuration: - Name: 'sts-stratis-test-pool' - Devices: Provided loop devices - Encrypted with provided key - Default settings

Example

@pytest.mark.parametrize('loop_devices', [2], indirect=True)
def test_pool(stratis_encrypted_pool):
    # Test encrypted operations
    assert stratis_encrypted_pool.is_encrypted
    fs = stratis_encrypted_pool.create_filesystem('test')
    assert fs.exists

Source code in sts_libs/src/sts/fixtures/stratis_fixtures.py

@pytest.fixture
def stratis_encrypted_pool(loop_devices: list[str], setup_stratis_key: str) -> Generator[StratisPool, None, None]:
    """Create encrypted test pool with loop devices.

    Creates and manages encrypted pool:
    - Uses loop devices as storage
    - Creates encrypted pool
    - Manages encryption key
    - Handles secure cleanup

    Args:
        loop_devices: Loop device fixture (requires 2 devices)
        setup_stratis_key: Stratis key fixture

    Pool Configuration:
    - Name: 'sts-stratis-test-pool'
    - Devices: Provided loop devices
    - Encrypted with provided key
    - Default settings

    Example:
        ```python
        @pytest.mark.parametrize('loop_devices', [2], indirect=True)
        def test_pool(stratis_encrypted_pool):
            # Test encrypted operations
            assert stratis_encrypted_pool.is_encrypted
            fs = stratis_encrypted_pool.create_filesystem('test')
            assert fs.exists
        ```
    """
    pool = StratisPool()
    pool.name = 'sts-stratis-test-pool'
    pool.blockdevs = loop_devices

    # Create encrypted pool
    config = PoolCreateConfig(key_desc=setup_stratis_key)
    if not pool.create(config):
        pytest.skip('Failed to create encrypted test pool')

    yield pool

    # Clean up
    pool.destroy()

stratis_failing_pool(scsi_debug_devices)

Create test pool with failing devices.

Creates pool for failure testing: - Uses SCSI debug devices - Injects device failures - Tests error handling - Manages cleanup

Parameters:

Name	Type	Description	Default
scsi_debug_devices	list[str]	SCSI debug device fixture	required

Failure Injection: - Every operation fails - Noisy error reporting - Tests error handling - Recovery procedures

Example

@pytest.mark.parametrize('scsi_debug_devices', [2], indirect=True)
def test_pool(stratis_failing_pool):
    # Test failure handling
    assert not stratis_failing_pool.stop()
    assert 'error' in stratis_failing_pool.status

Source code in sts_libs/src/sts/fixtures/stratis_fixtures.py

@pytest.fixture
def stratis_failing_pool(scsi_debug_devices: list[str]) -> Generator[StratisPool, None, None]:
    """Create test pool with failing devices.

    Creates pool for failure testing:
    - Uses SCSI debug devices
    - Injects device failures
    - Tests error handling
    - Manages cleanup

    Args:
        scsi_debug_devices: SCSI debug device fixture

    Failure Injection:
    - Every operation fails
    - Noisy error reporting
    - Tests error handling
    - Recovery procedures

    Example:
        ```python
        @pytest.mark.parametrize('scsi_debug_devices', [2], indirect=True)
        def test_pool(stratis_failing_pool):
            # Test failure handling
            assert not stratis_failing_pool.stop()
            assert 'error' in stratis_failing_pool.status
        ```
    """
    # Get first device for injection
    device = ScsiDebugDevice(scsi_debug_devices[0])

    # Inject failures (every operation fails with noisy error)
    device.inject_failure(every_nth=1, opts=1)

    # Create pool
    pool = StratisPool()
    pool.name = 'sts-stratis-test-pool'
    pool.blockdevs = [scsi_debug_devices[0]]  # Only use first device

    if not pool.create():
        pytest.skip('Failed to create test pool')

    yield pool

    # Clean up
    pool.destroy()

stratis_test_pool(loop_devices)

Create test pool with loop devices.

Creates and manages test pool: - Uses loop devices as storage - Creates standard pool - Handles cleanup - Supports testing operations

Parameters:

Name	Type	Description	Default
loop_devices	list[str]	Loop device fixture (requires 2 devices)	required

Pool Configuration: - Name: 'sts-stratis-test-pool' - Devices: Provided loop devices - Standard (non-encrypted) pool - Default settings

Example

@pytest.mark.parametrize('loop_devices', [2], indirect=True)
def test_pool(stratis_test_pool):
    # Test pool operations
    fs = stratis_test_pool.create_filesystem('test')
    assert fs.exists

Source code in sts_libs/src/sts/fixtures/stratis_fixtures.py

@pytest.fixture
def stratis_test_pool(loop_devices: list[str]) -> Generator[StratisPool, None, None]:
    """Create test pool with loop devices.

    Creates and manages test pool:
    - Uses loop devices as storage
    - Creates standard pool
    - Handles cleanup
    - Supports testing operations

    Args:
        loop_devices: Loop device fixture (requires 2 devices)

    Pool Configuration:
    - Name: 'sts-stratis-test-pool'
    - Devices: Provided loop devices
    - Standard (non-encrypted) pool
    - Default settings

    Example:
        ```python
        @pytest.mark.parametrize('loop_devices', [2], indirect=True)
        def test_pool(stratis_test_pool):
            # Test pool operations
            fs = stratis_test_pool.create_filesystem('test')
            assert fs.exists
        ```
    """
    pool = StratisPool()
    pool.name = 'sts-stratis-test-pool'
    pool.blockdevs = loop_devices

    # Create pool
    if not pool.create():
        pytest.skip('Failed to create test pool')

    yield pool

    # Clean up
    pool.destroy()

Target Fixtures

sts.fixtures.target_fixtures

Target test fixtures.

This module provides fixtures for testing storage targets: - Target creation and configuration - Backstore management (block, fileio, ramdisk) - ACL and authentication setup - LUN management

Fixture Dependencies: 1. _target_test (base fixture) - Installs target utilities - Manages target cleanup - Logs system information

1. backstore_*_setup (depends on _target_test)
2. block: Creates block backstore with loop device
3. fileio: Creates fileio backstore

4. ramdisk: Creates ramdisk backstore

5. iscsi_target_setup (depends on _target_test)

6. Creates iSCSI target
7. Configures ACLs and LUNs

8. Manages cleanup

9. configure_auth (depends on _target_test)

10. Sets up CHAP authentication
11. Configures mutual CHAP
12. Manages credentials

Common Usage: 1. Basic target testing: @pytest.mark.usefixtures('_target_test') def test_target(): # Create and test targets # Targets are cleaned up after test

1. Backstore testing: @pytest.mark.parametrize('backstore_block_setup', [{'name': 'test', 'size': 1024*1024}], indirect=True) def test_backstore(backstore_block_setup): # Test backstore operations

2. iSCSI target testing: @pytest.mark.parametrize('iscsi_target_setup', [{'t_iqn': 'iqn.test', 'n_luns': 2}], indirect=True) def test_iscsi(iscsi_target_setup): # Test iSCSI target operations

3. Authentication testing: @pytest.mark.parametrize('configure_auth', [{'t_iqn': 'iqn.test', 'chap_username': 'user', 'chap_password': 'pass'}], indirect=True) def test_auth(configure_auth): # Test authentication

Error Handling: - Package installation failures fail test - Target creation failures are handled - Resource cleanup runs on failure - Authentication errors are logged

backstore_block_setup(_target_test, request)

Create block backstore with loop device.

Creates block backstore using loop device: - Creates temporary loop device - Sets up block backstore - Manages cleanup - Supports custom size

Parameters:

Name	Type	Description	Default
request	SubRequest	Fixture request with parameters: - name: Backstore name - size: Loop device size in MB	required

Example

@pytest.mark.parametrize('backstore_block_setup', [{'name': 'test', 'size': 1024}], indirect=True)
def test_backstore(backstore_block_setup):
    assert backstore_block_setup.exists

Source code in sts_libs/src/sts/fixtures/target_fixtures.py

@pytest.fixture
def backstore_block_setup(_target_test: None, request: SubRequest) -> Generator[BackstoreBlock, None, None]:
    """Create block backstore with loop device.

    Creates block backstore using loop device:
    - Creates temporary loop device
    - Sets up block backstore
    - Manages cleanup
    - Supports custom size

    Args:
        request: Fixture request with parameters:
            - name: Backstore name
            - size: Loop device size in MB

    Example:
        ```python
        @pytest.mark.parametrize('backstore_block_setup', [{'name': 'test', 'size': 1024}], indirect=True)
        def test_backstore(backstore_block_setup):
            assert backstore_block_setup.exists
        ```
    """
    loop_dev = None
    backstore = None
    try:
        # Create loop device
        loop_dev = LoopDevice.create(
            name=request.param['name'],
            size_mb=request.param['size'] // (1024 * 1024),
        )
        if not loop_dev:
            pytest.skip('Failed to create loop device')

        # Create backstore
        backstore = BackstoreBlock(name=request.param['name'])
        backstore.create_backstore(dev=str(loop_dev.path))
        yield backstore

    except Exception:
        logging.exception('Failed to set up block backstore')
        raise

    finally:
        # Clean up
        if backstore:
            backstore.delete_backstore()
        if loop_dev:
            loop_dev.remove()

backstore_fileio_setup(_target_test, request)

Create fileio backstore.

Creates fileio backstore: - Creates backing file - Sets up fileio backstore - Manages cleanup - Supports custom size

Parameters:

Name	Type	Description	Default
request	SubRequest	Fixture request with parameters: - name: Backstore name - size: File size in bytes - file_or_dev: File path	required

Example

@pytest.mark.parametrize('backstore_fileio_setup', [{'name': 'test', 'size': 1024 * 1024}], indirect=True)
def test_backstore(backstore_fileio_setup):
    assert backstore_fileio_setup.exists

Source code in sts_libs/src/sts/fixtures/target_fixtures.py

@pytest.fixture
def backstore_fileio_setup(_target_test: None, request: SubRequest) -> Generator[BackstoreFileio, None, None]:
    """Create fileio backstore.

    Creates fileio backstore:
    - Creates backing file
    - Sets up fileio backstore
    - Manages cleanup
    - Supports custom size

    Args:
        request: Fixture request with parameters:
            - name: Backstore name
            - size: File size in bytes
            - file_or_dev: File path

    Example:
        ```python
        @pytest.mark.parametrize('backstore_fileio_setup', [{'name': 'test', 'size': 1024 * 1024}], indirect=True)
        def test_backstore(backstore_fileio_setup):
            assert backstore_fileio_setup.exists
        ```
    """
    backstore = None
    try:
        backstore = BackstoreFileio(name=request.param['name'])
        backstore.create_backstore(
            size=str(request.param['size']),
            file_or_dev=request.param.get('file_or_dev') or f'{request.param["name"]}_file',
        )
        yield backstore

    except Exception:
        logging.exception('Failed to set up fileio backstore')
        raise

    finally:
        if backstore:
            backstore.delete_backstore()

backstore_ramdisk_setup(_target_test, request)

Create ramdisk backstore.

Creates ramdisk backstore: - Allocates memory - Sets up ramdisk backstore - Manages cleanup - Supports custom size

Parameters:

Name	Type	Description	Default
request	SubRequest	Fixture request with parameters: - name: Backstore name - size: Size in bytes	required

Example

@pytest.mark.parametrize('backstore_ramdisk_setup', [{'name': 'test', 'size': 1024 * 1024}], indirect=True)
def test_backstore(backstore_ramdisk_setup):
    assert backstore_ramdisk_setup.exists

Source code in sts_libs/src/sts/fixtures/target_fixtures.py

@pytest.fixture
def backstore_ramdisk_setup(_target_test: None, request: SubRequest) -> Generator[BackstoreRamdisk, None, None]:
    """Create ramdisk backstore.

    Creates ramdisk backstore:
    - Allocates memory
    - Sets up ramdisk backstore
    - Manages cleanup
    - Supports custom size

    Args:
        request: Fixture request with parameters:
            - name: Backstore name
            - size: Size in bytes

    Example:
        ```python
        @pytest.mark.parametrize('backstore_ramdisk_setup', [{'name': 'test', 'size': 1024 * 1024}], indirect=True)
        def test_backstore(backstore_ramdisk_setup):
            assert backstore_ramdisk_setup.exists
        ```
    """
    backstore = None
    try:
        backstore = BackstoreRamdisk(name=request.param['name'])
        backstore.create_backstore(size=str(request.param['size']))
        yield backstore

    except Exception:
        logging.exception('Failed to set up ramdisk backstore')
        raise

    finally:
        if backstore:
            backstore.delete_backstore()

configure_auth(request)

Configure CHAP authentication.

Sets up CHAP authentication: - Creates target with auth - Configures CHAP credentials - Supports mutual CHAP - Manages cleanup

Parameters:

Name	Type	Description	Default
request	SubRequest	Fixture request with parameters: - t_iqn: Target IQN - i_iqn: Initiator IQN - chap_username: CHAP username - chap_password: CHAP password - chap_target_username: Mutual CHAP username (optional) - chap_target_password: Mutual CHAP password (optional) - tpg_or_acl: Configure TPG or ACL auth	required

Example

@pytest.mark.parametrize(
    'configure_auth', [{'t_iqn': 'iqn.test', 'chap_username': 'user', 'chap_password': 'pass'}], indirect=True
)
def test_auth(configure_auth):
    assert configure_auth.exists

Source code in sts_libs/src/sts/fixtures/target_fixtures.py

@pytest.fixture
def configure_auth(request: SubRequest) -> Generator[Iscsi, None, None]:
    """Configure CHAP authentication.

    Sets up CHAP authentication:
    - Creates target with auth
    - Configures CHAP credentials
    - Supports mutual CHAP
    - Manages cleanup

    Args:
        request: Fixture request with parameters:
            - t_iqn: Target IQN
            - i_iqn: Initiator IQN
            - chap_username: CHAP username
            - chap_password: CHAP password
            - chap_target_username: Mutual CHAP username (optional)
            - chap_target_password: Mutual CHAP password (optional)
            - tpg_or_acl: Configure TPG or ACL auth

    Example:
        ```python
        @pytest.mark.parametrize(
            'configure_auth', [{'t_iqn': 'iqn.test', 'chap_username': 'user', 'chap_password': 'pass'}], indirect=True
        )
        def test_auth(configure_auth):
            assert configure_auth.exists
        ```
    """
    target_wwn = request.param['t_iqn']
    target = Iscsi(target_wwn=target_wwn)

    try:
        # Create target
        target.create_target()

        # Add backstore
        backstore = BackstoreFileio(name='auth_test')
        backstore.create_backstore(size='1M', file_or_dev='auth_test_file')
        luns = IscsiLUN(target_wwn=target_wwn)
        luns.create_lun(storage_object=backstore.path)

        # Configure auth
        if request.param['tpg_or_acl'] == 'acl':
            acl = ACL(target_wwn=target_wwn, initiator_wwn=request.param['i_iqn'])
            acl.create_acl()
            acl.set_auth(
                userid=request.param['chap_username'],
                password=request.param['chap_password'],
                mutual_userid=request.param.get('chap_target_username', ''),
                mutual_password=request.param.get('chap_target_password', ''),
            )

        yield target

    finally:
        target.delete_target()

iscsi_target_setup(_target_test, request)

Create iSCSI target with ACLs and LUNs.

Creates complete iSCSI target: - Creates target with IQN - Sets up ACLs - Creates LUNs - Manages cleanup

Parameters:

Name	Type	Description	Default
request	SubRequest	Fixture request with parameters: - t_iqn: Target IQN (optional) - i_iqn: Initiator IQN (optional) - n_luns: Number of LUNs (optional) - back_size: Backstore size in bytes (optional)	required

Example

@pytest.mark.parametrize('iscsi_target_setup', [{'t_iqn': 'iqn.test', 'n_luns': 2}], indirect=True)
def test_target(iscsi_target_setup):
    assert iscsi_target_setup.exists

Source code in sts_libs/src/sts/fixtures/target_fixtures.py

@pytest.fixture(scope='class')
def iscsi_target_setup(_target_test: None, request: SubRequest) -> Generator[Iscsi, None, None]:
    """Create iSCSI target with ACLs and LUNs.

    Creates complete iSCSI target:
    - Creates target with IQN
    - Sets up ACLs
    - Creates LUNs
    - Manages cleanup

    Args:
        request: Fixture request with parameters:
            - t_iqn: Target IQN (optional)
            - i_iqn: Initiator IQN (optional)
            - n_luns: Number of LUNs (optional)
            - back_size: Backstore size in bytes (optional)

    Example:
        ```python
        @pytest.mark.parametrize('iscsi_target_setup', [{'t_iqn': 'iqn.test', 'n_luns': 2}], indirect=True)
        def test_target(iscsi_target_setup):
            assert iscsi_target_setup.exists
        ```
    """
    params = request.param
    with target_setup(
        t_iqn=params.get('t_iqn'),
        i_iqn=params.get('i_iqn'),
        n_luns=params.get('n_luns', 0),
        back_size=params.get('back_size'),
    ) as target:
        yield target

target_setup(*, t_iqn=None, i_iqn=None, n_luns=0, back_size=None)

Set up iSCSI target.

Creates and manages iSCSI target: - Creates target with IQN - Sets up ACLs if needed - Creates LUNs if needed - Manages cleanup

Parameters:

Name	Type	Description	Default
t_iqn	str | None	Target IQN	None
i_iqn	str | None	Initiator IQN	None
n_luns	int	Number of LUNs	0
back_size	int | None	Backstore size in bytes	None

Yields:

Type	Description
Iscsi	iSCSI target instance

Example

with target_setup(t_iqn='iqn.test', n_luns=2) as target:
    # Use target
    assert target.exists

Source code in sts_libs/src/sts/fixtures/target_fixtures.py

@contextmanager
def target_setup(
    *,
    t_iqn: str | None = None,
    i_iqn: str | None = None,
    n_luns: int = 0,
    back_size: int | None = None,
) -> Generator[Iscsi, None, None]:
    """Set up iSCSI target.

    Creates and manages iSCSI target:
    - Creates target with IQN
    - Sets up ACLs if needed
    - Creates LUNs if needed
    - Manages cleanup

    Args:
        t_iqn: Target IQN
        i_iqn: Initiator IQN
        n_luns: Number of LUNs
        back_size: Backstore size in bytes

    Yields:
        iSCSI target instance

    Example:
        ```python
        with target_setup(t_iqn='iqn.test', n_luns=2) as target:
            # Use target
            assert target.exists
        ```
    """
    target_wwn = t_iqn or DEFAULT_TARGET_IQN
    target = Iscsi(target_wwn=target_wwn)

    try:
        # Create target
        target.create_target()

        # Add ACL if needed
        if i_iqn:
            acl = ACL(target_wwn=target_wwn, initiator_wwn=i_iqn)
            acl.create_acl()

        # Add LUNs if needed
        if back_size and n_luns > 0:
            luns = IscsiLUN(target_wwn)
            for n in range(n_luns):
                name = f'backstore{n}'
                backstore = BackstoreFileio(name=name)
                backstore.create_backstore(size=str(back_size), file_or_dev=f'{name}_file')
                luns.create_lun(storage_object=backstore.path)

        yield target

    finally:
        target.delete_target()