tests(maas-billing): add token mint + utils and fixtures (#792)

* tests(maas-billing): add token mint + utils and fixtures

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

* tests(maas-billing): address review comments

* maas-billing: updated review comments

* maas-billing: Updated utils as per review comments

* maas-billing: address review comments

* maas-billing: Using existing llmd-utils-LLMInferenceService URL

* maas-billing:Updated as per review comments

* maas-billing: Updated conftest

* maas-billing: Updated conftest

---------

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

Author: 3 people

tests/model_serving/model_server/maas_billing/conftest.py

@@ -0,0 +1,46 @@
+from typing import Generator
+
+import pytest
+import requests
+from simple_logger.logger import get_logger
+
+from tests.model_serving.model_server.maas_billing.utils import (
+    detect_scheme_via_llmisvc,
+    host_from_ingress_domain,
+    mint_token,
+)
+
+LOGGER = get_logger(name=__name__)
+
+
+@pytest.fixture(scope="session")
+def request_session_http() -> Generator[requests.Session, None, None]:
+    session = requests.Session()
+    session.headers.update({"User-Agent": "odh-maas-billing-tests/1"})
+    session.verify = False
+    yield session
+    session.close()
+
+
+@pytest.fixture(scope="class")
+def minted_token(request_session_http, base_url: str, current_client_token: str) -> str:
+    """Mint a MaaS token once per test class and reuse it."""
+    resp, body = mint_token(
+        base_url=base_url,
+        oc_user_token=current_client_token,
+        minutes=30,
+        http_session=request_session_http,
+    )
+    LOGGER.info("Mint token response status=%s", resp.status_code)
+    assert resp.status_code in (200, 201), f"mint failed: {resp.status_code} {resp.text[:200]}"
+    token = body.get("token", "")
+    assert isinstance(token, str) and len(token) > 10, f"no usable token in response: {body}"
+    LOGGER.info(f"Minted MaaS token len={len(token)}")
+    return token
+
+
+@pytest.fixture(scope="module")
+def base_url(admin_client) -> str:
+    scheme = detect_scheme_via_llmisvc(client=admin_client, namespace="llm")
+    host = host_from_ingress_domain(client=admin_client)
+    return f"{scheme}://{host}/maas-api"

tests/model_serving/model_server/maas_billing/test_tokens.py

@@ -0,0 +1,18 @@
+import json
+
+from tests.model_serving.model_server.maas_billing.utils import b64url_decode
+
+
+class TestMintedToken:
+    def test_minted_token_generated(self, minted_token: str) -> None:
+        """Smoke: a MaaS token can be minted."""
+        assert isinstance(minted_token, str) and len(minted_token) > 10, "no usable token minted"
+
+    def test_minted_token_is_jwt(self, minted_token: str) -> None:
+        """Minted token looks like a JWT and has a JSON header."""
+        parts = minted_token.split(".")
+        assert len(parts) == 3, "not a JWT (expected header.payload.signature)"
+
+        header_json = b64url_decode(parts[0]).decode("utf-8")
+        header = json.loads(header_json)
+        assert isinstance(header, dict), "JWT header not a JSON object"

tests/model_serving/model_server/maas_billing/utils.py

@@ -0,0 +1,67 @@
+from typing import Dict
+
+import base64
+import requests
+from json import JSONDecodeError
+from ocp_resources.ingress_config_openshift_io import Ingress as IngressConfig
+from requests import Response
+from urllib.parse import urlparse
+from ocp_resources.llm_inference_service import LLMInferenceService
+from utilities.llmd_utils import get_llm_inference_url
+
+
+def host_from_ingress_domain(client) -> str:
+    """Return 'maas.<ingress-domain>'"""
+    ingress_config = IngressConfig(name="cluster", client=client, ensure_exists=True)
+    domain = ingress_config.instance.spec.get("domain")
+    assert domain, "Ingress 'cluster' missing spec.domain (ingresses.config.openshift.io)"
+    return f"maas.{domain}"
+
+
+def detect_scheme_via_llmisvc(client, namespace: str = "llm") -> str:
+    """
+    Using LLMInferenceService's URL to infer the scheme.
+    """
+    for inference_service in LLMInferenceService.get(dyn_client=client, namespace=namespace):
+        status_conditions = inference_service.instance.status.get("conditions", [])
+        service_is_ready = any(
+            condition_entry.get("type") == "Ready" and condition_entry.get("status") == "True"
+            for condition_entry in status_conditions
+        )
+        if service_is_ready:
+            url = get_llm_inference_url(llm_service=inference_service)
+            scheme = (urlparse(url).scheme or "").lower()
+            if scheme in ("http", "https"):
+                return scheme
+    return "http"
+
+
+def maas_auth_headers(token: str) -> Dict[str, str]:
+    """Build Authorization header for MaaS/Billing calls."""
+    return {"Authorization": f"Bearer {token}"}
+
+
+def mint_token(
+    base_url: str,
+    oc_user_token: str,
+    http_session: requests.Session,
+    minutes: int = 10,
+) -> tuple[Response, dict]:
+    """Mint a MaaS token."""
+    resp = http_session.post(
+        f"{base_url}/v1/tokens",
+        headers=maas_auth_headers(token=oc_user_token),
+        json={"ttl": f"{minutes}m"},
+        timeout=60,
+    )
+    try:
+        body = resp.json()
+    except (JSONDecodeError, ValueError):
+        body = {}
+    return resp, body
+
+
+def b64url_decode(encoded_str: str) -> bytes:
+    padding = "=" * (-len(encoded_str) % 4)
+    padded_bytes = (encoded_str + padding).encode(encoding="utf-8")
+    return base64.urlsafe_b64decode(s=padded_bytes)