X-Financial/server/tests/test_automation_eligibility.py

from __future__ import annotations

from decimal import Decimal

from app.services.automation_eligibility import (
    AutomationEligibilityCalculator,
    AutomationEligibilityInput,
    AutomationLevel,
    AutomationPolicy,
)


def _enabled_policy(**updates) -> AutomationPolicy:
    values = {
        "enabled": True,
        "enterprise_max_level": AutomationLevel.L4_LOW_RISK_STRAIGHT_THROUGH,
        "enterprise_whitelist": frozenset({"expense.prefill_fields"}),
        "amount_cap": Decimal("1000"),
        "min_confidence": 0.95,
        "min_evidence_completeness": 0.95,
        "min_historical_precision": 0.98,
        "min_historical_samples": 100,
        "minimum_sampling_rate": 0.1,
        "release_stage": "active",
    }
    values.update(updates)
    return AutomationPolicy(**values)


def _qualified_input(**updates) -> AutomationEligibilityInput:
    values = {
        "action_key": "expense.prefill_fields",
        "action_risk": "low",
        "amount": Decimal("0"),
        "confidence": 0.99,
        "evidence_completeness": 1.0,
        "reversible": True,
        "historical_precision": 0.995,
        "historical_samples": 1000,
        "sampling_rate": 0.2,
    }
    values.update(updates)
    return AutomationEligibilityInput(**values)


def test_default_policy_is_shadow_only() -> None:
    decision = AutomationEligibilityCalculator().calculate(_qualified_input())

    assert decision.eligible_level == AutomationLevel.L1_RECOMMEND
    assert decision.eligible_for_auto_execution is False
    assert decision.requires_human_confirmation is True


def test_hard_whitelist_cannot_be_extended_by_enterprise() -> None:
    decision = AutomationEligibilityCalculator().calculate(
        _qualified_input(action_key="expense.unknown_action"),
        _enabled_policy(enterprise_whitelist=frozenset({"expense.unknown_action"})),
    )

    assert decision.eligible_level == AutomationLevel.L1_RECOMMEND
    assert "action_not_in_hard_whitelist" in decision.reasons


def test_payment_and_high_risk_policy_actions_never_auto_execute() -> None:
    calculator = AutomationEligibilityCalculator()
    payment = calculator.calculate(
        _qualified_input(action_key="expense.execute_payment", moves_money=True),
        _enabled_policy(enterprise_whitelist=frozenset({"expense.execute_payment"})),
    )
    policy_change = calculator.calculate(
        _qualified_input(
            action_key="expense.publish_policy",
            changes_policy=True,
            action_risk="high",
        ),
        _enabled_policy(enterprise_whitelist=frozenset({"expense.publish_policy"})),
    )

    assert payment.eligible_level == AutomationLevel.L5_HUMAN_CONTROLLED
    assert policy_change.eligible_level == AutomationLevel.L5_HUMAN_CONTROLLED
    assert not payment.eligible_for_auto_execution
    assert not policy_change.eligible_for_auto_execution


def test_all_controls_can_reach_enterprise_cap() -> None:
    decision = AutomationEligibilityCalculator().calculate(
        _qualified_input(),
        _enabled_policy(enterprise_max_level=AutomationLevel.L3_REVERSIBLE_AUTO),
    )

    assert decision.eligible_level == AutomationLevel.L3_REVERSIBLE_AUTO
    assert decision.eligible_for_auto_execution is True
    assert "clamped_by_enterprise_level" in decision.reasons


def test_each_safety_gap_clamps_to_human_confirmation() -> None:
    calculator = AutomationEligibilityCalculator()
    candidates = [
        _qualified_input(amount=Decimal("1001")),
        _qualified_input(action_risk="medium"),
        _qualified_input(reversible=False),
        _qualified_input(confidence=0.7),
        _qualified_input(evidence_completeness=0.7),
        _qualified_input(historical_precision=None),
        _qualified_input(historical_samples=10),
    ]

    for candidate in candidates:
        decision = calculator.calculate(candidate, _enabled_policy())
        assert decision.eligible_level <= AutomationLevel.L2_PREFILL
        assert decision.requires_human_confirmation is True


def test_low_sampling_rate_prevents_all_auto_execution() -> None:
    decision = AutomationEligibilityCalculator().calculate(
        _qualified_input(sampling_rate=0.01),
        _enabled_policy(),
    )

    assert decision.eligible_level == AutomationLevel.L2_PREFILL
    assert decision.eligible_for_auto_execution is False
    assert "sampling_rate_below_threshold" in decision.reasons


def test_invalid_amount_and_enterprise_l0_cap_fail_closed() -> None:
    calculator = AutomationEligibilityCalculator()
    invalid_amount = calculator.calculate(
        _qualified_input(amount=Decimal("NaN")),
        _enabled_policy(),
    )
    human_only = calculator.calculate(
        _qualified_input(),
        _enabled_policy(enterprise_max_level=AutomationLevel.L0_EXPLAIN),
    )

    assert invalid_amount.eligible_level == AutomationLevel.L2_PREFILL
    assert "amount_invalid" in invalid_amount.reasons
    assert human_only.eligible_level == AutomationLevel.L0_EXPLAIN


def test_even_qualified_action_stays_non_executing_in_shadow() -> None:
    decision = AutomationEligibilityCalculator().calculate(
        _qualified_input(),
        _enabled_policy(release_stage="shadow"),
    )

    assert decision.eligible_level == AutomationLevel.L4_LOW_RISK_STRAIGHT_THROUGH
    assert decision.mode == "shadow_low_risk_straight_through"
    assert decision.eligible_for_auto_execution is False


def test_malformed_levels_and_sample_counts_fail_closed_without_runtime_error() -> None:
    decision = AutomationEligibilityCalculator().calculate(
        _qualified_input(historical_samples="not-a-number"),
        _enabled_policy(
            enterprise_max_level="not-a-level",
            min_historical_samples="not-a-count",
        ),
    )

    assert decision.eligible_level == AutomationLevel.L1_RECOMMEND
    assert decision.eligible_for_auto_execution is False
    assert decision.checks["historical_samples"] == 0
    assert decision.checks["historical_samples_valid"] is False
    assert decision.checks["minimum_historical_samples_valid"] is False
    assert decision.checks["enterprise_max_level"] == 1