test: add MaaS API key bulk revoke tests

Signed-off-by: Swati Mukund Bagal <sbagal@redhat.com>

Author: SB159

tests/model_serving/maas_billing/maas_subscription/conftest.py

@@ -657,6 +657,34 @@ def two_active_api_key_ids(
         )
 
 
+@pytest.fixture(scope="function")
+def three_active_api_key_ids(
+    request_session_http: requests.Session,
+    base_url: str,
+    ocp_token_for_actor: str,
+) -> Generator[list[str], Any, Any]:
+    """Create three active API keys and yield their IDs for bulk-revoke tests."""
+    key_ids = [
+        create_api_key(
+            base_url=base_url,
+            ocp_user_token=ocp_token_for_actor,
+            request_session_http=request_session_http,
+            api_key_name=f"e2e-bulk-key-{index}-{generate_random_name()}",
+        )[1]["id"]
+        for index in range(1, 4)
+    ]
+    LOGGER.info(f"three_active_api_key_ids: created keys {key_ids}")
+    yield key_ids
+    for key_id in key_ids:
+        LOGGER.info(f"three_active_api_key_ids: teardown revoking key {key_id}")
+        revoke_api_key(
+            request_session_http=request_session_http,
+            base_url=base_url,
+            key_id=key_id,
+            ocp_user_token=ocp_token_for_actor,
+        )
+
+
 @pytest.fixture(scope="function")
 def active_api_key_id(
     request_session_http: requests.Session,

tests/model_serving/maas_billing/maas_subscription/test_api_key_bulk_operations.py

@@ -0,0 +1,115 @@
+from __future__ import annotations
+
+import pytest
+import requests
+from simple_logger.logger import get_logger
+
+from tests.model_serving.maas_billing.maas_subscription.utils import (
+    bulk_revoke_api_keys,
+    get_api_key,
+    resolve_api_key_username,
+)
+
+LOGGER = get_logger(name=__name__)
+
+
+@pytest.mark.usefixtures(
+    "maas_subscription_controller_enabled_latest",
+    "maas_gateway_api",
+    "maas_api_gateway_reachable",
+)
+class TestAPIKeyBulkOperations:
+    """Tests for MaaS API key bulk revoke operations."""
+
+    @pytest.mark.tier1
+    @pytest.mark.parametrize("ocp_token_for_actor", [{"type": "admin"}], indirect=True)
+    def test_bulk_revoke_own_keys(
+        self,
+        request_session_http: requests.Session,
+        base_url: str,
+        ocp_token_for_actor: str,
+        three_active_api_key_ids: list[str],
+    ) -> None:
+        """Verify a user can bulk revoke all their own active API keys.
+        """
+        username = resolve_api_key_username(
+            request_session_http=request_session_http,
+            base_url=base_url,
+            key_id=three_active_api_key_ids[0],
+            ocp_user_token=ocp_token_for_actor,
+        )
+
+        bulk_resp, bulk_body = bulk_revoke_api_keys(
+            request_session_http=request_session_http,
+            base_url=base_url,
+            ocp_user_token=ocp_token_for_actor,
+            username=username,
+        )
+        assert bulk_resp.status_code == 200, (
+            f"Expected 200 on bulk-revoke for user {username}, "
+            f"got {bulk_resp.status_code}: {bulk_resp.text[:200]}"
+        )
+        revoked_count = bulk_body.get("revokedCount", 0)
+        assert revoked_count >= 3, f"Expected at least 3 revoked keys, got revokedCount={revoked_count}"
+        LOGGER.info(f"[bulk-revoke] User {username} bulk revoked {revoked_count} key(s)")
+
+        for key_id in three_active_api_key_ids:
+            get_resp, get_body = get_api_key(
+                request_session_http=request_session_http,
+                base_url=base_url,
+                key_id=key_id,
+                ocp_user_token=ocp_token_for_actor,
+            )
+            if get_resp.status_code == 200:
+                assert get_body.get("status") == "revoked", (
+                    f"Expected key id={key_id} to have status='revoked', got: {get_body.get('status')}"
+                )
+        LOGGER.info(f"[bulk-revoke] All {len(three_active_api_key_ids)} key(s) confirmed revoked")
+
+    @pytest.mark.tier1
+    @pytest.mark.parametrize("ocp_token_for_actor", [{"type": "free"}], indirect=True)
+    def test_bulk_revoke_other_user_forbidden(
+        self,
+        request_session_http: requests.Session,
+        base_url: str,
+        ocp_token_for_actor: str,
+    ) -> None:
+        """Verify a non-admin user gets 403 when attempting to bulk revoke another user's keys.
+        """
+        bulk_resp, _ = bulk_revoke_api_keys(
+            request_session_http=request_session_http,
+            base_url=base_url,
+            ocp_user_token=ocp_token_for_actor,
+            username="someotheruser",
+        )
+        assert bulk_resp.status_code == 403, (
+            f"Expected 403 (non-admin cannot bulk revoke other users), "
+            f"got {bulk_resp.status_code}: {bulk_resp.text[:200]}"
+        )
+        LOGGER.info("[bulk-revoke] Non-admin correctly received 403 when attempting to bulk revoke another user's keys")
+
+    @pytest.mark.tier1
+    @pytest.mark.parametrize("ocp_token_for_actor", [{"type": "free"}], indirect=True)
+    def test_bulk_revoke_admin_can_revoke_any_user(
+        self,
+        request_session_http: requests.Session,
+        base_url: str,
+        active_api_key_id: str,
+        free_user_username: str,
+        admin_ocp_token: str,
+    ) -> None:
+        """Verify an admin can bulk revoke any user's active API keys.
+        """
+        bulk_resp, bulk_body = bulk_revoke_api_keys(
+            request_session_http=request_session_http,
+            base_url=base_url,
+            ocp_user_token=admin_ocp_token,
+            username=free_user_username,
+        )
+        assert bulk_resp.status_code == 200, (
+            f"Expected 200 on admin bulk-revoke for user {free_user_username}, "
+            f"got {bulk_resp.status_code}: {bulk_resp.text[:200]}"
+        )
+        revoked_count = bulk_body.get("revokedCount", 0)
+        assert revoked_count >= 1, f"Expected at least 1 revoked key, got revokedCount={revoked_count}"
+        LOGGER.info(f"[bulk-revoke] Admin successfully revoked {revoked_count} key(s) for user {free_user_username}")

tests/model_serving/maas_billing/maas_subscription/utils.py

@@ -376,6 +376,37 @@ def revoke_api_key(
     return response, parsed_body
 
 
+def bulk_revoke_api_keys(
+    request_session_http: requests.Session,
+    base_url: str,
+    ocp_user_token: str,
+    username: str,
+    request_timeout_seconds: int = 60,
+) -> tuple[Response, dict[str, Any]]:
+    """
+    Bulk revoke all active API keys for a given user via MaaS API (POST /v1/api-keys/bulk-revoke).
+
+    """
+    url = f"{base_url}/v1/api-keys/bulk-revoke"
+    response = request_session_http.post(
+        url=url,
+        headers={
+            "Authorization": f"Bearer {ocp_user_token}",
+            "Content-Type": "application/json",
+        },
+        json={"username": username},
+        timeout=request_timeout_seconds,
+    )
+    LOGGER.info(f"bulk_revoke_api_keys: url={url} username={username} status={response.status_code}")
+    try:
+        parsed_body: dict[str, Any] = json.loads(response.text)
+    except json.JSONDecodeError as error:
+        raise AssertionError(
+            f"bulk_revoke_api_keys returned non-JSON response: status={response.status_code} body={response.text[:200]}"
+        ) from error
+    return response, parsed_body
+
+
 def get_maas_postgres_labels() -> dict[str, str]:
     return {
         "app": "postgres",