diff --git a/plugins/nemo-deployments/README.md b/plugins/nemo-deployments/README.md index 1d5e6d174d9..8af6d497a1c 100644 --- a/plugins/nemo-deployments/README.md +++ b/plugins/nemo-deployments/README.md @@ -104,3 +104,50 @@ omit the per-container `restart_policy` on init containers and run that container as a regular main container instead — the compiler does not fall back to legacy sidecar emulation (emptyDir readiness files, etc.) automatically. + +## Cluster validation + +`tests/integration/backends/k8s/` and `tests/integration/test_reconcile_k8s.py` +exercise the k8s backend and reconciler against a real cluster. Like the Docker +integration tests above, they skip automatically (CI stays green) when no +cluster is reachable. + +**Prerequisites:** + +- `kind` and `kubectl` on your `PATH` (for the local run) or an existing + cluster context (for the dev-blue run). +- `uv` for running the test suite. +- A kubeconfig pointing at a reachable cluster — either the one `kind create + cluster` sets automatically, or your own context via `kubectl config + use-context`. + +**Local run against kind** (no dev-blue access needed): + +```bash +kind create cluster # or reuse an existing one; sets your kubeconfig context +uv run pytest plugins/nemo-deployments/tests/integration/backends/k8s -v +uv run pytest plugins/nemo-deployments/tests/integration/test_reconcile_k8s.py -v +``` + +**Run against dev-blue** (uses your namespace and your own RBAC there): + +```bash +kubectl config use-context +export NMP_K8S_ITEST_NAMESPACE= # defaults to "default" +uv run pytest plugins/nemo-deployments/tests/integration/backends/k8s -v +``` + +Tests authenticate with whatever kubeconfig identity is currently active in +your shell, not the restricted `nmp-core` controller ServiceAccount, so they +validate backend behavior against a real API server but do **not** exercise +the RBAC scope granted by the deploy chart's `controller-role.yaml`. That +still needs a manual smoke test of the deployed platform pod. + +`test_reconcile_k8s.py` covers prerequisite gating (one deployment blocking on +another) but not PVC-mount gating: kind's default `local-path` StorageClass uses +`WaitForFirstConsumer` binding, so an unconsumed PVC never reaches `BOUND`, and +`DeploymentReconciler` requires a mounted volume to already be `BOUND` before it +creates the pod that would consume it — a chicken-and-egg specific to +`WaitForFirstConsumer` storage classes that doesn't arise on `Immediate`-binding +classes (the common case outside kind). The standalone PVC lifecycle scenario in +`test_k8s_backend.py` accepts `PENDING` as a valid terminal state for this reason. diff --git a/plugins/nemo-deployments/tests/integration/backends/k8s/test_k8s_backend.py b/plugins/nemo-deployments/tests/integration/backends/k8s/test_k8s_backend.py new file mode 100644 index 00000000000..50cb0f31ee1 --- /dev/null +++ b/plugins/nemo-deployments/tests/integration/backends/k8s/test_k8s_backend.py @@ -0,0 +1,293 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Integration tests for K8sDeploymentBackend against a real cluster (kind or kubeconfig). + +Namespace defaults to ``default`` (present on any cluster, and where a kind admin +kubeconfig has full rights). Point at a different namespace you control via +``NMP_K8S_ITEST_NAMESPACE`` — e.g. your dev-blue namespace, which only has the RBAC +verbs the deploy chart's ``controller-role.yaml`` grants (see AIRCORE-757 Phase 6). +""" + +from __future__ import annotations + +import asyncio +import os +from collections.abc import Iterator +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from kubeconfig_availability import skip_without_kubeconfig +from kubernetes.client.rest import ApiException +from nemo_deployments_plugin.backends.k8s.backend import K8sDeploymentBackend +from nemo_deployments_plugin.backends.k8s.client import k8s_client_module +from nemo_deployments_plugin.backends.labels import k8s_deployment_resource_name +from nemo_deployments_plugin.backends.registry import BACKEND_CLASSES +from nemo_deployments_plugin.constants import MANAGED_BY_LABEL +from nemo_deployments_plugin.entities import ConfigFile, Container, ContainerPort, Deployment, DeploymentConfig + +pytestmark = [ + pytest.mark.skipif("k8s" not in BACKEND_CLASSES, reason="K8sDeploymentBackend not registered"), + skip_without_kubeconfig, +] + +NAMESPACE = os.environ.get("NMP_K8S_ITEST_NAMESPACE", "default") +LABELS = {"managed-by": MANAGED_BY_LABEL} +POLL_ATTEMPTS = 60 +POLL_INTERVAL_SECONDS = 1 + + +@pytest.fixture +def mock_entities() -> AsyncMock: + return AsyncMock() + + +@pytest.fixture +def k8s_backend(mock_entities: AsyncMock) -> Iterator[K8sDeploymentBackend]: + mock_sdk = MagicMock() + with ( + patch("nemo_deployments_plugin.backends.k8s.backend.AsyncEntitiesResource"), + patch("nemo_deployments_plugin.backends.k8s.backend.NemoEntitiesClient", return_value=mock_entities), + ): + backend = K8sDeploymentBackend(mock_sdk, {"default_namespace": NAMESPACE, "request_timeout": 30}) + backend._entities = mock_entities + try: + yield backend + finally: + backend.shutdown() + + +def _never_config( + *, + name: str = "echo-cfg", + args: list[str] | None = None, + config_files: list[ConfigFile] | None = None, +) -> DeploymentConfig: + return DeploymentConfig( + name=name, + workspace="itest", + restart_policy="Never", # ty: ignore[unknown-argument] + containers=[ + Container(name="main", image="alpine:3.20", command=["sh", "-c"], args=args or ["echo hello-from-k8s"]) + ], + config_files=config_files or [], # ty: ignore[unknown-argument] + ) + + +def _always_http_config() -> DeploymentConfig: + return DeploymentConfig( + name="http-cfg", + workspace="itest", + restart_policy="Always", # ty: ignore[unknown-argument] + containers=[ + Container( + name="main", + image="nginx:alpine", + ports=[ContainerPort(containerPort=80, protocol="TCP", name="http")], + ) + ], + ) + + +def _configure_deployment_lookup( + mock_entities: AsyncMock, + *, + name: str, + config_name: str, + config: DeploymentConfig, + workspace: str = "itest", +) -> None: + """Wire the mocked entity client to resolve both the Deployment and its DeploymentConfig. + + Unlike the docker backend (which derives container identity from workspace/name alone), + the k8s backend's read/delete paths load the Deployment entity first to discover + ``deployment_config``, then load that DeploymentConfig — so a single blanket + ``get.return_value`` isn't enough here. + """ + deployment = Deployment(name=name, workspace=workspace, deployment_config=config_name) + + async def get_side_effect( + entity_type: type, + entity_name: str, + workspace: str | None = None, + ) -> Deployment | DeploymentConfig: + if entity_type is Deployment: + return deployment + if entity_type is DeploymentConfig: + return config + raise KeyError(entity_name) + + mock_entities.get.side_effect = get_side_effect + + +async def _wait_for_status( + k8s_backend: K8sDeploymentBackend, + name: str, + *, + terminal_statuses: tuple[str, ...], + attempts: int = POLL_ATTEMPTS, +) -> None: + for _ in range(attempts): + status = await k8s_backend.read_status(workspace="itest", name=name) + if status.status in terminal_statuses: + return + await asyncio.sleep(POLL_INTERVAL_SECONDS) + pytest.fail(f"{name!r} did not reach {terminal_statuses} within {attempts} attempts") + + +@pytest.mark.asyncio +async def test_pvc_lifecycle(k8s_backend: K8sDeploymentBackend) -> None: + created = await k8s_backend.create_volume( + workspace="itest-pvc", + name="data", + size="1Gi", + access_modes=["ReadWriteOnce"], + backend_config={}, + ) + try: + # Unconsumed PVCs stay Pending under kind's default WaitForFirstConsumer + # storage class, so both Pending and Bound are valid terminal states here. + assert created.status in ("PENDING", "BOUND") + + read = await k8s_backend.read_volume_status(workspace="itest-pvc", name="data") + assert read.status in ("PENDING", "BOUND") + finally: + deleted = await k8s_backend.delete_volume("itest-pvc", "data") + + assert deleted.status == "RELEASED" + + +@pytest.mark.asyncio +async def test_never_job_succeeds(k8s_backend: K8sDeploymentBackend, mock_entities: AsyncMock) -> None: + config = _never_config() + _configure_deployment_lookup(mock_entities, name="echo-job", config_name="echo-cfg", config=config) + + try: + created = await k8s_backend.create_deployment( + workspace="itest", + name="echo-job", + config_name="echo-cfg", + labels=LABELS, + backend_config={}, + ) + assert created.status in ("STARTING", "SUCCEEDED") + + await _wait_for_status(k8s_backend, "echo-job", terminal_statuses=("SUCCEEDED", "FAILED")) + status = await k8s_backend.read_status(workspace="itest", name="echo-job") + assert status.status == "SUCCEEDED" + assert status.exit_code == 0 + finally: + await k8s_backend.delete_deployment("itest", "echo-job") + + +@pytest.mark.asyncio +async def test_configmap_mount_round_trip(k8s_backend: K8sDeploymentBackend, mock_entities: AsyncMock) -> None: + """A ConfigFile mounted via ConfigMap is readable inside the container.""" + config = _never_config( + name="cm-cfg", + args=["cat /etc/nemo-config/hello.txt"], + config_files=[ConfigFile(path="/etc/nemo-config/hello.txt", content="hello-from-configmap")], + ) + _configure_deployment_lookup(mock_entities, name="cm-job", config_name="cm-cfg", config=config) + + try: + await k8s_backend.create_deployment( + workspace="itest", + name="cm-job", + config_name="cm-cfg", + labels=LABELS, + backend_config={}, + ) + await _wait_for_status(k8s_backend, "cm-job", terminal_statuses=("SUCCEEDED", "FAILED")) + status = await k8s_backend.read_status(workspace="itest", name="cm-job") + assert status.status == "SUCCEEDED" + + logs = await k8s_backend.get_logs(workspace="itest", name="cm-job") + assert any("hello-from-configmap" in line for line in logs.lines) + finally: + await k8s_backend.delete_deployment("itest", "cm-job") + + +@pytest.mark.asyncio +async def test_always_deployment_becomes_ready(k8s_backend: K8sDeploymentBackend, mock_entities: AsyncMock) -> None: + config = _always_http_config() + _configure_deployment_lookup(mock_entities, name="http-svc", config_name="http-cfg", config=config) + resource_name = k8s_deployment_resource_name("itest", "http-svc") + + try: + await k8s_backend.create_deployment( + workspace="itest", + name="http-svc", + config_name="http-cfg", + labels=LABELS, + backend_config={}, + ) + + await _wait_for_status(k8s_backend, "http-svc", terminal_statuses=("READY",)) + status = await k8s_backend.read_status(workspace="itest", name="http-svc") + assert status.status == "READY" + assert status.endpoints + + core_v1 = k8s_backend.clients.core_v1 + service = await asyncio.to_thread( + core_v1.read_namespaced_service, + name=resource_name, + namespace=NAMESPACE, + ) + assert service.metadata.name == resource_name + finally: + await k8s_backend.delete_deployment("itest", "http-svc") + + +@pytest.mark.asyncio +async def test_delete_rejects_foreign_resource(k8s_backend: K8sDeploymentBackend, mock_entities: AsyncMock) -> None: + """A resource with the plugin's expected name but no identity labels is not touched.""" + config = _never_config(name="foreign-cfg") + _configure_deployment_lookup(mock_entities, name="foreign-job", config_name="foreign-cfg", config=config) + resource_name = k8s_deployment_resource_name("itest", "foreign-job") + + batch_v1 = k8s_backend.clients.batch_v1 + k8s = k8s_client_module() + foreign_job = k8s.client.V1Job( + api_version="batch/v1", + kind="Job", + metadata=k8s.client.V1ObjectMeta(name=resource_name, labels={"owner": "someone-else"}), + spec=k8s.client.V1JobSpec( + template=k8s.client.V1PodTemplateSpec( + spec=k8s.client.V1PodSpec( + restart_policy="Never", + containers=[k8s.client.V1Container(name="main", image="alpine:3.20", command=["true"])], + ) + ) + ), + ) + + try: + await asyncio.to_thread( + batch_v1.create_namespaced_job, + namespace=NAMESPACE, + body=foreign_job, + ) + + result = await k8s_backend.delete_deployment("itest", "foreign-job") + assert result.status == "FAILED" + assert "not managed by this plugin" in result.status_message + + # Foreign resource must still exist. + existing = await asyncio.to_thread( + batch_v1.read_namespaced_job, + name=resource_name, + namespace=NAMESPACE, + ) + assert existing.metadata.name == resource_name + finally: + try: + await asyncio.to_thread( + batch_v1.delete_namespaced_job, + name=resource_name, + namespace=NAMESPACE, + propagation_policy="Background", + ) + except ApiException as exc: + if exc.status != 404: + raise diff --git a/plugins/nemo-deployments/tests/integration/conftest.py b/plugins/nemo-deployments/tests/integration/conftest.py index 33f8b3eae0a..bd983fc87a3 100644 --- a/plugins/nemo-deployments/tests/integration/conftest.py +++ b/plugins/nemo-deployments/tests/integration/conftest.py @@ -1,7 +1,7 @@ # SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: Apache-2.0 -"""Integration test fixtures for docker-backed deployments.""" +"""Integration test fixtures for docker- and k8s-backed deployments.""" from __future__ import annotations @@ -9,16 +9,25 @@ import pytest from docker_availability import DOCKER_AVAILABLE, skip_without_docker +from kubeconfig_availability import KUBECONFIG_AVAILABLE, skip_without_kubeconfig -__all__ = ["DOCKER_AVAILABLE", "skip_without_docker"] +__all__ = ["DOCKER_AVAILABLE", "KUBECONFIG_AVAILABLE", "skip_without_docker", "skip_without_kubeconfig"] -# All tests in this package share one Docker daemon and the same managed-by label -# namespace. Run them on a single xdist worker to avoid container/volume races. +# Tests under the same substrate share one daemon/cluster and the same managed-by +# label namespace. Run each substrate's tests on a single xdist worker to avoid +# container/volume/pod name races; docker and k8s tests don't collide with each +# other, so they get separate groups rather than one shared one. _DOCKER_INTEGRATION_XDIST_GROUP = "nemo_deployments_docker_integration" +_K8S_INTEGRATION_XDIST_GROUP = "nemo_deployments_k8s_integration" def pytest_collection_modifyitems(items: list[pytest.Item]) -> None: integration_dir = Path(__file__).parent.resolve() + k8s_dir = (integration_dir / "backends" / "k8s").resolve() + k8s_reconcile_test = (integration_dir / "test_reconcile_k8s.py").resolve() for item in items: - if integration_dir in Path(str(item.fspath)).resolve().parents: + item_path = Path(str(item.fspath)).resolve() + if k8s_dir in item_path.parents or item_path == k8s_reconcile_test: + item.add_marker(pytest.mark.xdist_group(_K8S_INTEGRATION_XDIST_GROUP)) + elif integration_dir in item_path.parents: item.add_marker(pytest.mark.xdist_group(_DOCKER_INTEGRATION_XDIST_GROUP)) diff --git a/plugins/nemo-deployments/tests/integration/kubeconfig_availability.py b/plugins/nemo-deployments/tests/integration/kubeconfig_availability.py new file mode 100644 index 00000000000..31e64607c59 --- /dev/null +++ b/plugins/nemo-deployments/tests/integration/kubeconfig_availability.py @@ -0,0 +1,25 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Shared Kubernetes cluster availability check for integration tests.""" + +from __future__ import annotations + +import pytest + +try: + from kubernetes import client, config + + try: + config.load_kube_config() + except config.ConfigException: + config.load_incluster_config() + client.VersionApi().get_code(_request_timeout=5) + KUBECONFIG_AVAILABLE: bool = True +except Exception: + KUBECONFIG_AVAILABLE = False + +skip_without_kubeconfig: pytest.MarkDecorator = pytest.mark.skipif( + not KUBECONFIG_AVAILABLE, + reason="No reachable Kubernetes cluster (set KUBECONFIG or run against kind)", +) diff --git a/plugins/nemo-deployments/tests/integration/test_reconcile_k8s.py b/plugins/nemo-deployments/tests/integration/test_reconcile_k8s.py new file mode 100644 index 00000000000..20c835e3712 --- /dev/null +++ b/plugins/nemo-deployments/tests/integration/test_reconcile_k8s.py @@ -0,0 +1,147 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Integration test for deployment reconciliation against a real k8s backend. + +Requires AIRCORE-757 K8sDeploymentBackend to be registered in BACKEND_CLASSES. + +No volume/PVC mount here (unlike ``test_reconcile_docker.py``'s puller/server chain): +kind's default ``local-path`` StorageClass uses ``WaitForFirstConsumer`` binding, so an +unconsumed PVC never reaches BOUND, and ``DeploymentReconciler`` gates deployment create +on the mounted volume already being BOUND (see ``volume_mounts_ready``) — a chicken-and-egg +that only resolves on storage classes with ``Immediate`` binding (the common case outside +kind, e.g. most cloud block-storage classes). Prerequisite gating alone is exercised here. +""" + +from __future__ import annotations + +import asyncio +import os +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from kubeconfig_availability import skip_without_kubeconfig +from nemo_deployments_plugin.backends.k8s.backend import K8sDeploymentBackend +from nemo_deployments_plugin.backends.registry import BACKEND_CLASSES, ExecutorRegistry +from nemo_deployments_plugin.config import ControllerConfig +from nemo_deployments_plugin.entities import Container, ContainerPort, Deployment, DeploymentConfig, Prerequisite +from nemo_deployments_plugin.reconciler.deployment_reconciler import DeploymentReconciler + +pytestmark = [ + pytest.mark.skipif("k8s" not in BACKEND_CLASSES, reason="Requires K8sDeploymentBackend (AIRCORE-757)"), + skip_without_kubeconfig, +] + +NAMESPACE = os.environ.get("NMP_K8S_ITEST_NAMESPACE", "default") +POLL_ATTEMPTS = 60 +POLL_INTERVAL_SECONDS = 1 + + +@pytest.fixture +def k8s_registry() -> ExecutorRegistry: + mock_sdk = MagicMock() + with ( + patch("nemo_deployments_plugin.backends.k8s.backend.AsyncEntitiesResource"), + patch("nemo_deployments_plugin.backends.k8s.backend.NemoEntitiesClient"), + ): + backend = K8sDeploymentBackend(mock_sdk, {"default_namespace": NAMESPACE, "request_timeout": 30}) + return ExecutorRegistry({"k8s": backend}, default_executor="k8s") + + +def _backend(k8s_registry: ExecutorRegistry) -> K8sDeploymentBackend: + backend = k8s_registry.resolve("k8s") + assert isinstance(backend, K8sDeploymentBackend) + return backend + + +@pytest.mark.asyncio +async def test_puller_server_prerequisite_chain(k8s_registry: ExecutorRegistry) -> None: + """Puller (Never) -> server (Always + prerequisite) driven end-to-end by the reconciler.""" + entities = AsyncMock() + entities.update = AsyncMock(side_effect=lambda entity: entity) + + puller_dep = Deployment(name="puller", workspace="itest", deployment_config="puller-cfg", status="PENDING") + server_dep = Deployment( + name="server", + workspace="itest", + deployment_config="server-cfg", + status="PENDING", + prerequisites=[Prerequisite(deployment_name="puller", condition="succeeded")], + ) + + puller_cfg = DeploymentConfig( + name="puller-cfg", + workspace="itest", + restart_policy="Never", # ty: ignore[unknown-argument] + containers=[Container(name="puller", image="alpine:3.20", command=["sh", "-c"], args=["echo pulled"])], + ) + server_cfg = DeploymentConfig( + name="server-cfg", + workspace="itest", + restart_policy="Always", # ty: ignore[unknown-argument] + containers=[ + Container(name="server", image="nginx:alpine", ports=[ContainerPort(name="http", containerPort=80)]) + ], + ) + + config_cache = { + ("itest", "puller-cfg"): puller_cfg, + ("itest", "server-cfg"): server_cfg, + } + + async def get_side_effect( + entity_type: type, + name: str, + workspace: str | None = None, + ) -> Deployment | DeploymentConfig: + if entity_type is Deployment: + if name == "puller": + return puller_dep + if name == "server": + return server_dep + raise KeyError(name) + if entity_type is DeploymentConfig: + return config_cache[(workspace or "itest", name)] + raise KeyError(name) + + entities.get.side_effect = get_side_effect + + backend = _backend(k8s_registry) + backend._entities = entities + + deployment_reconciler = DeploymentReconciler( + entities, + k8s_registry, + ControllerConfig(interval_seconds=1), + ) + deployment_reconciler.set_config_cache(config_cache) + + by_name = {("itest", "puller"): puller_dep, ("itest", "server"): server_dep} + + try: + # The "succeeded" prerequisite the server depends on requires both status and exit_code + # (see prerequisite.py); reconcile_one stops polling the backend once status is terminal, so + # wait for both together rather than breaking on status alone. + for _ in range(POLL_ATTEMPTS): + await deployment_reconciler.reconcile_one(puller_dep, deployments_by_name=by_name, volumes_by_name={}) + if puller_dep.status == "SUCCEEDED" and puller_dep.exit_code == 0: + break + await asyncio.sleep(POLL_INTERVAL_SECONDS) + assert puller_dep.status == "SUCCEEDED" + assert puller_dep.exit_code == 0 + + for _ in range(POLL_ATTEMPTS): + await deployment_reconciler.reconcile_one(server_dep, deployments_by_name=by_name, volumes_by_name={}) + if server_dep.status in ("READY", "STARTING"): + break + await asyncio.sleep(POLL_INTERVAL_SECONDS) + assert server_dep.status in ("READY", "STARTING") + finally: + results = await asyncio.gather( + backend.delete_deployment("itest", "puller"), + backend.delete_deployment("itest", "server"), + return_exceptions=True, + ) + for result in results: + if isinstance(result, Exception): + raise result