| """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) |
|
|
| |
| 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" |
|
|
| |
| total_pct = sum(e.contribution_percentage for e in gnn_exp.top_contributing_edges) |
| assert abs(total_pct - 100.0) < 1.0 |
|
|
