test(maas): add ephemeral API key cleanup tests (#1324)

SB159 · web-flow · commit 896819692345 · 2026-03-31T22:13:04.000Z
* test(maas): add ephemeral API key cleanup tests

Signed-off-by: Swati Mukund Bagal &lt;sbagal@redhat.com&gt;

* fix: address review comments

Signed-off-by: Swati Mukund Bagal &lt;sbagal@redhat.com&gt;

---------

Signed-off-by: Swati Mukund Bagal &lt;sbagal@redhat.com&gt;
diff --git a/tests/model_serving/maas_billing/maas_subscription/conftest.py b/tests/model_serving/maas_billing/maas_subscription/conftest.py
@@ -5,19 +5,24 @@
 import requests
 import structlog
 from kubernetes.dynamic import DynamicClient
+from ocp_resources.cron_job import CronJob
 from ocp_resources.llm_inference_service import LLMInferenceService
 from ocp_resources.maas_auth_policy import MaaSAuthPolicy
 from ocp_resources.maas_model_ref import MaaSModelRef
 from ocp_resources.maas_subscription import MaaSSubscription
 from ocp_resources.namespace import Namespace
+from ocp_resources.network_policy import NetworkPolicy
+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 +791,87 @@ def short_expiration_api_key_id(
         key_name_prefix="e2e-exp-short",
         expires_in="1h",
     )
+
+
+@pytest.fixture()
+def maas_cleanup_cronjob(
+    admin_client: DynamicClient,
+) -> CronJob:
+    """Return the maas-api-key-cleanup CronJob, asserting it exists."""
+    applications_namespace = py_config["applications_namespace"]
+    cronjob = CronJob(
+        client=admin_client,
+        name="maas-api-key-cleanup",
+        namespace=applications_namespace,
+    )
+    assert cronjob.exists, f"CronJob maas-api-key-cleanup not found in {applications_namespace}"
+    return cronjob
+
+
+@pytest.fixture()
+def maas_cleanup_networkpolicy(
+    admin_client: DynamicClient,
+) -> NetworkPolicy:
+    """Return the maas-api-cleanup-restrict NetworkPolicy, asserting it exists."""
+    applications_namespace = py_config["applications_namespace"]
+    network_policy = NetworkPolicy(
+        client=admin_client,
+        name="maas-api-cleanup-restrict",
+        namespace=applications_namespace,
+    )
+    assert network_policy.exists, f"NetworkPolicy maas-api-cleanup-restrict not found in {applications_namespace}"
+    return network_policy
+
+
+@pytest.fixture()
+def maas_api_pod_name(
+    admin_client: DynamicClient,
+) -> str:
+    """Return the name of the single running maas-api pod (exactly one pod is expected)."""
+    applications_namespace = py_config["applications_namespace"]
+    label_selector = ",".join(f"{k}={v}" for k, v in get_maas_api_labels().items())
+    pods = list(
+        Pod.get(
+            client=admin_client,
+            namespace=applications_namespace,
+            label_selector=label_selector,
+        )
+    )
+    assert len(pods) == 1, f"Expected exactly 1 maas-api pod in {applications_namespace}, found {len(pods)}"
+    assert pods[0].instance.status.phase == "Running", (
+        f"maas-api pod '{pods[0].name}' is not Running (phase={pods[0].instance.status.phase})"
+    )
+    return 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."""
+    creation_response, api_key_data = 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=creation_response, body=api_key_data, required_fields=("key", "id"))
+    LOGGER.info(
+        f"[ephemeral] Created ephemeral key: id={api_key_data['id']}, expiresAt={api_key_data.get('expiresAt')}"
+    )
+    yield api_key_data
+    revoke_response, _ = revoke_api_key(
+        request_session_http=request_session_http,
+        base_url=base_url,
+        key_id=api_key_data["id"],
+        ocp_user_token=ocp_token_for_actor,
+    )
+    if revoke_response.status_code not in (200, 404):
+        raise AssertionError(
+            f"Unexpected teardown status for ephemeral key id={api_key_data['id']}: {revoke_response.status_code}"
+        )
diff --git a/tests/model_serving/maas_billing/maas_subscription/test_api_key_ephemeral_cleanup.py b/tests/model_serving/maas_billing/maas_subscription/test_api_key_ephemeral_cleanup.py
@@ -0,0 +1,168 @@
+from typing import Any
+
+import portforward
+import pytest
+import requests
+import structlog
+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.maas_subscription.utils import search_active_api_keys
+from tests.model_serving.maas_billing.utils import build_maas_headers
+
+LOGGER = structlog.get_logger(name=__name__)
+
+
+@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, maas_cleanup_cronjob: CronJob) -> None:
+        """Verify the maas-api-key-cleanup CronJob exists with expected configuration."""
+        spec = maas_cleanup_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, maas_cleanup_networkpolicy: NetworkPolicy) -> None:
+        """Verify the cleanup NetworkPolicy restricts cleanup pod egress to maas-api only."""
+        spec = maas_cleanup_networkpolicy.instance.spec
+
+        assert spec.podSelector.matchLabels.get("app") == "maas-api-cleanup", (
+            f"NetworkPolicy should target app=maas-api-cleanup pods, got: {spec.podSelector.matchLabels}"
+        )
+        for policy_type in ("Egress", "Ingress"):
+            assert policy_type in spec.policyTypes, f"NetworkPolicy should control {policy_type} 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_ephemeral_key_visible_with_include_filter(
+        self,
+        request_session_http: requests.Session,
+        base_url: str,
+        ocp_token_for_actor: str,
+        ephemeral_api_key: dict[str, Any],
+    ) -> None:
+        """Verify ephemeral key is marked as ephemeral and visible when includeEphemeral=True."""
+        key_id = ephemeral_api_key["id"]
+
+        assert ephemeral_api_key.get("ephemeral") is True, "Key should be marked as ephemeral"
+
+        items = search_active_api_keys(
+            request_session_http=request_session_http,
+            base_url=base_url,
+            ocp_user_token=ocp_token_for_actor,
+            include_ephemeral=True,
+        )
+        assert key_id in [item["id"] for item in items], (
+            f"Ephemeral key {key_id} should appear in search with includeEphemeral=True"
+        )
+        LOGGER.info(f"[ephemeral] Ephemeral key {key_id} visible with includeEphemeral=True")
+
+    @pytest.mark.tier1
+    @pytest.mark.parametrize("ocp_token_for_actor", [{"type": "free"}], indirect=True)
+    def test_ephemeral_key_hidden_from_default_search(
+        self,
+        request_session_http: requests.Session,
+        base_url: str,
+        ocp_token_for_actor: str,
+        ephemeral_api_key: dict[str, Any],
+    ) -> None:
+        """Verify ephemeral key is hidden from default search when includeEphemeral is not set."""
+        key_id = ephemeral_api_key["id"]
+
+        default_items = search_active_api_keys(
+            request_session_http=request_session_http,
+            base_url=base_url,
+            ocp_user_token=ocp_token_for_actor,
+            include_ephemeral=False,
+        )
+        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} correctly hidden from default search")
+
+    @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")
diff --git a/tests/model_serving/maas_billing/maas_subscription/utils.py b/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,
@@ -466,6 +472,29 @@ def assert_api_key_created_ok(
         assert field in body, f"Response must contain '{field}'"
 
 
+def search_active_api_keys(
+    request_session_http: requests.Session,
+    base_url: str,
+    ocp_user_token: str,
+    include_ephemeral: bool = False,
+    request_timeout_seconds: int = 30,
+) -> list[dict[str, Any]]:
+    """POST /v1/api-keys/search for active keys and return the list of matching items."""
+    filters: dict[str, Any] = {"status": ["active"]}
+    if include_ephemeral:
+        filters["includeEphemeral"] = True
+    url = f"{base_url}/v1/api-keys/search"
+    resp = request_session_http.post(
+        url=url,
+        headers={"Authorization": f"Bearer {ocp_user_token}"},
+        json={"filters": filters, "pagination": {"limit": 50, "offset": 0}},
+        timeout=request_timeout_seconds,
+    )
+    assert resp.status_code == 200, f"Expected 200 from key search, got {resp.status_code}: {(resp.text or '')[:200]}"
+    body = resp.json()
+    return body.get("items") or body.get("data") or []
+
+
 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, (
