Unprivileged client: set as var before switching back to admin client (#224)

* Create size-labeler.yml

* Delete .github/workflows/size-labeler.yml

* model mesh - add auth tests

* xx

* fix: update unpriv client

* fix: raise on un

* fix: raise on un

* fix: remove test code

* fix: fail on login

Author: rnetser

tests/conftest.py

@@ -24,6 +24,7 @@
 from simple_logger.logger import get_logger
 
 from utilities.data_science_cluster_utils import update_components_in_dsc
+from utilities.exceptions import ClusterLoginError
 from utilities.general import get_s3_secret_dict
 from utilities.infra import (
     create_ns,
@@ -246,33 +247,36 @@ def unprivileged_client(
     non_admin_user_password: tuple[str, str],
 ) -> Generator[DynamicClient, Any, Any]:
     """
-    Provides none privileged API client. If non_admin_user_password is None, then it will yield admin_client.
+    Provides none privileged API client. If non_admin_user_password is None, then it will raise.
     """
     if non_admin_user_password is None:
-        yield admin_client
+        raise ValueError("Unprivileged user not provisioned")
 
     else:
         current_user = run_command(command=["oc", "whoami"])[1].strip()
+        non_admin_user_name = non_admin_user_password[0]
 
         if login_with_user_password(
             api_address=admin_client.configuration.host,
-            user=non_admin_user_password[0],
+            user=non_admin_user_name,
             password=non_admin_user_password[1],
         ):
             with open(kubconfig_filepath) as fd:
                 kubeconfig_content = yaml.safe_load(fd)
 
             unprivileged_context = kubeconfig_content["current-context"]
 
+            unprivileged_client = get_client(config_file=kubconfig_filepath, context=unprivileged_context)
+
             # Get back to admin account
             login_with_user_password(
                 api_address=admin_client.configuration.host,
                 user=current_user.strip(),
             )
-            yield get_client(config_file=kubconfig_filepath, context=unprivileged_context)
+            yield unprivileged_client
 
         else:
-            yield admin_client
+            raise ClusterLoginError(user=non_admin_user_name)
 
 
 @pytest.fixture(scope="session")

utilities/exceptions.py

@@ -84,3 +84,11 @@ class PodLogMissMatchError(Exception):
 
 class ResourceMismatch(Exception):
     pass
+
+
+class ClusterLoginError(Exception):
+    def __init__(self, user: str):
+        self.user = user
+
+    def __str__(self) -> str:
+        return f"Failed to log in as user {self.user}."

utilities/infra.py

@@ -39,7 +39,7 @@
 from utilities.constants import ApiGroups, Labels, Timeout
 from utilities.constants import KServeDeploymentType
 from utilities.constants import Annotations
-from utilities.exceptions import FailedPodsError
+from utilities.exceptions import ClusterLoginError, FailedPodsError
 from timeout_sampler import TimeoutExpiredError, TimeoutSampler, retry
 import utilities.general
 
@@ -328,9 +328,15 @@ def login_with_user_password(api_address: str, user: str, password: str | None =
     if password:
         login_command += f" -p '{password}'"
 
-    _, out, _ = run_command(command=shlex.split(login_command), hide_log_command=True)
+    _, out, err = run_command(command=shlex.split(login_command), hide_log_command=True)
 
-    return "Login successful" in out
+    if err and err.lower().startswith("error"):
+        raise ClusterLoginError(user=user)
+
+    if re.search(r"Login successful|Logged into", out):
+        return True
+
+    return False
 
 
 @cache