fund-tracer/backend/tests/test_rules.py

"""Unit tests for the rules engine using plain objects (no SQLAlchemy session)."""
from datetime import datetime, timezone
from types import SimpleNamespace
from uuid import uuid4

import pytest

from app.models.transaction import Direction
from app.models.evidence_image import SourceApp
from app.rules.dedup_rules import is_duplicate_pair
from app.rules.transit_rules import (
    is_self_transfer,
    is_fee_tolerant_transit_pair,
    mark_transit_transactions,
)
from app.rules.assessment_rules import classify_transaction


def _make_tx(**kwargs):
    defaults = dict(
        id=uuid4(), case_id=uuid4(), source_app=SourceApp.alipay,
        trade_time=datetime(2026, 3, 8, 10, 0, tzinfo=timezone.utc),
        amount=10000, direction=Direction.out,
        counterparty_name="测试对手方", counterparty_account="",
        self_account_tail_no="1234", order_no="ORD001",
        remark="测试", confidence=0.9, is_duplicate=False, is_transit=False,
        evidence_image_id=None, cluster_id=None,
    )
    defaults.update(kwargs)
    return SimpleNamespace(**defaults)


class TestDedupRules:
    def test_same_order_no(self):
        a = _make_tx(order_no="ORD001")
        b = _make_tx(order_no="ORD001", self_account_tail_no="5678")
        assert is_duplicate_pair(a, b)

    def test_different_order_no_different_counterparty(self):
        a = _make_tx(order_no="ORD001", counterparty_name="A", self_account_tail_no="1111")
        b = _make_tx(order_no="ORD002", counterparty_name="B", self_account_tail_no="2222")
        assert not is_duplicate_pair(a, b)

    def test_same_amount_close_time_same_tail_should_not_dedup(self):
        a = _make_tx(order_no="", amount=5000)
        b = _make_tx(
            order_no="",
            amount=5000,
            trade_time=datetime(2026, 3, 8, 10, 3, tzinfo=timezone.utc),
        )
        assert not is_duplicate_pair(a, b)

    def test_same_amount_far_time(self):
        a = _make_tx(order_no="", amount=5000)
        b = _make_tx(
            order_no="",
            amount=5000,
            trade_time=datetime(2026, 3, 8, 11, 0, tzinfo=timezone.utc),
        )
        assert not is_duplicate_pair(a, b)

    def test_same_amount_close_time_same_counterparty_should_not_dedup(self):
        a = _make_tx(order_no="", amount=8000, counterparty_name="刷单账户A")
        b = _make_tx(
            order_no="",
            amount=8000,
            counterparty_name="刷单账户A",
            trade_time=datetime(2026, 3, 8, 10, 2, tzinfo=timezone.utc),
        )
        assert not is_duplicate_pair(a, b)


class TestTransitRules:
    def test_keyword_match(self):
        tx = _make_tx(counterparty_name="支付宝-张某", direction=Direction.out)
        assert is_self_transfer(tx, [])

    def test_known_account_match(self):
        tx = _make_tx(counterparty_name="我的银行卡")
        assert is_self_transfer(tx, ["我的银行卡"])

    def test_not_transit(self):
        tx = _make_tx(counterparty_name="李*华", remark="投资款")
        assert not is_self_transfer(tx, [])

    def test_fee_tolerant_pair_match(self):
        out_tx = _make_tx(
            direction=Direction.out,
            amount=10000,
            trade_time=datetime(2026, 3, 8, 10, 0, tzinfo=timezone.utc),
        )
        in_tx = _make_tx(
            direction=Direction.in_,
            amount=9850,  # 1.5% diff
            trade_time=datetime(2026, 3, 8, 10, 1, 30, tzinfo=timezone.utc),
        )
        assert is_fee_tolerant_transit_pair(out_tx, in_tx)

    def test_fee_tolerant_pair_time_window_exceeded(self):
        out_tx = _make_tx(
            direction=Direction.out,
            amount=10000,
            trade_time=datetime(2026, 3, 8, 10, 0, tzinfo=timezone.utc),
        )
        in_tx = _make_tx(
            direction=Direction.in_,
            amount=9900,
            trade_time=datetime(2026, 3, 8, 10, 3, 1, tzinfo=timezone.utc),
        )
        assert not is_fee_tolerant_transit_pair(out_tx, in_tx)

    def test_wechat_recharge_in_then_out_marked_transit(self):
        in_tx = _make_tx(
            trade_time=datetime(2026, 3, 17, 21, 46, 0, tzinfo=timezone.utc),
            amount=50,
            direction=Direction.in_,
            counterparty_name="零钱充值-来自工商银行(3893)",
            counterparty_account="",
            self_account_tail_no="3893",
            order_no="",
            remark="",
            confidence=0.95,
            is_transit=False,
        )
        out_tx = _make_tx(
            trade_time=datetime(2026, 3, 17, 21, 46, 59, tzinfo=timezone.utc),
            amount=50,
            direction=Direction.out,
            counterparty_name="童年",
            counterparty_account="1154****0928",
            self_account_tail_no="3893",
            order_no="",
            remark="充值",
            confidence=0.98,
            is_transit=False,
        )
        txs = [in_tx, out_tx]
        mark_transit_transactions(txs, known_self_accounts=[])
        assert in_tx.is_transit
        assert out_tx.is_transit


class TestAssessmentRules:
    def test_transit_classified_as_low(self):
        tx = _make_tx(is_transit=True)
        level, reason, _ = classify_transaction(tx)
        assert level.value == "low"

    def test_high_confidence_fraud_keyword(self):
        tx = _make_tx(confidence=0.95, remark="投资款")
        level, reason, _ = classify_transaction(tx)
        assert level.value == "high"

    def test_income_classified_as_low(self):
        tx = _make_tx(direction=Direction.in_)
        level, _, _ = classify_transaction(tx)
        assert level.value == "low"