kubectl-ng config use/rename context

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,38 @@ 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}".')
+
+
+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")

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)

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)

examples/kubectl-ng/kubectl_ng/tests/test_kng_config.py

@@ -39,3 +39,40 @@ 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
+
+
+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