Something went wrong. Try again.
This repository has no description
Something went wrong. Try again.
8.5 kB · 280 lines
Python
at ci
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281import asyncioimport typing
import lvmd.proto as lvmdimport pytest
from .testutils import SIZE_GIB, SIZE_TEST_POOLfrom .testutils.stratis import STRATIS_UNIT, PoolSelection, Stratisdfrom .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]