dieya_risk_prediction_backend/tests/test_risk_prediction.py
kx_guo 1d58bb8913 按需求完善供应商频次分析
- 统一供应商异常统计为近3年数据口径
- 以近3年异常记录数作为合作频次代理值并明确提示
- 实现 A/B 缺陷占比与 S/A/B/C 层级判定
- 按供应商层级输出正常合作、加强巡检、驻厂及备选供应商等管理建议
- 完善风险卡片和供应商卡片的频次、层级、建议展示
- 补充供应商画像接口、测试和历史会话卡片兼容处理
2026-09-07 13:53:29 +08:00

54 lines
1.6 KiB
Python

from app.domain.risk_engine import BaselineRiskEngine, IndicatorValue
from app.main import app
from fastapi.testclient import TestClient
def test_preview_requires_authentication() -> None:
response = TestClient(app).post(
"/api/v1/risk-predictions/preview",
json={
"subjectId": "customer-001",
"indicators": [
{"name": "debtBurden", "value": 80, "weight": 3},
{"name": "cashflowVolatility", "value": 20, "weight": 1},
],
},
)
assert response.status_code == 401
def test_supplier_profile_requires_authentication() -> None:
response = TestClient(app).get("/api/v1/suppliers")
assert response.status_code == 401
def test_baseline_engine_calculates_weighted_score() -> None:
score = BaselineRiskEngine().predict(
[
IndicatorValue(name="debtBurden", value=80, weight=3),
IndicatorValue(name="cashflowVolatility", value=20, weight=1),
]
)
assert score.score == 35.0
assert score.level.value == "high"
assert [factor.contribution for factor in score.factors] == [15.0, 20.0]
def test_risk_level_boundaries_follow_prd() -> None:
engine = BaselineRiskEngine()
cases = (
(80, "low"),
(60, "moderately_low"),
(40, "medium"),
(20, "high"),
(19.99, "very_high"),
)
for score, expected_level in cases:
result = engine.predict([IndicatorValue(name="risk", value=100 - score, weight=1)])
assert result.score == score
assert result.level.value == expected_level