98 lines
3.5 KiB
Python
98 lines
3.5 KiB
Python
"""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
|
|
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, [])
|
|
|
|
|
|
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"
|