+
+
+
+
`,
footer: `
@@ -302,7 +306,8 @@ const ProjectsPage = (() => {
btn.innerHTML = '
Creating…';
try {
- await API.projects.add({ name, owner, scope, data_perimeter: perimeter });
+ const batchEnabled = document.getElementById('new-proj-batch').checked;
+ await API.projects.add({ name, owner, scope, data_perimeter: perimeter, batch_enabled: batchEnabled });
close();
Toast.success(`Project "${name}" created.`);
loadProjects();
diff --git a/infrastructure/k8s/backend-deployment.yaml b/infrastructure/k8s/backend-deployment.yaml
index d48bd73..3c9ec65 100644
--- a/infrastructure/k8s/backend-deployment.yaml
+++ b/infrastructure/k8s/backend-deployment.yaml
@@ -72,6 +72,9 @@ rules:
- apiGroups: [""]
resources: ["pods"]
verbs: [ "get", "list", "delete" ]
+- apiGroups: [""]
+ resources: ["pods/log"]
+ verbs: [ "get" ]
- apiGroups: [""]
resources: ["services"]
verbs: [ "get", "list", "create", "update", "patch", "delete" ]
diff --git a/infrastructure/k8s/minio-deployment.yaml b/infrastructure/k8s/minio-deployment.yaml
index daf0f8e..844c8f4 100644
--- a/infrastructure/k8s/minio-deployment.yaml
+++ b/infrastructure/k8s/minio-deployment.yaml
@@ -32,6 +32,8 @@ spec:
args:
- server
- /data
+ - --console-address
+ - ":9001"
env:
- name: MINIO_ROOT_USER
value: "minio_user"
@@ -39,6 +41,7 @@ spec:
value: "minio_password"
ports:
- containerPort: 9000
+ - containerPort: 9001
volumeMounts:
- name: minio-data
mountPath: /data
@@ -59,6 +62,9 @@ spec:
- name: api
port: 9000
targetPort: 9000
+ - name: console
+ port: 9001
+ targetPort: 9001
---
apiVersion: batch/v1
kind: Job
diff --git a/pyproject.toml b/pyproject.toml
index a8c29d5..dee2125 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -51,6 +51,7 @@ dev = [
"pytest-asyncio>=0.25.3",
"pytest-mock>=3.14.0",
"pytest-playwright",
+ "pytest-order>=1.2.0",
]
notebooks = [
"pyomo>=6.9.0",
diff --git a/tests/conftest.py b/tests/conftest.py
index 5c32f24..744e65d 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -55,13 +55,28 @@ def cleanup_project(project_name: str) -> None:
# Try CLI deletion first (this cleans DB + namespace)
run_cli("projects", "delete", project_name)
- # Force delete namespace if it still exists
+ # Delete namespace and wait for it to be fully removed
subprocess.run(
- ["kubectl", "delete", "namespace", project_name, "--ignore-not-found", "--wait=false"],
+ ["kubectl", "delete", "namespace", project_name, "--ignore-not-found"],
capture_output=True,
text=True,
+ timeout=120,
)
+ # Poll until namespace is actually gone (handles edge cases)
+ deadline = time.time() + 120
+ while time.time() < deadline:
+ result = subprocess.run(
+ ["kubectl", "get", "namespace", project_name],
+ capture_output=True,
+ text=True,
+ timeout=10,
+ )
+ if result.returncode != 0:
+ return
+ time.sleep(5)
+ print(f"[WARNING] Namespace {project_name} still exists after 120s cleanup timeout")
+
def cleanup_test_namespaces(project_names: list[str] | None = None) -> None:
"""Cleanup all test namespaces."""
diff --git a/tests/test_unitaires/adapters/test_k8s_batch_prediction_adapter.py b/tests/test_unitaires/adapters/test_k8s_batch_prediction_adapter.py
new file mode 100644
index 0000000..3cc7aa9
--- /dev/null
+++ b/tests/test_unitaires/adapters/test_k8s_batch_prediction_adapter.py
@@ -0,0 +1,129 @@
+# Philippe Stepniewski
+import os
+from datetime import datetime, timezone
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+os.environ.setdefault("PATH_LOG_EVENTS", "/tmp/test_log_events")
+os.environ.setdefault("MP_HOST_NAME", "localhost")
+os.environ.setdefault("MP_DEPLOYMENT_PATH", "/deploy")
+os.environ.setdefault("MP_DEPLOYMENT_PORT", "8000")
+os.environ.setdefault("MLFLOW_S3_ENDPOINT_URL", "http://localhost:9000")
+os.environ.setdefault("AWS_ACCESS_KEY_ID", "minio_user")
+os.environ.setdefault("AWS_SECRET_ACCESS_KEY", "minio_password")
+
+
+from backend.domain.entities.batch_prediction import BatchPredictionStatus
+
+
+@pytest.fixture
+def mock_k8s():
+ with (
+ patch("backend.infrastructure.k8s_deployment.config") as mock_config,
+ patch("backend.infrastructure.k8s_deployment.client") as mock_base_client,
+ patch("backend.infrastructure.k8s_batch_prediction_adapter.client") as mock_client,
+ ):
+ mock_config.load_kube_config.return_value = None
+ mock_base_client.CoreV1Api.return_value = MagicMock()
+ mock_base_client.AppsV1Api.return_value = MagicMock()
+ mock_base_client.NetworkingV1Api.return_value = MagicMock()
+ mock_batch_api = MagicMock()
+ mock_client.BatchV1Api.return_value = mock_batch_api
+ mock_client.V1Job = MagicMock()
+ mock_client.V1ObjectMeta = MagicMock()
+ mock_client.V1JobSpec = MagicMock()
+ mock_client.V1PodTemplateSpec = MagicMock()
+ mock_client.V1PodSpec = MagicMock()
+ mock_client.V1Container = MagicMock()
+ mock_client.V1EnvVar = MagicMock()
+ mock_client.V1DeleteOptions = MagicMock()
+
+ yield mock_batch_api, mock_client
+
+
+def test_create_batch_job(mock_k8s):
+ mock_batch_api, mock_client = mock_k8s
+
+ from backend.infrastructure.k8s_batch_prediction_adapter import K8sBatchPredictionAdapter
+
+ adapter = K8sBatchPredictionAdapter()
+
+ result = adapter.create_batch_job(
+ project_name="test-project",
+ model_name="my-model",
+ model_version="1",
+ input_path="test-project/my-model/1/abc/input.csv",
+ output_path="test-project/my-model/1/abc/predictions-abc.csv",
+ job_id="abc12345",
+ )
+
+ mock_batch_api.create_namespaced_job.assert_called_once()
+ assert result.project_name == "test-project"
+ assert result.model_name == "my-model"
+ assert result.model_version == "1"
+ assert result.status == BatchPredictionStatus.PENDING
+
+
+def test_map_job_status_succeeded(mock_k8s):
+ from backend.infrastructure.k8s_batch_prediction_adapter import K8sBatchPredictionAdapter
+
+ adapter = K8sBatchPredictionAdapter()
+
+ status = MagicMock()
+ status.succeeded = 1
+ status.failed = None
+ status.active = None
+ assert adapter._map_job_status(status) == BatchPredictionStatus.COMPLETED
+
+
+def test_map_job_status_failed(mock_k8s):
+ from backend.infrastructure.k8s_batch_prediction_adapter import K8sBatchPredictionAdapter
+
+ adapter = K8sBatchPredictionAdapter()
+
+ status = MagicMock()
+ status.succeeded = None
+ status.failed = 1
+ status.active = None
+ assert adapter._map_job_status(status) == BatchPredictionStatus.FAILED
+
+
+def test_map_job_status_active(mock_k8s):
+ from backend.infrastructure.k8s_batch_prediction_adapter import K8sBatchPredictionAdapter
+
+ adapter = K8sBatchPredictionAdapter()
+
+ status = MagicMock()
+ status.succeeded = None
+ status.failed = None
+ status.active = 1
+ assert adapter._map_job_status(status) == BatchPredictionStatus.RUNNING
+
+
+def test_map_job_status_pending(mock_k8s):
+ from backend.infrastructure.k8s_batch_prediction_adapter import K8sBatchPredictionAdapter
+
+ adapter = K8sBatchPredictionAdapter()
+
+ status = MagicMock()
+ status.succeeded = None
+ status.failed = None
+ status.active = None
+ assert adapter._map_job_status(status) == BatchPredictionStatus.PENDING
+
+
+def test_delete_batch_job(mock_k8s):
+ mock_batch_api, mock_client = mock_k8s
+
+ from backend.infrastructure.k8s_batch_prediction_adapter import K8sBatchPredictionAdapter
+
+ adapter = K8sBatchPredictionAdapter()
+
+ result = adapter.delete_batch_job("test-project", "my-job-id")
+
+ mock_batch_api.delete_namespaced_job.assert_called_once()
+ call_kwargs = mock_batch_api.delete_namespaced_job.call_args
+ assert call_kwargs[1]["name"] == "batch-my-job-id"
+ assert call_kwargs[1]["namespace"] == "test-project"
+ assert result is True
diff --git a/tests/test_unitaires/adapters/test_minio_storage_adapter.py b/tests/test_unitaires/adapters/test_minio_storage_adapter.py
new file mode 100644
index 0000000..4f11ef5
--- /dev/null
+++ b/tests/test_unitaires/adapters/test_minio_storage_adapter.py
@@ -0,0 +1,61 @@
+# Philippe Stepniewski
+from unittest.mock import MagicMock, patch
+
+import pytest
+from botocore.exceptions import ClientError
+
+from backend.infrastructure.minio_storage_adapter import BATCH_BUCKET, MinioStorageAdapter
+
+
+@pytest.fixture
+def adapter():
+ with patch("backend.infrastructure.minio_storage_adapter.boto3") as mock_boto3:
+ mock_client = MagicMock()
+ mock_boto3.client.return_value = mock_client
+ with patch.dict("os.environ", {"MLFLOW_S3_ENDPOINT_URL": "http://minio:9000"}):
+ a = MinioStorageAdapter()
+ yield a, mock_client
+
+
+def test_ensure_project_space_creates_bucket(adapter):
+ a, mock_client = adapter
+ mock_client.head_bucket.side_effect = ClientError({"Error": {"Code": "404"}}, "HeadBucket")
+
+ a.ensure_project_space("my-project")
+
+ mock_client.create_bucket.assert_called_once_with(Bucket=BATCH_BUCKET)
+
+
+def test_ensure_project_space_existing_bucket(adapter):
+ a, mock_client = adapter
+
+ a.ensure_project_space("my-project")
+
+ mock_client.create_bucket.assert_not_called()
+
+
+def test_remove_project_space_deletes_all_objects(adapter):
+ a, mock_client = adapter
+ paginator = MagicMock()
+ mock_client.get_paginator.return_value = paginator
+ paginator.paginate.return_value = [
+ {"Contents": [{"Key": "my-project/.keep"}, {"Key": "my-project/data.csv"}]},
+ ]
+
+ a.remove_project_space("my-project")
+
+ mock_client.delete_objects.assert_called_once_with(
+ Bucket=BATCH_BUCKET,
+ Delete={"Objects": [{"Key": "my-project/.keep"}, {"Key": "my-project/data.csv"}]},
+ )
+
+
+def test_remove_project_space_no_objects(adapter):
+ a, mock_client = adapter
+ paginator = MagicMock()
+ mock_client.get_paginator.return_value = paginator
+ paginator.paginate.return_value = [{"Contents": []}]
+
+ a.remove_project_space("my-project")
+
+ mock_client.delete_objects.assert_not_called()
diff --git a/tests/test_unitaires/use_cases/test_batch_predict.py b/tests/test_unitaires/use_cases/test_batch_predict.py
new file mode 100644
index 0000000..cee9374
--- /dev/null
+++ b/tests/test_unitaires/use_cases/test_batch_predict.py
@@ -0,0 +1,236 @@
+# Philippe Stepniewski
+import os
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+os.environ.setdefault("PATH_LOG_EVENTS", "/tmp/test_log_events")
+
+from backend.domain.entities.batch_prediction import BatchPrediction, BatchPredictionStatus
+from backend.domain.entities.project import Project
+from backend.domain.use_cases.batch_predict import (
+ delete_batch_prediction,
+ download_batch_result,
+ get_batch_prediction_status,
+ list_batch_predictions,
+ submit_batch_prediction,
+)
+
+
+@pytest.fixture
+def mock_batch_handler():
+ return MagicMock()
+
+
+@pytest.fixture
+def mock_object_storage():
+ return MagicMock()
+
+
+@pytest.fixture
+def mock_project_db_handler():
+ handler = MagicMock()
+ handler.get_project.return_value = Project(
+ name="test-project", owner="owner", scope="scope", data_perimeter="perimeter", batch_enabled=True
+ )
+ return handler
+
+
+@pytest.fixture
+def mock_registry():
+ return MagicMock()
+
+
+@pytest.fixture
+def sample_batch_prediction():
+ from datetime import datetime, timezone
+
+ return BatchPrediction(
+ job_id="abc12345",
+ project_name="test-project",
+ model_name="my-model",
+ model_version="1",
+ status=BatchPredictionStatus.PENDING,
+ input_path="test-project/my-model/1/abc12345/input.csv",
+ output_path="test-project/my-model/1/abc12345/predictions-abc12345.csv",
+ created_at=datetime.now(timezone.utc),
+ )
+
+
+def test_submit_uploads_file_and_creates_job(
+ mock_batch_handler, mock_object_storage, mock_project_db_handler, sample_batch_prediction
+):
+ mock_batch_handler.create_batch_job.return_value = sample_batch_prediction
+
+ result = submit_batch_prediction(
+ project_name="test-project",
+ model_name="my-model",
+ version="1",
+ file_content=b"col1,col2\n1,2\n3,4",
+ job_id="abc12345",
+ object_storage=mock_object_storage,
+ batch_handler=mock_batch_handler,
+ project_db_handler=mock_project_db_handler,
+ )
+
+ mock_object_storage.upload_file.assert_called_once()
+ mock_batch_handler.create_batch_job.assert_called_once()
+ assert result["project_name"] == "test-project"
+ assert result["model_name"] == "my-model"
+
+
+def test_submit_fails_if_batch_not_enabled(mock_batch_handler, mock_object_storage):
+ project_db = MagicMock()
+ project_db.get_project.return_value = Project(
+ name="test-project", owner="owner", scope="scope", data_perimeter="perimeter", batch_enabled=False
+ )
+
+ from fastapi import HTTPException
+
+ with pytest.raises(HTTPException) as exc_info:
+ submit_batch_prediction(
+ project_name="test-project",
+ model_name="my-model",
+ version="1",
+ file_content=b"col1,col2\n1,2",
+ job_id="abc12345",
+ object_storage=mock_object_storage,
+ batch_handler=mock_batch_handler,
+ project_db_handler=project_db,
+ )
+ assert exc_info.value.status_code == 400
+
+
+def test_get_status_delegates_to_handler(mock_batch_handler, sample_batch_prediction):
+ mock_batch_handler.get_job_status.return_value = sample_batch_prediction
+
+ result = get_batch_prediction_status("test-project", "abc12345", mock_batch_handler)
+
+ mock_batch_handler.get_job_status.assert_called_once_with("test-project", "abc12345")
+ assert result["job_id"] == "abc12345"
+ assert result["status"] == "pending"
+
+
+def test_list_delegates_to_handler(mock_batch_handler, sample_batch_prediction):
+ mock_batch_handler.list_batch_jobs.return_value = [sample_batch_prediction]
+
+ result = list_batch_predictions("test-project", mock_batch_handler)
+
+ mock_batch_handler.list_batch_jobs.assert_called_once_with("test-project")
+ assert len(result) == 1
+ assert result[0]["job_id"] == "abc12345"
+
+
+def test_download_returns_file_content(mock_batch_handler, mock_object_storage, sample_batch_prediction):
+ mock_batch_handler.get_job_status.return_value = sample_batch_prediction
+ mock_object_storage.download_file.return_value = b"prediction\n0.95\n0.32"
+
+ result = download_batch_result("test-project", "abc12345", mock_batch_handler, mock_object_storage)
+
+ mock_object_storage.download_file.assert_called_once_with(
+ "test-project", "my-model/1/abc12345/predictions-abc12345.csv"
+ )
+ assert result == b"prediction\n0.95\n0.32"
+
+
+def test_delete_cleans_up_job_and_storage(mock_batch_handler, mock_object_storage, sample_batch_prediction):
+ mock_batch_handler.get_job_status.return_value = sample_batch_prediction
+ mock_batch_handler.delete_batch_job.return_value = True
+ mock_object_storage.list_files.return_value = [
+ "my-model/1/abc12345/input.csv",
+ "my-model/1/abc12345/predictions-abc12345.csv",
+ ]
+
+ result = delete_batch_prediction("test-project", "abc12345", mock_batch_handler, mock_object_storage)
+
+ assert result is True
+ mock_batch_handler.delete_batch_job.assert_called_once_with("test-project", "abc12345")
+ assert mock_object_storage.delete_file.call_count == 2
+
+
+@patch("backend.domain.use_cases.batch_predict.build_model_docker_image")
+@patch("backend.domain.use_cases.batch_predict.check_docker_image_exists")
+def test_submit_builds_image_if_not_exists(
+ mock_check,
+ mock_build,
+ mock_batch_handler,
+ mock_object_storage,
+ mock_project_db_handler,
+ mock_registry,
+ sample_batch_prediction,
+):
+ mock_check.return_value = False
+ mock_build.return_value = 1
+ mock_batch_handler.create_batch_job.return_value = sample_batch_prediction
+
+ submit_batch_prediction(
+ project_name="test-project",
+ model_name="my-model",
+ version="1",
+ file_content=b"col1,col2\n1,2",
+ job_id="abc12345",
+ object_storage=mock_object_storage,
+ batch_handler=mock_batch_handler,
+ project_db_handler=mock_project_db_handler,
+ registry=mock_registry,
+ )
+
+ mock_check.assert_called_once()
+ mock_build.assert_called_once_with(mock_registry, "test-project", "my-model", "1")
+
+
+@patch("backend.domain.use_cases.batch_predict.build_model_docker_image")
+@patch("backend.domain.use_cases.batch_predict.check_docker_image_exists")
+def test_submit_skips_build_if_image_exists(
+ mock_check,
+ mock_build,
+ mock_batch_handler,
+ mock_object_storage,
+ mock_project_db_handler,
+ mock_registry,
+ sample_batch_prediction,
+):
+ mock_check.return_value = True
+ mock_batch_handler.create_batch_job.return_value = sample_batch_prediction
+
+ submit_batch_prediction(
+ project_name="test-project",
+ model_name="my-model",
+ version="1",
+ file_content=b"col1,col2\n1,2",
+ job_id="abc12345",
+ object_storage=mock_object_storage,
+ batch_handler=mock_batch_handler,
+ project_db_handler=mock_project_db_handler,
+ registry=mock_registry,
+ )
+
+ mock_check.assert_called_once()
+ mock_build.assert_not_called()
+
+
+@patch("backend.domain.use_cases.batch_predict.build_model_docker_image")
+@patch("backend.domain.use_cases.batch_predict.check_docker_image_exists")
+def test_submit_fails_if_build_fails(
+ mock_check, mock_build, mock_batch_handler, mock_object_storage, mock_project_db_handler, mock_registry
+):
+ mock_check.return_value = False
+ mock_build.return_value = 0
+
+ from fastapi import HTTPException
+
+ with pytest.raises(HTTPException) as exc_info:
+ submit_batch_prediction(
+ project_name="test-project",
+ model_name="my-model",
+ version="1",
+ file_content=b"col1,col2\n1,2",
+ job_id="abc12345",
+ object_storage=mock_object_storage,
+ batch_handler=mock_batch_handler,
+ project_db_handler=mock_project_db_handler,
+ registry=mock_registry,
+ )
+
+ assert exc_info.value.status_code == 500
+ assert "Failed to build model image" in exc_info.value.detail
diff --git a/tests/test_unitaires/use_cases/test_projects_usecases.py b/tests/test_unitaires/use_cases/test_projects_usecases.py
new file mode 100644
index 0000000..16c1e93
--- /dev/null
+++ b/tests/test_unitaires/use_cases/test_projects_usecases.py
@@ -0,0 +1,85 @@
+# Philippe Stepniewski
+import os
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+# Set required env var before importing the module that needs it
+os.environ.setdefault("PATH_LOG_EVENTS", "/tmp/test_log_events")
+
+from backend.domain.entities.project import Project
+from backend.domain.use_cases.projects_usecases import add_project, remove_project, update_project_batch_enabled
+
+
+@pytest.fixture
+def mock_project_db_handler():
+ handler = MagicMock()
+ handler.add_project.return_value = True
+ handler.remove_project.return_value = True
+ handler.update_batch_enabled.return_value = True
+ return handler
+
+
+@pytest.fixture
+def mock_object_storage():
+ return MagicMock()
+
+
+@patch("backend.domain.use_cases.projects_usecases.deploy_registry")
+def test_add_project_with_batch_enabled_creates_storage_space(
+ mock_deploy_registry, mock_project_db_handler, mock_object_storage
+):
+ project = Project(name="test-project", owner="owner", scope="scope", data_perimeter="perimeter", batch_enabled=True)
+
+ add_project(mock_project_db_handler, project, mock_object_storage)
+
+ mock_object_storage.ensure_project_space.assert_called_once_with("test-project")
+ mock_project_db_handler.add_project.assert_called_once_with(project)
+
+
+@patch("backend.domain.use_cases.projects_usecases.deploy_registry")
+def test_add_project_without_batch_does_not_create_storage_space(
+ mock_deploy_registry, mock_project_db_handler, mock_object_storage
+):
+ project = Project(
+ name="test-project", owner="owner", scope="scope", data_perimeter="perimeter", batch_enabled=False
+ )
+
+ add_project(mock_project_db_handler, project, mock_object_storage)
+
+ mock_object_storage.ensure_project_space.assert_not_called()
+ mock_project_db_handler.add_project.assert_called_once_with(project)
+
+
+@patch("backend.domain.use_cases.projects_usecases._remove_project_namespace")
+def test_remove_project_cleans_up_storage(mock_remove_ns, mock_project_db_handler, mock_object_storage):
+ remove_project(mock_project_db_handler, "test-project", mock_object_storage)
+
+ mock_object_storage.remove_project_space.assert_called_once_with("test-project")
+ mock_project_db_handler.remove_project.assert_called_once_with("test-project")
+
+
+@patch("backend.domain.use_cases.projects_usecases._remove_project_namespace")
+def test_remove_project_continues_if_storage_cleanup_fails(
+ mock_remove_ns, mock_project_db_handler, mock_object_storage
+):
+ mock_object_storage.remove_project_space.side_effect = Exception("Storage error")
+
+ result = remove_project(mock_project_db_handler, "test-project", mock_object_storage)
+
+ assert result is True
+ mock_project_db_handler.remove_project.assert_called_once_with("test-project")
+
+
+def test_update_batch_enabled_to_true_creates_space(mock_project_db_handler, mock_object_storage):
+ update_project_batch_enabled(mock_project_db_handler, "test-project", True, mock_object_storage)
+
+ mock_object_storage.ensure_project_space.assert_called_once_with("test-project")
+ mock_project_db_handler.update_batch_enabled.assert_called_once_with("test-project", True)
+
+
+def test_update_batch_enabled_to_false_removes_space(mock_project_db_handler, mock_object_storage):
+ update_project_batch_enabled(mock_project_db_handler, "test-project", False, mock_object_storage)
+
+ mock_object_storage.remove_project_space.assert_called_once_with("test-project")
+ mock_project_db_handler.update_batch_enabled.assert_called_once_with("test-project", False)
diff --git a/tests/tests_end_to_end/test_batch_prediction_e2e.py b/tests/tests_end_to_end/test_batch_prediction_e2e.py
new file mode 100644
index 0000000..3370cbc
--- /dev/null
+++ b/tests/tests_end_to_end/test_batch_prediction_e2e.py
@@ -0,0 +1,281 @@
+# Philippe Stepniewski
+"""
+End-to-end test: Batch prediction workflow.
+
+This test covers:
+1. Project creation with batch enabled
+2. Model training and push to MLflow
+3. Batch prediction submission (triggers auto-build of Docker image)
+4. Status polling until completion
+5. Result download and validation
+6. Cleanup (delete job + files + project)
+"""
+
+import os
+import random
+import string
+import subprocess
+import sys
+import time
+
+import mlflow
+import mlflow.sklearn
+import pytest
+import requests
+from sklearn.datasets import load_iris
+from sklearn.ensemble import RandomForestClassifier
+from sklearn.model_selection import train_test_split
+
+from tests.conftest import MP_HOSTNAME, cleanup_project, login, run_cli
+
+_this_module = sys.modules[__name__]
+
+PROJECT_SUFFIX = "".join(random.choices(string.ascii_lowercase, k=6))
+PROJECT_NAME = f"e2ebatch{PROJECT_SUFFIX}"
+MODEL_NAME = "batch_test_model"
+MODEL_VERSION = "1"
+
+
+def _setup_minikube_docker_env():
+ try:
+ result = subprocess.run(
+ ["minikube", "docker-env", "--shell", "bash"],
+ capture_output=True,
+ text=True,
+ timeout=30,
+ )
+ if result.returncode == 0:
+ env_lines = [line.strip() for line in result.stdout.split("\n") if line.startswith("export ")]
+ for line in env_lines:
+ var_assignment = line.replace("export ", "", 1)
+ if "=" in var_assignment:
+ key, value = var_assignment.split("=", 1)
+ value = value.strip("\"'")
+ os.environ[key] = value
+ except Exception as exc:
+ print(f"[DEBUG] Error setting up minikube docker env: {exc}")
+
+
+def _run_debug_cmd(label, cmd):
+ print(f"[DEBUG] {label}: {' '.join(cmd)}")
+ try:
+ result = subprocess.run(cmd, capture_output=True, text=True, timeout=40)
+ print(f"[DEBUG] {label} exit={result.returncode}")
+ if result.stdout:
+ print(f"[DEBUG] {label} stdout:\n{result.stdout}")
+ if result.stderr:
+ print(f"[DEBUG] {label} stderr:\n{result.stderr}")
+ except Exception as exc: # noqa: BLE001
+ print(f"[DEBUG] {label} error: {exc}")
+
+
+def _dump_batch_debug_info():
+ _run_debug_cmd("kubectl get jobs", ["kubectl", "get", "jobs", "-n", PROJECT_NAME, "-o", "wide"])
+ _run_debug_cmd("kubectl get pods", ["kubectl", "get", "pods", "-n", PROJECT_NAME, "-o", "wide"])
+ _run_debug_cmd(
+ "kubectl get events",
+ ["kubectl", "get", "events", "-n", PROJECT_NAME, "--sort-by=.metadata.creationTimestamp"],
+ )
+ result = subprocess.run(
+ ["kubectl", "get", "pods", "-n", PROJECT_NAME, "-l", "app=batch-prediction", "--no-headers"],
+ capture_output=True,
+ text=True,
+ timeout=20,
+ )
+ if result.returncode == 0 and result.stdout.strip():
+ for line in result.stdout.strip().splitlines():
+ pod_name = line.split()[0]
+ _run_debug_cmd(f"kubectl logs {pod_name}", ["kubectl", "logs", pod_name, "-n", PROJECT_NAME, "--tail=50"])
+
+
+@pytest.fixture(scope="module", autouse=True)
+def setup_and_teardown():
+ print(f"[DEBUG] Setting up batch e2e test with project {PROJECT_NAME}")
+ _setup_minikube_docker_env()
+ assert login() == 0, "Login failed"
+
+ yield
+
+ cleanup_project(PROJECT_NAME)
+
+
+# ── Track readiness (set via module attribute for robustness) ────
+_mlflow_ready = False
+
+
+def _skip_if_mlflow_not_ready():
+ if not getattr(_this_module, "_mlflow_ready", False):
+ pytest.skip("MLflow registry not ready")
+
+
+# ── Tests (ordered) ──────────────────────────────────────────────
+
+
+@pytest.mark.order(1)
+def test_create_project_with_batch():
+ result = run_cli("projects", "add", "--name", PROJECT_NAME, "--batch-enabled")
+ assert result.returncode == 0, f"Project creation failed: {result.stderr}"
+ assert "Project created successfully" in result.stdout
+
+
+@pytest.mark.order(2)
+def test_mlflow_registry_responds():
+ registry_url = f"http://{MP_HOSTNAME}/registry/{PROJECT_NAME}/"
+ deadline = time.time() + 300
+ while time.time() < deadline:
+ result = subprocess.run(
+ ["curl", "-s", "-o", "/dev/null", "-w", "%{http_code}", registry_url],
+ capture_output=True,
+ text=True,
+ )
+ if result.stdout == "200":
+ _this_module._mlflow_ready = True
+ return
+ print(f"[DEBUG] MLflow registry HTTP {result.stdout}, retrying in 5s...")
+ time.sleep(5)
+ assert False, "MLflow registry did not respond with 200 within 5 minutes"
+
+
+@pytest.mark.order(3)
+def test_train_and_push_model():
+ _skip_if_mlflow_not_ready()
+ api_url = f"http://{MP_HOSTNAME}/registry/{PROJECT_NAME}/api/2.0/mlflow/experiments/search"
+ response = requests.post(api_url, json={"max_results": 1}, timeout=10)
+ assert response.status_code == 200, f"MLflow API not ready: {response.status_code}"
+
+ data = load_iris()
+ x_train, _, y_train, _ = train_test_split(data.data, data.target, test_size=0.3, random_state=42)
+
+ mlflow.set_tracking_uri(f"http://{MP_HOSTNAME}/registry/{PROJECT_NAME}/")
+
+ with mlflow.start_run():
+ model = RandomForestClassifier(n_estimators=2, random_state=42)
+ model.fit(x_train, y_train)
+ mlflow.sklearn.log_model(model, "custom_model", registered_model_name=MODEL_NAME)
+
+ deadline = time.time() + 60
+ while time.time() < deadline:
+ result = run_cli("projects", "list-models", PROJECT_NAME)
+ if result.returncode == 0 and MODEL_NAME in result.stdout:
+ return
+ time.sleep(5)
+ assert False, f"Model {MODEL_NAME} not found after 60s"
+
+
+@pytest.mark.order(4)
+def test_submit_batch_prediction():
+ """Submit a batch prediction via the API. This triggers auto-build of the Docker image."""
+ _skip_if_mlflow_not_ready()
+
+ # Create a small CSV matching the iris model (4 features)
+ csv_content = "0,1,2,3\n5.1,3.5,1.4,0.2\n6.2,2.9,4.3,1.3\n7.1,3.0,5.9,2.1\n"
+ csv_path = f"/tmp/batch_e2e_test_{PROJECT_NAME}.csv"
+ with open(csv_path, "w") as f:
+ f.write(csv_content)
+
+ result = run_cli("batch", "submit", PROJECT_NAME, MODEL_NAME, MODEL_VERSION, "--file-path", csv_path)
+ assert result.returncode == 0, f"Batch submit failed: {result.stderr}\n{result.stdout}"
+ assert "Job ID:" in result.stdout or "job_id" in result.stdout
+
+ # Extract job_id from output
+ for line in result.stdout.splitlines():
+ if "Job ID:" in line:
+ _this_module._batch_job_id = line.split("Job ID:")[-1].strip()
+ print(f"[DEBUG] Batch job submitted with ID: {_this_module._batch_job_id}")
+ return
+
+ assert False, f"Could not extract job ID from output: {result.stdout}"
+
+
+_batch_job_id = None
+
+
+def _skip_if_no_job():
+ if not getattr(_this_module, "_batch_job_id", None):
+ pytest.skip("No batch job ID available")
+
+
+@pytest.mark.order(5)
+def test_batch_job_completes():
+ """Poll batch job status until it completes (building → pending → running → completed)."""
+ _skip_if_mlflow_not_ready()
+ _skip_if_no_job()
+
+ deadline = time.time() + 600 # 10 minutes — includes Docker image build
+ last_status = None
+ while time.time() < deadline:
+ result = run_cli("batch", "status", PROJECT_NAME, _this_module._batch_job_id)
+ output = result.stdout.lower()
+
+ if "completed" in output:
+ print(f"[DEBUG] Batch job {_this_module._batch_job_id} completed")
+ return
+
+ if "failed" in output:
+ print(f"[DEBUG] Batch job {_this_module._batch_job_id} FAILED")
+ _dump_batch_debug_info()
+ assert False, f"Batch job failed: {result.stdout}"
+
+ # Extract status for debug
+ for status_word in ["building", "pending", "running"]:
+ if status_word in output:
+ if status_word != last_status:
+ print(f"[DEBUG] Batch job status: {status_word}")
+ last_status = status_word
+ break
+
+ time.sleep(10)
+
+ _dump_batch_debug_info()
+ assert False, f"Batch job did not complete within 10 minutes. Last status: {last_status}"
+
+
+@pytest.mark.order(6)
+def test_download_batch_result():
+ """Download the batch prediction result and validate it."""
+ _skip_if_mlflow_not_ready()
+ _skip_if_no_job()
+
+ output_path = f"/tmp/batch_e2e_result_{PROJECT_NAME}.csv"
+ result = run_cli("batch", "download", PROJECT_NAME, _this_module._batch_job_id, "--output", output_path)
+ assert result.returncode == 0, f"Download failed: {result.stderr}\n{result.stdout}"
+ assert "downloaded" in result.stdout.lower()
+
+ assert os.path.exists(output_path), f"Output file not found at {output_path}"
+ with open(output_path) as f:
+ lines = f.readlines()
+ # Header + 3 prediction rows
+ assert len(lines) >= 4, f"Expected at least 4 lines (header + 3 rows), got {len(lines)}: {lines}"
+ assert "prediction" in lines[0].lower(), f"Expected 'prediction' header, got: {lines[0]}"
+ print(f"[DEBUG] Downloaded {len(lines) - 1} predictions")
+
+
+@pytest.mark.order(7)
+def test_list_batch_jobs():
+ """Verify the job appears in the list."""
+ _skip_if_mlflow_not_ready()
+ _skip_if_no_job()
+
+ result = run_cli("batch", "list", PROJECT_NAME)
+ assert result.returncode == 0, f"List failed: {result.stderr}"
+ assert (
+ _this_module._batch_job_id in result.stdout
+ ), f"Job {_this_module._batch_job_id} not in list output: {result.stdout}"
+
+
+@pytest.mark.order(8)
+def test_delete_batch_job():
+ """Delete the batch job and associated files."""
+ _skip_if_mlflow_not_ready()
+ _skip_if_no_job()
+
+ result = run_cli("batch", "delete", PROJECT_NAME, _this_module._batch_job_id)
+ assert result.returncode == 0, f"Delete failed: {result.stderr}\n{result.stdout}"
+ assert "deleted" in result.stdout.lower()
+
+
+@pytest.mark.order(9)
+def test_delete_project():
+ result = run_cli("projects", "delete", PROJECT_NAME)
+ assert result.returncode == 0, f"Delete failed: {result.stderr}"
+ assert "Project deleted successfully" in result.stdout
diff --git a/tests/tests_end_to_end/test_from_project_creation_to_model_predict.py b/tests/tests_end_to_end/test_from_project_creation_to_model_predict.py
index b3af7a5..e15857e 100644
--- a/tests/tests_end_to_end/test_from_project_creation_to_model_predict.py
+++ b/tests/tests_end_to_end/test_from_project_creation_to_model_predict.py
@@ -13,6 +13,7 @@
import random
import string
import subprocess
+import sys
import time
import os
@@ -27,6 +28,8 @@
from backend.utils import sanitize_ressource_name
from tests.conftest import cleanup_project, login, MP_HOSTNAME, run_cli
+_this_module = sys.modules[__name__]
+
# Use random suffix to avoid conflicts with previous test runs (db dropper jobs, etc.)
PROJECT_SUFFIX = "".join(random.choices(string.ascii_lowercase, k=6))
PROJECT_NAME = f"e2e{PROJECT_SUFFIX}"
@@ -59,6 +62,7 @@ def setup_and_teardown():
cleanup_project(PROJECT_NAME)
+@pytest.mark.order(1)
def test_health_endpoint_responds():
"""Test that the platform health endpoint responds."""
result = subprocess.run(
@@ -69,13 +73,15 @@ def test_health_endpoint_responds():
assert result.stdout == "200", f"Health endpoint did not respond with 200: {result.stdout}"
+@pytest.mark.order(2)
def test_project_creation():
"""Test project creation."""
result = run_cli("projects", "add", "--name", PROJECT_NAME)
assert result.returncode == 0, f"Project creation failed: {result.stderr}"
- assert "✅ Project created successfully" in result.stdout
+ assert "Project created successfully" in result.stdout
+@pytest.mark.order(3)
def test_mlflow_registry_responds():
"""Test that MLflow registry responds after project creation."""
registry_url = f"http://{MP_HOSTNAME}/registry/{PROJECT_NAME}/"
@@ -87,8 +93,7 @@ def test_mlflow_registry_responds():
text=True,
)
if result.stdout == "200":
- global _mlflow_ready
- _mlflow_ready = True
+ _this_module._mlflow_ready = True
return
print(f"[DEBUG] MLflow registry HTTP {result.stdout}, retrying in 5s...")
time.sleep(5)
@@ -96,12 +101,12 @@ def test_mlflow_registry_responds():
assert False, f"MLflow registry did not respond with 200 within 5 minutes"
-# Flag to track if MLflow is ready
+# Flag to track if MLflow is ready (set via module attribute for robustness)
_mlflow_ready = False
def _skip_if_mlflow_not_ready():
- if not _mlflow_ready:
+ if not getattr(_this_module, "_mlflow_ready", False):
pytest.skip("MLflow registry not ready - skipping dependent test")
@@ -223,6 +228,7 @@ def _dump_registry_status():
_run_debug_cmd("curl registry index", ["curl", "-v", "--max-time", "20", registry_url])
+@pytest.mark.order(4)
def test_train_and_push_model_to_mlflow():
"""Test model training and push to MLflow."""
_skip_if_mlflow_not_ready()
@@ -251,6 +257,7 @@ def test_train_and_push_model_to_mlflow():
assert False, f"Model {MODEL_NAME} not found after 60s. Last output: {result.stdout}"
+@pytest.mark.order(5)
def test_deploy_model():
"""Test model deployment."""
_skip_if_mlflow_not_ready()
@@ -284,6 +291,7 @@ def test_deploy_model():
assert False, f"Deployment {deployment_name} was not created within 5 minutes"
+@pytest.mark.order(6)
def test_deployed_model_health_check():
"""Test that deployed model responds to health check."""
_skip_if_mlflow_not_ready()
@@ -334,6 +342,7 @@ def test_deployed_model_health_check():
raise AssertionError("Model health endpoint did not respond with 200 within 5 minutes")
+@pytest.mark.order(7)
def test_deployed_model_returns_predictions():
"""Test that deployed model returns predictions."""
_skip_if_mlflow_not_ready()
@@ -375,6 +384,7 @@ def test_deployed_model_returns_predictions():
raise AssertionError("Could not get prediction from model within 2 minutes")
+@pytest.mark.order(8)
def test_list_deployed_models():
"""Test that list deployed models shows the deployed model."""
_skip_if_mlflow_not_ready()
@@ -383,6 +393,7 @@ def test_list_deployed_models():
assert MODEL_NAME in result.stdout or MODEL_NAME.replace("_", "-") in result.stdout
+@pytest.mark.order(9)
def test_undeploy_model():
"""Test model undeployment."""
_skip_if_mlflow_not_ready()
@@ -390,6 +401,7 @@ def test_undeploy_model():
assert result.returncode == 0, f"Undeploy failed: {result.stderr}"
+@pytest.mark.order(10)
def test_undeployed_model_is_removed():
"""Test that undeployed model no longer has a K8s deployment."""
_skip_if_mlflow_not_ready()
@@ -409,13 +421,15 @@ def test_undeployed_model_is_removed():
assert False, f"Deployment {deployment_name} should have been deleted after 30s"
+@pytest.mark.order(11)
def test_delete_project():
"""Test project deletion."""
result = run_cli("projects", "delete", PROJECT_NAME)
assert result.returncode == 0, f"Delete failed: {result.stderr}"
- assert "✅ Project deleted successfully" in result.stdout
+ assert "Project deleted successfully" in result.stdout
+@pytest.mark.order(12)
def test_project_registry_is_removed():
"""Test that project registry is no longer accessible after deletion."""
# Poll until registry is gone (instead of flat 60s sleep)
diff --git a/uv.lock b/uv.lock
index ed2ae0f..dd155f7 100644
--- a/uv.lock
+++ b/uv.lock
@@ -2212,6 +2212,7 @@ dev = [
{ name = "pytest-asyncio" },
{ name = "pytest-cov" },
{ name = "pytest-mock" },
+ { name = "pytest-order" },
{ name = "pytest-playwright" },
{ name = "ruff" },
]
@@ -2263,6 +2264,7 @@ dev = [
{ name = "pytest-asyncio", specifier = ">=0.25.3" },
{ name = "pytest-cov", specifier = ">=4.1.0" },
{ name = "pytest-mock", specifier = ">=3.14.0" },
+ { name = "pytest-order", specifier = ">=1.2.0" },
{ name = "pytest-playwright" },
{ name = "ruff", specifier = ">=0.1.4" },
]
@@ -3458,6 +3460,18 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/5a/cc/06253936f4a7fa2e0f48dfe6d851d9c56df896a9ab09ac019d70b760619c/pytest_mock-3.15.1-py3-none-any.whl", hash = "sha256:0a25e2eb88fe5168d535041d09a4529a188176ae608a6d249ee65abc0949630d", size = 10095, upload-time = "2025-09-16T16:37:25.734Z" },
]
+[[package]]
+name = "pytest-order"
+version = "1.3.0"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+ { name = "pytest" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/1d/66/02ae17461b14a52ce5a29ae2900156b9110d1de34721ccc16ccd79419876/pytest_order-1.3.0.tar.gz", hash = "sha256:51608fec3d3ee9c0adaea94daa124a5c4c1d2bb99b00269f098f414307f23dde", size = 47544, upload-time = "2024-08-22T12:29:54.512Z" }
+wheels = [
+ { url = "https://files.pythonhosted.org/packages/1b/73/59b038d1aafca89f8e9936eaa8ffa6bb6138d00459d13a32ce070be4f280/pytest_order-1.3.0-py3-none-any.whl", hash = "sha256:2cd562a21380345dd8d5774aa5fd38b7849b6ee7397ca5f6999bbe6e89f07f6e", size = 14609, upload-time = "2024-08-22T12:29:53.156Z" },
+]
+
[[package]]
name = "pytest-playwright"
version = "0.7.2"