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 开发）
118 lines
3.9 KiB
Python
118 lines
3.9 KiB
Python
"""T2.2 财务数据校验测试。
|
|
|
|
测试财务数据校验 Agent 的内部一致性检查、跨期一致性检查和历史偏差追踪。
|
|
"""
|
|
|
|
import uuid as uuid_mod
|
|
|
|
import pytest
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.company import Company
|
|
from app.models.financial_data import FinancialData
|
|
from app.models.tenant import Tenant
|
|
from app.models.user import User
|
|
from app.services.financial_validator import validate_financial_data
|
|
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_data(db_session: AsyncSession):
|
|
"""创建测试租户、用户和企业，返回 (tenant_id, company_id)。"""
|
|
unique = uuid_mod.uuid4().hex[:8]
|
|
tenant = Tenant(name=f"fin校验机构_{unique}")
|
|
db_session.add(tenant)
|
|
await db_session.flush()
|
|
|
|
user = User(
|
|
email=f"fin_{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"fin校验公司_{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)
|
|
|
|
|
|
class TestValidateFinancialData:
|
|
"""财务数据校验测试。"""
|
|
|
|
async def test_no_data(self, db_session: AsyncSession, seed_data):
|
|
"""无财务数据时返回可信度 0。"""
|
|
_, company_id = seed_data
|
|
result = await validate_financial_data(db_session, company_id, 2025, 6)
|
|
assert result["credibility_score"] == 0.0
|
|
assert "无财务数据" in result["issues"]
|
|
|
|
async def test_balance_sheet_consistent(self, db_session: AsyncSession, seed_data):
|
|
"""资产负债表平衡时通过校验。"""
|
|
_, company_id = seed_data
|
|
stmt = FinancialData(
|
|
company_id=company_id,
|
|
period_year=2025,
|
|
period_month=6,
|
|
statement_type="balance_sheet",
|
|
data_json={"total_assets": 1000, "total_liabilities": 600, "total_equity": 400},
|
|
)
|
|
db_session.add(stmt)
|
|
await db_session.flush()
|
|
await db_session.commit()
|
|
|
|
result = await validate_financial_data(db_session, company_id, 2025, 6)
|
|
assert result["credibility_score"] > 0
|
|
assert result["checks_passed"] >= 1
|
|
|
|
async def test_balance_sheet_inconsistent(self, db_session: AsyncSession, seed_data):
|
|
"""资产负债表不平衡时报错。"""
|
|
_, company_id = seed_data
|
|
stmt = FinancialData(
|
|
company_id=company_id,
|
|
period_year=2025,
|
|
period_month=6,
|
|
statement_type="balance_sheet",
|
|
data_json={"total_assets": 1000, "total_liabilities": 700, "total_equity": 200},
|
|
)
|
|
db_session.add(stmt)
|
|
await db_session.flush()
|
|
await db_session.commit()
|
|
|
|
result = await validate_financial_data(db_session, company_id, 2025, 6)
|
|
assert any("不平" in issue for issue in result["issues"])
|
|
|
|
async def test_income_negative_revenue(self, db_session: AsyncSession, seed_data):
|
|
"""收入为负数时标记异常。"""
|
|
_, company_id = seed_data
|
|
stmt = FinancialData(
|
|
company_id=company_id,
|
|
period_year=2025,
|
|
period_month=6,
|
|
statement_type="income",
|
|
data_json={"revenue": -100},
|
|
)
|
|
db_session.add(stmt)
|
|
await db_session.flush()
|
|
await db_session.commit()
|
|
|
|
result = await validate_financial_data(db_session, company_id, 2025, 6)
|
|
assert any("收入为负" in issue for issue in result["issues"])
|