import asyncio import typing import lvmd.proto as lvmd import pytest from .testutils import SIZE_GIB, SIZE_TEST_POOL from .testutils.stratis import STRATIS_UNIT, PoolSelection, Stratisd from .testutils.systemd import StartMode, Systemd if typing.TYPE_CHECKING: from collections.abc import Awaitable @pytest.mark.parametrize('selection', list(PoolSelection)) async def test_lvservice_create_remove( stratisd: Stratisd, lvservice: lvmd.LVServiceStub, selection: PoolSelection, ) -> None: resp = await lvservice.create_l_v( name='test', size_bytes=2 * SIZE_GIB, device_class=selection.device_class, ) assert resp.volume.name == 'test' assert resp.volume.size_bytes == 2 * SIZE_GIB assert resp.volume.path == f'/dev/stratis/{selection.pool}/test' await stratisd.reload() pool = stratisd.pool(selection) assert len(pool.filesystems) == 1 fs = pool.filesystems[0] assert fs.name == 'test' assert fs.size == 2 * SIZE_GIB assert fs.size_limit == 2 * SIZE_GIB assert fs.devnode == resp.volume.path await lvservice.remove_l_v( name='test', device_class=selection.device_class, ) await stratisd.reload() pool = stratisd.pool(selection) assert not pool.filesystems @pytest.mark.parametrize('selection', list(PoolSelection)) async def test_lvservice_snapshot( stratisd: Stratisd, lvservice: lvmd.LVServiceStub, selection: PoolSelection, ) -> None: await lvservice.create_l_v( name='test', size_bytes=2 * SIZE_GIB, device_class=selection.device_class, ) resp = await lvservice.create_l_v_snapshot( name='test-2', device_class=selection.device_class, source_volume='test', size_bytes=4 * SIZE_GIB, access_type='rw', ) assert resp.snapshot.name == 'test-2' assert resp.snapshot.size_bytes == 4 * SIZE_GIB assert resp.snapshot.path == f'/dev/stratis/{selection.pool}/test-2' await stratisd.reload() pool = stratisd.pool(selection) assert len(pool.filesystems) == 2 pool.filesystems.sort(key=lambda fs: fs.name) fs = pool.filesystems[1] assert fs.name == 'test-2' assert fs.size == 2 * SIZE_GIB assert fs.size_limit == 4 * SIZE_GIB assert fs.devnode == resp.snapshot.path @pytest.mark.parametrize('selection', list(PoolSelection)) async def test_lvservice_resize( stratisd: Stratisd, lvservice: lvmd.LVServiceStub, selection: PoolSelection, ) -> None: await lvservice.create_l_v( name='test', size_bytes=2 * SIZE_GIB, device_class=selection.device_class, ) await lvservice.resize_l_v( name='test', device_class=selection.device_class, size_bytes=4 * SIZE_GIB, ) await stratisd.reload() pool = stratisd.pool(selection) assert len(pool.filesystems) == 1 fs = pool.filesystems[0] assert fs.name == 'test' assert fs.size == 2 * SIZE_GIB assert fs.size_limit == 4 * SIZE_GIB @pytest.mark.parametrize( ('selection', 'restart'), [(selection, False) for selection in PoolSelection] + [(PoolSelection.DEFAULT, True)], ) async def test_vgservice_list( lvservice: lvmd.LVServiceStub, vgservice: lvmd.VGServiceStub, systemd: Systemd, selection: PoolSelection, restart: bool, ) -> None: resp = await vgservice.get_l_v_list(device_class=selection.device_class) assert not resp.volumes await lvservice.create_l_v( name='test-1', size_bytes=SIZE_GIB, device_class=selection.device_class, ) await lvservice.create_l_v( name='test-2', size_bytes=2 * SIZE_GIB, device_class=selection.device_class, ) if restart: await systemd.manager.wait_for_job_removal( systemd.manager.restart_unit(STRATIS_UNIT, StartMode.REPLACE) ) resp = await vgservice.get_l_v_list(device_class=selection.device_class) assert len(resp.volumes) == 2 resp.volumes.sort(key=lambda lv: lv.name) lv = resp.volumes[0] assert lv.name == 'test-1' assert not lv.tags assert lv.size_bytes == SIZE_GIB assert lv.path == f'/dev/stratis/{selection.pool}/test-1' lv = resp.volumes[1] assert lv.name == 'test-2' assert not lv.tags assert lv.size_bytes == 2 * SIZE_GIB assert lv.path == f'/dev/stratis/{selection.pool}/test-2' @pytest.mark.parametrize( ('selection', 'restart'), [(selection, False) for selection in PoolSelection] + [(PoolSelection.DEFAULT, True)], ) async def test_vgservice_free_bytes( lvservice: lvmd.LVServiceStub, vgservice: lvmd.VGServiceStub, systemd: Systemd, selection: PoolSelection, restart: bool, ) -> None: resp = await vgservice.get_free_bytes(device_class=selection.device_class) assert resp.free_bytes == selection.initial_free_bytes await lvservice.create_l_v( name='test-1', size_bytes=SIZE_GIB, device_class=selection.device_class, ) await lvservice.create_l_v( name='test-2', size_bytes=2 * SIZE_GIB, device_class=selection.device_class, ) if restart: await systemd.manager.wait_for_job_removal( systemd.manager.restart_unit(STRATIS_UNIT, StartMode.REPLACE) ) resp = await vgservice.get_free_bytes(device_class=selection.device_class) assert resp.free_bytes == selection.initial_free_bytes - 3 * SIZE_GIB async def test_vgservice_watch( lvservice: lvmd.LVServiceStub, vgservice: lvmd.VGServiceStub, systemd: Systemd, ) -> None: ORDERED_POOLS = [PoolSelection.PRIMARY, PoolSelection.SECONDARY] expected_free: dict[PoolSelection, int] = {} for selection in ORDERED_POOLS: expected_free[selection] = selection.initial_free_bytes pending_response: Awaitable[lvmd.WatchResponse] | None = None async def check_response(target: PoolSelection | None) -> None: nonlocal pending_response if pending_response is None: pending_response = anext(events) resp = await pending_response # We might get some redundant events while things are being changed, # so debounce them. while True: pending_response = asyncio.create_task(anext(events)) try: resp = await asyncio.wait_for(asyncio.shield(pending_response), 0.4) except TimeoutError: break else: pass resp.items.sort(key=lambda item: item.device_class) for i, selection in enumerate(ORDERED_POOLS): assert resp.items[i].device_class == selection.device_class, target assert resp.items[i].free_bytes == expected_free[selection], target assert resp.items[i].size_bytes == SIZE_TEST_POOL, target assert resp.free_bytes == expected_free[PoolSelection.PRIMARY] try: async with asyncio.timeout(10): events = vgservice.watch() await check_response(None) for selection in ORDERED_POOLS: await lvservice.create_l_v( name='test', size_bytes=2 * SIZE_GIB, device_class=selection.device_class, ) expected_free[selection] -= 2 * SIZE_GIB await check_response(selection) for selection in ORDERED_POOLS: await lvservice.resize_l_v( name='test', size_bytes=4 * SIZE_GIB, device_class=selection.device_class, ) expected_free[selection] -= 2 * SIZE_GIB await check_response(selection) await systemd.manager.wait_for_job_removal( systemd.manager.restart_unit(STRATIS_UNIT, StartMode.REPLACE) ) await check_response(None) for selection in ORDERED_POOLS: await lvservice.remove_l_v( name='test', device_class=selection.device_class, ) expected_free[selection] += 4 * SIZE_GIB await check_response(selection) finally: if pending_response is not None: # XXX: pretty sure this is a false positive? pending_response.cancel() # pyright: ignore[reportUnreachable]