From 1f46732c95688f246cff881985df363097f1e02f Mon Sep 17 00:00:00 2001 From: Jacob Tomlinson Date: Fri, 17 May 2024 16:15:59 +0100 Subject: [PATCH 1/2] Add kubectl-ng config use-context --- examples/kubectl-ng/kubectl_ng/_config.py | 25 ++++++++++++++--- .../kubectl-ng/kubectl_ng/_typer_utils.py | 28 +++++++++++++++++++ examples/kubectl-ng/kubectl_ng/cli.py | 26 +---------------- .../kubectl_ng/tests/test_kng_config.py | 7 +++++ 4 files changed, 57 insertions(+), 29 deletions(-) create mode 100644 examples/kubectl-ng/kubectl_ng/_typer_utils.py diff --git a/examples/kubectl-ng/kubectl_ng/_config.py b/examples/kubectl-ng/kubectl_ng/_config.py index ce98b577..310173cf 100644 --- a/examples/kubectl-ng/kubectl_ng/_config.py +++ b/examples/kubectl-ng/kubectl_ng/_config.py @@ -10,6 +10,8 @@ import kr8s +from ._typer_utils import register + console = Console() config = typer.Typer( @@ -19,7 +21,6 @@ ) -@config.command(name="current-context", help="Display the current-context") def config_current_context(): """Display the current context.""" try: @@ -29,7 +30,6 @@ def config_current_context(): raise typer.Exit(code=1) -@config.command(name="get-clusters", help="Display clusters defined in the kubeconfig") def config_get_clusters(): """Display clusters defined in the kubeconfig.""" try: @@ -45,7 +45,6 @@ def config_get_clusters(): console.print(table) -@config.command(name="get-users", help="Display users defined in the kubeconfig") def config_get_users(): """Display users defined in the kubeconfig.""" try: @@ -61,7 +60,6 @@ def config_get_users(): console.print(table) -@config.command(name="get-contexts", help="Describe one or many contexts") def config_get_contexts(name: Annotated[Optional[str], typer.Argument()] = None): """Display users defined in the kubeconfig.""" try: @@ -91,3 +89,22 @@ def config_get_contexts(name: Annotated[Optional[str], typer.Argument()] = None) raise typer.Exit(code=1) console.print(table) + + +async def config_use_context(context: Annotated[str, typer.Argument()]): + """Set the current-context in a kubeconfig file.""" + try: + api = await kr8s.asyncio.api() + await api.auth.kubeconfig.use_context(context) + except (ValueError, KeyError): + typer.echo(f"error: context {context} not found") + raise typer.Exit(code=1) + + console.print(f'Switched to context "{context}".') + + +register(config, config_current_context, "current-context") +register(config, config_get_clusters, "get-clusters") +register(config, config_get_users, "get-users") +register(config, config_get_contexts, "get-contexts") +register(config, config_use_context, "use-context") diff --git a/examples/kubectl-ng/kubectl_ng/_typer_utils.py b/examples/kubectl-ng/kubectl_ng/_typer_utils.py new file mode 100644 index 00000000..221d0358 --- /dev/null +++ b/examples/kubectl-ng/kubectl_ng/_typer_utils.py @@ -0,0 +1,28 @@ +# SPDX-FileCopyrightText: Copyright (c) 2024, Kr8s Developers (See LICENSE for list) +# SPDX-License-Identifier: BSD 3-Clause License + +import asyncio +from functools import wraps + +import typer + + +def _typer_async(f): + @wraps(f) + def wrapper(*args, **kwargs): + return asyncio.run(f(*args, **kwargs)) + + return wrapper + + +def register(app, func, alias=None): + if asyncio.iscoroutinefunction(func): + func = _typer_async(func) + if isinstance(func, typer.Typer): + assert alias, "Typer subcommand must have an alias." + app.add_typer(func, name=alias) + else: + if alias is not None: + app.command(alias)(func) + else: + app.command()(func) diff --git a/examples/kubectl-ng/kubectl_ng/cli.py b/examples/kubectl-ng/kubectl_ng/cli.py index 507aba9a..0751cae6 100644 --- a/examples/kubectl-ng/kubectl_ng/cli.py +++ b/examples/kubectl-ng/kubectl_ng/cli.py @@ -1,8 +1,5 @@ # SPDX-FileCopyrightText: Copyright (c) 2023-2024, Kr8s Developers (See LICENSE for list) # SPDX-License-Identifier: BSD 3-Clause License -import asyncio -from functools import wraps - import typer from ._api_resources import api_resources @@ -12,31 +9,10 @@ from ._delete import delete from ._exec import kexec from ._get import get +from ._typer_utils import register from ._version import version from ._wait import wait - -def _typer_async(f): - @wraps(f) - def wrapper(*args, **kwargs): - return asyncio.run(f(*args, **kwargs)) - - return wrapper - - -def register(app, func, alias=None): - if asyncio.iscoroutinefunction(func): - func = _typer_async(func) - if isinstance(func, typer.Typer): - assert alias, "Typer subcommand must have an alias." - app.add_typer(func, name=alias) - else: - if alias is not None: - app.command(alias)(func) - else: - app.command()(func) - - app = typer.Typer(no_args_is_help=True) register(app, api_resources) register(app, api_versions) diff --git a/examples/kubectl-ng/kubectl_ng/tests/test_kng_config.py b/examples/kubectl-ng/kubectl_ng/tests/test_kng_config.py index 6d1d3e4b..b6e1f481 100644 --- a/examples/kubectl-ng/kubectl_ng/tests/test_kng_config.py +++ b/examples/kubectl-ng/kubectl_ng/tests/test_kng_config.py @@ -39,3 +39,10 @@ def test_get_contexts(k8s_cluster): result = runner.invoke(app, ["config", "get-contexts", "foo"]) assert result.exit_code == 1 assert "foo not found" in result.stdout + + +def test_use_context(): + current_context = kr8s.api().auth.kubeconfig.current_context + result = runner.invoke(app, ["config", "use-context", current_context]) + assert result.exit_code == 0 + assert current_context in result.stdout From 9517063a846d0c530281e77aba52bf6ac4f67315 Mon Sep 17 00:00:00 2001 From: Jacob Tomlinson Date: Fri, 17 May 2024 16:21:28 +0100 Subject: [PATCH 2/2] Add kubectl-ng config rename-context --- examples/kubectl-ng/kubectl_ng/_config.py | 16 ++++++++++ .../kubectl_ng/tests/test_kng_config.py | 30 +++++++++++++++++++ 2 files changed, 46 insertions(+) diff --git a/examples/kubectl-ng/kubectl_ng/_config.py b/examples/kubectl-ng/kubectl_ng/_config.py index 310173cf..97cd833a 100644 --- a/examples/kubectl-ng/kubectl_ng/_config.py +++ b/examples/kubectl-ng/kubectl_ng/_config.py @@ -103,8 +103,24 @@ async def config_use_context(context: Annotated[str, typer.Argument()]): console.print(f'Switched to context "{context}".') +async def config_rename_context( + old_name: Annotated[str, typer.Argument()], + new_name: Annotated[str, typer.Argument()], +): + """Renames a context from the kubeconfig file.""" + try: + api = await kr8s.asyncio.api() + await api.auth.kubeconfig.rename_context(old_name, new_name) + except (ValueError, KeyError): + typer.echo(f"error: context {old_name} not found") + raise typer.Exit(code=1) + + console.print(f'Context "{old_name}" renamed to "{new_name}".') + + register(config, config_current_context, "current-context") register(config, config_get_clusters, "get-clusters") register(config, config_get_users, "get-users") register(config, config_get_contexts, "get-contexts") register(config, config_use_context, "use-context") +register(config, config_rename_context, "rename-context") diff --git a/examples/kubectl-ng/kubectl_ng/tests/test_kng_config.py b/examples/kubectl-ng/kubectl_ng/tests/test_kng_config.py index b6e1f481..0eb71fb6 100644 --- a/examples/kubectl-ng/kubectl_ng/tests/test_kng_config.py +++ b/examples/kubectl-ng/kubectl_ng/tests/test_kng_config.py @@ -46,3 +46,33 @@ def test_use_context(): result = runner.invoke(app, ["config", "use-context", current_context]) assert result.exit_code == 0 assert current_context in result.stdout + + +def test_rename_context(): + # Get current context + current_context = kr8s.api().auth.kubeconfig.current_context + result = runner.invoke(app, ["config", "current-context"]) + assert result.exit_code == 0 + assert current_context in result.stdout + + # Rename current context to foo + result = runner.invoke(app, ["config", "rename-context", current_context, "foo"]) + assert result.exit_code == 0 + assert current_context in result.stdout + assert "foo" in result.stdout + + # Check the context rename was successful + result = runner.invoke(app, ["config", "current-context"]) + assert result.exit_code == 0 + assert "foo" in result.stdout + + # Rename foo back to the original name + result = runner.invoke(app, ["config", "rename-context", "foo", current_context]) + assert result.exit_code == 0 + assert current_context in result.stdout + assert "foo" in result.stdout + + # Check the context revert was successful + result = runner.invoke(app, ["config", "current-context"]) + assert result.exit_code == 0 + assert current_context in result.stdout