conftest.py

import shlex
from collections.abc import Generator
from typing import Any

import pytest
from kubernetes.dynamic import DynamicClient
from ocp_resources.deployment import Deployment
from ocp_resources.namespace import Namespace
from ocp_resources.role_binding import RoleBinding
from ocp_resources.route import Route
from ocp_resources.service_account import ServiceAccount
from pyhelper_utils.shell import run_command
from simple_logger.logger import get_logger

from tests.model_explainability.evalhub.constants import EVALHUB_PROVIDERS_ACCESS_CLUSTER_ROLE
from tests.model_explainability.evalhub.utils import list_evalhub_providers
from utilities.certificates_utils import create_ca_bundle_file
from utilities.constants import Timeout
from utilities.resources.evalhub import EvalHub

LOGGER = get_logger(name=__name__)


@pytest.fixture(scope="class")
def evalhub_cr(
    admin_client: DynamicClient,
    model_namespace: Namespace,
) -> Generator[EvalHub, Any, Any]:
    """Create an EvalHub custom resource and wait for it to be ready."""
    with EvalHub(
        client=admin_client,
        name="evalhub",
        namespace=model_namespace.name,
        wait_for_resource=True,
    ) as evalhub:
        yield evalhub


@pytest.fixture(scope="class")
def evalhub_deployment(
    admin_client: DynamicClient,
    model_namespace: Namespace,
    evalhub_cr: EvalHub,
) -> Deployment:
    """Wait for the EvalHub deployment to become available."""
    deployment = Deployment(
        client=admin_client,
        name=evalhub_cr.name,
        namespace=model_namespace.name,
    )
    deployment.wait_for_replicas(timeout=Timeout.TIMEOUT_5MIN)
    return deployment


@pytest.fixture(scope="class")
def evalhub_route(
    admin_client: DynamicClient,
    model_namespace: Namespace,
    evalhub_deployment: Deployment,
) -> Route:
    """Get the Route created by the operator for the EvalHub service."""
    return Route(
        client=admin_client,
        name=evalhub_deployment.name,
        namespace=model_namespace.name,
        ensure_exists=True,
    )


@pytest.fixture(scope="class")
def evalhub_ca_bundle_file(
    admin_client: DynamicClient,
) -> str:
    """Create a CA bundle file for verifying the EvalHub route TLS certificate."""
    return create_ca_bundle_file(client=admin_client)


@pytest.fixture(scope="class")
def evalhub_scoped_sa(
    admin_client: DynamicClient,
    model_namespace: Namespace,
    evalhub_deployment: Deployment,
) -> Generator[ServiceAccount, Any, Any]:
    """ServiceAccount with providers access in the test namespace."""
    with ServiceAccount(
        client=admin_client,
        name="evalhub-test-user",
        namespace=model_namespace.name,
    ) as sa:
        yield sa


@pytest.fixture(scope="class")
def evalhub_providers_role_binding(
    admin_client: DynamicClient,
    model_namespace: Namespace,
    evalhub_scoped_sa: ServiceAccount,
) -> Generator[RoleBinding, Any, Any]:
    """RoleBinding granting the scoped SA providers access via the ClusterRole."""
    with RoleBinding(
        client=admin_client,
        name="evalhub-test-providers-access",
        namespace=model_namespace.name,
        role_ref_kind="ClusterRole",
        role_ref_name=EVALHUB_PROVIDERS_ACCESS_CLUSTER_ROLE,
        subjects_kind="ServiceAccount",
        subjects_name=evalhub_scoped_sa.name,
    ) as rb:
        yield rb


@pytest.fixture(scope="class")
def evalhub_scoped_token(
    evalhub_scoped_sa: ServiceAccount,
    model_namespace: Namespace,
) -> str:
    """Short-lived token for the scoped ServiceAccount."""
    return run_command(
        command=shlex.split(f"oc create token -n {model_namespace.name} {evalhub_scoped_sa.name} --duration=30m")
    )[1].strip()


@pytest.fixture(scope="class")
def evalhub_providers_response(
    model_namespace: Namespace,
    evalhub_scoped_token: str,
    evalhub_providers_role_binding: RoleBinding,
    evalhub_ca_bundle_file: str,
    evalhub_route: Route,
) -> dict:
    """Fetch the providers list once per test class."""
    return list_evalhub_providers(
        host=evalhub_route.host,
        token=evalhub_scoped_token,
        ca_bundle_file=evalhub_ca_bundle_file,
        tenant=model_namespace.name,
    )


@pytest.fixture(scope="class")
def evalhub_unauthorised_sa(
    admin_client: DynamicClient,
    model_namespace: Namespace,
    evalhub_deployment: Deployment,
) -> Generator[ServiceAccount, Any, Any]:
    """ServiceAccount without any EvalHub RBAC in the test namespace."""
    with ServiceAccount(
        client=admin_client,
        name="evalhub-no-access-user",
        namespace=model_namespace.name,
    ) as sa:
        yield sa


@pytest.fixture(scope="class")
def evalhub_unauthorised_token(
    evalhub_unauthorised_sa: ServiceAccount,
    model_namespace: Namespace,
) -> str:
    """Short-lived token for the unauthorised ServiceAccount."""
    return run_command(
        command=shlex.split(f"oc create token -n {model_namespace.name} {evalhub_unauthorised_sa.name} --duration=30m")
    )[1].strip()