Ml Testing

ML Testing

Overview

ML testing goes beyond traditional software testing to validate data quality, model behavior, and pipeline correctness. A comprehensive test suite catches bugs before they reach production.

ML Testing Pyramid

          ┌─────────────┐
          │   System    │  End-to-end pipeline tests
          │   Tests     │  (slow, expensive)
         ─┤            ├─
        ┌─┴─────────────┴─┐
        │  Integration    │  Component interaction tests
        │  Tests          │
       ─┤                ├─
      ┌─┴─────────────────┴─┐
      │  Model Tests        │  Quality, behavior, regression
      │                     │
     ─┤                    ├─
    ┌─┴─────────────────────┴─┐
    │  Feature / Transform    │  Pipeline correctness
    │  Tests                  │
   ─┤                        ├─
  ┌─┴─────────────────────────┴─┐
  │  Data Tests                 │  Schema, quality, freshness
  │                             │  (fast, cheap)
  └─────────────────────────────┘

Step-by-Step Instructions

1. Data Tests

import pytest
import pandas as pd
import pandera as pa

class TestDataQuality:
    @pytest.fixture
    def training_data(self):
        return pd.read_parquet("data/train.parquet")

    def test_no_nulls_in_required_columns(self, training_data):
        required = ["user_id", "target", "timestamp"]
        for col in required:
            assert training_data[col].isnull().sum() == 0, f"Nulls in {col}"

    def test_no_duplicate_ids(self, training_data):
        assert training_data["user_id"].is_unique

    def test_target_distribution(self, training_data):
        """Target should not be severely imbalanced."""
        ratio = training_data["target"].mean()
        assert 0.05 = 0) and np.all(proba = prob_low

    def test_minimum_functionality(self, model):
        """Model should correctly predict obvious cases."""
        obvious_positive = create_obvious_positive_case()
        obvious_negative = create_obvious_negative_case()

        assert model.predict(obvious_positive)[0] == 1
        assert model.predict(obvious_negative)[0] == 0

4. Quality Gate Tests

class TestModelQuality:
    """Quality gates that must pass before deployment."""

    @pytest.fixture
    def model_and_data(self):
        model = joblib.load("models/model.joblib")
        test_df = pd.read_parquet("data/test.parquet")
        X = test_df.drop("target", axis=1)
        y = test_df["target"]
        return model, X, y

    def test_accuracy_threshold(self, model_and_data):
        model, X, y = model_and_data
        accuracy = accuracy_score(y, model.predict(X))
        assert accuracy >= 0.85, f"Accuracy {accuracy} below 0.85"

    def test_f1_threshold(self, model_and_data):
        model, X, y = model_and_data
        f1 = f1_score(y, model.predict(X), average="weighted")
        assert f1 >= 0.80, f"F1 {f1} below 0.80"

    def test_no_regression(self, model_and_data):
        """New model should not be worse than baseline."""
        model, X, y = model_and_data
        baseline = joblib.load("models/baseline.joblib")

        new_f1 = f1_score(y, model.predict(X), average="weighted")
        baseline_f1 = f1_score(y, baseline.predict(X), average="weighted")

        assert new_f1 >= baseline_f1 - 0.01, \
            f"Regression: new F1 {new_f1} < baseline {baseline_f1}"

    def test_latency(self, model_and_data):
        """Prediction latency should be within SLA."""
        model, X, _ = model_and_data
        import time
        start = time.time()
        model.predict(X[:1])
        latency_ms = (time.time() - start) * 1000
        assert latency_ms < 100, f"Latency {latency_ms}ms exceeds 100ms SLA"

5. Pipeline Integration Tests

class TestTrainingPipeline:
    def test_pipeline_end_to_end(self, tmp_path):
        """Full pipeline should produce a valid model."""
        config = {
            "data": {"path": "data/sample.parquet"},
            "model": {"type": "random_forest", "params": {"n_estimators": 10}},
            "output": {"path": str(tmp_path / "model.joblib")},
        }
        result = run_training_pipeline(config)
        assert result["status"] == "success"
        assert (tmp_path / "model.joblib").exists()

    def test_pipeline_idempotent(self, tmp_path):
        """Running pipeline twice should produce equivalent results."""
        config = {"seed": 42, "output": str(tmp_path)}
        result1 = run_training_pipeline(config)
        result2 = run_training_pipeline(config)
        assert abs(result1["metrics"]["f1"] - result2["metrics"]["f1"]) < 0.001

Best Practices

1. Write data tests first - Catch issues at the source
2. Use behavioral tests - More robust than exact value assertions
3. Automate quality gates in CI/CD before model promotion
4. Test on representative slices not just overall metrics
5. Keep test data separate from training data
6. Use synthetic data for edge case testing
7. Test model serving endpoints, not just model objects
8. Run tests on every commit and before every deployment
9. Track test metrics over time to catch gradual degradation
10. Test the tests - Verify tests catch known failures

Scripts

• scripts/test_model.py - Comprehensive pytest-based model test suite
• scripts/test_data_pipeline.py - Data pipeline test suite

References

See [references/REFERENCE.md](references/REFERENCE.md) for testing strategies by model type.

Source & license

This open-source skill is cataloged on AgentStack and links to its original source — we do not rehost the code.

• Author: timwukp
• Source: timwukp/MLOps-agent-skills
• License: Apache-2.0

Install and usage instructions live in the source repository linked above.