Testability and Shift-Left Testing Questions

Policy Title: Organization ML Test SLAs & Model Change Policy — Outline

1. Purpose & Scope

• Apply to all production ML models, feature flags, and retrain pipelines across business lines.
• Owners: Model Owner, Release Manager, MLOps, Risk/Compliance.

2. Required Test Types (pre-gate)

• Data validation: schema checks, distribution drift tests, nulls, PII leakage scans.
• Training validation: cross-validation metrics, calibration, fairness checks, reproducibility (seeded runs).
• Performance & robustness: holdout eval, adversarial/noise tests, OOD detection.
• Integration & contract tests: API, latency, resource usage, dependency compatibility.
• Security & privacy: vulnerability scan, differential privacy checks where applicable.
• Explainability & docs: model card, training data summary, intended use & limitations.

3. Minimal Gating Thresholds (examples; business configurable)

• Accuracy/AUC/F1: not worse than baseline by >1% absolute or 5% relative.
• Latency: 95th percentile < X ms (service SLA).
• Error uplift: production-regression < 0.5% user impact.
• Fairness: specified subgroup metric gap < predefined delta.
• Reproducibility: identical seeds reproduce within tolerance.
• Any unmet threshold must be documented with mitigation plan.

4. Rollout Requirements

• Canary rollout 1–5% traffic for 24–72 hrs with real-time telemetry.
• Gradual ramping (5→25→50→100) with automated gating at each step.
• Automated rollback criteria (metric degradation, error rates, latency spikes).

5. Monitoring Obligations (post-deploy)

• Real-time: primary metric, downstream business KPIs, error/latency, data drift, prediction distribution.
• Alerts: tiered (P1/P2) with on-call responsibilities and runbooks.
• Scheduled: weekly model health report, quarterly audit, retrain triggers defined.
• Logging & retention: input snapshots, model versions, seed, hyperparams for N days.

6. Exceptions & Risk Escalation

• Exception process: submit risk assessment, compensating controls, temporary TTL approval by Risk + MLOps.
• High-risk models (safety/regulated): require stricter thresholds and manual review.

7. Governance & Versioning

• Enforce model registry (version, artifacts, lineage), CI/CD pipelines, immutable releases.

Onboarding Plan

• Phase 1: Leadership kickoff + policy doc, identify pilot teams.
• Phase 2: Templates & tooling (checklists, CI job templates, model-card generator, monitoring dashboards).
• Phase 3: Workshops: hands-on labs for implementing gates, mock deployments, and runbook drills.
• Phase 4: Pilot rollouts with MLOps mentor; collect feedback and iterate thresholds.
• Phase 5: Org-wide rollout, mandatory compliance by deadline, quarterly audits, and feedback loop.

Rationale: Combines engineering rigor (automated gates, reproducibility), business safety (gradual rollouts, rollback), and governance (audits, exceptions) while providing practical onboarding through tools, training, and pilots.

Line coverage is a blunt instrument for ML codebases: 80% shows many lines execute in tests, but not that the right behaviors, data edge cases, or model-level properties are validated. Meaningful test coverage for ML combines unit tests of utilities, integration tests of data pipelines and training loops, and behavioral tests that assert model outputs, performance, robustness, and reproducibility under realistic data scenarios.

Key additional quality signals / metrics (with why they matter):

• Mutation score: measures whether tests catch injected faults; low mutation escape means brittle gaps in assertions.
• Test dataset coverage (feature-space / scenario coverage): ensures tests exercise rare classes, boundary values, and important feature combinations.
• Model-regression checks (CI automated): track metrics (accuracy, AUC, latency) vs baseline with tolerances to catch performance regressions.
• Data validation and schema tests (e.g., using Great Expectations): detect upstream data drift, missing/null distributions, and type changes.
• End-to-end integration and canary test pass rate: validates training→serving pipeline and deployment correctness in staging.
• Determinism / reproducibility metrics: percentage of runs that produce identical weights/metrics with fixed seeds; flags nondeterministic components.
• Test flakiness / stability (failure rate over time) and test runtime: high flakiness or long tests reduce CI trust and feedback speed.
• Adversarial / robustness tests and calibration checks: measure resilience to input perturbations and confidence calibration (ECE).

Together these signals give a multi-dimensional view: code exercised, behaviors asserted, data assumptions enforced, model quality preserved, and CI reliability—far more meaningful than line percentage alone.

Approach: create a tiny deterministic PyTorch nn.Module, set seeds, trace it to TorchScript, save/load from a temp file, run both models on the same fixed input, and assert outputs are close with specified rtol/atol.

import tempfile
import torch
import torch.nn as nn
import pytest

@pytest.mark.parametrize("rtol,atol", [(1e-3, 1e-6), (1e-2, 1e-5)])
def test_torchscript_serialization_outputs_close(rtol, atol):
    # Control randomness for determinism
    torch.manual_seed(42)
    
    # Simple model
    class SimpleModel(nn.Module):
        def __init__(self):
            super().__init__()
            self.lin = nn.Linear(4, 3, bias=True)
            nn.init.constant_(self.lin.weight, 0.5)
            nn.init.constant_(self.lin.bias, 0.1)
        def forward(self, x):
            return torch.relu(self.lin(x))
    
    model = SimpleModel().eval()  # eval to avoid training randomness

    # Deterministic input
    torch.manual_seed(0)
    inp = torch.randn(2, 4, dtype=torch.float32)

    # Get reference output
    with torch.no_grad():
        ref_out = model(inp)

    # Trace and serialize to temporary file
    traced = torch.jit.trace(model, inp)
    with tempfile.NamedTemporaryFile(suffix=".pt") as tmp:
        torch.jit.save(traced, tmp.name)
        loaded = torch.jit.load(tmp.name)
        loaded.eval()
        with torch.no_grad():
            loaded_out = loaded(inp)

    # Assert outputs are numerically close within tolerances
    assert torch.allclose(ref_out, loaded_out, rtol=rtol, atol=atol), (
        f"Outputs differ: max abs diff={(ref_out - loaded_out).abs().max().item():.6f}"
    )

Key points:

• Seed control (torch.manual_seed) ensures repeatable weights and inputs.
• Use eval() and torch.no_grad() to avoid nondeterminism from dropout/batchnorm.
• Test uses parametrized tolerances to reflect acceptable numerical differences from JIT tracing/serialization.

I avoid arbitrary sleeps because they make tests slow and nondeterministic. Instead, I try to control the source of asynchrony.

Patterns I use:

• Expectations or latches: wait for a specific callback or event, with a timeout only as a safety net.
• Fake clocks or test schedulers: for timers, debounce logic, or reactive streams, I advance virtual time manually instead of waiting in real time.
• Injected executors/dispatchers: run work on a test-controlled thread or immediate executor.
• Async/await or promise testing: await the final state rather than polling.
• For streams, subscribe in the test, emit events explicitly, and assert the sequence.

The key is to make the async dependency injectable so the test controls when work happens. That keeps tests fast, repeatable, and easy to debug. If a test still needs a timeout, I keep it short and use it only to fail fast, not to "solve" synchronization. The best async tests are ones where the test owns time and input, not the app.

Approach: use pytest.mark.parametrize to iterate over imputers and scalers (and their combinations) so one test covers many preprocessing scenarios without duplication. Each param set is small and readable; pytest will show which combo failed via ids.

import numpy as np
import pytest
from sklearn.impute import SimpleImputer
from sklearn.preprocessing import StandardScaler, MinMaxScaler
from sklearn.pipeline import Pipeline

# small synthetic dataset with missing values
X = np.array([
    [1.0,   np.nan],
    [2.0,  10.0],
    [np.nan, 20.0],
    [4.0,  30.0]
])

imputers = [
    (SimpleImputer(strategy="mean"), "imputer-mean"),
    (SimpleImputer(strategy="median"), "imputer-median"),
]

scalers = [
    (StandardScaler(), "std"),
    (MinMaxScaler(), "minmax"),
]

@pytest.mark.parametrize("imputer,imputer_id", imputers, ids=[i[1] for i in imputers])
@pytest.mark.parametrize("scaler,scaler_id", scalers, ids=[s[1] for s in scalers])
def test_pipeline_impute_then_scale(imputer, imputer_id, scaler, scaler_id):
    """
    For each (imputer, scaler) combo:
    - fit_transform should produce no NaNs
    - shape preserved
    - scaler properties: minmax in [0,1]; std has zero mean approx for each column
    """
    pipe = Pipeline([("imputer", imputer), ("scaler", scaler)])
    Xt = pipe.fit_transform(X)

    # shape and no missing values
    assert Xt.shape == X.shape
    assert not np.isnan(Xt).any()

    # scaler-specific checks
    if isinstance(scaler, MinMaxScaler):
        assert np.all(Xt >= -1e-8) and np.all(Xt <= 1 + 1e-8)
    else:  # StandardScaler
        # columns should have mean ~0
        col_means = Xt.mean(axis=0)
        assert np.allclose(col_means, 0.0, atol=1e-6)

Key points:

• Parametrization reduces duplicate test code while covering all combos.
• ids make failures easy to identify.
• Add more imputers/scalers or edge-case datasets to expand coverage.