354 lines
14 KiB
Python
354 lines
14 KiB
Python
"""批次 B1 · AI 应收风险预警测试。
|
||
|
||
覆盖：三因子评分逻辑（回款变慢 / 欠款上升 / 频次骤降各有正分）、分档、
|
||
风险排行、预警去重、API 全链路、无 LLM KEY 时的降级路径、租户隔离。
|
||
"""
|
||
|
||
import pytest
|
||
from datetime import date, timedelta
|
||
from decimal import Decimal
|
||
from model_bakery import baker
|
||
from rest_framework.test import APIClient
|
||
from rest_framework_simplejwt.tokens import RefreshToken
|
||
|
||
from apps.catalog.models import Product
|
||
from apps.inventory.models import Warehouse
|
||
from apps.partner.models import Customer
|
||
from apps.finance.models import Receivable, Receipt, Allocation
|
||
from apps.notify.models import Notification, AlertRule
|
||
from apps.sales.models import SalesBill
|
||
from apps.ai import risk as ai_risk
|
||
|
||
|
||
@pytest.fixture
|
||
def auth_client(db, user, tenant):
|
||
c = APIClient()
|
||
refresh = RefreshToken.for_user(user)
|
||
c.credentials(
|
||
HTTP_AUTHORIZATION=f"Bearer {refresh.access_token}",
|
||
HTTP_X_TENANT_ID=tenant.code,
|
||
)
|
||
return c
|
||
|
||
|
||
@pytest.fixture
|
||
def customer(db, tenant):
|
||
return baker.make(Customer, tenant=tenant, code="C001", name="张三便利店",
|
||
credit_limit=Decimal("5000"))
|
||
|
||
|
||
@pytest.fixture
|
||
def warehouse(db, tenant):
|
||
return baker.make(Warehouse, tenant=tenant, code="WH01", name="主仓")
|
||
|
||
|
||
@pytest.fixture
|
||
def product(db, tenant):
|
||
return baker.make(Product, tenant=tenant, code="P001", name="可乐",
|
||
sale_price=Decimal("10"))
|
||
|
||
|
||
def _receivable(tenant, customer, amount, *, bill_date, paid=0, no="RC001"):
|
||
return baker.make(
|
||
Receivable, tenant=tenant, customer=customer, bill_no=no,
|
||
bill_date=bill_date, total_amount=Decimal(str(amount)),
|
||
paid_amount=Decimal(str(paid)),
|
||
status="paid" if paid and Decimal(str(paid)) >= Decimal(str(amount)) else
|
||
("partial" if paid else "open"),
|
||
)
|
||
|
||
|
||
def _settle(tenant, customer, recv, *, receipt_date, method="现金"):
|
||
"""造一笔收款 + 核销，用于回款周期计算。"""
|
||
receipt = baker.make(
|
||
Receipt, tenant=tenant, customer=customer, bill_no=f"SK{recv.bill_no}",
|
||
bill_date=receipt_date, amount=recv.total_amount, status="posted", method=method,
|
||
)
|
||
baker.make(
|
||
Allocation, tenant=tenant, kind="receipt",
|
||
receivable=recv, receipt=receipt, amount=recv.total_amount,
|
||
)
|
||
return receipt
|
||
|
||
|
||
# ---------- 因子：回款周期漂移 ----------
|
||
|
||
|
||
def test_cycle_drift_zero_when_no_samples(tenant, customer):
|
||
"""无核销记录 → 不计分（不误报）。"""
|
||
out = ai_risk.cycle_drift_score(tenant, customer)
|
||
assert Decimal(str(out["score"])) == 0
|
||
assert "样本不足" in out["reason"]
|
||
|
||
|
||
def test_cycle_drift_positive_when_slower(tenant, customer):
|
||
"""历史 10 天回款，近期 40 天回款 → 明显变慢，得分高。"""
|
||
today = date.today()
|
||
# 基线：1 年前 3 笔，10 天回款
|
||
for i in range(3):
|
||
d = today - timedelta(days=300 + i * 10)
|
||
r = _receivable(tenant, customer, 100, bill_date=d, no=f"RC-B{i}")
|
||
_settle(tenant, customer, r, receipt_date=d + timedelta(days=10))
|
||
# 近期：40 天回款（仍在 90 天窗口内）
|
||
r2 = _receivable(tenant, customer, 200, bill_date=today - timedelta(days=50), no="RC-N1")
|
||
_settle(tenant, customer, r2, receipt_date=today - timedelta(days=10))
|
||
out = ai_risk.cycle_drift_score(tenant, customer, today=today)
|
||
assert Decimal(str(out["score"])) > 50, out
|
||
assert out["recent_days"] > out["baseline_days"]
|
||
|
||
|
||
def test_cycle_drift_zero_when_faster(tenant, customer):
|
||
"""近期回款更快 → 不得分。"""
|
||
today = date.today()
|
||
for i in range(3):
|
||
d = today - timedelta(days=300 + i * 10)
|
||
r = _receivable(tenant, customer, 100, bill_date=d, no=f"RC-S{i}")
|
||
_settle(tenant, customer, r, receipt_date=d + timedelta(days=60))
|
||
r2 = _receivable(tenant, customer, 200, bill_date=today - timedelta(days=30), no="RC-F1")
|
||
_settle(tenant, customer, r2, receipt_date=today - timedelta(days=25))
|
||
out = ai_risk.cycle_drift_score(tenant, customer, today=today)
|
||
assert Decimal(str(out["score"])) == 0, out
|
||
|
||
|
||
# ---------- 因子：欠款趋势 ----------
|
||
|
||
|
||
def test_outstanding_trend_positive_on_growth(tenant, customer):
|
||
"""上月欠 1000，本月欠 1600（+60%）→ 高分。"""
|
||
today = date.today()
|
||
this_month_start = today.replace(day=1)
|
||
last_month_end = this_month_start - timedelta(days=1)
|
||
_receivable(tenant, customer, 1000, bill_date=last_month_end - timedelta(days=5),
|
||
no="RC-OLD")
|
||
_receivable(tenant, customer, 600, bill_date=today, no="RC-NEW")
|
||
# 注意：本月口径含上月未结，故 as_of 今天 = 1600，上月 = 1000
|
||
out = ai_risk.outstanding_trend_score(tenant, customer, today=today)
|
||
assert Decimal(str(out["score"])) > 50, out
|
||
assert out["current"] > out["previous"]
|
||
|
||
|
||
def test_outstanding_trend_zero_on_decline(tenant, customer):
|
||
today = date.today()
|
||
this_month_start = today.replace(day=1)
|
||
last_month_end = this_month_start - timedelta(days=1)
|
||
_receivable(tenant, customer, 2000, bill_date=last_month_end - timedelta(days=3),
|
||
paid=1500, no="RC-OLD2")
|
||
out = ai_risk.outstanding_trend_score(tenant, customer, today=today)
|
||
assert Decimal(str(out["score"])) == 0, out
|
||
|
||
|
||
def test_outstanding_trend_new_debt_when_no_history(tenant, customer):
|
||
"""上月无欠款、本月有 → 50 分（新增欠款，但不直接满分）。"""
|
||
today = date.today()
|
||
_receivable(tenant, customer, 800, bill_date=today, no="RC-FRESH")
|
||
out = ai_risk.outstanding_trend_score(tenant, customer, today=today)
|
||
assert Decimal(str(out["score"])) == 50, out
|
||
assert "新增" in out["reason"]
|
||
|
||
|
||
# ---------- 因子：开单频次骤降 ----------
|
||
|
||
|
||
def test_order_drop_zero_without_outstanding(tenant, customer, warehouse, product):
|
||
out = ai_risk.order_drop_score(tenant, customer)
|
||
assert Decimal(str(out["score"])) == 0
|
||
assert "无欠款" in out["reason"]
|
||
|
||
|
||
def test_order_drop_high_when_long_gap(tenant, customer, warehouse, product):
|
||
"""有欠款 + 90 天未开单 → 100 分。"""
|
||
today = date.today()
|
||
_receivable(tenant, customer, 500, bill_date=today - timedelta(days=100), no="RC-GAP")
|
||
from apps.sales import services as sales_services
|
||
bill = sales_services.create_sales_bill(
|
||
tenant=tenant, customer=customer, warehouse=warehouse,
|
||
lines=[{"product": product, "quantity": 1, "unit_price": 10}],
|
||
bill_date=today - timedelta(days=95),
|
||
)
|
||
out = ai_risk.order_drop_score(tenant, customer, today=today)
|
||
assert Decimal(str(out["score"])) == 100, out
|
||
assert out["days_since_last_bill"] >= 90
|
||
|
||
|
||
def test_order_drop_zero_within_30_days(tenant, customer, warehouse, product):
|
||
today = date.today()
|
||
_receivable(tenant, customer, 500, bill_date=today, no="RC-RECENT")
|
||
from apps.sales import services as sales_services
|
||
sales_services.create_sales_bill(
|
||
tenant=tenant, customer=customer, warehouse=warehouse,
|
||
lines=[{"product": product, "quantity": 1, "unit_price": 10}],
|
||
bill_date=today - timedelta(days=5),
|
||
)
|
||
out = ai_risk.order_drop_score(tenant, customer, today=today)
|
||
assert Decimal(str(out["score"])) == 0, out
|
||
|
||
|
||
# ---------- 综合评分 ----------
|
||
|
||
|
||
def test_score_customer_high_risk_combo(tenant, customer, warehouse, product):
|
||
"""回款变慢 + 欠款上升 + 长期未开单 → high 档。"""
|
||
today = date.today()
|
||
this_month_start = today.replace(day=1)
|
||
last_month_end = this_month_start - timedelta(days=1)
|
||
# 基线快回款
|
||
for i in range(3):
|
||
d = today - timedelta(days=320 + i * 10)
|
||
r = _receivable(tenant, customer, 100, bill_date=d, no=f"RC-H{i}")
|
||
_settle(tenant, customer, r, receipt_date=d + timedelta(days=5))
|
||
# 近期待收：慢回款 + 未结欠款
|
||
r_recent = _receivable(tenant, customer, 300, bill_date=today - timedelta(days=45),
|
||
no="RC-HN")
|
||
_settle(tenant, customer, r_recent, receipt_date=today - timedelta(days=3))
|
||
_receivable(tenant, customer, 900, bill_date=last_month_end - timedelta(days=2),
|
||
no="RC-HOLD1")
|
||
_receivable(tenant, customer, 700, bill_date=today, no="RC-HOLD2")
|
||
|
||
data = ai_risk.score_customer(tenant, customer, today=today)
|
||
assert data["level"] == "high", data
|
||
assert data["score"] >= 70
|
||
assert set(data["factors"]) == {"cycle_drift", "outstanding_trend", "order_drop"}
|
||
|
||
|
||
def test_score_customer_low_when_healthy(tenant, customer, warehouse, product):
|
||
"""按时回款 + 无欠款 → low 档，分数低。"""
|
||
today = date.today()
|
||
r = _receivable(tenant, customer, 300, bill_date=today - timedelta(days=20),
|
||
paid=300, no="RC-OK")
|
||
_settle(tenant, customer, r, receipt_date=today - timedelta(days=15))
|
||
data = ai_risk.score_customer(tenant, customer, today=today)
|
||
assert data["level"] == "low", data
|
||
assert data["score"] < 40
|
||
|
||
|
||
def test_risk_ranking_orders_by_score(tenant, warehouse, product):
|
||
"""排行按分数降序，且只含有效应收客户。"""
|
||
today = date.today()
|
||
c_high = baker.make(Customer, tenant=tenant, code="C-H", name="高风险客户",
|
||
credit_limit=Decimal("1000"))
|
||
c_low = baker.make(Customer, tenant=tenant, code="C-L", name="低风险客户",
|
||
credit_limit=Decimal("1000"))
|
||
# 高风险：大额欠款 + 长期无单
|
||
_receivable(tenant, c_high, 2000, bill_date=today - timedelta(days=120), no="RC-HH")
|
||
# 低风险：今天刚开单 + 小额欠款
|
||
_receivable(tenant, c_low, 50, bill_date=today, no="RC-LL")
|
||
from apps.sales import services as sales_services
|
||
for c in (c_low,):
|
||
sales_services.create_sales_bill(
|
||
tenant=tenant, customer=c, warehouse=warehouse,
|
||
lines=[{"product": product, "quantity": 1, "unit_price": 10}],
|
||
bill_date=today - timedelta(days=2),
|
||
)
|
||
|
||
ranking = ai_risk.risk_ranking(tenant, top_n=5, today=today)
|
||
assert len(ranking) == 2
|
||
assert ranking[0]["customer_code"] == "C-H"
|
||
assert ranking[0]["score"] >= ranking[1]["score"]
|
||
|
||
|
||
def test_risk_ranking_empty_without_receivables(tenant, customer):
|
||
assert ai_risk.risk_ranking(tenant, top_n=5) == []
|
||
|
||
|
||
# ---------- API ----------
|
||
|
||
|
||
def test_risk_ranking_api(db, auth_client, tenant, customer):
|
||
today = date.today()
|
||
_receivable(tenant, customer, 1200, bill_date=today - timedelta(days=100), no="RC-API")
|
||
resp = auth_client.get("/api/v1/ai/risk/ranking/?top_n=5")
|
||
assert resp.status_code == 200, resp.content
|
||
body = resp.json()
|
||
assert body["llm_enhanced"] is False # 测试环境无 KEY → 降级
|
||
assert body["count"] >= 1
|
||
assert body["results"][0]["customer_code"] == "C001"
|
||
|
||
|
||
def test_customer_risk_api(db, auth_client, tenant, customer):
|
||
resp = auth_client.get(f"/api/v1/ai/risk/customer/{customer.id}/")
|
||
assert resp.status_code == 200, resp.content
|
||
body = resp.json()
|
||
assert body["customer_code"] == "C001"
|
||
assert "factors" in body
|
||
|
||
|
||
def test_customer_risk_api_404_other_tenant(db, auth_client, tenant, other_tenant):
|
||
other = baker.make(Customer, tenant=other_tenant, code="X1", name="别家客户")
|
||
resp = auth_client.get(f"/api/v1/ai/risk/customer/{other.id}/")
|
||
assert resp.status_code == 404
|
||
|
||
|
||
def test_collection_advice_falls_back_to_statistics(db, auth_client, tenant, customer):
|
||
"""无 KEY → advice 回落到统计理由，且 llm_enhanced=False（不 500）。"""
|
||
today = date.today()
|
||
_receivable(tenant, customer, 2000, bill_date=today - timedelta(days=120), no="RC-ADV")
|
||
resp = auth_client.get(f"/api/v1/ai/risk/advice/{customer.id}/")
|
||
assert resp.status_code == 200, resp.content
|
||
body = resp.json()
|
||
assert body["llm_enhanced"] is False
|
||
assert body["advice"] == body["stat_reason"]
|
||
assert "风险分" in body["advice"]
|
||
|
||
|
||
# ---------- 预警规则 + 扫描 ----------
|
||
|
||
|
||
def test_risk_scan_creates_notification(db, auth_client, tenant, customer):
|
||
today = date.today()
|
||
_receivable(tenant, customer, 3000, bill_date=today - timedelta(days=120), no="RC-SCAN")
|
||
AlertRule.objects.create(
|
||
tenant=tenant, code="ai_risk_score", name="AI 风险预警",
|
||
rule_type="risk_score", threshold=Decimal("10"), is_enabled=True,
|
||
)
|
||
resp = auth_client.post("/api/v1/ai/risk/scan/", {}, format="json")
|
||
assert resp.status_code == 200, resp.content
|
||
assert resp.json()["created_count"] >= 1
|
||
n = Notification.objects.filter(tenant=tenant, extra_data__rule="risk_score")
|
||
assert n.count() >= 1
|
||
assert "应收风险" in n.first().title
|
||
|
||
|
||
def test_risk_scan_dedupes_same_day(db, auth_client, tenant, customer):
|
||
"""同日二次扫描不重复发同客户预警。"""
|
||
today = date.today()
|
||
_receivable(tenant, customer, 3000, bill_date=today - timedelta(days=120), no="RC-DD")
|
||
AlertRule.objects.create(
|
||
tenant=tenant, code="ai_risk_score", name="AI 风险预警",
|
||
rule_type="risk_score", threshold=Decimal("10"), is_enabled=True,
|
||
)
|
||
auth_client.post("/api/v1/ai/risk/scan/", {}, format="json")
|
||
resp2 = auth_client.post("/api/v1/ai/risk/scan/", {}, format="json")
|
||
assert resp2.json()["created_count"] == 0
|
||
|
||
|
||
def test_risk_scan_skipped_without_enabled_rule(db, auth_client, tenant, customer):
|
||
"""规则未启用 → 不扫描（尊重配置）。"""
|
||
today = date.today()
|
||
_receivable(tenant, customer, 3000, bill_date=today - timedelta(days=120), no="RC-NR")
|
||
resp = auth_client.post("/api/v1/ai/risk/scan/", {}, format="json")
|
||
assert resp.status_code == 200
|
||
assert resp.json()["created_count"] == 0
|
||
|
||
|
||
def test_init_default_rules_includes_risk(db, tenant):
|
||
from apps.notify.services import init_default_alert_rules
|
||
init_default_alert_rules(tenant)
|
||
assert AlertRule.objects.filter(tenant=tenant, rule_type="risk_score").exists()
|
||
|
||
|
||
def test_run_all_alert_checks_includes_risk_count(db, tenant, customer):
|
||
from apps.notify.services import run_all_alert_checks
|
||
result = run_all_alert_checks(tenant)
|
||
assert "risk_score_alerts_count" in result
|
||
assert "total_alerts" in result
|
||
|
||
|
||
def test_llm_unavailable_by_default(db, settings):
|
||
"""默认无 KEY → available() False（保证离线可跑）。"""
|
||
from apps.ai import llm
|
||
settings.AI_PROVIDER = ""
|
||
settings.AI_API_KEY = ""
|
||
assert llm.available() is False
|
||
assert llm.chat([{"role": "user", "content": "hi"}]) is None
|
||
assert llm.collect_advice("hi") is None
|