warnet auth --revert

src/warnet/constants.py

@@ -83,6 +83,7 @@ class AnnexMember(Enum):
 
 # Kubeconfig related stuffs
 KUBECONFIG = os.environ.get("KUBECONFIG", os.path.expanduser("~/.kube/config"))
+KUBECONFIG_UNDO = KUBECONFIG + "_warnet_undo"
 
 # TODO: all of this logging stuff should be a helm chart
 LOGGING_CONFIG = {

src/warnet/users.py

@@ -5,14 +5,20 @@
 
 import click
 
-from warnet.constants import KUBECONFIG
+from warnet.constants import KUBECONFIG, KUBECONFIG_UNDO
 from warnet.k8s import K8sError, open_kubeconfig, write_kubeconfig
 
 
 @click.command()
-@click.argument("auth_config", type=str)
-def auth(auth_config):
+@click.option("--revert", is_flag=True, default=False, show_default=True)
+@click.argument("auth_config", type=str, required=False)
+def auth(revert, auth_config):
     """Authenticate with a Warnet cluster using a kubernetes config file"""
+    if revert:
+        auth_config = KUBECONFIG_UNDO
+    elif not auth_config:
+        raise click.UsageError("Missing argument: AUTH_CONFIG")
+
     try:
         auth_config = open_kubeconfig(auth_config)
     except K8sError as e:
@@ -38,6 +44,13 @@ def auth(auth_config):
         click.secho(f"Could not open KUBECONFIG: {KUBECONFIG}", fg="red")
         sys.exit(1)
 
+    try:
+        write_kubeconfig(base_config, KUBECONFIG_UNDO)
+        click.secho(f"Backed up current kubeconfig to: {KUBECONFIG_UNDO}", fg="green")
+    except K8sError as e:
+        click.secho(e, fg="yellow")
+        click.secho(f"Could not backup current kubeconfig to {KUBECONFIG_UNDO}", fg="red")
+
     if not is_first_config:
         for category in ["clusters", "users", "contexts"]:
             if category in auth_config:

test/namespace_admin_test.py

@@ -147,8 +147,7 @@ def two_namespaces_are_validated(self) -> bool:
         return self.red_namespace in maybe_namespaces
 
     def return_to_initial_context(self):
-        cmd = f"kubectl config use-context {self.initial_context}"
-        self.log.info(run_command(cmd))
+        self.log.info(self.warnet("auth --revert"))
         self.wait_for_predicate(self.this_is_the_current_context(self.initial_context))
 
     def this_is_the_current_context(self, context: str) -> Callable[[], bool]: