From 9617a055a62391a683e6d7a367f1495ef22e172d Mon Sep 17 00:00:00 2001 From: Codemageddon Date: Wed, 28 Oct 2026 00:05:59 +0100 Subject: [PATCH 1/2] Tolerate objects that do not fit the model in list() and watch() Add invalid=InvalidObjects.INCLUDE to Api.list() and Api.watch(). An object that fails validation comes back as an InvalidObject (raw data, the ValidationError, parsed metadata): in ListWithInvalid.invalid for list(), or as an InvalidWatchEvent for watch(), which keeps streaming. Only ADDED/MODIFIED/DELETED objects whose metadata parses with a name are tolerated; ERROR events, unparseable bookmarks and corrupt lines still raise. Overloads keep the return types of existing calls unchanged. Add WatchEvent.is_initial_events_end for the bookmark that ends a send_initial_events watch. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 4 + CLAUDE.md | 10 +- docs/operations/crud.md | 16 ++ docs/operations/watch.md | 50 +++- docs/reference/core.md | 4 + kubex/api/api.py | 145 ++++++++++- kubex/core/invalid_object.py | 123 +++++++++ kubex/core/params.py | 13 + kubex/core/watch_event.py | 71 ++++++ test/e2e/test_invalid_objects.py | 220 ++++++++++++++++ test/test_invalid_objects.py | 414 +++++++++++++++++++++++++++++++ 11 files changed, 1056 insertions(+), 14 deletions(-) create mode 100644 kubex/core/invalid_object.py create mode 100644 test/e2e/test_invalid_objects.py create mode 100644 test/test_invalid_objects.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 3a921140..0f6b6c49 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -21,6 +21,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - `kubex_core.models.types`: `Time`, `MicroTime`, `Quantity`, `NullAsEmptyList`, `NullAsEmptyDict`. - `ObjectMetadata.managed_fields`/`self_link` and `ListMetadata.shard_info`. +- `invalid=InvalidObjects.INCLUDE` for `Api.list()` and `Api.watch()`: objects that do not fit + the model come back as `InvalidObject`s (in `ListWithInvalid.invalid`, or as + `InvalidWatchEvent`s) instead of raising. +- `WatchEvent.is_initial_events_end`. ### Fixed diff --git a/CLAUDE.md b/CLAUDE.md index 738c959c..840fb974 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -99,13 +99,16 @@ kubex/ # Main package — PEP 420 namespace package (no ├── response.py # Response dataclass + HeadersWrapper ├── params.py # API option classes (ListOptions, GetOptions, DeleteOptions, ExecOptions, AttachOptions, PortForwardOptions, etc.) # + Timeout, TimeoutTypes — HTTP timeout configuration + # + InvalidObjects enum (RAISE/INCLUDE) for list()/watch(invalid=) + ├── invalid_object.py # InvalidObject (raw + ValidationError + parsed ObjectMetadata) and ListWithInvalid(ListEntity) + # — what list()/watch(invalid=InvalidObjects.INCLUDE) return for objects that do not fit the model ├── json_patch.py # JSON Patch RFC 6902 operation models (JsonPatchAdd, etc.) + JsonPatch RootModel ├── json_pointer.py # JSON Pointer RFC 6901 implementation (JsonPointer custom str type) ├── patch.py # Patch protocol + ApplyPatch, MergePatch, StrategicMergePatch # Re-exports JsonPatch models and JsonPointer from json_patch/json_pointer ├── subresource.py # Subresource definitions ├── exec_channels.py # WebSocket channel protocol (V5ChannelProtocol, channel ID constants, select_protocol()) + portforward helpers (data_channel_for_port_index, error_channel_for_port_index, port_prefix_encode/decode) - ├── watch_event.py # WatchEvent[ResourceType] and EventType enum + ├── watch_event.py # WatchEvent[ResourceType] (+ is_initial_events_end), InvalidWatchEvent and EventType enum └── request_builder/ # Constructs HTTP requests from API calls ├── _paths.py # path_segment()/object_url()/collection_url() — the only way builders build URLs: # validates (no empty/None name — that would hit the collection URL) and percent-encodes names/namespaces @@ -172,6 +175,7 @@ test/ # Test suite │ ├── test_portforward.py # E2E tests for Pod portforward subresource (forward() + listen() against K3S) │ ├── test_bearer_token_auth.py # E2E tests for bearer-token auth (no proxy): valid SA token succeeds, invalid token returns 401 │ ├── test_api_safety.py # E2E: empty-name rejection, dry-run delete, escaped names, server-side apply +│ ├── test_invalid_objects.py # E2E: list()/watch(invalid=INCLUDE) with a CRD object the model rejects; initial-events-end bookmark │ └── proxy/ # E2E tests for trust_env proxy support (Squid on shared Docker network) │ ├── _constants.py # SQUID_PROXY_USER / SQUID_PROXY_PASSWORD shared by conftest and tests │ ├── conftest.py # Fixtures: Docker network, K3S-in-network, Squid, proxy_url, proxy config, @@ -190,6 +194,7 @@ test/ # Test suite │ └── test_refreshable_token.py # Token refresh and async lock tests ├── test_error_handling.py # handle_request_error() and exception hierarchy tests ├── test_metadata_accessor.py # MetadataAccessor (get, list, patch) tests +├── test_invalid_objects.py # list()/watch(invalid=) parsing rules, is_initial_events_end, overload types (assert_type) ├── test_api_request_validation.py # Api-level name/namespace validation (nothing reaches the client) ├── test_models/ # Unit tests for kubex-core models, field types and generated-model behaviour │ └── test_eviction.py # Eviction model tests @@ -300,6 +305,9 @@ All models inherit from `BaseK8sModel` which uses `alias_generator=to_camel`, `p ### Configuration auto-loading `create_client()` → tries kubeconfig file first → falls back to in-cluster pod environment. +### Objects that do not fit the model +`Api.list()` and `Api.watch()` take `invalid: InvalidObjectsTypes` (`InvalidObjects` enum or `"raise"`/`"include"`); overloads pick the return type. `RAISE` (default) raises pydantic's `ValidationError`. `INCLUDE` returns `ListWithInvalid` (`items` + `invalid: list[InvalidObject]`) or yields `InvalidWatchEvent` and keeps streaming. Only ADDED/MODIFIED/DELETED objects are tolerated, and only when their metadata parses with a name; ERROR events, unparseable bookmarks and non-JSON lines still raise. `MetadataAccessor` has no such option (`PartialObjectMetadata` has only optional fields). The watch loop parses before `yield`, so the yield is never inside an `except`. + ### Ellipsis sentinel for optional overrides `Ellipsis` (`...`) is used as a sentinel to distinguish "not provided" (use the default) from `None` (explicitly disabled). This pattern is used in two places: - **Namespace**: `...` = use the `Api` instance default; `None` = explicitly no namespace. diff --git a/docs/operations/crud.md b/docs/operations/crud.md index b6507f43..5e5e480e 100644 --- a/docs/operations/crud.md +++ b/docs/operations/crud.md @@ -60,6 +60,22 @@ Pass `namespace=None` to list across all namespaces even when the `Api` instance all_pods = await api.list(namespace=None) ``` +### Objects that do not fit the model + +By default, `list()` raises pydantic's `ValidationError` if any item does not fit the model. Pass `invalid=InvalidObjects.INCLUDE` (or the string `"include"`) to get a `ListWithInvalid` instead: `items` holds the objects that fit, and `invalid` the others, each an `InvalidObject` with `raw`, `error` and `metadata`: + +```python +from kubex.core.params import InvalidObjects + +pods = await api.list(invalid=InvalidObjects.INCLUDE) +for pod in pods.items: + print(pod.metadata.name) +for bad in pods.invalid: + print(f"{bad.metadata.name} does not fit the model: {bad.error}") +``` + +`ListWithInvalid` is a `ListEntity`, so `metadata.resource_version` and `metadata.continue_` work as usual; `invalid` is not serialized. The list itself is still checked strictly, and an item without a readable `metadata.name` still raises. See [Objects that do not fit the model](watch.md#objects-that-do-not-fit-the-model) for when this happens and what other calls do with such objects. + ## create Create a new resource. The returned object is the server-assigned representation (including `metadata.name` when `generate_name` was used): diff --git a/docs/operations/watch.md b/docs/operations/watch.md index 46352583..07cab583 100644 --- a/docs/operations/watch.md +++ b/docs/operations/watch.md @@ -24,6 +24,7 @@ Each yielded value is a `WatchEvent[ResourceType]`: |---|---|---| | `event.type` | `EventType` | `ADDED`, `MODIFIED`, `DELETED`, or `BOOKMARK` | | `event.object` | `ResourceType | Bookmark` | Fully parsed resource (or `Bookmark` for bookmark events) | +| `event.is_initial_events_end` | `bool` | `True` only for the bookmark that ends the initial events (see [`send_initial_events` pattern](#send_initial_events-pattern)) | `ERROR` events are never yielded: the generator raises the matching exception instead — most importantly `Gone` (HTTP 410) when the requested `resourceVersion` is too old. See [Resuming after `Gone`](#resuming-after-gone). @@ -75,17 +76,22 @@ async for event in api.watch(namespace=None): `send_initial_events=True` makes the stream start with synthetic `ADDED` events for every existing resource, then a `BOOKMARK` annotated with `k8s.io/initial-events-end: "true"`, then live updates — current state and future changes in one stream. The server requires bookmarks for this, so `allow_bookmarks` defaults to `True` when `send_initial_events=True` (passing `allow_bookmarks=False` together with it raises `ValueError`). It needs the API server's `WatchList` feature; a server without it rejects the request with `422`. +`event.is_initial_events_end` is `True` only for that bookmark. Periodic checkpoint bookmarks look the same apart from the annotation, so they return `False`: + ```python async for event in api.watch(send_initial_events=True): - if event.type == EventType.BOOKMARK: - annotations = event.object.metadata.annotations or {} - if annotations.get("k8s.io/initial-events-end") == "true": - print("initial state complete, watching for changes now") - continue - print(event.type, event.object.metadata.name) + if event.is_initial_events_end: + print("initial state complete, watching for changes now") + elif event.type != EventType.BOOKMARK: + print(event.type, event.object.metadata.name) ``` -Not every bookmark marks the end of the initial events — periodic checkpoint bookmarks look the same apart from the annotation. +### Where to resume from + +The initial `ADDED` events carry each object's own `resourceVersion`, in no particular order, so none of them is a point to resume from. The end bookmark's `resourceVersion` is the first one: + +- If the stream ends before the end bookmark, discard what you collected and start again with `send_initial_events=True`. +- After it, resume with `watch(resource_version=...)` from the last `resourceVersion` you saw (an event's or a bookmark's), as in [Resuming after `Gone`](#resuming-after-gone). ## Resuming after `Gone` @@ -123,6 +129,36 @@ async def watch_forever(api: Api[Pod]) -> None: Events between the list and the watch are not lost: watching from the list's `resourceVersion` delivers everything that changed after it. Changes that happened while the watcher was disconnected for longer than the server keeps history are covered by the re-list, so `handle_existing` should reconcile (update or create local state, and drop objects that no longer exist) rather than assume a clean slate. +## Objects that do not fit the model + +By default, an object the model cannot parse raises pydantic's `ValidationError`, which ends the stream. This mostly happens with custom resources: the API server does not re-validate objects stored before a CRD schema change, so an old object can break a model written for the current schema. Reconnecting does not help, because the same object comes back. + +Pass `invalid=InvalidObjects.INCLUDE` (or the string `"include"`) to keep going. Such an object then arrives as an `InvalidWatchEvent` instead of a `WatchEvent`: + +| Attribute | Type | Description | +|---|---|---| +| `event.type` | `EventType` | `ADDED`, `MODIFIED` or `DELETED` | +| `event.object.raw` | `dict[str, Any]` | The object as the server sent it | +| `event.object.error` | `ValidationError` | Why it does not fit the model | +| `event.object.metadata` | `ObjectMetadata` | Its metadata; always has a `name` | + +```python +from kubex.core.params import InvalidObjects +from kubex.core.watch_event import InvalidWatchEvent + +async for event in api.watch(invalid=InvalidObjects.INCLUDE): + if isinstance(event, InvalidWatchEvent): + print(f"skipping {event.object.metadata.name}: {event.object.error}") + continue + handle(event) +``` + +Both kinds of event have `type` and `object.metadata`, so code that only needs an object's name, namespace or `resourceVersion` (keeping a resume point, or dropping a deleted object from a cache) works on both without the `isinstance` check. + +Only `ADDED`, `MODIFIED` and `DELETED` objects are tolerated. These still raise, because they mean the stream itself is broken: an `ERROR` event, a bookmark that does not parse, an object without a readable `metadata.name`, and a line that is not JSON. + +`list()` takes the same option; see [Objects that do not fit the model](crud.md#objects-that-do-not-fit-the-model) for lists. Other calls still parse the object with the model: `get()` raises `ValidationError` for it, and so do `delete()`, `patch()` and `replace()`, after the server has carried out the change. Use [`api.metadata`](../subresources/metadata.md), which only parses metadata (for example, to remove a finalizer), or a looser model for those. + ## Server-side timeout Pass `timeout_seconds=` to set a server-side timeout on the watch call (sent as the Kubernetes `timeoutSeconds` query parameter). The server closes the stream cleanly after this many seconds and the `async for` loop simply ends; reconnect as shown above: diff --git a/docs/reference/core.md b/docs/reference/core.md index 8a8acba9..9099092a 100644 --- a/docs/reference/core.md +++ b/docs/reference/core.md @@ -20,6 +20,10 @@ Auto-generated reference for the `kubex.core` module. ::: kubex.core.watch_event +## Objects that do not fit the model + +::: kubex.core.invalid_object + ## Patch types ::: kubex.core.patch diff --git a/kubex/api/api.py b/kubex/api/api.py index c8cfb4a5..2a2b64fb 100644 --- a/kubex/api/api.py +++ b/kubex/api/api.py @@ -4,17 +4,22 @@ from typing import ( AsyncGenerator, Generic, + Literal, Type, + overload, ) from pydantic import ValidationError from kubex.client.client import BaseClient, create_client +from kubex.core.invalid_object import ListWithInvalid, _validate_list_with_invalid from kubex.core.params import ( DeleteOptions, DryRunTypes, FieldValidationTypes, GetOptions, + InvalidObjects, + InvalidObjectsTypes, ListOptions, PatchOptions, PostOptions, @@ -26,7 +31,7 @@ ) from kubex.core.patch import Patch from kubex.core.request_builder.builder import RequestBuilder -from kubex.core.watch_event import WatchEvent +from kubex.core.watch_event import InvalidWatchEvent, WatchEvent, _parse_watch_event from kubex_core.models.list_entity import ListEntity from kubex_core.models.resource_config import Scope from kubex_core.models.status import Status @@ -165,6 +170,7 @@ async def get( response = await self._client.request(request) return self._resource.model_validate_json(response.content) + @overload async def list( self, *, @@ -177,6 +183,54 @@ async def list( version_match: VersionMatchTypes = None, resource_version: ResourceVersionTypes = None, request_timeout: ApiRequestTimeoutTypes = Ellipsis, + invalid: Literal[InvalidObjects.RAISE, "raise"] = InvalidObjects.RAISE, + ) -> ListEntity[ResourceType]: ... + + @overload + async def list( + self, + *, + namespace: ApiNamespaceTypes = Ellipsis, + label_selector: str | None = None, + field_selector: str | None = None, + timeout_seconds: int | None = None, + limit: int | None = None, + continue_token: str | None = None, + version_match: VersionMatchTypes = None, + resource_version: ResourceVersionTypes = None, + request_timeout: ApiRequestTimeoutTypes = Ellipsis, + invalid: Literal[InvalidObjects.INCLUDE, "include"], + ) -> ListWithInvalid[ResourceType]: ... + + @overload + async def list( + self, + *, + namespace: ApiNamespaceTypes = Ellipsis, + label_selector: str | None = None, + field_selector: str | None = None, + timeout_seconds: int | None = None, + limit: int | None = None, + continue_token: str | None = None, + version_match: VersionMatchTypes = None, + resource_version: ResourceVersionTypes = None, + request_timeout: ApiRequestTimeoutTypes = Ellipsis, + invalid: InvalidObjectsTypes, + ) -> ListEntity[ResourceType]: ... + + async def list( + self, + *, + namespace: ApiNamespaceTypes = Ellipsis, + label_selector: str | None = None, + field_selector: str | None = None, + timeout_seconds: int | None = None, + limit: int | None = None, + continue_token: str | None = None, + version_match: VersionMatchTypes = None, + resource_version: ResourceVersionTypes = None, + request_timeout: ApiRequestTimeoutTypes = Ellipsis, + invalid: InvalidObjectsTypes = InvalidObjects.RAISE, ) -> ListEntity[ResourceType]: """List objects of kind. @@ -205,15 +259,25 @@ async def list( request_timeout: HTTP-level timeout override for this call. A number is interpreted as the total timeout in seconds. Pass ``None`` to disable timeouts entirely for this call. Omit to use the client default. + invalid: What to do with an object that does not fit the model. + ``InvalidObjects.RAISE`` (the default) raises pydantic's + ``ValidationError``. ``InvalidObjects.INCLUDE`` returns a + ``ListWithInvalid`` that keeps such objects in ``invalid``, each an + ``InvalidObject`` with its raw data, the error and its metadata. Returns: ListEntity[ResourceType]: the list of resources, with the list's - ``metadata`` (``resource_version``, ``continue_``). + ``metadata`` (``resource_version``, ``continue_``). A + ``ListWithInvalid`` with ``invalid=InvalidObjects.INCLUDE``. Raises: ValueError: an argument is invalid, e.g. a namespace given for a cluster-scoped resource, or an unknown option value (e.g. for ``version_match``). + ValidationError: an object does not fit the model (unless + ``invalid=InvalidObjects.INCLUDE``), or the response is not a + valid list; with ``INCLUDE``, also an object without a readable + ``metadata.name``. KubexApiError: the API server rejected the request; the subclass matches the HTTP status (e.g. ``Forbidden``, ``UnprocessableEntity``). KubexConnectionError: the API server could not be reached. @@ -221,6 +285,7 @@ async def list( ConfigurationError: credentials could not be obtained, e.g. an exec plugin or OIDC token refresh failed. """ + invalid_objects = InvalidObjects(invalid) _namespace = ensure_optional_namespace( namespace, self._namespace, self._resource.__RESOURCE_CONFIG__.scope ) @@ -238,6 +303,10 @@ async def list( ) response = await self._client.request(request) list_model = self._resource.__RESOURCE_CONFIG__.list_model + if invalid_objects is InvalidObjects.INCLUDE: + return _validate_list_with_invalid( + self._resource, list_model, response.content + ) return list_model.model_validate_json(response.content) async def create( @@ -602,6 +671,51 @@ async def replace( response = await self._client.request(request) return self._resource.model_validate_json(response.content) + @overload + def watch( + self, + *, + namespace: ApiNamespaceTypes = Ellipsis, + label_selector: str | None = None, + field_selector: str | None = None, + allow_bookmarks: bool | None = None, + send_initial_events: bool | None = None, + resource_version: ResourceVersionTypes = None, + timeout_seconds: int | None = None, + request_timeout: ApiRequestTimeoutTypes = Ellipsis, + invalid: Literal[InvalidObjects.RAISE, "raise"] = InvalidObjects.RAISE, + ) -> AsyncGenerator[WatchEvent[ResourceType], None]: ... + + @overload + def watch( + self, + *, + namespace: ApiNamespaceTypes = Ellipsis, + label_selector: str | None = None, + field_selector: str | None = None, + allow_bookmarks: bool | None = None, + send_initial_events: bool | None = None, + resource_version: ResourceVersionTypes = None, + timeout_seconds: int | None = None, + request_timeout: ApiRequestTimeoutTypes = Ellipsis, + invalid: Literal[InvalidObjects.INCLUDE, "include"], + ) -> AsyncGenerator[WatchEvent[ResourceType] | InvalidWatchEvent, None]: ... + + @overload + def watch( + self, + *, + namespace: ApiNamespaceTypes = Ellipsis, + label_selector: str | None = None, + field_selector: str | None = None, + allow_bookmarks: bool | None = None, + send_initial_events: bool | None = None, + resource_version: ResourceVersionTypes = None, + timeout_seconds: int | None = None, + request_timeout: ApiRequestTimeoutTypes = Ellipsis, + invalid: InvalidObjectsTypes, + ) -> AsyncGenerator[WatchEvent[ResourceType] | InvalidWatchEvent, None]: ... + async def watch( self, *, @@ -613,7 +727,8 @@ async def watch( resource_version: ResourceVersionTypes = None, timeout_seconds: int | None = None, request_timeout: ApiRequestTimeoutTypes = Ellipsis, - ) -> AsyncGenerator[WatchEvent[ResourceType], None]: + invalid: InvalidObjectsTypes = InvalidObjects.RAISE, + ) -> AsyncGenerator[WatchEvent[ResourceType] | InvalidWatchEvent, None]: """Watch for changes to the specified resource. Args: @@ -631,7 +746,8 @@ async def watch( ``send_initial_events=True`` raises ``ValueError``. send_initial_events: Start the stream with synthetic ``ADDED`` events for the current state, followed by a bookmark annotated with - ``k8s.io/initial-events-end``. + ``k8s.io/initial-events-end`` (see + ``WatchEvent.is_initial_events_end``). resource_version: Only deliver changes made after this resource version, e.g. the list ``metadata.resource_version`` from a previous ``list()``. If not provided, the stream starts with synthetic @@ -645,17 +761,28 @@ async def watch( timeouts entirely for this call. Omit to use the client's default for streams, which does not cut off an idle watch. An explicit short read/total timeout will terminate a quiet stream. + invalid: What to do with an ``ADDED``, ``MODIFIED`` or ``DELETED`` + object that does not fit the model. ``InvalidObjects.RAISE`` (the + default) raises pydantic's ``ValidationError``, which ends the + stream. ``InvalidObjects.INCLUDE`` yields an ``InvalidWatchEvent`` + for it and the stream goes on. Yields: WatchEvent[ResourceType]: one event per change, with its ``type`` (``ADDED``, ``MODIFIED``, ``DELETED`` or ``BOOKMARK``) and ``object`` (the resource, or a ``Bookmark`` for ``BOOKMARK`` - events). ``ERROR`` events are raised as exceptions instead. + events). ``ERROR`` events are raised as exceptions instead. With + ``invalid=InvalidObjects.INCLUDE``, an ``InvalidWatchEvent`` for + each object that does not fit the model. Raises: ValueError: an argument is invalid, e.g. a namespace given for a cluster-scoped resource, or ``allow_bookmarks=False`` together with ``send_initial_events=True``. + ValidationError: an object does not fit the model (unless + ``invalid=InvalidObjects.INCLUDE``), or a bookmark does not parse; + with ``INCLUDE``, also an object without a readable + ``metadata.name``. Gone: the requested ``resource_version`` is too old (HTTP 410), either on connect or as an ``ERROR`` event mid-stream; re-list and watch again from the new ``resourceVersion``. @@ -668,6 +795,7 @@ async def watch( ConfigurationError: credentials could not be obtained, e.g. an exec plugin or OIDC token refresh failed. """ + invalid_objects = InvalidObjects(invalid) _namespace = ensure_optional_namespace( namespace, self._namespace, self._resource.__RESOURCE_CONFIG__.scope ) @@ -685,7 +813,12 @@ async def watch( request_timeout=request_timeout, ) async for line in self._client.stream_lines(request): - yield WatchEvent(self._resource, json.loads(line)) + # Parse before yielding: the yield stays outside any ``except``, so an + # exception thrown into the generator is never mistaken for ours. + event = _parse_watch_event( + self._resource, json.loads(line), invalid_objects + ) + yield event async def create_api( diff --git a/kubex/core/invalid_object.py b/kubex/core/invalid_object.py new file mode 100644 index 00000000..387fc762 --- /dev/null +++ b/kubex/core/invalid_object.py @@ -0,0 +1,123 @@ +"""Objects that do not fit their resource model. + +``Api.list()`` and ``Api.watch()`` raise pydantic's ``ValidationError`` when an +object from the API server does not fit the model, for example a custom resource +stored before its CRD schema changed. With ``invalid=InvalidObjects.INCLUDE`` they +keep such an object as an ``InvalidObject`` instead, so one bad object neither +fails a whole list nor ends a watch. +""" + +from __future__ import annotations + +import json +from dataclasses import dataclass +from typing import Any, Generic, Type, cast + +from pydantic import Field, GetCoreSchemaHandler, ValidationError +from pydantic_core import core_schema + +from kubex_core.models.list_entity import ListEntity +from kubex_core.models.metadata import ObjectMetadata +from kubex_core.models.typing import ResourceType + + +@dataclass(frozen=True, eq=False) +class InvalidObject: + """An object from the API server that does not fit the resource model. + + Attributes: + raw: The object as the server sent it. + error: Why it does not fit the model. + metadata: The object's metadata. It always parsed and always has a + ``name``, so the object can still be identified, e.g. to drop it from + a cache on a ``DELETED`` event. + """ + + raw: dict[str, Any] + error: ValidationError + metadata: ObjectMetadata + + def __repr__(self) -> str: + return ( + f"InvalidObject(kind={self.raw.get('kind')!r}, " + f"namespace={self.metadata.namespace!r}, name={self.metadata.name!r}, " + f"errors={self.error.error_count()})" + ) + + @classmethod + def __get_pydantic_core_schema__( + cls, source: Any, handler: GetCoreSchemaHandler + ) -> core_schema.CoreSchema: + # Lets ``ListWithInvalid`` hold these in a field; never parsed from JSON. + return core_schema.is_instance_schema(cls) + + +class ListWithInvalid(ListEntity[ResourceType], Generic[ResourceType]): + """A list whose objects that do not fit the model are kept apart. + + Returned by ``Api.list(invalid=InvalidObjects.INCLUDE)``. ``items`` holds the + objects that fit the model, in server order; ``invalid`` holds the rest. The + list's ``metadata`` (``resource_version``, ``continue_``) is the server's, so + paging and watching from the list work as usual. ``invalid`` is not serialized. + """ + + invalid: list[InvalidObject] = Field(default_factory=list, exclude=True) + + +def _invalid_object_or_raise(raw: Any, error: ValidationError) -> InvalidObject: + """Wrap ``raw``, which failed validation with ``error``, or re-raise ``error``. + + An object is only tolerated when its metadata parses and has a name: without + one the caller cannot tell which object it is, which means the response + itself is broken rather than one object in it. + """ + if isinstance(raw, dict): + try: + metadata = ObjectMetadata.model_validate(raw.get("metadata")) + except ValidationError: + pass + else: + if metadata.name: + return InvalidObject(raw=raw, error=error, metadata=metadata) + raise error + + +def _validate_list_with_invalid( + resource_type: Type[ResourceType], + list_model: Type[ListEntity[ResourceType]], + content: bytes, +) -> ListWithInvalid[ResourceType]: + """Parse a list response, keeping items that do not fit ``resource_type`` apart. + + The list itself (``apiVersion``, ``kind``, ``metadata``) is validated as + strictly as by ``list_model``; only its items are tolerated. + """ + raw = json.loads(content) + raw_items = raw.get("items") if isinstance(raw, dict) else None + items: list[ResourceType] = [] + invalid: list[InvalidObject] = [] + if isinstance(raw_items, list): + envelope = list_model.model_validate({**raw, "items": []}) + for raw_item in raw_items: + try: + items.append(resource_type.model_validate(raw_item)) + except ValidationError as error: + invalid.append(_invalid_object_or_raise(raw_item, error)) + else: + # No items (``null`` or missing), or not a list at all: nothing to + # tolerate, so the strict model decides. + envelope = list_model.model_validate(raw) + items = list(envelope.items) + model = cast( + "Type[ListWithInvalid[ResourceType]]", + cast(Any, ListWithInvalid)[resource_type], + ) + return model.model_construct( + envelope.model_fields_set, + **(envelope.model_extra or {}), + api_version=envelope.api_version, + kind=envelope.kind, + metadata=envelope.metadata, + items=items, + invalid=invalid, + ) diff --git a/kubex/core/params.py b/kubex/core/params.py index f8443ba0..f1d8c591 100644 --- a/kubex/core/params.py +++ b/kubex/core/params.py @@ -97,12 +97,25 @@ class DryRun(str, Enum): ALL = "All" +class InvalidObjects(str, Enum): + """What ``list()`` and ``watch()`` do with an object that does not fit the model. + + ``RAISE`` (the default) raises pydantic's ``ValidationError``. ``INCLUDE`` keeps + the object as an ``InvalidObject`` instead: ``list()`` returns it in + ``ListWithInvalid.invalid`` and ``watch()`` yields an ``InvalidWatchEvent``. + """ + + RAISE = "raise" + INCLUDE = "include" + + PropagationPolicyTypes = ( PropagationPolicy | Literal["Background", "Foreground", "Orphan"] | None ) DryRunTypes = DryRun | bool | Literal["All"] | None VersionMatchTypes = VersionMatch | Literal["Exact", "NotOlderThan"] | None FieldValidationTypes = FieldValidation | Literal["Ignore", "Strict", "Warn"] | None +InvalidObjectsTypes = InvalidObjects | Literal["raise", "include"] UidTypes = UUID | str | None ResourceVersionTypes = str | None NamespaceTypes = str | None diff --git a/kubex/core/watch_event.py b/kubex/core/watch_event.py index dafdde46..ddd9976e 100644 --- a/kubex/core/watch_event.py +++ b/kubex/core/watch_event.py @@ -1,3 +1,6 @@ +from __future__ import annotations + +from dataclasses import dataclass from enum import Enum from typing import Any, Generic, Type @@ -7,6 +10,12 @@ from kubex_core.models.typing import ResourceType from .exceptions import build_exception +from .invalid_object import InvalidObject, _invalid_object_or_raise +from .params import InvalidObjects + +INITIAL_EVENTS_END_ANNOTATION = "k8s.io/initial-events-end" +"""Annotation on the bookmark that ends the initial events of a watch started +with ``send_initial_events=True``.""" class EventType(str, Enum): @@ -44,5 +53,67 @@ def __init__( case _: self.object = self._resource_type.model_validate(raw_event["object"]) + @property + def is_initial_events_end(self) -> bool: + """Whether this is the bookmark that ends the initial events. + + A watch started with ``send_initial_events=True`` first sends an ``ADDED`` + event for every existing object, then this bookmark. From here on the + stream carries live changes, and the bookmark's + ``object.metadata.resource_version`` is the first one a watch can resume + from. Other bookmarks are periodic checkpoints and return ``False``. + """ + if self.type is not EventType.BOOKMARK or not isinstance(self.object, Bookmark): + return False + annotations = self.object.metadata.annotations or {} + return annotations.get(INITIAL_EVENTS_END_ANNOTATION) == "true" + def __repr__(self) -> str: return f"WatchEvent(type={self.type}, object={self.object})" + + +@dataclass(frozen=True, eq=False) +class InvalidWatchEvent: + """A watch event whose object does not fit the resource model. + + Yielded by ``Api.watch(invalid=InvalidObjects.INCLUDE)`` in place of a + ``WatchEvent`` for an ``ADDED``, ``MODIFIED`` or ``DELETED`` event; the stream + goes on. Like a ``WatchEvent`` it has ``type`` and ``object.metadata``, so code + that only needs an object's name, namespace or resource version works on both. + + Attributes: + type: ``ADDED``, ``MODIFIED`` or ``DELETED``. + object: The object, with its raw data, the validation error and its + metadata. + """ + + type: EventType + object: InvalidObject + + @property + def is_initial_events_end(self) -> bool: + """Always ``False``: only a bookmark ends the initial events.""" + return False + + +def _parse_watch_event( + resource_type: Type[ResourceType], + raw_event: dict[str, Any], + invalid: InvalidObjects, +) -> WatchEvent[ResourceType] | InvalidWatchEvent: + """Parse one decoded watch line. + + With ``InvalidObjects.INCLUDE``, an ``ADDED``/``MODIFIED``/``DELETED`` object + that does not fit ``resource_type`` becomes an ``InvalidWatchEvent``. Nothing + else is tolerated: an ``ERROR`` event raises its API error, and a bookmark + that does not parse raises, as both mean the stream itself is wrong. + """ + try: + return WatchEvent(resource_type, raw_event) + except ValidationError as error: + event_type = EventType(raw_event["type"]) + if invalid is InvalidObjects.RAISE or event_type is EventType.BOOKMARK: + raise + return InvalidWatchEvent( + event_type, _invalid_object_or_raise(raw_event["object"], error) + ) diff --git a/test/e2e/test_invalid_objects.py b/test/e2e/test_invalid_objects.py new file mode 100644 index 00000000..82b71835 --- /dev/null +++ b/test/e2e/test_invalid_objects.py @@ -0,0 +1,220 @@ +"""E2E: objects that do not fit the model, and the end of a watch's initial events. + +A custom resource whose CRD accepts any ``spec`` stands in for an object stored +before its CRD schema changed: ``LooseGadget`` writes a ``spec.replicas`` that +the stricter ``Gadget`` model rejects. +""" + +from __future__ import annotations + +from typing import Any, ClassVar, Literal + +import anyio +import pytest +from pydantic import Field, ValidationError + +from kubex.api import Api +from kubex.client import BaseClient +from kubex.core.exceptions import Conflict, NotFound, TooManyRequests +from kubex.core.params import InvalidObjects +from kubex.core.watch_event import EventType, InvalidWatchEvent, WatchEvent +from kubex.k8s.v1_35.apiextensions_k8s_io.v1.custom_resource_definition import ( + CustomResourceDefinition, +) +from kubex.k8s.v1_35.core.v1.config_map import ConfigMap +from kubex_core.models.base import BaseK8sModel +from kubex_core.models.interfaces import NamespaceScopedEntity +from kubex_core.models.metadata import ObjectMetadata +from kubex_core.models.resource_config import ResourceConfig, Scope + +pytestmark = pytest.mark.anyio + +_GROUP = "invalid-objects.kubex.test" +_CRD_NAME = f"gadgets.{_GROUP}" +_TIMEOUT = 60 + + +class GadgetSpec(BaseK8sModel): + replicas: int + + +class Gadget(NamespaceScopedEntity): + __RESOURCE_CONFIG__: ClassVar[ResourceConfig["Gadget"]] = ResourceConfig( + version="v1", + kind="Gadget", + group=_GROUP, + plural="gadgets", + scope=Scope.NAMESPACE, + ) + api_version: Literal["invalid-objects.kubex.test/v1"] = Field( + default="invalid-objects.kubex.test/v1", alias="apiVersion" + ) + kind: Literal["Gadget"] = Field(default="Gadget") + spec: GadgetSpec | None = None + + +class LooseGadget(NamespaceScopedEntity): + """The same resource with any ``spec``: writes objects ``Gadget`` rejects.""" + + __RESOURCE_CONFIG__: ClassVar[ResourceConfig["LooseGadget"]] = ResourceConfig( + version="v1", + kind="Gadget", + group=_GROUP, + plural="gadgets", + scope=Scope.NAMESPACE, + ) + api_version: Literal["invalid-objects.kubex.test/v1"] = Field( + default="invalid-objects.kubex.test/v1", alias="apiVersion" + ) + kind: Literal["Gadget"] = Field(default="Gadget") + spec: dict[str, Any] | None = None + + +_CRD = { + "apiVersion": "apiextensions.k8s.io/v1", + "kind": "CustomResourceDefinition", + "metadata": {"name": _CRD_NAME}, + "spec": { + "group": _GROUP, + "scope": "Namespaced", + "names": {"plural": "gadgets", "singular": "gadget", "kind": "Gadget"}, + "versions": [ + { + "name": "v1", + "served": True, + "storage": True, + "schema": { + "openAPIV3Schema": { + "type": "object", + "properties": { + "spec": { + "type": "object", + "x-kubernetes-preserve-unknown-fields": True, + } + }, + } + }, + } + ], + }, +} + + +@pytest.fixture +async def gadgets( + client: BaseClient, tmp_namespace_name: str +) -> tuple[Api[Gadget], Api[LooseGadget]]: + """Install the CRD (once per cluster) and create a valid and an invalid gadget.""" + crd_api: Api[CustomResourceDefinition] = Api( + CustomResourceDefinition, client=client + ) + try: + await crd_api.create(CustomResourceDefinition.model_validate(_CRD)) + except Conflict: + pass # installed by an earlier test; the cluster lives for the session + loose: Api[LooseGadget] = Api( + LooseGadget, client=client, namespace=tmp_namespace_name + ) + with anyio.fail_after(_TIMEOUT): + while True: + try: + await loose.list() + break + except (NotFound, TooManyRequests): # not served yet / storage starting + await anyio.sleep(0.2) + await loose.create( + LooseGadget(metadata=ObjectMetadata(name="good"), spec={"replicas": 1}) + ) + await loose.create( + LooseGadget(metadata=ObjectMetadata(name="bad"), spec={"replicas": "three"}) + ) + return Api(Gadget, client=client, namespace=tmp_namespace_name), loose + + +async def test_watch_include_goes_on_past_invalid_objects( + gadgets: tuple[Api[Gadget], Api[LooseGadget]], +) -> None: + api, loose = gadgets + initial: dict[str | None, WatchEvent[Gadget] | InvalidWatchEvent] = {} + deleted: WatchEvent[Gadget] | InvalidWatchEvent | None = None + synced = False + + with anyio.fail_after(_TIMEOUT): + async for event in api.watch( + send_initial_events=True, invalid=InvalidObjects.INCLUDE + ): + if event.is_initial_events_end: + synced = True + # `api.delete` would parse the deleted object with `Gadget` too. + await loose.delete("bad") + elif not synced and event.type is not EventType.BOOKMARK: + initial[event.object.metadata.name] = event + elif event.type is EventType.DELETED: + deleted = event + break + + good = initial["good"] + assert isinstance(good, WatchEvent) and isinstance(good.object, Gadget) + assert good.object.spec is not None and good.object.spec.replicas == 1 + + bad = initial["bad"] + assert isinstance(bad, InvalidWatchEvent) + assert bad.type is EventType.ADDED + assert bad.object.raw["spec"] == {"replicas": "three"} + assert "replicas" in str(bad.object.error) + + # A DELETED event for an invalid object still says which object went away. + assert isinstance(deleted, InvalidWatchEvent) + assert deleted.object.metadata.name == "bad" + assert deleted.object.metadata.uid == bad.object.metadata.uid + + +async def test_watch_raises_on_invalid_objects_by_default( + gadgets: tuple[Api[Gadget], Api[LooseGadget]], +) -> None: + api, _ = gadgets + with pytest.raises(ValidationError), anyio.fail_after(_TIMEOUT): + async for _ in api.watch(send_initial_events=True): + pass + + +async def test_list_include_keeps_invalid_objects_apart( + gadgets: tuple[Api[Gadget], Api[LooseGadget]], +) -> None: + api, _ = gadgets + + result = await api.list(invalid=InvalidObjects.INCLUDE) + + assert [gadget.metadata.name for gadget in result.items] == ["good"] + assert [obj.metadata.name for obj in result.invalid] == ["bad"] + assert result.metadata.resource_version + with pytest.raises(ValidationError): + await api.list() + + +async def test_initial_events_end_bookmark_is_a_resume_point( + client: BaseClient, tmp_namespace_name: str +) -> None: + api: Api[ConfigMap] = Api(ConfigMap, client=client, namespace=tmp_namespace_name) + for name in ("cm-1", "cm-2"): + await api.create(ConfigMap(metadata=ObjectMetadata(name=name))) + + added: set[str | None] = set() + resume_from: str | None = None + with anyio.fail_after(_TIMEOUT): + async for event in api.watch(send_initial_events=True): + if event.is_initial_events_end: + resume_from = event.object.metadata.resource_version + break + if event.type is EventType.ADDED: + added.add(event.object.metadata.name) + assert {"cm-1", "cm-2"} <= added + assert resume_from + + # Resuming from the bookmark delivers only what changed after the snapshot. + await api.create(ConfigMap(metadata=ObjectMetadata(name="cm-3"))) + with anyio.fail_after(_TIMEOUT): + async for event in api.watch(resource_version=resume_from): + assert event.type is EventType.ADDED + assert event.object.metadata.name == "cm-3" + break diff --git a/test/test_invalid_objects.py b/test/test_invalid_objects.py new file mode 100644 index 00000000..54df327d --- /dev/null +++ b/test/test_invalid_objects.py @@ -0,0 +1,414 @@ +"""Objects that do not fit the model: ``invalid=InvalidObjects.INCLUDE`` on +``Api.watch()``/``Api.list()``, and ``WatchEvent.is_initial_events_end``.""" + +from __future__ import annotations + +import json +from typing import TYPE_CHECKING, Any, assert_type + +import pytest +from pydantic import ValidationError + +from kubex.api import Api +from kubex.core.exceptions import Gone +from kubex.core.invalid_object import InvalidObject, ListWithInvalid +from kubex.core.params import InvalidObjects +from kubex.core.watch_event import ( + EventType, + InvalidWatchEvent, + WatchEvent, + _parse_watch_event, +) +from kubex.k8s.v1_35.core.v1.pod import Pod +from kubex_core.models.list_entity import ListEntity +from test.stub_client import StubClient + + +@pytest.fixture +def anyio_backend() -> str: + return "asyncio" + + +def _pod(name: str, rv: str = "1", **extra: Any) -> dict[str, Any]: + return { + "apiVersion": "v1", + "kind": "Pod", + "metadata": {"name": name, "namespace": "default", "resourceVersion": rv}, + **extra, + } + + +def _bad_pod(name: str, rv: str = "1") -> dict[str, Any]: + # `spec.containers` must be a list. + return _pod(name, rv, spec={"containers": "not-a-list"}) + + +def _event(event_type: str, obj: Any) -> dict[str, Any]: + return {"type": event_type, "object": obj} + + +def _bookmark(rv: str, annotations: dict[str, str] | None = None) -> dict[str, Any]: + metadata: dict[str, Any] = {"resourceVersion": rv} + if annotations is not None: + metadata["annotations"] = annotations + return _event("BOOKMARK", {"kind": "Pod", "apiVersion": "v1", "metadata": metadata}) + + +_END_BOOKMARK = _bookmark("10", {"k8s.io/initial-events-end": "true"}) + +_GONE = _event( + "ERROR", + { + "apiVersion": "v1", + "kind": "Status", + "metadata": {}, + "status": "Failure", + "code": 410, + "reason": "Expired", + "message": "too old resource version", + }, +) + + +def _lines(*events: dict[str, Any]) -> list[str]: + return [json.dumps(event) for event in events] + + +def _list_json(*items: Any, **extra: Any) -> bytes: + return json.dumps( + { + "apiVersion": "v1", + "kind": "PodList", + "metadata": {"resourceVersion": "100", "continue": "next-page"}, + "items": list(items), + **extra, + } + ).encode() + + +# --- Parsing one watch event --------------------------------------------------- + + +@pytest.mark.parametrize("event_type", ["ADDED", "MODIFIED", "DELETED"]) +def test_include_turns_an_invalid_object_into_an_invalid_watch_event( + event_type: str, +) -> None: + raw = _bad_pod("bad", rv="7") + event = _parse_watch_event(Pod, _event(event_type, raw), InvalidObjects.INCLUDE) + + assert isinstance(event, InvalidWatchEvent) + assert event.type is EventType(event_type) + assert event.object.raw == raw + assert event.object.metadata.name == "bad" + assert event.object.metadata.namespace == "default" + assert event.object.metadata.resource_version == "7" + assert isinstance(event.object.error, ValidationError) + assert "containers" in str(event.object.error) + + +def test_include_returns_a_watch_event_for_a_valid_object() -> None: + event = _parse_watch_event(Pod, _event("ADDED", _pod("ok")), InvalidObjects.INCLUDE) + assert isinstance(event, WatchEvent) + assert isinstance(event.object, Pod) + + +def test_raise_keeps_raising_validation_error() -> None: + with pytest.raises(ValidationError): + _parse_watch_event(Pod, _event("ADDED", _bad_pod("bad")), InvalidObjects.RAISE) + + +def test_include_still_raises_for_a_bookmark_that_does_not_parse() -> None: + bookmark_without_metadata = _event("BOOKMARK", {"kind": "Pod"}) + with pytest.raises(ValidationError): + _parse_watch_event(Pod, bookmark_without_metadata, InvalidObjects.INCLUDE) + + +def test_include_still_raises_error_events() -> None: + with pytest.raises(Gone): + _parse_watch_event(Pod, _GONE, InvalidObjects.INCLUDE) + + +@pytest.mark.parametrize( + "obj", + [ + pytest.param(None, id="null-object"), + pytest.param(["not", "an", "object"], id="list-object"), + pytest.param({"kind": "Pod", "spec": {"containers": 1}}, id="no-metadata"), + pytest.param( + {"kind": "Pod", "metadata": {"namespace": "default"}, "spec": 1}, + id="no-name", + ), + pytest.param({"kind": "Pod", "metadata": "nope", "spec": 1}, id="bad-metadata"), + ], +) +def test_include_raises_when_the_object_cannot_be_identified(obj: Any) -> None: + with pytest.raises(ValidationError): + _parse_watch_event(Pod, _event("MODIFIED", obj), InvalidObjects.INCLUDE) + + +def test_invalid_object_repr_is_short() -> None: + event = _parse_watch_event( + Pod, _event("ADDED", _bad_pod("bad")), InvalidObjects.INCLUDE + ) + assert isinstance(event, InvalidWatchEvent) + assert repr(event.object) == ( + "InvalidObject(kind='Pod', namespace='default', name='bad', errors=1)" + ) + + +# --- is_initial_events_end ----------------------------------------------------- + + +def test_end_bookmark_is_initial_events_end() -> None: + assert WatchEvent(Pod, _END_BOOKMARK).is_initial_events_end + + +@pytest.mark.parametrize( + "raw_event", + [ + pytest.param(_bookmark("11"), id="periodic-bookmark"), + pytest.param(_bookmark("11", {"other": "true"}), id="other-annotation"), + pytest.param( + _bookmark("11", {"k8s.io/initial-events-end": "false"}), id="false-value" + ), + pytest.param(_event("ADDED", _pod("ok")), id="not-a-bookmark"), + pytest.param( + _event( + "ADDED", + { + **_pod("ok"), + "metadata": { + "name": "ok", + "annotations": {"k8s.io/initial-events-end": "true"}, + }, + }, + ), + id="annotated-object", + ), + ], +) +def test_other_events_are_not_initial_events_end(raw_event: dict[str, Any]) -> None: + assert not WatchEvent(Pod, raw_event).is_initial_events_end + + +def test_invalid_watch_event_is_never_initial_events_end() -> None: + event = _parse_watch_event( + Pod, _event("ADDED", _bad_pod("bad")), InvalidObjects.INCLUDE + ) + assert not event.is_initial_events_end + + +# --- Api.watch ----------------------------------------------------------------- + + +@pytest.mark.anyio +@pytest.mark.parametrize("invalid", [InvalidObjects.INCLUDE, "include"]) +async def test_watch_include_goes_on_after_an_invalid_object( + invalid: InvalidObjects, +) -> None: + client = StubClient( + stream_lines=_lines( + _event("ADDED", _pod("a", rv="1")), + _event("MODIFIED", _bad_pod("bad", rv="2")), + _END_BOOKMARK, + _event("DELETED", _bad_pod("bad", rv="11")), + _event("ADDED", _pod("b", rv="12")), + ) + ) + api: Api[Pod] = Api(Pod, client=client, namespace="default") + + events = [event async for event in api.watch(invalid=invalid)] + + assert [ + (type(e).__name__, e.type, e.object.metadata.resource_version) for e in events + ] == [ + ("WatchEvent", EventType.ADDED, "1"), + ("InvalidWatchEvent", EventType.MODIFIED, "2"), + ("WatchEvent", EventType.BOOKMARK, "10"), + ("InvalidWatchEvent", EventType.DELETED, "11"), + ("WatchEvent", EventType.ADDED, "12"), + ] + assert [e.is_initial_events_end for e in events] == [ + False, + False, + True, + False, + False, + ] + + +@pytest.mark.anyio +async def test_watch_raises_on_an_invalid_object_by_default() -> None: + client = StubClient( + stream_lines=_lines( + _event("ADDED", _pod("a")), + _event("MODIFIED", _bad_pod("bad")), + _event("ADDED", _pod("b")), + ) + ) + api: Api[Pod] = Api(Pod, client=client, namespace="default") + seen: list[str | None] = [] + + with pytest.raises(ValidationError): + async for event in api.watch(): + seen.append(event.object.metadata.name) + + assert seen == ["a"] + + +@pytest.mark.anyio +async def test_watch_include_still_raises_on_a_corrupt_line() -> None: + client = StubClient(stream_lines=['{"type": "ADDED", "object": ']) + api: Api[Pod] = Api(Pod, client=client, namespace="default") + with pytest.raises(json.JSONDecodeError): + async for _ in api.watch(invalid=InvalidObjects.INCLUDE): + pass + + +@pytest.mark.anyio +async def test_watch_rejects_an_unknown_invalid_value() -> None: + client = StubClient() + api: Api[Pod] = Api(Pod, client=client, namespace="default") + with pytest.raises(ValueError, match="yield"): + async for _ in api.watch(invalid="yield"): # type: ignore[call-overload] + pass + assert client.requests == [] + + +# --- Api.list ------------------------------------------------------------------ + + +@pytest.mark.anyio +@pytest.mark.parametrize("invalid", [InvalidObjects.INCLUDE, "include"]) +async def test_list_include_keeps_invalid_objects_apart( + invalid: InvalidObjects, +) -> None: + bad = _bad_pod("bad", rv="2") + client = StubClient(response_content=_list_json(_pod("a"), bad, _pod("b"))) + api: Api[Pod] = Api(Pod, client=client, namespace="default") + + result = await api.list(invalid=invalid) + + assert isinstance(result, ListWithInvalid) + assert isinstance(result, ListEntity) + assert all(isinstance(item, Pod) for item in result.items) + assert [item.metadata.name for item in result.items] == ["a", "b"] + assert len(result.invalid) == 1 + assert isinstance(result.invalid[0], InvalidObject) + assert result.invalid[0].raw == bad + assert result.invalid[0].metadata.name == "bad" + assert result.metadata.resource_version == "100" + assert result.metadata.continue_ == "next-page" + assert (result.api_version, result.kind) == ("v1", "PodList") + + +@pytest.mark.anyio +async def test_list_raises_on_an_invalid_object_by_default() -> None: + client = StubClient(response_content=_list_json(_pod("a"), _bad_pod("bad"))) + api: Api[Pod] = Api(Pod, client=client, namespace="default") + with pytest.raises(ValidationError): + await api.list() + + +@pytest.mark.anyio +async def test_list_include_with_only_valid_objects() -> None: + client = StubClient(response_content=_list_json(_pod("a"), _pod("b"))) + api: Api[Pod] = Api(Pod, client=client, namespace="default") + + result = await api.list(invalid=InvalidObjects.INCLUDE) + + assert [item.metadata.name for item in result.items] == ["a", "b"] + assert result.invalid == [] + + +@pytest.mark.anyio +async def test_list_include_with_null_items() -> None: + content = json.dumps( + {"apiVersion": "v1", "kind": "PodList", "metadata": {}, "items": None} + ).encode() + api: Api[Pod] = Api(Pod, client=StubClient(response_content=content)) + + result = await api.list(invalid=InvalidObjects.INCLUDE, namespace="default") + + assert result.items == [] + assert result.invalid == [] + + +@pytest.mark.anyio +@pytest.mark.parametrize( + "content", + [ + pytest.param( + json.dumps( + {"apiVersion": "v1", "kind": "NotAList", "metadata": {}, "items": []} + ).encode(), + id="wrong-kind", + ), + pytest.param( + json.dumps( + {"apiVersion": "v1", "kind": "PodList", "metadata": {}, "items": {}} + ).encode(), + id="items-not-a-list", + ), + pytest.param(b"[]", id="not-an-object"), + pytest.param( + _list_json({"kind": "Pod", "metadata": {}, "spec": 1}), id="item-no-name" + ), + ], +) +async def test_list_include_still_raises_for_a_broken_list(content: bytes) -> None: + api: Api[Pod] = Api(Pod, client=StubClient(response_content=content)) + with pytest.raises(ValidationError): + await api.list(invalid=InvalidObjects.INCLUDE, namespace="default") + + +@pytest.mark.anyio +async def test_list_include_serializes_like_a_list_without_invalid() -> None: + client = StubClient( + response_content=_list_json(_pod("a"), _bad_pod("bad"), futureField="kept") + ) + api: Api[Pod] = Api(Pod, client=client, namespace="default") + + result = await api.list(invalid=InvalidObjects.INCLUDE) + dumped = result.model_dump(by_alias=True, exclude_none=True, mode="json") + + assert "invalid" not in dumped + assert dumped["futureField"] == "kept" + assert dumped["kind"] == "PodList" + assert [item["metadata"]["name"] for item in dumped["items"]] == ["a"] + # Items keep their full Pod shape, not just the BaseEntity fields. + assert dumped["items"][0]["metadata"]["namespace"] == "default" + + +@pytest.mark.anyio +async def test_list_rejects_an_unknown_invalid_value() -> None: + client = StubClient() + api: Api[Pod] = Api(Pod, client=client, namespace="default") + with pytest.raises(ValueError, match="collect"): + await api.list(invalid="collect") # type: ignore[call-overload] + assert client.requests == [] + + +# --- Static types (checked by mypy, never run) ---------------------------------- + +if TYPE_CHECKING: + + async def _check_types(api: Api[Pod], policy: InvalidObjects) -> None: + assert_type(await api.list(), ListEntity[Pod]) + assert_type(await api.list(invalid=InvalidObjects.RAISE), ListEntity[Pod]) + assert_type( + await api.list(invalid=InvalidObjects.INCLUDE), ListWithInvalid[Pod] + ) + assert_type(await api.list(invalid="include"), ListWithInvalid[Pod]) + assert_type(await api.list(invalid=policy), ListEntity[Pod]) + + async for plain in api.watch(): + assert_type(plain, WatchEvent[Pod]) + async for raised in api.watch(invalid="raise"): + assert_type(raised, WatchEvent[Pod]) + async for included in api.watch(invalid=InvalidObjects.INCLUDE): + assert_type(included, WatchEvent[Pod] | InvalidWatchEvent) + # Both kinds of event carry the object's metadata. + assert_type(included.object.metadata.name, str | None) + async for dynamic in api.watch(invalid=policy): + assert_type(dynamic, WatchEvent[Pod] | InvalidWatchEvent) From 7d35035ce324abd28b6cffc27cc9eb5f28ab37e5 Mon Sep 17 00:00:00 2001 From: Codemageddon Date: Wed, 28 Oct 2026 00:26:47 +0100 Subject: [PATCH 2/2] Raise ValidationError for a non-JSON body in list(invalid=INCLUDE) The tolerant list path parsed the body with json.loads, so a body that is not JSON or not UTF-8 raised JSONDecodeError/UnicodeDecodeError instead of the ValidationError the default mode and the Api.list() docstring promise. Fall back to the strict list model's model_validate_json in that case. Co-Authored-By: Claude Opus 5.5 --- kubex/core/invalid_object.py | 8 +++++++- test/test_invalid_objects.py | 5 +++++ 2 files changed, 12 insertions(+), 1 deletion(-) diff --git a/kubex/core/invalid_object.py b/kubex/core/invalid_object.py index 387fc762..53c14506 100644 --- a/kubex/core/invalid_object.py +++ b/kubex/core/invalid_object.py @@ -92,7 +92,13 @@ def _validate_list_with_invalid( The list itself (``apiVersion``, ``kind``, ``metadata``) is validated as strictly as by ``list_model``; only its items are tolerated. """ - raw = json.loads(content) + try: + raw = json.loads(content) + except ValueError: + # Not JSON (or not UTF-8): raise the strict model's ``ValidationError``, + # as without ``INCLUDE``. Should pydantic parse it after all, keep ours. + list_model.model_validate_json(content) + raise raw_items = raw.get("items") if isinstance(raw, dict) else None items: list[ResourceType] = [] invalid: list[InvalidObject] = [] diff --git a/test/test_invalid_objects.py b/test/test_invalid_objects.py index 54df327d..f7025a8a 100644 --- a/test/test_invalid_objects.py +++ b/test/test_invalid_objects.py @@ -354,6 +354,11 @@ async def test_list_include_with_null_items() -> None: pytest.param( _list_json({"kind": "Pod", "metadata": {}, "spec": 1}), id="item-no-name" ), + pytest.param(b"Bad Gateway", id="not-json"), + pytest.param( + b'{"apiVersion": "v1", "kind": "PodList", "items": [', id="cut-off" + ), + pytest.param(b'{"kind": "\xff"}', id="not-utf8"), ], ) async def test_list_include_still_raises_for_a_broken_list(content: bytes) -> None: