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

debugfs_module_reader(managed_module)

Fixture to prepare and provide access to a module's debugfs directory.

Relies on the 'managed_module' fixture to ensure the module is loaded. Ensures debugfs is mounted and the module's debugfs directory exists.

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

@pytest.fixture
def debugfs_module_reader(managed_module: ModuleInfo) -> Generator[Directory, None, None]:
    """Fixture to prepare and provide access to a module's debugfs directory.

    Relies on the 'managed_module' fixture to ensure the module is loaded.
    Ensures debugfs is mounted and the module's debugfs directory exists.
    """
    module_name = managed_module.name
    if not module_name:
        pytest.skip(f'Managed module fixture yielded ModuleInfo with name=None for {managed_module}')

    logging.info(f'Setting up debugfs reader for already loaded module: {module_name}')

    # Module loading is handled by the managed_module fixture dependency.
    # We can assume managed_module.loaded is True here, otherwise the test would have skipped.

    # Ensure debugfs is mounted
    debugfs_path = Path('/sys/kernel/debug')
    if not debugfs_path.is_mount():
        logging.info(f'Debugfs not mounted at {debugfs_path}, attempting mount.')
        try:
            # Check if debugfs filesystem type is already mounted somewhere else
            mount_output = run('mount')
            if f'debugfs on {debugfs_path}' not in mount_output.stdout and ' type debugfs' in mount_output.stdout:
                existing_mount = [line for line in mount_output.stdout.splitlines() if ' type debugfs' in line]
                logging.warning(f'Debugfs already mounted elsewhere: {existing_mount}. Proceeding anyway.')
            # Only attempt mount if not already mounted at the target path
            elif f'debugfs on {debugfs_path}' not in mount_output.stdout:
                run(f'mount -t debugfs none {debugfs_path}')

            if not debugfs_path.is_mount():
                pytest.skip(f'Failed to mount debugfs at {debugfs_path} after attempt')
        except OSError as e:
            pytest.skip(f'OS error mounting debugfs at {debugfs_path}: {e}')

    # Check module debugfs directory
    module_debugfs_dir = debugfs_path / module_name
    if not module_debugfs_dir.exists():
        pytest.skip(f'Debugfs directory {module_debugfs_dir} not found for module {module_name}')
    if not module_debugfs_dir.is_dir():
        pytest.skip(f'Path {module_debugfs_dir} exists but is not a directory')

    logging.info(f'Providing Directory object for {module_debugfs_dir}')
    yield Directory(module_debugfs_dir)

    # No specific cleanup needed here; module unloading is handled by managed_module teardown.
    logging.info(f'Finished using debugfs reader for module: {module_name}')

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) - size_mb: Size of each device in MB (default: 1024) Set via parametrize: @pytest.mark.parametrize('loop_devices', [2], indirect=True) Or with custom size: @pytest.mark.parametrize('loop_devices', [{'count': 1, 'size_mb': 4096}], indirect=True)

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(scope='class')
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)
    - size_mb: Size of each device in MB (default: 1024)
      Set via parametrize: @pytest.mark.parametrize('loop_devices', [2], indirect=True)
      Or with custom size: @pytest.mark.parametrize('loop_devices', [{'count': 1, 'size_mb': 4096}], indirect=True)

    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)
        ```
    """
    # Handle different parameter formats
    param = getattr(request, 'param', 1)
    if isinstance(param, dict):
        count = param.get('count', 1)
        size_mb = param.get('size_mb', 1024)
    else:
        count = param  # Backward compatibility for just count
        size_mb = 1024  # Default size
    devices = []

    # Create devices one by one
    for _ in range(count):
        device = LoopDevice.create(size_mb=size_mb)
        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()

prepare_1minutetip_disk()

This fixture is used to prepare a spare disk for testing on specific 1minutetip flavor (ci.m1.small.ephemeral).

It will wipe /dev/vdb and return a single-item list of BlockDevice objects.

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

@pytest.fixture(scope='class')
def prepare_1minutetip_disk() -> list[BlockDevice]:
    """This fixture is used to prepare a spare disk for testing on specific 1minutetip flavor (ci.m1.small.ephemeral).

    It will wipe /dev/vdb and return a single-item list of BlockDevice objects.
    """
    flag = Path('/var/tmp/STS_PREPARE_1MINUTETIP_DISK_FLAG')
    disk_path = '/dev/vdb'
    try:
        disk = BlockDevice(disk_path)
    except DeviceNotFoundError:
        pytest.fail(f'Disk {disk_path} not found')

    # Wipe disk if flag file does not exist, this is to avoid wiping the disk multiple times.
    # We need to remove partition table that is always there after provisioning.
    if not flag.exists():
        assert disk.wipe_device()
        flag.touch()
    else:
        logging.info(f'Disk {disk_path} already wiped')

    # Return a list of one device to be used with setup_vg fixture
    return [disk]

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) == 4
    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(scope='class')
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) == 4
            assert all(d.startswith('/dev/sd') for d in scsi_debug_devices)
        ```
    """
    count = getattr(request, 'param', 1)  # Default to 1 device if not specified
    total = count**2  # expected_devices = num_tgts * add_host
    logging.info(f'Creating {total} 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) < total:
        device.remove()
        pytest.skip(f'Expected {total} SCSI debug devices, got {len(devices or [])}')

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

    # Clean up
    device.remove()

timed_operation()

Fixture providing timed operation context manager.

Example

def test_example(timed_operation):
    with timed_operation('My operation'):
        do_something()

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

@pytest.fixture
def timed_operation() -> Callable[[str], AbstractContextManager[None]]:
    """Fixture providing timed operation context manager.

    Example:
        ```python
        def test_example(timed_operation):
            with timed_operation('My operation'):
                do_something()
        ```
    """

    @contextmanager
    def _timed_operation(description: str) -> Generator[None, None, None]:
        start = datetime.now(tz=timezone.utc)
        logging.info(f'Starting: {description}')
        try:
            yield
        finally:
            duration = datetime.now(tz=timezone.utc) - start
            logging.info(f'Completed: {description} (took {duration.total_seconds():.1f}s)')

    return _timed_operation

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 paths.

Returns:

Type	Description
Callable[[str], list[Path]]	Function to get device paths with optional vendor parameter

Example

def test_device(get_test_device):
    devices = get_test_device('LIO-ORG')  # or get_test_device('COMPELNT')
    for device in devices:
        assert device.exists()

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

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

    Returns:
        Function to get device paths with optional vendor parameter

    Example:
        ```python
        def test_device(get_test_device):
            devices = get_test_device('LIO-ORG')  # or get_test_device('COMPELNT')
            for device in devices:
                assert device.exists()
        ```
    """

    def _get_test_device(vendor: str = 'LIO-ORG') -> list[Path]:
        """Get test device paths.

        Args:
            vendor: SCSI vendor name to search for (default: 'LIO-ORG')

        Returns:
            List of device paths

        Raises:
            AssertionError: If no devices found
        """

        def _extract_device_paths(devices: Sequence[StorageDevice]) -> list[Path]:
            """Extract valid device paths from device objects."""
            return [Path(str(device.path)) for device in devices if device.path]

        mp_service = MultipathService()
        if mp_service.is_running():
            devices = MultipathDevice.get_all()
            if devices:
                device_paths = _extract_device_paths(devices)
                if device_paths:
                    return device_paths

        devices = ScsiDevice.get_by_vendor(vendor)
        assert devices, f'No {vendor} devices found'

        device_paths = _extract_device_paths(devices)
        assert device_paths, f'No valid device paths found for {vendor} devices'
        return device_paths

    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: pytest.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()
        ```
    """
    assert ensure_installed('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: pytest.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

binary_metadata_file()

Create a pre-allocated binary file for thin metadata operations.

This fixture creates a 5MB binary file in /var/tmp that can be used as output for thin_restore or other DMPD tools that require pre-allocated files.

Yields:

Name	Type	Description
Path	Path	Path to the allocated binary metadata file in /var/tmp

Example

def test_thin_restore(binary_metadata_file, metadata_backup):
    xml_file = metadata_backup['metadata_backup_path']
    binary_file = binary_metadata_file

    # thin_restore can now write to the pre-allocated binary file
    restore_result = dmpd.thin_restore(input=str(xml_file), output=str(binary_file))
    assert restore_result.succeeded

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

@pytest.fixture
def binary_metadata_file() -> Generator[Path, None, None]:
    """Create a pre-allocated binary file for thin metadata operations.

    This fixture creates a 5MB binary file in /var/tmp that can be used as output
    for thin_restore or other DMPD tools that require pre-allocated files.

    Yields:
        Path: Path to the allocated binary metadata file in /var/tmp

    Example:
        ```python
        def test_thin_restore(binary_metadata_file, metadata_backup):
            xml_file = metadata_backup['metadata_backup_path']
            binary_file = binary_metadata_file

            # thin_restore can now write to the pre-allocated binary file
            restore_result = dmpd.thin_restore(input=str(xml_file), output=str(binary_file))
            assert restore_result.succeeded
        ```
    """
    binary_file = Path('/var/tmp/thin_check_metadata.bin')
    assert fallocate(str(binary_file), length='5M'), f'Failed to allocate {binary_file}'
    logging.info(f'Allocated binary metadata file: {binary_file} (5MB)')

    yield binary_file

    # Cleanup
    if binary_file.exists():
        binary_file.unlink()
        logging.debug(f'Cleaned up binary metadata file: {binary_file}')

cache_metadata_backup(cache_metadata_swap)

Create cache metadata backup files for testing.

Creates cache metadata dump and prepares repair files for testing.

Parameters:

Name	Type	Description	Default
cache_metadata_swap	dict[str, Any]	Cache metadata swap setup from cache_metadata_swap fixture	required

Yields:

Name	Type	Description
dict	dict[str, Any]	Extended information with backup file paths

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

@pytest.fixture
def cache_metadata_backup(cache_metadata_swap: dict[str, Any]) -> Generator[dict[str, Any], None, None]:
    """Create cache metadata backup files for testing.

    Creates cache metadata dump and prepares repair files for testing.

    Args:
        cache_metadata_swap: Cache metadata swap setup from cache_metadata_swap fixture

    Yields:
        dict: Extended information with backup file paths
    """
    cache_info = cache_metadata_swap.copy()
    cache_metadata_dev = cache_info['cache_metadata_dev']
    cache_dump_path = Path('/var/tmp/cache_dump')
    cache_repair_path = Path('/var/tmp/cache_repair')

    # Create cache metadata dump (to match testing expectations)
    dump_result = dmpd.cache_dump(cache_metadata_dev, output=str(cache_dump_path))
    assert dump_result.succeeded

    # Create empty repair file with proper allocation (5MB should be enough)
    assert fallocate(cache_repair_path, length='5M')

    cache_info.update(
        {
            'cache_dump_path': cache_dump_path,
            'cache_repair_path': cache_repair_path,
        }
    )

    yield cache_info

    # Cleanup files
    run(f'rm -f {cache_dump_path} {cache_repair_path}')

cache_metadata_swap(cache_split, swap_volume)

Perform cache metadata swap operation.

Swaps cache metadata to the swap volume for DMPD testing.

Parameters:

Name	Type	Description	Default
cache_split	dict[str, Any]	Cache split setup from cache_split fixture	required
swap_volume	dict[str, Any]	Swap volume setup from swap_volume fixture	required

Returns:

Name	Type	Description
dict	dict[str, Any]	Combined information with cache metadata device details

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

@pytest.fixture
def cache_metadata_swap(cache_split: dict[str, Any], swap_volume: dict[str, Any]) -> dict[str, Any]:
    """Perform cache metadata swap operation.

    Swaps cache metadata to the swap volume for DMPD testing.

    Args:
        cache_split: Cache split setup from cache_split fixture
        swap_volume: Swap volume setup from swap_volume fixture

    Returns:
        dict: Combined information with cache metadata device details
    """
    cache_info = cache_split.copy()
    swap_info = swap_volume

    # Ensure both fixtures reference the same VG
    assert cache_info['vg_name'] == swap_info['vg_name'], 'Cache and swap must be in same VG'

    vg_name = cache_info['vg_name']
    cache_pool_name = cache_info['cache_pool_name']
    swap_name = swap_info['swap_name']

    # Deactivate volumes before metadata swap
    cache_pool_lv = LogicalVolume(name=cache_pool_name, vg=vg_name)
    cache_pool_lv.deactivate()  # Ignore errors
    swap_lv = LogicalVolume(name=swap_name, vg=vg_name)
    swap_lv.deactivate()
    run('udevadm settle')

    # Swap cache metadata to swap volume (matching setup logic)
    convert_result = run(f'lvconvert -y --cachepool {vg_name}/{cache_pool_name} --poolmetadata {vg_name}/{swap_name}')
    assert convert_result.succeeded

    # Activate swap volume (now containing cache metadata)
    swap_lv = LogicalVolume(name=swap_name, vg=vg_name)
    swap_lv.activate()
    run('udevadm settle')

    # Use swap LV as cache metadata device
    cache_metadata_dev = f'/dev/{vg_name}/{swap_name}'

    # Combine information from both fixtures
    combined_info = cache_info.copy()
    combined_info.update(swap_info)
    combined_info.update(
        {
            'cache_metadata_dev': cache_metadata_dev,
            'cache_metadata_swapped': True,
        }
    )

    return combined_info

cache_pool(cache_volumes)

Create cache pool by merging cache data and metadata volumes.

Parameters:

Name	Type	Description	Default
cache_volumes	dict[str, Any]	Cache volumes setup from cache_volumes fixture	required

Returns:

Name	Type	Description
dict	dict[str, Any]	Extended cache information with pool details

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

@pytest.fixture
def cache_pool(cache_volumes: dict[str, Any]) -> dict[str, Any]:
    """Create cache pool by merging cache data and metadata volumes.

    Args:
        cache_volumes: Cache volumes setup from cache_volumes fixture

    Returns:
        dict: Extended cache information with pool details
    """
    cache_info = cache_volumes.copy()
    vg_name = cache_info['vg_name']
    cache_data_name = cache_info['cache_data_name']
    cache_meta_name = cache_info['cache_meta_name']

    # Use lvm convert to create cache pool (matching setup logic)
    convert_result = run(
        f'lvconvert -y --type cache-pool --cachemode writeback '
        f'--poolmetadata {vg_name}/{cache_meta_name} {vg_name}/{cache_data_name}'
    )
    assert convert_result.succeeded

    cache_info.update(
        {
            'cache_pool_created': True,
            'cache_pool_name': cache_data_name,  # Pool takes the name of data LV
            'cache_pool_path': f'/dev/{vg_name}/{cache_data_name}',
        }
    )

    return cache_info

cache_split(cache_volume)

Split cache volume to separate cache pool and origin.

Parameters:

Name	Type	Description	Default
cache_volume	dict[str, Any]	Cache volume setup from cache_volume fixture	required

Returns:

Name	Type	Description
dict	dict[str, Any]	Extended cache information with split cache details

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

@pytest.fixture
def cache_split(cache_volume: dict[str, Any]) -> dict[str, Any]:
    """Split cache volume to separate cache pool and origin.

    Args:
        cache_volume: Cache volume setup from cache_volume fixture

    Returns:
        dict: Extended cache information with split cache details
    """
    cache_info = cache_volume.copy()
    vg_name = cache_info['vg_name']
    cached_lv_name = cache_info['cached_lv_name']

    # Split cache (matching setup logic)
    split_result = run(f'lvconvert -y --splitcache {vg_name}/{cached_lv_name}')
    assert split_result.succeeded

    cache_info.update(
        {
            'cache_split': True,
        }
    )

    return cache_info

cache_volume(cache_pool)

Create cached volume by adding origin to cache pool.

Parameters:

Name	Type	Description	Default
cache_pool	dict[str, Any]	Cache pool setup from cache_pool fixture	required

Returns:

Name	Type	Description
dict	dict[str, Any]	Extended cache information with cached volume details

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

@pytest.fixture
def cache_volume(cache_pool: dict[str, Any]) -> dict[str, Any]:
    """Create cached volume by adding origin to cache pool.

    Args:
        cache_pool: Cache pool setup from cache_pool fixture

    Returns:
        dict: Extended cache information with cached volume details
    """
    cache_info = cache_pool.copy()
    vg_name = cache_info['vg_name']
    cache_origin_name = cache_info['cache_origin_name']
    cache_pool_name = cache_info['cache_pool_name']

    # Convert origin LV to cached LV
    convert_result = run(
        f'lvconvert -y --type cache --cachepool {vg_name}/{cache_pool_name} {vg_name}/{cache_origin_name}'
    )
    assert convert_result.succeeded

    # Create ext4 filesystem on cached volume (matching setup logic)
    assert mkfs(f'/dev/{vg_name}/{cache_origin_name}', 'ext4', force=True)

    cache_info.update(
        {
            'cache_volume_created': True,
            'cached_lv_name': cache_origin_name,  # Origin LV becomes the cached LV
            'cached_lv_path': f'/dev/{vg_name}/{cache_origin_name}',
        }
    )

    return cache_info

cache_volumes(setup_loopdev_vg)

Create cache volumes for testing.

Creates cache metadata, origin, and data logical volumes that can be used for creating cache pools and cached volumes.

Parameters:

Name	Type	Description	Default
setup_loopdev_vg	str	Volume group name from setup_loopdev_vg fixture	required

Yields:

Name	Type	Description
dict	dict[str, Any]	Information about the created cache volumes

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

@pytest.fixture
def cache_volumes(setup_loopdev_vg: str) -> Generator[dict[str, Any], None, None]:
    """Create cache volumes for testing.

    Creates cache metadata, origin, and data logical volumes that can be used
    for creating cache pools and cached volumes.

    Args:
        setup_loopdev_vg: Volume group name from setup_loopdev_vg fixture

    Yields:
        dict: Information about the created cache volumes
    """
    vg_name = setup_loopdev_vg
    cache_meta_name = 'cache_meta'
    cache_origin_name = 'cache_origin'
    cache_data_name = 'cache_data'

    # Create cache metadata LV (12MB as per original setup)
    cache_meta_lv = LogicalVolume(name=cache_meta_name, vg=vg_name)
    assert cache_meta_lv.create(size='12M')

    # Create cache origin LV (300MB as per original setup)
    cache_origin_lv = LogicalVolume(name=cache_origin_name, vg=vg_name)
    assert cache_origin_lv.create(size='300M')

    # Create cache data LV (100MB as per original setup)
    cache_data_lv = LogicalVolume(name=cache_data_name, vg=vg_name)
    assert cache_data_lv.create(size='100M')

    cache_info = {
        'vg_name': vg_name,
        'cache_meta_name': cache_meta_name,
        'cache_origin_name': cache_origin_name,
        'cache_data_name': cache_data_name,
        'cache_meta_path': f'/dev/{vg_name}/{cache_meta_name}',
        'cache_origin_path': f'/dev/{vg_name}/{cache_origin_name}',
        'cache_data_path': f'/dev/{vg_name}/{cache_data_name}',
        'cache_meta_lv': cache_meta_lv,
        'cache_origin_lv': cache_origin_lv,
        'cache_data_lv': cache_data_lv,
    }

    yield cache_info

    # Cleanup
    cache_data_lv.remove()
    cache_origin_lv.remove()
    cache_meta_lv.remove()

install_dmpd(_lvm_test)

Install required packages for device-mapper-persistent-data tools.

This fixture installs the device-mapper-persistent-data package which provides cache metadata tools like cache_check, cache_dump, cache_repair, etc.

Example

@pytest.mark.usefixtures('install_dmpd_packages')
def test_cache_tools():
    # DMPD tools are now available
    pass

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

@pytest.fixture(scope='class')
def install_dmpd(_lvm_test: None) -> None:
    """Install required packages for device-mapper-persistent-data tools.

    This fixture installs the device-mapper-persistent-data package which provides
    cache metadata tools like cache_check, cache_dump, cache_repair, etc.

    Example:
        ```python
        @pytest.mark.usefixtures('install_dmpd_packages')
        def test_cache_tools():
            # DMPD tools are now available
            pass
        ```
    """
    system = SystemManager()
    package = 'device-mapper-persistent-data'

    assert system.package_manager.install(package), f'Failed to install {package}'

load_vdo_module(_lvm_test)

Load the appropriate VDO kernel module based on kernel version.

This fixture installs the VDO package and loads the correct VDO kernel module depending on the system's kernel version: - For kernel 6.9+: uses dm-vdo module (built into kernel) - For kernel 6.8 and earlier: uses kvdo module (from kmod-kvdo package)

The fixture handles kernel version detection and falls back to dm-vdo if version parsing fails.

Parameters:

Name	Type	Description	Default
_lvm_test	None	LVM test fixture dependency (ensures LVM setup is complete)	required

Returns:

Name	Type	Description
str	str	Name of the loaded VDO module ('dm-vdo' or 'kvdo')

Raises:

Type	Description
AssertionError	If VDO package installation or module loading fails

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

@pytest.fixture(scope='class')
def load_vdo_module(_lvm_test: None) -> str:
    """Load the appropriate VDO kernel module based on kernel version.

    This fixture installs the VDO package and loads the correct VDO kernel module
    depending on the system's kernel version:
    - For kernel 6.9+: uses dm-vdo module (built into kernel)
    - For kernel 6.8 and earlier: uses kvdo module (from kmod-kvdo package)

    The fixture handles kernel version detection and falls back to dm-vdo if
    version parsing fails.

    Args:
        _lvm_test: LVM test fixture dependency (ensures LVM setup is complete)

    Returns:
        str: Name of the loaded VDO module ('dm-vdo' or 'kvdo')

    Raises:
        AssertionError: If VDO package installation or module loading fails
    """
    module = 'dm_vdo'
    system = SystemManager()
    assert system.package_manager.install(VDO_PACKAGE_NAME)
    log_package_versions(VDO_PACKAGE_NAME)
    try:
        k_version = system.info.kernel
        if k_version:
            k_version = k_version.split('.')
            # dm-vdo is available from kernel 6.9, for older version it's available
            # from kmod-kvdo package
            if int(k_version[0]) < 6 or (int(k_version[0]) == 6 and int(k_version[1]) <= 8):
                logging.info('Using kmod-kvdo')
                assert system.package_manager.install('kmod-kvdo')
                log_package_versions('kmod-kvdo')
                module = 'kvdo'
    except (ValueError, IndexError):
        # if we can't get kernel version, just try to load dm-vdo
        logging.warning('Unable to parse kernel version; defaulting to dm-vdo')

    kmod = ModuleManager()
    assert kmod.load(name=module)

    return module

lv_fixture(_lvm_test, setup_vg, request)

Create a logical volume for testing with automatic cleanup.

This fixture creates either a COW (regular) or thin logical volume depending on the parameters provided. It handles proper cleanup including thin pool removal for thin volumes.

Parameters:

Name	Type	Description	Default
_lvm_test	None	LVM test fixture dependency	required
setup_vg	str	Volume group name from setup_vg fixture	required
request	FixtureRequest	Pytest fixture request for accessing parameters	required

Yields:

Name	Type	Description
str	str	Device path to the created logical volume

Default Configuration

• lv_type: 'cow' (regular logical volume)
• extents: '25%vg'

Parameters (via pytest.mark.parametrize with indirect=True): - lv_type (str): 'cow' for regular LV or 'thin' for thin LV (default: 'cow') - lv_name (str): Name for the logical volume (default: based on lv_type) - extents (str): Size specification in extents (default: '25%vg') - pool_name (str): Thin pool name (only for thin type, default: 'stspool1_25vg') - virtualsize (str): Virtual size for thin LV (only for thin type, default: '512M')

Examples:

Basic COW usage with default parameters:

def test_cow_operations(lv_fixture):
    dev_path = lv_fixture
    # dev_path is '/dev/vgname/stscow25vglv1'

With thin LV:

@pytest.mark.parametrize('lv_fixture', [{'lv_type': 'thin'}], indirect=True)
def test_thin_operations(lv_fixture):
    dev_path = lv_fixture
    # dev_path is '/dev/vgname/ststhin25vglv1'

With custom configuration:

@pytest.mark.parametrize(
    'lv_fixture',
    [
        {'lv_type': 'cow', 'extents': '50%vg', 'lv_name': 'custom_lv'},
        {'lv_type': 'thin', 'virtualsize': '1G', 'pool_name': 'mypool'},
    ],
    indirect=True,
    ids=['large-cow', 'large-thin'],
)
def test_various_lvs(lv_fixture):
    dev_path = lv_fixture

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

@pytest.fixture
def lv_fixture(_lvm_test: None, setup_vg: str, request: pytest.FixtureRequest) -> Generator[str, None, None]:
    """Create a logical volume for testing with automatic cleanup.

    This fixture creates either a COW (regular) or thin logical volume depending on
    the parameters provided. It handles proper cleanup including thin pool removal
    for thin volumes.

    Args:
        _lvm_test: LVM test fixture dependency
        setup_vg: Volume group name from setup_vg fixture
        request: Pytest fixture request for accessing parameters

    Yields:
        str: Device path to the created logical volume

    Default Configuration:
        - lv_type: 'cow' (regular logical volume)
        - extents: '25%vg'

    Parameters (via pytest.mark.parametrize with indirect=True):
        - lv_type (str): 'cow' for regular LV or 'thin' for thin LV (default: 'cow')
        - lv_name (str): Name for the logical volume (default: based on lv_type)
        - extents (str): Size specification in extents (default: '25%vg')
        - pool_name (str): Thin pool name (only for thin type, default: 'stspool1_25vg')
        - virtualsize (str): Virtual size for thin LV (only for thin type, default: '512M')

    Examples:
        Basic COW usage with default parameters:

        ```python
        def test_cow_operations(lv_fixture):
            dev_path = lv_fixture
            # dev_path is '/dev/vgname/stscow25vglv1'
        ```

        With thin LV:

        ```python
        @pytest.mark.parametrize('lv_fixture', [{'lv_type': 'thin'}], indirect=True)
        def test_thin_operations(lv_fixture):
            dev_path = lv_fixture
            # dev_path is '/dev/vgname/ststhin25vglv1'
        ```

        With custom configuration:

        ```python
        @pytest.mark.parametrize(
            'lv_fixture',
            [
                {'lv_type': 'cow', 'extents': '50%vg', 'lv_name': 'custom_lv'},
                {'lv_type': 'thin', 'virtualsize': '1G', 'pool_name': 'mypool'},
            ],
            indirect=True,
            ids=['large-cow', 'large-thin'],
        )
        def test_various_lvs(lv_fixture):
            dev_path = lv_fixture
        ```
    """
    params = getattr(request, 'param', {})
    lv_type = params.get('lv_type', 'cow')
    vg_name = setup_vg

    # Set defaults based on lv_type
    if lv_type == 'thin':
        default_lv_name = 'ststhin25vglv1'
        default_pool_name = 'stspool1_25vg'
    else:
        default_lv_name = 'stscow25vglv1'
        default_pool_name = None

    lv_name = params.get('lv_name', getenv('LV_NAME', default_lv_name))
    extents = params.get('extents', '25%vg')
    pool_name = params.get('pool_name', getenv('THIN_POOL_NAME', default_pool_name)) if lv_type == 'thin' else None
    virtualsize = params.get('virtualsize', '512M')

    lv = LogicalVolume(name=lv_name, vg=vg_name)

    if lv_type == 'thin':
        assert pool_name is not None, 'pool_name is required for thin LV'
        assert lv.create(
            type='thin',
            thinpool=pool_name,
            extents=extents.upper(),
            virtualsize=virtualsize,
        )
    else:
        assert lv.create(extents=extents)

    yield f'/dev/{vg_name}/{lv_name}'

    # Cleanup
    if lv_type == 'thin' and pool_name:
        pool = ThinPool(name=pool_name, vg=vg_name)
        assert pool.remove_with_thin_volumes(force='', yes='')
    else:
        lv = LogicalVolume(name=lv_name, vg=vg_name)
        assert lv.remove()

lvm2_version()

Get LVM2 package version as VersionInfo.

This fixture retrieves the LVM2 version from the installed package and returns it as a VersionInfo object for proper version comparison.

Returns:

Name	Type	Description
VersionInfo	VersionInfo	Parsed LVM2 version or VersionInfo(0, 0, 0) if not found

Example

def test_with_version(lvm2_version):
    if lvm2_version >= VersionInfo.from_string('2.02.171-6'):
        # Handle new behavior
    else:
        # Handle old behavior

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

@pytest.fixture(scope='class')
def lvm2_version() -> VersionInfo:
    """Get LVM2 package version as VersionInfo.

    This fixture retrieves the LVM2 version from the installed package
    and returns it as a VersionInfo object for proper version comparison.

    Returns:
        VersionInfo: Parsed LVM2 version or VersionInfo(0, 0, 0) if not found

    Example:
        ```python
        def test_with_version(lvm2_version):
            if lvm2_version >= VersionInfo.from_string('2.02.171-6'):
                # Handle new behavior
            else:
                # Handle old behavior
        ```
    """
    result = run('rpm -q lvm2 --queryformat "%{VERSION}-%{RELEASE}"')
    if result.rc == 0:
        return VersionInfo.from_string(result.stdout.strip())

    # Fallback to lvm version command
    result = run('lvm version')
    if result.rc == 0:
        lines = result.stdout.strip().split('\n')
        for line in lines:
            if 'LVM version:' in line:
                return VersionInfo.from_string(line.split(':')[1].strip())

    return VersionInfo(0, 0, 0)

lvm_config()

Provide an LvmConfig instance for LVM configuration management.

This fixture provides access to LVM configuration settings through the LvmConfig class. Use this for tests that need to read or modify lvm.conf settings.

Yields:

Name	Type	Description
LvmConfig	LvmConfig	An LvmConfig instance for configuration management

Example

def test_config_value(lvm_config):
    threshold = lvm_config.get_thin_pool_autoextend_threshold()
    assert threshold == '100'

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

@pytest.fixture
def lvm_config() -> LvmConfig:
    """Provide an LvmConfig instance for LVM configuration management.

    This fixture provides access to LVM configuration settings through the LvmConfig class.
    Use this for tests that need to read or modify lvm.conf settings.

    Yields:
        LvmConfig: An LvmConfig instance for configuration management

    Example:
        ```python
        def test_config_value(lvm_config):
            threshold = lvm_config.get_thin_pool_autoextend_threshold()
            assert threshold == '100'
        ```
    """
    return LvmConfig()

lvm_config_restore(lvm_config)

Provide LvmConfig with automatic restoration of thin pool settings after test.

This fixture saves the current thin pool configuration values before the test and restores them after the test completes. Use this when your test modifies LVM configuration settings that should be restored.

Parameters:

Name	Type	Description	Default
lvm_config	LvmConfig	LvmConfig fixture dependency	required

Yields:

Name	Type	Description
LvmConfig	LvmConfig	The same LvmConfig instance with restoration on cleanup

Example

def test_modify_config(lvm_config_restore):
    config = lvm_config_restore
    config.set_thin_pool_metadata_require_separate_pvs('1')
    # ... test ...
    # Configuration is automatically restored after test

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

@pytest.fixture
def lvm_config_restore(lvm_config: LvmConfig) -> Generator[LvmConfig, None, None]:
    """Provide LvmConfig with automatic restoration of thin pool settings after test.

    This fixture saves the current thin pool configuration values before the test
    and restores them after the test completes. Use this when your test modifies
    LVM configuration settings that should be restored.

    Args:
        lvm_config: LvmConfig fixture dependency

    Yields:
        LvmConfig: The same LvmConfig instance with restoration on cleanup

    Example:
        ```python
        def test_modify_config(lvm_config_restore):
            config = lvm_config_restore
            config.set_thin_pool_metadata_require_separate_pvs('1')
            # ... test ...
            # Configuration is automatically restored after test
        ```
    """
    # Save original values
    original_values = {
        LvmConfig.THIN_POOL_METADATA_REQUIRE_SEPARATE_PVS: lvm_config.get_thin_pool_metadata_require_separate_pvs(),
        LvmConfig.THIN_POOL_AUTOEXTEND_THRESHOLD: lvm_config.get_thin_pool_autoextend_threshold(),
        LvmConfig.THIN_POOL_AUTOEXTEND_PERCENT: lvm_config.get_thin_pool_autoextend_percent(),
    }

    yield lvm_config

    # Restore original values
    for key, value in original_values.items():
        if value is not None:
            lvm_config.set(key, value)

metadata_backup(metadata_swap)

Create metadata backup files for testing.

Creates metadata backup using thin_dump and prepares repair file for testing.

Parameters:

Name	Type	Description	Default
metadata_swap	dict[str, Any]	Metadata swap setup from metadata_swap fixture	required

Yields:

Name	Type	Description
dict	dict[str, Any]	Extended information with backup file paths

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

@pytest.fixture
def metadata_backup(metadata_swap: dict[str, Any]) -> Generator[dict[str, Any], None, None]:
    """Create metadata backup files for testing.

    Creates metadata backup using thin_dump and prepares repair file for testing.

    Args:
        metadata_swap: Metadata swap setup from metadata_swap fixture

    Yields:
        dict: Extended information with backup file paths
    """
    vol_info = metadata_swap.copy()
    metadata_dev = vol_info['metadata_dev']
    metadata_backup_path = Path('/var/tmp/metadata')
    metadata_repair_path = Path('/var/tmp/metadata_repair')

    # Create metadata backup using thin_dump (matching backup_metadata from main.fmf)
    backup_cmd = f'thin_dump --format xml --repair {metadata_dev} --output {metadata_backup_path}'
    backup_result = run(backup_cmd)
    assert backup_result.succeeded

    # Create proper metadata files for testing
    # 1. Create empty repair file with proper allocation (5MB should be enough)
    assert fallocate(metadata_repair_path, length='5M')

    # 2. Create a working metadata file that thin_repair can actually repair
    metadata_working_path = Path('/var/tmp/metadata_working')
    assert fallocate(metadata_working_path, length='5M')

    # 3. Populate the working metadata file with valid data from backup
    restore_working_cmd = f'thin_restore -i {metadata_backup_path} -o {metadata_working_path}'
    restore_working_result = run(restore_working_cmd)
    assert restore_working_result.succeeded, f'Failed to create working metadata: {restore_working_result.stderr}'

    # Update vol_info to include all metadata files
    vol_info.update(
        {
            'metadata_backup_path': metadata_backup_path,
            'metadata_repair_path': metadata_repair_path,
            'metadata_working_path': metadata_working_path,
        }
    )

    yield vol_info

    # Cleanup files
    run(f'rm -f {metadata_backup_path} {metadata_repair_path} {metadata_working_path}')

metadata_snapshot(thin_volumes_with_lifecycle)

Create metadata snapshot for thin pool.

Creates a metadata snapshot while the thin pool is active and handles the suspend/message/resume sequence for snapshot creation.

Parameters:

Name	Type	Description	Default
thin_volumes_with_lifecycle	dict[str, Any]	Thin volumes setup from thin_volumes_with_lifecycle fixture	required

Yields:

Name	Type	Description
dict	dict[str, Any]	Pool information with snapshot status

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

@pytest.fixture
def metadata_snapshot(thin_volumes_with_lifecycle: dict[str, Any]) -> Generator[dict[str, Any], None, None]:
    """Create metadata snapshot for thin pool.

    Creates a metadata snapshot while the thin pool is active and handles
    the suspend/message/resume sequence for snapshot creation.

    Args:
        thin_volumes_with_lifecycle: Thin volumes setup from thin_volumes_with_lifecycle fixture

    Yields:
        dict: Pool information with snapshot status
    """
    pool_info = thin_volumes_with_lifecycle.copy()
    vg_name = pool_info['vg_name']
    pool_name = pool_info['pool_name']

    udevadm_settle()

    # Create metadata snapshot while pool is still active
    pool_device = f'/dev/mapper/{vg_name}-{pool_name}-tpool'

    # Suspend -> message -> resume sequence (matching metadata_snapshot from setup.py)
    suspend_result = run(f'dmsetup suspend {pool_device}')
    assert suspend_result.succeeded

    message_result = run(f'dmsetup message {pool_device} 0 reserve_metadata_snap')
    assert message_result.succeeded

    resume_result = run(f'dmsetup resume {pool_device}')
    assert resume_result.succeeded

    # Now deactivate thin volumes (matching deactivate_thinvols from setup)
    for i in range(int(pool_info['thin_count'])):
        thin_name = f'{pool_info["thin_base_name"]}{i}'
        thin_lv = LogicalVolume(name=thin_name, vg=vg_name)
        thin_lv.deactivate()

    udevadm_settle()

    pool_info.update(
        {
            'pool_device': pool_device,
            'has_snapshot': True,
        }
    )

    yield pool_info

    # Release metadata snapshot
    run(f'dmsetup message {pool_device} 0 release_metadata_snap')

metadata_swap(metadata_snapshot, swap_volume)

Perform metadata swap operation between thin pool and swap volume.

Deactivates the thin pool and swap volume, then uses lvconvert to swap the metadata from the thin pool to the swap volume.

Parameters:

Name	Type	Description	Default
metadata_snapshot	dict[str, Any]	Metadata snapshot setup from metadata_snapshot fixture	required
swap_volume	dict[str, Any]	Swap volume setup from swap_volume fixture	required

Returns:

Name	Type	Description
dict	dict[str, Any]	Combined information with metadata device details

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

@pytest.fixture
def metadata_swap(metadata_snapshot: dict[str, Any], swap_volume: dict[str, Any]) -> dict[str, Any]:
    """Perform metadata swap operation between thin pool and swap volume.

    Deactivates the thin pool and swap volume, then uses lvconvert to swap
    the metadata from the thin pool to the swap volume.

    Args:
        metadata_snapshot: Metadata snapshot setup from metadata_snapshot fixture
        swap_volume: Swap volume setup from swap_volume fixture

    Returns:
        dict: Combined information with metadata device details
    """
    pool_info = metadata_snapshot.copy()
    swap_info = swap_volume

    # Ensure both fixtures reference the same VG
    assert pool_info['vg_name'] == swap_info['vg_name'], 'Pool and swap must be in same VG'

    vg_name = pool_info['vg_name']
    pool_name = pool_info['pool_name']
    swap_name = swap_info['swap_name']

    # Deactivate pool and swap (matching swap_metadata logic from setup.py)
    pool_lv = LogicalVolume(name=pool_name, vg=vg_name)
    pool_lv.deactivate()
    swap_lv = LogicalVolume(name=swap_name, vg=vg_name)
    swap_lv.deactivate()

    logging.info(run('lvs').stdout)
    udevadm_settle()

    # Swap metadata using lv_convert --poolmetadata (exact logic from setup.py)
    # This converts the swap LV to hold the thin pool's metadata
    convert_cmd = f'lvconvert -y --thinpool {vg_name}/{pool_name} --poolmetadata {vg_name}/{swap_name}'
    convert_result = run(convert_cmd)
    assert convert_result.succeeded

    # Activate swap volume (now containing metadata)
    swap_lv = LogicalVolume(name=swap_name, vg=vg_name)
    swap_lv.activate()

    # Use swap LV as metadata device (it now contains the metadata)
    metadata_dev = f'/dev/{vg_name}/{swap_name}'

    # Combine information from both fixtures
    combined_info = pool_info.copy()
    combined_info.update(swap_info)
    combined_info.update(
        {
            'metadata_dev': metadata_dev,
        }
    )

    return combined_info

mount_lv_fixture(_lvm_test, setup_vg, request)

Mount a logical volume on a test directory with automatic cleanup.

This fixture creates a logical volume (COW or thin), formats it with a filesystem, mounts it, and handles cleanup. It combines the functionality of lv_fixture and mount operations.

Parameters:

Name	Type	Description	Default
_lvm_test	None	LVM test fixture dependency	required
setup_vg	str	Volume group name from setup_vg fixture	required
request	FixtureRequest	Pytest fixture request for accessing parameters	required

Yields:

Name	Type	Description
Directory	Directory	Directory representation of mount point

Default Configuration

• lv_type: 'cow' (regular logical volume)
• fs_type: 'xfs'
• mount_point: '/mnt/lvmntdir'

Parameters (via pytest.mark.parametrize with indirect=True): - lv_type (str): 'cow' for regular LV or 'thin' for thin LV (default: 'cow') - lv_name (str): Name for the logical volume (default: based on lv_type) - extents (str): Size specification in extents (default: '25%vg') - pool_name (str): Thin pool name (only for thin type) - virtualsize (str): Virtual size for thin LV (default: '512M') - fs_type (str): Filesystem type (default: 'xfs') - mount_point (str): Mount point path (default: '/mnt/lvmntdir')

Examples:

Basic COW usage:

def test_filesystem_operations(mount_lv_fixture):
    mnt_dir = mount_lv_fixture
    # Write files to mnt_dir.path

With thin LV and ext4:

@pytest.mark.parametrize('mount_lv_fixture', [{'lv_type': 'thin', 'fs_type': 'ext4'}], indirect=True)
def test_thin_ext4(mount_lv_fixture):
    mnt_dir = mount_lv_fixture

Multiple configurations:

@pytest.mark.parametrize(
    'mount_lv_fixture',
    [
        {'lv_type': 'cow', 'fs_type': 'xfs'},
        {'lv_type': 'thin', 'fs_type': 'xfs'},
        {'lv_type': 'cow', 'fs_type': 'ext4'},
        {'lv_type': 'thin', 'fs_type': 'ext4'},
    ],
    indirect=True,
    ids=['cow-xfs', 'thin-xfs', 'cow-ext4', 'thin-ext4'],
)
def test_all_combinations(mount_lv_fixture):
    mnt_dir = mount_lv_fixture

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

@pytest.fixture
def mount_lv_fixture(
    _lvm_test: None, setup_vg: str, request: pytest.FixtureRequest
) -> Generator[Directory, None, None]:
    """Mount a logical volume on a test directory with automatic cleanup.

    This fixture creates a logical volume (COW or thin), formats it with a filesystem,
    mounts it, and handles cleanup. It combines the functionality of lv_fixture and
    mount operations.

    Args:
        _lvm_test: LVM test fixture dependency
        setup_vg: Volume group name from setup_vg fixture
        request: Pytest fixture request for accessing parameters

    Yields:
        Directory: Directory representation of mount point

    Default Configuration:
        - lv_type: 'cow' (regular logical volume)
        - fs_type: 'xfs'
        - mount_point: '/mnt/lvmntdir'

    Parameters (via pytest.mark.parametrize with indirect=True):
        - lv_type (str): 'cow' for regular LV or 'thin' for thin LV (default: 'cow')
        - lv_name (str): Name for the logical volume (default: based on lv_type)
        - extents (str): Size specification in extents (default: '25%vg')
        - pool_name (str): Thin pool name (only for thin type)
        - virtualsize (str): Virtual size for thin LV (default: '512M')
        - fs_type (str): Filesystem type (default: 'xfs')
        - mount_point (str): Mount point path (default: '/mnt/lvmntdir')

    Examples:
        Basic COW usage:

        ```python
        def test_filesystem_operations(mount_lv_fixture):
            mnt_dir = mount_lv_fixture
            # Write files to mnt_dir.path
        ```

        With thin LV and ext4:

        ```python
        @pytest.mark.parametrize('mount_lv_fixture', [{'lv_type': 'thin', 'fs_type': 'ext4'}], indirect=True)
        def test_thin_ext4(mount_lv_fixture):
            mnt_dir = mount_lv_fixture
        ```

        Multiple configurations:

        ```python
        @pytest.mark.parametrize(
            'mount_lv_fixture',
            [
                {'lv_type': 'cow', 'fs_type': 'xfs'},
                {'lv_type': 'thin', 'fs_type': 'xfs'},
                {'lv_type': 'cow', 'fs_type': 'ext4'},
                {'lv_type': 'thin', 'fs_type': 'ext4'},
            ],
            indirect=True,
            ids=['cow-xfs', 'thin-xfs', 'cow-ext4', 'thin-ext4'],
        )
        def test_all_combinations(mount_lv_fixture):
            mnt_dir = mount_lv_fixture
        ```
    """
    params = getattr(request, 'param', {})
    lv_type = params.get('lv_type', 'cow')
    vg_name = setup_vg

    # Set defaults based on lv_type
    if lv_type == 'thin':
        default_lv_name = 'ststhinmntlv1'
        default_pool_name = 'stspool1_mnt'
        default_mount_point = '/mnt/thinlvmntdir'
    else:
        default_lv_name = 'stscowmntlv1'
        default_pool_name = None
        default_mount_point = '/mnt/lvcowmntdir'

    lv_name = params.get('lv_name', default_lv_name)
    extents = params.get('extents', '25%vg')
    pool_name = params.get('pool_name', default_pool_name) if lv_type == 'thin' else None
    virtualsize = params.get('virtualsize', '512M')
    fs_type = params.get('fs_type', 'xfs')
    mount_point = params.get('mount_point', default_mount_point)

    # Create LV
    lv = LogicalVolume(name=lv_name, vg=vg_name)
    if lv_type == 'thin':
        assert pool_name is not None, 'pool_name is required for thin LV'
        assert lv.create(
            type='thin',
            thinpool=pool_name,
            extents=extents.upper(),
            virtualsize=virtualsize,
        )
    else:
        assert lv.create(extents=extents)

    dev_path = f'/dev/{vg_name}/{lv_name}'

    # Create filesystem
    assert mkfs(device=dev_path, fs_type=fs_type)

    # Create mount point directory
    mnt_dir = Directory(Path(mount_point), create=True)
    assert mnt_dir.exists, f'Failed to create mount point directory {mount_point}'

    # Mount the LV
    assert mount(device=dev_path, mountpoint=mount_point)

    yield mnt_dir

    # Cleanup
    assert umount(mountpoint=mount_point)
    mnt_dir.remove_dir()

    if lv_type == 'thin' and pool_name:
        pool = ThinPool(name=pool_name, vg=vg_name)
        assert pool.remove_with_thin_volumes(force='', yes='')
    else:
        lv = LogicalVolume(name=lv_name, vg=vg_name)
        assert lv.remove()

mounted_thin_and_regular_lvs(thin_and_regular_lvs)

Mount thin and regular LVs for performance testing.

Creates ext4 filesystems on both volumes and mounts them. Useful for I/O performance comparison tests.

Parameters:

Name	Type	Description	Default
thin_and_regular_lvs	dict[str, Any]	Volume information from thin_and_regular_lvs fixture	required

Yields:

Name	Type	Description
dict	dict[str, Any]	Extended information including: - All fields from thin_and_regular_lvs - thin_lv_mnt: Path to thin volume mount point - regular_lv_mnt: Path to regular volume mount point - filesystem: Filesystem type used (ext4)

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

@pytest.fixture
def mounted_thin_and_regular_lvs(thin_and_regular_lvs: dict[str, Any]) -> Generator[dict[str, Any], None, None]:
    """Mount thin and regular LVs for performance testing.

    Creates ext4 filesystems on both volumes and mounts them.
    Useful for I/O performance comparison tests.

    Args:
        thin_and_regular_lvs: Volume information from thin_and_regular_lvs fixture

    Yields:
        dict: Extended information including:
            - All fields from thin_and_regular_lvs
            - thin_lv_mnt: Path to thin volume mount point
            - regular_lv_mnt: Path to regular volume mount point
            - filesystem: Filesystem type used (ext4)
    """
    vol_info = thin_and_regular_lvs.copy()
    thin_device = vol_info['thin_device']
    regular_device = vol_info['regular_device']

    thin_lv_mnt = Path('/mnt/thin_lv')
    regular_lv_mnt = Path('/mnt/regular_lv')

    # Create directories and filesystems
    thin_dir = Directory(thin_lv_mnt, create=True)
    regular_dir = Directory(regular_lv_mnt, create=True)

    filesystem = 'ext4'  # Use ext4 for consistent performance testing
    assert mkfs(thin_device, filesystem, force=True)
    assert mkfs(regular_device, filesystem, force=True)

    assert mount(thin_device, thin_lv_mnt)
    assert mount(regular_device, regular_lv_mnt)

    vol_info.update(
        {
            'thin_lv_mnt': thin_lv_mnt,
            'regular_lv_mnt': regular_lv_mnt,
            'filesystem': filesystem,
        }
    )

    yield vol_info

    # Cleanup
    umount(thin_lv_mnt)
    umount(regular_lv_mnt)
    thin_dir.remove_dir()
    regular_dir.remove_dir()

multiple_mntpoints_fixture(_lvm_test, setup_vg, request)

Create multiple logical volumes with mounted filesystems for testing.

This fixture creates multiple logical volumes (COW or thin) within a volume group, formats them with filesystems, and mounts them to separate mount points. It provides a unified interface replacing the separate cow/thin and xfs/ext4 fixtures.

Parameters:

Name	Type	Description	Default
_lvm_test	None	LVM test fixture dependency	required
setup_vg	str	Volume group name from setup_vg fixture	required
request	FixtureRequest	Pytest fixture request for accessing parameters	required

Yields:

Type	Description
list[Directory]	list[Directory]: List of Directory objects representing the mount points

Default Configuration

• lv_type: 'cow'
• fs_type: 'xfs'
• num_of_mntpoints: 6

Parameters (via pytest.mark.parametrize with indirect=True): - lv_type (str): 'cow' for regular LVs or 'thin' for thin LVs (default: 'cow') - fs_type (str): Filesystem type - 'xfs' or 'ext4' (default: 'xfs') - num_of_mntpoints (int): Number of mount points to create (default: 6) - lv_name (str): Base name for logical volumes - mount_point (str): Base mount point path - pool_name (str): Base name for thin pools (thin only) - virtualsize (str): Virtual size for thin LVs (default: '512M') - percentage_of_vg_to_use (int): Total VG percentage to use (default: 50)

Examples:

Basic COW with xfs (default):

def test_cow_xfs(multiple_mntpoints_fixture):
    mntpoints = multiple_mntpoints_fixture
    assert len(mntpoints) == 6

Thin with ext4:

@pytest.mark.parametrize('multiple_mntpoints_fixture', [{'lv_type': 'thin', 'fs_type': 'ext4'}], indirect=True)
def test_thin_ext4(multiple_mntpoints_fixture):
    mntpoints = multiple_mntpoints_fixture

All combinations for comprehensive testing:

@pytest.mark.parametrize(
    'multiple_mntpoints_fixture',
    [
        {'lv_type': 'cow', 'fs_type': 'xfs'},
        {'lv_type': 'thin', 'fs_type': 'xfs'},
        {'lv_type': 'cow', 'fs_type': 'ext4'},
        {'lv_type': 'thin', 'fs_type': 'ext4'},
    ],
    indirect=True,
    ids=['cow-xfs', 'thin-xfs', 'cow-ext4', 'thin-ext4'],
)
def test_all_fs_types(multiple_mntpoints_fixture):
    mntpoints = multiple_mntpoints_fixture

Custom number of mount points:

@pytest.mark.parametrize(
    'multiple_mntpoints_fixture', [{'num_of_mntpoints': 10, 'percentage_of_vg_to_use': 80}], indirect=True
)
def test_many_volumes(multiple_mntpoints_fixture):
    assert len(multiple_mntpoints_fixture) == 10

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

@pytest.fixture
def multiple_mntpoints_fixture(
    _lvm_test: None, setup_vg: str, request: pytest.FixtureRequest
) -> Generator[list[Directory], None, None]:
    """Create multiple logical volumes with mounted filesystems for testing.

    This fixture creates multiple logical volumes (COW or thin) within a volume group,
    formats them with filesystems, and mounts them to separate mount points. It provides
    a unified interface replacing the separate cow/thin and xfs/ext4 fixtures.

    Args:
        _lvm_test: LVM test fixture dependency
        setup_vg: Volume group name from setup_vg fixture
        request: Pytest fixture request for accessing parameters

    Yields:
        list[Directory]: List of Directory objects representing the mount points

    Default Configuration:
        - lv_type: 'cow'
        - fs_type: 'xfs'
        - num_of_mntpoints: 6

    Parameters (via pytest.mark.parametrize with indirect=True):
        - lv_type (str): 'cow' for regular LVs or 'thin' for thin LVs (default: 'cow')
        - fs_type (str): Filesystem type - 'xfs' or 'ext4' (default: 'xfs')
        - num_of_mntpoints (int): Number of mount points to create (default: 6)
        - lv_name (str): Base name for logical volumes
        - mount_point (str): Base mount point path
        - pool_name (str): Base name for thin pools (thin only)
        - virtualsize (str): Virtual size for thin LVs (default: '512M')
        - percentage_of_vg_to_use (int): Total VG percentage to use (default: 50)

    Examples:
        Basic COW with xfs (default):

        ```python
        def test_cow_xfs(multiple_mntpoints_fixture):
            mntpoints = multiple_mntpoints_fixture
            assert len(mntpoints) == 6
        ```

        Thin with ext4:

        ```python
        @pytest.mark.parametrize('multiple_mntpoints_fixture', [{'lv_type': 'thin', 'fs_type': 'ext4'}], indirect=True)
        def test_thin_ext4(multiple_mntpoints_fixture):
            mntpoints = multiple_mntpoints_fixture
        ```

        All combinations for comprehensive testing:

        ```python
        @pytest.mark.parametrize(
            'multiple_mntpoints_fixture',
            [
                {'lv_type': 'cow', 'fs_type': 'xfs'},
                {'lv_type': 'thin', 'fs_type': 'xfs'},
                {'lv_type': 'cow', 'fs_type': 'ext4'},
                {'lv_type': 'thin', 'fs_type': 'ext4'},
            ],
            indirect=True,
            ids=['cow-xfs', 'thin-xfs', 'cow-ext4', 'thin-ext4'],
        )
        def test_all_fs_types(multiple_mntpoints_fixture):
            mntpoints = multiple_mntpoints_fixture
        ```

        Custom number of mount points:

        ```python
        @pytest.mark.parametrize(
            'multiple_mntpoints_fixture', [{'num_of_mntpoints': 10, 'percentage_of_vg_to_use': 80}], indirect=True
        )
        def test_many_volumes(multiple_mntpoints_fixture):
            assert len(multiple_mntpoints_fixture) == 10
        ```
    """
    params = getattr(request, 'param', {})

    yield from _create_multiple_lv_mntpoints(
        vg_name=setup_vg,
        **params,
    )

multiple_thin_pools(setup_loopdev_vg)

Create multiple thin pools with thin volumes for concurrent testing.

Creates configurable number of thin pools, each with a thin volume, for testing concurrent operations.

Parameters:

Name	Type	Description	Default
setup_loopdev_vg	str	Volume group name from setup_loopdev_vg fixture	required

Yields:

Name	Type	Description
list	list[ThinPool]	List of ThinPool objects

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

@pytest.fixture
def multiple_thin_pools(setup_loopdev_vg: str) -> Generator[list[ThinPool], None, None]:
    """Create multiple thin pools with thin volumes for concurrent testing.

    Creates configurable number of thin pools, each with a thin volume,
    for testing concurrent operations.

    Args:
        setup_loopdev_vg: Volume group name from setup_loopdev_vg fixture

    Yields:
        list: List of ThinPool objects
    """
    vg_name = setup_loopdev_vg
    pools: list[ThinPool] = []

    # Create 3 thin pools with thin volumes
    for i in range(1, 4):
        pool_name = f'pool{i}'
        lv_name = f'lv{i}'

        pool = ThinPool.create_thin_pool(pool_name, vg_name, size='30M')
        pool.create_thin_volume(lv_name, virtualsize='50M')
        pools.append(pool)

    yield pools

    # Cleanup
    for pool in pools:
        pool.remove_with_thin_volumes()

regular_lv_fixture(setup_loopdev_vg, request)

Create a regular (non-thin) logical volume for testing.

This fixture creates a regular LV that can be used for conversion to thin, testing, or other operations that require a standard logical volume.

Parameters:

Name	Type	Description	Default
setup_loopdev_vg	str	Volume group name from setup_loopdev_vg fixture	required
request	FixtureRequest	Pytest fixture request for accessing parameters	required

Yields:

Name	Type	Description
LogicalVolume	LogicalVolume	The created regular logical volume

Default Configuration

• lv_name: 'lv'
• size: '50M' (or extents if specified)

Parameters (via pytest.mark.parametrize with indirect=True): - lv_name (str): Name for the LV (default: 'lv') - size (str): Size of the LV (default: '50M') - extents (str): Size in extents (alternative to size) - inactive (bool): Create as inactive LV with '-an' flag (default: False) - zero (str): Zero option for LV creation (default: None) - skip_cleanup (bool): Skip automatic cleanup (default: False) - use when LV will be converted to thin pool and cleanup handled separately

Examples:

Basic usage:

def test_regular_lv(regular_lv_fixture):
    lv = regular_lv_fixture
    assert lv.report.lv_attr[0] == '-'  # Regular volume

With custom size:

@pytest.mark.parametrize('regular_lv_fixture', [{'size': '100M', 'lv_name': 'data'}], indirect=True)
def test_large_lv(regular_lv_fixture):
    lv = regular_lv_fixture

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

@pytest.fixture
def regular_lv_fixture(setup_loopdev_vg: str, request: pytest.FixtureRequest) -> Generator[LogicalVolume, None, None]:
    """Create a regular (non-thin) logical volume for testing.

    This fixture creates a regular LV that can be used for conversion to thin,
    testing, or other operations that require a standard logical volume.

    Args:
        setup_loopdev_vg: Volume group name from setup_loopdev_vg fixture
        request: Pytest fixture request for accessing parameters

    Yields:
        LogicalVolume: The created regular logical volume

    Default Configuration:
        - lv_name: 'lv'
        - size: '50M' (or extents if specified)

    Parameters (via pytest.mark.parametrize with indirect=True):
        - lv_name (str): Name for the LV (default: 'lv')
        - size (str): Size of the LV (default: '50M')
        - extents (str): Size in extents (alternative to size)
        - inactive (bool): Create as inactive LV with '-an' flag (default: False)
        - zero (str): Zero option for LV creation (default: None)
        - skip_cleanup (bool): Skip automatic cleanup (default: False) - use when
          LV will be converted to thin pool and cleanup handled separately

    Examples:
        Basic usage:

        ```python
        def test_regular_lv(regular_lv_fixture):
            lv = regular_lv_fixture
            assert lv.report.lv_attr[0] == '-'  # Regular volume
        ```

        With custom size:

        ```python
        @pytest.mark.parametrize('regular_lv_fixture', [{'size': '100M', 'lv_name': 'data'}], indirect=True)
        def test_large_lv(regular_lv_fixture):
            lv = regular_lv_fixture
        ```
    """
    vg_name = setup_loopdev_vg
    params = getattr(request, 'param', {})

    lv_name = params.get('lv_name', 'lv')
    size = params.get('size', '50M')
    extents = params.get('extents')
    inactive = params.get('inactive', False)
    zero = params.get('zero')
    skip_cleanup = params.get('skip_cleanup', False)

    lv = LogicalVolume(name=lv_name, vg=vg_name)

    # Build create options
    create_args = []
    create_kwargs: dict[str, str] = {}

    if inactive:
        create_args.append('-an')
    if zero:
        create_kwargs['zero'] = zero
    if extents:
        create_kwargs['extents'] = extents
    else:
        create_kwargs['size'] = size

    assert lv.create(*create_args, **create_kwargs)

    logging.info(f'Created regular LV {lv_name} in VG {vg_name}')

    yield lv

    # Cleanup - skip if LV was converted to pool (conversion tests handle cleanup)
    if not skip_cleanup:
        lv.remove(force='', yes='')
        logging.info(f'Cleaned up regular LV {lv_name}')

restored_thin_pool(metadata_backup)

Restore thin pool to a usable state after metadata operations.

WARNING: Use this fixture ONLY for tests that specifically need an active thin pool (like thin_trim). Most DMPD tools are designed to work with "broken" metadata and should use setup_thin_metadata_for_dmpd instead, which preserves the intentionally inconsistent metadata state.

This fixture uses thin_restore to repair the metadata and make the pool activatable.

Parameters:

Name	Type	Description	Default
metadata_backup	dict[str, Any]	Metadata backup setup from metadata_backup fixture	required

Yields:

Name	Type	Description
dict	dict[str, Any]	Pool information with restored pool that can be activated

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

@pytest.fixture
def restored_thin_pool(metadata_backup: dict[str, Any]) -> Generator[dict[str, Any], None, None]:
    """Restore thin pool to a usable state after metadata operations.

    WARNING: Use this fixture ONLY for tests that specifically need an active thin pool
    (like thin_trim). Most DMPD tools are designed to work with "broken" metadata and
    should use setup_thin_metadata_for_dmpd instead, which preserves the intentionally
    inconsistent metadata state.

    This fixture uses thin_restore to repair the metadata and make the pool activatable.

    Args:
        metadata_backup: Metadata backup setup from metadata_backup fixture

    Yields:
        dict: Pool information with restored pool that can be activated
    """
    vol_info = metadata_backup.copy()
    vg_name = vol_info['vg_name']
    pool_name = vol_info['pool_name']
    swap_name = vol_info['swap_name']
    metadata_backup_path = vol_info['metadata_backup_path']
    metadata_dev = vol_info['metadata_dev']

    # Step 1: Use thin_restore to repair the metadata in the swap device
    logging.info(f'Restoring metadata to repair inconsistencies in {metadata_dev}')
    restore_cmd = f'thin_restore -i {metadata_backup_path} -o {metadata_dev}'
    restore_result = run(restore_cmd)
    assert restore_result.succeeded, f'Failed to restore metadata: {restore_result.stderr}'

    # Step 2: Deactivate both volumes before swapping metadata back
    pool_lv = LogicalVolume(name=pool_name, vg=vg_name)
    pool_lv.deactivate()  # Pool might already be deactivated
    swap_lv = LogicalVolume(name=swap_name, vg=vg_name)
    swap_lv.deactivate()
    udevadm_settle()

    # Step 3: "Swap back metadata" - restore the fixed metadata to the pool
    # This matches the "Swapping back metadata" step from python-stqe cleanup
    swap_back_cmd = f'lvconvert -y --thinpool {vg_name}/{pool_name} --poolmetadata {vg_name}/{swap_name}'
    swap_back_result = run(swap_back_cmd)
    assert swap_back_result.succeeded, f'Failed to swap metadata back to pool: {swap_back_result.stderr}'

    # Step 4: Reactivate the swap volume and verify device accessibility
    swap_lv = LogicalVolume(name=swap_name, vg=vg_name)
    activate_swap_result = swap_lv.activate()
    assert activate_swap_result, 'Failed to reactivate swap volume'
    udevadm_settle()

    # Verify the metadata device exists and update the path if needed
    metadata_dev_path = f'/dev/{vg_name}/{swap_name}'
    check_dev = run(f'ls -la {metadata_dev_path}')
    if not check_dev.succeeded:
        # Try alternative device path
        metadata_dev_path = f'/dev/mapper/{vg_name}-{swap_name}'
        check_dev_alt = run(f'ls -la {metadata_dev_path}')
        assert check_dev_alt.succeeded, f'Swap device not accessible at {metadata_dev_path}'

    # Update the metadata_dev path to the verified working path
    vol_info['metadata_dev'] = metadata_dev_path

    # Now the pool should have the fixed metadata and be activatable
    vol_info.update(
        {
            'pool_can_activate': True,
            'metadata_restored': True,
            'metadata_swapped_back': True,
        }
    )

    yield vol_info

    # Leave pool in deactivated state for cleanup
    pool_lv = LogicalVolume(name=pool_name, vg=vg_name)
    pool_lv.deactivate()  # Ignore errors

setup_cache_metadata_for_dmpd(install_dmpd, cache_metadata_backup)

Set up cache metadata configuration for DMPD tool testing.

This fixture creates the necessary cache metadata setup that DMPD cache tools can operate on. Unlike thin metadata which intentionally creates "broken" state, cache metadata swap creates a working metadata device that cache tools can analyze.

Parameters:

Name	Type	Description	Default
install_dmpd	None	DMPD package installation fixture	required
cache_metadata_backup	dict[str, Any]	Cache metadata backup setup from cache_metadata_backup fixture	required

Returns:

Name	Type	Description
dict	dict[str, Any]	Extended cache information for DMPD testing

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

@pytest.fixture
def setup_cache_metadata_for_dmpd(install_dmpd: None, cache_metadata_backup: dict[str, Any]) -> dict[str, Any]:
    """Set up cache metadata configuration for DMPD tool testing.

    This fixture creates the necessary cache metadata setup that DMPD cache tools
    can operate on. Unlike thin metadata which intentionally creates "broken" state,
    cache metadata swap creates a working metadata device that cache tools can analyze.

    Args:
        install_dmpd: DMPD package installation fixture
        cache_metadata_backup: Cache metadata backup setup from cache_metadata_backup fixture

    Returns:
        dict: Extended cache information for DMPD testing
    """
    # DMPD packages are installed via install_dmpd fixture
    _ = install_dmpd

    # Use cache_metadata_backup which provides working cache metadata for DMPD testing
    return cache_metadata_backup.copy()

setup_loopdev_vg(_lvm_test, loop_devices)

Set up a volume group using loop devices.

This fixture creates a volume group using the provided loop devices. The volume group name can be customized using the STS_VG_NAME environment variable, otherwise defaults to 'stsvg0'.

Parameters:

Name	Type	Description	Default
loop_devices	list[str]	List of loop device paths to use as physical volumes.	required

Yields:

Name	Type	Description
str	str	The name of the created volume group.

Examples:

Basic usage with custom loop device configuration:

@pytest.mark.parametrize('loop_devices', [{'count': 1, 'size_mb': 4096}], indirect=True)
@pytest.mark.usefixtures('setup_loopdev_vg')
def test_large_vg_operations(setup_loopdev_vg):
    vg_name = setup_loopdev_vg
    # Create logical volumes in the 4GB VG
    lv = LogicalVolume(name='testlv', vg=vg_name, size='1G')
    assert lv.create()

Using with multiple loop devices:

@pytest.mark.parametrize('loop_devices', [{'count': 2, 'size_mb': 2048}], indirect=True)
@pytest.mark.usefixtures('setup_loopdev_vg')
def test_multi_pv_vg(setup_loopdev_vg):
    vg_name = setup_loopdev_vg
    vg = VolumeGroup(name=vg_name)
    assert vg.exists()

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

@pytest.fixture
def setup_loopdev_vg(_lvm_test: None, loop_devices: list[str]) -> Generator[str, None, None]:
    """Set up a volume group using loop devices.

    This fixture creates a volume group using the provided loop devices.
    The volume group name can be customized using the STS_VG_NAME environment
    variable, otherwise defaults to 'stsvg0'.

    Args:
        loop_devices: List of loop device paths to use as physical volumes.

    Yields:
        str: The name of the created volume group.

    Examples:
        Basic usage with custom loop device configuration:

        ```python
        @pytest.mark.parametrize('loop_devices', [{'count': 1, 'size_mb': 4096}], indirect=True)
        @pytest.mark.usefixtures('setup_loopdev_vg')
        def test_large_vg_operations(setup_loopdev_vg):
            vg_name = setup_loopdev_vg
            # Create logical volumes in the 4GB VG
            lv = LogicalVolume(name='testlv', vg=vg_name, size='1G')
            assert lv.create()
        ```

        Using with multiple loop devices:

        ```python
        @pytest.mark.parametrize('loop_devices', [{'count': 2, 'size_mb': 2048}], indirect=True)
        @pytest.mark.usefixtures('setup_loopdev_vg')
        def test_multi_pv_vg(setup_loopdev_vg):
            vg_name = setup_loopdev_vg
            vg = VolumeGroup(name=vg_name)
            assert vg.exists()
        ```
    """
    vg_name = getenv('STS_VG_NAME', 'stsvg0')
    pvs = []

    try:
        for device in loop_devices:
            pv = PhysicalVolume(name=device, path=device)
            assert pv.create(), f'Failed to create PV on device {device}'
            pvs.append(pv)

        vg = 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:
        vg = 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}')

setup_thin_metadata_for_dmpd(install_dmpd, metadata_backup)

Set up thin metadata configuration for DMPD tool testing with snapshot support.

This fixture creates the intended "broken" metadata state that DMPD tools are designed to detect, analyze, and repair. The metadata swap operation intentionally leaves the thin pool in an inconsistent state (transaction_id mismatch) to test that DMPD tools can properly handle corrupted/problematic metadata scenarios.

Parameters:

Name	Type	Description	Default
install_dmpd	None	DMPD package installation fixture	required
metadata_backup	dict[str, Any]	Metadata backup setup from metadata_backup fixture	required

Returns:

Name	Type	Description
dict	dict[str, Any]	Extended volume information with intentionally inconsistent metadata for testing

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

@pytest.fixture
def setup_thin_metadata_for_dmpd(install_dmpd: None, metadata_backup: dict[str, Any]) -> dict[str, Any]:
    """Set up thin metadata configuration for DMPD tool testing with snapshot support.

    This fixture creates the intended "broken" metadata state that DMPD tools are designed
    to detect, analyze, and repair. The metadata swap operation intentionally leaves the
    thin pool in an inconsistent state (transaction_id mismatch) to test that DMPD tools
    can properly handle corrupted/problematic metadata scenarios.

    Args:
        install_dmpd: DMPD package installation fixture
        metadata_backup: Metadata backup setup from metadata_backup fixture

    Returns:
        dict: Extended volume information with intentionally inconsistent metadata for testing
    """
    # DMPD packages are installed via install_dmpd fixture
    _ = install_dmpd

    # Use metadata_backup which preserves the "broken" metadata state for DMPD testing
    return metadata_backup.copy()

setup_thin_pool_with_vols(thin_volumes_with_lifecycle, swap_volume)

Set up thin pool with thin volumes for DMPD testing.

This is a backward-compatible fixture that combines the modular fixtures to recreate the original functionality. Uses the new modular approach internally.

Parameters:

Name	Type	Description	Default
thin_volumes_with_lifecycle	dict[str, str]	Thin volumes setup from thin_volumes_with_lifecycle fixture	required
swap_volume	dict[str, str]	Swap volume setup from swap_volume fixture	required

Returns:

Name	Type	Description
dict	dict[str, str]	Information about created volumes (compatible with original format)

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

@pytest.fixture
def setup_thin_pool_with_vols(
    thin_volumes_with_lifecycle: dict[str, str], swap_volume: dict[str, str]
) -> dict[str, str]:
    """Set up thin pool with thin volumes for DMPD testing.

    This is a backward-compatible fixture that combines the modular fixtures
    to recreate the original functionality. Uses the new modular approach internally.

    Args:
        thin_volumes_with_lifecycle: Thin volumes setup from thin_volumes_with_lifecycle fixture
        swap_volume: Swap volume setup from swap_volume fixture

    Returns:
        dict: Information about created volumes (compatible with original format)
    """
    pool_info = thin_volumes_with_lifecycle.copy()
    swap_info = swap_volume

    # Ensure both fixtures reference the same VG
    assert pool_info['vg_name'] == swap_info['vg_name'], 'Pool and swap must be in same VG'

    # Combine information from both fixtures to match original format
    volume_info = pool_info.copy()
    volume_info.update(
        {
            'swap_name': swap_info['swap_name'],
            'swap_path': swap_info['swap_path'],
        }
    )

    return volume_info

setup_vg(_lvm_test, 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(
    _lvm_test: None, 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 = 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 = 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 = 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}')

swap_volume(setup_loopdev_vg)

Create a swap volume for metadata operations.

Creates a 75MB swap logical volume that can be used for metadata swapping.

Parameters:

Name	Type	Description	Default
setup_loopdev_vg	str	Volume group name from setup_loopdev_vg fixture	required

Yields:

Name	Type	Description
dict	dict[str, Any]	Information about the created swap volume

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

@pytest.fixture
def swap_volume(setup_loopdev_vg: str) -> Generator[dict[str, Any], None, None]:
    """Create a swap volume for metadata operations.

    Creates a 75MB swap logical volume that can be used for metadata swapping.

    Args:
        setup_loopdev_vg: Volume group name from setup_loopdev_vg fixture

    Yields:
        dict: Information about the created swap volume
    """
    vg_name = setup_loopdev_vg
    swap_name = 'swapvol'

    # Create swap LV (75MB as per original setup)
    swap_lv = LogicalVolume(name=swap_name, vg=vg_name)
    assert swap_lv.create(size='75M')

    swap_info = {
        'vg_name': vg_name,
        'swap_name': swap_name,
        'swap_path': f'/dev/{vg_name}/{swap_name}',
        'swap_lv': swap_lv,
    }

    yield swap_info

    # Cleanup
    swap_lv.remove()

temp_mount_fixture(request)

Create a temporary mount point with automatic cleanup.

This fixture creates a mount point directory and handles cleanup of the directory and any temp files after the test.

Parameters:

Name	Type	Description	Default
request	FixtureRequest	Pytest fixture request for accessing parameters	required

Yields:

Name	Type	Description
dict	dict[str, Any]	Mount information including: - mount_point: Path to the mount point - mount_dir: Directory object - temp_files: List to track temp files for cleanup

Default Configuration

• mount_point: '/mnt/test'

Parameters (via pytest.mark.parametrize with indirect=True): - mount_point (str): Path for the mount point (default: '/mnt/test')

Examples:

def test_with_mount(temp_mount_fixture, regular_lv_fixture):
    mount_info = temp_mount_fixture
    lv = regular_lv_fixture

    mkfs(lv.device_path, 'ext4')
    mount(lv.device_path, mount_info['mount_point'])
    # ... test operations ...
    umount(mount_info['mount_point'])

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

@pytest.fixture
def temp_mount_fixture(request: pytest.FixtureRequest) -> Generator[dict[str, Any], None, None]:
    """Create a temporary mount point with automatic cleanup.

    This fixture creates a mount point directory and handles cleanup of
    the directory and any temp files after the test.

    Args:
        request: Pytest fixture request for accessing parameters

    Yields:
        dict: Mount information including:
            - mount_point: Path to the mount point
            - mount_dir: Directory object
            - temp_files: List to track temp files for cleanup

    Default Configuration:
        - mount_point: '/mnt/test'

    Parameters (via pytest.mark.parametrize with indirect=True):
        - mount_point (str): Path for the mount point (default: '/mnt/test')

    Examples:
        ```python
        def test_with_mount(temp_mount_fixture, regular_lv_fixture):
            mount_info = temp_mount_fixture
            lv = regular_lv_fixture

            mkfs(lv.device_path, 'ext4')
            mount(lv.device_path, mount_info['mount_point'])
            # ... test operations ...
            umount(mount_info['mount_point'])
        ```
    """
    params = getattr(request, 'param', {})
    mount_point = Path(params.get('mount_point', '/mnt/test'))

    mount_dir = Directory(mount_point, create=True)
    temp_files: list[str] = []

    mount_info = {
        'mount_point': mount_point,
        'mount_dir': mount_dir,
        'temp_files': temp_files,
    }

    yield mount_info

    # Cleanup - unmount if mounted, remove dir, clean temp files
    umount(mount_point)

    if mount_dir.exists:
        mount_dir.remove_dir()

    for temp_file in temp_files:
        run(f'rm -f {temp_file}')

thin_and_regular_lvs(setup_loopdev_vg)

Create a thin pool with thin volume and a regular LV for performance testing.

Creates: - Thin pool (1G) - Thin volume (900M virtual size) - Regular logical volume (900M)

This fixture is useful for comparing thin vs regular LV performance.

Parameters:

Name	Type	Description	Default
setup_loopdev_vg	str	Volume group name from setup_loopdev_vg fixture	required

Yields:

Name	Type	Description
dict	dict[str, Any]	Information about created volumes including: - vg_name: Volume group name - pool: ThinPool object - regular_lv: LogicalVolume object - thin_device: Device path for thin volume - regular_device: Device path for regular volume

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

@pytest.fixture
def thin_and_regular_lvs(setup_loopdev_vg: str) -> Generator[dict[str, Any], None, None]:
    """Create a thin pool with thin volume and a regular LV for performance testing.

    Creates:
    - Thin pool (1G)
    - Thin volume (900M virtual size)
    - Regular logical volume (900M)

    This fixture is useful for comparing thin vs regular LV performance.

    Args:
        setup_loopdev_vg: Volume group name from setup_loopdev_vg fixture

    Yields:
        dict: Information about created volumes including:
            - vg_name: Volume group name
            - pool: ThinPool object
            - regular_lv: LogicalVolume object
            - thin_device: Device path for thin volume
            - regular_device: Device path for regular volume
    """
    vg_name = setup_loopdev_vg

    # Create thin pool and volumes
    pool = ThinPool.create_thin_pool('pool', vg_name, size='1G')
    pool.create_thin_volume('thin_lv', virtualsize='900M')
    regular_lv = LogicalVolume(name='regular_lv', vg=vg_name)
    assert regular_lv.create(size='900M')

    yield {
        'vg_name': vg_name,
        'pool': pool,
        'regular_lv': regular_lv,
        'thin_device': f'/dev/mapper/{vg_name}-thin_lv',
        'regular_device': f'/dev/mapper/{vg_name}-regular_lv',
    }

    # Cleanup
    regular_lv.remove(force='', yes='')
    pool.remove_with_thin_volumes(force='', yes='')

thin_pool_with_volume(setup_loopdev_vg)

Create a thin pool with a single thin volume for testing.

Creates a basic thin pool and one thin volume for simple activation and attribute testing scenarios.

Parameters:

Name	Type	Description	Default
setup_loopdev_vg	str	Volume group name from setup_loopdev_vg fixture	required

Yields:

Name	Type	Description
dict	dict[str, Any]	Information about the thin pool and thin volume

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

@pytest.fixture
def thin_pool_with_volume(setup_loopdev_vg: str) -> Generator[dict[str, Any], None, None]:
    """Create a thin pool with a single thin volume for testing.

    Creates a basic thin pool and one thin volume for simple activation and
    attribute testing scenarios.

    Args:
        setup_loopdev_vg: Volume group name from setup_loopdev_vg fixture

    Yields:
        dict: Information about the thin pool and thin volume
    """
    vg_name = setup_loopdev_vg
    pool_name = 'pool'
    lv_name = 'lv1'

    # Create thin pool
    pool = ThinPool.create_thin_pool(pool_name, vg_name, size='100M')

    # Create thin volume
    thin_lv = pool.create_thin_volume(lv_name, virtualsize='50M')

    yield {
        'vg_name': vg_name,
        'pool': pool,
        'thin_lv': thin_lv,
        'pool_name': pool_name,
        'lv_name': lv_name,
    }

    # Cleanup
    pool.remove_with_thin_volumes()

thin_volumes_with_lifecycle(setup_loopdev_vg)

Create thin volumes and perform filesystem lifecycle operations.

Creates a 3GB thin pool and 10 thin volumes of 300MB each, then performs filesystem operations (create, mount, unmount, deactivate) to generate metadata activity.

Parameters:

Name	Type	Description	Default
setup_loopdev_vg	str	Volume group name from setup_loopdev_vg fixture	required

Yields:

Name	Type	Description
dict	dict[str, Any]	Extended pool information with thin volume details

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

@pytest.fixture
def thin_volumes_with_lifecycle(setup_loopdev_vg: str) -> Generator[dict[str, Any], None, None]:
    """Create thin volumes and perform filesystem lifecycle operations.

    Creates a 3GB thin pool and 10 thin volumes of 300MB each, then performs
    filesystem operations (create, mount, unmount, deactivate) to generate
    metadata activity.

    Args:
        setup_loopdev_vg: Volume group name from setup_loopdev_vg fixture

    Yields:
        dict: Extended pool information with thin volume details
    """
    vg_name = setup_loopdev_vg
    pool_name = 'thinpool'

    # Create thin pool (3GB to accommodate 10x300MB thin volumes with filesystem support)
    pool = ThinPool.create_thin_pool(pool_name, vg_name, size='3G')

    pool_info: dict[str, Any] = {
        'vg_name': vg_name,
        'pool_name': pool_name,
        'pool_path': f'/dev/{vg_name}/{pool_name}',
    }
    thin_base_name = 'thinvol'
    run(f'vgchange {vg_name} --setautoactivation n')
    # Create 10 thin volumes of 300MB each (minimum size for filesystem support)
    thin_lvs = []
    for i in range(10):
        thin_name = f'{thin_base_name}{i}'
        thin_lv = LogicalVolume(name=thin_name, vg=vg_name)
        assert thin_lv.create(type='thin', thinpool=pool_name, virtualsize='300M')
        thin_lvs.append(thin_lv)
    for thin_lv in thin_lvs:
        # Create filesystem and mount/unmount to generate metadata activity
        # This matches the mount_lv/umount_lv logic from setup.py
        thin_path = f'/dev/{vg_name}/{thin_lv.name}'
        mount_point = f'/mnt/{thin_lv.name}'

        mnt_dir = Directory(Path(mount_point), create=True)
        assert mkfs(device=thin_path, fs_type='xfs')
        assert mount(device=thin_path, mountpoint=mount_point)
        random_data_file = f'{mount_point}/random_data.txt'
        assert write_data(target=random_data_file, source='/dev/urandom', count=10, bs='1M'), (
            f'Failed to write random data to {random_data_file}'
        )
        logging.info(f'Wrote 10MB of random data to {random_data_file}')
        udevadm_settle()
        assert umount(mountpoint=mount_point)
        mnt_dir.remove_dir()

        # Deactivate thin LV with verification
        assert thin_lv.deactivate()

    pool_info.update(
        {
            'thin_count': 10,
            'thin_base_name': thin_base_name,
            'thin_lvs': thin_lvs,
        }
    )

    yield pool_info

    # Cleanup thin volumes and pool
    for thin_lv in thin_lvs:
        thin_lv.remove()
    pool.remove_with_thin_volumes(force='', yes='')

thinpool_fixture(setup_loopdev_vg, request)

Create a thin pool for testing with automatic cleanup.

This fixture creates a thin pool and automatically removes all thin volumes and the pool itself during cleanup. It supports parameterization for flexible pool configuration.

Parameters:

Name	Type	Description	Default
setup_loopdev_vg	str	Volume group name from setup_loopdev_vg fixture	required
request	FixtureRequest	Pytest fixture request for accessing parameters	required

Yields:

Name	Type	Description
ThinPool	ThinPool	The created thin pool object that can be used to create thin volumes

Default Configuration

• pool_name: 'pool'
• size: '500M'

Parameters (via pytest.mark.parametrize with indirect=True): - pool_name (str): Name for the thin pool (default: 'pool') - size (str): Size of the thin pool (default: '500M') - extents (str): Size in extents (alternative to size) - discards (str): Discard mode - 'passdown', 'nopassdown', or 'ignore' (default: None) - stripes (str): Number of stripes for the pool (default: None) - stripesize (str): Stripe size in KB (default: None) - chunksize (str): Chunk size for the pool (default: None) - poolmetadatasize (str): Metadata size for the pool (default: None) - poolmetadataspare (str): Create pool metadata spare - 'y' or 'n' (default: None) - create_thin_volume (bool): Whether to create a thin volume (default: False) - thin_volume_name (str): Name for the thin volume (default: 'lv1') - thin_volume_size (str): Virtual size for thin volume (default: '50M')

Examples:

Basic usage with default parameters:

def test_thin_operations(thinpool_fixture):
    pool = thinpool_fixture
    lv = pool.create_thin_volume('test_lv', virtualsize='100M')
    # ... test operations ...
    # Cleanup is automatic - all thin volumes and pool are removed

With custom pool size:

@pytest.mark.parametrize('thinpool_fixture', [{'size': '1G'}], indirect=True)
def test_large_pool(thinpool_fixture):
    pool = thinpool_fixture
    lv = pool.create_thin_volume('large_lv', virtualsize='500M')

With multiple parameter combinations:

@pytest.mark.parametrize(
    'thinpool_fixture',
    [
        {'size': '200M', 'pool_name': 'small_pool'},
        {'size': '1G', 'pool_name': 'large_pool'},
    ],
    indirect=True,
    ids=['small', 'large'],
)
def test_various_sizes(thinpool_fixture):
    pool = thinpool_fixture
    # Test with different pool configurations

Combined with loop_devices parameterization:

@pytest.mark.parametrize('loop_devices', [{'count': 1, 'size_mb': 4096}], indirect=True)
@pytest.mark.parametrize('thinpool_fixture', [{'size': '3G'}], indirect=True)
def test_discard_performance(thinpool_fixture):
    pool = thinpool_fixture
    discard_lv = pool.create_thin_volume('discard', virtualsize='1T')
    # ... performance test ...

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

@pytest.fixture
def thinpool_fixture(setup_loopdev_vg: str, request: pytest.FixtureRequest) -> Generator[ThinPool, None, None]:
    """Create a thin pool for testing with automatic cleanup.

    This fixture creates a thin pool and automatically removes all thin volumes
    and the pool itself during cleanup. It supports parameterization for flexible
    pool configuration.

    Args:
        setup_loopdev_vg: Volume group name from setup_loopdev_vg fixture
        request: Pytest fixture request for accessing parameters

    Yields:
        ThinPool: The created thin pool object that can be used to create thin volumes

    Default Configuration:
        - pool_name: 'pool'
        - size: '500M'

    Parameters (via pytest.mark.parametrize with indirect=True):
        - pool_name (str): Name for the thin pool (default: 'pool')
        - size (str): Size of the thin pool (default: '500M')
        - extents (str): Size in extents (alternative to size)
        - discards (str): Discard mode - 'passdown', 'nopassdown', or 'ignore' (default: None)
        - stripes (str): Number of stripes for the pool (default: None)
        - stripesize (str): Stripe size in KB (default: None)
        - chunksize (str): Chunk size for the pool (default: None)
        - poolmetadatasize (str): Metadata size for the pool (default: None)
        - poolmetadataspare (str): Create pool metadata spare - 'y' or 'n' (default: None)
        - create_thin_volume (bool): Whether to create a thin volume (default: False)
        - thin_volume_name (str): Name for the thin volume (default: 'lv1')
        - thin_volume_size (str): Virtual size for thin volume (default: '50M')

    Examples:
        Basic usage with default parameters:

        ```python
        def test_thin_operations(thinpool_fixture):
            pool = thinpool_fixture
            lv = pool.create_thin_volume('test_lv', virtualsize='100M')
            # ... test operations ...
            # Cleanup is automatic - all thin volumes and pool are removed
        ```

        With custom pool size:

        ```python
        @pytest.mark.parametrize('thinpool_fixture', [{'size': '1G'}], indirect=True)
        def test_large_pool(thinpool_fixture):
            pool = thinpool_fixture
            lv = pool.create_thin_volume('large_lv', virtualsize='500M')
        ```

        With multiple parameter combinations:

        ```python
        @pytest.mark.parametrize(
            'thinpool_fixture',
            [
                {'size': '200M', 'pool_name': 'small_pool'},
                {'size': '1G', 'pool_name': 'large_pool'},
            ],
            indirect=True,
            ids=['small', 'large'],
        )
        def test_various_sizes(thinpool_fixture):
            pool = thinpool_fixture
            # Test with different pool configurations
        ```

        Combined with loop_devices parameterization:

        ```python
        @pytest.mark.parametrize('loop_devices', [{'count': 1, 'size_mb': 4096}], indirect=True)
        @pytest.mark.parametrize('thinpool_fixture', [{'size': '3G'}], indirect=True)
        def test_discard_performance(thinpool_fixture):
            pool = thinpool_fixture
            discard_lv = pool.create_thin_volume('discard', virtualsize='1T')
            # ... performance test ...
        ```
    """
    vg_name = setup_loopdev_vg
    params = getattr(request, 'param', {})

    pool_name = params.get('pool_name', 'pool')
    size = params.get('size')
    extents = params.get('extents')
    discards = params.get('discards')
    stripes = params.get('stripes')
    stripesize = params.get('stripesize')
    chunksize = params.get('chunksize')
    poolmetadatasize = params.get('poolmetadatasize')
    poolmetadataspare = params.get('poolmetadataspare')
    create_thin_volume = params.get('create_thin_volume', False)
    thin_volume_name = params.get('thin_volume_name', 'lv1')
    thin_volume_size = params.get('thin_volume_size', '50M')

    # Default size if neither size nor extents specified
    if not size and not extents:
        size = '500M'

    # Build options dict, filtering out None values
    pool_options: dict[str, str] = {}
    if size:
        pool_options['size'] = size
    if extents:
        pool_options['extents'] = extents
    if discards:
        pool_options['discards'] = discards
    if stripes:
        pool_options['stripes'] = stripes
    if stripesize:
        pool_options['stripesize'] = stripesize
    if chunksize:
        pool_options['chunksize'] = chunksize
    if poolmetadatasize:
        pool_options['poolmetadatasize'] = poolmetadatasize
    if poolmetadataspare:
        pool_options['poolmetadataspare'] = poolmetadataspare

    pool = ThinPool.create_thin_pool(
        pool_name,
        vg_name,
        **pool_options,
    )
    logging.info(f'Created thin pool {pool_name} with size {size or extents} in VG {vg_name}')

    if create_thin_volume:
        pool.create_thin_volume(thin_volume_name, virtualsize=thin_volume_size)
        logging.info(f'Created thin volume {thin_volume_name} with size {thin_volume_size}')

    yield pool

    # Cleanup - remove all thin volumes and the pool
    pool.remove_with_thin_volumes(force='', yes='')
    logging.info(f'Cleaned up thin pool {pool_name} and all its volumes')

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_clevis_test()

Set up Tang server for Stratis Clevis encryption testing.

This fixture configures the Tang server environment: - Ensures Tang server packages are installed - Starts Tang service - Gets server information for encryption - Handles cleanup

Package Installation: - tang: Tang server package - curl: For HTTP requests - jose: For JWK operations - jq: For JSON processing

Service Management: 1. Installs required packages 2. Starts Tang service 3. Gets server thumbprint 4. Cleans up after tests

Returns:

Type	Description
None	Dictionary containing:
None	thumbprint: Server thumbprint for verification
None	url: Tang server URL

Example

def test_tang(stratis_clevis_test):
    # Create encrypted pool with Tang
    config = PoolCreateConfig(
        clevis='tang', tang_url=stratis_clevis_test['url'], thumbprint=stratis_clevis_test['thumbprint']
    )

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

@pytest.fixture
def stratis_clevis_test() -> Generator[dict[str, str], None, None]:
    """Set up Tang server for Stratis Clevis encryption testing.

    This fixture configures the Tang server environment:
    - Ensures Tang server packages are installed
    - Starts Tang service
    - Gets server information for encryption
    - Handles cleanup

    Package Installation:
    - tang: Tang server package
    - curl: For HTTP requests
    - jose: For JWK operations
    - jq: For JSON processing

    Service Management:
    1. Installs required packages
    2. Starts Tang service
    3. Gets server thumbprint
    4. Cleans up after tests

    Returns:
        Dictionary containing:
        - thumbprint: Server thumbprint for verification
        - url: Tang server URL

    Example:
        ```python
        def test_tang(stratis_clevis_test):
            # Create encrypted pool with Tang
            config = PoolCreateConfig(
                clevis='tang', tang_url=stratis_clevis_test['url'], thumbprint=stratis_clevis_test['thumbprint']
            )
        ```
    """
    system = SystemManager()
    system_info = SystemInfo()
    tang_service = 'tangd.socket'

    # Install required packages
    required_packages = ['tang', 'curl', 'jose', 'jq', 'coreutils']
    assert ensure_installed(*required_packages), 'Failed to install required packages'

    # Start Tang service if not running
    if not system.is_service_running(tang_service):
        assert system.service_start(tang_service), f'Failed to start {tang_service}'

    # Get server thumbprint
    cmd = (
        f'curl -s {system_info.hostname}/adv | '
        f'jq -r .payload | '
        f'base64 -d | '
        f'jose jwk use -i- -r -u verify -o- | '
        f'jose jwk thp -i-'
    )
    result = run(cmd=cmd)
    assert result.succeeded, 'Failed to get Tang server thumbprint'
    assert result.stdout.strip(), 'Empty thumbprint received'

    # Prepare server information
    clevis_info = {'thumbprint': result.stdout.strip(), 'url': f'http://{system_info.hostname}'}

    yield clevis_info

    # Clean up
    if system.is_service_running(tang_service):
        assert system.service_stop(tang_service), f'Failed to stop {tang_service}'

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_extend_lvm(_lvm_test, loop_devices)

Create a logical volume 70%vg size which we will use to test stratis extend-data.

First two devices from loop_devices are used for creating stratis pool. This fixture will use third and fourth loop device as PV.

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

@pytest.fixture
def stratis_extend_lvm(
    _lvm_test: None,
    loop_devices: list[str],
) -> Generator[LogicalVolume, None, None]:
    """Create a logical volume 70%vg size which we will use to test stratis extend-data.

    First two devices from loop_devices are used for creating stratis pool.
    This fixture will use third and fourth loop device as PV.
    """
    vg_name = getenv('STRATIS_VG_NAME', 'sts-stratis-volume-group')
    lv_name = getenv('STRATIS_LV_NAME', 'sts-stratis-logical-volume')
    pvs = []
    assert len(loop_devices) > 4, 'Not enough loop devices is available'
    devices = loop_devices[2:4]
    try:
        # Create PVs
        for device in devices:
            pv = PhysicalVolume(name=device, path=device)
            assert pv.create(), f'Failed to create PV on device {device}'
            pvs.append(pv)

        # Create VG
        vg = VolumeGroup(name=vg_name, pvs=devices)
        assert vg.create(), f'Failed to create VG {vg_name}'
        lv = LogicalVolume(name=lv_name, vg=vg_name)
        assert lv.create(extents='70%vg')
        yield lv

    finally:
        # Cleanup in reverse order
        vg = 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}')

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_key_desc_pool(loop_devices, setup_stratis_key)

Create a pool with keyring encryption.

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

@pytest.fixture
def stratis_key_desc_pool(loop_devices: list[str], setup_stratis_key: str) -> Generator[StratisPool, None, None]:
    """Create a pool with keyring encryption."""
    pool = StratisPool()
    pool.name = 'sts-stratis-test-pool'
    pool.blockdevs = loop_devices[:2]  # Use first two devices initially
    config = PoolCreateConfig(key_desc=setup_stratis_key)
    assert pool.create(config)
    yield pool
    pool.destroy()

stratis_no_enc_pool(loop_devices)

Create a pool without encryption.

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

@pytest.fixture
def stratis_no_enc_pool(loop_devices: list[str]) -> Generator[StratisPool, None, None]:
    """Create a pool without encryption."""
    pool = StratisPool()
    pool.name = 'sts-stratis-test-pool'
    pool.blockdevs = loop_devices[:2]  # Use first two devices initially
    assert pool.create()
    yield pool
    pool.destroy()

stratis_tang_pool(loop_devices, stratis_clevis_test)

Create a pool with Tang encryption.

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

@pytest.fixture
def stratis_tang_pool(
    loop_devices: list[str], stratis_clevis_test: dict[str, str]
) -> Generator[StratisPool, None, None]:
    """Create a pool with Tang encryption."""
    pool = StratisPool()
    pool.name = 'sts-stratis-test-pool'
    pool.blockdevs = loop_devices[:2]  # Use first two devices initially
    config = PoolCreateConfig(
        clevis='tang', tang_url=stratis_clevis_test['url'], thumbprint=stratis_clevis_test['thumbprint']
    )
    assert pool.create(config)
    yield pool
    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

13. loopback_devices

14. Creates loopback devices for testing
15. Supports custom block size and device count
16. Manages device cleanup automatically
17. Yields list of BlockDevice instances

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

4. Loopback device testing: # Using test parametrization @pytest.mark.parametrize('block_size', [512, 1024]) def test_loopback(loopback_devices): # Uses block_size from parametrize, device_count=2 (default)

# Using fixture parametrization @pytest.mark.parametrize('loopback_devices', [{'device_count': 4, 'block_size': 1024}], indirect=True) def test_loopback_custom(loopback_devices): # Uses custom device_count and block_size # Devices are automatically cleaned up

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	FixtureRequest	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: pytest.FixtureRequest) -> 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	FixtureRequest	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: pytest.FixtureRequest
) -> 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	FixtureRequest	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: pytest.FixtureRequest
) -> 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	FixtureRequest	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: pytest.FixtureRequest) -> 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	FixtureRequest	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: pytest.FixtureRequest) -> 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

loopback_devices(request)

Create loopback devices for testing.

This fixture creates loopback devices with the specified block size from the parametrized test, yields them for testing, and ensures proper cleanup afterward.

Parameters (from test parametrization or fixture params): block_size: Block size for devices (default: 512) device_count: Number of devices to create (default: 2)

Yields:

Type	Description
list[BlockDevice]	List of BlockDevice instances representing the created loopback devices

Example

# Using test parametrization
@pytest.mark.parametrize('block_size', [512, 1024])
def test_loopback(loopback_devices):
    # Uses block_size from parametrize, device_count=2 (default)

# Using fixture parametrization
@pytest.mark.parametrize('loopback_devices', [{'device_count': 4, 'block_size': 1024}], indirect=True)
def test_loopback(loopback_devices):
    # Uses custom device_count and block_size

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

@pytest.fixture
def loopback_devices(request: pytest.FixtureRequest) -> Generator[list[BlockDevice], None, None]:
    """Create loopback devices for testing.

    This fixture creates loopback devices with the specified block size
    from the parametrized test, yields them for testing, and ensures
    proper cleanup afterward.

    Parameters (from test parametrization or fixture params):
        block_size: Block size for devices (default: 512)
        device_count: Number of devices to create (default: 2)

    Yields:
        List of BlockDevice instances representing the created loopback devices

    Example:
        ```python
        # Using test parametrization
        @pytest.mark.parametrize('block_size', [512, 1024])
        def test_loopback(loopback_devices):
            # Uses block_size from parametrize, device_count=2 (default)

        # Using fixture parametrization
        @pytest.mark.parametrize('loopback_devices', [{'device_count': 4, 'block_size': 1024}], indirect=True)
        def test_loopback(loopback_devices):
            # Uses custom device_count and block_size
        ```
    """
    # Get parameters with defaults
    if hasattr(request, 'param') and isinstance(request.param, dict):
        # Fixture is parametrized with dict
        device_count = request.param.get('device_count', 2)
        block_size = request.param.get('block_size', 512)
    else:
        # Get from test parametrization or use defaults
        device_count = getattr(request.node.callspec, 'params', {}).get('device_count', 2)
        block_size = getattr(request.node.callspec, 'params', {}).get('block_size', 512)

    # Create devices
    devices = create_loopback_devices(device_count, block_size=block_size)

    try:
        yield devices
    finally:
        # Clean up
        cleanup_loopback_devices(devices)

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()
        # Clean up backstore files
        for n in range(n_luns):
            Path(f'backstore{n}_file').unlink(missing_ok=True)