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 开发）
128 lines
4.2 KiB
Python
128 lines
4.2 KiB
Python
"""T2.3 投资协议解析与监控测试。
|
|
|
|
测试 AI 协议解析和条款监控引擎。
|
|
"""
|
|
|
|
import uuid as uuid_mod
|
|
|
|
import pytest
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.agreement import InvestmentAgreement
|
|
from app.models.company import Company
|
|
from app.models.tenant import Tenant
|
|
from app.models.user import User
|
|
from app.services.agreement_monitor import check_clause_triggers
|
|
from app.services.agreement_parser import parse_agreement
|
|
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_company(db_session: AsyncSession):
|
|
"""创建测试企业和租户，返回 company_id。"""
|
|
unique = uuid_mod.uuid4().hex[:8]
|
|
tenant = Tenant(name=f"agree机构_{unique}")
|
|
db_session.add(tenant)
|
|
await db_session.flush()
|
|
|
|
user = User(
|
|
email=f"agree_{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"agree公司_{unique}", industry="AI", tenant_id=tenant.id)
|
|
db_session.add(company)
|
|
await db_session.flush()
|
|
await db_session.commit()
|
|
return str(company.id)
|
|
|
|
|
|
class TestParseAgreement:
|
|
"""协议解析测试。"""
|
|
|
|
async def test_parse_returns_dict(self):
|
|
"""AI 解析协议返回字典结构。"""
|
|
result = await parse_agreement("本投资协议约定估值 5000 万元，对赌条款要求 2025 年营收达到 2000 万元。")
|
|
assert isinstance(result, dict)
|
|
|
|
async def test_parse_empty_text(self):
|
|
"""空文本仍返回结构化空结果。"""
|
|
result = await parse_agreement("")
|
|
assert isinstance(result, dict)
|
|
|
|
|
|
class TestCheckClauseTriggers:
|
|
"""条款监控测试。"""
|
|
|
|
async def test_no_agreements(self, db_session: AsyncSession, seed_company):
|
|
"""无协议时返回空列表。"""
|
|
company_id = seed_company
|
|
result = await check_clause_triggers(db_session, company_id)
|
|
assert result == []
|
|
|
|
async def test_agreement_with_rules(self, db_session: AsyncSession, seed_company):
|
|
"""有监控规则的协议生成预警。"""
|
|
company_id = seed_company
|
|
agreement = InvestmentAgreement(
|
|
company_id=company_id,
|
|
title="A 轮投资协议",
|
|
status="active",
|
|
monitoring_rules=[
|
|
{"rule": "营收不低于 1000 万", "metric": "revenue", "threshold": "1000"},
|
|
],
|
|
)
|
|
db_session.add(agreement)
|
|
await db_session.flush()
|
|
await db_session.commit()
|
|
|
|
result = await check_clause_triggers(db_session, company_id)
|
|
assert len(result) == 1
|
|
assert result[0]["agreement_title"] == "A 轮投资协议"
|
|
assert result[0]["rule"] == "营收不低于 1000 万"
|
|
|
|
async def test_inactive_agreement_skipped(self, db_session: AsyncSession, seed_company):
|
|
"""非 active 状态的协议不生成预警。"""
|
|
company_id = seed_company
|
|
agreement = InvestmentAgreement(
|
|
company_id=company_id,
|
|
title="已终止协议",
|
|
status="terminated",
|
|
monitoring_rules=[{"rule": "测试规则", "metric": "revenue", "threshold": "100"}],
|
|
)
|
|
db_session.add(agreement)
|
|
await db_session.flush()
|
|
await db_session.commit()
|
|
|
|
result = await check_clause_triggers(db_session, company_id)
|
|
assert result == []
|
|
|
|
async def test_agreement_no_rules_skipped(self, db_session: AsyncSession, seed_company):
|
|
"""无监控规则的协议不生成预警。"""
|
|
company_id = seed_company
|
|
agreement = InvestmentAgreement(
|
|
company_id=company_id,
|
|
title="无规则协议",
|
|
status="active",
|
|
monitoring_rules=None,
|
|
)
|
|
db_session.add(agreement)
|
|
await db_session.flush()
|
|
await db_session.commit()
|
|
|
|
result = await check_clause_triggers(db_session, company_id)
|
|
assert result == []
|