19040252246

Committed 03 Nov 2025 03:41PM UTC coverage: 75.065% (+0.04%) from 75.024%

Build # 19040252246

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #90 from karellen/fix_help_empty_repo_update

Helm fails when update is issued but no repositories are available

Run Details

639 of 1010 branches covered (63.27%)

Branch coverage included in aggregate %.

2 of 2 new or added lines in 1 file covered. (100.0%)

33 existing lines in 3 files now uncovered.

2519 of 3197 relevant lines covered (78.79%)

4.72 hits per line

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

79.0

/src/main/python/kubernator/plugins/helm.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 json
import logging
import os
import tarfile
import tempfile
from hashlib import sha256
from pathlib import Path
from shutil import which, copy
from typing import Sequence

import yaml
from jsonschema import Draft7Validator

from kubernator.api import (KubernatorPlugin, Globs, StripNL,
                            scan_dir,
                            load_file,
                            FileType,
                            calling_frame_source,
                            validator_with_defaults,
                            get_golang_os,
                            get_golang_machine,
                            prepend_os_path, TemplateEngine
                            )
from kubernator.plugins.k8s_api import K8SResource
from kubernator.proc import DEVNULL

logger = logging.getLogger("kubernator.helm")
proc_logger = logger.getChild("proc")
stdout_logger = StripNL(proc_logger.info)
stderr_logger = StripNL(proc_logger.warning)

HELM_SCHEMA = {
    "properties": {
        "repository": {
            "type": "string",
            "description": "repository hosting the charts"
        },
        "chart": {
            "type": "string",
            "description": "chart name within the repository"
        },
        "version": {
            "type": "string",
            "description": "chart version"
        },
        "name": {
            "type": "string",
            "description": "name of the particular release"
        },
        "namespace": {
            "type": "string",
            "description": "namespace where the chart is going to be deployed"
        },
        "include-crds": {
            "type": "boolean",
            "default": True,
            "description": "generate CRDs for the helm"
        },
        "values": {
            "type": "object",
            "additionalProperties": True,
            "description": "inline Helm chart values"
        },
        "values-file": {
            "type": "string",
            "description": "reference to the file containing Helm chart values"
        }
    },
    "type": "object",
    "required": ["chart", "name", "namespace"]
}

Draft7Validator.check_schema(HELM_SCHEMA)
HELM_VALIDATOR_CLS = validator_with_defaults(Draft7Validator)
HELM_VALIDATOR = HELM_VALIDATOR_CLS(HELM_SCHEMA, format_checker=Draft7Validator.FORMAT_CHECKER)


class HelmPlugin(KubernatorPlugin):
    logger = logger

    _name = "helm"

    def __init__(self):
        self.context = None
        self.repositories = set()
        self.helm_dir = None
        self.template_engine = TemplateEngine(logger)

        self._repositories_not_populated = True

    def set_context(self, context):
        self.context = context

    def stanza(self):
        context = self.context
        stanza = [context.helm.helm_file, f"--kubeconfig={context.kubeconfig.kubeconfig}"]
        if logger.getEffectiveLevel() < logging.INFO:
            stanza.append("--debug")
        return stanza

    def register(self, version=None):
        context = self.context
        context.app.register_plugin("kubeconfig")
        context.app.register_plugin("k8s")

        if version:
            # Download and use specific version
            helm_url = f"https://get.helm.sh/helm-v{version}-{get_golang_os()}-{get_golang_machine()}.tar.gz"
            helm_file_dl, _ = context.app.download_remote_file(logger, helm_url, "bin")
            helm_file_dl = str(helm_file_dl)
            self.helm_dir = tempfile.TemporaryDirectory()
            context.app.register_cleanup(self.helm_dir)

            helm_file = str(Path(self.helm_dir.name) / "helm")
            helm_tar = tarfile.open(helm_file_dl)
            helm_tar.extractall(self.helm_dir.name)

            copy(Path(self.helm_dir.name) / f"{get_golang_os()}-{get_golang_machine()}" / "helm", helm_file)

            os.chmod(helm_file, 0o500)
            prepend_os_path(self.helm_dir.name)
        else:
            # Use current version
            helm_file = which("helm")
            if not helm_file:
                raise RuntimeError("`helm` cannot be found and no version has been specified")

            logger.debug("Found Helm in %r", helm_file)

        context.globals.helm = dict(default_includes=Globs(["*.helm.yaml", "*.helm.yml"], True),
                                    default_excludes=Globs([".*"], True),
                                    namespace_transformer=True,
                                    helm_file=helm_file,
                                    stanza=self.stanza,
                                    add_helm_template=self.add_helm_template,
                                    add_helm=self.add_helm,
                                    )

    def handle_init(self):
        version = self.context.app.run_capturing_out(self.stanza() + ["version", "--template", "{{.Version}}"],
                                                     logger.error)
        logger.info("Found Helm version %s", version)

    def handle_start(self):
        pass

    def handle_before_dir(self, cwd: Path):
        context = self.context

        context.helm.default_includes = Globs(context.helm.default_includes)
        context.helm.default_excludes = Globs(context.helm.default_excludes)
        context.helm.includes = Globs(context.helm.default_includes)
        context.helm.excludes = Globs(context.helm.default_excludes)

        # Exclude Helm YAMLs from K8S resource loading
        context.k8s.excludes.add("*.helm.yaml")
        context.k8s.excludes.add("*.helm.yml")

    def handle_after_dir(self, cwd: Path):
        context = self.context
        helm = context.helm

        for f in scan_dir(logger, cwd, lambda d: d.is_file(), helm.excludes, helm.includes):
            p = cwd / f.name
            display_p = context.app.display_path(p)
            logger.debug("Adding Helm template from %s", display_p)

            helm_templates = load_file(logger, p, FileType.YAML, display_p,
                                       self.template_engine,
                                       {"ktor": context})

            for helm_template in helm_templates:
                self._add_helm(helm_template, display_p)

    def add_helm_template(self, template):
        return self._add_helm(template, calling_frame_source())

    def add_helm(self, **kwargs):
        return self._internal_add_helm(calling_frame_source(), **kwargs)

    def _add_helm(self, template, source):
        errors = list(HELM_VALIDATOR.iter_errors(template))
        if errors:
            for error in errors:
                self.logger.error("Error detected in Helm template from %s", source, exc_info=error)
            raise errors[0]

        return self._internal_add_helm(source, **{k.replace("-", "_"): v for k, v in template.items()})

    def _add_repository(self, repository: str):
        def _update_repositories():
            if self.repositories:
                self.context.app.run(self.stanza() + ["repo", "update"],
                                     stdout_logger,
                                     stderr_logger).wait()

        if self._repositories_not_populated:
            preexisting_repositories = json.loads(
                self.context.app.run_capturing_out(self.stanza() + ["repo", "list", "-o", "json"], stderr_logger))
            for pr in preexisting_repositories:
                repo_name = pr["name"]
                repo_url = pr["url"]
                logger.debug("Recording pre-existing repository %s mapping %s", repo_name, repo_url)
                self.repositories.add(repo_name)
            _update_repositories()
            self._repositories_not_populated = False

        repository_hash = sha256(repository.encode("UTF-8")).hexdigest()
        logger.debug("Repository %s mapping to %s", repository, repository_hash)
        if repository_hash not in self.repositories:
            logger.info("Adding and updating repository %s mapping to %s", repository, repository_hash)
            self.context.app.run(self.stanza() + ["repo", "add", repository_hash, repository],
                                 stdout_logger,
                                 stderr_logger).wait()
            _update_repositories()
            self.repositories.add(repository_hash)

        return repository_hash

    def _internal_add_helm(self, source, *, chart, name, namespace, include_crds,
                           values=None, values_file=None, repository=None, version=None):
        if values and values_file:
            raise RuntimeError(f"In {source} either values or values file may be specified, but not both")

        if (repository and chart and chart.startswith("oci://") or
                not repository and chart and not chart.startswith("oci://")):
            raise RuntimeError(
                f"In {source} either repository must be specified or OCI-chart must be used, but not both")

        if not version and repository:
            raise RuntimeError(f"In {source} version must be specified unless OCI-chart is used")

        if values_file:
            values_file = Path(values_file)
            if not values_file.is_absolute():
                values_file = self.context.app.cwd / values_file
            values = list(load_file(logger, values_file, FileType.YAML,
                                    template_engine=self.template_engine,
                                    template_context={"ktor": self.context,
                                                      "helm": {"chart": chart,
                                                               "name": name,
                                                               "namespace": namespace,
                                                               "include_crds": include_crds,
                                                               "repository": repository,
                                                               "version": version,
                                                               }}))
            values = values[0] if values else {}

        version_spec = []
        if repository:
            repository_hash = self._add_repository(repository)
            chart_name = f"{repository_hash}/{chart}"
        else:
            chart_name = chart

        if version:
            version_spec = ["--version", version]

        stdin = DEVNULL

        if values:
            def write_stdin():
                return json.dumps(values)

            stdin = write_stdin

        resources = self.context.app.run_capturing_out(self.stanza() +
                                                       ["template",
                                                        name,
                                                        chart_name,
                                                        "-n", namespace,
                                                        "-a", ",".join(self.context.k8s.get_api_versions())
                                                        ] +
                                                       version_spec +
                                                       (["--include-crds"] if include_crds else []) +
                                                       ["-f", "-"],
                                                       stderr_logger,
                                                       stdin=stdin,
                                                       )

        def helm_namespace_transformer(resources: Sequence[K8SResource],
                                       resource: K8SResource):
            if resource.rdef.namespaced and not resource.namespace:
                resource.namespace = namespace
                return resource

        if self.context.helm.namespace_transformer:
            self.context.k8s.add_transformer(helm_namespace_transformer)

        self.context.k8s.add_resources(yaml.safe_load_all(resources), source)

        if self.context.helm.namespace_transformer:
            self.context.k8s.remove_transformer(helm_namespace_transformer)

    def __repr__(self):
        return "Helm Plugin"

1	# -- coding: utf-8 --
2	#
3	# Copyright 2020 Express Systems USA, Inc
4	# Copyright 2021 Karellen, Inc.
5	#
6	# Licensed under the Apache License, Version 2.0 (the "License");
7	# you may not use this file except in compliance with the License.
8	# You may obtain a copy of the License at
9	#
10	# http://www.apache.org/licenses/LICENSE-2.0
11	#
12	# Unless required by applicable law or agreed to in writing, software
13	# distributed under the License is distributed on an "AS IS" BASIS,
14	# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15	# See the License for the specific language governing permissions and
16	# limitations under the License.
17	#
18
19	import json	6✔
20	import logging	6✔
21	import os	6✔
22	import tarfile	6✔
23	import tempfile	6✔
24	from hashlib import sha256	6✔
25	from pathlib import Path	6✔
26	from shutil import which, copy	6✔
27	from typing import Sequence	6✔
28
29	import yaml	6✔
30	from jsonschema import Draft7Validator	6✔
31
32	from kubernator.api import (KubernatorPlugin, Globs, StripNL,	6✔
33	scan_dir,
34	load_file,
35	FileType,
36	calling_frame_source,
37	validator_with_defaults,
38	get_golang_os,
39	get_golang_machine,
40	prepend_os_path, TemplateEngine
41	)
42	from kubernator.plugins.k8s_api import K8SResource	6✔
43	from kubernator.proc import DEVNULL	6✔
44
45	logger = logging.getLogger("kubernator.helm")	6✔
46	proc_logger = logger.getChild("proc")	6✔
47	stdout_logger = StripNL(proc_logger.info)	6✔
48	stderr_logger = StripNL(proc_logger.warning)	6✔
49
50	HELM_SCHEMA = {	6✔
51	"properties": {
52	"repository": {
53	"type": "string",
54	"description": "repository hosting the charts"
55	},
56	"chart": {
57	"type": "string",
58	"description": "chart name within the repository"
59	},
60	"version": {
61	"type": "string",
62	"description": "chart version"
63	},
64	"name": {
65	"type": "string",
66	"description": "name of the particular release"
67	},
68	"namespace": {
69	"type": "string",
70	"description": "namespace where the chart is going to be deployed"
71	},
72	"include-crds": {
73	"type": "boolean",
74	"default": True,
75	"description": "generate CRDs for the helm"
76	},
77	"values": {
78	"type": "object",
79	"additionalProperties": True,
80	"description": "inline Helm chart values"
81	},
82	"values-file": {
83	"type": "string",
84	"description": "reference to the file containing Helm chart values"
85	}
86	},
87	"type": "object",
88	"required": ["chart", "name", "namespace"]
89	}
90
91	Draft7Validator.check_schema(HELM_SCHEMA)	6✔
92	HELM_VALIDATOR_CLS = validator_with_defaults(Draft7Validator)	6✔
93	HELM_VALIDATOR = HELM_VALIDATOR_CLS(HELM_SCHEMA, format_checker=Draft7Validator.FORMAT_CHECKER)	6✔
94
95
96	class HelmPlugin(KubernatorPlugin):	6✔
97	logger = logger	6✔
98
99	_name = "helm"	6✔
100
101	def __init__(self):	6✔
102	self.context = None	6✔
103	self.repositories = set()	6✔
104	self.helm_dir = None	6✔
105	self.template_engine = TemplateEngine(logger)	6✔
106
107	self._repositories_not_populated = True	6✔
108
109	def set_context(self, context):	6✔
110	self.context = context	6✔
111
112	def stanza(self):	6✔
113	context = self.context	6✔
114	stanza = [context.helm.helm_file, f"--kubeconfig={context.kubeconfig.kubeconfig}"]	6✔
115	if logger.getEffectiveLevel() < logging.INFO:	6!
116	stanza.append("--debug")	6✔
117	return stanza	6✔
118
119	def register(self, version=None):	6✔
120	context = self.context	6✔
121	context.app.register_plugin("kubeconfig")	6✔
122	context.app.register_plugin("k8s")	6✔
123
124	if version:	6!
125	# Download and use specific version
126	helm_url = f"https://get.helm.sh/helm-v{version}-{get_golang_os()}-{get_golang_machine()}.tar.gz"	6✔
127	helm_file_dl, _ = context.app.download_remote_file(logger, helm_url, "bin")	6✔
128	helm_file_dl = str(helm_file_dl)	6✔
129	self.helm_dir = tempfile.TemporaryDirectory()	6✔
130	context.app.register_cleanup(self.helm_dir)	6✔
131
132	helm_file = str(Path(self.helm_dir.name) / "helm")	6✔
133	helm_tar = tarfile.open(helm_file_dl)	6✔
134	helm_tar.extractall(self.helm_dir.name)	6✔
135
136	copy(Path(self.helm_dir.name) / f"{get_golang_os()}-{get_golang_machine()}" / "helm", helm_file)	6✔
137
138	os.chmod(helm_file, 0o500)	6✔
139	prepend_os_path(self.helm_dir.name)	6✔
140	else:
141	# Use current version
142	helm_file = which("helm")	×
UNCOV 143	if not helm_file:	×
144	raise RuntimeError("`helm` cannot be found and no version has been specified")	×
145
UNCOV 146	logger.debug("Found Helm in %r", helm_file)	×
147
148	context.globals.helm = dict(default_includes=Globs([".helm.yaml", ".helm.yml"], True),	6✔
149	default_excludes=Globs([".*"], True),
150	namespace_transformer=True,
151	helm_file=helm_file,
152	stanza=self.stanza,
153	add_helm_template=self.add_helm_template,
154	add_helm=self.add_helm,
155	)
156
157	def handle_init(self):	6✔
158	version = self.context.app.run_capturing_out(self.stanza() + ["version", "--template", "{{.Version}}"],	6✔
159	logger.error)
160	logger.info("Found Helm version %s", version)	6✔
161
162	def handle_start(self):	6✔
163	pass	6✔
164
165	def handle_before_dir(self, cwd: Path):	6✔
166	context = self.context	6✔
167
168	context.helm.default_includes = Globs(context.helm.default_includes)	6✔
169	context.helm.default_excludes = Globs(context.helm.default_excludes)	6✔
170	context.helm.includes = Globs(context.helm.default_includes)	6✔
171	context.helm.excludes = Globs(context.helm.default_excludes)	6✔
172
173	# Exclude Helm YAMLs from K8S resource loading
174	context.k8s.excludes.add("*.helm.yaml")	6✔
175	context.k8s.excludes.add("*.helm.yml")	6✔
176
177	def handle_after_dir(self, cwd: Path):	6✔
178	context = self.context	6✔
179	helm = context.helm	6✔
180
181	for f in scan_dir(logger, cwd, lambda d: d.is_file(), helm.excludes, helm.includes):	6✔
182	p = cwd / f.name	6✔
183	display_p = context.app.display_path(p)	6✔
184	logger.debug("Adding Helm template from %s", display_p)	6✔
185
186	helm_templates = load_file(logger, p, FileType.YAML, display_p,	6✔
187	self.template_engine,
188	{"ktor": context})
189
190	for helm_template in helm_templates:	6✔
191	self._add_helm(helm_template, display_p)	6✔
192
193	def add_helm_template(self, template):	6✔
UNCOV 194	return self._add_helm(template, calling_frame_source())	×
195
196	def add_helm(self, **kwargs):	6✔
UNCOV 197	return self._internal_add_helm(calling_frame_source(), **kwargs)	×
198
199	def _add_helm(self, template, source):	6✔
200	errors = list(HELM_VALIDATOR.iter_errors(template))	6✔
201	if errors:	6!
202	for error in errors:	×
UNCOV 203	self.logger.error("Error detected in Helm template from %s", source, exc_info=error)	×
UNCOV 204	raise errors[0]	×
205
206	return self._internal_add_helm(source, **{k.replace("-", "_"): v for k, v in template.items()})	6✔
207
208	def _add_repository(self, repository: str):	6✔
209	def _update_repositories():	6✔
210	if self.repositories:	6!
211	self.context.app.run(self.stanza() + ["repo", "update"],	6✔
212	stdout_logger,
213	stderr_logger).wait()
214
215	if self._repositories_not_populated:	6!
216	preexisting_repositories = json.loads(	6✔
217	self.context.app.run_capturing_out(self.stanza() + ["repo", "list", "-o", "json"], stderr_logger))
218	for pr in preexisting_repositories:	6✔
219	repo_name = pr["name"]	6✔
220	repo_url = pr["url"]	6✔
221	logger.debug("Recording pre-existing repository %s mapping %s", repo_name, repo_url)	6✔
222	self.repositories.add(repo_name)	6✔
223	_update_repositories()	6✔
224	self._repositories_not_populated = False	6✔
225
226	repository_hash = sha256(repository.encode("UTF-8")).hexdigest()	6✔
227	logger.debug("Repository %s mapping to %s", repository, repository_hash)	6✔
228	if repository_hash not in self.repositories:	6!
UNCOV 229	logger.info("Adding and updating repository %s mapping to %s", repository, repository_hash)	×
UNCOV 230	self.context.app.run(self.stanza() + ["repo", "add", repository_hash, repository],	×
231	stdout_logger,
232	stderr_logger).wait()
UNCOV 233	_update_repositories()	×
UNCOV 234	self.repositories.add(repository_hash)	×
235
236	return repository_hash	6✔
237
238	def _internal_add_helm(self, source, *, chart, name, namespace, include_crds,	6✔
239	values=None, values_file=None, repository=None, version=None):
240	if values and values_file:	6!
UNCOV 241	raise RuntimeError(f"In {source} either values or values file may be specified, but not both")	×
242
243	if (repository and chart and chart.startswith("oci://") or	6✔
244	not repository and chart and not chart.startswith("oci://")):
245	raise RuntimeError(	6✔
246	f"In {source} either repository must be specified or OCI-chart must be used, but not both")
247
248	if not version and repository:	6✔
249	raise RuntimeError(f"In {source} version must be specified unless OCI-chart is used")	6✔
250
251	if values_file:	6!
UNCOV 252	values_file = Path(values_file)	×
UNCOV 253	if not values_file.is_absolute():	×
UNCOV 254	values_file = self.context.app.cwd / values_file	×
UNCOV 255	values = list(load_file(logger, values_file, FileType.YAML,	×
256	template_engine=self.template_engine,
257	template_context={"ktor": self.context,
258	"helm": {"chart": chart,
259	"name": name,
260	"namespace": namespace,
261	"include_crds": include_crds,
262	"repository": repository,
263	"version": version,
264	}}))
265	values = values[0] if values else {}	×
266
267	version_spec = []	6✔
268	if repository:	6✔
269	repository_hash = self._add_repository(repository)	6✔
270	chart_name = f"{repository_hash}/{chart}"	6✔
271	else:
272	chart_name = chart	6✔
273
274	if version:	6✔
275	version_spec = ["--version", version]	6✔
276
277	stdin = DEVNULL	6✔
278
279	if values:	6!
UNCOV 280	def write_stdin():	×
UNCOV 281	return json.dumps(values)	×
282
UNCOV 283	stdin = write_stdin	×
284
285	resources = self.context.app.run_capturing_out(self.stanza() +	6✔
286	["template",
287	name,
288	chart_name,
289	"-n", namespace,
290	"-a", ",".join(self.context.k8s.get_api_versions())
291	] +
292	version_spec +
293	(["--include-crds"] if include_crds else []) +
294	["-f", "-"],
295	stderr_logger,
296	stdin=stdin,
297	)
298
299	def helm_namespace_transformer(resources: Sequence[K8SResource],	6✔
300	resource: K8SResource):
301	if resource.rdef.namespaced and not resource.namespace:	6!
UNCOV 302	resource.namespace = namespace	×
UNCOV 303	return resource	×
304
305	if self.context.helm.namespace_transformer:	6!
306	self.context.k8s.add_transformer(helm_namespace_transformer)	6✔
307
308	self.context.k8s.add_resources(yaml.safe_load_all(resources), source)	6✔
309
310	if self.context.helm.namespace_transformer:	6!
311	self.context.k8s.remove_transformer(helm_namespace_transformer)	6✔
312
313	def __repr__(self):	6✔
314	return "Helm Plugin"	6✔

karellen / kubernator / 19040252246

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