test: update API key authorization tests

Signed-off-by: Swati Mukund Bagal <[email protected]>

Author: SB159

tests/model_serving/maas_billing/maas_subscription/conftest.py

@@ -7,7 +7,6 @@
 import pytest
 import requests
 from kubernetes.dynamic import DynamicClient
-from ocp_resources.cluster_role_binding import ClusterRoleBinding
 from ocp_resources.data_science_cluster import DataScienceCluster
 from ocp_resources.deployment import Deployment
 from ocp_resources.llm_inference_service import LLMInferenceService
@@ -38,10 +37,11 @@
 from tests.model_serving.maas_billing.utils import build_maas_headers
 from utilities.constants import DscComponents
 from utilities.general import generate_random_name
-from utilities.infra import create_inference_token, create_ns, login_with_user_password
+from utilities.infra import create_inference_token, create_ns, get_openshift_token, login_with_user_password
 from utilities.llmd_constants import ContainerImages, ModelStorage
 from utilities.llmd_utils import create_llmisvc
 from utilities.plugins.constant import OpenAIEnpoints
+from utilities.resources.auth import Auth
 
 LOGGER = get_logger(name=__name__)
 
@@ -687,7 +687,7 @@ def free_user_username(
         key_id=active_api_key_id,
         ocp_user_token=ocp_token_for_actor,
     )
-    LOGGER.info(f"free_user_username: resolved username='{username}' from key id={active_api_key_id}")
+    LOGGER.info(f"free_user_username: resolved username from key id={active_api_key_id}")
     return username
 
 
@@ -705,7 +705,7 @@ def admin_username(
         key_id=admin_active_api_key_id,
         ocp_user_token=admin_ocp_token,
     )
-    LOGGER.info(f"admin_username: resolved username='{username}' from key id={admin_active_api_key_id}")
+    LOGGER.info(f"admin_username: resolved username from key id={admin_active_api_key_id}")
     return username
 
 
@@ -726,26 +726,14 @@ def admin_active_api_key_id(
 
 @pytest.fixture(scope="class")
 def admin_ocp_token(admin_client: DynamicClient) -> Generator[str, Any, Any]:
-    """OCP bearer token for a dedicated SA with cluster-admin ClusterRole, recognised as admin by MaaS API."""
-    applications_namespace = py_config["applications_namespace"]
-    sa_name = f"maas-e2e-admin-{generate_random_name()}"
-
-    with ServiceAccount(
-        client=admin_client,
-        namespace=applications_namespace,
-        name=sa_name,
-        teardown=True,
-    ) as sa:
-        sa.wait(timeout=60)
-
-        with ClusterRoleBinding(
-            client=admin_client,
-            name=sa_name,
-            cluster_role="cluster-admin",
-            subjects=[{"kind": "ServiceAccount", "name": sa_name, "namespace": applications_namespace}],
-            teardown=True,
-        ):
-            yield create_inference_token(model_service_account=sa)
+    """Temporarily adds dedicated-admins to Auth CR adminGroups so the admin token is recognised by MaaS."""
+    auth = Auth(client=admin_client, name="auth")
+    current_groups: list[str] = list(auth.instance.spec.adminGroups or [])
+    patched_groups = list(set(current_groups + ["dedicated-admins"]))
+    LOGGER.info(f"admin_ocp_token: patching Auth CR adminGroups to {patched_groups}")
+
+    with ResourceEditor(patches={auth: {"spec": {"adminGroups": patched_groups}}}):
+        yield get_openshift_token(client=admin_client)
 
 
 @pytest.fixture(scope="function")

tests/model_serving/maas_billing/maas_subscription/test_api_key_authorization.py

@@ -41,16 +41,14 @@ def test_admin_manage_other_users_keys(
             pagination={"limit": 50, "offset": 0},
         )
         assert list_resp.status_code == 200, (
-            f"Expected 200 on admin search for username='{free_user_username}', "
-            f"got {list_resp.status_code}: {list_resp.text[:200]}"
+            f"Expected 200 on admin search by username, got {list_resp.status_code}: {list_resp.text[:200]}"
         )
         items: list[dict] = list_body.get("items") or list_body.get("data") or []
         key_ids = [item["id"] for item in items]
         assert active_api_key_id in key_ids, (
-            f"Expected free user's key id={active_api_key_id} in admin search results "
-            f"for username='{free_user_username}', found ids={key_ids}"
+            f"Expected free user's key id={active_api_key_id} in admin search results, found ids={key_ids}"
         )
-        LOGGER.info(f"[authz] Admin found {len(items)} active key(s) for user='{free_user_username}'")
+        LOGGER.info(f"[authz] Admin found {len(items)} active key(s) for the free user")
 
         revoke_resp, revoke_body = revoke_api_key(
             request_session_http=request_session_http,
@@ -100,7 +98,7 @@ def test_non_admin_cannot_access_other_users_keys(
         )
         LOGGER.info(f"[authz] Free user correctly received 404 on DELETE of admin's key id={admin_active_api_key_id}")
 
-    @pytest.mark.tier1
+    @pytest.mark.tier2
     def test_non_admin_search_only_returns_own_keys(
         self,
         request_session_http: requests.Session,
@@ -134,7 +132,7 @@ def test_non_admin_search_only_returns_own_keys(
             f"[authz] Free user search returned {len(items)} key(s) — own key present, admin's key correctly excluded"
         )
 
-    @pytest.mark.tier1
+    @pytest.mark.tier2
     def test_non_admin_cannot_search_by_other_username(
         self,
         request_session_http: requests.Session,
@@ -152,7 +150,11 @@ def test_non_admin_cannot_search_by_other_username(
             pagination={"limit": 50, "offset": 0},
         )
         assert list_resp.status_code == 403, (
-            f"Expected 403 when free user searches by admin username='{admin_username}', "
+            f"Expected 403 when free user searches by another user's username, "
             f"got {list_resp.status_code}: {list_resp.text[:200]}"
         )
+<<<<<<< HEAD
         LOGGER.info(f"[authz] Free user correctly received 403 when searching by admin username='{admin_username}'")
+=======
+        LOGGER.info("[authz] Free user correctly received 403 when searching by another user's username")
+>>>>>>> 626a88d (test: update API key authorization tests)

tests/model_serving/maas_billing/maas_subscription/utils.py

@@ -321,12 +321,16 @@ def create_and_yield_api_key_id(
     LOGGER.info(f"create_and_yield_api_key_id: created key id={body['id']} name={key_name}")
     yield body["id"]
     LOGGER.info(f"create_and_yield_api_key_id: teardown revoking key id={body['id']}")
-    revoke_api_key(
+    revoke_resp, _ = revoke_api_key(
         request_session_http=request_session_http,
         base_url=base_url,
         key_id=body["id"],
         ocp_user_token=ocp_user_token,
     )
+    if revoke_resp.status_code not in (200, 404):
+        raise AssertionError(
+            f"Unexpected teardown status for key id={body['id']}: {revoke_resp.status_code} {revoke_resp.text[:200]}"
+        )
 
 
 def revoke_api_key(

utilities/constants.py

@@ -140,6 +140,7 @@ class ApiGroups:
     KSERVE: str = "serving.kserve.io"
     KUADRANT_IO: str = "kuadrant.io"
     MAAS_IO: str = "maas.opendatahub.io"
+    AUTH_IO: str = "SERVICES_PLATFORM_OPENDATAHUB_IO"
 
 
 class Annotations:

utilities/resources/auth.py

@@ -0,0 +1,52 @@
+# Generated using https://github.com/RedHatQE/openshift-python-wrapper/blob/main/scripts/resource/README.md
+
+
+from typing import Any
+
+from ocp_resources.exceptions import MissingRequiredArgumentError
+from ocp_resources.resource import Resource
+
+
+class Auth(Resource):
+    """
+    Auth is the Schema for the auths API
+    """
+
+    api_group: str = Resource.ApiGroup.SERVICES_PLATFORM_OPENDATAHUB_IO
+
+    def __init__(
+        self,
+        admin_groups: list[Any] | None = None,
+        allowed_groups: list[Any] | None = None,
+        **kwargs: Any,
+    ) -> None:
+        r"""
+        Args:
+            admin_groups (list[Any]): No field description from API
+
+            allowed_groups (list[Any]): No field description from API
+
+        """
+        super().__init__(**kwargs)
+
+        self.admin_groups = admin_groups
+        self.allowed_groups = allowed_groups
+
+    def to_dict(self) -> None:
+
+        super().to_dict()
+
+        if not self.kind_dict and not self.yaml_file:
+            if self.admin_groups is None:
+                raise MissingRequiredArgumentError(argument="self.admin_groups")
+
+            if self.allowed_groups is None:
+                raise MissingRequiredArgumentError(argument="self.allowed_groups")
+
+            self.res["spec"] = {}
+            _spec = self.res["spec"]
+
+            _spec["adminGroups"] = self.admin_groups
+            _spec["allowedGroups"] = self.allowed_groups
+
+    # End of generated code