fix: make a copy of the credentials before to use it (#891)

* fix: make a copy of the credentials before to use it

* test: add retry logic for RBAC propagation delays

* fix: add retry logic for RBAC proxy initialization in negative test

* fix: improve RBAC proxy initialization error handling

* test: dedicated helper for RBAC connection retry logic

Author: fege

tests/model_registry/model_catalog/test_default_model_catalog.py

@@ -7,6 +7,8 @@
 from ocp_resources.deployment import Deployment
 from simple_logger.logger import get_logger
 from typing import Self, Any
+from timeout_sampler import TimeoutSampler
+from kubernetes.dynamic.exceptions import ResourceNotFoundError
 
 from ocp_resources.pod import Pod
 from ocp_resources.config_map import ConfigMap
@@ -170,10 +172,20 @@ def test_model_catalog_default_catalog_sources(
         Validate specific user can access default model catalog source
         """
         LOGGER.info("Attempting client connection with token")
-        result = execute_get_command(
-            url=f"{model_catalog_rest_url[0]}sources",
-            headers=get_rest_headers(token=user_token_for_api_calls),
-        )["items"]
+
+        url = f"{model_catalog_rest_url[0]}sources"
+        headers = get_rest_headers(token=user_token_for_api_calls)
+
+        # Retry for up to 2 minutes to allow RBAC propagation
+        # Accept ResourceNotFoundError (401) as a transient error during RBAC propagation
+        sampler = TimeoutSampler(
+            wait_timeout=120,
+            sleep=5,
+            func=lambda: execute_get_command(url=url, headers=headers)["items"],
+            exceptions_dict={ResourceNotFoundError: []},
+        )
+
+        result = next(iter(sampler))
         assert result
         items_to_validate = []
         if pytestconfig.option.pre_upgrade or pytestconfig.option.post_upgrade:

tests/model_registry/rbac/conftest.py

@@ -122,16 +122,16 @@ def created_role_binding_user(
     mr_access_role: Role,
     user_credentials_rbac: dict[str, str],
 ) -> Generator[RoleBinding, None, None]:
-    if is_byoidc:
-        user_credentials_rbac["username"] = "mr-non-admin"
-    LOGGER.info(f"Using user {user_credentials_rbac['username']}")
+    # Determine the username to use without mutating the shared fixture
+    username = "mr-non-admin" if is_byoidc else user_credentials_rbac["username"]
+    LOGGER.info(f"Using user {username}")
     yield from create_role_binding(
         admin_client=admin_client,
         model_registry_namespace=model_registry_namespace,
         name="test-model-registry-access",
         mr_access_role=mr_access_role,
         subjects_kind="User",
-        subjects_name=user_credentials_rbac["username"],
+        subjects_name=username,
     )
 
 

tests/model_registry/rbac/test_mr_rbac.py

@@ -91,15 +91,17 @@ def test_user_added_to_group(
         1. After adding the user to the appropriate group, they gain access
         """
         # Wait for access to be granted
-        user_credentials_rbac["username"] = "mr-user1"
+        # Create a copy to avoid mutating the shared fixture
+        creds_copy = user_credentials_rbac.copy()
+        creds_copy["username"] = "mr-user1"
         sampler = TimeoutSampler(
             wait_timeout=240,
             sleep=5,
             func=assert_positive_mr_registry,
             model_registry_instance_rest_endpoint=model_registry_instance_rest_endpoint[0],
             token=get_openshift_token()
             if not is_byoidc
-            else get_mr_user_token(admin_client=admin_client, user_credentials_rbac=user_credentials_rbac),
+            else get_mr_user_token(admin_client=admin_client, user_credentials_rbac=creds_copy),
         )
         for _ in sampler:
             break  # Break after first successful iteration
@@ -145,13 +147,15 @@ def test_add_single_user_role_binding(
         2. The user can access the Model Registry after being granted access
         """
         if is_byoidc:
-            user_credentials_rbac["username"] = "mr-non-admin"
+            # Create a copy to avoid mutating the shared fixture
+            creds_copy = user_credentials_rbac.copy()
+            creds_copy["username"] = "mr-non-admin"
             sampler = TimeoutSampler(
                 wait_timeout=120,
                 sleep=5,
                 func=assert_positive_mr_registry,
                 model_registry_instance_rest_endpoint=model_registry_instance_rest_endpoint[0],
-                token=get_mr_user_token(admin_client=admin_client, user_credentials_rbac=user_credentials_rbac),
+                token=get_mr_user_token(admin_client=admin_client, user_credentials_rbac=creds_copy),
             )
             for _ in sampler:
                 break  # Break after first successful iteration

tests/model_registry/rbac/test_mr_rbac_sa.py

@@ -1,12 +1,13 @@
 # AI Disclaimer: Google Gemini 2.5 pro has been used to generate a majority of this code, with human review and editing.
 import pytest
-from typing import Self
+from typing import Any, Self
 from simple_logger.logger import get_logger
 from model_registry import ModelRegistry as ModelRegistryClient
 from tests.model_registry.rbac.utils import build_mr_client_args
 from utilities.infra import create_inference_token
-from mr_openapi.exceptions import ForbiddenException
+from mr_openapi.exceptions import ForbiddenException, UnauthorizedException
 from ocp_resources.service_account import ServiceAccount
+from timeout_sampler import TimeoutSampler, retry
 
 LOGGER = get_logger(name=__name__)
 
@@ -44,12 +45,11 @@ def test_service_account_access_denied(
         )
         LOGGER.debug(f"Attempting client connection with args: {client_args}")
 
-        # Expect an exception related to HTTP 403
-        with pytest.raises(ForbiddenException) as exc_info:
-            _ = ModelRegistryClient(**client_args)
+        # Retry for up to 2 minutes if we get UnauthorizedException (401) during kube-rbac-proxy initialization
+        # Expect ForbiddenException (403) once kube-rbac-proxy is fully initialized
+        http_error = _try_connection_expect_forbidden(client_args=client_args)
 
         # Verify the status code from the caught exception
-        http_error = exc_info.value
         assert http_error.body is not None, "HTTPError should have a response object"
         LOGGER.info(f"Received expected HTTP error: Status Code {http_error.status}")
         assert http_error.status == 403, f"Expected HTTP 403 Forbidden, but got {http_error.status}"
@@ -69,21 +69,44 @@ def test_service_account_access_granted(
         LOGGER.info(f"Targeting Model Registry REST endpoint: {model_registry_instance_rest_endpoint[0]}")
         LOGGER.info("Applied RBAC Role/Binding via fixtures. Expecting access GRANT.")
 
-        # Create a fresh token to bypass OAuth proxy cache from previous test
+        # Create a fresh token to bypass kube-rbac-proxy cache from previous test
         fresh_token = create_inference_token(model_service_account=service_account)
+        client_args = build_mr_client_args(
+            rest_endpoint=model_registry_instance_rest_endpoint[0], token=fresh_token, author="rbac-test-granted"
+        )
+        LOGGER.debug(f"Attempting client connection with args: {client_args}")
+
+        # Retry for up to 2 minutes to allow RBAC propagation
+        # Accept UnauthorizedException (401) as a transient error during RBAC propagation
+        sampler = TimeoutSampler(
+            wait_timeout=120,
+            sleep=5,
+            func=lambda: ModelRegistryClient(**client_args),
+            exceptions_dict={UnauthorizedException: []},
+        )
 
         try:
-            client_args = build_mr_client_args(
-                rest_endpoint=model_registry_instance_rest_endpoint[0], token=fresh_token, author="rbac-test-granted"
-            )
-            LOGGER.debug(f"Attempting client connection with args: {client_args}")
-            mr_client_success = ModelRegistryClient(**client_args)
+            # Get the first successful result
+            mr_client_success = next(iter(sampler))
             assert mr_client_success is not None, "Client initialization failed after granting permissions"
             LOGGER.info("Client instantiated successfully after granting permissions.")
-
         except Exception as e:
-            # If we get an exception here, it's unexpected, especially 403
-            LOGGER.error(f"Received unexpected general error after granting access: {e}", exc_info=True)
+            LOGGER.error(f"Failed to access Model Registry after granting permissions: {e}", exc_info=True)
             raise
 
         LOGGER.info("--- RBAC Test Completed Successfully ---")
+
+
+@retry(wait_timeout=120, sleep=5, exceptions_dict={UnauthorizedException: []})
+def _try_connection_expect_forbidden(client_args: dict[str, Any]) -> ForbiddenException:
+    """
+    Attempts to create a ModelRegistryClient and expects ForbiddenException.
+    Retries on UnauthorizedException (401) during kube-rbac-proxy initialization.
+    Returns the ForbiddenException when received.
+    """
+    try:
+        ModelRegistryClient(**client_args)
+        raise AssertionError("Expected ForbiddenException but client connection succeeded")
+    except ForbiddenException as e:
+        # This is what we want - 403 Forbidden
+        return e