updates to MR related tests to ensure they work against BYOIDC (#862)

* updates to MR related tests to ensure they work against BYOIDC

* address tox failure

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

---------

Co-authored-by: Luca Giorgi <lgiorgi@redhat.com>

Author: dbasunag

tests/conftest.py

@@ -45,6 +45,7 @@
     login_with_user_password,
     get_openshift_token,
     download_oc_console_cli,
+    get_cluster_authentication,
 )
 from utilities.constants import (
     AcceleratorType,
@@ -379,10 +380,7 @@ def kubconfig_filepath() -> str:
 
 @pytest.fixture(scope="session")
 def cluster_authentication(admin_client: DynamicClient) -> Authentication | None:
-    auth = Authentication(client=admin_client, name="cluster")
-    if auth.exists:
-        return auth
-    return None
+    return get_cluster_authentication(admin_client=admin_client)
 
 
 @pytest.fixture(scope="session")

tests/model_registry/conftest.py

@@ -35,7 +35,12 @@
 from model_registry.types import RegisteredModel
 
 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
+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,13 +491,14 @@ 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.
     """
     if is_byoidc:
-        pytest.skip("Working on OIDC support for tests that use test_idp_user")
+        # For BYOIDC, we would be using a preconfigured group and username for actual api calls.
+        yield
     else:
         user_credentials_rbac = request.getfixturevalue(argname="user_credentials_rbac")
         _ = request.getfixturevalue(argname="created_htpasswd_secret")
@@ -544,67 +550,84 @@ 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]
+    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()
+        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")
-def user_credentials_rbac() -> dict[str, str]:
-    random_str = generate_random_name()
-    return {
-        "username": f"test-user-{random_str}",
-        "password": f"test-password-{random_str}",
-        "idp_name": f"test-htpasswd-idp-{random_str}",
-        "secret_name": f"test-htpasswd-secret-{random_str}",
-    }
+def user_credentials_rbac(
+    is_byoidc: bool,
+) -> dict[str, str]:
+    if is_byoidc:
+        byoidc_creds = get_byoidc_user_credentials(username="mr-non-admin")
+        return {
+            "username": byoidc_creds["username"],
+            "password": byoidc_creds["password"],
+            "idp_name": "byoidc",
+            "secret_name": None,
+        }
+    else:
+        random_str = generate_random_name()
+        return {
+            "username": f"test-user-{random_str}",
+            "password": f"test-password-{random_str}",
+            "idp_name": f"test-htpasswd-idp-{random_str}",
+            "secret_name": f"test-htpasswd-secret-{random_str}",
+        }
 
 
 @pytest.fixture(scope="class")

tests/model_registry/model_catalog/conftest.py

@@ -25,9 +25,9 @@
     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
-from utilities.user_utils import UserTestSession
+from utilities.infra import get_openshift_token, create_inference_token, login_with_user_password
 
 
 LOGGER = get_logger(name=__name__)
@@ -93,10 +93,12 @@ def update_configmap_data_add_model(
 
 @pytest.fixture(scope="class")
 def user_token_for_api_calls(
+    is_byoidc: bool,
+    admin_client: DynamicClient,
     request: pytest.FixtureRequest,
     original_user: str,
     api_server_url: str,
-    test_idp_user: UserTestSession,
+    user_credentials_rbac: dict[str, str],
     service_account: ServiceAccount,
 ) -> Generator[str, None, None]:
     param = getattr(request, "param", {})
@@ -106,8 +108,20 @@ def user_token_for_api_calls(
         LOGGER.info("Logging in as admin user")
         yield get_openshift_token()
     elif user == "test":
-        # TODO: implement byoidc check in get_openshift_token
-        yield get_openshift_token()
+        if not is_byoidc:
+            login_with_user_password(
+                api_address=api_server_url,
+                user=user_credentials_rbac["username"],
+                password=user_credentials_rbac["password"],
+            )
+            yield get_openshift_token()
+            LOGGER.info(f"Logging in as {original_user}")
+            login_with_user_password(
+                api_address=api_server_url,
+                user=original_user,
+            )
+        else:
+            yield get_mr_user_token(admin_client=admin_client, user_credentials_rbac=user_credentials_rbac)
     elif user == "sa_user":
         yield create_inference_token(service_account)
     else:

tests/model_registry/rbac/conftest.py

@@ -57,6 +57,7 @@ def add_user_to_group(
 
 @pytest.fixture(scope="function")
 def model_registry_group_with_user(
+    is_byoidc: bool,
     admin_client: DynamicClient,
     test_idp_user: UserTestSession,
 ) -> Generator[Group, None, None]:
@@ -71,24 +72,28 @@ def model_registry_group_with_user(
     Yields:
         Group: The group with the test user added
     """
-    group_name = f"{MR_INSTANCE_NAME}-users"
-    group = Group(
-        client=admin_client,
-        name=group_name,
-        wait_for_resource=True,
-    )
+    if is_byoidc:
+        # this is no op. byoidc already has a group with user model-registry-user
+        yield
+    else:
+        group_name = f"{MR_INSTANCE_NAME}-users"
+        group = Group(
+            client=admin_client,
+            name=group_name,
+            wait_for_resource=True,
+        )
 
-    # Add user to group
-    with ResourceEditor(
-        patches={
-            group: {
-                "metadata": {"name": group_name},
-                "users": [test_idp_user.username],
+        # Add user to group
+        with ResourceEditor(
+            patches={
+                group: {
+                    "metadata": {"name": group_name},
+                    "users": [test_idp_user.username],
+                }
             }
-        }
-    ) as _:
-        LOGGER.info(f"Added user {test_idp_user.username} to {group_name} group")
-        yield group
+        ) as _:
+            LOGGER.info(f"Added user {test_idp_user.username} to {group_name} group")
+            yield group
 
 
 @pytest.fixture(scope="function")
@@ -111,18 +116,22 @@ def created_role_binding_group(
 
 @pytest.fixture(scope="function")
 def created_role_binding_user(
+    is_byoidc: bool,
     admin_client: DynamicClient,
     model_registry_namespace: str,
     mr_access_role: Role,
-    test_idp_user: UserTestSession,
+    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']}")
     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=test_idp_user.username,
+        subjects_name=user_credentials_rbac["username"],
     )
 
 
@@ -223,17 +232,26 @@ def model_registry_instance_parametrized(
 
 @pytest.fixture()
 def login_as_test_user(
-    api_server_url: str, original_user: str, test_idp_user: UserTestSession
+    is_byoidc: bool, api_server_url: str, original_user: str, test_idp_user: UserTestSession
 ) -> Generator[None, None, None]:
-    LOGGER.info(f"Logging in as {test_idp_user.username}")
-    login_with_user_password(
-        api_address=api_server_url,
-        user=test_idp_user.username,
-        password=test_idp_user.password,
-    )
-    yield
-    LOGGER.info(f"Logging in as {original_user}")
-    login_with_user_password(
-        api_address=api_server_url,
-        user=original_user,
-    )
+    if is_byoidc:
+        yield
+    else:
+        LOGGER.info(f"Logging in as {test_idp_user.username}")
+        login_with_user_password(
+            api_address=api_server_url,
+            user=test_idp_user.username,
+            password=test_idp_user.password,
+        )
+        yield
+        LOGGER.info(f"Logging in as {original_user}")
+        login_with_user_password(
+            api_address=api_server_url,
+            user=original_user,
+        )
+
+
+@pytest.fixture()
+def skip_test_on_byoidc(is_byoidc: bool) -> None:
+    if is_byoidc:
+        pytest.skip(reason="This test is meant to skip on byoidc.")

tests/model_registry/rbac/multiple_instance_utils.py

@@ -3,11 +3,11 @@
 
 from tests.model_registry.constants import (
     MODEL_REGISTRY_DB_SECRET_STR_DATA,
-    MR_INSTANCE_NAME,
     DB_BASE_RESOURCES_NAME,
     NUM_MR_INSTANCES,
     MODEL_REGISTRY_DB_SECRET_ANNOTATIONS,
     OAUTH_PROXY_CONFIG_DICT,
+    MR_INSTANCE_BASE_NAME,
 )
 from tests.model_registry.utils import (
     get_model_registry_db_label_dict,
@@ -76,9 +76,9 @@
 
 model_registry_instance_params = [
     {
-        "name": f"{MR_INSTANCE_NAME}{index}",
+        "name": f"{MR_INSTANCE_BASE_NAME}{index}",
         "namespace": ns_name,
-        "label": get_mr_standard_labels(resource_name=f"{MR_INSTANCE_NAME}{index}"),
+        "label": get_mr_standard_labels(resource_name=f"{MR_INSTANCE_BASE_NAME}{index}"),
         "grpc": {},
         "rest": {},
         "mysql": get_mysql_config(base_name=f"{DB_BASE_RESOURCES_NAME}{index}", namespace=ns_name, db_backend="mysql"),