flux_local.manifest

   1"""Representation of the contents of a cluster.
   2
   3A manifest may be built directly from the local context of a cluster, or may be
   4serialized and stored and checked into the cluster for use in other applications
   5e.g. such as writing management plan for resources.
   6"""
   7
   8import base64
   9from dataclasses import dataclass, field
  10import logging
  11from pathlib import Path
  12from typing import Any, Optional, cast, ClassVar
  13
  14import aiofiles
  15from mashumaro.codecs.yaml import yaml_decode, yaml_encode
  16from mashumaro import DataClassDictMixin, field_options
  17from mashumaro.config import BaseConfig
  18
  19from .exceptions import InputException
  20
  21__all__ = [
  22    "read_manifest",
  23    "write_manifest",
  24    "Manifest",
  25    "Cluster",
  26    "Kustomization",
  27    "HelmRepository",
  28    "HelmRelease",
  29    "HelmChart",
  30    "ConfigMap",
  31    "Secret",
  32]
  33
  34_LOGGER = logging.getLogger(__name__)
  35
  36
  37# Match a prefix of apiVersion to ensure we have the right type of object.
  38# We don't check specific versions for forward compatibility on upgrade.
  39FLUXTOMIZE_DOMAIN = "kustomize.toolkit.fluxcd.io"
  40KUSTOMIZE_DOMAIN = "kustomize.config.k8s.io"
  41HELM_REPO_DOMAIN = "source.toolkit.fluxcd.io"
  42HELM_RELEASE_DOMAIN = "helm.toolkit.fluxcd.io"
  43GIT_REPOSITORY_DOMAIN = "source.toolkit.fluxcd.io"
  44OCI_REPOSITORY_DOMAIN = "source.toolkit.fluxcd.io"
  45CRD_KIND = "CustomResourceDefinition"
  46SECRET_KIND = "Secret"
  47CONFIG_MAP_KIND = "ConfigMap"
  48DEFAULT_NAMESPACE = "flux-system"
  49VALUE_PLACEHOLDER_TEMPLATE = "..PLACEHOLDER_{name}.."
  50HELM_RELEASE = "HelmRelease"
  51HELM_REPO_KIND = "HelmRepository"
  52HELM_CHART = "HelmChart"
  53GIT_REPOSITORY = "GitRepository"
  54HELM_REPOSITORY = "HelmRepository"
  55OCI_REPOSITORY = "OCIRepository"
  56KUSTOMIZE_KIND = "Kustomization"
  57
  58
  59REPO_TYPE_DEFAULT = "default"
  60REPO_TYPE_OCI = "oci"
  61
  62# Strip any annotations from kustomize that contribute to diff noise when
  63# objects are re-ordered in the output
  64STRIP_ATTRIBUTES = [
  65    "config.kubernetes.io/index",
  66    "internal.config.kubernetes.io/index",
  67]
  68
  69
  70def _check_version(doc: dict[str, Any], version: str) -> None:
  71    """Assert that the resource has the specified version."""
  72    if not (api_version := doc.get("apiVersion")):
  73        raise InputException(f"Invalid object missing apiVersion: {doc}")
  74    if not api_version.startswith(version):
  75        raise InputException(f"Invalid object expected '{version}': {doc}")
  76
  77
  78@dataclass
  79class BaseManifest(DataClassDictMixin):
  80    """Base class for all manifest objects."""
  81
  82    def compact_dict(self) -> dict[str, Any]:
  83        """Return a compact dictionary representation of the object.
  84
  85        This is similar to `dict()` but with a specific implementation for serializing
  86        with variable fields removed.
  87        """
  88        return self.to_dict()
  89
  90    @classmethod
  91    def parse_yaml(cls, content: str) -> "BaseManifest":
  92        """Parse a serialized manifest."""
  93        return yaml_decode(content, cls)
  94
  95    def yaml(self, exclude: dict[str, Any] | None = None) -> str:
  96        """Return a YAML string representation of compact_dict."""
  97        return yaml_encode(self, self.__class__)  # type: ignore[return-value]
  98
  99    class Config(BaseConfig):
 100        omit_none = True
 101
 102
 103@dataclass(frozen=True, order=True)
 104class NamedResource:
 105    """Identifier for a kubernetes resource."""
 106
 107    kind: str
 108    namespace: str | None
 109    name: str
 110
 111    @property
 112    def namespaced_name(self) -> str:
 113        if self.namespace:
 114            return f"{self.namespace}/{self.name}"
 115        return self.name
 116
 117    def __str__(self) -> str:
 118        """Return the kind and namespaced name concatenated as an id."""
 119        return f"{self.kind}/{self.namespaced_name}"
 120
 121
 122@dataclass
 123class RawObject(BaseManifest):
 124    """Raw kubernetes object."""
 125
 126    kind: str
 127    """The kind of the object."""
 128
 129    api_version: str
 130    """The apiVersion of the object."""
 131
 132    name: str
 133    """The name of the object."""
 134
 135    namespace: str | None
 136    """The namespace of the object."""
 137
 138    spec: dict[str, Any] | None = None
 139    """The spec of the object."""
 140
 141    @classmethod
 142    def parse_doc(cls, doc: dict[str, Any]) -> "RawObject":
 143        """Parse a RawObject from a raw kubernetes object."""
 144        if not (api_version := doc.get("apiVersion")):
 145            raise InputException("Invalid object missing apiVersion: {doc}")
 146        if not (metadata := doc.get("metadata")):
 147            raise InputException(f"Invalid object missing metadata: {doc}")
 148        if not (name := metadata.get("name")):
 149            raise InputException(f"Invalid object missing metadata.name: {doc}")
 150        return cls(
 151            kind=doc["kind"],
 152            api_version=api_version,
 153            name=name,
 154            namespace=metadata.get("namespace", DEFAULT_NAMESPACE),
 155            spec=doc.get("spec", {}),
 156        )
 157
 158
 159@dataclass
 160class HelmChart(BaseManifest):
 161    """A representation of an instantiation of a chart for a HelmRelease."""
 162
 163    kind: ClassVar[str] = HELM_CHART
 164    """The kind of the object."""
 165
 166    name: str
 167    """The name of the chart within the HelmRepository."""
 168
 169    version: Optional[str] = field(metadata={"serialize": "omit"})
 170    """The version of the chart."""
 171
 172    repo_name: str
 173    """The short name of the repository."""
 174
 175    repo_namespace: str
 176    """The namespace of the repository."""
 177
 178    repo_kind: str = HELM_REPO_KIND
 179    """The kind of the soruceRef of the repository (e.g. HelmRepository, GitRepository)."""
 180
 181    @classmethod
 182    def parse_doc(cls, doc: dict[str, Any], default_namespace: str) -> "HelmChart":
 183        """Parse a HelmChart from a HelmRelease resource object."""
 184        _check_version(doc, HELM_RELEASE_DOMAIN)
 185        if not (spec := doc.get("spec")):
 186            raise InputException(f"Invalid {cls} missing spec: {doc}")
 187        chart_ref = spec.get("chartRef")
 188        chart = spec.get("chart")
 189        if not chart_ref and not chart:
 190            raise InputException(
 191                f"Invalid {cls} missing spec.chart or spec.chartRef: {doc}"
 192            )
 193        if chart_ref:
 194            if not (kind := chart_ref.get("kind")):
 195                raise InputException(f"Invalid {cls} missing spec.chartRef.kind: {doc}")
 196            if not (name := chart_ref.get("name")):
 197                raise InputException(f"Invalid {cls} missing spec.chartRef.name: {doc}")
 198
 199            return cls(
 200                name=name,
 201                version=None,
 202                repo_name=name,
 203                repo_namespace=chart_ref.get("namespace", default_namespace),
 204                repo_kind=kind,
 205            )
 206        if not (chart_spec := chart.get("spec")):
 207            raise InputException(f"Invalid {cls} missing spec.chart.spec: {doc}")
 208        if not (chart := chart_spec.get("chart")):
 209            raise InputException(f"Invalid {cls} missing spec.chart.spec.chart: {doc}")
 210        version = chart_spec.get("version")
 211        if not (source_ref := chart_spec.get("sourceRef")):
 212            raise InputException(
 213                f"Invalid {cls} missing spec.chart.spec.sourceRef: {doc}"
 214            )
 215        if "name" not in source_ref:
 216            raise InputException(f"Invalid {cls} missing sourceRef fields: {doc}")
 217        return cls(
 218            name=chart,
 219            version=version,
 220            repo_name=source_ref["name"],
 221            repo_namespace=source_ref.get("namespace", default_namespace),
 222            repo_kind=source_ref.get("kind", HELM_REPO_KIND),
 223        )
 224
 225    @property
 226    def repo_full_name(self) -> str:
 227        """Identifier for the HelmRepository."""
 228        return f"{self.repo_namespace}-{self.repo_name}"
 229
 230    @property
 231    def chart_name(self) -> str:
 232        """Identifier for the HelmChart."""
 233        return f"{self.repo_full_name}/{self.name}"
 234
 235
 236@dataclass
 237class ValuesReference(BaseManifest):
 238    """A reference to a resource containing values for a HelmRelease."""
 239
 240    kind: str
 241    """The kind of resource."""
 242
 243    name: str
 244    """The name of the resource."""
 245
 246    values_key: str = field(
 247        metadata=field_options(alias="valuesKey"), default="values.yaml"
 248    )
 249    """The key in the resource that contains the values."""
 250
 251    target_path: Optional[str] = field(
 252        metadata=field_options(alias="targetPath"), default=None
 253    )
 254    """The path in the HelmRelease values to store the values."""
 255
 256    optional: bool = False
 257    """Whether the reference is optional."""
 258
 259
 260@dataclass
 261class LocalObjectReference(BaseManifest):
 262    """A reference to a local object."""
 263
 264    name: str
 265    """The name of the object."""
 266
 267
 268@dataclass
 269class HelmRelease(BaseManifest):
 270    """A representation of a Flux HelmRelease."""
 271
 272    kind: ClassVar[str] = HELM_RELEASE
 273    """The kind of the object."""
 274
 275    name: str
 276    """The name of the HelmRelease."""
 277
 278    namespace: str
 279    """The namespace that owns the HelmRelease."""
 280
 281    chart: HelmChart
 282    """A mapping to a specific helm chart for this HelmRelease."""
 283
 284    target_namespace: str | None = field(metadata={"serialize": "omit"}, default=None)
 285    """The namespace to target when performing the operation."""
 286
 287    values: Optional[dict[str, Any]] = field(
 288        metadata={"serialize": "omit"}, default=None
 289    )
 290    """The values to install in the chart."""
 291
 292    values_from: Optional[list[ValuesReference]] = field(
 293        metadata={"serialize": "omit"}, default=None
 294    )
 295    """A list of values to reference from an ConfigMap or Secret."""
 296
 297    images: list[str] | None = field(default=None)
 298    """The list of images referenced in the HelmRelease."""
 299
 300    labels: dict[str, str] | None = field(metadata={"serialize": "omit"}, default=None)
 301    """A list of labels on the HelmRelease."""
 302
 303    disable_schema_validation: bool = field(
 304        metadata={"serialize": "omit"}, default=False
 305    )
 306    """Prevents Helm from validating the values against the JSON Schema."""
 307
 308    disable_openapi_validation: bool = field(
 309        metadata={"serialize": "omit"}, default=False
 310    )
 311    """Prevents Helm from validating the values against the Kubernetes OpenAPI Schema."""
 312
 313    @classmethod
 314    def parse_doc(cls, doc: dict[str, Any]) -> "HelmRelease":
 315        """Parse a HelmRelease from a kubernetes resource object."""
 316        _check_version(doc, HELM_RELEASE_DOMAIN)
 317        if not (metadata := doc.get("metadata")):
 318            raise InputException(f"Invalid {cls} missing metadata: {doc}")
 319        if not (name := metadata.get("name")):
 320            raise InputException(f"Invalid {cls} missing metadata.name: {doc}")
 321        if not (namespace := metadata.get("namespace")):
 322            raise InputException(f"Invalid {cls} missing metadata.namespace: {doc}")
 323        chart = HelmChart.parse_doc(doc, namespace)
 324        spec = doc["spec"]
 325        values_from: list[ValuesReference] | None = None
 326        if values_from_dict := spec.get("valuesFrom"):
 327            values_from = [
 328                ValuesReference.from_dict(subdoc) for subdoc in values_from_dict
 329            ]
 330        disable_schema_validation = any(
 331            bag.get("disableSchemaValidation")
 332            for key in ("install", "upgrade")
 333            if (bag := spec.get(key)) is not None
 334        )
 335        disable_openapi_validation = any(
 336            bag.get("disableOpenAPIValidation")
 337            for key in ("install", "upgrade")
 338            if (bag := spec.get(key)) is not None
 339        )
 340        return HelmRelease(
 341            name=name,
 342            namespace=namespace,
 343            target_namespace=spec.get("targetNamespace"),
 344            chart=chart,
 345            values=spec.get("values"),
 346            values_from=values_from,
 347            labels=metadata.get("labels"),
 348            disable_schema_validation=disable_schema_validation,
 349            disable_openapi_validation=disable_openapi_validation,
 350        )
 351
 352    @property
 353    def release_name(self) -> str:
 354        """Identifier for the HelmRelease."""
 355        return f"{self.namespace}-{self.name}"
 356
 357    @property
 358    def release_namespace(self) -> str:
 359        """Actual namespace where the HelmRelease will be installed to."""
 360        if self.target_namespace:
 361            return self.target_namespace
 362        return self.namespace
 363
 364    @property
 365    def repo_name(self) -> str:
 366        """Identifier for the HelmRepository identified in the HelmChart."""
 367        return f"{self.chart.repo_namespace}-{self.chart.repo_name}"
 368
 369    @property
 370    def namespaced_name(self) -> str:
 371        """Return the namespace and name concatenated as an id."""
 372        return f"{self.namespace}/{self.name}"
 373
 374    @property
 375    def resource_dependencies(self) -> list[NamedResource]:
 376        """Return the list of input dependencies for the HelmRelease."""
 377        deps = [
 378            NamedResource(
 379                kind=HELM_RELEASE,
 380                name=self.name,
 381                namespace=self.namespace,
 382            )
 383        ]
 384        if self.chart:
 385            deps.append(
 386                NamedResource(
 387                    kind=self.chart.repo_kind,
 388                    name=self.chart.repo_name,
 389                    namespace=self.chart.repo_namespace,
 390                )
 391            )
 392        names_seen = set()
 393        for ref in self.values_from or ():
 394            if ref.name in names_seen:
 395                continue
 396            names_seen.add(ref.name)
 397            deps.append(
 398                NamedResource(kind=ref.kind, name=ref.name, namespace=self.namespace)
 399            )
 400        return deps
 401
 402
 403@dataclass
 404class HelmRepository(BaseManifest):
 405    """A representation of a flux HelmRepository."""
 406
 407    kind: ClassVar[str] = HELM_REPO_KIND
 408    """The kind of the object."""
 409
 410    name: str
 411    """The name of the HelmRepository."""
 412
 413    namespace: str
 414    """The namespace of owning the HelmRepository."""
 415
 416    url: str
 417    """The URL to the repository of helm charts."""
 418
 419    repo_type: str | None = None
 420    """The type of the HelmRepository."""
 421
 422    @classmethod
 423    def parse_doc(cls, doc: dict[str, Any]) -> "HelmRepository":
 424        """Parse a HelmRepository from a kubernetes resource."""
 425        _check_version(doc, HELM_REPO_DOMAIN)
 426        if not (metadata := doc.get("metadata")):
 427            raise InputException(f"Invalid {cls} missing metadata: {doc}")
 428        if not (name := metadata.get("name")):
 429            raise InputException(f"Invalid {cls} missing metadata.name: {doc}")
 430        if not (namespace := metadata.get("namespace")):
 431            raise InputException(f"Invalid {cls} missing metadata.namespace: {doc}")
 432        if not (spec := doc.get("spec")):
 433            raise InputException(f"Invalid {cls} missing spec: {doc}")
 434        if not (url := spec.get("url")):
 435            raise InputException(f"Invalid {cls} missing spec.url: {doc}")
 436        return cls(
 437            name=name,
 438            namespace=namespace,
 439            url=url,
 440            repo_type=spec.get("type", REPO_TYPE_DEFAULT),
 441        )
 442
 443    @property
 444    def repo_name(self) -> str:
 445        """Identifier for the HelmRepository."""
 446        return f"{self.namespace}-{self.name}"
 447
 448    def helm_chart_name(self, chart: HelmChart) -> str:
 449        """Get the name or URL for a specific chart for helm template."""
 450        if self.repo_type == REPO_TYPE_OCI:
 451            # For OCI repositories, we need to append the chart short name to the URL
 452            return f"{self.url}/{chart.name}"
 453        # For regular helm repositories, we just return the full chart name
 454        return chart.chart_name
 455
 456
 457@dataclass
 458class GitRepositoryRef:
 459    """GitRepositoryRef defines the Git ref used for pull and checkout operations."""
 460
 461    branch: str | None = field(default=None)
 462    """The Git branch to checkout, defaults to master."""
 463
 464    tag: str | None = field(default=None)
 465    """The Git tag to checkout."""
 466
 467    semver: str | None = field(default=None)
 468    """The Git tag semver expression."""
 469
 470    commit: str | None = field(default=None)
 471    """The Git commit SHA to checkout."""
 472
 473    @classmethod
 474    def parse_doc(cls, doc: dict[str, Any]) -> "GitRepositoryRef":
 475        """Parse a GitRepositoryRef from a kubernetes resource."""
 476        return cls(
 477            branch=doc.get("branch"),
 478            tag=doc.get("tag"),
 479            semver=doc.get("semver"),
 480            commit=doc.get("commit"),
 481        )
 482
 483    @property
 484    def ref_str(self) -> str | None:
 485        """Get the reference string for the GitRepository."""
 486        if self.commit:
 487            return f"commit:{self.commit}"
 488        if self.tag:
 489            return f"tag:{self.tag}"
 490        if self.branch:
 491            return f"branch:{self.branch}"
 492        if self.semver:
 493            return f"semver:{self.semver}"
 494        return None
 495
 496    class Config(BaseConfig):
 497        omit_none = True
 498
 499
 500@dataclass
 501class GitRepository(BaseManifest):
 502    """GitRepository represents a Git repository."""
 503
 504    kind: ClassVar[str] = GIT_REPOSITORY
 505    """The kind of the object."""
 506
 507    name: str
 508    """The name of the GitRepository."""
 509
 510    namespace: str
 511    """The namespace of owning the GitRepository."""
 512
 513    url: str
 514    """The URL to the repository."""
 515
 516    ref: GitRepositoryRef | None = None
 517    """The Git reference to use for pull and checkout operations."""
 518
 519    @classmethod
 520    def parse_doc(cls, doc: dict[str, Any]) -> "GitRepository":
 521        """Parse a GitRepxository from a kubernetes resource."""
 522        _check_version(doc, GIT_REPOSITORY_DOMAIN)
 523        if not (metadata := doc.get("metadata")):
 524            raise InputException(f"Invalid {cls} missing metadata: {doc}")
 525        if not (name := metadata.get("name")):
 526            raise InputException(f"Invalid {cls} missing metadata.name: {doc}")
 527        if not (namespace := metadata.get("namespace")):
 528            raise InputException(f"Invalid {cls} missing metadata.namespace: {doc}")
 529        if not (spec := doc.get("spec")):
 530            raise InputException(f"Invalid {cls} missing spec: {doc}")
 531        if not (url := spec.get("url")):
 532            raise InputException(f"Invalid {cls} missing spec.url: {doc}")
 533
 534        ref = None
 535        if ref_dict := spec.get("ref"):
 536            ref = GitRepositoryRef.parse_doc(ref_dict)
 537
 538        return cls(
 539            name=name,
 540            namespace=namespace,
 541            url=url,
 542            ref=ref,
 543        )
 544
 545    @property
 546    def repo_name(self) -> str:
 547        """Identifier for the GitRepository."""
 548        return f"{self.namespace}-{self.name}"
 549
 550
 551@dataclass
 552class OCIRepositoryRef:
 553    """OCIRepositoryRef defines the image reference for the OCIRepository's URL."""
 554
 555    digest: str | None = None
 556    """The image digest to pull, takes precedence over SemVer."""
 557
 558    tag: str | None = None
 559    """The image tag to pull, defaults to latest."""
 560
 561    semver: str | None = None
 562    """The range of tags to pull selecting the latest within the range."""
 563
 564    semver_filter: str | None = None
 565    """A regex pattern to filter the tags within the SemVer range."""
 566
 567    @classmethod
 568    def parse_doc(cls, doc: dict[str, Any]) -> "OCIRepositoryRef":
 569        """Parse a dictionary into an OCIRepositoryRef."""
 570        return cls(
 571            digest=doc.get("digest"),
 572            tag=doc.get("tag"),
 573            semver=doc.get("semver"),
 574            semver_filter=doc.get("semverFilter"),
 575        )
 576
 577    class Config(BaseConfig):
 578        omit_none = True
 579
 580
 581@dataclass
 582class OCIRepository(BaseManifest):
 583    """A representation of a flux OCIRepository."""
 584
 585    kind: ClassVar[str] = OCI_REPOSITORY
 586    """The kind of the object."""
 587
 588    name: str
 589    """The name of the OCIRepository."""
 590
 591    namespace: str
 592    """The namespace of owning the OCIRepository."""
 593
 594    url: str
 595    """The URL to the repository."""
 596
 597    ref: OCIRepositoryRef | None = None
 598    """The OCI reference (tag or digest) to use."""
 599
 600    secret_ref: LocalObjectReference | None = field(
 601        metadata=field_options(alias="secretRef"), default=None
 602    )
 603    """The local secret reference."""
 604
 605    @classmethod
 606    def parse_doc(cls, doc: dict[str, Any]) -> "OCIRepository":
 607        """Parse a HelmRepository from a kubernetes resource."""
 608        _check_version(doc, OCI_REPOSITORY_DOMAIN)
 609        if not (metadata := doc.get("metadata")):
 610            raise InputException(f"Invalid {cls} missing metadata: {doc}")
 611        if not (name := metadata.get("name")):
 612            raise InputException(f"Invalid {cls} missing metadata.name: {doc}")
 613        if not (namespace := metadata.get("namespace")):
 614            raise InputException(f"Invalid {cls} missing metadata.namespace: {doc}")
 615        if not (spec := doc.get("spec")):
 616            raise InputException(f"Invalid {cls} missing spec: {doc}")
 617        if not (url := spec.get("url")):
 618            raise InputException(f"Invalid {cls} missing spec.url: {doc}")
 619        repo_ref: OCIRepositoryRef | None = None
 620        if ref := spec.get("ref"):
 621            repo_ref = OCIRepositoryRef.parse_doc(ref)
 622        secret_ref: LocalObjectReference | None = None
 623        if secret_ref_dict := spec.get("secretRef"):
 624            secret_ref = LocalObjectReference.from_dict(secret_ref_dict)
 625        return cls(
 626            name=name,
 627            namespace=namespace,
 628            url=url,
 629            ref=repo_ref,
 630            secret_ref=secret_ref,
 631        )
 632
 633    def version(self) -> str | None:
 634        """Get the version of the OCI repository."""
 635        if self.ref is not None:
 636            if self.ref.semver_filter:
 637                raise ValueError(
 638                    f"OCIRepository has unsupported field semvar_filter: {self.ref}"
 639                )
 640            if self.ref.digest:
 641                return self.ref.digest
 642            if self.ref.tag:
 643                return self.ref.tag
 644            if self.ref.semver:
 645                return self.ref.semver
 646        return None
 647
 648    def versioned_url(self) -> str:
 649        """Get the URL with the version."""
 650        if self.ref is None:
 651            return self.url
 652        if self.ref.digest:
 653            return f"{self.url}@{self.ref.digest}"
 654        if self.ref.tag:
 655            return f"{self.url}:{self.ref.tag}"
 656        if self.ref.semver:
 657            return f"{self.url}:{self.ref.semver}"
 658        return self.url
 659
 660    @property
 661    def repo_name(self) -> str:
 662        """Identifier for the OCIRepository."""
 663        return f"{self.namespace}-{self.name}"
 664
 665
 666@dataclass
 667class ConfigMap(BaseManifest):
 668    """A ConfigMap is an API object used to store data in key-value pairs."""
 669
 670    kind: ClassVar[str] = CONFIG_MAP_KIND
 671    """The kind of the ConfigMap."""
 672
 673    name: str
 674    """The name of the ConfigMap."""
 675
 676    namespace: str | None = None
 677    """The namespace of the kustomization."""
 678
 679    data: dict[str, Any] | None = field(metadata={"serialize": "omit"}, default=None)
 680    """The data in the ConfigMap."""
 681
 682    binary_data: dict[str, Any] | None = field(
 683        metadata={"serialize": "omit"}, default=None
 684    )
 685    """The binary data in the ConfigMap."""
 686
 687    @classmethod
 688    def parse_doc(cls, doc: dict[str, Any]) -> "ConfigMap":
 689        """Parse a config map object from a kubernetes resource."""
 690        _check_version(doc, "v1")
 691        if not (metadata := doc.get("metadata")):
 692            raise InputException(f"Invalid {cls} missing metadata: {doc}")
 693        if not (name := metadata.get("name")):
 694            raise InputException(f"Invalid {cls} missing metadata.name: {doc}")
 695        namespace = metadata.get("namespace")
 696        return ConfigMap(
 697            name=name,
 698            namespace=namespace,
 699            data=doc.get("data"),
 700            binary_data=doc.get("binaryData"),
 701        )
 702
 703
 704@dataclass
 705class Secret(BaseManifest):
 706    """A Secret contains a small amount of sensitive data."""
 707
 708    kind: ClassVar[str] = SECRET_KIND
 709    """The kind of the Secret."""
 710
 711    name: str
 712    """The name of the Secret."""
 713
 714    namespace: str | None = None
 715    """The namespace of the Secret."""
 716
 717    data: dict[str, Any] | None = field(metadata={"serialize": "omit"}, default=None)
 718    """The data in the Secret."""
 719
 720    string_data: dict[str, Any] | None = field(
 721        metadata={"serialize": "omit"}, default=None
 722    )
 723    """The string data in the Secret."""
 724
 725    @classmethod
 726    def parse_doc(cls, doc: dict[str, Any], *, wipe_secrets: bool = True) -> "Secret":
 727        """Parse a secret object from a kubernetes resource."""
 728        _check_version(doc, "v1")
 729        if not (metadata := doc.get("metadata")):
 730            raise InputException(f"Invalid {cls} missing metadata: {doc}")
 731        if not (name := metadata.get("name")):
 732            raise InputException(f"Invalid {cls} missing metadata.name: {doc}")
 733        namespace = metadata.get("namespace")
 734        # While secrets are not typically stored in the cluster, we replace with
 735        # placeholder values anyway.
 736        data = doc.get("data")
 737        if data and wipe_secrets:
 738            for key, value in data.items():
 739                data[key] = base64.b64encode(
 740                    VALUE_PLACEHOLDER_TEMPLATE.format(name=key).encode()
 741                )
 742        string_data = doc.get("stringData")
 743        if string_data and wipe_secrets:
 744            for key, value in string_data.items():
 745                string_data[key] = VALUE_PLACEHOLDER_TEMPLATE.format(name=key)
 746        return Secret(
 747            name=name, namespace=namespace, data=data, string_data=string_data
 748        )
 749
 750
 751@dataclass
 752class SubstituteReference(BaseManifest):
 753    """SubstituteReference contains a reference to a resource containing the variables name and value."""
 754
 755    kind: str
 756    """The kind of resource."""
 757
 758    name: str
 759    """The name of the resource."""
 760
 761    optional: bool = False
 762    """Whether the reference is optional."""
 763
 764
 765@dataclass
 766class Kustomization(BaseManifest):
 767    """A Kustomization is a set of declared cluster artifacts.
 768
 769    This represents a flux Kustomization that points to a path that
 770    contains typical `kustomize` Kustomizations on local disk that
 771    may be flat or contain overlays.
 772    """
 773
 774    kind: ClassVar[str] = KUSTOMIZE_KIND
 775    """The kind of the object."""
 776
 777    name: str
 778    """The name of the kustomization."""
 779
 780    namespace: str | None
 781    """The namespace of the kustomization."""
 782
 783    path: str
 784    """The local repo path to the kustomization contents."""
 785
 786    helm_repos: list[HelmRepository] = field(default_factory=list)
 787    """The set of HelmRepositories represented in this kustomization."""
 788
 789    oci_repos: list[OCIRepository] = field(default_factory=list)
 790    """The set of OCIRepositories represented in this kustomization."""
 791
 792    helm_releases: list[HelmRelease] = field(default_factory=list)
 793    """The set of HelmRelease represented in this kustomization."""
 794
 795    config_maps: list[ConfigMap] = field(default_factory=list)
 796    """The list of config maps referenced in the kustomization."""
 797
 798    secrets: list[Secret] = field(default_factory=list)
 799    """The list of secrets referenced in the kustomization."""
 800
 801    source_path: str | None = field(metadata={"serialize": "omit"}, default=None)
 802    """Optional source path for this Kustomization, relative to the build path."""
 803
 804    source_kind: str | None = field(metadata={"serialize": "omit"}, default=None)
 805    """The sourceRef kind that provides this Kustomization e.g. GitRepository etc."""
 806
 807    source_name: str | None = field(metadata={"serialize": "omit"}, default=None)
 808    """The name of the sourceRef that provides this Kustomization."""
 809
 810    source_namespace: str | None = field(metadata={"serialize": "omit"}, default=None)
 811    """The namespace of the sourceRef that provides this Kustomization."""
 812
 813    target_namespace: str | None = field(metadata={"serialize": "omit"}, default=None)
 814    """The namespace to target when performing the operation."""
 815
 816    contents: dict[str, Any] | None = field(
 817        metadata={"serialize": "omit"}, default=None
 818    )
 819    """Contents of the raw Kustomization document."""
 820
 821    images: list[str] | None = field(default=None)
 822    """The list of images referenced in the kustomization."""
 823
 824    postbuild_substitute: Optional[dict[str, Any]] = field(
 825        metadata={"serialize": "omit"}, default=None
 826    )
 827    """A map of key/value pairs to substitute into the final YAML manifest, after building."""
 828
 829    postbuild_substitute_from: Optional[list[SubstituteReference]] = field(
 830        metadata={"serialize": "omit"}, default=None
 831    )
 832    """A list of substitutions to reference from an ConfigMap or Secret."""
 833
 834    depends_on: list[str] | None = field(metadata={"serialize": "omit"}, default=None)
 835    """A list of namespaced names that this Kustomization depends on."""
 836
 837    labels: dict[str, str] | None = field(metadata={"serialize": "omit"}, default=None)
 838    """A list of labels on the Kustomization."""
 839
 840    @classmethod
 841    def parse_doc(cls, doc: dict[str, Any]) -> "Kustomization":
 842        """Parse a partial Kustomization from a kubernetes resource."""
 843        _check_version(doc, FLUXTOMIZE_DOMAIN)
 844        if not (metadata := doc.get("metadata")):
 845            raise InputException(f"Invalid {cls} missing metadata: {doc}")
 846        if not (name := metadata.get("name")):
 847            raise InputException(f"Invalid {cls} missing metadata.name: {doc}")
 848        if not (namespace := metadata.get("namespace")):
 849            raise InputException(f"Invalid {cls} missing metadata.namespace: {doc}")
 850        if not (spec := doc.get("spec")):
 851            raise InputException(f"Invalid {cls} missing spec: {doc}")
 852        path = spec.get("path", "")
 853        source_path = metadata.get("annotations", {}).get("config.kubernetes.io/path")
 854        source_ref = spec.get("sourceRef", {})
 855        postbuild = spec.get("postBuild", {})
 856        substitute_from: list[SubstituteReference] | None = None
 857        if substitute_from_dict := postbuild.get("substituteFrom"):
 858            substitute_from = [
 859                SubstituteReference(**subdoc) for subdoc in substitute_from_dict
 860            ]
 861        depends_on = []
 862        for dependency in spec.get("dependsOn", ()):
 863            if not (dep_name := dependency.get("name")):
 864                raise InputException(f"Invalid {cls} missing dependsOn.name: {doc}")
 865            dep_namespace = dependency.get("namespace", namespace)
 866            depends_on.append(f"{dep_namespace}/{dep_name}")
 867        return Kustomization(
 868            name=name,
 869            namespace=namespace,
 870            path=path,
 871            source_path=source_path,
 872            source_kind=source_ref.get("kind"),
 873            source_name=source_ref.get("name"),
 874            source_namespace=source_ref.get("namespace", namespace),
 875            target_namespace=spec.get("targetNamespace"),
 876            contents=doc,
 877            postbuild_substitute=postbuild.get("substitute"),
 878            postbuild_substitute_from=substitute_from,
 879            depends_on=depends_on,
 880            labels=metadata.get("labels"),
 881        )
 882
 883    @property
 884    def id_name(self) -> str:
 885        """Identifier for the Kustomization in tests"""
 886        return f"{self.path}"
 887
 888    @property
 889    def namespaced_name(self) -> str:
 890        """Return the namespace and name concatenated as an id."""
 891        return f"{self.namespace}/{self.name}"
 892
 893    def validate_depends_on(self, all_ks: set[str]) -> None:
 894        """Validate depends_on values are all correct given the list of Kustomizations."""
 895        depends_on = set(self.depends_on or {})
 896        if missing := (depends_on - all_ks):
 897            _LOGGER.warning(
 898                "Kustomization %s has dependsOn with invalid names: %s",
 899                self.namespaced_name,
 900                missing,
 901            )
 902            self.depends_on = list(depends_on - missing)
 903
 904    def update_postbuild_substitutions(self, substitutions: dict[str, Any]) -> None:
 905        """Update the postBuild.substitutions in the extracted values and raw doc contents."""
 906        if self.postbuild_substitute is None:
 907            self.postbuild_substitute = {}
 908        self.postbuild_substitute.update(substitutions)
 909        if self.contents:
 910            post_build = self.contents["spec"]["postBuild"]
 911            if (substitute := post_build.get("substitute")) is None:
 912                substitute = {}
 913                post_build["substitute"] = substitute
 914            substitute.update(substitutions)
 915
 916
 917@dataclass
 918class Cluster(BaseManifest):
 919    """A set of nodes that run containerized applications.
 920
 921    Many flux git repos will only have a single flux cluster, though
 922    a repo may also contain multiple (e.g. dev an prod).
 923    """
 924
 925    path: str
 926    """The local git repo path to the Kustomization objects for the cluster."""
 927
 928    kustomizations: list[Kustomization] = field(default_factory=list)
 929    """A list of flux Kustomizations for the cluster."""
 930
 931    @property
 932    def id_name(self) -> str:
 933        """Identifier for the Cluster in tests."""
 934        return f"{self.path}"
 935
 936    @property
 937    def helm_repos(self) -> list[HelmRepository]:
 938        """Return the list of HelmRepository objects from all Kustomizations."""
 939        return [
 940            repo
 941            for kustomization in self.kustomizations
 942            for repo in kustomization.helm_repos
 943        ]
 944
 945    @property
 946    def oci_repos(self) -> list[OCIRepository]:
 947        """Return the list of OCIRepository objects from all Kustomizations."""
 948        return [
 949            repo
 950            for kustomization in self.kustomizations
 951            for repo in kustomization.oci_repos
 952        ]
 953
 954    @property
 955    def helm_releases(self) -> list[HelmRelease]:
 956        """Return the list of HelmRelease objects from all Kustomizations."""
 957        return [
 958            release
 959            for kustomization in self.kustomizations
 960            for release in kustomization.helm_releases
 961        ]
 962
 963
 964@dataclass
 965class Manifest(BaseManifest):
 966    """Holds information about cluster and applications contained in a repo."""
 967
 968    clusters: list[Cluster]
 969    """A list of Clusters represented in the repo."""
 970
 971
 972async def read_manifest(manifest_path: Path) -> Manifest:
 973    """Return the contents of a serialized manifest file.
 974
 975    A manifest file is typically created by `flux-local get cluster -o yaml` or
 976    similar command.
 977    """
 978    async with aiofiles.open(str(manifest_path)) as manifest_file:
 979        content = await manifest_file.read()
 980        if not content:
 981            raise ValueError("validation error for Manifest file {manifest_path}")
 982        return cast(Manifest, Manifest.parse_yaml(content))
 983
 984
 985async def write_manifest(manifest_path: Path, manifest: Manifest) -> None:
 986    """Write the specified manifest content to disk."""
 987    content = manifest.yaml()
 988    async with aiofiles.open(str(manifest_path), mode="w") as manifest_file:
 989        await manifest_file.write(content)
 990
 991
 992async def update_manifest(manifest_path: Path, manifest: Manifest) -> None:
 993    """Write the specified manifest only if changed."""
 994    new_content = manifest.yaml()
 995    async with aiofiles.open(str(manifest_path)) as manifest_file:
 996        content = await manifest_file.read()
 997    if content == new_content:
 998        return
 999    async with aiofiles.open(str(manifest_path), mode="w") as manifest_file:
1000        await manifest_file.write(new_content)
1001
1002
1003def parse_raw_obj(obj: dict[str, Any], *, wipe_secrets: bool = True) -> BaseManifest:
1004    """Parse a raw kubernetes object into a BaseManifest."""
1005    if not (kind := obj.get("kind")):
1006        raise InputException(f"Invalid object missing kind: {obj}")
1007    if not (api_version := obj.get("apiVersion")):
1008        raise InputException(f"Invalid object missing apiVersion: {obj}")
1009    if kind == KUSTOMIZE_KIND and api_version.startswith(FLUXTOMIZE_DOMAIN):
1010        return Kustomization.parse_doc(obj)
1011    if kind == HELM_RELEASE:
1012        return HelmRelease.parse_doc(obj)
1013    if kind == HELM_REPOSITORY:
1014        return HelmRepository.parse_doc(obj)
1015    if kind == GIT_REPOSITORY:
1016        return GitRepository.parse_doc(obj)
1017    if kind == OCI_REPOSITORY:
1018        return OCIRepository.parse_doc(obj)
1019    if kind == CONFIG_MAP_KIND:
1020        return ConfigMap.parse_doc(obj)
1021    if kind == SECRET_KIND:
1022        return Secret.parse_doc(obj, wipe_secrets=wipe_secrets)
1023    return RawObject.parse_doc(obj)
1024
1025
1026def is_kustomization(obj: dict[str, Any]) -> bool:
1027    """Check if the object is a Kustomization."""
1028    return obj.get("kind") == KUSTOMIZE_KIND and obj.get("apiVersion", "").startswith(
1029        KUSTOMIZE_DOMAIN
1030    )
1031
1032
1033def _strip_attrs(metadata: dict[str, Any], strip_attributes: list[str]) -> None:
1034    """Update the resource object, stripping any requested labels to simplify diff."""
1035
1036    for attr_key in ("annotations", "labels"):
1037        if not (val := metadata.get(attr_key)):
1038            continue
1039        for key in strip_attributes:
1040            if key in val:
1041                del val[key]
1042            if not val:
1043                del metadata[attr_key]
1044                break
1045
1046
1047def strip_resource_attributes(
1048    resource: dict[str, Any], strip_attributes: list[str]
1049) -> None:
1050    """Strip any annotations from kustomize that contribute to diff noise when objects are re-ordered in the output."""
1051    _strip_attrs(resource["metadata"], strip_attributes)
1052    # Remove common noisy labels in commonly used templates
1053    if (
1054        (spec := resource.get("spec"))
1055        and (templ := spec.get("template"))
1056        and (meta := templ.get("metadata"))
1057    ):
1058        _strip_attrs(meta, strip_attributes)
1059    if (
1060        resource["kind"] == "List"
1061        and (items := resource.get("items"))
1062        and isinstance(items, list)
1063    ):
1064        for item in items:
1065            if not (item_meta := item.get("metadata")):
1066                continue
1067            _strip_attrs(item_meta, strip_attributes)