maas-billing: refactor RBAC fixtures

Author: SB159

tests/model_serving/model_server/maas_billing/conftest.py

@@ -4,7 +4,7 @@
 import pytest
 import requests
 from simple_logger.logger import get_logger
-from utilities.plugins.constant import RestHeader, OpenAIEnpoints
+from utilities.plugins.constant import OpenAIEnpoints
 
 from kubernetes.dynamic import DynamicClient
 from ocp_resources.infrastructure import Infrastructure
@@ -15,15 +15,14 @@
 from utilities.infra import login_with_user_password, get_openshift_token
 from utilities.general import wait_for_oauth_openshift_deployment
 from ocp_resources.secret import Secret
-from timeout_sampler import TimeoutExpiredError
-
-
 from tests.model_serving.model_server.maas_billing.utils import (
     detect_scheme_via_llmisvc,
     host_from_ingress_domain,
     mint_token,
     llmis_name,
     create_maas_group,
+    build_maas_headers,
+    get_maas_models_response,
 )
 
 
@@ -53,7 +52,7 @@ def minted_token(request_session_http, base_url: str, current_client_token: str)
         minutes=30,
         http_session=request_session_http,
     )
-    LOGGER.info("Mint token response status=%s", resp.status_code)
+    LOGGER.info(f"Mint token response status={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}"
@@ -81,22 +80,20 @@ def model_url(admin_client) -> str:
 
 @pytest.fixture
 def maas_headers(minted_token: str) -> dict:
-    """Common headers for MaaS API calls."""
-    return {"Authorization": f"Bearer {minted_token}", **RestHeader.HEADERS}
+    return build_maas_headers(token=minted_token)
 
 
 @pytest.fixture
 def maas_models(
-    request_session_http: requests.Session,
-    base_url: str,
-    maas_headers: dict,
+    request_session_http,
+    base_url,
+    maas_headers,
 ):
-    """
-    Call /v1/models once and return the list of models.
-
-    """
-    models_url = f"{base_url}{MODELS_INFO}"
-    resp = request_session_http.get(models_url, headers=maas_headers, timeout=60)
+    resp = get_maas_models_response(
+        session=request_session_http,
+        base_url=base_url,
+        headers=maas_headers,
+    )
 
     assert resp.status_code == 200, f"/v1/models failed: {resp.status_code} {resp.text[:200]}"
 
@@ -133,113 +130,129 @@ def maas_user_credentials_both() -> dict[str, str]:
 
 
 @pytest.fixture(scope="session")
-def maas_rbac_idp_env(
-    admin_client: DynamicClient,
+def maas_htpasswd_files(
     maas_user_credentials_both: dict[str, str],
-    is_byoidc: bool,
-) -> Generator[dict[str, str], None, None]:
-    """
-    - Creates a single htpasswd Secret with FREE + PREMIUM users.
-    - Adds a temporary MaaS HTPasswd IDP to oauth/cluster using ResourceEditor.
-    - Waits for oauth-openshift rollout after patch.
-    - On teardown, ResourceEditor restores the original OAuth spec and we wait again.
-    - Deletes the temporary htpasswd Secret.
+) -> Generator[tuple[str, str, str, str], None, None]:
     """
-    if is_byoidc:
-        pytest.skip("Working on OIDC support for tests that use htpasswd IDP for MaaS")
+    Create per-user htpasswd files for FREE and PREMIUM users and return
+    their file paths + base64 contents.
 
+    Cleanup of the temp files happens at teardown.
+    """
     free_username = maas_user_credentials_both["free_user"]
     free_password = maas_user_credentials_both["free_pass"]
     premium_username = maas_user_credentials_both["premium_user"]
     premium_password = maas_user_credentials_both["premium_pass"]
-    secret_name = maas_user_credentials_both["secret_name"]
-    idp_name = maas_user_credentials_both["idp_name"]
 
     free_htpasswd_file_path, free_htpasswd_b64 = create_htpasswd_file(
         username=free_username,
         password=free_password,
     )
-
     premium_htpasswd_file_path, premium_htpasswd_b64 = create_htpasswd_file(
         username=premium_username,
         password=premium_password,
     )
 
     try:
-        free_bytes = base64.b64decode(s=free_htpasswd_b64)
-        premium_bytes = base64.b64decode(s=premium_htpasswd_b64)
-        combined_bytes = free_bytes + b"\n" + premium_bytes
-        combined_htpasswd_b64 = base64.b64encode(s=combined_bytes).decode("utf-8")
-
-        oauth_resource = OAuth(name="cluster", client=admin_client)
-        oauth_spec = getattr(oauth_resource.instance, "spec", {}) or {}
-        existing_identity_providers = oauth_spec.get("identityProviders") or []
-
-        maas_identity_provider = {
-            "name": idp_name,
-            "mappingMethod": "claim",
-            "type": "HTPasswd",
-            "challenge": True,
-            "login": True,
-            "htpasswd": {"fileData": {"name": secret_name}},
-        }
-
-        updated_identity_providers = existing_identity_providers + [maas_identity_provider]
-
-        LOGGER.info(
-            f"MaaS RBAC: creating shared htpasswd Secret '{secret_name}' for users "
-            f"'{free_username}' and '{premium_username}'"
+        yield (
+            free_htpasswd_file_path,
+            free_htpasswd_b64,
+            premium_htpasswd_file_path,
+            premium_htpasswd_b64,
         )
+    finally:
+        free_htpasswd_file_path.unlink(missing_ok=True)
+        premium_htpasswd_file_path.unlink(missing_ok=True)
 
-        # --- update OAuth + create Secret ---
-        with (
-            Secret(
-                client=admin_client,
-                name=secret_name,
-                namespace="openshift-config",
-                htpasswd=combined_htpasswd_b64,
-                type="Opaque",
-                teardown=True,
-                wait_for_resource=True,
-            ),
-            ResourceEditor(patches={oauth_resource: {"spec": {"identityProviders": updated_identity_providers}}}),
-        ):
-            LOGGER.info(f"MaaS RBAC: updating OAuth with MaaS htpasswd IDP '{maas_identity_provider['name']}'")
-
-            wait_for_oauth_openshift_deployment()
-            LOGGER.info(f"MaaS RBAC: OAuth updated with MaaS IDP '{maas_identity_provider['name']}'")
-
-            # >>> this is the yield that pytest uses <<<
-            yield maas_user_credentials_both
-
-            LOGGER.info("MaaS RBAC: restoring OAuth identityProviders to original state")
-
-        # --- after exit: secret deleted + OAuth restored ---
-        try:
-            wait_for_oauth_openshift_deployment()
-            LOGGER.info("MaaS RBAC: oauth-openshift rollout completed after restoring OAuth.")
-        except TimeoutExpiredError as timeout_error:
-            LOGGER.warning(
-                f"MaaS RBAC: timeout while waiting for oauth-openshift rollout after restoring OAuth. "
-                f"Continuing teardown. Details: {timeout_error}"
-            )
 
-        # --- final sanity check ---
-        oauth_resource_after = OAuth(name="cluster", client=admin_client)
-        oauth_spec_after = getattr(oauth_resource_after.instance, "spec", {}) or {}
-        current_idps = oauth_spec_after.get("identityProviders") or []
+@pytest.fixture(scope="session")
+def maas_htpasswd_oauth_idp(
+    admin_client: DynamicClient,
+    maas_user_credentials_both: dict[str, str],
+    maas_htpasswd_files: tuple[str, str, str, str],
+    is_byoidc: bool,
+):
+    """
+    - Combines FREE + PREMIUM htpasswd entries into a single Secret.
+    - Adds the MaaS HTPasswd IDP to oauth/cluster using ResourceEditor.
+    - Waits for oauth-openshift rollout after patch.
+    - On teardown, waits again and verifies the IDP is gone.
+    """
+    if is_byoidc:
+        pytest.skip("Working on OIDC support for tests that use htpasswd IDP for MaaS")
+
+    (
+        _free_htpasswd_file_path,
+        free_htpasswd_b64,
+        _premium_htpasswd_file_path,
+        premium_htpasswd_b64,
+    ) = maas_htpasswd_files
 
-        if any(idp.get("name") == idp_name for idp in current_idps):
-            LOGGER.warning(
-                f"MaaS RBAC: temporary MaaS IDP '{idp_name}' is STILL present in OAuth spec "
-                f"after teardown — please investigate cluster OAuth configuration."
-            )
-        else:
-            LOGGER.info("MaaS RBAC: OAuth identityProviders restoration & cleanup completed")
+    free_username = maas_user_credentials_both["free_user"]
+    premium_username = maas_user_credentials_both["premium_user"]
+    secret_name = maas_user_credentials_both["secret_name"]
+    idp_name = maas_user_credentials_both["idp_name"]
 
-    finally:
-        free_htpasswd_file_path.unlink(missing_ok=True)
-        premium_htpasswd_file_path.unlink(missing_ok=True)
+    free_bytes = base64.b64decode(s=free_htpasswd_b64)
+    premium_bytes = base64.b64decode(s=premium_htpasswd_b64)
+    combined_bytes = free_bytes + b"\n" + premium_bytes
+    combined_htpasswd_b64 = base64.b64encode(s=combined_bytes).decode("utf-8")
+
+    oauth_resource = OAuth(name="cluster", client=admin_client)
+    oauth_spec = getattr(oauth_resource.instance, "spec", {}) or {}
+    existing_idps = oauth_spec.get("identityProviders") or []
+
+    maas_idp = {
+        "name": idp_name,
+        "mappingMethod": "claim",
+        "type": "HTPasswd",
+        "challenge": True,
+        "login": True,
+        "htpasswd": {"fileData": {"name": secret_name}},
+    }
+
+    updated_idps = existing_idps + [maas_idp]
+
+    LOGGER.info(
+        f"MaaS RBAC: creating shared htpasswd Secret '{secret_name}' "
+        f"for users '{free_username}' and '{premium_username}'"
+    )
+
+    with (
+        Secret(
+            client=admin_client,
+            name=secret_name,
+            namespace="openshift-config",
+            htpasswd=combined_htpasswd_b64,
+            type="Opaque",
+            teardown=True,
+            wait_for_resource=True,
+        ),
+        ResourceEditor(patches={oauth_resource: {"spec": {"identityProviders": updated_idps}}}),
+    ):
+        LOGGER.info(f"MaaS RBAC: updating OAuth with MaaS htpasswd IDP '{maas_idp['name']}'")
+        wait_for_oauth_openshift_deployment()
+        LOGGER.info(f"MaaS RBAC: OAuth updated with MaaS IDP '{maas_idp['name']}'")
+        yield
+
+    # teardown checks
+    wait_for_oauth_openshift_deployment()
+
+    oauth_after = OAuth(name="cluster", client=admin_client)
+    idps_after = getattr(oauth_after.instance, "spec", {}).get("identityProviders") or []
+    if any(idp.get("name") == idp_name for idp in idps_after):
+        pytest.fail(f"MaaS RBAC: cleanup failed, IDP {idp_name} still present after teardown")
+
+    LOGGER.info("MaaS RBAC: OAuth identityProviders restoration & cleanup completed")
+
+
+@pytest.fixture(scope="session")
+def maas_rbac_idp_env(
+    maas_htpasswd_oauth_idp,
+    maas_user_credentials_both: dict[str, str],
+):
+
+    return maas_user_credentials_both
 
 
 @pytest.fixture(scope="session")
@@ -289,7 +302,7 @@ def maas_premium_user_session(
     original_user: str,
     maas_api_server_url: str,
     is_byoidc: bool,
-    maas_rbac_idp_env: dict[str, str],  # <-- same outer fixture
+    maas_rbac_idp_env: dict[str, str],
 ) -> Generator[UserTestSession, None, None]:
     if is_byoidc:
         pytest.skip("Working on OIDC support for tests that use htpasswd IDP for MaaS")
@@ -368,27 +381,26 @@ def ocp_token_for_actor(
     """
     Log in as the requested actor ('admin' / 'free' / 'premium')
     and yield the OpenShift token for that user.
-
     """
     actor_param = getattr(request, "param", None)
 
     if isinstance(actor_param, dict):
-        actor_kind = actor_param.get("kind", "admin")
+        actor_type = actor_param.get("type", "admin")
     else:
-        actor_kind = actor_param or "admin"
+        actor_type = actor_param or "admin"
 
-    if actor_kind == "admin":
+    if actor_type == "admin":
         LOGGER.info("MaaS RBAC: using existing admin session to obtain token")
         yield get_openshift_token(client=admin_client)
     else:
-        if actor_kind == "free":
+        if actor_type == "free":
             user_session = maas_free_user_session
-        elif actor_kind == "premium":
+        elif actor_type == "premium":
             user_session = maas_premium_user_session
         else:
-            raise ValueError(f"Unknown actor kind: {actor_kind!r}")
+            raise ValueError(f"Unknown actor type: {actor_type!r}")
 
-        LOGGER.info(f"MaaS RBAC: logging in as MaaS {actor_kind} user '{user_session.username}'")
+        LOGGER.info(f"MaaS RBAC: logging in as MaaS {actor_type} user '{user_session.username}'")
         login_successful = login_with_user_password(
             api_address=maas_api_server_url,
             user=user_session.username,
@@ -426,11 +438,31 @@ def maas_token_for_actor(
         http_session=request_session_http,
         minutes=30,
     )
-    LOGGER.info("MaaS RBAC: mint token status=%s", response.status_code)
+    LOGGER.info(f"MaaS RBAC: mint token status={response.status_code}")
     assert response.status_code in (200, 201), f"mint failed: {response.status_code} {response.text[:200]}"
 
     token = body.get("token", "")
     assert isinstance(token, str) and len(token) > 10, "no usable MaaS token in response"
 
-    LOGGER.info("MaaS RBAC: minted MaaS token len=%s for current actor", len(token))
+    LOGGER.info(f"MaaS RBAC: minted MaaS token len={len(token)} for current actor")
     return token
+
+
+@pytest.fixture
+def maas_headers_for_actor(maas_token_for_actor: str) -> dict:
+    """Headers for the current actor (admin/free/premium)."""
+    return build_maas_headers(token=maas_token_for_actor)
+
+
+@pytest.fixture
+def maas_models_response_for_actor(
+    request_session_http: requests.Session,
+    base_url: str,
+    maas_headers_for_actor: dict,
+):
+    """Raw /v1/models response for the current actor."""
+    return get_maas_models_response(
+        session=request_session_http,
+        base_url=base_url,
+        headers=maas_headers_for_actor,
+    )