823 lines
29 KiB
Python
823 lines
29 KiB
Python
from __future__ import annotations
|
||
|
||
from datetime import datetime
|
||
from types import SimpleNamespace
|
||
|
||
import pytest
|
||
import pytest_asyncio
|
||
from fastapi import FastAPI
|
||
from httpx import ASGITransport, AsyncClient
|
||
from sqlalchemy import JSON
|
||
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
||
|
||
from app.api.ai_steward_api import routes_executive_briefing
|
||
from app.core.database import get_db
|
||
from app.dependencies.auth import get_current_user
|
||
from app.dependencies.tenant import get_tenant_id
|
||
from app.models.base import Base
|
||
from app.models.cognitive_conflict import CognitiveConflict
|
||
from app.models.feedback_center import FeedbackCenterRecord
|
||
from app.models.integration import Integration
|
||
from app.models.notification import Notification
|
||
from app.models.portfolio_snapshot import PortfolioSnapshot
|
||
from app.models.project import Project
|
||
from app.models.project_signal_link import ProjectSignalLink
|
||
from app.models.signal import Signal
|
||
from app.models.signal_source_registry import SignalSourceRegistry
|
||
from app.models.task import Task
|
||
from app.models.truth_node import TruthNode
|
||
from app.models.user import User
|
||
from app.models.user_workload import UserWorkload
|
||
from app.services.executive_briefing_service import ExecutiveBriefingService
|
||
|
||
|
||
def _use_sqlite_json_types() -> None:
|
||
Integration.__table__.c.config.type = JSON()
|
||
User.__table__.c.assigned_domains.type = JSON()
|
||
TruthNode.__table__.c.content.type = JSON()
|
||
Task.__table__.c.ai_recommended_assignees.type = JSON()
|
||
Task.__table__.c.ai_breakdown_payload.type = JSON()
|
||
Task.__table__.c.manual_overrides.type = JSON()
|
||
Task.__table__.c.required_skill_tag_ids.type = JSON()
|
||
SignalSourceRegistry.__table__.c.config.type = JSON()
|
||
SignalSourceRegistry.__table__.c.raw_context.type = JSON()
|
||
Notification.__table__.c.payload.type = JSON()
|
||
FeedbackCenterRecord.__table__.c.before_value.type = JSON()
|
||
FeedbackCenterRecord.__table__.c.after_value.type = JSON()
|
||
FeedbackCenterRecord.__table__.c.ai_snapshot.type = JSON()
|
||
FeedbackCenterRecord.__table__.c.reason_candidates.type = JSON()
|
||
CognitiveConflict.__table__.c.involved_user_ids.type = JSON()
|
||
PortfolioSnapshot.__table__.c.summary.type = JSON()
|
||
PortfolioSnapshot.__table__.c.payload.type = JSON()
|
||
|
||
|
||
@pytest_asyncio.fixture
|
||
async def db_session():
|
||
engine = create_async_engine("sqlite+aiosqlite:///:memory:")
|
||
_use_sqlite_json_types()
|
||
async with engine.begin() as conn:
|
||
await conn.run_sync(
|
||
lambda sync_conn: Base.metadata.create_all(
|
||
sync_conn,
|
||
tables=[
|
||
Integration.__table__,
|
||
User.__table__,
|
||
Project.__table__,
|
||
Signal.__table__,
|
||
TruthNode.__table__,
|
||
Task.__table__,
|
||
UserWorkload.__table__,
|
||
SignalSourceRegistry.__table__,
|
||
Notification.__table__,
|
||
FeedbackCenterRecord.__table__,
|
||
CognitiveConflict.__table__,
|
||
PortfolioSnapshot.__table__,
|
||
ProjectSignalLink.__table__,
|
||
],
|
||
)
|
||
)
|
||
maker = async_sessionmaker(engine, expire_on_commit=False)
|
||
async with maker() as session:
|
||
yield session
|
||
await engine.dispose()
|
||
|
||
|
||
def _build_app(*, db_session, role: str = "ADMIN") -> FastAPI:
|
||
app = FastAPI()
|
||
app.include_router(routes_executive_briefing.router)
|
||
|
||
async def _fake_get_db():
|
||
yield db_session
|
||
|
||
app.dependency_overrides[get_db] = _fake_get_db
|
||
app.dependency_overrides[get_tenant_id] = lambda: 1
|
||
app.dependency_overrides[get_current_user] = lambda: SimpleNamespace(
|
||
id=9001,
|
||
tenant_id=1,
|
||
name="管理员",
|
||
username="admin",
|
||
role=role,
|
||
)
|
||
return app
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_executive_briefing_snapshot_and_live_are_enveloped(db_session, monkeypatch) -> None:
|
||
async def fake_get_briefing(self, *, mode: str): # noqa: ANN001
|
||
_ = self
|
||
return {
|
||
"brief_date": "2026-03-30",
|
||
"mode": mode,
|
||
"hero": {
|
||
"title": "老板晨报",
|
||
"decision_count": 2,
|
||
"anomaly_count": 4,
|
||
},
|
||
"sections": [
|
||
{
|
||
"key": "decisions",
|
||
"title": "今日需拍板",
|
||
"summary": "有 2 项需要拍板",
|
||
"metrics": [{"label": "需拍板", "value": 2, "tone": "warning"}],
|
||
"items": [
|
||
{
|
||
"id": "conflict:1",
|
||
"title": "skill_mismatch",
|
||
"display_title": "技能错配",
|
||
"display_subtitle": "技能要求与成员能力不匹配",
|
||
"display_source": "协作冲突",
|
||
"technical_identifier": "skill_mismatch",
|
||
"detail": "技能要求与成员能力不匹配",
|
||
"detail_lines": ["技能要求与成员能力不匹配"],
|
||
"meta": [],
|
||
"tone": "warning",
|
||
"kind": "conflict",
|
||
}
|
||
],
|
||
"drilldown_tab": "signals",
|
||
}
|
||
],
|
||
}
|
||
|
||
monkeypatch.setattr(ExecutiveBriefingService, "get_briefing", fake_get_briefing)
|
||
|
||
app = _build_app(db_session=db_session)
|
||
transport = ASGITransport(app=app)
|
||
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
||
snapshot_response = await client.get("/ai-steward/executive-briefing", params={"mode": "snapshot"})
|
||
live_response = await client.get("/ai-steward/executive-briefing", params={"mode": "live"})
|
||
|
||
assert snapshot_response.status_code == 200
|
||
snapshot_payload = snapshot_response.json()
|
||
assert snapshot_payload["code"] == 0
|
||
assert snapshot_payload["message"] == "success"
|
||
assert snapshot_payload["data"]["mode"] == "snapshot"
|
||
assert snapshot_payload["data"]["hero"]["decision_count"] == 2
|
||
|
||
assert live_response.status_code == 200
|
||
live_payload = live_response.json()
|
||
assert live_payload["code"] == 0
|
||
assert live_payload["data"]["mode"] == "live"
|
||
assert live_payload["data"]["sections"][0]["key"] == "decisions"
|
||
assert live_payload["data"]["sections"][0]["items"][0]["display_title"] == "技能错配"
|
||
assert live_payload["data"]["sections"][0]["items"][0]["display_subtitle"] == "技能要求与成员能力不匹配"
|
||
assert live_payload["data"]["sections"][0]["items"][0]["display_source"] == "协作冲突"
|
||
assert live_payload["data"]["sections"][0]["items"][0]["technical_identifier"] == "skill_mismatch"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_executive_briefing_config_round_trip_is_enveloped(db_session) -> None:
|
||
app = _build_app(db_session=db_session)
|
||
transport = ASGITransport(app=app)
|
||
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
||
default_response = await client.get("/ai-steward/executive-briefing/config")
|
||
update_response = await client.put(
|
||
"/ai-steward/executive-briefing/config",
|
||
json={
|
||
"enabled": True,
|
||
"send_time": "09:15",
|
||
"send_cycle": "daily",
|
||
"delivery_target_type": "management_group",
|
||
"delivery_target_ref": "oc_management_group",
|
||
"timezone": "Asia/Shanghai",
|
||
},
|
||
)
|
||
refreshed_response = await client.get("/ai-steward/executive-briefing/config")
|
||
|
||
assert default_response.status_code == 200
|
||
default_payload = default_response.json()
|
||
assert default_payload["code"] == 0
|
||
assert default_payload["data"]["send_time"] == "08:30"
|
||
assert default_payload["data"]["send_cycle"] == "workdays"
|
||
|
||
assert update_response.status_code == 200
|
||
update_payload = update_response.json()
|
||
assert update_payload["code"] == 0
|
||
assert update_payload["data"]["send_time"] == "09:15"
|
||
assert update_payload["data"]["delivery_target_ref"] == "oc_management_group"
|
||
assert update_payload["data"]["updated_by"] == 9001
|
||
|
||
refreshed_payload = refreshed_response.json()
|
||
assert refreshed_payload["code"] == 0
|
||
assert refreshed_payload["data"]["send_time"] == "09:15"
|
||
assert refreshed_payload["data"]["send_cycle"] == "daily"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_executive_briefing_config_rejects_non_manager_users_with_envelope(db_session) -> None:
|
||
app = _build_app(db_session=db_session, role="BE")
|
||
transport = ASGITransport(app=app)
|
||
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
||
response = await client.put(
|
||
"/ai-steward/executive-briefing/config",
|
||
json={
|
||
"enabled": True,
|
||
"send_time": "09:15",
|
||
"send_cycle": "daily",
|
||
"delivery_target_type": "management_group",
|
||
"delivery_target_ref": "oc_management_group",
|
||
"timezone": "Asia/Shanghai",
|
||
},
|
||
)
|
||
|
||
assert response.status_code == 403
|
||
payload = response.json()
|
||
assert payload["code"] == 403
|
||
assert payload["data"] is None
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_executive_briefing_service_persists_config_in_integration_row(db_session) -> None:
|
||
service = ExecutiveBriefingService(db_session, tenant_id=1)
|
||
|
||
updated = await service.update_config(
|
||
payload={
|
||
"enabled": True,
|
||
"send_time": "09:45",
|
||
"send_cycle": "daily",
|
||
"delivery_target_type": "management_group",
|
||
"delivery_target_ref": "oc_management_group_2",
|
||
"timezone": "Asia/Shanghai",
|
||
},
|
||
operator_user_id=777,
|
||
)
|
||
await db_session.commit()
|
||
|
||
assert updated["send_time"] == "09:45"
|
||
assert updated["updated_by"] == 777
|
||
|
||
row = (
|
||
await db_session.execute(
|
||
Integration.__table__.select().where(Integration.tenant_id == 1, Integration.type == "executive_briefing")
|
||
)
|
||
).mappings().one()
|
||
assert row["config"]["send_time"] == "09:45"
|
||
assert row["config"]["delivery_target_ref"] == "oc_management_group_2"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_executive_briefing_service_formats_management_summary_text(db_session) -> None:
|
||
service = ExecutiveBriefingService(db_session, tenant_id=1)
|
||
|
||
text = service.render_management_group_message(
|
||
{
|
||
"brief_date": "2026-03-30",
|
||
"hero": {
|
||
"title": "简报中心",
|
||
"decision_count": 2,
|
||
"anomaly_count": 4,
|
||
"baseline_generated_at": datetime(2026, 3, 30, 8, 30, 0).isoformat(),
|
||
},
|
||
"sections": [
|
||
{
|
||
"title": "今日需拍板",
|
||
"summary": "有 2 项需要拍板",
|
||
"items": [
|
||
{"title": "P0 信号无人接", "detail": "项目 A"},
|
||
{"title": "持续过载成员", "detail": "张三 128%"},
|
||
],
|
||
}
|
||
],
|
||
}
|
||
)
|
||
|
||
assert "简报中心" in text
|
||
assert "今日需拍板" in text
|
||
assert "P0 信号无人接" in text
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_executive_briefing_pending_signal_sources_include_display_fields(db_session) -> None:
|
||
owner = User(
|
||
tenant_id=1,
|
||
username="pm",
|
||
email="pm@example.com",
|
||
password_hash="hash",
|
||
name="产品经理",
|
||
role="ADMIN",
|
||
status="active",
|
||
is_active=True,
|
||
)
|
||
db_session.add(owner)
|
||
await db_session.flush()
|
||
|
||
project = Project(
|
||
tenant_id=1,
|
||
name="cashflow",
|
||
title="现金流看板",
|
||
status="active",
|
||
)
|
||
db_session.add(project)
|
||
await db_session.flush()
|
||
|
||
db_session.add(
|
||
SignalSourceRegistry(
|
||
tenant_id=1,
|
||
project_id=project.id,
|
||
created_by_user_id=owner.id,
|
||
source_kind="github",
|
||
source_name="现金流日报仓库",
|
||
source_identifier="daily_cashflow_analysis",
|
||
description="每日现金流分析脚本仓库",
|
||
requested_via="ai_steward",
|
||
status="pending",
|
||
onboarding_status="pending",
|
||
config={},
|
||
raw_context={},
|
||
)
|
||
)
|
||
await db_session.commit()
|
||
|
||
service = ExecutiveBriefingService(db_session, tenant_id=1)
|
||
items = await service._list_pending_signal_sources(limit=5)
|
||
|
||
assert len(items) == 1
|
||
assert items[0]["display_title"] == "现金流日报仓库"
|
||
assert items[0]["display_source"] == "GitHub 仓库"
|
||
assert items[0]["technical_identifier"] == "daily_cashflow_analysis"
|
||
assert "daily_cashflow_analysis" not in items[0]["display_subtitle"]
|
||
|
||
|
||
def test_snapshot_payload_without_display_fields_is_not_current() -> None:
|
||
payload = {
|
||
"hero": {"title": "简报中心"},
|
||
"sections": [
|
||
{
|
||
"key": "decisions",
|
||
"title": "今日需拍板",
|
||
"summary": "有 1 项需要拍板",
|
||
"items": [
|
||
{
|
||
"id": "conflict:1",
|
||
"title": "skill_mismatch",
|
||
"detail": "技能要求与成员能力不匹配",
|
||
"detail_lines": ["技能要求与成员能力不匹配"],
|
||
"meta": [],
|
||
"tone": "warning",
|
||
"kind": "conflict",
|
||
}
|
||
],
|
||
}
|
||
],
|
||
}
|
||
|
||
assert ExecutiveBriefingService._snapshot_payload_is_current(payload) is False
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_executive_briefing_service_builds_detailed_signal_items_and_target_params(db_session) -> None:
|
||
owner = User(
|
||
tenant_id=1,
|
||
username="boss",
|
||
email="boss@example.com",
|
||
password_hash="hash",
|
||
name="李雷",
|
||
role="BOSS",
|
||
status="active",
|
||
is_active=True,
|
||
)
|
||
db_session.add(owner)
|
||
await db_session.flush()
|
||
|
||
project = Project(
|
||
tenant_id=1,
|
||
name="cashflow-forecast",
|
||
title="现金流平台",
|
||
description="现金流预测与预警",
|
||
owner_id=owner.id,
|
||
status="active",
|
||
)
|
||
db_session.add(project)
|
||
await db_session.flush()
|
||
|
||
signal = Signal(
|
||
tenant_id=1,
|
||
source="feishu",
|
||
title="现金流预测功能需求",
|
||
summary="需要尽快评估接入方案",
|
||
priority="high",
|
||
processed=False,
|
||
matched_project_id=project.id,
|
||
status="pending",
|
||
planner_status="pending",
|
||
planner_assignee_user_id=owner.id,
|
||
created_at=datetime(2026, 3, 30, 7, 50, 0),
|
||
)
|
||
db_session.add(signal)
|
||
await db_session.commit()
|
||
|
||
service = ExecutiveBriefingService(db_session, tenant_id=1)
|
||
payload = await service.build_brief_payload(
|
||
mode="live",
|
||
baseline_generated_at=None,
|
||
live_generated_at=datetime(2026, 3, 30, 8, 30, 0),
|
||
)
|
||
|
||
assert payload["hero"]["title"] == "简报中心"
|
||
decisions = next(section for section in payload["sections"] if section["key"] == "decisions")
|
||
signal_item = next(item for item in decisions["items"] if item["kind"] == "signal")
|
||
|
||
assert signal_item["id"] == f"signal:{signal.id}"
|
||
assert signal_item["project_name"] == "现金流平台"
|
||
assert signal_item["owner_name"] == "李雷"
|
||
assert signal_item["occurred_at"] == "2026-03-30T07:50:00"
|
||
assert "优先级 高优" in signal_item["detail_lines"]
|
||
assert "状态 待处理" in signal_item["detail_lines"]
|
||
assert any(meta["label"] == "优先级" and meta["value"] == "高优" for meta in signal_item["meta"])
|
||
assert any(meta["label"] == "负责人" and meta["value"] == "李雷" for meta in signal_item["meta"])
|
||
assert signal_item["target"]["type"] == "project_board"
|
||
assert signal_item["target"]["params"]["stage"] == "signal"
|
||
assert signal_item["target"]["params"]["signal_keyword"] == "现金流预测功能需求"
|
||
assert signal_item["target"]["params"]["signal_status"] == "pending"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_executive_briefing_excludes_unassigned_signals_and_aligns_metrics(db_session) -> None:
|
||
owner = User(
|
||
tenant_id=1,
|
||
username="boss",
|
||
email="boss@example.com",
|
||
password_hash="hash",
|
||
name="李雷",
|
||
role="BOSS",
|
||
status="active",
|
||
is_active=True,
|
||
)
|
||
db_session.add(owner)
|
||
await db_session.flush()
|
||
|
||
project = Project(
|
||
tenant_id=1,
|
||
name="cashflow-forecast",
|
||
title="现金流平台",
|
||
description="现金流预测与预警",
|
||
owner_id=owner.id,
|
||
status="active",
|
||
)
|
||
db_session.add(project)
|
||
await db_session.flush()
|
||
|
||
db_session.add_all(
|
||
[
|
||
Signal(
|
||
tenant_id=1,
|
||
source="feishu",
|
||
title="现金流预测功能需求",
|
||
summary="需要尽快评估接入方案",
|
||
priority="high",
|
||
processed=False,
|
||
matched_project_id=project.id,
|
||
status="pending",
|
||
planner_status="pending",
|
||
planner_assignee_user_id=owner.id,
|
||
created_at=datetime(2026, 3, 30, 7, 50, 0),
|
||
),
|
||
Signal(
|
||
tenant_id=1,
|
||
source="feishu",
|
||
title="孤立高优信号",
|
||
summary="还没有归属项目",
|
||
priority="high",
|
||
processed=False,
|
||
matched_project_id=None,
|
||
status="pending",
|
||
planner_status="pending",
|
||
planner_assignee_user_id=None,
|
||
created_at=datetime(2026, 3, 30, 7, 45, 0),
|
||
),
|
||
Signal(
|
||
tenant_id=1,
|
||
source="feishu",
|
||
title=None,
|
||
summary=None,
|
||
priority="high",
|
||
processed=False,
|
||
matched_project_id=None,
|
||
status="pending",
|
||
planner_status="pending",
|
||
planner_assignee_user_id=None,
|
||
created_at=datetime(2026, 3, 30, 7, 40, 0),
|
||
),
|
||
]
|
||
)
|
||
await db_session.commit()
|
||
|
||
service = ExecutiveBriefingService(db_session, tenant_id=1)
|
||
payload = await service.build_brief_payload(
|
||
mode="live",
|
||
baseline_generated_at=None,
|
||
live_generated_at=datetime(2026, 3, 30, 8, 30, 0),
|
||
)
|
||
|
||
decisions = next(section for section in payload["sections"] if section["key"] == "decisions")
|
||
input_health = next(section for section in payload["sections"] if section["key"] == "input_health")
|
||
|
||
decision_signal_items = [item for item in decisions["items"] if item["kind"] == "signal"]
|
||
input_signal_items = [item for item in input_health["items"] if item["kind"] == "signal"]
|
||
|
||
assert [item["title"] for item in decision_signal_items] == ["现金流预测功能需求"]
|
||
assert [item["title"] for item in input_signal_items] == ["现金流预测功能需求"]
|
||
assert all(item["project_id"] == project.id for item in decision_signal_items)
|
||
assert all(item["project_id"] == project.id for item in input_signal_items)
|
||
assert all(item["title"] != "待处理事项" for item in decision_signal_items + input_signal_items)
|
||
|
||
high_priority_metric = next(metric for metric in decisions["metrics"] if metric["label"] == "高优信号")
|
||
recent_signal_metric = next(metric for metric in input_health["metrics"] if metric["label"] == "24h 新信号")
|
||
pending_signal_metric = next(metric for metric in input_health["metrics"] if metric["label"] == "待处理信号")
|
||
|
||
assert high_priority_metric["value"] == 1
|
||
assert recent_signal_metric["value"] == 1
|
||
assert pending_signal_metric["value"] == 1
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_executive_briefing_snapshot_rebuilds_outdated_payload_for_detail_ui(db_session, monkeypatch) -> None:
|
||
fixed_now = datetime(2026, 3, 30, 9, 0, 0)
|
||
monkeypatch.setattr("app.services.executive_briefing_service.now_cn_naive", lambda: fixed_now)
|
||
|
||
owner = User(
|
||
tenant_id=1,
|
||
username="boss",
|
||
email="boss@example.com",
|
||
password_hash="hash",
|
||
name="李雷",
|
||
role="BOSS",
|
||
status="active",
|
||
is_active=True,
|
||
)
|
||
db_session.add(owner)
|
||
await db_session.flush()
|
||
|
||
project = Project(
|
||
tenant_id=1,
|
||
name="cashflow-forecast",
|
||
title="现金流平台",
|
||
description="现金流预测与预警",
|
||
owner_id=owner.id,
|
||
status="active",
|
||
)
|
||
db_session.add(project)
|
||
await db_session.flush()
|
||
|
||
signal = Signal(
|
||
tenant_id=1,
|
||
source="feishu",
|
||
title="现金流预测功能需求",
|
||
summary="需要尽快评估接入方案",
|
||
priority="high",
|
||
processed=False,
|
||
matched_project_id=project.id,
|
||
status="pending",
|
||
planner_status="pending",
|
||
planner_assignee_user_id=owner.id,
|
||
created_at=datetime(2026, 3, 30, 7, 50, 0),
|
||
)
|
||
db_session.add(signal)
|
||
await db_session.flush()
|
||
|
||
outdated_snapshot = PortfolioSnapshot(
|
||
tenant_id=1,
|
||
scope_type="executive_brief",
|
||
scope_ref="tenant:1",
|
||
summary={},
|
||
payload={
|
||
"brief_date": "2026-03-30",
|
||
"mode": "snapshot",
|
||
"hero": {
|
||
"title": "老板晨报中心",
|
||
"decision_count": 1,
|
||
"anomaly_count": 1,
|
||
"baseline_generated_at": datetime(2026, 3, 30, 8, 30, 0).isoformat(),
|
||
},
|
||
"sections": [
|
||
{
|
||
"key": "decisions",
|
||
"title": "今日需拍板",
|
||
"summary": "旧快照",
|
||
"metrics": [],
|
||
"items": [
|
||
{
|
||
"kind": "signal",
|
||
"tone": "warning",
|
||
"title": "📈 现金流预测功能需求",
|
||
"detail": "优先级 高优｜状态 待处理｜03-27 12:06",
|
||
"target": {"ref": "signals", "type": "ai_steward_tab", "label": "信号捕获"},
|
||
"project_id": project.id,
|
||
}
|
||
],
|
||
}
|
||
],
|
||
},
|
||
created_at=datetime(2026, 3, 30, 8, 30, 0),
|
||
)
|
||
db_session.add(outdated_snapshot)
|
||
await db_session.commit()
|
||
|
||
service = ExecutiveBriefingService(db_session, tenant_id=1)
|
||
payload = await service.get_briefing(mode="snapshot")
|
||
|
||
assert payload["hero"]["title"] == "简报中心"
|
||
assert payload["mode"] == "snapshot"
|
||
assert payload["hero"]["baseline_generated_at"] == datetime(2026, 3, 30, 8, 30, 0).isoformat()
|
||
|
||
decisions = next(section for section in payload["sections"] if section["key"] == "decisions")
|
||
signal_item = next(item for item in decisions["items"] if item["kind"] == "signal")
|
||
assert signal_item["id"] == f"signal:{signal.id}"
|
||
assert signal_item["target"]["type"] == "project_board"
|
||
assert signal_item["target"]["params"]["stage"] == "signal"
|
||
assert signal_item["detail_lines"]
|
||
assert signal_item["meta"]
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_executive_briefing_signal_source_target_does_not_apply_mismatched_keyword_filter(db_session) -> None:
|
||
owner = User(
|
||
tenant_id=1,
|
||
username="boss",
|
||
email="boss@example.com",
|
||
password_hash="hash",
|
||
name="李雷",
|
||
role="BOSS",
|
||
status="active",
|
||
is_active=True,
|
||
)
|
||
db_session.add(owner)
|
||
await db_session.flush()
|
||
|
||
project = Project(
|
||
tenant_id=1,
|
||
name="cashflow-forecast",
|
||
title="现金流平台",
|
||
description="现金流预测与预警",
|
||
owner_id=owner.id,
|
||
status="active",
|
||
)
|
||
db_session.add(project)
|
||
await db_session.flush()
|
||
|
||
source = SignalSourceRegistry(
|
||
tenant_id=1,
|
||
project_id=project.id,
|
||
created_by_user_id=owner.id,
|
||
source_kind="scheduled",
|
||
source_name="每日现金流分析",
|
||
source_identifier="daily_cashflow_analysis",
|
||
onboarding_status="pending",
|
||
created_at=datetime(2026, 3, 30, 8, 10, 0),
|
||
)
|
||
db_session.add(source)
|
||
await db_session.commit()
|
||
|
||
service = ExecutiveBriefingService(db_session, tenant_id=1)
|
||
payload = await service.build_brief_payload(
|
||
mode="live",
|
||
baseline_generated_at=None,
|
||
live_generated_at=datetime(2026, 3, 30, 8, 30, 0),
|
||
)
|
||
|
||
input_health = next(section for section in payload["sections"] if section["key"] == "input_health")
|
||
source_item = next(item for item in input_health["items"] if item["kind"] == "signal_source")
|
||
assert source_item["target"]["type"] == "project_board"
|
||
assert source_item["target"]["params"] == {"stage": "signal"}
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_executive_briefing_capacity_ratio_matches_displayed_hours(db_session) -> None:
|
||
user = User(
|
||
tenant_id=1,
|
||
username="owner",
|
||
email="owner@example.com",
|
||
password_hash="hash",
|
||
name="杨宇",
|
||
role="ADMIN",
|
||
status="active",
|
||
is_active=True,
|
||
)
|
||
db_session.add(user)
|
||
await db_session.flush()
|
||
|
||
db_session.add(
|
||
UserWorkload(
|
||
tenant_id=1,
|
||
user_id=user.id,
|
||
current_assigned_hours=95.6,
|
||
max_available_hours_per_week=60,
|
||
workload_ratio=797,
|
||
)
|
||
)
|
||
await db_session.commit()
|
||
|
||
service = ExecutiveBriefingService(db_session, tenant_id=1)
|
||
payload = await service.build_brief_payload(
|
||
mode="live",
|
||
baseline_generated_at=None,
|
||
live_generated_at=datetime(2026, 3, 30, 8, 30, 0),
|
||
)
|
||
|
||
capacity = next(section for section in payload["sections"] if section["key"] == "team_capacity")
|
||
overload_item = next(item for item in capacity["items"] if item["kind"] == "capacity")
|
||
assert overload_item["title"] == "杨宇"
|
||
assert overload_item["detail_lines"][0] == "159% 负载"
|
||
assert any(meta["label"] == "负载" and meta["value"] == "159%" for meta in overload_item["meta"])
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_executive_briefing_localizes_conflict_and_signal_source_codes(db_session) -> None:
|
||
owner = User(
|
||
tenant_id=1,
|
||
username="boss",
|
||
email="boss@example.com",
|
||
password_hash="hash",
|
||
name="李雷",
|
||
role="BOSS",
|
||
status="active",
|
||
is_active=True,
|
||
)
|
||
db_session.add(owner)
|
||
await db_session.flush()
|
||
|
||
project = Project(
|
||
tenant_id=1,
|
||
name="cashflow-forecast",
|
||
title="现金流平台",
|
||
description="现金流预测与预警",
|
||
owner_id=owner.id,
|
||
status="active",
|
||
)
|
||
db_session.add(project)
|
||
await db_session.flush()
|
||
|
||
truth_node = TruthNode(
|
||
tenant_id=1,
|
||
project_id=project.id,
|
||
created_by=owner.id,
|
||
node_type="risk",
|
||
title="风险节点",
|
||
content={"summary": "风险节点"},
|
||
version=1,
|
||
status="active",
|
||
)
|
||
db_session.add(truth_node)
|
||
await db_session.flush()
|
||
|
||
db_session.add(
|
||
SignalSourceRegistry(
|
||
tenant_id=1,
|
||
project_id=project.id,
|
||
created_by_user_id=owner.id,
|
||
source_kind="user_insight",
|
||
source_name="用户洞察数据源",
|
||
source_identifier="ai-customer-service-insight",
|
||
description="用户洞察相关信号源",
|
||
requested_via="ai_steward",
|
||
status="pending",
|
||
onboarding_status="pending",
|
||
config={},
|
||
raw_context={},
|
||
)
|
||
)
|
||
db_session.add_all(
|
||
[
|
||
CognitiveConflict(
|
||
tenant_id=1,
|
||
truth_node_id=truth_node.id,
|
||
conflict_type="quality_drop",
|
||
description="质量下降预警",
|
||
involved_user_ids=[],
|
||
status="active",
|
||
),
|
||
CognitiveConflict(
|
||
tenant_id=1,
|
||
truth_node_id=truth_node.id,
|
||
conflict_type="schedule_delay",
|
||
description="任务超时预警",
|
||
involved_user_ids=[],
|
||
status="active",
|
||
),
|
||
]
|
||
)
|
||
await db_session.commit()
|
||
|
||
service = ExecutiveBriefingService(db_session, tenant_id=1)
|
||
payload = await service.build_brief_payload(
|
||
mode="live",
|
||
baseline_generated_at=None,
|
||
live_generated_at=datetime(2026, 3, 30, 8, 30, 0),
|
||
)
|
||
|
||
decisions = next(section for section in payload["sections"] if section["key"] == "decisions")
|
||
input_health = next(section for section in payload["sections"] if section["key"] == "input_health")
|
||
|
||
source_item = next(item for item in input_health["items"] if item["kind"] == "signal_source")
|
||
assert source_item["display_source"] == "用户洞察来源"
|
||
assert source_item["display_subtitle"].startswith("用户洞察来源")
|
||
assert "user_insight" not in source_item["display_subtitle"]
|
||
|
||
conflict_titles = {
|
||
item["technical_identifier"]: item["display_title"]
|
||
for item in decisions["items"]
|
||
if item["kind"] == "conflict"
|
||
}
|
||
assert conflict_titles["quality_drop"] == "质量下降预警"
|
||
assert conflict_titles["schedule_delay"] == "进度延期预警"
|