fix: address review comments

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

Author: SB159

tests/model_serving/maas_billing/maas_subscription/conftest.py

@@ -5,11 +5,13 @@
 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
@@ -791,23 +793,55 @@ def short_expiration_api_key_id(
     )
 
 
+@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 first running maas-api pod."""
+    """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())
-    all_pods = list(
+    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
+    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()
@@ -817,7 +851,7 @@ def ephemeral_api_key(
     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(
+    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,
@@ -826,16 +860,18 @@ def ephemeral_api_key(
         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(
+    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=create_body["id"],
+        key_id=api_key_data["id"],
         ocp_user_token=ocp_token_for_actor,
     )
-    if revoke_resp.status_code not in (200, 404):
+    if revoke_response.status_code not in (200, 404):
         raise AssertionError(
-            f"Unexpected teardown status for ephemeral key id={create_body['id']}: {revoke_resp.status_code}"
+            f"Unexpected teardown status for ephemeral key id={api_key_data['id']}: {revoke_response.status_code}"
         )

tests/model_serving/maas_billing/maas_subscription/test_api_key_ephemeral_cleanup.py

@@ -1,23 +1,18 @@
-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.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__)
 
-MAAS_CLEANUP_CRONJOB_NAME = "maas-api-key-cleanup"
-MAAS_CLEANUP_NETWORKPOLICY_NAME = "maas-api-cleanup-restrict"
-
 
 @pytest.mark.usefixtures(
     "maas_subscription_controller_enabled_latest",
@@ -28,18 +23,9 @@ 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:
+    def test_cronjob_exists_and_configured(self, maas_cleanup_cronjob: CronJob) -> 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
+        spec = maas_cleanup_cronjob.instance.spec
 
         assert spec.schedule == "*/15 * * * *", f"Expected schedule '*/15 * * * *', got '{spec.schedule}'"
         assert spec.concurrencyPolicy == "Forbid", (
@@ -62,26 +48,15 @@ def test_cronjob_exists_and_configured(self, admin_client: DynamicClient) -> Non
         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:
+    def test_cleanup_networkpolicy_exists(self, maas_cleanup_networkpolicy: NetworkPolicy) -> 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
+        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}"
         )
-        assert "Egress" in spec.policyTypes, "NetworkPolicy should control Egress traffic"
-        assert "Ingress" in spec.policyTypes, "NetworkPolicy should control Ingress traffic"
+        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"
@@ -93,57 +68,51 @@ def test_cleanup_networkpolicy_exists(self, admin_client: DynamicClient) -> None
 
     @pytest.mark.tier1
     @pytest.mark.parametrize("ocp_token_for_actor", [{"type": "free"}], indirect=True)
-    def test_create_ephemeral_key(
+    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 keys are visible with includeEphemeral filter but hidden by default."""
+        """Verify ephemeral key is marked as ephemeral and visible when includeEphemeral=True."""
         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]}"
+        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,
         )
-        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"
         )
+        LOGGER.info(f"[ephemeral] Ephemeral key {key_id} visible 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]}"
+    @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,
         )
-        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")
+        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)

tests/model_serving/maas_billing/maas_subscription/utils.py

@@ -472,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, (