test(maas): add MaaS API and controller component health tests

tests/model_serving/model_server/maas_billing/maas_subscription/component_health/test_maas_api_health.py

@@ -0,0 +1,66 @@
+import pytest
+from kubernetes.dynamic import DynamicClient
+from ocp_resources.data_science_cluster import DataScienceCluster
+from ocp_resources.deployment import Deployment
+from pytest_testconfig import config as py_config
+from simple_logger.logger import get_logger
+
+from utilities.constants import DscComponents
+from utilities.general import wait_for_pods_running
+
+LOGGER = get_logger(name=__name__)
+
+
+@pytest.mark.component_health
+@pytest.mark.usefixtures("maas_controller_enabled_latest")
+class TestMaaSApiComponentHealth:
+    def test_maas_management_state(
+        self,
+        dsc_resource: DataScienceCluster,
+    ) -> None:
+        """Verify modelsAsService managementState is MANAGED in DSC."""
+        assert (
+            dsc_resource.instance.spec.components[DscComponents.KSERVE].modelsAsService.managementState
+            == DscComponents.ManagementState.MANAGED
+        )
+
+    def test_maas_condition_in_dsc(
+        self,
+        dsc_resource: DataScienceCluster,
+    ) -> None:
+        """Verify ModelsAsServiceReady condition is True in DSC."""
+        for condition in dsc_resource.instance.status.conditions:
+            if condition.type == "ModelsAsServiceReady":
+                assert condition.status == "True"
+                break
+        else:
+            pytest.fail("ModelsAsServiceReady condition not found in DSC")
+
+    def test_maas_api_deployment_available(
+        self,
+        admin_client: DynamicClient,
+    ) -> None:
+        """Verify maas-api deployment Available=True."""
+        applications_namespace = py_config["applications_namespace"]
+
+        maas_api_deployment = Deployment(
+            client=admin_client,
+            name="maas-api",
+            namespace=applications_namespace,
+            ensure_exists=True,
+        )
+        maas_api_deployment.wait_for_condition(
+            condition="Available",
+            status="True",
+            timeout=120,
+        )
+
+    def test_maas_api_pods_health(
+        self,
+        admin_client: DynamicClient,
+    ) -> None:
+        """Verify maas-api pods are Running/Ready."""
+        applications_namespace = py_config["applications_namespace"]
+        LOGGER.info(f"Checking maas-api pods in namespace {applications_namespace}")
+
+        wait_for_pods_running(admin_client=admin_client, namespace_name=applications_namespace)

tests/model_serving/model_server/maas_billing/maas_subscription/component_health/test_maas_controller_health.py

@@ -0,0 +1,97 @@
+import pytest
+from kubernetes.dynamic import DynamicClient
+from ocp_resources.custom_resource_definition import CustomResourceDefinition
+from ocp_resources.data_science_cluster import DataScienceCluster
+from ocp_resources.deployment import Deployment
+from pytest_testconfig import config as py_config
+from simple_logger.logger import get_logger
+
+from utilities.constants import ApiGroups
+from utilities.general import wait_for_pods_running
+
+LOGGER = get_logger(name=__name__)
+
+
+@pytest.mark.component_health
+@pytest.mark.usefixtures("maas_controller_enabled_latest")
+class TestMaaSController:
+    def test_maas_condition_in_dsc(
+        self,
+        dsc_resource: DataScienceCluster,
+    ) -> None:
+        """Verify ModelsAsServiceReady condition is True in DSC (MR-style loop + break + else fail)."""
+        for condition in dsc_resource.instance.status.conditions:
+            if condition.type == "ModelsAsServiceReady":
+                assert condition.status == "True"
+                break
+        else:
+            pytest.fail("ModelsAsServiceReady condition not found in DSC")
+
+    def test_maas_controller_crds_exist(
+        self,
+        admin_client: DynamicClient,
+    ) -> None:
+        """Verify MaaS controller CRDs exist (no CustomResourceDefinition.get usage)."""
+        expected_crd_names = (
+            f"maasmodels.{ApiGroups.MAAS_IO}",
+            f"maasauthpolicies.{ApiGroups.MAAS_IO}",
+            f"maassubscriptions.{ApiGroups.MAAS_IO}",
+        )
+        missing_crds = []
+        for crd_name in expected_crd_names:
+            crd_resource = CustomResourceDefinition(
+                client=admin_client,
+                name=crd_name,
+                ensure_exists=True,
+            )
+            if not crd_resource.exists:
+                missing_crds.append(crd_name)
+
+        assert not missing_crds, f"Missing expected CRDs: {', '.join(missing_crds)}"
+
+    def test_maas_controller_deployment_available(
+        self,
+        admin_client: DynamicClient,
+    ) -> None:
+        """Verify maas-controller Deployment is Available."""
+        applications_namespace = py_config["applications_namespace"]
+
+        controller_deployment = Deployment(
+            client=admin_client,
+            name="maas-controller",
+            namespace=applications_namespace,
+            ensure_exists=True,
+        )
+        controller_deployment.wait_for_condition(
+            condition="Available",
+            status="True",
+            timeout=120,
+        )
+
+    def test_maas_controller_pods_health(
+        self,
+        admin_client: DynamicClient,
+    ) -> None:
+        """Verify maas-controller pods are Running."""
+        applications_namespace = py_config["applications_namespace"]
+        LOGGER.info(f"Testing Pods in namespace {applications_namespace} for MaaS Controller health")
+        wait_for_pods_running(admin_client=admin_client, namespace_name=applications_namespace)
+
+    @pytest.mark.parametrize(
+        "resource_fixture, kind_label",
+        [
+            ("maas_model_tinyllama_free", "MaaSModel"),
+            ("maas_auth_policy_tinyllama_free", "MaaSAuthPolicy"),
+            ("maas_subscription_tinyllama_free", "MaaSSubscription"),
+        ],
+    )
+    def test_maas_subscription_stack_ready_for_free_model(
+        self,
+        request: pytest.FixtureRequest,
+        resource_fixture: str,
+        kind_label: str,
+    ) -> None:
+        """Verify the MaaS subscription flow objects are created and Ready."""
+        resource = request.getfixturevalue(argname=resource_fixture)
+        LOGGER.info(f"Checking {kind_label} {resource.name} is Ready")
+        resource.wait_for_condition(condition="Ready", status="True", timeout=300)