selfrelease
fad458b2a7
- UIUX 文档：填充 19 个缺口（多主体画像/健康度/AI+看板/增长域/洞察域/创始人端/OODA/助推/商密） - UIUX 文档：插入 6 个新章节（十四~十九），旧章节重编号为二十~三十一，更新目录和交叉引用 - 作业指导书 x5：导航改为 6 域分组，新增 Context Bar/工作模式/Insight Rail/决策线程/多工作区等 UI 概念 - 新建 docs/2-task-uiux.md：50 个代码落地开发任务，按 P0-P6 分优先级 + 8 Sprint 规划 - 后端/前端：大量新增模型、路由、组件（来自之前 Phase 开发）
361 lines
13 KiB
Python
361 lines
13 KiB
Python
"""Phase 2 Service 层测试 — 财务校验/协议监控/弱信号关联/健康度/及时性追踪。"""
|
|
|
|
import uuid as uuid_mod
|
|
|
|
import pytest
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.services.financial_validator import validate_financial_data
|
|
from app.services.agreement_monitor import check_clause_triggers
|
|
from app.services.signal_correlator import correlate_signals
|
|
from app.services.health_calculator import calculate_health_score, determine_trend
|
|
from app.services.report_tracker import compute_timeliness
|
|
from app.models.financial_data import FinancialData
|
|
from app.models.agreement import InvestmentAgreement
|
|
from app.models.report import MonthlyReport
|
|
from app.models.company import Company
|
|
from app.models.user import User
|
|
from app.models.tenant import Tenant
|
|
from tests.conftest import test_session_factory
|
|
|
|
|
|
@pytest.fixture
|
|
async def db_session():
|
|
"""创建数据库会话。"""
|
|
async with test_session_factory() as session:
|
|
yield session
|
|
await session.rollback()
|
|
|
|
|
|
@pytest.fixture
|
|
async def seed_tenant_company(db_session: AsyncSession):
|
|
"""创建测试租户和企业，返回 (tenant_id, company_id, user_id)。
|
|
|
|
每次调用使用唯一邮箱避免唯一约束冲突。
|
|
"""
|
|
unique = uuid_mod.uuid4().hex[:8]
|
|
tenant = Tenant(name=f"service测试机构_{unique}")
|
|
db_session.add(tenant)
|
|
await db_session.flush()
|
|
|
|
user = User(
|
|
email=f"svc_{unique}@example.com",
|
|
name="测试用户",
|
|
role="investor",
|
|
tenant_id=tenant.id,
|
|
password_hash="fake_hash",
|
|
is_active=True,
|
|
)
|
|
db_session.add(user)
|
|
await db_session.flush()
|
|
|
|
company = Company(
|
|
name=f"service测试公司_{unique}",
|
|
industry="AI",
|
|
tenant_id=tenant.id,
|
|
)
|
|
db_session.add(company)
|
|
await db_session.flush()
|
|
await db_session.commit()
|
|
|
|
return str(tenant.id), str(company.id), str(user.id)
|
|
|
|
|
|
class TestFinancialValidator:
|
|
"""财务数据校验服务测试。"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_validate_no_data(self, db_session: AsyncSession):
|
|
"""无财务数据时应返回可信度 0 和提示。"""
|
|
result = await validate_financial_data(db_session, "nonexistent-id", 2025, 6)
|
|
assert result["credibility_score"] == 0.0
|
|
assert "无财务数据" in result["issues"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_validate_balance_sheet_balanced(self, db_session: AsyncSession, seed_tenant_company):
|
|
"""资产负债表平衡时应通过校验。"""
|
|
_, company_id, _ = seed_tenant_company
|
|
|
|
fd = FinancialData(
|
|
company_id=company_id,
|
|
statement_type="balance_sheet",
|
|
period_year=2025,
|
|
period_month=6,
|
|
data_json={"total_assets": 1000000, "total_liabilities": 600000, "total_equity": 400000},
|
|
)
|
|
db_session.add(fd)
|
|
await db_session.commit()
|
|
|
|
result = await validate_financial_data(db_session, company_id, 2025, 6)
|
|
assert result["checks_total"] >= 1
|
|
assert result["checks_passed"] >= 1
|
|
assert result["credibility_score"] > 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_validate_balance_sheet_unbalanced(self, db_session: AsyncSession, seed_tenant_company):
|
|
"""资产负债表不平时应报告问题。"""
|
|
_, company_id, _ = seed_tenant_company
|
|
|
|
fd = FinancialData(
|
|
company_id=company_id,
|
|
statement_type="balance_sheet",
|
|
period_year=2025,
|
|
period_month=6,
|
|
data_json={"total_assets": 1000000, "total_liabilities": 700000, "total_equity": 200000},
|
|
)
|
|
db_session.add(fd)
|
|
await db_session.commit()
|
|
|
|
result = await validate_financial_data(db_session, company_id, 2025, 6)
|
|
assert any("不平" in issue for issue in result["issues"])
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_validate_negative_revenue(self, db_session: AsyncSession, seed_tenant_company):
|
|
"""收入为负数时应报告异常。"""
|
|
_, company_id, _ = seed_tenant_company
|
|
|
|
fd = FinancialData(
|
|
company_id=company_id,
|
|
statement_type="income",
|
|
period_year=2025,
|
|
period_month=6,
|
|
data_json={"revenue": -50000},
|
|
)
|
|
db_session.add(fd)
|
|
await db_session.commit()
|
|
|
|
result = await validate_financial_data(db_session, company_id, 2025, 6)
|
|
assert any("负数" in issue for issue in result["issues"])
|
|
|
|
|
|
class TestAgreementMonitor:
|
|
"""协议条款监控服务测试。"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_check_no_agreements(self, db_session: AsyncSession, seed_tenant_company):
|
|
"""无协议时应返回空列表。"""
|
|
_, company_id, _ = seed_tenant_company
|
|
alerts = await check_clause_triggers(db_session, company_id)
|
|
assert alerts == []
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_check_agreement_with_rules(self, db_session: AsyncSession, seed_tenant_company):
|
|
"""有监控规则的协议应生成预警。"""
|
|
_, company_id, _ = seed_tenant_company
|
|
|
|
agreement = InvestmentAgreement(
|
|
company_id=company_id,
|
|
title="A轮投资协议",
|
|
status="active",
|
|
monitoring_rules=[
|
|
{"rule": "营收低于 100 万", "metric": "revenue", "threshold": 1000000},
|
|
],
|
|
)
|
|
db_session.add(agreement)
|
|
await db_session.commit()
|
|
|
|
alerts = await check_clause_triggers(db_session, company_id)
|
|
assert len(alerts) == 1
|
|
assert alerts[0]["agreement_title"] == "A轮投资协议"
|
|
assert "营收" in alerts[0]["rule"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_check_inactive_agreement_ignored(self, db_session: AsyncSession, seed_tenant_company):
|
|
"""非 active 状态的协议不应生成预警。"""
|
|
_, company_id, _ = seed_tenant_company
|
|
|
|
agreement = InvestmentAgreement(
|
|
company_id=company_id,
|
|
title="已终止协议",
|
|
status="terminated",
|
|
monitoring_rules=[{"rule": "test", "metric": "revenue", "threshold": 100}],
|
|
)
|
|
db_session.add(agreement)
|
|
await db_session.commit()
|
|
|
|
alerts = await check_clause_triggers(db_session, company_id)
|
|
assert alerts == []
|
|
|
|
|
|
class TestSignalCorrelator:
|
|
"""弱信号关联引擎测试。"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_single_signal_no_correlation(self):
|
|
"""单个信号不应产生关联。"""
|
|
result = await correlate_signals([{"signal_type": "tech", "confidence": 0.8}])
|
|
assert result == []
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_empty_signals(self):
|
|
"""空信号列表不应产生关联。"""
|
|
result = await correlate_signals([])
|
|
assert result == []
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cross_dimension_correlation(self):
|
|
"""不同类型信号应产生跨维度关联。"""
|
|
signals = [
|
|
{"id": "s1", "signal_type": "tech", "confidence": 0.7},
|
|
{"id": "s2", "signal_type": "market", "confidence": 0.6},
|
|
{"id": "s3", "signal_type": "org", "confidence": 0.5},
|
|
]
|
|
result = await correlate_signals(signals)
|
|
assert len(result) >= 1
|
|
assert result[0]["correlation_type"] == "cross_dimension"
|
|
assert result[0]["risk_probability"] > 0
|
|
assert result[0]["risk_probability"] <= 0.95
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_same_type_no_cross_dimension(self):
|
|
"""同类型信号不应产生跨维度关联。"""
|
|
signals = [
|
|
{"id": "s1", "signal_type": "tech", "confidence": 0.7},
|
|
{"id": "s2", "signal_type": "tech", "confidence": 0.6},
|
|
]
|
|
result = await correlate_signals(signals)
|
|
assert result == []
|
|
|
|
|
|
class TestHealthCalculator:
|
|
"""健康度评分计算测试（补充已有 test_health_calculator.py 的边界用例）。"""
|
|
|
|
def test_empty_data_returns_zeros(self):
|
|
"""空数据应返回全 0。"""
|
|
result = calculate_health_score({})
|
|
assert result["total_score"] == 0.0
|
|
assert result["financial_score"] == 0.0
|
|
|
|
def test_healthy_company_high_score(self):
|
|
"""健康企业应获得高分。"""
|
|
data = {
|
|
"cash_balance": {"runway_months": 18},
|
|
"revenue": {"yoy_change": "+25%"},
|
|
"burn_rate": {"trend": "down"},
|
|
"headcount": {"new_hires": 10, "departures": 2},
|
|
"key_metrics": [
|
|
{"name": "AI 推理量", "change": "+30%"},
|
|
{"name": "推理成本", "change": "-15%"},
|
|
],
|
|
}
|
|
result = calculate_health_score(data)
|
|
assert result["total_score"] > 70
|
|
assert result["financial_score"] > 80
|
|
|
|
def test_unhealthy_company_low_score(self):
|
|
"""不健康企业应获得低分。"""
|
|
data = {
|
|
"cash_balance": {"runway_months": 2},
|
|
"revenue": {"yoy_change": "-30%"},
|
|
"burn_rate": {"trend": "up"},
|
|
"headcount": {"new_hires": 0, "departures": 8},
|
|
}
|
|
result = calculate_health_score(data)
|
|
assert result["total_score"] < 50
|
|
assert result["financial_score"] < 40
|
|
|
|
def test_score_range_0_to_100(self):
|
|
"""所有维度分数应在 0-100 范围内。"""
|
|
data = {
|
|
"cash_balance": {"runway_months": 0},
|
|
"revenue": {"yoy_change": "-100%"},
|
|
"burn_rate": {"trend": "up"},
|
|
"headcount": {"new_hires": 0, "departures": 100},
|
|
}
|
|
result = calculate_health_score(data)
|
|
for v in result.values():
|
|
assert 0 <= v <= 100
|
|
|
|
def test_determine_trend_up(self):
|
|
"""评分上升 >5 应为 up。"""
|
|
assert determine_trend(80, 70) == "up"
|
|
|
|
def test_determine_trend_down(self):
|
|
"""评分下降 >5 应为 down。"""
|
|
assert determine_trend(60, 70) == "down"
|
|
|
|
def test_determine_trend_stable(self):
|
|
"""评分变化 <=5 应为 stable。"""
|
|
assert determine_trend(72, 70) == "stable"
|
|
|
|
def test_determine_trend_no_previous(self):
|
|
"""无上期评分应为 stable。"""
|
|
assert determine_trend(75, None) == "stable"
|
|
|
|
|
|
class TestReportTracker:
|
|
"""月报提交及时性追踪测试。"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_compute_timeliness_no_reports(self, db_session: AsyncSession, seed_tenant_company):
|
|
"""无月报时应返回空列表。"""
|
|
tenant_id, _, _ = seed_tenant_company
|
|
result = await compute_timeliness(db_session, tenant_id)
|
|
assert result == []
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_compute_timeliness_on_time(self, db_session: AsyncSession, seed_tenant_company):
|
|
"""按时提交应延迟 0 天。"""
|
|
tenant_id, company_id, user_id = seed_tenant_company
|
|
|
|
# 2025年6月月报，应提交日期为 2025-07-10
|
|
report = MonthlyReport(
|
|
company_id=company_id,
|
|
period_year=2025,
|
|
period_month=6,
|
|
status="submitted",
|
|
raw_content="6月月报",
|
|
submitted_at=__import__("datetime").datetime(2025, 7, 10, tzinfo=__import__("datetime").timezone.utc),
|
|
structured_data={"revenue": 100, "cash_balance": {}, "burn_rate": {}, "headcount": {}, "key_metrics": [], "highlights": "", "concerns": ""},
|
|
)
|
|
db_session.add(report)
|
|
await db_session.commit()
|
|
|
|
result = await compute_timeliness(db_session, tenant_id)
|
|
assert len(result) == 1
|
|
assert result[0]["delay_days"] == 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_compute_timeliness_late(self, db_session: AsyncSession, seed_tenant_company):
|
|
"""延迟提交应 delay_days > 0。"""
|
|
tenant_id, company_id, _ = seed_tenant_company
|
|
|
|
report = MonthlyReport(
|
|
company_id=company_id,
|
|
period_year=2025,
|
|
period_month=6,
|
|
status="submitted",
|
|
raw_content="6月月报",
|
|
submitted_at=__import__("datetime").datetime(2025, 7, 20, tzinfo=__import__("datetime").timezone.utc),
|
|
structured_data={"revenue": 100},
|
|
)
|
|
db_session.add(report)
|
|
await db_session.commit()
|
|
|
|
result = await compute_timeliness(db_session, tenant_id)
|
|
assert len(result) == 1
|
|
assert result[0]["delay_days"] == 10
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_compute_timeliness_quality_score(self, db_session: AsyncSession, seed_tenant_company):
|
|
"""数据质量评分应基于结构化字段完整度。"""
|
|
tenant_id, company_id, _ = seed_tenant_company
|
|
|
|
# 只填了 1/8 个字段
|
|
report = MonthlyReport(
|
|
company_id=company_id,
|
|
period_year=2025,
|
|
period_month=6,
|
|
status="submitted",
|
|
raw_content="6月月报",
|
|
submitted_at=__import__("datetime").datetime(2025, 7, 10, tzinfo=__import__("datetime").timezone.utc),
|
|
structured_data={"revenue": 100},
|
|
)
|
|
db_session.add(report)
|
|
await db_session.commit()
|
|
|
|
result = await compute_timeliness(db_session, tenant_id)
|
|
assert len(result) == 1
|
|
# 1/8 = 12.5%
|
|
assert result[0]["quality_score"] == 12.5
|