Skip to content

Refactor test to verify TrustyAIService works with multiple namespaces #408

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
adolfo-ab merged 3 commits into from
Jul 24, 2025
Merged

Refactor test to verify TrustyAIService works with multiple namespaces #408

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
0b53f4d
[WIP] Refactor to verify TrustyAIService works with multiple ns
kpunwatk Jul 8, 2025
cb2738d
Merge branch 'main' into multi_ns
kpunwatk Jul 23, 2025
e3af2d8
[pre-commit.ci] auto fixes from pre-commit.com hooks
pre-commit-ci[bot] Jul 23, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
220 changes: 0 additions & 220 deletions tests/model_explainability/trustyai_service/service/conftest.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,52 +1,24 @@
from typing import Generator, Any

import pytest
from _pytest.fixtures import FixtureRequest
from kubernetes.dynamic import DynamicClient
from ocp_resources.config_map import ConfigMap
from ocp_resources.inference_service import InferenceService
from ocp_resources.maria_db import MariaDB
from ocp_resources.namespace import Namespace
from ocp_resources.pod import Pod
from ocp_resources.role import Role
from ocp_resources.role_binding import RoleBinding
from ocp_resources.secret import Secret
from ocp_resources.service import Service
from ocp_resources.service_account import ServiceAccount
from ocp_resources.serving_runtime import ServingRuntime
from ocp_resources.trustyai_service import TrustyAIService


from tests.model_explainability.trustyai_service.constants import (
TAI_METRICS_CONFIG,
TAI_DATA_CONFIG,
TAI_PVC_STORAGE_CONFIG,
GAUSSIAN_CREDIT_MODEL,
GAUSSIAN_CREDIT_MODEL_STORAGE_PATH,
GAUSSIAN_CREDIT_MODEL_RESOURCES,
KSERVE_MLSERVER,
KSERVE_MLSERVER_CONTAINERS,
KSERVE_MLSERVER_SUPPORTED_MODEL_FORMATS,
KSERVE_MLSERVER_ANNOTATIONS,
XGBOOST,
TAI_DB_STORAGE_CONFIG,
ISVC_GETTER,
)
from tests.model_explainability.trustyai_service.trustyai_service_utils import (
TRUSTYAI_SERVICE_NAME,
wait_for_isvc_deployment_registered_by_trustyai_service,
)
from tests.model_explainability.trustyai_service.utils import (
create_trustyai_service,
create_isvc_getter_service_account,
create_isvc_getter_role,
create_isvc_getter_role_binding,
create_isvc_getter_token_secret,
)
from utilities.constants import KServeDeploymentType, Labels
from utilities.inference_utils import create_isvc
from utilities.infra import create_ns, create_inference_token
from utilities.minio import create_minio_data_connection_secret


INVALID_TLS_CERTIFICATE: str = "LS0tLS1CRUdJTiBDRVJUSUZJQ0FURS0tLS0tCk1JSUJnRENDQVNlZ0F3SUJBZ0lRRGtTcXVuUWRzRmZwdi8zSm\
Expand All @@ -59,150 +31,6 @@
LS0t" # pragma: allowlist secret


@pytest.fixture(scope="class")
def model_namespace_2(
request: FixtureRequest,
pytestconfig: pytest.Config,
admin_client: DynamicClient,
teardown_resources: bool,
) -> Generator[Namespace, Any, Any]:
if request.param.get("modelmesh-enabled"):
request.getfixturevalue(argname="enabled_modelmesh_in_dsc")

if pytestconfig.option.post_upgrade:
ns = Namespace(client=admin_client, name=request.param["name"])
yield ns
ns.clean_up()
else:
with create_ns(
admin_client=admin_client,
pytest_request=request,
teardown=teardown_resources,
) as ns:
yield ns


@pytest.fixture(scope="class")
def minio_data_connection_2(
request: FixtureRequest,
admin_client: DynamicClient,
model_namespace_2: Namespace,
minio_service: Service,
) -> Generator[Secret, Any, Any]:
with create_minio_data_connection_secret(
minio_service=minio_service,
model_namespace=model_namespace_2.name,
aws_s3_bucket=request.param["bucket"],
client=admin_client,
) as secret:
yield secret


@pytest.fixture(scope="class")
def trustyai_service_with_pvc_storage_2(
pytestconfig: pytest.Config,
admin_client: DynamicClient,
model_namespace_2: Namespace,
cluster_monitoring_config: ConfigMap,
user_workload_monitoring_config: ConfigMap,
teardown_resources: bool,
) -> Generator[TrustyAIService, Any, Any]:
trustyai_service_kwargs = {
"client": admin_client,
"namespace": model_namespace_2.name,
"name": TRUSTYAI_SERVICE_NAME,
}

if pytestconfig.option.post_upgrade:
trustyai_service = TrustyAIService(**trustyai_service_kwargs)
yield trustyai_service
trustyai_service.clean_up()

else:
yield from create_trustyai_service(
**trustyai_service_kwargs,
storage=TAI_PVC_STORAGE_CONFIG,
metrics=TAI_METRICS_CONFIG,
data=TAI_DATA_CONFIG,
wait_for_replicas=True,
teardown=teardown_resources,
)


@pytest.fixture(scope="class")
def gaussian_credit_model_2(
pytestconfig: pytest.Config,
admin_client: DynamicClient,
model_namespace_2: Namespace,
minio_pod: Pod,
minio_service: Service,
minio_data_connection_2: Secret,
mlserver_runtime_2: ServingRuntime,
trustyai_service_with_pvc_storage_2: TrustyAIService,
teardown_resources: bool,
) -> Generator[InferenceService, Any, Any]:
gaussian_credit_model_kwargs = {
"client": admin_client,
"namespace": model_namespace_2.name,
"name": GAUSSIAN_CREDIT_MODEL,
}

if pytestconfig.option.post_upgrade:
isvc = InferenceService(**gaussian_credit_model_kwargs)
yield isvc
isvc.clean_up()
else:
with create_isvc(
deployment_mode=KServeDeploymentType.SERVERLESS,
model_format=XGBOOST,
runtime=mlserver_runtime_2.name,
storage_key=minio_data_connection_2.name,
storage_path=GAUSSIAN_CREDIT_MODEL_STORAGE_PATH,
enable_auth=True,
wait_for_predictor_pods=False,
resources=GAUSSIAN_CREDIT_MODEL_RESOURCES,
teardown=teardown_resources,
**gaussian_credit_model_kwargs,
) as isvc:
wait_for_isvc_deployment_registered_by_trustyai_service(
client=admin_client,
isvc=isvc,
runtime_name=mlserver_runtime_2.name,
)
yield isvc


@pytest.fixture(scope="class")
def mlserver_runtime_2(
pytestconfig: pytest.Config,
admin_client: DynamicClient,
minio_data_connection: Secret,
model_namespace_2: Namespace,
teardown_resources: bool,
) -> Generator[ServingRuntime, Any, Any]:
mlserver_runtime_kwargs = {
"client": admin_client,
"namespace": model_namespace_2.name,
"name": KSERVE_MLSERVER,
}

if pytestconfig.option.post_upgrade:
serving_runtime = ServingRuntime(**mlserver_runtime_kwargs)
yield serving_runtime
serving_runtime.clean_up()
else:
with ServingRuntime(
containers=KSERVE_MLSERVER_CONTAINERS,
supported_model_formats=KSERVE_MLSERVER_SUPPORTED_MODEL_FORMATS,
protocol_versions=["v2"],
annotations=KSERVE_MLSERVER_ANNOTATIONS,
label={Labels.OpenDataHub.DASHBOARD: "true"},
teardown=teardown_resources,
**mlserver_runtime_kwargs,
) as mlserver:
yield mlserver


@pytest.fixture(scope="class")
def trustyai_service_with_invalid_db_cert(
admin_client: DynamicClient,
Expand Down Expand Up @@ -236,51 +64,3 @@ def trustyai_invalid_db_ca_secret(
data_dict={"ca.crt": INVALID_TLS_CERTIFICATE},
) as secret:
yield secret


@pytest.fixture(scope="class")
def isvc_getter_service_account_2(
admin_client: DynamicClient, model_namespace_2: Namespace
) -> Generator[ServiceAccount, Any, Any]:
yield from create_isvc_getter_service_account(client=admin_client, namespace=model_namespace_2, name=ISVC_GETTER)


@pytest.fixture(scope="class")
def isvc_getter_role_2(admin_client: DynamicClient, model_namespace_2: Namespace) -> Generator[Role, Any, Any]:
yield from create_isvc_getter_role(client=admin_client, namespace=model_namespace_2, name=ISVC_GETTER)


@pytest.fixture(scope="class")
def isvc_getter_role_binding_2(
admin_client: DynamicClient,
model_namespace_2: Namespace,
isvc_getter_role_2: Role,
isvc_getter_service_account_2: ServiceAccount,
) -> Generator[RoleBinding, Any, Any]:
yield from create_isvc_getter_role_binding(
client=admin_client,
namespace=model_namespace_2,
role=isvc_getter_role_2,
service_account=isvc_getter_service_account_2,
name=ISVC_GETTER,
)


@pytest.fixture(scope="class")
def isvc_getter_token_secret_2(
admin_client: DynamicClient,
model_namespace_2: Namespace,
isvc_getter_service_account_2: ServiceAccount,
isvc_getter_role_binding_2: RoleBinding,
) -> Generator[Secret, Any, Any]:
yield from create_isvc_getter_token_secret(
client=admin_client,
name="sa-token",
namespace=model_namespace_2,
service_account=isvc_getter_service_account_2,
)


@pytest.fixture(scope="class")
def isvc_getter_token_2(isvc_getter_service_account_2: ServiceAccount, isvc_getter_token_secret_2: Secret) -> str:
return create_inference_token(model_service_account=isvc_getter_service_account_2)
Loading