feat(backend): AI 服务层 — 千问流式 LLM + 月报解析 + 健康度计算 + 风险检测 + Copilot
- LLM 客户端:全部 SSE 流式输出,兼容 OpenAI 接口
- AI 月报解析:SSE 流式端点 POST /reports/{id}/parse
- 健康度计算引擎:四维评分(财务/经营/AI商业化/AI成本)
- 风险自动检测引擎:6 条规则自动检测指标越界
- AI Copilot:SSE 流式对话 POST /copilot/chat
- 权限中间件:角色级 + 字段级权限控制
- 测试:21 个新测试(健康度 8 + 风险检测 8 + 权限 5),总计 64 passed
This commit is contained in:
@@ -0,0 +1,65 @@
|
||||
"""健康度计算引擎测试。"""
|
||||
|
||||
from app.services.health_calculator import calculate_health_score, determine_trend
|
||||
|
||||
|
||||
class TestCalculateHealthScore:
|
||||
"""健康度评分计算。"""
|
||||
|
||||
def test_empty_data(self):
|
||||
"""空数据应返回全 0。"""
|
||||
result = calculate_health_score({})
|
||||
assert result["total_score"] == 0.0
|
||||
assert result["financial_score"] == 0.0
|
||||
|
||||
def test_healthy_company(self):
|
||||
"""健康企业:跑道充足 + 营收增长 + 烧钱下降。"""
|
||||
data = {
|
||||
"revenue": {"yoy_change": "30"},
|
||||
"cash_balance": {"runway_months": 18},
|
||||
"burn_rate": {"trend": "down"},
|
||||
"headcount": {"new_hires": 5, "departures": 1},
|
||||
"key_metrics": [{"name": "ARR", "change": "+25%"}],
|
||||
}
|
||||
result = calculate_health_score(data)
|
||||
assert result["total_score"] > 70
|
||||
assert result["financial_score"] > 80
|
||||
|
||||
def test_unhealthy_company(self):
|
||||
"""不健康企业:跑道短 + 营收下滑 + 烧钱上升。"""
|
||||
data = {
|
||||
"revenue": {"yoy_change": "-20"},
|
||||
"cash_balance": {"runway_months": 2},
|
||||
"burn_rate": {"trend": "up"},
|
||||
"headcount": {"new_hires": 0, "departures": 8},
|
||||
}
|
||||
result = calculate_health_score(data)
|
||||
assert result["total_score"] < 50
|
||||
assert result["financial_score"] < 30
|
||||
|
||||
def test_score_range(self):
|
||||
"""评分应在 0-100 范围内。"""
|
||||
data = {
|
||||
"revenue": {"yoy_change": "1000"},
|
||||
"cash_balance": {"runway_months": 100},
|
||||
"burn_rate": {"trend": "down"},
|
||||
}
|
||||
result = calculate_health_score(data)
|
||||
for key, val in result.items():
|
||||
assert 0 <= val <= 100
|
||||
|
||||
|
||||
class TestDetermineTrend:
|
||||
"""趋势判断。"""
|
||||
|
||||
def test_up(self):
|
||||
assert determine_trend(80, 60) == "up"
|
||||
|
||||
def test_down(self):
|
||||
assert determine_trend(50, 70) == "down"
|
||||
|
||||
def test_stable(self):
|
||||
assert determine_trend(60, 62) == "stable"
|
||||
|
||||
def test_no_previous(self):
|
||||
assert determine_trend(70, None) == "stable"
|
||||
@@ -0,0 +1,43 @@
|
||||
"""权限中间件测试。"""
|
||||
|
||||
from types import SimpleNamespace
|
||||
|
||||
from app.core.permissions import filter_fields, require_min_role, require_role, ROLE_HIERARCHY
|
||||
|
||||
|
||||
class TestRoleHierarchy:
|
||||
"""角色层级。"""
|
||||
|
||||
def test_admin_highest(self):
|
||||
assert ROLE_HIERARCHY["admin"] > ROLE_HIERARCHY["investor"]
|
||||
assert ROLE_HIERARCHY["admin"] > ROLE_HIERARCHY["founder"]
|
||||
|
||||
def test_investor_above_founder(self):
|
||||
assert ROLE_HIERARCHY["investor"] > ROLE_HIERARCHY["founder"]
|
||||
|
||||
|
||||
class TestFilterFields:
|
||||
"""字段级权限过滤。"""
|
||||
|
||||
def test_investor_sees_all(self):
|
||||
"""investor 可见全部字段。"""
|
||||
data = {"name": "公司A", "total_funding": "1亿", "description": "测试"}
|
||||
user = SimpleNamespace(role="investor")
|
||||
result = filter_fields("company", data, user)
|
||||
assert result == data
|
||||
|
||||
def test_founder_filtered(self):
|
||||
"""founder 只能看限定字段。"""
|
||||
data = {"name": "公司A", "total_funding": "1亿", "description": "测试", "id": "123"}
|
||||
user = SimpleNamespace(role="founder")
|
||||
result = filter_fields("company", data, user)
|
||||
assert "name" in result
|
||||
assert "id" in result
|
||||
assert "total_funding" not in result
|
||||
|
||||
def test_admin_sees_all(self):
|
||||
"""admin 可见全部字段。"""
|
||||
data = {"name": "公司A", "total_funding": "1亿"}
|
||||
user = SimpleNamespace(role="admin")
|
||||
result = filter_fields("company", data, user)
|
||||
assert result == data
|
||||
@@ -0,0 +1,61 @@
|
||||
"""风险检测引擎测试。"""
|
||||
|
||||
from app.services.risk_engine import detect_risks
|
||||
|
||||
|
||||
class TestDetectRisks:
|
||||
"""风险自动检测。"""
|
||||
|
||||
def test_empty_data(self):
|
||||
"""空数据不应检测到风险。"""
|
||||
risks = detect_risks({}, "company-1")
|
||||
assert len(risks) == 0
|
||||
|
||||
def test_low_runway_critical(self):
|
||||
"""跑道 < 3 月应触发 critical 风险。"""
|
||||
data = {"cash_balance": {"runway_months": 2}}
|
||||
risks = detect_risks(data, "company-1")
|
||||
assert any(r["severity"] == "critical" for r in risks)
|
||||
assert any("3 个月" in r["title"] for r in risks)
|
||||
|
||||
def test_low_runway_warning(self):
|
||||
"""跑道 3-6 月应触发 high 风险。"""
|
||||
data = {"cash_balance": {"runway_months": 4}}
|
||||
risks = detect_risks(data, "company-1")
|
||||
assert any(r["severity"] == "high" and "6 个月" in r["title"] for r in risks)
|
||||
|
||||
def test_burn_rate_up(self):
|
||||
"""烧钱率上升应触发 medium 风险。"""
|
||||
data = {"burn_rate": {"trend": "up"}}
|
||||
risks = detect_risks(data, "company-1")
|
||||
assert any(r["severity"] == "medium" and "烧钱率" in r["title"] for r in risks)
|
||||
|
||||
def test_revenue_decline(self):
|
||||
"""营收下滑应触发 high 风险。"""
|
||||
data = {"revenue": {"yoy_change": "-15"}}
|
||||
risks = detect_risks(data, "company-1")
|
||||
assert any(r["severity"] == "high" and "营收" in r["title"] for r in risks)
|
||||
|
||||
def test_high_departures(self):
|
||||
"""高离职率应触发 medium 风险。"""
|
||||
data = {"headcount": {"new_hires": 2, "departures": 8}}
|
||||
risks = detect_risks(data, "company-1")
|
||||
assert any(r["severity"] == "medium" and "流失" in r["title"] for r in risks)
|
||||
|
||||
def test_healthy_company_no_risks(self):
|
||||
"""健康企业不应检测到风险。"""
|
||||
data = {
|
||||
"revenue": {"yoy_change": "20"},
|
||||
"cash_balance": {"runway_months": 18},
|
||||
"burn_rate": {"trend": "down"},
|
||||
"headcount": {"new_hires": 5, "departures": 1},
|
||||
}
|
||||
risks = detect_risks(data, "company-1")
|
||||
assert len(risks) == 0
|
||||
|
||||
def test_company_id_in_risks(self):
|
||||
"""风险事件应包含 company_id。"""
|
||||
data = {"cash_balance": {"runway_months": 2}}
|
||||
risks = detect_risks(data, "test-company-id")
|
||||
for r in risks:
|
||||
assert r["company_id"] == "test-company-id"
|
||||
Reference in New Issue
Block a user