test: add MaaS API key tests

tests/model_serving/maas_billing/maas_subscription/conftest.py

@@ -28,6 +28,7 @@
     create_maas_subscription,
     get_maas_postgres_resources,
     patch_llmisvc_with_maas_router_and_tiers,
+    revoke_api_key,
     wait_for_postgres_connection_log,
     wait_for_postgres_deployment_ready,
 )
@@ -620,3 +621,88 @@ def free_actor_premium_subscription(
             f"on premium model {maas_model_tinyllama_premium.name}"
         )
         yield sub_for_free_actor
+
+
+@pytest.fixture(scope="function")
+def two_active_api_key_ids(
+    request_session_http: requests.Session,
+    base_url: str,
+    ocp_token_for_actor: str,
+) -> Generator[list[str], Any, Any]:
+    """
+    Create two active API keys and return their IDs for list tests.
+    """
+    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-fixture-list-{i}-{generate_random_name()}",
+        )[1]["id"]
+        for i in range(1, 3)
+    ]
+    LOGGER.info(f"two_active_api_key_ids: created keys {ids}")
+    yield ids
+    for key_id in ids:
+        LOGGER.info(f"Fixture 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,
+    base_url: str,
+    ocp_token_for_actor: str,
+) -> Generator[str, Any, Any]:
+    """
+    Create a single active API key and return its ID for revoke tests.
+    """
+    key_name = f"e2e-fixture-key-{generate_random_name()}"
+    _, body = create_api_key(
+        base_url=base_url,
+        ocp_user_token=ocp_token_for_actor,
+        request_session_http=request_session_http,
+        api_key_name=key_name,
+    )
+    LOGGER.info(f"active_api_key_id: created key id={body['id']}")
+    yield body["id"]
+    LOGGER.info(f"Fixture teardown: revoking key {body['id']}")
+    revoke_api_key(
+        request_session_http=request_session_http,
+        base_url=base_url,
+        key_id=body["id"],
+        ocp_user_token=ocp_token_for_actor,
+    )
+
+
+@pytest.fixture(scope="function")
+def revoked_api_key_id(
+    request_session_http: requests.Session,
+    base_url: str,
+    ocp_token_for_actor: str,
+    active_api_key_id: str,
+) -> str:
+    """
+    Revoke the active API key and return its ID.
+
+    Asserts the DELETE response confirms status='revoked'.
+    Used as a precondition fixture for tests that verify revoked state persists.
+    """
+    revoke_resp, revoke_body = revoke_api_key(
+        request_session_http=request_session_http,
+        base_url=base_url,
+        key_id=active_api_key_id,
+        ocp_user_token=ocp_token_for_actor,
+    )
+    assert revoke_resp.status_code == 200, (
+        f"Expected 200 on DELETE /v1/api-keys/{active_api_key_id}, "
+        f"got {revoke_resp.status_code}: {revoke_resp.text[:200]}"
+    )
+    assert revoke_body.get("status") == "revoked", f"Expected status='revoked' in DELETE response, got: {revoke_body}"
+    LOGGER.info(f"revoked_api_key_id: revoked key id={active_api_key_id}")
+    return active_api_key_id

tests/model_serving/maas_billing/maas_subscription/test_api_key_crud.py

@@ -0,0 +1,145 @@
+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 (
+    create_api_key,
+    get_api_key,
+    list_api_keys,
+)
+from utilities.general import generate_random_name
+
+LOGGER = get_logger(name=__name__)
+
+
+@pytest.mark.usefixtures(
+    "maas_subscription_controller_enabled_latest",
+    "maas_gateway_api",
+    "maas_api_gateway_reachable",
+)
+class TestAPIKeyCRUD:
+    """Tests for MaaS API key lifecycle: create, list, and revoke."""
+
+    @pytest.mark.tier1
+    @pytest.mark.parametrize("ocp_token_for_actor", [{"type": "admin"}], indirect=True)
+    def test_create_api_key(
+        self,
+        request_session_http: requests.Session,
+        base_url: str,
+        ocp_token_for_actor: str,
+    ) -> None:
+        """Verify API key creation and show-once behavior."""
+
+        key_name = f"e2e-crud-create-{generate_random_name()}"
+
+        _, body = create_api_key(
+            base_url=base_url,
+            ocp_user_token=ocp_token_for_actor,
+            request_session_http=request_session_http,
+            api_key_name=key_name,
+        )
+
+        for field in ("id", "key", "name"):
+            assert field in body, f"Expected '{field}' field in create response"
+
+        key = body["key"]
+        assert key.startswith("sk-oai-"), "Expected key to start with 'sk-oai-' prefix"
+        assert len(key) > len("sk-oai-"), "Key body after prefix must not be empty"
+
+        LOGGER.info(f"[create] Created key id={body['id']}, key_prefix=sk-oai-***")
+
+        get_resp, get_body = get_api_key(
+            request_session_http=request_session_http,
+            base_url=base_url,
+            key_id=body["id"],
+            ocp_user_token=ocp_token_for_actor,
+        )
+        assert get_resp.status_code == 200, (
+            f"Expected 200 on GET /v1/api-keys/{body['id']}, got {get_resp.status_code}: {get_resp.text[:200]}"
+        )
+        assert "key" not in get_body, "Plaintext key must not be returned by GET after creation (show-once pattern)"
+
+    @pytest.mark.tier1
+    @pytest.mark.parametrize("ocp_token_for_actor", [{"type": "admin"}], indirect=True)
+    def test_list_api_keys(
+        self,
+        request_session_http: requests.Session,
+        base_url: str,
+        ocp_token_for_actor: str,
+        two_active_api_key_ids: list[str],
+    ) -> None:
+        """Verify active API keys are listed and no plaintext key is exposed."""
+
+        list_resp, list_body = list_api_keys(
+            request_session_http=request_session_http,
+            base_url=base_url,
+            ocp_user_token=ocp_token_for_actor,
+            filters={"status": ["active"]},
+            sort={"by": "created_at", "order": "desc"},
+            pagination={"limit": 50, "offset": 0},
+        )
+        assert list_resp.status_code == 200, (
+            f"Expected 200 on POST /v1/api-keys/search, got {list_resp.status_code}: {list_resp.text[:200]}"
+        )
+
+        items: list[dict] = list_body.get("items") or list_body.get("data") or []
+        assert len(items) >= 2, f"Expected at least 2 active keys, got {len(items)}"
+
+        key_ids = [item["id"] for item in items]
+        for created_id in two_active_api_key_ids:
+            assert created_id in key_ids, f"Created key id={created_id} not found in listed keys"
+
+        for item in items:
+            assert "key" not in item, f"Plaintext key must not appear in any list item: {item}"
+
+        LOGGER.info(f"[list] Found {len(items)} active keys")
+
+    @pytest.mark.tier1
+    @pytest.mark.parametrize("ocp_token_for_actor", [{"type": "admin"}], indirect=True)
+    def test_list_api_keys_pagination(
+        self,
+        request_session_http: requests.Session,
+        base_url: str,
+        ocp_token_for_actor: str,
+    ) -> None:
+        """Verify that the search endpoint respects the pagination limit."""
+
+        page_resp, page_body = list_api_keys(
+            request_session_http=request_session_http,
+            base_url=base_url,
+            ocp_user_token=ocp_token_for_actor,
+            filters={"status": ["active"]},
+            sort={"by": "created_at", "order": "desc"},
+            pagination={"limit": 1, "offset": 0},
+        )
+        assert page_resp.status_code == 200, (
+            f"Expected 200 on paginated search, got {page_resp.status_code}: {page_resp.text[:200]}"
+        )
+        paged_items: list[dict] = page_body.get("items") or page_body.get("data") or []
+        assert len(paged_items) <= 1, f"Expected at most 1 item with limit=1, got {len(paged_items)}"
+        LOGGER.info(f"[list] Pagination limit=1 returned {len(paged_items)} item(s)")
+
+    @pytest.mark.tier1
+    @pytest.mark.parametrize("ocp_token_for_actor", [{"type": "admin"}], indirect=True)
+    def test_revoke_api_key(
+        self,
+        request_session_http: requests.Session,
+        base_url: str,
+        ocp_token_for_actor: str,
+        revoked_api_key_id: str,
+    ) -> None:
+        """Verify a revoked API key shows status='revoked' on subsequent GET."""
+
+        get_resp, get_body = get_api_key(
+            request_session_http=request_session_http,
+            base_url=base_url,
+            key_id=revoked_api_key_id,
+            ocp_user_token=ocp_token_for_actor,
+        )
+        assert get_resp.status_code == 200, (
+            f"Expected 200 on GET after revoke, got {get_resp.status_code}: {get_resp.text[:200]}"
+        )
+        assert get_body.get("status") == "revoked", f"Expected status='revoked' on GET after revoke, got: {get_body}"
+        LOGGER.info(f"[revoke] Key {revoked_api_key_id} confirmed revoked")

tests/model_serving/maas_billing/maas_subscription/utils.py

@@ -4,7 +4,7 @@
 from collections.abc import Generator, Sequence
 from contextlib import contextmanager
 from typing import Any
-from urllib.parse import urlparse
+from urllib.parse import quote, urlparse
 
 import pytest
 import requests
@@ -219,6 +219,95 @@ def create_api_key(
     return response, parsed_body
 
 
+def get_api_key(
+    request_session_http: requests.Session,
+    base_url: str,
+    key_id: str,
+    ocp_user_token: str,
+    request_timeout_seconds: int = 60,
+) -> tuple[Response, dict[str, Any]]:
+    """
+    Fetch a single API key by ID via MaaS API (GET /v1/api-keys/{id}).
+    """
+    url = f"{base_url}/v1/api-keys/{quote(key_id, safe='')}"
+    response = request_session_http.get(
+        url=url,
+        headers={"Authorization": f"Bearer {ocp_user_token}"},
+        timeout=request_timeout_seconds,
+    )
+    LOGGER.info(f"get_api_key: url={url} key_id={key_id} status={response.status_code}")
+    try:
+        parsed_body: dict[str, Any] = json.loads(response.text)
+    except json.JSONDecodeError as error:
+        raise AssertionError(
+            f"get_api_key returned non-JSON response: status={response.status_code} body={response.text[:200]}"
+        ) from error
+    return response, parsed_body
+
+
+def list_api_keys(
+    request_session_http: requests.Session,
+    base_url: str,
+    ocp_user_token: str,
+    filters: dict[str, Any] | None = None,
+    sort: dict[str, Any] | None = None,
+    pagination: dict[str, Any] | None = None,
+    request_timeout_seconds: int = 60,
+) -> tuple[Response, dict[str, Any]]:
+    """
+    Search/list API keys via MaaS API (POST /v1/api-keys/search).
+    """
+    url = f"{base_url}/v1/api-keys/search"
+    payload: dict[str, Any] = {}
+    if filters is not None:
+        payload["filters"] = filters
+    if sort is not None:
+        payload["sort"] = sort
+    if pagination is not None:
+        payload["pagination"] = pagination
+
+    response = request_session_http.post(
+        url=url,
+        headers={"Authorization": f"Bearer {ocp_user_token}"},
+        json=payload,
+        timeout=request_timeout_seconds,
+    )
+    LOGGER.info(f"list_api_keys: url={url} status={response.status_code} items_count=pending_parse")
+    try:
+        parsed_body: dict[str, Any] = json.loads(response.text)
+    except json.JSONDecodeError as error:
+        raise AssertionError(
+            f"list_api_keys returned non-JSON response: status={response.status_code} body={response.text[:200]}"
+        ) from error
+    return response, parsed_body
+
+
+def revoke_api_key(
+    request_session_http: requests.Session,
+    base_url: str,
+    key_id: str,
+    ocp_user_token: str,
+    request_timeout_seconds: int = 60,
+) -> tuple[Response, dict[str, Any]]:
+    """
+    Revoke an API key via MaaS API (DELETE /v1/api-keys/{id}).
+    """
+    url = f"{base_url}/v1/api-keys/{quote(key_id, safe='')}"
+    response = request_session_http.delete(
+        url=url,
+        headers={"Authorization": f"Bearer {ocp_user_token}"},
+        timeout=request_timeout_seconds,
+    )
+    LOGGER.info(f"revoke_api_key: url={url} key_id={key_id} status={response.status_code}")
+    try:
+        parsed_body: dict[str, Any] = json.loads(response.text)
+    except json.JSONDecodeError as error:
+        raise AssertionError(
+            f"revoke_api_key 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",