Something went wrong. Try again.
This repository has no description
Something went wrong. Try again.
1.6 kB · 53 lines
Python
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354from pathlib import Path
from cloudcoil.apimachinery import ObjectMetafrom cloudcoil.client import Configfrom cloudcoil.resources import Unstructured
import cloudcoil.models.kubernetes as k8simport pytest
KUBECONFIG = Path(__file__).parent.parent / 'vm' / '_state' / 'kubeconfig.yaml'TEST_NAMESPACE = 'test'
async def pod_logs( config: Config, pod: k8s.core.v1.Pod, *, container: str | None = None,) -> str: # XXX: pyright says this is Unknown because it doesn't like the # 'from cloudcoil import apimachinery' # noqa: ERA001 assert isinstance(pod.metadata, ObjectMeta) # pyright: ignore[reportUnknownMemberType]
assert pod.metadata.namespace is not None assert pod.metadata.name is not None
if container is None: assert pod.spec is not None assert len(pod.spec.containers) == 1 container = pod.spec.containers[0].name
resp = await config.async_client.get( f'/api/v1/namespaces/{pod.metadata.namespace}' + f'/pods/{pod.metadata.name}/log?container={container}' ) resp.raise_for_status() return resp.text
async def wait_for_pod_success(pod: k8s.core.v1.Pod) -> k8s.core.v1.Pod: async for _, obj in await pod.async_watch(): if isinstance(obj, Unstructured): continue
if (status := obj.status) is None: continue
if status.phase == 'Succeeded': return obj elif status.phase == 'Failed': pytest.fail('pod failed') else: pytest.fail('never got final pod event')