tests/test_explainability.py

"""
Tests for SHAP Explainability Engine.

Integration tests verifying SHAP functionality with trained models.
"""

import base64
import tempfile
from pathlib import Path

import joblib
import numpy as np
import pandas as pd
import pytest

from src.explainability import FraudExplainer
from src.models.pipeline import create_fraud_pipeline


@pytest.fixture
def sample_transaction():
    """Create a valid sample transaction for testing."""
    return pd.DataFrame(
        [
            {
                "trans_date_trans_time": "2020-01-01 12:00:00",
                "amt": 150.0,
                "lat": 40.7128,
                "long": -74.0060,
                "merch_lat": 40.7200,
                "merch_long": -74.0100,
                "job": "Engineer, biomedical",
                "category": "grocery_pos",
                "gender": "M",
                "dob": "1990-01-01",
                "trans_count_24h": 3,
                "amt_to_avg_ratio_24h": 1.2,
                "amt_relative_to_all_time": 1.1,
            }
        ]
    )


@pytest.fixture
def trained_pipeline():
    """Create a minimal trained pipeline for testing."""
    # Create and quickly train a pipeline
    params = {"max_depth": 3, "n_estimators": 10, "learning_rate": 0.3}
    pipeline = create_fraud_pipeline(params)

    # Generate minimal training data
    np.random.seed(42)
    n_samples = 100

    X_train = pd.DataFrame(
        {
            "trans_date_trans_time": pd.date_range("2019-01-01", periods=n_samples, freq="h"),
            "amt": np.random.uniform(10, 500, n_samples),
            "lat": np.random.uniform(30, 45, n_samples),
            "long": np.random.uniform(-120, -70, n_samples),
            "merch_lat": np.random.uniform(30, 45, n_samples),
            "merch_long": np.random.uniform(-120, -70, n_samples),
            "job": np.random.choice(["Engineer, biomedical", "Data scientist"], n_samples),
            "category": np.random.choice(["grocery_pos", "gas_transport"], n_samples),
            "gender": np.random.choice(["M", "F"], n_samples),
            "dob": ["1990-01-01"] * n_samples,
            "trans_count_24h": np.random.randint(1, 10, n_samples),
            "amt_to_avg_ratio_24h": np.random.uniform(0.5, 2.0, n_samples),
            "amt_relative_to_all_time": np.random.uniform(0.5, 2.0, n_samples),
        }
    )

    y_train = np.random.randint(0, 2, n_samples)

    # Train pipeline
    pipeline.fit(X_train, y_train)

    # Save to temp file
    with tempfile.NamedTemporaryFile(delete=False, suffix=".pkl") as f:
        joblib.dump(pipeline, f.name)
        temp_path = f.name

    yield temp_path

    # Cleanup
    Path(temp_path).unlink()


class TestFraudExplainer:
    """Test suite for FraudExplainer class."""

    def test_initialization(self, trained_pipeline):
        """Test that explainer initializes without errors."""
        explainer = FraudExplainer(trained_pipeline)

        assert explainer.pipeline is not None
        assert explainer.model is not None
        assert explainer.preprocessor is not None
        assert explainer.explainer is not None
        assert len(explainer.feature_names) > 0

    def test_initialization_invalid_path(self):
        """Test that explainer raises error for invalid path."""
        with pytest.raises(FileNotFoundError):
            FraudExplainer("/nonexistent/path.pkl")

    def test_generate_waterfall(self, trained_pipeline, sample_transaction):
        """Test waterfall plot generation for a single transaction."""
        explainer = FraudExplainer(trained_pipeline)

        # Generate waterfall (base64)
        waterfall_b64 = explainer.generate_waterfall(sample_transaction)

        # Verify it's a valid base64 string
        assert isinstance(waterfall_b64, str)
        assert len(waterfall_b64) > 0

        # Verify it decodes to valid bytes
        try:
            decoded = base64.b64decode(waterfall_b64)
            assert len(decoded) > 0
        except Exception as e:
            pytest.fail(f"Failed to decode base64: {e}")

    def test_generate_waterfall_multiple_transactions_fails(
        self, trained_pipeline, sample_transaction
    ):
        """Test that waterfall fails with multiple transactions."""
        explainer = FraudExplainer(trained_pipeline)

        # Create 2 transactions
        two_transactions = pd.concat([sample_transaction, sample_transaction])

        with pytest.raises(ValueError, match="Expected 1 transaction"):
            explainer.generate_waterfall(two_transactions)

    def test_generate_summary(self, trained_pipeline, sample_transaction):
        """Test summary plot generation for multiple transactions."""
        explainer = FraudExplainer(trained_pipeline)

        # Create sample of 10 transactions
        sample = pd.concat([sample_transaction] * 10, ignore_index=True)

        # Generate summary plot
        summary_b64 = explainer.generate_summary(sample)

        # Verify it's a valid base64 string
        assert isinstance(summary_b64, str)
        assert len(summary_b64) > 0

        # Verify it decodes
        try:
            decoded = base64.b64decode(summary_b64)
            assert len(decoded) > 0
        except Exception as e:
            pytest.fail(f"Failed to decode base64: {e}")

    def test_explain_prediction(self, trained_pipeline, sample_transaction):
        """Test comprehensive prediction explanation."""
        explainer = FraudExplainer(trained_pipeline)

        explanation = explainer.explain_prediction(sample_transaction, threshold=0.5)

        # Verify structure
        assert "prediction" in explanation
        assert "decision" in explanation
        assert "shap_values" in explanation
        assert "top_features" in explanation
        assert "base_value" in explanation

        # Verify types
        assert isinstance(explanation["prediction"], float)
        assert explanation["decision"] in ["BLOCK", "APPROVE"]
        assert isinstance(explanation["shap_values"], dict)
        assert isinstance(explanation["top_features"], list)
        assert len(explanation["top_features"]) == 5

        # Verify top_features structure
        for feature in explanation["top_features"]:
            assert "feature" in feature
            assert "impact" in feature
            assert "abs_impact" in feature

    def test_no_value_error_raised(self, trained_pipeline, sample_transaction):
        """Test that no ValueError is raised during normal operation."""
        explainer = FraudExplainer(trained_pipeline)

        # This should not raise ValueError
        try:
            waterfall = explainer.generate_waterfall(sample_transaction)
            summary = explainer.generate_summary(sample_transaction)
            explanation = explainer.explain_prediction(sample_transaction)
        except ValueError as e:
            pytest.fail(f"Unexpected ValueError raised: {e}")

    def test_shap_values_calculation(self, trained_pipeline, sample_transaction):
        """Test SHAP value calculation."""
        explainer = FraudExplainer(trained_pipeline)

        shap_values, X_transformed = explainer.calculate_shap_values(sample_transaction)

        # Verify shapes
        assert shap_values.shape[0] == 1  # 1 transaction
        assert shap_values.shape[1] == len(explainer.feature_names)
        assert X_transformed.shape[0] == 1
        assert X_transformed.shape[1] == len(explainer.feature_names)