Move the byoidc secret to a different ns and address coderabbit comments

Author: dbasunag

tests/model_registry/conftest.py

@@ -34,8 +34,13 @@
 from pytest_testconfig import config as py_config
 from model_registry.types import RegisteredModel
 
-from tests.model_registry.rbac.utils import wait_for_oauth_openshift_deployment, get_byoidc_user_credentials
-from tests.model_registry.utils import generate_namespace_name, get_rest_headers, wait_for_default_resource_cleanedup
+from tests.model_registry.rbac.utils import wait_for_oauth_openshift_deployment
+from tests.model_registry.utils import (
+    generate_namespace_name,
+    get_rest_headers,
+    wait_for_default_resource_cleanedup,
+    get_byoidc_user_credentials,
+)
 from utilities.general import generate_random_name, wait_for_pods_running
 
 from tests.model_registry.constants import (
@@ -486,7 +491,7 @@ def test_idp_user(
     original_user: str,
     api_server_url: str,
     is_byoidc: bool,
-) -> Generator[UserTestSession, None, None]:
+) -> Generator[UserTestSession | None, None, None]:
     """
     Session-scoped fixture that creates a test IDP user and cleans it up after all tests.
     Returns a UserTestSession object that contains all necessary credentials and contexts.
@@ -545,56 +550,62 @@ def original_user() -> str:
 
 @pytest.fixture(scope="module")
 def created_htpasswd_secret(
-    original_user: str, user_credentials_rbac: dict[str, str]
-) -> Generator[UserTestSession, None, None]:
+    is_byoidc: bool, original_user: str, user_credentials_rbac: dict[str, str]
+) -> Generator[UserTestSession | None, None, None]:
     """
     Session-scoped fixture that creates a test IDP user and cleans it up after all tests.
     Returns a UserTestSession object that contains all necessary credentials and contexts.
     """
+    if is_byoidc:
+        yield
 
-    temp_path, htpasswd_b64 = create_htpasswd_file(
-        username=user_credentials_rbac["username"], password=user_credentials_rbac["password"]
-    )
-    try:
-        LOGGER.info(f"Creating secret {user_credentials_rbac['secret_name']} in openshift-config namespace")
-        with Secret(
-            name=user_credentials_rbac["secret_name"],
-            namespace="openshift-config",
-            htpasswd=htpasswd_b64,
-            type="Opaque",
-            wait_for_resource=True,
-        ) as secret:
-            yield secret
-    finally:
-        # Clean up the temporary file
-        temp_path.unlink(missing_ok=True)
+    else:
+        temp_path, htpasswd_b64 = create_htpasswd_file(
+            username=user_credentials_rbac["username"], password=user_credentials_rbac["password"]
+        )
+        try:
+            LOGGER.info(f"Creating secret {user_credentials_rbac['secret_name']} in openshift-config namespace")
+            with Secret(
+                name=user_credentials_rbac["secret_name"],
+                namespace="openshift-config",
+                htpasswd=htpasswd_b64,
+                type="Opaque",
+                wait_for_resource=True,
+            ) as secret:
+                yield secret
+        finally:
+            # Clean up the temporary file
+            temp_path.unlink(missing_ok=True)
 
 
 @pytest.fixture(scope="module")
 def updated_oauth_config(
-    admin_client: DynamicClient, original_user: str, user_credentials_rbac: dict[str, str]
+    is_byoidc: bool, admin_client: DynamicClient, original_user: str, user_credentials_rbac: dict[str, str]
 ) -> Generator[Any, None, None]:
-    # Get current providers and add the new one
-    oauth = OAuth(name="cluster")
-    identity_providers = oauth.instance.spec.identityProviders
-
-    new_idp = {
-        "name": user_credentials_rbac["idp_name"],
-        "mappingMethod": "claim",
-        "type": "HTPasswd",
-        "htpasswd": {"fileData": {"name": user_credentials_rbac["secret_name"]}},
-    }
-    updated_providers = identity_providers + [new_idp]
-
-    LOGGER.info("Updating OAuth")
-    identity_providers_patch = ResourceEditor(patches={oauth: {"spec": {"identityProviders": updated_providers}}})
-    identity_providers_patch.update(backup_resources=True)
-    # Wait for OAuth server to be ready
-    wait_for_oauth_openshift_deployment()
-    LOGGER.info(f"Added IDP {user_credentials_rbac['idp_name']} to OAuth configuration")
-    yield
-    identity_providers_patch.restore()
-    wait_for_oauth_openshift_deployment()
+    if is_byoidc:
+        yield
+    else:
+        # Get current providers and add the new one
+        oauth = OAuth(name="cluster")
+        identity_providers = oauth.instance.spec.identityProviders
+
+        new_idp = {
+            "name": user_credentials_rbac["idp_name"],
+            "mappingMethod": "claim",
+            "type": "HTPasswd",
+            "htpasswd": {"fileData": {"name": user_credentials_rbac["secret_name"]}},
+        }
+        updated_providers = identity_providers + [new_idp]
+
+        LOGGER.info("Updating OAuth")
+        identity_providers_patch = ResourceEditor(patches={oauth: {"spec": {"identityProviders": updated_providers}}})
+        identity_providers_patch.update(backup_resources=True)
+        # Wait for OAuth server to be ready
+        wait_for_oauth_openshift_deployment()
+        LOGGER.info(f"Added IDP {user_credentials_rbac['idp_name']} to OAuth configuration")
+        yield
+        identity_providers_patch.restore()
+        wait_for_oauth_openshift_deployment()
 
 
 @pytest.fixture(scope="module")

tests/model_registry/model_catalog/conftest.py

@@ -18,14 +18,14 @@
     REDHAT_AI_CATALOG_ID,
 )
 from tests.model_registry.constants import CUSTOM_CATALOG_ID1
-from tests.model_registry.rbac.utils import get_mr_user_token
 from tests.model_registry.utils import (
     get_rest_headers,
     is_model_catalog_ready,
     get_model_catalog_pod,
     wait_for_model_catalog_api,
     execute_get_command,
     get_model_str,
+    get_mr_user_token,
 )
 from utilities.infra import get_openshift_token, create_inference_token, login_with_user_password
 

tests/model_registry/rbac/test_mr_rbac.py

@@ -27,15 +27,14 @@
     build_mr_client_args,
     assert_positive_mr_registry,
     assert_forbidden_access,
-    get_mr_user_token,
 )
 from tests.model_registry.constants import NUM_MR_INSTANCES
 from utilities.infra import get_openshift_token
 from mr_openapi.exceptions import ForbiddenException
 from utilities.user_utils import UserTestSession
 from kubernetes.dynamic import DynamicClient
 from ocp_resources.model_registry_modelregistry_opendatahub_io import ModelRegistry
-from tests.model_registry.utils import get_mr_service_by_label, get_endpoint_from_mr_service
+from tests.model_registry.utils import get_mr_service_by_label, get_endpoint_from_mr_service, get_mr_user_token
 from tests.model_registry.rbac.utils import grant_mr_access, revoke_mr_access
 from utilities.constants import Protocols
 

tests/model_registry/rbac/utils.py

@@ -1,6 +1,3 @@
-import base64
-
-import requests
 from typing import Any, Dict, Generator, List
 
 from kubernetes.dynamic import DynamicClient
@@ -9,11 +6,10 @@
 from ocp_resources.deployment import Deployment
 from ocp_resources.role import Role
 from ocp_resources.role_binding import RoleBinding
-from ocp_resources.secret import Secret
 from utilities.constants import Protocols
 import logging
 from model_registry import ModelRegistry as ModelRegistryClient
-from utilities.infra import get_openshift_token, get_cluster_authentication
+from utilities.infra import get_openshift_token
 from mr_openapi.exceptions import ForbiddenException
 
 LOGGER = logging.getLogger(__name__)
@@ -182,91 +178,3 @@ def assert_forbidden_access(endpoint: str, token: str) -> None:
     except ForbiddenException:
         # This is what we want - access is properly forbidden
         pass
-
-
-def get_byoidc_issuer_url(admin_client: DynamicClient) -> str:
-    authentication = get_cluster_authentication(admin_client=admin_client)
-    assert authentication is not None
-    url = authentication.instance.spec.oidcProviders[0].issuer.issuerURL
-    assert url is not None
-    return url
-
-
-def get_mr_user_token(admin_client: DynamicClient, user_credentials_rbac: dict[str, str]) -> str:
-    url = f"{get_byoidc_issuer_url(admin_client=admin_client)}/protocol/openid-connect/token"
-    headers = {"Content-Type": "application/x-www-form-urlencoded", "User-Agent": "python-requests"}
-
-    data = {
-        "username": user_credentials_rbac["username"],
-        "password": user_credentials_rbac["password"],
-        "grant_type": "password",
-        "client_id": "oc-cli",
-        "scope": "openid",
-    }
-
-    try:
-        LOGGER.info(f"Requesting token for user {user_credentials_rbac['username']} in byoidc environment")
-        response = requests.post(
-            url=url,
-            headers=headers,
-            data=data,
-            allow_redirects=True,
-            timeout=30,
-            verify=True,  # Set to False if you need to skip SSL verification
-        )
-        response.raise_for_status()
-        json_response = response.json()
-
-        # Validate that we got an access token
-        if "id_token" not in json_response:
-            LOGGER.error("Warning: No access_token in response")
-            raise AssertionError(f"No access_token in response: {json_response}")
-        return json_response["id_token"]
-    except Exception as e:
-        raise e
-
-
-def get_byoidc_user_credentials(username: str = None) -> Dict[str, str]:
-    """
-    Get user credentials from byoidc-credentials secret.
-
-    Args:
-        username: Specific username to look up. If None, returns first user.
-
-    Returns:
-        Dictionary with username and password for the specified user.
-
-    Raises:
-        ValueError: If username not found or no users/passwords in secret.
-        AssertionError: If users or passwords lists are empty.
-    """
-    credentials_secret = Secret(name="byoidc-credentials", namespace="default", ensure_exists=True)
-    credential_data = credentials_secret.instance.data
-    user_names = base64.b64decode(credential_data.users).decode().split(",")
-    passwords = base64.b64decode(credential_data.passwords).decode().split(",")
-
-    # Assert that both lists are not empty
-    assert user_names and user_names != [""], "No usernames found in byoidc-credentials secret"
-    assert passwords and passwords != [""], "No passwords found in byoidc-credentials secret"
-
-    # Use specified username or default to first user
-    requested_username = username if username else user_names[0]
-
-    if requested_username and requested_username in user_names:
-        # Find the index of the requested username
-        user_index = user_names.index(requested_username)
-        if user_index < len(passwords):
-            selected_username = user_names[user_index]
-            selected_password = passwords[user_index]
-        else:
-            raise ValueError(f"Password not found for user '{requested_username}' at index {user_index}")
-    elif requested_username:
-        raise ValueError(f"Username '{requested_username}' not found in byoidc credentials")
-    else:
-        raise ValueError("No users found in byoidc credentials")
-
-    LOGGER.info(f"Using byoidc-credentials username='{selected_username}'")
-    return {
-        "username": selected_username,
-        "password": selected_password,
-    }

tests/model_registry/utils.py

@@ -1,6 +1,7 @@
+import base64
 import json
 import time
-from typing import Any, List
+from typing import Any, List, Dict
 
 import requests
 from kubernetes.dynamic import DynamicClient
@@ -33,6 +34,7 @@
 from model_registry.types import RegisteredModel
 
 from utilities.general import wait_for_pods_running
+from utilities.infra import get_cluster_authentication
 
 ADDRESS_ANNOTATION_PREFIX: str = "routing.opendatahub.io/external-address-"
 MARIA_DB_IMAGE = (
@@ -797,3 +799,91 @@ def wait_for_default_resource_cleanedup(admin_client: DynamicClient, namespace_n
     if not objects_not_deleted:
         return True
     raise ResourceNotDeleted(f"Following objects are not deleted: {objects_not_deleted}")
+
+
+def get_mr_user_token(admin_client: DynamicClient, user_credentials_rbac: dict[str, str]) -> str:
+    url = f"{get_byoidc_issuer_url(admin_client=admin_client)}/protocol/openid-connect/token"
+    headers = {"Content-Type": "application/x-www-form-urlencoded", "User-Agent": "python-requests"}
+
+    data = {
+        "username": user_credentials_rbac["username"],
+        "password": user_credentials_rbac["password"],
+        "grant_type": "password",
+        "client_id": "oc-cli",
+        "scope": "openid",
+    }
+
+    try:
+        LOGGER.info(f"Requesting token for user {user_credentials_rbac['username']} in byoidc environment")
+        response = requests.post(
+            url=url,
+            headers=headers,
+            data=data,
+            allow_redirects=True,
+            timeout=30,
+            verify=True,  # Set to False if you need to skip SSL verification
+        )
+        response.raise_for_status()
+        json_response = response.json()
+
+        # Validate that we got an access token
+        if "id_token" not in json_response:
+            LOGGER.error("Warning: No access_token in response")
+            raise AssertionError(f"No access_token in response: {json_response}")
+        return json_response["id_token"]
+    except Exception as e:
+        raise e
+
+
+def get_byoidc_issuer_url(admin_client: DynamicClient) -> str:
+    authentication = get_cluster_authentication(admin_client=admin_client)
+    assert authentication is not None
+    url = authentication.instance.spec.oidcProviders[0].issuer.issuerURL
+    assert url is not None
+    return url
+
+
+def get_byoidc_user_credentials(username: str = None) -> Dict[str, str]:
+    """
+    Get user credentials from byoidc-credentials secret.
+
+    Args:
+        username: Specific username to look up. If None, returns first user.
+
+    Returns:
+        Dictionary with username and password for the specified user.
+
+    Raises:
+        ValueError: If username not found or no users/passwords in secret.
+        AssertionError: If users or passwords lists are empty.
+    """
+    credentials_secret = Secret(name="byoidc-credentials", namespace="oidc", ensure_exists=True)
+    credential_data = credentials_secret.instance.data
+    user_names = base64.b64decode(credential_data.users).decode().split(",")
+    passwords = base64.b64decode(credential_data.passwords).decode().split(",")
+
+    # Assert that both lists are not empty
+    assert user_names and user_names != [""], "No usernames found in byoidc-credentials secret"
+    assert passwords and passwords != [""], "No passwords found in byoidc-credentials secret"
+
+    # Use specified username or default to first user
+    requested_username = username if username else user_names[0]
+
+    if requested_username and requested_username in user_names:
+        # Find the index of the requested username
+        user_index = user_names.index(requested_username)
+        if user_index < len(passwords):
+            selected_username = user_names[user_index]
+            selected_password = passwords[user_index]
+        else:
+            raise ValueError(f"Password not found for user '{requested_username}' at index {user_index}")
+    elif requested_username:
+        raise ValueError(f"Username '{requested_username}' not found in byoidc credentials")
+    else:
+        raise ValueError("No users found in byoidc credentials")
+
+    LOGGER.info(f"Using byoidc-credentials username='{selected_username}'")
+    return {
+        "username": selected_username,
+        "password": selected_password,
+    }