milvus-io--milvus/tests/python_client/cdc/testcases/test_rbac.py

"""
CDC sync tests for RBAC operations.
"""

import time
import pytest
from common.common_type import CaseLabel
from .base import TestCDCSyncBase, logger


@pytest.mark.tags(CaseLabel.CDC)
class TestCDCSyncRBAC(TestCDCSyncBase):
    """Test CDC sync for RBAC operations."""

    def setup_method(self):
        """Setup for each test method."""
        self.resources_to_cleanup = []

    def teardown_method(self):
        """Cleanup after each test method - only cleanup upstream, downstream will sync."""
        upstream_client = getattr(self, "_upstream_client", None)

        if upstream_client:
            for resource_type, resource_name in self.resources_to_cleanup:
                if resource_type == "user":
                    self.cleanup_user(upstream_client, resource_name)
                elif resource_type == "role":
                    self.cleanup_role(upstream_client, resource_name)

            time.sleep(1)  # Allow cleanup to sync to downstream

    def test_create_role(self, upstream_client, downstream_client, sync_timeout):
        """Test CREATE_ROLE operation sync."""
        # Store upstream client for teardown
        self._upstream_client = upstream_client

        role_name = self.gen_unique_name("test_role_create")
        self.resources_to_cleanup.append(("role", role_name))

        # Initial cleanup
        self.cleanup_role(upstream_client, role_name)

        # Create role in upstream
        upstream_client.create_role(role_name)
        assert role_name in upstream_client.list_roles()

        # Wait for sync to downstream
        def check_sync():
            return role_name in downstream_client.list_roles()

        assert self.wait_for_sync(check_sync, sync_timeout, f"create role {role_name}")

    def test_drop_role(self, upstream_client, downstream_client, sync_timeout):
        """Test DROP_ROLE operation sync."""
        # Store upstream client for teardown
        self._upstream_client = upstream_client

        role_name = self.gen_unique_name("test_role_drop")
        self.resources_to_cleanup.append(("role", role_name))

        # Initial cleanup
        self.cleanup_role(upstream_client, role_name)

        # Create role first
        upstream_client.create_role(role_name)

        # Wait for creation to sync
        def check_create():
            return role_name in downstream_client.list_roles()

        assert self.wait_for_sync(
            check_create, sync_timeout, f"create role {role_name}"
        )

        # Drop role in upstream
        upstream_client.drop_role(role_name)
        assert role_name not in upstream_client.list_roles()

        # Wait for drop to sync
        def check_drop():
            return role_name not in downstream_client.list_roles()

        assert self.wait_for_sync(check_drop, sync_timeout, f"drop role {role_name}")

    def test_create_user(self, upstream_client, downstream_client, sync_timeout):
        """Test CREATE_USER operation sync."""
        # Store upstream client for teardown
        self._upstream_client = upstream_client

        username = self.gen_unique_name("test_user_create", max_length=31)
        password = "TestPass123!"
        self.resources_to_cleanup.append(("user", username))

        # Initial cleanup
        self.cleanup_user(upstream_client, username)

        # Create user in upstream
        upstream_client.create_user(username, password)
        assert username in upstream_client.list_users()

        # Wait for sync to downstream
        def check_sync():
            return username in downstream_client.list_users()

        assert self.wait_for_sync(check_sync, sync_timeout, f"create user {username}")

    def test_drop_user(self, upstream_client, downstream_client, sync_timeout):
        """Test DROP_USER operation sync."""
        # Store upstream client for teardown
        self._upstream_client = upstream_client

        username = self.gen_unique_name("test_user_drop", max_length=31)
        password = "TestPass123!"
        self.resources_to_cleanup.append(("user", username))

        # Initial cleanup
        self.cleanup_user(upstream_client, username)

        # Create user first
        upstream_client.create_user(username, password)

        # Wait for creation to sync
        def check_create():
            return username in downstream_client.list_users()

        assert self.wait_for_sync(check_create, sync_timeout, f"create user {username}")

        # Drop user in upstream
        upstream_client.drop_user(username)
        assert username not in upstream_client.list_users()

        # Wait for drop to sync
        def check_drop():
            return username not in downstream_client.list_users()

        assert self.wait_for_sync(check_drop, sync_timeout, f"drop user {username}")

    def test_grant_role(self, upstream_client, downstream_client, sync_timeout):
        """Test GRANT_ROLE operation sync."""
        # Store upstream client for teardown
        self._upstream_client = upstream_client

        username = self.gen_unique_name("test_user_grant", max_length=31)
        role_name = self.gen_unique_name("test_role_grant")
        password = "TestPass123!"
        self.resources_to_cleanup.append(("user", username))
        self.resources_to_cleanup.append(("role", role_name))

        # Initial cleanup
        self.cleanup_user(upstream_client, username)
        self.cleanup_role(upstream_client, role_name)

        # Create user and role
        upstream_client.create_user(username, password)
        upstream_client.create_role(role_name)

        # Wait for creation to sync
        def check_create():
            return (
                username in downstream_client.list_users()
                and role_name in downstream_client.list_roles()
            )

        assert self.wait_for_sync(
            check_create, sync_timeout, "create user/role for grant"
        )

        # Grant role to user
        upstream_client.grant_role(username, role_name)

        # Wait for grant to sync
        def check_grant():
            # Allow operation to propagate
            time.sleep(2)
            # Verify role is bound to user using describe_user
            try:
                user_info = downstream_client.describe_user(username)
                user_roles = user_info.get("roles", [])
                print(f"DEBUG: user_roles in check_grant: {user_roles}")
                return role_name in user_roles
            except Exception as e:
                logger.debug(f"Error checking user roles: {e}")
                return False

        assert self.wait_for_sync(
            check_grant, sync_timeout, f"grant role {role_name} to user {username}"
        )

    def test_revoke_role(self, upstream_client, downstream_client, sync_timeout):
        """Test REVOKE_ROLE operation sync."""
        # Store upstream client for teardown
        self._upstream_client = upstream_client

        username = self.gen_unique_name("test_user_revoke", max_length=31)
        role_name = self.gen_unique_name("test_role_revoke")
        password = "TestPass123!"
        self.resources_to_cleanup.append(("user", username))
        self.resources_to_cleanup.append(("role", role_name))

        # Initial cleanup
        self.cleanup_user(upstream_client, username)
        self.cleanup_role(upstream_client, role_name)

        # Create user and role, then grant role
        upstream_client.create_user(username, password)
        upstream_client.create_role(role_name)
        upstream_client.grant_role(username, role_name)

        # Wait for setup to sync
        time.sleep(5)

        # Revoke role from user
        upstream_client.revoke_role(username, role_name)

        # Wait for revoke to sync
        def check_revoke():
            time.sleep(2)  # Allow operation to propagate
            # Verify role is removed from user using describe_user
            try:
                user_info = downstream_client.describe_user(username)
                user_roles = user_info.get("roles", [])
                print(f"DEBUG: user_roles in check_revoke: {user_roles}")
                return role_name not in user_roles
            except Exception as e:
                logger.debug(f"Error checking user roles: {e}")
                return False

        assert self.wait_for_sync(
            check_revoke, sync_timeout, f"revoke role {role_name} from user {username}"
        )

    def test_grant_privilege(self, upstream_client, downstream_client, sync_timeout):
        """Test GRANT_PRIVILEGE operation sync."""
        # Store upstream client for teardown
        self._upstream_client = upstream_client

        role_name = self.gen_unique_name("test_role_priv_grant")
        self.resources_to_cleanup.append(("role", role_name))

        # Initial cleanup
        self.cleanup_role(upstream_client, role_name)

        # Create role
        upstream_client.create_role(role_name)

        # Wait for creation to sync
        def check_create():
            return role_name in downstream_client.list_roles()

        assert self.wait_for_sync(
            check_create, sync_timeout, f"create role for privilege {role_name}"
        )

        # Grant privilege to role
        upstream_client.grant_privilege(
            role_name=role_name,
            object_type="Collection",
            privilege="Search",
            object_name="*",
        )

        # check privilege in upstream
        upstream_privilege = upstream_client.describe_role(role_name)
        print(f"DEBUG: upstream_privilege in check_grant: {upstream_privilege}")

        # Wait for privilege grant to sync
        def check_grant():
            time.sleep(2)
            # Verify privilege is actually granted using describe_role
            try:
                role_info = downstream_client.describe_role(role_name)
                print(f"DEBUG: role_info in check_grant: {role_info}")
                for privilege_info in role_info["privileges"]:
                    print(f"DEBUG: privilege_info in check_grant: {privilege_info}")
                    print(
                        f"DEBUG: privilege_info.get('object_type') in check_grant: {privilege_info.get('object_type')}"
                    )
                    print(
                        f"DEBUG: privilege_info.get('privilege') in check_grant: {privilege_info.get('privilege')}"
                    )
                    print(
                        f"DEBUG: privilege_info.get('object_name') in check_grant: {privilege_info.get('object_name')}"
                    )
                    if (
                        privilege_info.get("object_type") == "Collection"
                        and privilege_info.get("privilege") == "Search"
                        and privilege_info.get("object_name") == "*"
                    ):
                        return True
                return False
            except Exception as e:
                logger.debug(f"Error checking role privileges: {e}")
                return False

        assert self.wait_for_sync(
            check_grant, sync_timeout, f"grant privilege to role {role_name}"
        )

    def test_revoke_privilege(self, upstream_client, downstream_client, sync_timeout):
        """Test REVOKE_PRIVILEGE operation sync."""
        # Store upstream client for teardown
        self._upstream_client = upstream_client

        role_name = self.gen_unique_name("test_role_priv_revoke")
        self.resources_to_cleanup.append(("role", role_name))

        # Initial cleanup
        self.cleanup_role(upstream_client, role_name)

        # Create role and grant privilege
        upstream_client.create_role(role_name)

        upstream_client.grant_privilege(
            role_name=role_name,
            object_type="Collection",
            privilege="Search",
            object_name="*",
        )

        # Wait for setup
        time.sleep(3)

        # Revoke privilege from role
        upstream_client.revoke_privilege(
            role_name=role_name,
            object_type="Collection",
            privilege="Search",
            object_name="*",
        )

        # Wait for revoke to sync
        def check_revoke():
            time.sleep(2)
            # Verify privilege is actually revoked using describe_role
            try:
                role_info = downstream_client.describe_role(role_name)
                for privilege_info in role_info["privileges"]:
                    if (
                        privilege_info.get("object_type") == "Collection"
                        and privilege_info.get("privilege") == "Search"
                        and privilege_info.get("object_name") == "*"
                    ):
                        return False  # Should not find the privilege
                return True
            except Exception as e:
                logger.debug(f"Error checking role privileges: {e}")
                return False

        assert self.wait_for_sync(
            check_revoke, sync_timeout, f"revoke privilege from role {role_name}"
        )

    def test_update_password(
        self, upstream_client, downstream_client, sync_timeout, downstream_uri
    ):
        """Test UPDATE_PASSWORD operation sync."""
        # Store upstream client for teardown
        self._upstream_client = upstream_client

        username = self.gen_unique_name("test_user_pwd", max_length=31)
        old_password = "OldPass123!"
        new_password = "NewPass456!"
        self.resources_to_cleanup.append(("user", username))

        # Initial cleanup
        self.cleanup_user(upstream_client, username)

        # Create user first
        upstream_client.create_user(username, old_password)

        # Wait for creation to sync
        def check_create():
            return username in downstream_client.list_users()

        assert self.wait_for_sync(check_create, sync_timeout, f"create user {username}")

        # Update password in upstream
        upstream_client.update_password(username, old_password, new_password)

        # Wait for password update to sync
        def check_password_update():
            time.sleep(2)
            # Verify password update by trying to create a client with new credentials
            try:
                from pymilvus import MilvusClient

                # Try to connect with new password
                test_client = MilvusClient(
                    uri=downstream_uri, token=f"{username}:{new_password}"
                )
                # Simple operation to verify connection works
                test_client.list_collections()
                test_client.close()
                return True
            except Exception as e:
                logger.debug(f"Error verifying new password: {e}")
                return False

        assert self.wait_for_sync(
            check_password_update, sync_timeout, f"update password for user {username}"
        )

    def test_create_privilege_group(
        self, upstream_client, downstream_client, sync_timeout
    ):
        """Test CREATE_PRIVILEGE_GROUP operation sync."""
        # Store upstream client for teardown
        self._upstream_client = upstream_client

        group_name = self.gen_unique_name("test_priv_group")
        # Note: privilege groups need special cleanup handling

        # Initial cleanup - try to drop if exists
        try:
            upstream_client.drop_privilege_group(group_name)
        except:
            pass  # Ignore if doesn't exist

        # Create privilege group in upstream
        upstream_client.create_privilege_group(group_name)

        # Verify in upstream
        def check_create():
            try:
                upstream_groups = upstream_client.list_privilege_groups()
                print(f"DEBUG: upstream_groups in check_create: {upstream_groups}")
                return group_name in [
                    group["privilege_group"] for group in upstream_groups
                ]
            except Exception as e:
                logger.debug(f"Error checking privilege groups: {e}")
                return False

        assert self.wait_for_sync(
            check_create, sync_timeout, f"create privilege group {group_name}"
        )

        # Wait for sync to downstream
        def check_sync():
            try:
                downstream_groups = downstream_client.list_privilege_groups()
                return group_name in [
                    group["privilege_group"] for group in downstream_groups
                ]
            except Exception as e:
                logger.debug(f"Error checking privilege groups: {e}")
                return False

        assert self.wait_for_sync(
            check_sync, sync_timeout, f"create privilege group {group_name}"
        )

        # Cleanup
        try:
            upstream_client.drop_privilege_group(group_name)
        except:
            pass

    def test_drop_privilege_group(
        self, upstream_client, downstream_client, sync_timeout
    ):
        """Test DROP_PRIVILEGE_GROUP operation sync."""
        # Store upstream client for teardown
        self._upstream_client = upstream_client

        group_name = self.gen_unique_name("test_priv_group_drop")

        # Initial cleanup
        try:
            upstream_client.drop_privilege_group(group_name)
        except:
            pass

        # Create privilege group first
        upstream_client.create_privilege_group(group_name)

        # Wait for creation to sync
        def check_create():
            try:
                return group_name in [
                    group["privilege_group"]
                    for group in downstream_client.list_privilege_groups()
                ]
            except Exception as e:
                logger.debug(f"Error checking privilege groups: {e}")
                return False

        assert self.wait_for_sync(
            check_create, sync_timeout, f"create privilege group {group_name}"
        )

        # Drop privilege group in upstream
        upstream_client.drop_privilege_group(group_name)

        # Verify dropped in upstream
        upstream_groups = upstream_client.list_privilege_groups()
        assert group_name not in [group["privilege_group"] for group in upstream_groups]

        # Wait for drop to sync
        def check_drop():
            try:
                downstream_groups = downstream_client.list_privilege_groups()
                return group_name not in [
                    group["privilege_group"] for group in downstream_groups
                ]
            except Exception as e:
                logger.debug(f"Error checking privilege groups: {e}")
                return False

        assert self.wait_for_sync(
            check_drop, sync_timeout, f"drop privilege group {group_name}"
        )

    def test_grant_privilege_v2(self, upstream_client, downstream_client, sync_timeout):
        """Test GRANT_PRIVILEGE_V2 operation sync."""
        # Store upstream client for teardown
        self._upstream_client = upstream_client

        role_name = self.gen_unique_name("test_role_priv_v2")
        self.resources_to_cleanup.append(("role", role_name))

        # Initial cleanup
        self.cleanup_role(upstream_client, role_name)

        # Create role
        upstream_client.create_role(role_name)

        # Wait for creation to sync
        def check_create():
            return role_name in downstream_client.list_roles()

        assert self.wait_for_sync(
            check_create, sync_timeout, f"create role for privilege v2 {role_name}"
        )

        # Grant privilege using v2 API
        upstream_client.grant_privilege_v2(
            role_name=role_name, privilege="Query", collection_name="*"
        )
        # check privilege in upstream
        upstream_privilege = upstream_client.describe_role(role_name)["privileges"]
        print(f"DEBUG: upstream_privilege in check_grant: {upstream_privilege}")

        # Wait for privilege grant to sync
        def check_grant():
            time.sleep(2)
            # Verify privilege is actually granted using describe_role
            try:
                role_info = downstream_client.describe_role(role_name)
                for privilege_info in role_info["privileges"]:
                    if (
                        privilege_info.get("privilege") == "Query"
                        and privilege_info.get("object_name") == "*"
                    ):
                        return True
                return False
            except Exception as e:
                logger.debug(f"Error checking role privileges v2: {e}")
                return False

        assert self.wait_for_sync(
            check_grant, sync_timeout, f"grant privilege v2 to role {role_name}"
        )

    def test_revoke_privilege_v2(
        self, upstream_client, downstream_client, sync_timeout
    ):
        """Test REVOKE_PRIVILEGE_V2 operation sync."""
        # Store upstream client for teardown
        self._upstream_client = upstream_client

        role_name = self.gen_unique_name("test_role_revoke_v2")
        self.resources_to_cleanup.append(("role", role_name))

        # Initial cleanup
        self.cleanup_role(upstream_client, role_name)

        # Create role and grant privilege using v2 API
        upstream_client.create_role(role_name)
        upstream_client.grant_privilege_v2(
            role_name=role_name, privilege="Query", collection_name="*"
        )

        # Wait for setup
        time.sleep(3)

        # check privilege in upstream
        upstream_privilege = upstream_client.describe_role(role_name)["privileges"]
        print(f"DEBUG: upstream_privilege in check_revoke: {upstream_privilege}")

        # check privilege in downstream
        def check_grant():
            try:
                role_info = downstream_client.describe_role(role_name)
                for privilege_info in role_info["privileges"]:
                    if (
                        privilege_info.get("privilege") == "Query"
                        and privilege_info.get("object_name") == "*"
                    ):
                        return True
                return False
            except Exception as e:
                logger.debug(f"Error checking role privileges v2: {e}")
                return False

        assert self.wait_for_sync(
            check_grant, sync_timeout, f"grant privilege v2 to role {role_name}"
        )

        # Revoke privilege using v2 API
        upstream_client.revoke_privilege_v2(
            role_name=role_name, privilege="Query", collection_name="*"
        )
        # check privilege in upstream
        upstream_privilege = upstream_client.describe_role(role_name)["privileges"]
        print(f"DEBUG: upstream_privilege in check_revoke: {upstream_privilege}")

        # Wait for revoke to sync
        def check_revoke():
            time.sleep(2)
            # Verify privilege is actually revoked using describe_role
            try:
                role_info = downstream_client.describe_role(role_name)
                for privilege_info in role_info["privileges"]:
                    if (
                        privilege_info.get("privilege") == "Query"
                        and privilege_info.get("object_name") == "*"
                    ):
                        return False  # Should not find the privilege
                return True
            except Exception as e:
                logger.debug(f"Error checking role privileges v2: {e}")
                return False

        assert self.wait_for_sync(
            check_revoke, sync_timeout, f"revoke privilege v2 from role {role_name}"
        )

    def test_add_privileges_to_group(
        self, upstream_client, downstream_client, sync_timeout
    ):
        """Test ADD_PRIVILEGES_TO_GROUP operation sync."""
        # Store upstream client for teardown
        self._upstream_client = upstream_client

        group_name = self.gen_unique_name("test_priv_group_add")

        # Initial cleanup
        try:
            upstream_client.drop_privilege_group(group_name)
        except:
            pass

        # Create privilege group first
        upstream_client.create_privilege_group(group_name)

        # Wait for creation to sync
        def check_create():
            try:
                downstream_groups = downstream_client.list_privilege_groups()
                print(f"DEBUG: downstream_groups in check_create: {downstream_groups}")
                return group_name in [
                    group["privilege_group"] for group in downstream_groups
                ]
            except Exception as e:
                logger.debug(f"Error checking privilege groups: {e}")
                return False

        assert self.wait_for_sync(
            check_create, sync_timeout, f"create privilege group {group_name}"
        )

        # Add privileges to group in upstream
        privileges_to_add = ["Search", "Query"]
        upstream_client.add_privileges_to_group(group_name, privileges_to_add)

        # Wait for add privileges to sync
        def check_upstream():
            client = upstream_client
            try:
                groups = client.list_privilege_groups()
                print(f"DEBUG: groups in check_add: {groups}")
                assert group_name in [group["privilege_group"] for group in groups]
                group_info = None
                for group in groups:
                    if group["privilege_group"] == group_name:
                        group_info = group
                        break
                assert group_info is not None
                assert "Search" in group_info["privileges"]
                assert "Query" in group_info["privileges"]
                return True

            except Exception as e:
                logger.debug(f"Error checking privilege groups: {e}")
                return False

        assert self.wait_for_sync(
            check_upstream, sync_timeout, f"add privileges to group {group_name}"
        )

        # Wait for add privileges to sync
        def check_downstream():
            client = upstream_client
            try:
                groups = client.list_privilege_groups()
                print(f"DEBUG: groups in check_add: {groups}")
                assert group_name in [group["privilege_group"] for group in groups]
                group_info = None
                for group in groups:
                    if group["privilege_group"] == group_name:
                        group_info = group
                        break
                assert group_info is not None
                assert "Search" in group_info["privileges"]
                assert "Query" in group_info["privileges"]
                return True

            except Exception as e:
                logger.debug(f"Error checking privilege groups: {e}")
                return False

        assert self.wait_for_sync(
            check_downstream, sync_timeout, f"add privileges to group {group_name}"
        )

        # Cleanup
        try:
            upstream_client.drop_privilege_group(group_name)
        except:
            pass

    def test_remove_privileges_from_group(
        self, upstream_client, downstream_client, sync_timeout
    ):
        """Test REMOVE_PRIVILEGES_FROM_GROUP operation sync."""
        # Store upstream client for teardown
        self._upstream_client = upstream_client

        group_name = self.gen_unique_name("test_priv_group_add")

        # Initial cleanup
        try:
            upstream_client.drop_privilege_group(group_name)
        except:
            pass

        # Create privilege group first
        upstream_client.create_privilege_group(group_name)

        # Wait for creation to sync
        def check_create():
            try:
                downstream_groups = downstream_client.list_privilege_groups()
                print(f"DEBUG: downstream_groups in check_create: {downstream_groups}")
                return group_name in [
                    group["privilege_group"] for group in downstream_groups
                ]
            except Exception as e:
                logger.debug(f"Error checking privilege groups: {e}")
                return False

        assert self.wait_for_sync(
            check_create, sync_timeout, f"create privilege group {group_name}"
        )

        # Add privileges to group in upstream
        privileges_to_add = ["Search", "Query"]
        upstream_client.add_privileges_to_group(group_name, privileges_to_add)

        # Wait for add privileges to sync
        def check_upstream():
            client = upstream_client
            try:
                groups = client.list_privilege_groups()
                print(f"DEBUG: groups in check_add: {groups}")
                assert group_name in [group["privilege_group"] for group in groups]
                group_info = None
                for group in groups:
                    if group["privilege_group"] == group_name:
                        group_info = group
                        break
                assert group_info is not None
                assert "Search" in group_info["privileges"]
                assert "Query" in group_info["privileges"]
                return True

            except Exception as e:
                logger.debug(f"Error checking privilege groups: {e}")
                return False

        assert self.wait_for_sync(
            check_upstream, sync_timeout, f"add privileges to group {group_name}"
        )

        # Wait for add privileges to sync
        def check_downstream():
            client = upstream_client
            try:
                groups = client.list_privilege_groups()
                print(f"DEBUG: groups in check_add: {groups}")
                assert group_name in [group["privilege_group"] for group in groups]
                group_info = None
                for group in groups:
                    if group["privilege_group"] == group_name:
                        group_info = group
                        break
                assert group_info is not None
                assert "Search" in group_info["privileges"]
                assert "Query" in group_info["privileges"]
                return True

            except Exception as e:
                logger.debug(f"Error checking privilege groups: {e}")
                return False

        assert self.wait_for_sync(
            check_downstream, sync_timeout, f"add privileges to group {group_name}"
        )

        # remove privileges from group in upstream
        privileges_to_remove = ["Search", "Query"]
        upstream_client.remove_privileges_from_group(group_name, privileges_to_remove)

        # Wait for remove privileges to sync
        def check_upstream():
            client = upstream_client
            try:
                groups = client.list_privilege_groups()
                print(f"DEBUG: groups in check_remove: {groups}")
                assert group_name in [group["privilege_group"] for group in groups]
                group_info = None
                for group in groups:
                    if group["privilege_group"] == group_name:
                        group_info = group
                        break
                assert group_info is not None
                assert "Search" not in group_info["privileges"]
                assert "Query" not in group_info["privileges"]
                return True

            except Exception as e:
                logger.debug(f"Error checking privilege groups: {e}")
                return False

        assert self.wait_for_sync(
            check_upstream, sync_timeout, f"remove privileges from group {group_name}"
        )

        # Wait for remove privileges to sync
        def check_downstream():
            client = downstream_client
            try:
                groups = client.list_privilege_groups()
                print(f"DEBUG: groups in check_remove: {groups}")
                assert group_name in [group["privilege_group"] for group in groups]
                group_info = None
                for group in groups:
                    if group["privilege_group"] == group_name:
                        group_info = group
                        break
                assert group_info is not None
                assert "Search" not in group_info["privileges"]
                assert "Query" not in group_info["privileges"]
                return True

            except Exception as e:
                logger.debug(f"Error checking privilege groups: {e}")
                return False

        assert self.wait_for_sync(
            check_downstream, sync_timeout, f"remove privileges from group {group_name}"
        )