24580658204

Committed 17 Apr 2026 06:29PM UTC coverage: 77.939% (+1.0%) from 76.951%

Build # 24580658204

Build Type

push

github

Committed by

web-flow

Commit Message

Refactor k8s apply: SSA test-ops, 409 retry, watch-based delete, imperative API (#102)

## Summary

- Harden the K8s apply path with SSA-derived JSON-patch `test` ops for
`/metadata/uid` and `/metadata/resourceVersion`, and wrap the SSA patch
branch in a 409-retry loop that recomputes the patch on conflict.
- Extend `K8SResource` with `list()`, `watch()`, `delete(wait=True)`
that uses watch + `get()` recheck, and add a `resource_generator()`
extension seam on `KubernetesPlugin`.
- Add `ktor.k8s.resource(manifest)` — a fully-wired `K8SResource`
factory for imperative CRUD from `.kubernator.py` scripts, bypassing the
declarative apply lifecycle.
- Thread the applied manifest as a 4th return from `_apply_resource` and
its inner patch/create/delete callables for external consumers.
- Move dump-file open/close from `app.py` into
`KubernetesPlugin.handle_apply`; fix dump-mode `patch_func` to seed
uid/RV on a copy of the local manifest by extracting them from the
patch's own test ops (via `jsonpointer`), so the simulated post-patch
manifest can be returned honestly.
- Rewrite `delete_create` and `resource_version_merge` integration tests
as single-phase using the new imperative API, replacing `kubectl`-based
phase-1 setup and the `TEST_PHASE` env re-invocation.
- Add unit test for 409 retry and integration tests for `watch()` and
`resource_generator`. Port `resource_version_merge` from minikube to
kind.

## Test plan

- [x] `pyb -vX run_unit_tests` (32 tests, all green — includes new
`k8s_apply_retry_tests`)
- [x] `pyb -vX run_integration_tests -P
integrationtest_file_glob=delete_create_tests.py` (single-phase, kind,
immutable-field delete+recreate path)
- [x] `pyb -vX run_integration_tests -P
integrationtest_file_glob=resource_version_merge_tests.py`
(single-phase, kind, dump mode patch verification)
- [ ] Full `pyb -vX run_integration_tests` suite via CI across the
Python 3.10–3.14 matrix

Coverage Stats

614 of 976 branches covered (62.91%)

Branch coverage included in aggregate %.

137 of 163 new or added lines in 4 files covered. (84.05%)

5 existing lines in 3 files now uncovered.

3039 of 3711 relevant lines covered (81.89%)

4.09 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

80.94

/src/main/python/kubernator/plugins/k8s_api.py

# -*- coding: utf-8 -*-
#
#   Copyright 2020 Express Systems USA, Inc
#   Copyright 2021 Karellen, Inc.
#
#   Licensed under the Apache License, Version 2.0 (the "License");
#   you may not use this file except in compliance with the License.
#   You may obtain a copy of the License at
#
#       http://www.apache.org/licenses/LICENSE-2.0
#
#   Unless required by applicable law or agreed to in writing, software
#   distributed under the License is distributed on an "AS IS" BASIS,
#   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#   See the License for the specific language governing permissions and
#   limitations under the License.
#

import base64
import json
import re
from collections import namedtuple
from collections.abc import Callable, Mapping, MutableMapping, Sequence, Iterable
from enum import Enum, auto
from functools import partial, wraps
from pathlib import Path
from typing import Union, Optional

import yaml
from jsonschema._format import FormatChecker
from jsonschema._keywords import required
from jsonschema.exceptions import ValidationError
from jsonschema.validators import extend, Draft7Validator
from openapi_schema_validator import OAS31Validator

from kubernator.api import load_file, FileType, load_remote_file, calling_frame_source, parse_yaml_docs


def api_exc_normalize_body(e):
    """Parse a raw ApiException body (JSON or YAML) into a Python object in-place.

    Idempotent: if e.body is already a parsed object, it is left unchanged.
    """
    if not isinstance(e.body, (str, bytes)):
        return
    if e.headers and "content-type" in e.headers:
        content_type = e.headers["content-type"]
        if content_type == "application/json" or content_type.endswith("+json"):
            e.body = json.loads(e.body)
        elif (content_type in ("application/yaml", "application/x-yaml", "text/yaml",
                               "text/x-yaml") or content_type.endswith("+yaml")):
            e.body = yaml.safe_load(e.body)


def api_exc_format_body(e):
    """Format an ApiException body back to a string for human-readable display.

    Idempotent: if e.body is already a string/bytes, it is left unchanged.
    """
    if not isinstance(e.body, (str, bytes)):
        e.body = json.dumps(e.body, indent=4)


def _normalize_api_exc(func):
    """Decorator: normalize ApiException body before propagating."""
    @wraps(func)
    def wrapper(*args, **kwargs):
        from kubernetes.client import ApiException
        try:
            return func(*args, **kwargs)
        except ApiException as e:
            api_exc_normalize_body(e)
            raise
    return wrapper


K8S_WARNING_HEADER = re.compile(r'(?:,\s*)?(\d{3})\s+(\S+)\s+"(.+?)(?<!\\)"(?:\s+\"(.+?)(?<!\\)\")?\s*')
UPPER_FOLLOWED_BY_LOWER_RE = re.compile(r"(.)([A-Z][a-z]+)")
LOWER_OR_NUM_FOLLOWED_BY_UPPER_RE = re.compile(r"([a-z0-9])([A-Z])")

K8S_MINIMAL_RESOURCE_SCHEMA = {
    "properties": {
        "apiVersion": {
            "type": "string"
        },
        "kind": {
            "type": "string"
        },
        "metadata": {
            "type": "object",
            "properties": {
                "name": {
                    "type": "string"
                },
                "namespace": {
                    "type": "string"
                }
            },
            "required": ["name"]
        }
    },
    "type": "object",
    "required": ["apiVersion", "kind"]
}
K8S_MINIMAL_RESOURCE_VALIDATOR = Draft7Validator(K8S_MINIMAL_RESOURCE_SCHEMA)

CLUSTER_RESOURCE_PATH = re.compile(r"^/apis?/(?:[^/]+/){1,2}([^/]+)$")
NAMESPACED_RESOURCE_PATH = re.compile(r"^/apis?/(?:[^/]+/){1,2}namespaces/[^/]+/([^/]+)$")


class K8SResourcePatchType(Enum):
    JSON_PATCH = auto()
    SERVER_SIDE_PATCH = auto()


class K8SPropagationPolicy(Enum):
    BACKGROUND = ("Background",)
    FOREGROUND = ("Foreground",)
    ORPHAN = ("Orphan",)

    def __init__(self, policy):
        self.policy = policy


def is_integer(instance):
    # bool inherits from int, so ensure bools aren't reported as ints
    if isinstance(instance, bool):
        return False
    return isinstance(instance, int)


def is_string(instance):
    return isinstance(instance, str)


def type_validator(validator, data_type, instance, schema):
    if instance is None:
        return

    if data_type == "string" and schema.get("format") == "int-or-string":
        if not (is_string(instance) or is_integer(instance)):
            yield ValidationError("%r is not of type %s" % (instance, "int-or-string"))
    elif not validator.is_type(instance, data_type):
        yield ValidationError("%r is not of type %s" % (instance, data_type))


K8SValidator = extend(OAS31Validator, validators={
    "type": type_validator,
    "required": required
})

k8s_format_checker = FormatChecker()


@k8s_format_checker.checks("int32")
def check_int32(value):
    return value is not None and (-2147483648 < value < 2147483647)


@k8s_format_checker.checks("int64")
def check_int64(value):
    return value is not None and (-9223372036854775808 < value < 9223372036854775807)


@k8s_format_checker.checks("float")
def check_float(value):
    return value is not None and (-3.4E+38 < value < +3.4E+38)


@k8s_format_checker.checks("double")
def check_double(value):
    return value is not None and (-1.7E+308 < value < +1.7E+308)


@k8s_format_checker.checks("byte", ValueError)
def check_byte(value):
    if value is None:
        return False
    base64.b64decode(value, validate=True)
    return True


@k8s_format_checker.checks("int-or-string")
def check_int_or_string(value):
    return check_int32(value) if is_integer(value) else is_string(value)


def to_group_and_version(api_version):
    group, _, version = api_version.partition("/")
    if not version:
        version = group
        group = ""
    return group, version


def to_k8s_resource_def_key(manifest):
    return K8SResourceDefKey(*to_group_and_version(manifest["apiVersion"]),
                             manifest["kind"])


class K8SResourceDefKey(namedtuple("K8SResourceDefKey", ["group", "version", "kind"])):
    __slots__ = ()

    def __str__(self):
        return f"{self.group}{'/' if self.group else '/'}{self.version}/{self.kind}"


class K8SResourceDef:
    def __init__(self, key, singular, plural, namespaced, custom, schema):
        self.key = key
        self.singular = singular
        self.plural = plural
        self.namespaced = namespaced
        self.custom = custom
        self.schema = schema

        self._api_get = None
        self._api_create = None
        self._api_patch = None
        self._api_delete = None
        self._api_list = None

    @property
    def group(self) -> str:
        return self.key.group

    @property
    def version(self) -> str:
        return self.key.version

    @property
    def kind(self) -> str:
        return self.key.kind

    @property
    def has_api(self) -> bool:
        return self.custom or self.plural

    @property
    def get(self):
        return self._api_get

    @property
    def create(self):
        return self._api_create

    @property
    def patch(self):
        return self._api_patch

    @property
    def delete(self):
        return self._api_delete

    @property
    def list(self):
        return self._api_list

    def __eq__(self, o: object) -> bool:
        if not isinstance(o, K8SResourceDef):
            return False

        return (self.key == o.key and
                self.singular == o.singular and
                self.plural == o.plural and
                self.namespaced == o.namespaced and
                self.custom == o.custom)

    def __hash__(self) -> int:
        return self.key.__hash__()

    def __str__(self):
        return f"{self.key=}, {self.singular=}, {self.plural=}, {self.namespaced=}, {self.custom=}"

    @classmethod
    def from_manifest(cls, key: K8SResourceDefKey,
                      schema,
                      paths: Mapping[K8SResourceDefKey, Mapping[str, Mapping]]):
        singular = key.kind.lower()

        plural = None
        namespaced = False

        if singular == "namespace":
            plural = "namespaces"
        else:
            for path in paths.get(key, ()):
                if m := NAMESPACED_RESOURCE_PATH.fullmatch(path):
                    plural = m[1]
                    namespaced = True
                    break
                elif m := CLUSTER_RESOURCE_PATH.fullmatch(path):
                    plural = m[1]

        yield K8SResourceDef(key, singular, plural, namespaced, False, schema)

    @classmethod
    def from_resource(cls, resource: "K8SResource"):
        manifest = resource.manifest
        spec = manifest["spec"]
        group = spec["group"]
        names = spec["names"]
        kind = names["kind"]
        singular = names.get("singular", names["kind"].lower())
        plural = names["plural"]
        namespaced = spec["scope"] == "Namespaced"

        for version_spec in spec["versions"]:
            version = version_spec["name"]
            if resource.version == "v1":
                schema = version_spec["schema"]["openAPIV3Schema"]
            else:
                schema = spec["validation"]["openAPIV3Schema"]
            yield K8SResourceDef(K8SResourceDefKey(group, version, kind), singular, plural, namespaced, True, schema)

    def populate_api(self, k8s_client_module, k8s_client):
        if not self.has_api:
            raise RuntimeError(f"{self} has no API")

        if self._api_get:
            return

        group = self.group or "core"
        version = self.version
        kind = self.kind

        if self.custom:
            k8s_api = k8s_client_module.CustomObjectsApi(k8s_client)

            kwargs = {"group": group,
                      "version": version,
                      "plural": self.plural}
            if self.namespaced:
                self._api_get = partial(k8s_api.get_namespaced_custom_object, **kwargs)
                self._api_patch = partial(k8s_api.patch_namespaced_custom_object, **kwargs)
                self._api_create = partial(k8s_api.create_namespaced_custom_object, **kwargs)
                self._api_delete = partial(k8s_api.delete_namespaced_custom_object, **kwargs)
                self._api_list = partial(k8s_api.list_namespaced_custom_object, **kwargs)
            else:
                self._api_get = partial(k8s_api.get_cluster_custom_object, **kwargs)
                self._api_patch = partial(k8s_api.patch_cluster_custom_object, **kwargs)
                self._api_create = partial(k8s_api.create_cluster_custom_object, **kwargs)
                self._api_delete = partial(k8s_api.delete_cluster_custom_object, **kwargs)
                self._api_list = partial(k8s_api.list_cluster_custom_object, **kwargs)
        else:
            # Take care for the case e.g. api_type is "apiextensions.k8s.io"
            # Only replace the last instance
            group = "".join(group.rsplit(".k8s.io", 1))

            # convert group name from DNS subdomain format to
            # python class name convention
            group = "".join(word.capitalize() for word in group.split('.'))
            fcn_to_call = f"{group}{version.capitalize()}Api"
            k8s_api = getattr(k8s_client_module, fcn_to_call)(k8s_client)

            # Replace CamelCased action_type into snake_case
            kind = UPPER_FOLLOWED_BY_LOWER_RE.sub(r"\1_\2", kind)
            kind = LOWER_OR_NUM_FOLLOWED_BY_UPPER_RE.sub(r"\1_\2", kind).lower()

            if self.namespaced:
                self._api_get = getattr(k8s_api, f"read_namespaced_{kind}")
                self._api_patch = getattr(k8s_api, f"patch_namespaced_{kind}")
                self._api_create = getattr(k8s_api, f"create_namespaced_{kind}")
                self._api_delete = getattr(k8s_api, f"delete_namespaced_{kind}")
                self._api_list = getattr(k8s_api, f"list_namespaced_{kind}")
            else:
                self._api_get = getattr(k8s_api, f"read_{kind}")
                self._api_patch = getattr(k8s_api, f"patch_{kind}")
                self._api_create = getattr(k8s_api, f"create_{kind}")
                self._api_delete = getattr(k8s_api, f"delete_{kind}")
                self._api_list = getattr(k8s_api, f"list_{kind}")


class K8SResourceKey(namedtuple("K8SResourceKey", ["group", "kind", "name", "namespace"])):
    __slots__ = ()

    def __str__(self):
        return (f"{self.group}{'/' if self.group else 'v1/'}{self.kind}"
                f"/{self.name}{f'.{self.namespace}' if self.namespace else ''}")


class K8SResource:
    _k8s_client_version = None
    _k8s_field_validation = None
    _k8s_field_validation_patched = None
    _logger = None
    _api_warnings = None

    def __init__(self, manifest: dict, rdef: K8SResourceDef, source: Union[str, Path] = None):
        self.key = self.get_manifest_key(manifest)

        self.manifest = manifest
        self.rdef = rdef
        self.source = source

    @property
    def group(self) -> str:
        return self.key.group

    @property
    def version(self) -> str:
        return self.rdef.version

    @property
    def kind(self) -> str:
        return self.key.kind

    @property
    def name(self) -> str:
        return self.key.name

    @name.setter
    def name(self, value):
        self.manifest["metadata"]["name"] = value
        self.key = self.get_manifest_key(self.manifest)

    @property
    def namespace(self) -> Optional[str]:
        return self.key.namespace

    @namespace.setter
    def namespace(self, value):
        self.manifest["metadata"]["namespace"] = value
        self.key = self.get_manifest_key(self.manifest)

    @property
    def api_version(self) -> str:
        return self.manifest["apiVersion"]

    @property
    def schema(self) -> dict:
        return self.rdef.schema

    @property
    def is_crd(self):
        return self.group == "apiextensions.k8s.io" and self.kind == "CustomResourceDefinition"

    def __str__(self):
        return f"{self.api_version}/{self.kind}/{self.name}{'.' + self.namespace if self.namespace else ''}"

    @_normalize_api_exc
    def get(self):
        rdef = self.rdef
        kwargs = {"name": self.name,
                  "_preload_content": False}
        if rdef.namespaced:
            kwargs["namespace"] = self.namespace
        return json.loads(self.rdef.get(**kwargs).data)

    @_normalize_api_exc
    def create(self, dry_run=True):
        rdef = self.rdef
        kwargs = {"body": self.manifest,
                  "_preload_content": False,
                  "field_manager": "kubernator",
                  }

        # `and not self.rdef.custom` to be removed after solving https://github.com/kubernetes-client/gen/issues/259
        if self._k8s_field_validation_patched or not self.rdef.custom:
            kwargs["field_validation"] = self._k8s_field_validation
        if rdef.namespaced:
            kwargs["namespace"] = self.namespace
        if dry_run:
            kwargs["dry_run"] = "All"
        resp = rdef.create(**kwargs)
        self._process_response_headers(resp)
        return json.loads(resp.data)

    @_normalize_api_exc
    def patch(self, json_patch, *, patch_type: K8SResourcePatchType, force=False, dry_run=True):
        rdef = self.rdef
        kwargs = {"name": self.name,
                  "body": json_patch,
                  "_preload_content": False,
                  "field_manager": "kubernator",
                  }

        # `and not self.rdef.custom` to be removed after solving https://github.com/kubernetes-client/gen/issues/259
        if self._k8s_field_validation_patched or not self.rdef.custom:
            kwargs["field_validation"] = self._k8s_field_validation
        if patch_type == K8SResourcePatchType.SERVER_SIDE_PATCH:
            kwargs["force"] = force
        if rdef.namespaced:
            kwargs["namespace"] = self.namespace
        if dry_run:
            kwargs["dry_run"] = "All"

        def select_header_content_type_patch(content_types):
            if patch_type == K8SResourcePatchType.JSON_PATCH:
                return "application/json-patch+json"
            if patch_type == K8SResourcePatchType.SERVER_SIDE_PATCH:
                return "application/apply-patch+yaml"
            raise NotImplementedError

        if isinstance(rdef.patch, partial):
            api_client = rdef.patch.func.__self__.api_client
        else:
            api_client = rdef.patch.__self__.api_client

        old_func = api_client.select_header_content_type
        try:
            api_client.select_header_content_type = select_header_content_type_patch
            resp = rdef.patch(**kwargs)
            self._process_response_headers(resp)
            return json.loads(resp.data)
        finally:
            api_client.select_header_content_type = old_func

    @_normalize_api_exc
    def delete(self, *, dry_run=True, propagation_policy=K8SPropagationPolicy.BACKGROUND, wait=True):
        from kubernetes.client import ApiException
        rdef = self.rdef
        kwargs = {"name": self.name,
                  "_preload_content": False,
                  "propagation_policy": propagation_policy.policy
                  }
        if rdef.namespaced:
            kwargs["namespace"] = self.namespace
        if dry_run:
            kwargs["dry_run"] = "All"

        result = json.loads(rdef.delete(**kwargs).data)

        if wait and not dry_run:
            # Wait for the resource to actually disappear by watching for the
            # DELETED event. If the resource is already gone before the watch
            # opens (race), the watch produces no events; loop with a short
            # server-side timeout and re-check existence via get() to break out.
            while True:
                for event in self.watch(timeout_seconds=10):
                    if event["type"] == "DELETED":
                        return result
                try:
                    self.get()
                except ApiException as e:
                    if e.status == 404:
                        return result
                    raise

        return result

    def watch(self, *, timeout_seconds=None):
        from kubernetes import watch as k8s_watch
        rdef = self.rdef
        kwargs = {"field_selector": f"metadata.name={self.name}"}
        if rdef.namespaced:
            kwargs["namespace"] = self.namespace
        if timeout_seconds is not None:
            kwargs["timeout_seconds"] = timeout_seconds
        return k8s_watch.Watch().stream(rdef.list, **kwargs)

    @staticmethod
    def get_manifest_key(manifest):
        return K8SResourceKey(to_group_and_version(manifest["apiVersion"])[0],
                              manifest["kind"],
                              manifest["metadata"]["name"],
                              manifest["metadata"].get("namespace"))

    @staticmethod
    def get_manifest_description(manifest: dict, source=None):
        api_version = manifest.get("apiVersion")
        kind = manifest.get("kind")
        metadata = manifest.get("metadata")
        name = None
        namespace = None
        if metadata:
            name = metadata.get("name")
            namespace = metadata.get("namespace")
        return (f"{api_version or 'unknown'}/{kind or '<unknown>'}/"
                f"{name or '<unknown>'}{'.' + namespace if namespace else ''}")

    def __eq__(self, other):
        if not isinstance(other, K8SResource):
            return False
        return self.key == other.key and self.manifest == other.manifest

    def _process_response_headers(self, resp):
        headers = resp.headers
        warn_headers = headers.get("Warning")
        if warn_headers:
            for warn in K8S_WARNING_HEADER.findall(warn_headers):
                code, _, msg, _ = warn
                code = int(code)
                msg = msg.encode("utf-8").decode("unicode_escape")
                if code == 299:
                    self._api_warnings(self, msg)
                else:
                    self._logger.warning("Unknown API warning received for resource %s from %s: code %d: %s",
                                         self, self.source, code, msg)


class K8SResourcePluginMixin:
    def __init__(self):
        self.resource_definitions: MutableMapping[K8SResourceDefKey, K8SResourceDef] = {}
        self.resource_paths: MutableMapping[K8SResourceDefKey, MutableMapping[str, dict]] = {}
        self.resources: MutableMapping[K8SResourceKey, K8SResource] = {}

        self.resource_definitions_schema = None

    def add_resources(self, manifests: Union[str, list, dict], source: Union[str, Path] = None):
        if not source:
            source = calling_frame_source()

        if isinstance(manifests, str):
            manifests = list(parse_yaml_docs(manifests, source))

        if isinstance(manifests, (Mapping, dict)):
            return self.add_resource(manifests, source)
        else:
            return [self.add_resource(m, source) for m in manifests if m]

    def add_resource(self, manifest: dict, source: Union[str, Path] = None):
        if not source:
            source = calling_frame_source()
        resource = self._create_resource(manifest, source)

        try:
            trans_resource = self._transform_resource(list(self.resources.values()), resource)
        except Exception as e:
            self.logger.error("An error occurred running transformers on %s", resource, exc_info=e)
            raise

        errors = list(self._validate_resource(trans_resource.manifest, source))
        if errors:
            for error in errors:
                if source:
                    self.logger.error("Error detected in re-transformed K8S resource %s generated through %s",
                                      trans_resource, source, exc_info=error)
            raise errors[0]

        return self._add_resource(trans_resource, source)

    def add_crds(self, manifests: Union[str, list, dict], source: Union[str, Path] = None):
        if not source:
            source = calling_frame_source()

        if isinstance(manifests, str):
            manifests = list(parse_yaml_docs(manifests, source))

        if isinstance(manifests, (Mapping, dict)):
            return self.add_crd(manifests, source)
        else:
            return [self.add_crd(m, source) for m in manifests if m]

    def add_crd(self, manifest: dict, source: Union[str, Path] = None):
        if not source:
            source = calling_frame_source()
        resource = self._create_resource(manifest, source)
        if not resource.is_crd:
            resource_description = K8SResource.get_manifest_description(manifest, source)
            raise ValueError(f"K8S manifest {resource_description} from {source} is not a CRD")

        self._add_crd(resource)
        return resource

    def create_resource(self, manifest: dict, source: Union[str, Path] = None):
        """Create K8S resource without adding it"""
        if not source:
            source = calling_frame_source()

        return self._create_resource(manifest, source)

    def add_local_resources(self, path: Path, file_type: FileType, source: str = None):
        manifests = load_file(self.logger, path, file_type)

        return [self.add_resource(m, source or path) for m in manifests if m]

    def add_remote_resources(self, url: str, file_type: FileType, *, sub_category: Optional[str] = None,
                             source: str = None):
        manifests = load_remote_file(self.logger, url, file_type, sub_category=sub_category)

        return [self.add_resource(m, source or url) for m in manifests if m]

    def add_local_crds(self, path: Path, file_type: FileType, source: str = None):
        manifests = load_file(self.logger, path, file_type)

        return [self.add_crd(m, source or path) for m in manifests if m]

    def add_remote_crds(self, url: str, file_type: FileType, *, sub_category: Optional[str] = None,
                        source: str = None):
        manifests = load_remote_file(self.logger, url, file_type, sub_category=sub_category)

        return [self.add_crd(m, source or url) for m in manifests if m]

    def get_api_versions(self):
        api_versions = set()
        for rdef in self.resource_definitions:
            api_version = f"{f'{rdef.group}/' if rdef.group else ''}{rdef.version}"
            if api_version not in api_versions:
                api_versions.add(api_version)
        return sorted(api_versions)

    def _create_resource(self, manifest: dict, source: Union[str, Path] = None):
        resource_description = K8SResource.get_manifest_description(manifest, source)

        new_manifest = self._patch_manifest(manifest, resource_description)
        if new_manifest != manifest:
            manifest = new_manifest
            resource_description = K8SResource.get_manifest_description(manifest, source)

        self.logger.debug("Validating K8S manifest for %s", resource_description)
        errors = list(self._validate_resource(manifest, source))
        if errors:
            for error in errors:
                self.logger.error("Error detected in K8S manifest %s from %s: \n%s",
                                  resource_description, source or "<unknown>", yaml.safe_dump(manifest, None),
                                  exc_info=error)
            raise errors[0]

        rdef = self._get_manifest_rdef(manifest)
        return K8SResource(manifest, rdef, source)

    def _add_resource(self, resource: K8SResource, source):
        if resource.key in self.resources:
            existing_resource = self.resources[resource.key]
            if resource != existing_resource:
                raise ValidationError("resource %s from %s already exists and was added from %s" %
                                      (resource.key, resource.source, existing_resource.source))
            self.logger.trace("K8S resource for %s from %s is already present and is identical", resource, source)
            return existing_resource

        self.logger.info("Adding K8S resource for %s from %s", resource, source)
        self.resources[resource.key] = resource

        if resource.is_crd:
            self._add_crd(resource)

        return resource

    def _patch_manifest(self,
                        manifest: dict,
                        resource_description: str):
        return manifest

    def _transform_resource(self,
                            resources: Sequence[K8SResource],
                            resource: K8SResource) -> K8SResource:
        return resource

    def _filter_resources(self, func: Callable[[K8SResource], bool]):
        yield from filter(func, self.resources.values())

    def _validate_resource(self, manifest: dict, source: Union[str, Path] = None):
        for error in self._yield_manifest_rdef(manifest):
            if isinstance(error, Exception):
                yield error
            else:
                rdef = error
                k8s_validator = K8SValidator(rdef.schema,
                                             format_checker=k8s_format_checker)
                yield from k8s_validator.iter_errors(manifest)

    def _get_manifest_rdef(self, manifest):
        for error in self._yield_manifest_rdef(manifest):
            if isinstance(error, Exception):
                raise error
            else:
                return error

    def _yield_manifest_rdef(self, manifest):
        error = None
        for error in K8S_MINIMAL_RESOURCE_VALIDATOR.iter_errors(manifest):
            yield error

        if error:
            return

        key = K8SResourceDefKey(*to_group_and_version(manifest["apiVersion"]), manifest["kind"])

        try:
            yield self.resource_definitions[key]
        except KeyError:
            yield ValidationError("%s is not a defined Kubernetes resource" % (key,),
                                  validator=K8S_MINIMAL_RESOURCE_VALIDATOR,
                                  validator_value=key,
                                  instance=manifest,
                                  schema=K8S_MINIMAL_RESOURCE_SCHEMA)

    def _add_crd(self, resource: K8SResource):
        for crd in K8SResourceDef.from_resource(resource):
            self.logger.info("Adding K8S CRD definition %s", crd.key)
            self.resource_definitions[crd.key] = crd

    def _populate_resource_definitions(self):
        k8s_def = self.resource_definitions_schema

        def k8s_resource_def_key(v: Mapping[str, Union[list, Mapping]]) -> Iterable[K8SResourceDefKey]:
            gvks = v.get("x-kubernetes-group-version-kind")
            if gvks:
                if isinstance(gvks, Mapping):
                    gvk = gvks
                    yield K8SResourceDefKey(gvk["group"],
                                            gvk["version"],
                                            gvk["kind"])
                else:
                    for gvk in gvks:
                        yield K8SResourceDefKey(gvk["group"],
                                                gvk["version"],
                                                gvk["kind"])

        paths = k8s_def["paths"]
        for path, actions in paths.items():
            path_rdk = None
            path_actions = []
            for action, action_details in actions.items():
                if action == "parameters":
                    continue
                rdks = list(k8s_resource_def_key(action_details))
                if rdks:
                    assert len(rdks) == 1
                    rdk = rdks[0]
                    if path_rdk:
                        if path_rdk != rdk:
                            raise ValueError(f"Encountered path action x-kubernetes-group-version-kind conflict: "
                                             f"{path}: {actions}")
                        path_actions.append(action_details["x-kubernetes-action"])
                    else:
                        path_rdk = rdk

            if path_rdk:
                rdef_paths = self.resource_paths.get(path_rdk)
                if not rdef_paths:
                    rdef_paths = {}
                    self.resource_paths[path_rdk] = rdef_paths
                rdef_paths[path] = actions

        for k, schema in k8s_def["definitions"].items():
            # This short-circuits the resolution of the references to the top of the document
            schema["definitions"] = k8s_def["definitions"]
            for key in k8s_resource_def_key(schema):
                for rdef in K8SResourceDef.from_manifest(key, schema, self.resource_paths):
                    self.resource_definitions[key] = rdef

karellen / kubernator / 24580658204

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous