test(maas): update subscription tests with new API key changes

tests/model_serving/model_server/maas_billing/maas_subscription/conftest.py

@@ -1,7 +1,5 @@
-from collections.abc import Generator
-from typing import Any
-
 import pytest
+import requests
 from kubernetes.dynamic import DynamicClient
 from ocp_resources.llm_inference_service import LLMInferenceService
 from ocp_resources.namespace import Namespace
@@ -10,9 +8,11 @@
 from simple_logger.logger import get_logger
 
 from tests.model_serving.model_server.maas_billing.maas_subscription.utils import (
+    create_api_key,
     patch_llmisvc_with_maas_router_and_tiers,
 )
 from tests.model_serving.model_server.maas_billing.utils import build_maas_headers
+from utilities.general import generate_random_name
 from utilities.infra import create_inference_token, login_with_user_password
 from utilities.llmd_constants import ContainerImages, ModelStorage
 from utilities.llmd_utils import create_llmisvc
@@ -255,6 +255,58 @@ def model_url_tinyllama_premium(
     return url
 
 
+@pytest.fixture(scope="class")
+def maas_api_key_for_actor(
+    request_session_http: requests.Session,
+    base_url: str,
+    ocp_token_for_actor: str,
+    maas_controller_enabled_latest: None,
+    maas_gateway_api: None,
+    maas_api_gateway_reachable: None,
+) -> str:
+    """
+    Create an API key for the current actor (admin/free/premium).
+
+    Flow:
+    - Use OpenShift token (ocp_token_for_actor) to create an API key via MaaS API.
+    - Use the plaintext API key for gateway inference: Authorization: Bearer <sk-...>.
+    """
+    api_key_name = f"odh-sub-tests-{generate_random_name()}"
+
+    response, body = create_api_key(
+        base_url=base_url,
+        ocp_user_token=ocp_token_for_actor,
+        request_session_http=request_session_http,
+        api_key_name=api_key_name,
+        request_timeout_seconds=60,
+    )
+
+    LOGGER.info(
+        f"MaaS subscription: create api-key name={api_key_name} "
+        f"status={response.status_code} body={(response.text or '')[:200]}"
+    )
+
+    assert response.status_code in (200, 201), (
+        f"api-key create failed: status={response.status_code} body={(response.text or '')[:200]}"
+    )
+
+    api_key = body.get("key", "")
+    assert isinstance(api_key, str) and api_key.startswith("sk-"), (
+        f"No plaintext api key returned in body['key']; body={body}"
+    )
+
+    LOGGER.info(f"MaaS subscription: created api-key name={api_key_name} key_prefix={api_key[:8]}")
+    return api_key
+
+
+@pytest.fixture(scope="class")
+def maas_headers_for_actor_api_key(maas_api_key_for_actor: str) -> dict[str, str]:
+    """
+    Headers for gateway inference using API key (new implementation).
+    """
+    return build_maas_headers(token=maas_api_key_for_actor)
+
+
 @pytest.fixture(scope="class")
 def maas_wrong_group_service_account_token(
     maas_api_server_url: str,

tests/model_serving/model_server/maas_billing/maas_subscription/test_multiple_subscriptions_per_model.py

@@ -35,7 +35,7 @@ class TestMultipleSubscriptionsPerModel:
     Validates behavior when multiple subscriptions exist for the same model.
     """
 
-    @pytest.mark.sanity
+    @pytest.mark.smoke
     @pytest.mark.parametrize("ocp_token_for_actor", [{"type": "free"}], indirect=True)
     def test_user_in_one_of_two_subscriptions_can_access_model(
         self,
@@ -44,8 +44,8 @@ def test_user_in_one_of_two_subscriptions_can_access_model(
         maas_free_group: str,
         maas_model_tinyllama_free,
         model_url_tinyllama_free: str,
-        ocp_token_for_actor: str,
         maas_subscription_tinyllama_free,
+        maas_headers_for_actor_api_key: dict[str, str],
     ) -> None:
         """
         Create a second subscription for a different group the user is NOT in.
@@ -66,7 +66,7 @@ def test_user_in_one_of_two_subscriptions_can_access_model(
         ) as extra_subscription:
             extra_subscription.wait_for_condition(condition="Ready", status="True", timeout=300)
 
-            headers = build_maas_headers(token=ocp_token_for_actor)
+            headers = dict(maas_headers_for_actor_api_key)
             payload = chat_payload_for_url(model_url=model_url_tinyllama_free)
 
             explicit_headers = dict(headers)
@@ -90,7 +90,7 @@ def test_user_in_one_of_two_subscriptions_can_access_model(
                 f"{(response.text or '')[:200]}"
             )
 
-    @pytest.mark.sanity
+    @pytest.mark.smoke
     @pytest.mark.parametrize("ocp_token_for_actor", [{"type": "free"}], indirect=True)
     def test_high_priority_subscription_allows_access_when_explicitly_selected(
         self,
@@ -99,8 +99,9 @@ def test_high_priority_subscription_allows_access_when_explicitly_selected(
         maas_free_group: str,
         maas_model_tinyllama_free,
         model_url_tinyllama_free: str,
-        ocp_token_for_actor: str,
+        # ocp_token_for_actor: str,
         maas_subscription_tinyllama_free,
+        maas_headers_for_actor_api_key: dict[str, str],
     ) -> None:
         """
         Create a second (higher priority) subscription for the same group + model.
@@ -122,7 +123,7 @@ def test_high_priority_subscription_allows_access_when_explicitly_selected(
         ) as high_tier_subscription:
             high_tier_subscription.wait_for_condition(condition="Ready", status="True", timeout=300)
 
-            headers = build_maas_headers(token=ocp_token_for_actor)
+            headers = dict(maas_headers_for_actor_api_key)
             payload = chat_payload_for_url(model_url=model_url_tinyllama_free)
 
             explicit_headers = dict(headers)
@@ -141,7 +142,7 @@ def test_high_priority_subscription_allows_access_when_explicitly_selected(
                 f"got {response.status_code}: {(response.text or '')[:200]}"
             )
 
-    @pytest.mark.sanity
+    @pytest.mark.smoke
     def test_service_account_cannot_use_subscription_it_does_not_belong_to(
         self,
         request_session_http: requests.Session,

tests/model_serving/model_server/maas_billing/maas_subscription/utils.py

@@ -12,6 +12,7 @@
 from ocp_resources.llm_inference_service import LLMInferenceService
 from ocp_resources.resource import ResourceEditor
 from pytest_testconfig import config as py_config
+from requests import Response
 from simple_logger.logger import get_logger
 from timeout_sampler import TimeoutSampler
 
@@ -168,3 +169,48 @@ def create_maas_subscription(
         teardown=teardown,
         wait_for_resource=wait_for_resource,
     )
+
+
+def maas_auth_headers_for_ocp_token(ocp_user_token: str) -> dict[str, str]:
+    """
+    Authorization header for MaaS API calls using an OpenShift user token.
+    (Used for API key creation.)
+    """
+    return {"Authorization": f"Bearer {ocp_user_token}"}
+
+
+def create_api_key(
+    *,
+    base_url: str,
+    ocp_user_token: str,
+    request_session_http: requests.Session,
+    api_key_name: str,
+    request_timeout_seconds: int = 60,
+) -> 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-...).
+    """
+    api_keys_url = f"{base_url}/v1/api-keys"
+
+    response = request_session_http.post(
+        url=api_keys_url,
+        headers={
+            **maas_auth_headers_for_ocp_token(ocp_user_token=ocp_user_token),
+            "Content-Type": "application/json",
+        },
+        json={"name": api_key_name},
+        timeout=request_timeout_seconds,
+    )
+
+    LOGGER.info(f"create_api_key: url={api_keys_url} status={response.status_code} body={response.text}")
+
+    try:
+        parsed_body: dict[str, Any] = json.loads(response.text)
+    except json.JSONDecodeError:
+        LOGGER.error(f"Unable to parse API key response: {response.text}")
+        parsed_body = {}
+
+    return response, parsed_body