19283361828

Committed 12 Nov 2025 01:25AM UTC coverage: 77.093% (-0.4%) from 77.467%

Build # 19283361828

Build Type

Pull #93

github

Committed by

web-flow

Commit Message

Merge ed6f54729 into 135c7b281

Pull Request Pull Request #93: Try enabling MacOS Latest (26) to see if it works

Run Details

587 of 934 branches covered (62.85%)

Branch coverage included in aggregate %.

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

9 existing lines in 3 files now uncovered.

2526 of 3104 relevant lines covered (81.38%)

1.63 hits per line

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

82.14

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

# -*- coding: utf-8 -*-
#
#   Copyright 2020 Express Systems USA, Inc
#   Copyright 2023 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 tempfile
from pathlib import Path

from kubernator.api import (KubernatorPlugin,
                            StripNL,
                            get_golang_os,
                            get_golang_machine,
                            prepend_os_path,
                            get_cache_dir,
                            CalledProcessError
                            )

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

MINIKUBE_MAX_VERSION_LEGACY = "1.36.0"


class MinikubePlugin(KubernatorPlugin):
    logger = logger

    _name = "minikube"

    def __init__(self):
        self.context = None
        self.minikube_dir = None
        self.minikube_home_dir = None
        self.kubeconfig_dir = None

        super().__init__()

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

    def get_latest_minikube_version(self):
        context = self.context
        versions = context.app.run_capturing_out(["git", "ls-remote", "-t", "--refs",
                                                  "https://github.com/kubernetes/minikube", "v*"],
                                                 stderr_logger)

        # 06e3b0cf7999f74fc52af362b42fb21076ade64a        refs/tags/v1.9.1
        # "refs/tags/v1.9.1"
        # "1.9.1"
        # ("1","9","1")
        # (1, 9, 1)
        # sort and get latest, which is the last/highest
        # "v1.9.1"
        return (".".join(map(str, sorted(list(map(lambda v: tuple(map(int, v)),
                                                  filter(lambda v: len(v) == 3,
                                                         map(lambda line: line.split()[1][11:].split("."),
                                                             versions.splitlines(False))))))[-1])))

    def cmd(self, *extra_args):
        stanza, env = self._stanza(list(extra_args))
        return self.context.app.run(stanza, stdout_logger, stderr_logger, env=env).wait()

    def cmd_out(self, *extra_args):
        stanza, env = self._stanza(list(extra_args))
        return self.context.app.run_capturing_out(stanza, stderr_logger, env=env)

    def _stanza(self, extra_args):
        context = self.context
        minikube = context.minikube
        stanza = [context.minikube.minikube_file, "-p", minikube.profile] + extra_args
        env = dict(os.environ)
        env["MINIKUBE_HOME"] = str(self.minikube_home_dir)
        env["KUBECONFIG"] = str(minikube.kubeconfig)
        return stanza, env

    def register(self, minikube_version=None, profile="default", k8s_version=None,
                 keep_running=False, start_fresh=False,
                 nodes=1, driver=None, cpus="no-limit", extra_args=None, extra_addons=None):
        context = self.context

        context.app.register_plugin("kubeconfig")

        if not k8s_version:
            msg = "No Kubernetes version is specified for Minikube"
            logger.critical(msg)
            raise RuntimeError(msg)

        k8s_version_tuple = tuple(map(int, k8s_version.split(".")))

        if not minikube_version:
            minikube_version = self.get_latest_minikube_version()
            logger.info("No minikube version is specified, latest is %s", minikube_version)
            if k8s_version_tuple < (1, 28, 0):
                logger.info("While latest minikube version is %s, "
                            "the requested K8S version %s requires %s or earlier - choosing %s",
                            minikube_version, k8s_version, MINIKUBE_MAX_VERSION_LEGACY,
                            MINIKUBE_MAX_VERSION_LEGACY)
                minikube_version = MINIKUBE_MAX_VERSION_LEGACY

        minikube_dl_file, _ = context.app.download_remote_file(logger,
                                                               f"https://github.com/kubernetes/minikube/releases"
                                                               f"/download/v{minikube_version}/"
                                                               f"minikube-{get_golang_os()}-{get_golang_machine()}",
                                                               "bin")

        os.chmod(minikube_dl_file, 0o500)
        self.minikube_dir = tempfile.TemporaryDirectory()
        context.app.register_cleanup(self.minikube_dir)

        minikube_file = Path(self.minikube_dir.name) / "minikube"
        minikube_file.symlink_to(minikube_dl_file)
        prepend_os_path(self.minikube_dir.name)
        version_out: str = self.context.app.run_capturing_out([str(minikube_file), "version", "--short"],
                                                              stderr_logger).strip()
        version = version_out[1:]
        logger.info("Found minikube %s in %s", version, minikube_file)

        profile_dir = get_cache_dir("minikube")
        self.minikube_home_dir = profile_dir
        self.minikube_home_dir.mkdir(parents=True, exist_ok=True)
        self.kubeconfig_dir = profile_dir / ".kube" / profile
        self.kubeconfig_dir.mkdir(parents=True, exist_ok=True)

        if not driver:
            driver = "docker"
            if get_golang_os() == "darwin":
                logger.debug("Auto-detecting Minikube driver on MacOS...")
                cmd_debug_logger = StripNL(proc_logger.debug)
                try:
                    context.app.run(["docker", "info"], cmd_debug_logger, cmd_debug_logger).wait()
                    logger.info("Docker is functional, selecting 'docker' as the driver for Minikube")
                except (FileNotFoundError, CalledProcessError) as e:
                    logger.trace("Docker is NOT functional", exc_info=e)
                    driver = "hyperkit"
                    try:
                        context.app.run(["hyperkit", "-v"], cmd_debug_logger, cmd_debug_logger).wait()
                        logger.info("Hyperkit is functional, selecting 'hyperkit' as the driver for Minikube")
                    except (FileNotFoundError, CalledProcessError) as e:
                        logger.trace("Hyperkit is NOT functional", exc_info=e)
                        driver = "podman"
                        try:
                            context.app.run(["podman", "info"], cmd_debug_logger, cmd_debug_logger).wait()
                            logger.info("Podman is functional, selecting 'podman' as the driver for Minikube")
                        except (FileNotFoundError, CalledProcessError) as e:
                            logger.trace("Podman is NOT functional", exc_info=e)
                            raise RuntimeError("No Minikube driver is functional on MacOS. "
                                               "Tried 'docker', 'hyperkit' and 'podman'!")

        context.globals.minikube = dict(version=version,
                                        minikube_file=str(minikube_file),
                                        profile=profile,
                                        k8s_version=k8s_version,
                                        k8s_version_tuple=k8s_version_tuple,
                                        start_fresh=start_fresh,
                                        keep_running=keep_running,
                                        nodes=nodes,
                                        driver=driver,
                                        cpus=cpus,
                                        extra_args=extra_args or [],
                                        extra_addons=extra_addons or [],
                                        kubeconfig=str(self.kubeconfig_dir / "config"),
                                        cmd=self.cmd,
                                        cmd_out=self.cmd_out
                                        )
        context.kubeconfig.kubeconfig = context.minikube.kubeconfig

        logger.info("Minikube Home is %s", self.minikube_home_dir)
        logger.info("Minikube Kubeconfig is %s", context.minikube.kubeconfig)

    def minikube_is_running(self):
        try:
            out = self.cmd_out("status", "-o", "json")
            logger.info("Minikube profile %r is running: %s", self.context.minikube.profile,
                        out.strip())
            return True
        except CalledProcessError as e:
            logger.info("Minikube profile %r is not running: %s", self.context.minikube.profile,
                        e.output.strip())
            return False

    def minikube_start(self):
        minikube = self.context.minikube
        if not self.minikube_is_running():
            logger.info("Starting minikube profile %r...", minikube.profile)
            args = ["start",
                    "--driver", str(minikube.driver),
                    "--kubernetes-version", str(minikube.k8s_version),
                    "--wait", "apiserver",
                    "--nodes", str(minikube.nodes)]

            addons = []
            if minikube.k8s_version_tuple >= (1, 28):
                addons += ["volumesnapshots", "csi-hostpath-driver"]

            if minikube.extra_addons:
                addons += minikube.extra_addons

            if addons:
                args += ["--addons", ",".join(addons)]

            if minikube.driver == "docker":
                args.extend(["--cpus", str(minikube.cpus)])

            self.cmd(*args)
        else:
            logger.warning("Minikube profile %r is already running!", minikube.profile)

        logger.info("Updating minikube profile %r context", minikube.profile)
        self.cmd("update-context")

        if minikube.k8s_version_tuple >= (1, 28):
            logger.info("Disabling old storage addons")
            self.cmd("addons", "disable", "storage-provisioner")
            self.cmd("addons", "disable", "default-storageclass")

        logger.info("Running initialization scripts for profile %r", minikube.profile)
        self.context.app.register_plugin("kubectl", version=minikube.k8s_version)
        if minikube.k8s_version_tuple >= (1, 28):
            storage_class = self.context.kubectl.get("storageclass", "csi-hostpath-sc")
            self.context.kubectl.run("delete", "storageclass", "csi-hostpath-sc")
            storage_class["metadata"]["annotations"]["storageclass.kubernetes.io/is-default-class"] = "true"
            storage_class["volumeBindingMode"] = "WaitForFirstConsumer"

            def write_stdin():
                return json.dumps(storage_class)

            self.context.kubectl.run("create", "-f", "-", stdin=write_stdin)

    def minikube_stop(self):
        minikube = self.context.minikube
        if self.minikube_is_running():
            logger.info("Shutting down minikube profile %r...", minikube.profile)
            try:
                self.cmd("stop", "-o", "json")
            except CalledProcessError as e:
                # Workaround for minikube 1.35.0 https://github.com/kubernetes/minikube/issues/20302
                if e.returncode != 82:
                    raise

    def minikube_delete(self):
        minikube = self.context.minikube
        self.minikube_stop()
        logger.warning("Deleting minikube profile %r!", minikube.profile)
        self.cmd("delete")

    def handle_start(self):
        minikube = self.context.minikube
        if minikube.start_fresh:
            self.minikube_delete()

        self.minikube_start()

    def handle_shutdown(self):
        minikube = self.context.minikube
        if not minikube.keep_running:
            self.minikube_stop()
        else:
            logger.warning("Will keep minikube profile %s running!", minikube.profile)

    def __repr__(self):
        return "Minikube Plugin"

1	# -- coding: utf-8 --
2	#
3	# Copyright 2020 Express Systems USA, Inc
4	# Copyright 2023 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	import json	2✔
19	import logging	2✔
20	import os	2✔
21	import tempfile	2✔
22	from pathlib import Path	2✔
23
24	from kubernator.api import (KubernatorPlugin,	2✔
25	StripNL,
26	get_golang_os,
27	get_golang_machine,
28	prepend_os_path,
29	get_cache_dir,
30	CalledProcessError
31	)
32
33	logger = logging.getLogger("kubernator.minikube")	2✔
34	proc_logger = logger.getChild("proc")	2✔
35	stdout_logger = StripNL(proc_logger.info)	2✔
36	stderr_logger = StripNL(proc_logger.warning)	2✔
37
38	MINIKUBE_MAX_VERSION_LEGACY = "1.36.0"	2✔
39
40
41	class MinikubePlugin(KubernatorPlugin):	2✔
42	logger = logger	2✔
43
44	_name = "minikube"	2✔
45
46	def __init__(self):	2✔
47	self.context = None	2✔
48	self.minikube_dir = None	2✔
49	self.minikube_home_dir = None	2✔
50	self.kubeconfig_dir = None	2✔
51
52	super().__init__()	2✔
53
54	def set_context(self, context):	2✔
55	self.context = context	2✔
56
57	def get_latest_minikube_version(self):	2✔
58	context = self.context	2✔
59	versions = context.app.run_capturing_out(["git", "ls-remote", "-t", "--refs",	2✔
60	"https://github.com/kubernetes/minikube", "v*"],
61	stderr_logger)
62
63	# 06e3b0cf7999f74fc52af362b42fb21076ade64a refs/tags/v1.9.1
64	# "refs/tags/v1.9.1"
65	# "1.9.1"
66	# ("1","9","1")
67	# (1, 9, 1)
68	# sort and get latest, which is the last/highest
69	# "v1.9.1"
70	return (".".join(map(str, sorted(list(map(lambda v: tuple(map(int, v)),	2✔
71	filter(lambda v: len(v) == 3,
72	map(lambda line: line.split()[1][11:].split("."),
73	versions.splitlines(False))))))[-1])))
74
75	def cmd(self, *extra_args):	2✔
76	stanza, env = self._stanza(list(extra_args))	2✔
77	return self.context.app.run(stanza, stdout_logger, stderr_logger, env=env).wait()	2✔
78
79	def cmd_out(self, *extra_args):	2✔
80	stanza, env = self._stanza(list(extra_args))	2✔
81	return self.context.app.run_capturing_out(stanza, stderr_logger, env=env)	2✔
82
83	def _stanza(self, extra_args):	2✔
84	context = self.context	2✔
85	minikube = context.minikube	2✔
86	stanza = [context.minikube.minikube_file, "-p", minikube.profile] + extra_args	2✔
87	env = dict(os.environ)	2✔
88	env["MINIKUBE_HOME"] = str(self.minikube_home_dir)	2✔
89	env["KUBECONFIG"] = str(minikube.kubeconfig)	2✔
90	return stanza, env	2✔
91
92	def register(self, minikube_version=None, profile="default", k8s_version=None,	2✔
93	keep_running=False, start_fresh=False,
94	nodes=1, driver=None, cpus="no-limit", extra_args=None, extra_addons=None):
95	context = self.context	2✔
96
97	context.app.register_plugin("kubeconfig")	2✔
98
99	if not k8s_version:	2!
UNCOV 100	msg = "No Kubernetes version is specified for Minikube"	×
101	logger.critical(msg)	×
102	raise RuntimeError(msg)	×
103
104	k8s_version_tuple = tuple(map(int, k8s_version.split(".")))	2✔
105
106	if not minikube_version:	2!
107	minikube_version = self.get_latest_minikube_version()	2✔
108	logger.info("No minikube version is specified, latest is %s", minikube_version)	2✔
109	if k8s_version_tuple < (1, 28, 0):	2✔
110	logger.info("While latest minikube version is %s, "	2✔
111	"the requested K8S version %s requires %s or earlier - choosing %s",
112	minikube_version, k8s_version, MINIKUBE_MAX_VERSION_LEGACY,
113	MINIKUBE_MAX_VERSION_LEGACY)
114	minikube_version = MINIKUBE_MAX_VERSION_LEGACY	2✔
115
116	minikube_dl_file, _ = context.app.download_remote_file(logger,	2✔
117	f"https://github.com/kubernetes/minikube/releases"
118	f"/download/v{minikube_version}/"
119	f"minikube-{get_golang_os()}-{get_golang_machine()}",
120	"bin")
121
122	os.chmod(minikube_dl_file, 0o500)	2✔
123	self.minikube_dir = tempfile.TemporaryDirectory()	2✔
124	context.app.register_cleanup(self.minikube_dir)	2✔
125
126	minikube_file = Path(self.minikube_dir.name) / "minikube"	2✔
127	minikube_file.symlink_to(minikube_dl_file)	2✔
128	prepend_os_path(self.minikube_dir.name)	2✔
129	version_out: str = self.context.app.run_capturing_out([str(minikube_file), "version", "--short"],	2✔
130	stderr_logger).strip()
131	version = version_out[1:]	2✔
132	logger.info("Found minikube %s in %s", version, minikube_file)	2✔
133
134	profile_dir = get_cache_dir("minikube")	2✔
135	self.minikube_home_dir = profile_dir	2✔
136	self.minikube_home_dir.mkdir(parents=True, exist_ok=True)	2✔
137	self.kubeconfig_dir = profile_dir / ".kube" / profile	2✔
138	self.kubeconfig_dir.mkdir(parents=True, exist_ok=True)	2✔
139
140	if not driver:	2!
141	driver = "docker"	2✔
142	if get_golang_os() == "darwin":	2!
UNCOV 143	logger.debug("Auto-detecting Minikube driver on MacOS...")	×
UNCOV 144	cmd_debug_logger = StripNL(proc_logger.debug)	×
145	try:	×
146	context.app.run(["docker", "info"], cmd_debug_logger, cmd_debug_logger).wait()	×
147	logger.info("Docker is functional, selecting 'docker' as the driver for Minikube")	×
148	except (FileNotFoundError, CalledProcessError) as e:	×
149	logger.trace("Docker is NOT functional", exc_info=e)	×
150	driver = "hyperkit"	×
151	try:	×
152	context.app.run(["hyperkit", "-v"], cmd_debug_logger, cmd_debug_logger).wait()	×
153	logger.info("Hyperkit is functional, selecting 'hyperkit' as the driver for Minikube")	×
154	except (FileNotFoundError, CalledProcessError) as e:	×
155	logger.trace("Hyperkit is NOT functional", exc_info=e)	×
156	driver = "podman"	×
157	try:	×
158	context.app.run(["podman", "info"], cmd_debug_logger, cmd_debug_logger).wait()	×
159	logger.info("Podman is functional, selecting 'podman' as the driver for Minikube")	×
160	except (FileNotFoundError, CalledProcessError) as e:	×
161	logger.trace("Podman is NOT functional", exc_info=e)	×
162	raise RuntimeError("No Minikube driver is functional on MacOS. "	×
163	"Tried 'docker', 'hyperkit' and 'podman'!")
164
165	context.globals.minikube = dict(version=version,	2✔
166	minikube_file=str(minikube_file),
167	profile=profile,
168	k8s_version=k8s_version,
169	k8s_version_tuple=k8s_version_tuple,
170	start_fresh=start_fresh,
171	keep_running=keep_running,
172	nodes=nodes,
173	driver=driver,
174	cpus=cpus,
175	extra_args=extra_args or [],
176	extra_addons=extra_addons or [],
177	kubeconfig=str(self.kubeconfig_dir / "config"),
178	cmd=self.cmd,
179	cmd_out=self.cmd_out
180	)
181	context.kubeconfig.kubeconfig = context.minikube.kubeconfig	2✔
182
183	logger.info("Minikube Home is %s", self.minikube_home_dir)	2✔
184	logger.info("Minikube Kubeconfig is %s", context.minikube.kubeconfig)	2✔
185
186	def minikube_is_running(self):	2✔
187	try:	2✔
188	out = self.cmd_out("status", "-o", "json")	2✔
189	logger.info("Minikube profile %r is running: %s", self.context.minikube.profile,	2✔
190	out.strip())
191	return True	2✔
192	except CalledProcessError as e:	2✔
193	logger.info("Minikube profile %r is not running: %s", self.context.minikube.profile,	2✔
194	e.output.strip())
195	return False	2✔
196
197	def minikube_start(self):	2✔
198	minikube = self.context.minikube	2✔
199	if not self.minikube_is_running():	2✔
200	logger.info("Starting minikube profile %r...", minikube.profile)	2✔
201	args = ["start",	2✔
202	"--driver", str(minikube.driver),
203	"--kubernetes-version", str(minikube.k8s_version),
204	"--wait", "apiserver",
205	"--nodes", str(minikube.nodes)]
206
207	addons = []	2✔
208	if minikube.k8s_version_tuple >= (1, 28):	2✔
209	addons += ["volumesnapshots", "csi-hostpath-driver"]	2✔
210
211	if minikube.extra_addons:	2!
UNCOV 212	addons += minikube.extra_addons	×
213
214	if addons:	2✔
215	args += ["--addons", ",".join(addons)]	2✔
216
217	if minikube.driver == "docker":	2!
218	args.extend(["--cpus", str(minikube.cpus)])	2✔
219
220	self.cmd(*args)	2✔
221	else:
222	logger.warning("Minikube profile %r is already running!", minikube.profile)	2✔
223
224	logger.info("Updating minikube profile %r context", minikube.profile)	2✔
225	self.cmd("update-context")	2✔
226
227	if minikube.k8s_version_tuple >= (1, 28):	2✔
228	logger.info("Disabling old storage addons")	2✔
229	self.cmd("addons", "disable", "storage-provisioner")	2✔
230	self.cmd("addons", "disable", "default-storageclass")	2✔
231
232	logger.info("Running initialization scripts for profile %r", minikube.profile)	2✔
233	self.context.app.register_plugin("kubectl", version=minikube.k8s_version)	2✔
234	if minikube.k8s_version_tuple >= (1, 28):	2✔
235	storage_class = self.context.kubectl.get("storageclass", "csi-hostpath-sc")	2✔
236	self.context.kubectl.run("delete", "storageclass", "csi-hostpath-sc")	2✔
237	storage_class["metadata"]["annotations"]["storageclass.kubernetes.io/is-default-class"] = "true"	2✔
238	storage_class["volumeBindingMode"] = "WaitForFirstConsumer"	2✔
239
240	def write_stdin():	2✔
241	return json.dumps(storage_class)	2✔
242
243	self.context.kubectl.run("create", "-f", "-", stdin=write_stdin)	2✔
244
245	def minikube_stop(self):	2✔
246	minikube = self.context.minikube	2✔
247	if self.minikube_is_running():	2✔
248	logger.info("Shutting down minikube profile %r...", minikube.profile)	2✔
249	try:	2✔
250	self.cmd("stop", "-o", "json")	2✔
UNCOV 251	except CalledProcessError as e:	×
252	# Workaround for minikube 1.35.0 https://github.com/kubernetes/minikube/issues/20302
253	if e.returncode != 82:	×
UNCOV 254	raise	×
255
256	def minikube_delete(self):	2✔
257	minikube = self.context.minikube	2✔
258	self.minikube_stop()	2✔
259	logger.warning("Deleting minikube profile %r!", minikube.profile)	2✔
260	self.cmd("delete")	2✔
261
262	def handle_start(self):	2✔
263	minikube = self.context.minikube	2✔
264	if minikube.start_fresh:	2✔
265	self.minikube_delete()	2✔
266
267	self.minikube_start()	2✔
268
269	def handle_shutdown(self):	2✔
270	minikube = self.context.minikube	2✔
271	if not minikube.keep_running:	2✔
272	self.minikube_stop()	2✔
273	else:
274	logger.warning("Will keep minikube profile %s running!", minikube.profile)	2✔
275
276	def __repr__(self):	2✔
277	return "Minikube Plugin"	2✔

karellen / kubernator / 19283361828

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