test: add list subscriptions and list subscriptions for model tests (#1346)

* test: add list subscriptions and list subscriptions for model tests

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

* address review comments

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

* fix:address coderabit comment

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

---------

Signed-off-by: Swati Mukund Bagal <sbagal@redhat.com>
Co-authored-by: Debarati Basu-Nag <dbasunag@redhat.com>

Author: SB159

tests/model_serving/maas_billing/conftest.py

@@ -555,6 +555,12 @@ def maas_headers_for_actor(maas_token_for_actor: str) -> dict:
     return build_maas_headers(token=maas_token_for_actor)
 
 
+@pytest.fixture(scope="class")
+def ocp_headers_for_actor(ocp_token_for_actor: str) -> dict[str, str]:
+    """Headers built from the OCP token for the current actor (admin/free/premium)."""
+    return build_maas_headers(token=ocp_token_for_actor)
+
+
 @pytest.fixture(scope="class")
 def maas_models_response_for_actor(
     request_session_http: requests.Session,

tests/model_serving/maas_billing/maas_subscription/conftest.py

@@ -135,6 +135,12 @@ def maas_subscription_tinyllama_premium(
         yield maas_subscription_premium
 
 
+@pytest.fixture(scope="class")
+def models_url(base_url: str) -> str:
+    """GET /v1/models endpoint URL."""
+    return f"{base_url}/v1/models"
+
+
 @pytest.fixture(scope="class")
 def model_url_tinyllama_free(
     maas_scheme: str,

tests/model_serving/maas_billing/maas_subscription/test_list_subscriptions.py

@@ -0,0 +1,113 @@
+from __future__ import annotations
+
+import pytest
+import requests
+import structlog
+from ocp_resources.maas_subscription import MaaSSubscription
+
+from tests.model_serving.maas_billing.maas_subscription.utils import assert_subscription_info_schema
+
+LOGGER = structlog.get_logger(name=__name__)
+
+
+@pytest.mark.usefixtures(
+    "maas_unprivileged_model_namespace",
+    "maas_subscription_controller_enabled_latest",
+    "maas_gateway_api",
+    "maas_api_gateway_reachable",
+    "maas_model_tinyllama_free",
+    "maas_auth_policy_tinyllama_free",
+    "maas_subscription_tinyllama_free",
+)
+class TestListSubscriptions:
+    """Verify a user can list their subscriptions."""
+
+    @pytest.mark.tier1
+    @pytest.mark.parametrize("ocp_token_for_actor", [{"type": "free"}], indirect=True)
+    def test_authenticated_user_sees_accessible_subscriptions(
+        self,
+        request_session_http: requests.Session,
+        base_url: str,
+        ocp_headers_for_actor: dict[str, str],
+        maas_subscription_tinyllama_free: MaaSSubscription,
+    ) -> None:
+        """Verify authenticated user gets their accessible subscriptions."""
+        response = request_session_http.get(
+            url=f"{base_url}/v1/subscriptions", headers=ocp_headers_for_actor, timeout=30
+        )
+
+        assert response.status_code == 200, f"Expected 200, got {response.status_code}: {(response.text or '')[:200]}"
+
+        subscriptions = response.json()
+        assert isinstance(subscriptions, list), f"Expected array response, got {type(subscriptions).__name__}"
+        assert len(subscriptions) >= 1, f"Expected at least 1 subscription, got {len(subscriptions)}"
+
+        for subscription in subscriptions:
+            assert_subscription_info_schema(subscription=subscription)
+
+        subscription_ids = [subscription["subscription_id_header"] for subscription in subscriptions]
+        assert maas_subscription_tinyllama_free.name in subscription_ids, (
+            f"Expected '{maas_subscription_tinyllama_free.name}' in accessible subscriptions, got {subscription_ids}"
+        )
+        LOGGER.info(
+            f"[subscriptions] GET /v1/subscriptions -> {len(subscriptions)} subscription(s): {subscription_ids}"
+        )
+
+    @pytest.mark.tier1
+    def test_unauthenticated_returns_401(
+        self,
+        request_session_http: requests.Session,
+        base_url: str,
+    ) -> None:
+        """Verify request without auth header returns 401."""
+        response = request_session_http.get(url=f"{base_url}/v1/subscriptions", timeout=30)
+
+        assert response.status_code == 401, f"Expected 401, got {response.status_code}: {(response.text or '')[:200]}"
+        LOGGER.info(f"[subscriptions] GET /v1/subscriptions (no auth) -> {response.status_code}")
+
+    @pytest.mark.tier1
+    @pytest.mark.parametrize("ocp_token_for_actor", [{"type": "free"}], indirect=True)
+    def test_subscription_response_includes_model_refs_with_rate_limits(
+        self,
+        request_session_http: requests.Session,
+        base_url: str,
+        ocp_headers_for_actor: dict[str, str],
+        maas_subscription_tinyllama_free: MaaSSubscription,
+    ) -> None:
+        """Verify subscription response includes model_refs with rate limit info."""
+        response = request_session_http.get(
+            url=f"{base_url}/v1/subscriptions", headers=ocp_headers_for_actor, timeout=30
+        )
+
+        assert response.status_code == 200, f"Expected 200, got {response.status_code}: {(response.text or '')[:200]}"
+
+        subscriptions = response.json()
+        assert isinstance(subscriptions, list), f"Expected array response, got {type(subscriptions).__name__}"
+
+        target_subscription = next(
+            (
+                subscription
+                for subscription in subscriptions
+                if subscription["subscription_id_header"] == maas_subscription_tinyllama_free.name
+            ),
+            None,
+        )
+        assert target_subscription is not None, (
+            f"Subscription '{maas_subscription_tinyllama_free.name}' not found in "
+            f"{[subscription['subscription_id_header'] for subscription in subscriptions]}"
+        )
+
+        assert len(target_subscription["model_refs"]) >= 1, "Expected at least 1 model_ref"
+        model_ref = target_subscription["model_refs"][0]
+        assert isinstance(model_ref["name"], str), "model_ref name must be a string"
+
+        assert "token_rate_limits" in model_ref, "model_ref missing 'token_rate_limits'"
+        assert len(model_ref["token_rate_limits"]) >= 1, "Expected at least 1 token_rate_limit"
+        rate_limit = model_ref["token_rate_limits"][0]
+        assert "limit" in rate_limit, "token_rate_limit missing 'limit'"
+        assert "window" in rate_limit, "token_rate_limit missing 'window'"
+
+        LOGGER.info(
+            f"[subscriptions] Subscription '{target_subscription['subscription_id_header']}' "
+            f"model_refs: {target_subscription['model_refs']}"
+        )

tests/model_serving/maas_billing/maas_subscription/test_list_subscriptions_for_model.py

@@ -0,0 +1,108 @@
+from __future__ import annotations
+
+import pytest
+import requests
+import structlog
+from ocp_resources.maas_model_ref import MaaSModelRef
+from ocp_resources.maas_subscription import MaaSSubscription
+
+from tests.model_serving.maas_billing.maas_subscription.utils import assert_subscription_info_schema
+
+LOGGER = structlog.get_logger(name=__name__)
+
+
+@pytest.mark.usefixtures(
+    "maas_unprivileged_model_namespace",
+    "maas_subscription_controller_enabled_latest",
+    "maas_gateway_api",
+    "maas_api_gateway_reachable",
+    "maas_model_tinyllama_free",
+    "maas_auth_policy_tinyllama_free",
+    "maas_subscription_tinyllama_free",
+    "maas_model_tinyllama_premium",
+    "maas_auth_policy_tinyllama_premium",
+    "maas_subscription_tinyllama_premium",
+)
+class TestListSubscriptionsForModel:
+    """Verify only subscriptions belonging to a given model are returned."""
+
+    @pytest.mark.tier1
+    @pytest.mark.parametrize("ocp_token_for_actor", [{"type": "free"}], indirect=True)
+    def test_returns_only_subscriptions_for_requested_model(
+        self,
+        request_session_http: requests.Session,
+        base_url: str,
+        ocp_headers_for_actor: dict[str, str],
+        maas_model_tinyllama_free: MaaSModelRef,
+        maas_subscription_tinyllama_free: MaaSSubscription,
+        maas_subscription_tinyllama_premium: MaaSSubscription,
+    ) -> None:
+        """Verify endpoint returns only subscriptions referencing the requested model."""
+        model_name = maas_model_tinyllama_free.name
+
+        response = request_session_http.get(
+            url=f"{base_url}/v1/model/{model_name}/subscriptions", headers=ocp_headers_for_actor, timeout=30
+        )
+
+        assert response.status_code == 200, f"Expected 200, got {response.status_code}: {(response.text or '')[:200]}"
+
+        subscriptions = response.json()
+        assert isinstance(subscriptions, list), f"Expected array response, got {type(subscriptions).__name__}"
+
+        subscription_ids = [subscription["subscription_id_header"] for subscription in subscriptions]
+
+        assert maas_subscription_tinyllama_free.name in subscription_ids, (
+            f"Expected '{maas_subscription_tinyllama_free.name}' in results for "
+            f"model '{model_name}', got {subscription_ids}"
+        )
+        assert maas_subscription_tinyllama_premium.name not in subscription_ids, (
+            f"'{maas_subscription_tinyllama_premium.name}' should not appear in results "
+            f"for model '{model_name}', got {subscription_ids}"
+        )
+
+        for subscription in subscriptions:
+            assert_subscription_info_schema(subscription=subscription)
+
+        LOGGER.info(
+            f"[subscriptions] GET /v1/model/{model_name}/subscriptions "
+            f"-> {len(subscriptions)} subscription(s): {subscription_ids}"
+        )
+
+    @pytest.mark.tier1
+    @pytest.mark.parametrize("ocp_token_for_actor", [{"type": "free"}], indirect=True)
+    def test_unknown_model_returns_empty_list(
+        self,
+        request_session_http: requests.Session,
+        base_url: str,
+        ocp_headers_for_actor: dict[str, str],
+    ) -> None:
+        """Verify GET /v1/model/{unknown-model}/subscriptions returns 200 with an empty list."""
+        unknown_model = "nonexistent-model-xyz"
+
+        response = request_session_http.get(
+            url=f"{base_url}/v1/model/{unknown_model}/subscriptions", headers=ocp_headers_for_actor, timeout=30
+        )
+
+        assert response.status_code == 200, f"Expected 200, got {response.status_code}: {(response.text or '')[:200]}"
+
+        subscriptions = response.json()
+        assert isinstance(subscriptions, list), f"Expected array response, got {type(subscriptions).__name__}"
+        assert len(subscriptions) == 0, (
+            f"Expected empty list for unknown model, got {len(subscriptions)}: {subscriptions}"
+        )
+        LOGGER.info(f"[subscriptions] GET /v1/model/{unknown_model}/subscriptions -> [] (empty list)")
+
+    @pytest.mark.tier1
+    def test_unauthenticated_returns_401(
+        self,
+        request_session_http: requests.Session,
+        base_url: str,
+        maas_model_tinyllama_free: MaaSModelRef,
+    ) -> None:
+        """Verify request without auth header returns 401."""
+        model_name = maas_model_tinyllama_free.name
+
+        response = request_session_http.get(url=f"{base_url}/v1/model/{model_name}/subscriptions", timeout=30)
+
+        assert response.status_code == 401, f"Expected 401, got {response.status_code}: {(response.text or '')[:200]}"
+        LOGGER.info(f"[subscriptions] GET /v1/model/{model_name}/subscriptions (no auth) -> {response.status_code}")

tests/model_serving/maas_billing/maas_subscription/utils.py

@@ -433,3 +433,26 @@ def wait_for_postgres_connection_log(
             return
 
     raise TimeoutError(f"PostgreSQL pod in namespace {namespace} did not report accepting connections")
+
+
+def assert_subscription_info_schema(subscription: dict[str, Any]) -> None:
+    """Assert a SubscriptionInfo object has the expected structure and field types."""
+    assert "subscription_id_header" in subscription, f"Missing subscription_id_header: {subscription}"
+    assert isinstance(subscription["subscription_id_header"], str), "subscription_id_header must be string"
+    assert "subscription_description" in subscription, f"Missing subscription_description: {subscription}"
+    assert isinstance(subscription["subscription_description"], str), "subscription_description must be string"
+    assert "priority" in subscription, f"Missing priority: {subscription}"
+    assert isinstance(subscription["priority"], int), "priority must be integer"
+    assert "model_refs" in subscription, f"Missing model_refs: {subscription}"
+    assert isinstance(subscription["model_refs"], list), "model_refs must be a list"
+    for model_ref in subscription["model_refs"]:
+        assert "name" in model_ref, f"model_ref missing name: {model_ref}"
+        assert isinstance(model_ref["name"], str), "model_ref name must be string"
+    if "display_name" in subscription:
+        assert isinstance(subscription["display_name"], str), "display_name must be string"
+    if "organization_id" in subscription:
+        assert isinstance(subscription["organization_id"], str), "organization_id must be string"
+    if "cost_center" in subscription:
+        assert isinstance(subscription["cost_center"], str), "cost_center must be string"
+    if "labels" in subscription:
+        assert isinstance(subscription["labels"], dict), "labels must be a dict"