fix: address comment

Author: fege

tests/model_registry/rbac/conftest.py

@@ -10,10 +10,12 @@
 from ocp_resources.role_binding import RoleBinding
 from ocp_resources.role import Role
 from ocp_resources.group import Group
+from ocp_resources.resource import ResourceEditor
 from kubernetes.dynamic import DynamicClient
 from pyhelper_utils.shell import run_command
 from tests.model_registry.utils import generate_random_name, generate_namespace_name
 from utilities.user_utils import create_test_idp, UserTestSession
+from tests.model_registry.rbac.group_utils import create_group
 from tests.model_registry.constants import MR_INSTANCE_NAME
 
 
@@ -93,63 +95,60 @@ def sa_token(service_account: ServiceAccount) -> str:
 
 
 @pytest.fixture(scope="function")
-def new_group(
+def add_user_to_group(
     request: pytest.FixtureRequest,
     admin_client: DynamicClient,
     test_idp_user_session: UserTestSession,
 ) -> Generator[str, None, None]:
-    """
-    Fixture to create a new OpenShift group and add a user, then delete the group after the test.
-    The group name is passed as a parameter to the fixture and the user is taken from test_idp_user_session.
-    """
-
     group_name = request.param
-    with Group(
-        client=admin_client,
-        name=group_name,
+    with create_group(
+        admin_client=admin_client,
+        group_name=group_name,
         users=[test_idp_user_session.username],
-        wait_for_resource=True,
-    ) as _:
-        LOGGER.info(f"Group {group_name} created successfully.")
+    ) as group_name:
         yield group_name
-        LOGGER.info(f"Group {group_name} deletion initiated by context manager.")
 
 
 @pytest.fixture(scope="function")
 def model_registry_group_with_user(
+    request: pytest.FixtureRequest,
     admin_client: DynamicClient,
     test_idp_user_session: UserTestSession,
 ) -> Generator[Group, None, None]:
     """
-    Fixture to manage a test user in the Model Registry group.
+    Fixture to manage a test user in a specified group.
     Adds the user to the group before the test, then removes them after.
 
     Args:
+        request: The pytest request object containing the group name parameter
         admin_client: The admin client for accessing the cluster
         test_idp_user_session: The test user session containing user information
 
     Yields:
-        Group: The Model Registry group with the test user added
+        Group: The group with the test user added
+
+    Note:
+        The group name should be passed as a parameter to the fixture using pytest.mark.parametrize
     """
-    model_registry_users_group = f"{MR_INSTANCE_NAME}-users"
+    group_name = request.param
     group = Group(
         client=admin_client,
-        name=model_registry_users_group,
+        name=group_name,
         wait_for_resource=True,
     )
 
     # Add user to group
-    group.update(
-        resource_dict={"metadata": {"name": model_registry_users_group}, "users": [test_idp_user_session.username]}
-    )
-    LOGGER.info(f"Added user {test_idp_user_session.username} to {model_registry_users_group} group")
-
-    try:
+    with ResourceEditor(
+        patches={
+            group: {
+                "metadata": {"name": group_name},
+                "users": [test_idp_user_session.username],
+            }
+        }
+    ) as _:
+        LOGGER.info(f"Added user {test_idp_user_session.username} to {group_name} group")
         yield group
-    finally:
-        # Remove user from group
-        group.update(resource_dict={"metadata": {"name": model_registry_users_group}, "users": []})
-        LOGGER.info(f"Removed user {test_idp_user_session.username} from {model_registry_users_group} group")
+        LOGGER.info(f"Removed user {test_idp_user_session.username} from {group_name} group")
 
 
 @pytest.fixture(scope="session")

tests/model_registry/rbac/group_utils.py

@@ -0,0 +1,44 @@
+from contextlib import contextmanager
+from typing import Generator
+from simple_logger.logger import get_logger
+from kubernetes.dynamic import DynamicClient
+from ocp_resources.group import Group
+
+LOGGER = get_logger(name=__name__)
+
+
+@contextmanager
+def create_group(
+    admin_client: DynamicClient,
+    group_name: str,
+    users: list[str] | None = None,
+    wait_for_resource: bool = True,
+) -> Generator[str, None, None]:
+    """
+    Factory function to create an OpenShift group with optional users.
+    Uses context manager to ensure proper cleanup.
+
+    Args:
+        admin_client: The admin client to use for group operations
+        group_name: Name of the group to create
+        users: Optional list of usernames to add to the group
+        wait_for_resource: Whether to wait for the group to be ready
+
+    Yields:
+        The group name
+
+    Example:
+        with create_group(admin_client, "my-group", users=["user1", "user2"]) as group_name:
+            # Use the group
+            pass
+        # Group is automatically deleted
+    """
+    with Group(
+        client=admin_client,
+        name=group_name,
+        users=users or [],
+        wait_for_resource=wait_for_resource,
+    ) as _:
+        LOGGER.info(f"Group {group_name} created successfully.")
+        yield group_name
+        LOGGER.info(f"Group {group_name} deletion initiated by context manager.")

tests/model_registry/rbac/test_mr_rbac.py

@@ -3,12 +3,8 @@
 from simple_logger.logger import get_logger
 
 from model_registry import ModelRegistry as ModelRegistryClient
-from tests.model_registry.constants import MR_NAMESPACE
-from tests.model_registry.rbac.utils import (
-    assert_positive_mr_registry,
-    get_mr_client_args,
-    verify_group_membership,
-)
+from tests.model_registry.constants import MR_NAMESPACE, MR_INSTANCE_NAME
+from tests.model_registry.rbac.utils import assert_positive_mr_registry, get_mr_client_args
 from utilities.infra import switch_user_context
 from kubernetes.dynamic import DynamicClient
 from ocp_resources.group import Group
@@ -37,7 +33,6 @@
         })
     ],
     indirect=True,
-    scope="class",
 )
 @pytest.mark.usefixtures("updated_dsc_component_state_scope_class")
 class TestUserPermission:
@@ -89,10 +84,7 @@ def test_user_permission(
         context_to_use = (
             test_idp_user_session.original_context if use_admin_context else test_idp_user_session.user_context
         )
-        LOGGER.info(
-            f"-----Testing Model Registry access for user '{test_idp_user_session.username}' "
-            f"- Expected Access: {'Granted' if use_admin_context else 'Denied'}-----"
-        )
+
         with switch_user_context(context_to_use):
             _, client_args = get_mr_client_args(
                 model_registry_instance=model_registry_instance,
@@ -111,7 +103,16 @@ def test_user_permission(
                 LOGGER.info("Successfully received expected HTTP 403 status code")
 
     @pytest.mark.sanity
-    @pytest.mark.usefixtures("model_registry_group_with_user")
+    @pytest.mark.parametrize(
+        "model_registry_group_with_user",
+        [
+            pytest.param(
+                f"{MR_INSTANCE_NAME}-users",
+                id="model_registry_users",
+            ),
+        ],
+        indirect=["model_registry_group_with_user"],
+    )
     def test_user_added_to_group(
         self: Self,
         model_registry_instance: ModelRegistry,
@@ -139,15 +140,6 @@ def test_user_added_to_group(
             AssertionError: If access permissions don't match expectations
             ForbiddenException: Expected before group addition, unexpected after
         """
-        LOGGER.info(
-            "-----Test that a user can access to the Model Registry once added to a "
-            "group that has the permissions to access it-----"
-        )
-        # Verify group membership
-        verify_group_membership(
-            group=model_registry_group_with_user,
-            username=test_idp_user_session.username,
-        )
 
         # Wait for access to be granted
         with switch_user_context(test_idp_user_session.user_context):
@@ -165,16 +157,16 @@ def test_user_added_to_group(
 
     @pytest.mark.sanity
     @pytest.mark.parametrize(
-        "new_group",
+        "add_user_to_group",
         [
             pytest.param(
                 NEW_GROUP_NAME,
                 id="new_group",
             ),
         ],
-        indirect=["new_group"],
+        indirect=["add_user_to_group"],
     )
-    @pytest.mark.usefixtures("mr_access_role", "new_group")
+    @pytest.mark.usefixtures("mr_access_role", "add_user_to_group")
     def test_create_group(
         self: Self,
         model_registry_instance: ModelRegistry,
@@ -187,7 +179,7 @@ def test_create_group(
         Test creating a new group and granting it Model Registry access.
 
         This test verifies that:
-        1. A new group can be created
+        1. A new group can be created and user added to it
         2. The group can be granted Model Registry access via RoleBinding
         3. Users in the group can access the Model Registry
 
@@ -201,10 +193,6 @@ def test_create_group(
         Raises:
             AssertionError: If group creation or access permissions don't match expectations
         """
-        LOGGER.info(
-            "-----Test that a new group can be granted access to Model Registry and user added to it can access MR-----"
-        )
-        LOGGER.info(f"Group {NEW_GROUP_NAME} created and user {test_idp_user_session.username} added to it")
 
         with RoleBinding(
             client=admin_client,
@@ -250,10 +238,7 @@ def test_add_single_user(
         Raises:
             AssertionError: If access permissions don't match expectations
         """
-        LOGGER.info(
-            "-----Test that adding a single user to the Model Registry's permitted list allows "
-            "that user to access the MR-----"
-        )
+
         with RoleBinding(
             client=admin_client,
             namespace=model_registry_namespace,

tests/model_registry/rbac/utils.py

@@ -7,7 +7,6 @@
 from ocp_resources.model_registry import ModelRegistry
 from utilities.infra import get_openshift_token
 from kubernetes.dynamic import DynamicClient
-from ocp_resources.group import Group
 
 LOGGER = logging.getLogger(__name__)
 
@@ -94,24 +93,3 @@ def get_mr_client_args(
     svc = get_mr_service_by_label(client=admin_client, ns=namespace_instance, mr_instance=model_registry_instance)
     endpoint = get_endpoint_from_mr_service(svc=svc, protocol=Protocols.REST)
     return token, build_mr_client_args(rest_endpoint=endpoint, token=token, author=author)
-
-
-def verify_group_membership(
-    group: Group,
-    username: str,
-) -> None:
-    """
-    Verify that a user is a member of a group.
-
-    Args:
-        group: The Group object to check membership in
-        username: The username to verify membership for
-
-    Raises:
-        AssertionError: If the user is not a member of the group
-    """
-    group_name = group.instance.get("metadata", {}).get("name")
-    LOGGER.info(f"Verifying user {username} is in group {group_name}")
-    users = group.instance.get("users", []) or []
-    assert username in users, f"User {username} not in group {group_name}. Current users: {users}"
-    LOGGER.info(f"Verified user {username} is in {group_name} group")