test(maas): add ephemeral API key cleanup tests

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

Author: SB159

tests/model_serving/maas_billing/maas_subscription/conftest.py

@@ -10,14 +10,17 @@
 from ocp_resources.maas_model_ref import MaaSModelRef
 from ocp_resources.maas_subscription import MaaSSubscription
 from ocp_resources.namespace import Namespace
+from ocp_resources.pod import Pod
 from ocp_resources.resource import ResourceEditor
 from ocp_resources.service_account import ServiceAccount
 from pytest_testconfig import config as py_config
 
 from tests.model_serving.maas_billing.maas_subscription.utils import (
+    assert_api_key_created_ok,
     create_and_yield_api_key_id,
     create_api_key,
     create_maas_subscription,
+    get_maas_api_labels,
     patch_llmisvc_with_maas_router_and_tiers,
     resolve_api_key_username,
     revoke_api_key,
@@ -786,3 +789,53 @@ def short_expiration_api_key_id(
         key_name_prefix="e2e-exp-short",
         expires_in="1h",
     )
+
+
+@pytest.fixture()
+def maas_api_pod_name(
+    admin_client: DynamicClient,
+) -> str:
+    """Return the name of the first running maas-api pod."""
+    applications_namespace = py_config["applications_namespace"]
+    label_selector = ",".join(f"{k}={v}" for k, v in get_maas_api_labels().items())
+    all_pods = list(
+        Pod.get(
+            client=admin_client,
+            namespace=applications_namespace,
+            label_selector=label_selector,
+        )
+    )
+    running_pods = [pod for pod in all_pods if pod.instance.status.phase == "Running"]
+    assert running_pods, f"No Running maas-api pods found in {applications_namespace}"
+    return running_pods[0].name
+
+
+@pytest.fixture()
+def ephemeral_api_key(
+    request_session_http: requests.Session,
+    base_url: str,
+    ocp_token_for_actor: str,
+) -> Generator[dict[str, Any]]:
+    """Create an ephemeral API key and revoke it on teardown."""
+    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=f"e2e-ephemeral-{generate_random_name()}",
+        expires_in="1h",
+        ephemeral=True,
+        raise_on_error=False,
+    )
+    assert_api_key_created_ok(resp=create_resp, body=create_body, required_fields=("key", "id"))
+    LOGGER.info(f"[ephemeral] Created ephemeral key: id={create_body['id']}, expiresAt={create_body.get('expiresAt')}")
+    yield create_body
+    revoke_resp, _ = 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,
+    )
+    if revoke_resp.status_code not in (200, 404):
+        raise AssertionError(
+            f"Unexpected teardown status for ephemeral key id={create_body['id']}: {revoke_resp.status_code}"
+        )

tests/model_serving/maas_billing/maas_subscription/test_api_key_ephemeral_cleanup.py

@@ -0,0 +1,199 @@
+from __future__ import annotations
+
+from typing import Any
+
+import portforward
+import pytest
+import requests
+import structlog
+from kubernetes.dynamic import DynamicClient
+from ocp_resources.cron_job import CronJob
+from ocp_resources.network_policy import NetworkPolicy
+from pytest_testconfig import config as py_config
+
+from tests.model_serving.maas_billing.utils import build_maas_headers
+
+LOGGER = structlog.get_logger(name=__name__)
+
+MAAS_CLEANUP_CRONJOB_NAME = "maas-api-key-cleanup"
+MAAS_CLEANUP_NETWORKPOLICY_NAME = "maas-api-cleanup-restrict"
+
+
+@pytest.mark.usefixtures(
+    "maas_subscription_controller_enabled_latest",
+    "maas_gateway_api",
+    "maas_api_gateway_reachable",
+)
+class TestEphemeralKeyCleanup:
+    """Tests for ephemeral API key cleanup (CronJob + internal endpoint)."""
+
+    @pytest.mark.tier1
+    def test_cronjob_exists_and_configured(self, admin_client: DynamicClient) -> None:
+        """Verify the maas-api-key-cleanup CronJob exists with expected configuration."""
+        applications_namespace = py_config["applications_namespace"]
+
+        cronjob = CronJob(
+            client=admin_client,
+            name=MAAS_CLEANUP_CRONJOB_NAME,
+            namespace=applications_namespace,
+        )
+        assert cronjob.exists, f"CronJob {MAAS_CLEANUP_CRONJOB_NAME} not found in {applications_namespace}"
+
+        spec = cronjob.instance.spec
+
+        assert spec.schedule == "*/15 * * * *", f"Expected schedule '*/15 * * * *', got '{spec.schedule}'"
+        assert spec.concurrencyPolicy == "Forbid", (
+            "CronJob should use Forbid concurrency policy to prevent overlapping runs"
+        )
+
+        containers = spec.jobTemplate.spec.template.spec.containers
+        assert len(containers) >= 1, "CronJob should have at least one container"
+        container_spec = containers[0]
+        cmd_str = " ".join(container_spec.command or [])
+        assert "/internal/v1/api-keys/cleanup" in cmd_str, (
+            f"CronJob command should target the internal cleanup endpoint, got: {cmd_str}"
+        )
+
+        sec_ctx = getattr(container_spec, "securityContext", None)
+        assert sec_ctx is not None, "Cleanup container should have securityContext configured"
+        assert sec_ctx.runAsNonRoot is True, "Cleanup container should run as non-root"
+        assert sec_ctx.readOnlyRootFilesystem is True, "Cleanup container should have read-only root filesystem"
+
+        LOGGER.info(f"[ephemeral] CronJob validated: schedule={spec.schedule}, concurrency={spec.concurrencyPolicy}")
+
+    @pytest.mark.tier1
+    def test_cleanup_networkpolicy_exists(self, admin_client: DynamicClient) -> None:
+        """Verify the cleanup NetworkPolicy restricts cleanup pod egress to maas-api only."""
+        applications_namespace = py_config["applications_namespace"]
+
+        network_policy = NetworkPolicy(
+            client=admin_client,
+            name=MAAS_CLEANUP_NETWORKPOLICY_NAME,
+            namespace=applications_namespace,
+        )
+        assert network_policy.exists, (
+            f"NetworkPolicy {MAAS_CLEANUP_NETWORKPOLICY_NAME} not found in {applications_namespace}"
+        )
+
+        spec = network_policy.instance.spec
+
+        assert spec.podSelector.matchLabels.get("app") == "maas-api-cleanup", (
+            f"NetworkPolicy should target app=maas-api-cleanup pods, got: {spec.podSelector.matchLabels}"
+        )
+        assert "Egress" in spec.policyTypes, "NetworkPolicy should control Egress traffic"
+        assert "Ingress" in spec.policyTypes, "NetworkPolicy should control Ingress traffic"
+
+        ingress_rules = getattr(spec, "ingress", None)
+        assert ingress_rules in ([], None), "Cleanup pods should have no inbound traffic allowed"
+
+        egress_rules = getattr(spec, "egress", None)
+        assert egress_rules, "NetworkPolicy should define at least one egress rule"
+
+        LOGGER.info("[ephemeral] NetworkPolicy validated: cleanup pods restricted to maas-api egress only")
+
+    @pytest.mark.tier1
+    @pytest.mark.parametrize("ocp_token_for_actor", [{"type": "free"}], indirect=True)
+    def test_create_ephemeral_key(
+        self,
+        request_session_http: requests.Session,
+        base_url: str,
+        ocp_token_for_actor: str,
+        ephemeral_api_key: dict[str, Any],
+    ) -> None:
+        """Verify ephemeral keys are visible with includeEphemeral filter but hidden by default."""
+        key_id = ephemeral_api_key["id"]
+        api_keys_endpoint = f"{base_url}/v1/api-keys"
+        auth_header = build_maas_headers(token=ocp_token_for_actor)
+
+        assert ephemeral_api_key.get("ephemeral") is True, "Key should be marked as ephemeral"
+
+        r_search = request_session_http.post(
+            url=f"{api_keys_endpoint}/search",
+            headers=auth_header,
+            json={
+                "filters": {"status": ["active"], "includeEphemeral": True},
+                "pagination": {"limit": 50, "offset": 0},
+            },
+            timeout=30,
+        )
+        assert r_search.status_code == 200, (
+            f"Expected 200 from search with includeEphemeral=True, "
+            f"got {r_search.status_code}: {(r_search.text or '')[:200]}"
+        )
+        search_body = r_search.json()
+        items: list[dict[str, Any]] = search_body.get("items") or search_body.get("data") or []
+        assert key_id in [item["id"] for item in items], (
+            f"Ephemeral key {key_id} should appear in search with includeEphemeral=True"
+        )
+
+        r_default = request_session_http.post(
+            url=f"{api_keys_endpoint}/search",
+            headers=auth_header,
+            json={
+                "filters": {"status": ["active"]},
+                "pagination": {"limit": 50, "offset": 0},
+            },
+            timeout=30,
+        )
+        assert r_default.status_code == 200, (
+            f"Expected 200 from default search, got {r_default.status_code}: {(r_default.text or '')[:200]}"
+        )
+        default_body = r_default.json()
+        default_items: list[dict[str, Any]] = default_body.get("items") or default_body.get("data") or []
+        assert key_id not in [item["id"] for item in default_items], (
+            "Ephemeral key should be excluded from default search (includeEphemeral defaults to False)"
+        )
+        LOGGER.info(f"[ephemeral] Ephemeral key {key_id} visibility verified")
+
+    @pytest.mark.tier1
+    @pytest.mark.parametrize("ocp_token_for_actor", [{"type": "free"}], indirect=True)
+    def test_trigger_cleanup_preserves_active_keys(
+        self,
+        request_session_http: requests.Session,
+        base_url: str,
+        ocp_token_for_actor: str,
+        ephemeral_api_key: dict[str, Any],
+        maas_api_pod_name: str,
+    ) -> None:
+        """Verify the cleanup endpoint does not delete active (non-expired) ephemeral keys."""
+        applications_namespace = py_config["applications_namespace"]
+        key_id = ephemeral_api_key["id"]
+        api_keys_endpoint = f"{base_url}/v1/api-keys"
+        auth_header = build_maas_headers(token=ocp_token_for_actor)
+
+        LOGGER.info(f"[ephemeral] Triggering cleanup via port-forward into pod={maas_api_pod_name}")
+
+        with portforward.forward(
+            pod_or_service=maas_api_pod_name,
+            namespace=applications_namespace,
+            from_port=8080,
+            to_port=8080,
+            waiting=20,
+        ):
+            cleanup_response = requests.post(
+                url="http://localhost:8080/internal/v1/api-keys/cleanup",
+                timeout=30,
+            )
+
+        assert cleanup_response.status_code == 200, (
+            f"Cleanup endpoint returned unexpected status: {cleanup_response.status_code}: "
+            f"{(cleanup_response.text or '')[:200]}"
+        )
+        cleanup_resp = cleanup_response.json()
+        deleted_count = cleanup_resp.get("deletedCount", -1)
+        assert deleted_count >= 0, f"Cleanup response should have non-negative deletedCount, got: {cleanup_resp}"
+        LOGGER.info(f"[ephemeral] Cleanup completed: deletedCount={deleted_count}")
+
+        r_get = request_session_http.get(
+            url=f"{api_keys_endpoint}/{key_id}",
+            headers=auth_header,
+            timeout=30,
+        )
+        assert r_get.status_code == 200, (
+            f"Active ephemeral key {key_id} should survive cleanup, got {r_get.status_code}: {(r_get.text or '')[:200]}"
+        )
+        get_body = r_get.json()
+        assert get_body.get("status") == "active", (
+            f"Key should still be active after cleanup, got: {get_body.get('status')}"
+        )
+        LOGGER.info(f"[ephemeral] Active key {key_id} survived cleanup correctly")

tests/model_serving/maas_billing/maas_subscription/utils.py

@@ -188,6 +188,7 @@ def create_api_key(
     expires_in: str | None = None,
     raise_on_error: bool = True,
     subscription: str | None = None,
+    ephemeral: bool = False,
 ) -> tuple[Response, dict[str, Any]]:
     """
     Create an API key via MaaS API and return (response, parsed_body).
@@ -203,6 +204,9 @@ def create_api_key(
         subscription: Optional MaaSSubscription name to bind at mint time.
             When provided, the key is bound to this subscription for inference.
             When None, the API auto-selects the highest-priority subscription.
+        ephemeral: When True, marks the key as short-lived/programmatic.
+            Ephemeral keys are hidden from default search results and are
+            cleaned up automatically by the cleanup CronJob after expiration.
     """
     api_keys_url = f"{base_url}/v1/api-keys"
 
@@ -211,6 +215,8 @@ def create_api_key(
         payload["expiresIn"] = expires_in
     if subscription is not None:
         payload["subscription"] = subscription
+    if ephemeral:
+        payload["ephemeral"] = True
 
     response = request_session_http.post(
         url=api_keys_url,
@@ -473,6 +479,29 @@ def assert_api_key_get_ok(resp: Response, body: dict[str, Any], key_id: str) ->
     )
 
 
+def assert_subscription_info_schema(sub: dict[str, Any]) -> None:
+    """Assert a SubscriptionInfo object has the expected structure and field types."""
+    assert "subscription_id_header" in sub, f"Missing subscription_id_header: {sub}"
+    assert isinstance(sub["subscription_id_header"], str), "subscription_id_header must be string"
+    assert "subscription_description" in sub, f"Missing subscription_description: {sub}"
+    assert isinstance(sub["subscription_description"], str), "subscription_description must be string"
+    assert "priority" in sub, f"Missing priority: {sub}"
+    assert isinstance(sub["priority"], int), "priority must be integer"
+    assert "model_refs" in sub, f"Missing model_refs: {sub}"
+    assert isinstance(sub["model_refs"], list), "model_refs must be a list"
+    for ref in sub["model_refs"]:
+        assert "name" in ref, f"model_ref missing name: {ref}"
+        assert isinstance(ref["name"], str), "model_ref name must be string"
+    if "display_name" in sub:
+        assert isinstance(sub["display_name"], str), "display_name must be string"
+    if "organization_id" in sub:
+        assert isinstance(sub["organization_id"], str), "organization_id must be string"
+    if "cost_center" in sub:
+        assert isinstance(sub["cost_center"], str), "cost_center must be string"
+    if "labels" in sub:
+        assert isinstance(sub["labels"], dict), "labels must be a dict"
+
+
 def get_maas_postgres_labels() -> dict[str, str]:
     return {
         "app": "postgres",