File size: 5,752 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 | """Unit tests for Advanced AI Explainability Portal (Counterfactuals, Decision Replay, GNNExplainer)."""
from __future__ import annotations
from app.application.services.explainability_service import ExplainabilityService
from app.domain.enums import AlertSeverity, AlertStatus
from app.domain.investigation_entities import Alert
def _make_sample_alert(alert_id: str = "alt_1001", risk_score: float = 750.0) -> Alert:
return Alert(
id=alert_id,
bank_id="bank_a",
transaction_id="tx_9988",
risk_score=risk_score,
severity=AlertSeverity.HIGH,
status=AlertStatus.NEW,
reason_codes=["HIGH-AMT", "GEO-RISK", "VEL-001"],
confidence=0.92,
involved_entity_ids=["cust_a1"],
top_features=[{"feature": "transaction_amount", "contribution": 0.42}],
risk_factors=["Transaction amount exceeds threshold"],
historical_evidence=["Prior alert 2 days ago"],
model_confidence=0.92,
)
class TestCounterfactualEngine:
"""Verify counterfactual remediation path generation."""
def test_generate_counterfactuals_clears_alert(self):
svc = ExplainabilityService()
alert = _make_sample_alert(risk_score=750.0)
cf = svc.generate_counterfactuals(alert, target_score=350.0)
assert cf.alert_id == alert.id
assert cf.original_score == 750.0
assert cf.remediated_score <= 350.0
assert cf.is_cleared is True
assert len(cf.changes) >= 1
assert "CLEARED" in cf.summary_text
def test_counterfactual_remediates_high_amount_and_geo(self):
svc = ExplainabilityService()
alert = _make_sample_alert(risk_score=850.0)
cf = svc.generate_counterfactuals(alert, target_score=300.0)
features_changed = [c.feature for c in cf.changes]
assert "transaction_amount" in features_changed or "country_code" in features_changed
def test_counterfactual_executes_real_risk_engine_search(self):
"""Verifies that counterfactual remediated_score is computed by real RiskScoringEngine and strictly clears target threshold."""
from app.application.services.risk_engine import RiskScoringEngine
svc = ExplainabilityService()
engine = RiskScoringEngine()
alert = _make_sample_alert(risk_score=850.0)
target_threshold = 350.0
result = svc.generate_counterfactuals(alert, target_score=target_threshold, risk_engine=engine)
assert result.original_score == 850.0
assert result.remediated_score <= target_threshold
assert result.is_cleared is True
assert len(result.changes) >= 1
assert isinstance(result.remediated_score, float)
def test_counterfactual_honest_failure_when_unreachable(self):
"""Verifies that when target threshold is deliberately unreachable within the search budget (e.g. 10.0),
the search reports failure honestly rather than forcing a fake cleared status.
"""
from app.application.services.risk_engine import RiskScoringEngine
svc = ExplainabilityService()
engine = RiskScoringEngine()
alert = _make_sample_alert(risk_score=850.0)
unreachable_target = 10.0
result = svc.generate_counterfactuals(alert, target_score=unreachable_target, risk_engine=engine)
# Confirm search reports failure honestly
assert result.original_score == 850.0
assert result.is_cleared is False
assert result.remediated_score <= result.original_score
assert result.remediated_score > unreachable_target
assert "did not reach" in result.summary_text
class TestDecisionReplayAudit:
"""Verify deterministic decision replay inference audit."""
def test_replay_inference_audit_reproduces_score(self):
svc = ExplainabilityService()
alert = _make_sample_alert(risk_score=680.0)
audit = svc.replay_inference_audit(alert)
assert audit.alert_id == alert.id
assert audit.model_version == "v1.4.2-champion"
assert audit.model_auc > 0.90
assert len(audit.policy_rules_evaluated) == 9
assert audit.reconstructed_risk_score > 0
assert audit.audit_matched is not None
class TestGNNExplainer:
"""Verify GNNExplainer graph attribution over entity neighborhood."""
def test_explain_gnn_embedding_calculates_edge_contributions(self):
from app.application.services.graph_engine import GraphEngine
from app.domain.enums import EntityType, RelationshipType
from app.domain.investigation_entities import Entity, Relationship
ge = GraphEngine()
ge.register_entity(
Entity(id="cust_a1", entity_type=EntityType.CUSTOMER, privacy_id="priv_a1", bank_id="bank_a")
)
ge.register_entity(
Entity(id="dev_x99", entity_type=EntityType.DEVICE, privacy_id="priv_x99", bank_id="bank_a")
)
ge.add_relationship(
Relationship(
source_entity_id="cust_a1",
target_entity_id="dev_x99",
relationship_type=RelationshipType.SHARES_DEVICE,
)
)
svc = ExplainabilityService()
gnn_exp = svc.explain_gnn_embedding("cust_a1")
assert gnn_exp.node_id == "cust_a1"
assert gnn_exp.subgraph_nodes_count >= 1
assert len(gnn_exp.top_contributing_edges) >= 1
assert "Primary GNN Driver" in gnn_exp.primary_driver_text
assert gnn_exp.top_contributing_edges[0].target == "dev_x99"
# Verify sum of contribution percentages equals 100%
total_pct = sum(e.contribution_percentage for e in gnn_exp.top_contributing_edges)
assert abs(total_pct - 100.0) < 1.0
|