18180360618

Committed 02 Oct 2025 01:15AM UTC coverage: 75.914% (-0.1%) from 76.019%

Build # 18180360618

Build Type

push

github

Committed by

arcivanov

Commit Message

Release 1.0.21

Run Details

615 of 961 branches covered (64.0%)

Branch coverage included in aggregate %.

2395 of 3004 relevant lines covered (79.73%)

4.78 hits per line

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

79.88

/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 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):
        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,
                                        start_fresh=start_fresh,
                                        keep_running=keep_running,
                                        nodes=nodes,
                                        driver=driver,
                                        cpus=cpus,
                                        extra_args=extra_args 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)]

            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")

    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
19	import logging	6✔
20	import os	6✔
21	import tempfile	6✔
22	from pathlib import Path	6✔
23
24	from kubernator.api import (KubernatorPlugin,	6✔
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")	6✔
34	proc_logger = logger.getChild("proc")	6✔
35	stdout_logger = StripNL(proc_logger.info)	6✔
36	stderr_logger = StripNL(proc_logger.warning)	6✔
37
38	MINIKUBE_MAX_VERSION_LEGACY = "1.36.0"	6✔
39
40
41	class MinikubePlugin(KubernatorPlugin):	6✔
42	logger = logger	6✔
43
44	_name = "minikube"	6✔
45
46	def __init__(self):	6✔
47	self.context = None	6✔
48	self.minikube_dir = None	6✔
49	self.minikube_home_dir = None	6✔
50	self.kubeconfig_dir = None	6✔
51
52	super().__init__()	6✔
53
54	def set_context(self, context):	6✔
55	self.context = context	6✔
56
57	def get_latest_minikube_version(self):	6✔
58	context = self.context	6✔
59	versions = context.app.run_capturing_out(["git", "ls-remote", "-t", "--refs",	6✔
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)),	6✔
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):	6✔
76	stanza, env = self._stanza(list(extra_args))	6✔
77	return self.context.app.run(stanza, stdout_logger, stderr_logger, env=env).wait()	6✔
78
79	def cmd_out(self, *extra_args):	6✔
80	stanza, env = self._stanza(list(extra_args))	6✔
81	return self.context.app.run_capturing_out(stanza, stderr_logger, env=env)	6✔
82
83	def _stanza(self, extra_args):	6✔
84	context = self.context	6✔
85	minikube = context.minikube	6✔
86	stanza = [context.minikube.minikube_file, "-p", minikube.profile] + extra_args	6✔
87	env = dict(os.environ)	6✔
88	env["MINIKUBE_HOME"] = str(self.minikube_home_dir)	6✔
89	env["KUBECONFIG"] = str(minikube.kubeconfig)	6✔
90	return stanza, env	6✔
91
92	def register(self, minikube_version=None, profile="default", k8s_version=None,	6✔
93	keep_running=False, start_fresh=False,
94	nodes=1, driver=None, cpus="no-limit", extra_args=None):
95	context = self.context	6✔
96
97	context.app.register_plugin("kubeconfig")	6✔
98
99	if not k8s_version:	6!
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(".")))	6✔
105
106	if not minikube_version:	6!
107	minikube_version = self.get_latest_minikube_version()	6✔
108	logger.info("No minikube version is specified, latest is %s", minikube_version)	6✔
109	if k8s_version_tuple < (1, 28, 0):	6✔
110	logger.info("While latest minikube version is %s, "	6✔
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	6✔
115
116	minikube_dl_file, _ = context.app.download_remote_file(logger,	6✔
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)	6✔
123	self.minikube_dir = tempfile.TemporaryDirectory()	6✔
124	context.app.register_cleanup(self.minikube_dir)	6✔
125
126	minikube_file = Path(self.minikube_dir.name) / "minikube"	6✔
127	minikube_file.symlink_to(minikube_dl_file)	6✔
128	prepend_os_path(self.minikube_dir.name)	6✔
129	version_out: str = self.context.app.run_capturing_out([str(minikube_file), "version", "--short"],	6✔
130	stderr_logger).strip()
131	version = version_out[1:]	6✔
132	logger.info("Found minikube %s in %s", version, minikube_file)	6✔
133
134	profile_dir = get_cache_dir("minikube")	6✔
135	self.minikube_home_dir = profile_dir	6✔
136	self.minikube_home_dir.mkdir(parents=True, exist_ok=True)	6✔
137	self.kubeconfig_dir = profile_dir / ".kube" / profile	6✔
138	self.kubeconfig_dir.mkdir(parents=True, exist_ok=True)	6✔
139
140	if not driver:	6!
141	driver = "docker"	6✔
142	if get_golang_os() == "darwin":	6!
143	logger.debug("Auto-detecting Minikube driver on MacOS...")	×
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,	6✔
166	minikube_file=str(minikube_file),
167	profile=profile,
168	k8s_version=k8s_version,
169	start_fresh=start_fresh,
170	keep_running=keep_running,
171	nodes=nodes,
172	driver=driver,
173	cpus=cpus,
174	extra_args=extra_args or [],
175	kubeconfig=str(self.kubeconfig_dir / "config"),
176	cmd=self.cmd,
177	cmd_out=self.cmd_out
178	)
179	context.kubeconfig.kubeconfig = context.minikube.kubeconfig	6✔
180
181	logger.info("Minikube Home is %s", self.minikube_home_dir)	6✔
182	logger.info("Minikube Kubeconfig is %s", context.minikube.kubeconfig)	6✔
183
184	def minikube_is_running(self):	6✔
185	try:	6✔
186	out = self.cmd_out("status", "-o", "json")	6✔
187	logger.info("Minikube profile %r is running: %s", self.context.minikube.profile,	6✔
188	out.strip())
189	return True	6✔
190	except CalledProcessError as e:	6✔
191	logger.info("Minikube profile %r is not running: %s", self.context.minikube.profile,	6✔
192	e.output.strip())
193	return False	6✔
194
195	def minikube_start(self):	6✔
196	minikube = self.context.minikube	6✔
197	if not self.minikube_is_running():	6✔
198	logger.info("Starting minikube profile %r...", minikube.profile)	6✔
199	args = ["start",	6✔
200	"--driver", str(minikube.driver),
201	"--kubernetes-version", str(minikube.k8s_version),
202	"--wait", "apiserver",
203	"--nodes", str(minikube.nodes)]
204
205	if minikube.driver == "docker":	6!
206	args.extend(["--cpus", str(minikube.cpus)])	6✔
207
208	self.cmd(*args)	6✔
209	else:
210	logger.warning("Minikube profile %r is already running!", minikube.profile)	6✔
211
212	logger.info("Updating minikube profile %r context", minikube.profile)	6✔
213	self.cmd("update-context")	6✔
214
215	def minikube_stop(self):	6✔
216	minikube = self.context.minikube	6✔
217	if self.minikube_is_running():	6✔
218	logger.info("Shutting down minikube profile %r...", minikube.profile)	6✔
219	try:	6✔
220	self.cmd("stop", "-o", "json")	6✔
221	except CalledProcessError as e:	×
222	# Workaround for minikube 1.35.0 https://github.com/kubernetes/minikube/issues/20302
223	if e.returncode != 82:	×
224	raise	×
225
226	def minikube_delete(self):	6✔
227	minikube = self.context.minikube	6✔
228	self.minikube_stop()	6✔
229	logger.warning("Deleting minikube profile %r!", minikube.profile)	6✔
230	self.cmd("delete")	6✔
231
232	def handle_start(self):	6✔
233	minikube = self.context.minikube	6✔
234	if minikube.start_fresh:	6✔
235	self.minikube_delete()	6✔
236
237	self.minikube_start()	6✔
238
239	def handle_shutdown(self):	6✔
240	minikube = self.context.minikube	6✔
241	if not minikube.keep_running:	6✔
242	self.minikube_stop()	6✔
243	else:
244	logger.warning("Will keep minikube profile %s running!", minikube.profile)	6✔
245
246	def __repr__(self):	6✔
247	return "Minikube Plugin"	6✔

karellen / kubernator / 18180360618

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