File size: 8,391 Bytes
eb0499d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 | """Robustness & Failure Injection Test Suite for Explainability (XAI) Subsystem.
Stress-tests every explainability algorithm by injecting hostile boundary conditions:
GEX1: NaN Risk Score in Alert
GEX2: NaN Feature Values in Transaction Dict
GEX3: +Inf / -Inf Feature Amounts & Risk Scores
GEX4: Empty Transaction Dictionary ({})
GEX5: Empty Alert Reason Codes & Risk Factors
GEX6: Missing Features (9 out of 10 features absent)
GEX7: Malformed Feature Data Types (strings, lists, None for numeric features)
GEX8: Extreme Floating-Point Values (1e308, -1e308, 1e-308)
GEX9: Empty Category & MCC Strings in Real-Time Explainer
GEX10: Empty Feature Vector List ([]) in Real-Time SHAP Engine
GEX11: Path Traversal & Special Characters in Transaction ID
GEX12: Non-Existent Node ID in GNN Embedding Explainer
GEX13: Target Counterfactual Score Higher than Original Score
GEX14: Extremely Low Target Counterfactual Score (0.0 or negative)
"""
from __future__ import annotations
from pathlib import Path
import sys
import math
import pytest
import numpy as np
PROJECT_ROOT = str(Path(__file__).resolve().parents[3] / "backend")
sys.path.insert(0, PROJECT_ROOT)
from app.application.services.explainability_service import ExplainabilityService
from app.domain.realtime_explainer import FastInferenceExplainer
from app.domain.investigation_entities import Alert, AlertSeverity
explainer_service = ExplainabilityService()
fast_explainer = FastInferenceExplainer()
def assert_finite(val: float, name: str) -> None:
"""Assert that a value is non-NaN and non-Inf."""
assert not math.isnan(val), f"{name} must not be NaN, got {val}"
assert not math.isinf(val), f"{name} must not be Inf, got {val}"
# =====================================================================
# GEX1 & GEX2: NaN Inputs
# =====================================================================
def test_gex1_nan_risk_score_in_alert():
"""GEX1: NaN alert risk_score - must not crash or return NaN breakdown scores."""
alert = Alert(
id="alt_nan_01",
risk_score=float("nan"),
severity=AlertSeverity.HIGH,
model_confidence=0.85,
bank_id="bank_a",
reason_codes=["ML-HIGH"],
)
try:
rpt = explainer_service.explain_alert(alert)
for s in rpt.risk_score_breakdown:
assert_finite(s.normalized_score, "normalized_score")
except (ValueError, FloatingPointError):
pass # Explicit domain exception is acceptable
def test_gex2_nan_feature_values_in_shap():
"""GEX2: NaN feature values in compute_shap_values - must fail-closed deterministically (INV-C/INV-08)."""
txn_dict = {
"transaction_amount": float("nan"),
"velocity": float("nan"),
"merchant_risk_score": 0.5,
}
with pytest.raises(ValueError, match="Feature 'transaction_amount' contains non-finite value: nan"):
explainer_service.compute_shap_values(txn_dict)
# =====================================================================
# GEX3: Infinite (+/-Inf) Values
# =====================================================================
def test_gex3_positive_infinity_amount():
"""GEX3a: +Inf transaction_amount - must fail-closed deterministically (INV-C/INV-08)."""
txn_dict = {"transaction_amount": float("inf"), "velocity": 5.0}
with pytest.raises(ValueError, match="Feature 'transaction_amount' contains non-finite value: inf"):
explainer_service.compute_shap_values(txn_dict)
def test_gex3_negative_infinity_risk_score():
"""GEX3b: -Inf alert risk_score - successfully handled post-remediation (BUG-EX-01 fixed)."""
alert = Alert(
id="alt_inf_01",
risk_score=float("-inf"),
severity=AlertSeverity.LOW,
model_confidence=0.5,
bank_id="bank_a",
reason_codes=[],
)
report = explainer_service.explain_alert(alert)
assert report is not None
assert report.alert_id == "alt_inf_01"
def test_gex7_malformed_feature_data_types():
"""GEX7: Malformed string amount - must fail-closed deterministically without coercion (INV-D/INV-09)."""
txn_dict = {
"transaction_amount": "invalid_string_1000",
"velocity": [1, 2, 3],
"merchant_risk_score": None,
}
with pytest.raises(ValueError, match="Invalid numeric value for feature 'transaction_amount'"):
explainer_service.compute_shap_values(txn_dict)
# =====================================================================
# GEX8: Extreme Floating-Point Values (1e308, -1e308, 1e-308)
# =====================================================================
def test_gex8_extreme_floating_point_values():
"""GEX8: Extreme values (1e308) in compute_shap_values - must not overflow."""
txn_dict = {
"transaction_amount": 1e308,
"velocity": 1e-308,
"account_age_days": -1e308,
}
contributions = explainer_service.compute_shap_values(txn_dict)
assert len(contributions) == 10
for f in contributions:
assert_finite(f["contribution"], f"contribution for {f['feature']}")
# =====================================================================
# GEX9 & GEX10: Real-Time Explainer Edge Cases
# =====================================================================
def test_gex9_empty_category_strings_realtime():
"""GEX9: Empty string merchant_category in explain_realtime_score - no exception."""
attributions = fast_explainer.explain_realtime_score(
amount=100.0, velocity_1h=1, merchant_category="", risk_score=0.1
)
assert isinstance(attributions, list)
def test_gex10_empty_feature_vector_list_realtime():
"""GEX10: Empty list [] passed as feature_vector to compute_shap - returns fallback attributions."""
res = fast_explainer.compute_shap("tx_empty_01", [])
assert res["status"] == "COMPLETED"
assert isinstance(res["shap_values"], list)
# =====================================================================
# GEX11: Path Traversal & Special Characters in Transaction ID
# =====================================================================
def test_gex11_path_traversal_transaction_id():
"""GEX11: Transaction ID with path traversal `../../etc/passwd` - sanitized key in Redis/cache."""
res = fast_explainer.compute_shap("../../etc/passwd", {"amount": 500.0})
assert res["transaction_id"] == "../../etc/passwd"
assert res["status"] == "COMPLETED"
# =====================================================================
# GEX12: Non-Existent Node ID in GNN Explainer
# =====================================================================
def test_gex12_non_existent_gnn_node():
"""GEX12: Non-existent node_id in explain_gnn_embedding - safely returns zero-edge isolated report."""
gnn_rpt = explainer_service.explain_gnn_embedding("non_existent_node_99999")
assert gnn_rpt.node_id == "non_existent_node_99999"
assert len(gnn_rpt.top_contributing_edges) == 0
assert gnn_rpt.subgraph_edges_count == 0
assert "Isolated entity" in gnn_rpt.primary_driver_text
# =====================================================================
# GEX13 & GEX14: Counterfactual Edge Cases
# =====================================================================
def test_gex13_target_score_higher_than_original():
"""GEX13: Target score higher than original alert score - is_cleared must be True."""
alert = Alert(
id="alt_cf_01",
risk_score=200.0,
severity=AlertSeverity.LOW,
model_confidence=0.9,
bank_id="bank_a",
reason_codes=[],
)
cf = explainer_service.generate_counterfactuals(alert, target_score=350.0)
assert cf.is_cleared is True
assert cf.remediated_score <= 350.0
def test_gex14_extreme_low_target_score():
"""GEX14: Target score = 0.0 - counterfactuals clamp remediated score to min 50.0."""
alert = Alert(
id="alt_cf_02",
risk_score=800.0,
severity=AlertSeverity.CRITICAL,
model_confidence=0.95,
bank_id="bank_a",
reason_codes=["HIGH-AMT", "GEO-RISK"],
)
cf = explainer_service.generate_counterfactuals(alert, target_score=0.0)
assert cf.remediated_score >= 50.0 # Clamped to minimum 50.0
assert isinstance(cf.changes, list)
if __name__ == "__main__":
pytest.main([__file__, "-v"])
|