test: add MaaS API key expiration tests (#1292)

SB159 · pre-commit-ci[bot] · dbasunag · web-flow · commit 9ef3f9dd8d21 · 2026-03-25T13:27:05.000Z
* test: add MaaS API key expiration tests Signed-off-by: Swati Mukund Bagal <sbagal@redhat.com> * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * fix: address review comments Signed-off-by: Swati Mukund Bagal <sbagal@redhat.com> --------- Signed-off-by: Swati Mukund Bagal <sbagal@redhat.com> Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com> Co-authored-by: Debarati Basu-Nag <dbasunag@redhat.com>
diff --git a/tests/model_serving/maas_billing/maas_subscription/conftest.py b/tests/model_serving/maas_billing/maas_subscription/conftest.py
@@ -809,3 +809,19 @@ def revoked_api_key_id(
     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
+
+
+@pytest.fixture(scope="function")
+def short_expiration_api_key_id(
+    request_session_http: requests.Session,
+    base_url: str,
+    ocp_token_for_actor: str,
+) -> Generator[str, Any, Any]:
+    """Create an API key with 1-hour expiration, yield its ID, and revoke on teardown."""
+    yield from create_and_yield_api_key_id(
+        request_session_http=request_session_http,
+        base_url=base_url,
+        ocp_user_token=ocp_token_for_actor,
+        key_name_prefix="e2e-exp-short",
+        expires_in="1h",
+    )
diff --git a/tests/model_serving/maas_billing/maas_subscription/test_api_key_expiration.py b/tests/model_serving/maas_billing/maas_subscription/test_api_key_expiration.py
@@ -0,0 +1,163 @@
+from __future__ import annotations
+
+import pytest
+import requests
+
+from tests.model_serving.maas_billing.maas_subscription.utils import (
+    assert_api_key_created_ok,
+    assert_api_key_get_ok,
+    create_api_key,
+    get_api_key,
+    revoke_api_key,
+)
+from utilities.general import generate_random_name
+from utilities.opendatahub_logger import get_logger
+
+LOGGER = get_logger(name=__name__)
+
+MAAS_API_KEY_MAX_EXPIRATION_DAYS = 30
+
+
+@pytest.mark.parametrize("ocp_token_for_actor", [{"type": "admin"}], indirect=True)
+@pytest.mark.usefixtures(
+    "maas_subscription_controller_enabled_latest",
+    "maas_gateway_api",
+    "maas_api_gateway_reachable",
+)
+class TestAPIKeyExpiration:
+    """Tests for API key expiration policy enforcement."""
+
+    @pytest.mark.tier1
+    def test_create_key_within_expiration_limit(
+        self,
+        request_session_http: requests.Session,
+        base_url: str,
+        ocp_token_for_actor: str,
+    ) -> None:
+        """Verify creating an API key with expiration below the limit succeeds."""
+        expires_in_hours = max((MAAS_API_KEY_MAX_EXPIRATION_DAYS // 2) * 24, 24)
+        key_name = f"e2e-exp-within-{generate_random_name()}"
+
+        create_resp, create_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,
+            expires_in=f"{expires_in_hours}h",
+            raise_on_error=False,
+        )
+        assert_api_key_created_ok(resp=create_resp, body=create_body, required_fields=("key", "expiresAt"))
+        LOGGER.info(
+            f"[expiration] Created key within limit: expires_in={expires_in_hours}h, "
+            f"expiresAt={create_body.get('expiresAt')}"
+        )
+        revoke_api_key(
+            request_session_http=request_session_http,
+            base_url=base_url,
+            key_id=create_body["id"],
+            ocp_user_token=ocp_token_for_actor,
+        )
+
+    @pytest.mark.tier1
+    def test_create_key_at_expiration_limit(
+        self,
+        request_session_http: requests.Session,
+        base_url: str,
+        ocp_token_for_actor: str,
+    ) -> None:
+        """Verify creating an API key with expiration exactly at the limit succeeds."""
+        expires_in_hours = MAAS_API_KEY_MAX_EXPIRATION_DAYS * 24
+        key_name = f"e2e-exp-at-limit-{generate_random_name()}"
+
+        create_resp, create_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,
+            expires_in=f"{expires_in_hours}h",
+            raise_on_error=False,
+        )
+        assert_api_key_created_ok(resp=create_resp, body=create_body, required_fields=("key", "expiresAt"))
+        LOGGER.info(
+            f"[expiration] Created key at limit: expires_in={expires_in_hours}h "
+            f"({MAAS_API_KEY_MAX_EXPIRATION_DAYS} days)"
+        )
+        revoke_api_key(
+            request_session_http=request_session_http,
+            base_url=base_url,
+            key_id=create_body["id"],
+            ocp_user_token=ocp_token_for_actor,
+        )
+
+    @pytest.mark.tier1
+    def test_create_key_exceeds_expiration_limit(
+        self,
+        request_session_http: requests.Session,
+        base_url: str,
+        ocp_token_for_actor: str,
+    ) -> None:
+        """Verify creating an API key with expiration beyond the limit returns 400."""
+        exceeds_days = MAAS_API_KEY_MAX_EXPIRATION_DAYS * 2
+        key_name = f"e2e-exp-exceeds-{generate_random_name()}"
+
+        create_resp, _ = 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,
+            expires_in=f"{exceeds_days * 24}h",
+            raise_on_error=False,
+        )
+        assert create_resp.status_code == 400, (
+            f"Expected 400 for expiration exceeding limit "
+            f"({exceeds_days} days > {MAAS_API_KEY_MAX_EXPIRATION_DAYS} days limit), "
+            f"got {create_resp.status_code}: {create_resp.text[:200]}"
+        )
+        error_text = create_resp.text.lower()
+        assert "exceed" in error_text or "maximum" in error_text, (
+            f"Expected error body to mention 'exceed' or 'maximum': {create_resp.text[:200]}"
+        )
+        LOGGER.info(
+            f"[expiration] Correctly rejected key: {exceeds_days} days > {MAAS_API_KEY_MAX_EXPIRATION_DAYS} days limit"
+        )
+
+    @pytest.mark.tier1
+    def test_create_key_without_expiration(
+        self,
+        request_session_http: requests.Session,
+        base_url: str,
+        ocp_token_for_actor: str,
+        active_api_key_id: str,
+    ) -> None:
+        """Verify a key created without an expiration field has no expiresAt value."""
+        get_resp, get_body = get_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_api_key_get_ok(resp=get_resp, body=get_body, key_id=active_api_key_id)
+        expires_at = get_body.get("expiresAt")
+        assert expires_at is None, f"Expected no 'expiresAt' for key created without expiration, got: {expires_at!r}"
+        LOGGER.info(f"[expiration] Key without expiration field: expiresAt={expires_at!r}")
+
+    @pytest.mark.tier2
+    def test_create_key_with_short_expiration(
+        self,
+        request_session_http: requests.Session,
+        base_url: str,
+        ocp_token_for_actor: str,
+        short_expiration_api_key_id: str,
+    ) -> None:
+        """Verify a key created with a 1-hour expiration has a non-null expirationDate value."""
+        get_resp, get_body = get_api_key(
+            request_session_http=request_session_http,
+            base_url=base_url,
+            key_id=short_expiration_api_key_id,
+            ocp_user_token=ocp_token_for_actor,
+        )
+        assert_api_key_get_ok(resp=get_resp, body=get_body, key_id=short_expiration_api_key_id)
+        assert get_body.get("expirationDate"), (
+            f"Expected non-null 'expirationDate' for 1h key, got: {get_body.get('expirationDate')!r}"
+        )
+        LOGGER.info(f"[expiration] 1h key expirationDate={get_body['expirationDate']}")
diff --git a/tests/model_serving/maas_billing/maas_subscription/utils.py b/tests/model_serving/maas_billing/maas_subscription/utils.py
@@ -185,28 +185,42 @@ def create_api_key(
     request_session_http: requests.Session,
     api_key_name: str,
     request_timeout_seconds: int = 60,
+    expires_in: str | None = None,
+    raise_on_error: bool = True,
 ) -> tuple[Response, dict[str, Any]]:
     """
     Create an API key via MaaS API and return (response, parsed_body).
 
     Uses ocp_user_token for auth against maas-api.
     Expects plaintext key in body["key"] (sk-...).
+
+    Args:
+        expires_in: Optional expiration duration string (e.g. "24h", "720h").
+            When None, no expiresIn field is sent and the key does not expire.
+        raise_on_error: When True (default), raises AssertionError for non-200/201
+            responses. Set to False when testing error cases (e.g. 400 rejection).
     """
     api_keys_url = f"{base_url}/v1/api-keys"
 
+    payload: dict[str, Any] = {"name": api_key_name}
+    if expires_in is not None:
+        payload["expiresIn"] = expires_in
+
     response = request_session_http.post(
         url=api_keys_url,
         headers={
             "Authorization": f"Bearer {ocp_user_token}",
             "Content-Type": "application/json",
         },
-        json={"name": api_key_name},
+        json=payload,
         timeout=request_timeout_seconds,
     )
 
     LOGGER.info(f"create_api_key: url={api_keys_url} status={response.status_code}")
     if response.status_code not in (200, 201):
-        raise AssertionError(f"api-key create failed: status={response.status_code}")
+        if raise_on_error:
+            raise AssertionError(f"api-key create failed: status={response.status_code}")
+        return response, {}
 
     try:
         parsed_body: dict[str, Any] = json.loads(response.text)
@@ -326,6 +340,7 @@ def create_and_yield_api_key_id(
     base_url: str,
     ocp_user_token: str,
     key_name_prefix: str,
+    expires_in: str | None = None,
 ) -> Generator[str]:
     """Create an API key, yield its ID, and revoke it on teardown."""
     key_name = f"{key_name_prefix}-{generate_random_name()}"
@@ -334,6 +349,7 @@ def create_and_yield_api_key_id(
         ocp_user_token=ocp_user_token,
         request_session_http=request_session_http,
         api_key_name=key_name,
+        expires_in=expires_in,
     )
     LOGGER.info(f"create_and_yield_api_key_id: created key id={body['id']} name={key_name}")
     yield body["id"]
@@ -431,6 +447,26 @@ def assert_bulk_revoke_success(
     return revoked_count
 
 
+def assert_api_key_created_ok(
+    resp: Response,
+    body: dict[str, Any],
+    required_fields: tuple[str, ...] = ("key",),
+) -> None:
+    """Assert an API key creation response has a success status and expected fields."""
+    assert resp.status_code in (200, 201), (
+        f"Expected 200/201 for API key creation, got {resp.status_code}: {resp.text[:200]}"
+    )
+    for field in required_fields:
+        assert field in body, f"Response must contain '{field}'"
+
+
+def assert_api_key_get_ok(resp: Response, body: dict[str, Any], key_id: str) -> None:
+    """Assert a GET /v1/api-keys/{id} response has status 200."""
+    assert resp.status_code == 200, (
+        f"Expected 200 on GET /v1/api-keys/{key_id}, got {resp.status_code}: {resp.text[:200]}"
+    )
+
+
 def get_maas_postgres_labels() -> dict[str, str]:
     return {
         "app": "postgres",
