Something went wrong. Try again.
This repository has no description
Something went wrong. Try again.
5.1 kB · 149 lines
Python
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150import asyncio
from cloudcoil.apimachinery import ObjectMeta, Quantityfrom cloudcoil.client import Config
import cloudcoil.models.kubernetes as k8simport cloudcoil.models.volumesnapshot as k8s_vsimport pytest
from .testutils import SIZE_GIBfrom .testutils.k8s import TEST_NAMESPACE, pod_logs, wait_for_pod_successfrom .testutils.stratis import PoolSelection
async def run_pod(*, config: Config, pvc: str, filename: str, size_mi: int) -> None: script = f""" while true; do free=$(df -B1M --output=avail /test | tail -1) echo "$free" [[ "$free" -lt {size_mi} ]] || break sleep 1 done dd if=/dev/zero of=/test/{filename} bs=1M count={size_mi} echo ok. """
pod = await k8s.core.v1.Pod( metadata=ObjectMeta(name='fill'), spec=k8s.core.v1.PodSpec( restart_policy='Never', containers=[ k8s.core.v1.Container( name='fill', command=['bash', '-exc', script], image='debian:trixie-slim', volume_mounts=[ k8s.core.v1.VolumeMount(name='test', mount_path='/test'), ], ), ], volumes=[ k8s.core.v1.Volume( name='test', persistent_volume_claim=k8s.core.v1.PersistentVolumeClaimVolumeSource( claim_name=pvc, ), ) ], ), ).async_create()
pod = await wait_for_pod_success(pod)
logs = await pod_logs(config, pod) assert logs.endswith('ok.\n')
await pod.async_remove()
@pytest.mark.parametrize('selection', [PoolSelection.PRIMARY, PoolSelection.SECONDARY])async def test_e2e(k8s_config: Config, selection: PoolSelection) -> None: async with asyncio.timeout(120): pvc_name = f'test-{selection.device_class}' snap_name = f'test-snap-{selection.device_class}'
await k8s.core.v1.Namespace( metadata=ObjectMeta(name=TEST_NAMESPACE), ).async_create()
pvc = await k8s.core.v1.PersistentVolumeClaim( metadata=ObjectMeta(name=pvc_name), spec=k8s.core.v1.PersistentVolumeClaimSpec( access_modes=['ReadWriteOnce'], resources=k8s.core.v1.VolumeResourceRequirements( requests={'storage': Quantity('2Gi')}, ), storage_class_name=selection.provisioner, ), ).async_create()
await run_pod( config=k8s_config, pvc=pvc_name, filename='bigfile', size_mi=1900, )
pvc = await pvc.async_fetch() assert pvc.spec is not None assert pvc.spec.resources is not None pvc.spec.resources.requests = {'storage': Quantity('3Gi')} pvc = await pvc.async_save()
await run_pod( config=k8s_config, pvc=pvc_name, filename='bigfile.2', size_mi=1000, )
await k8s_vs.snapshot.v1.VolumeSnapshotClass( metadata=ObjectMeta(name=selection.provisioner), driver='topolvm.io', deletion_policy='Delete', ).async_create()
await k8s_vs.snapshot.v1.VolumeSnapshot( metadata=ObjectMeta(name=snap_name), spec=k8s_vs.snapshot.v1.VolumeSnapshotSpec( source=k8s_vs.snapshot.v1.SourceModel( persistent_volume_claim_name=pvc_name, ), volume_snapshot_class_name=selection.provisioner, ), ).async_create()
pvc = await k8s.core.v1.PersistentVolumeClaim( metadata=ObjectMeta(name=snap_name), spec=k8s.core.v1.PersistentVolumeClaimSpec( access_modes=['ReadWriteOnce'], data_source=k8s.core.v1.TypedLocalObjectReference( name=snap_name, kind='VolumeSnapshot', api_group='snapshot.storage.k8s.io', ), resources=k8s.core.v1.VolumeResourceRequirements( requests={'storage': Quantity('4Gi')}, ), storage_class_name=selection.provisioner, ), ).async_create()
await run_pod( config=k8s_config, pvc=snap_name, filename='bigfile.3', size_mi=1000, )
node = await k8s.core.v1.Node.async_get('lima-topolvm-stratis-test') # XXX: pyright says this is Unknown because it doesn't like the # 'from cloudcoil import apimachinery' # noqa: ERA001 assert isinstance(node.metadata, ObjectMeta) # pyright: ignore[reportUnknownMemberType] assert node.metadata.annotations is not None
free_bytes = int( node.metadata.annotations[f'capacity.topolvm.io/{selection.device_class}'] ) assert free_bytes == selection.initial_free_bytes - (4 + 3 + 3) * SIZE_GIB