PMFlow/backend/tests/test_ai_pmo_full_scenario.py

641 lines
28 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

import json
import os
import unittest
from datetime import datetime
from tempfile import NamedTemporaryFile
from unittest.mock import AsyncMock, patch
from sqlalchemy import and_, select
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from app.api.users import _create_user_with_defaults
from app.models import (
Base,
Budget,
Project,
ProjectDomain,
ProjectSignalLink,
Signal,
SkillTag,
Task,
TruthNode,
User,
UserProject,
UserSkillBinding,
UserWorkload,
WorkItem,
)
from app.schemas import UserCreate
from app.services.ai_pmo_cadence_service import AIPMOCadenceService
from app.services.ai_pmo_kernel_service import AIPMOKernelService
from app.services.signal_capture import ProactiveSignalCaptureService
from app.services.skill_service import recommend_personnel
from app.services import workload_service
class AIPMOFullScenarioTest(unittest.IsolatedAsyncioTestCase):
async def asyncSetUp(self) -> None:
tmp = NamedTemporaryFile(prefix="pmflow-ai-pmo-scenario-", suffix=".db", delete=False)
tmp.close()
self.db_path = tmp.name
self.engine = create_async_engine(f"sqlite+aiosqlite:///{self.db_path}", future=True)
self.session_factory = async_sessionmaker(self.engine, expire_on_commit=False, class_=AsyncSession)
async with self.engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
async with self.session_factory() as db:
self.policy_skill_id = await self._create_skill_tag(db, name="政策分析", slug="policy-analysis")
self.research_skill_id = await self._create_skill_tag(db, name="行业研究", slug="market-research")
self.web_skill_id = await self._create_skill_tag(db, name="爬虫采集", slug="web-crawling")
self.graph_skill_id = await self._create_skill_tag(db, name="知识图谱", slug="knowledge-graph")
self.pipeline_skill_id = await self._create_skill_tag(db, name="数据管道", slug="data-pipeline")
boss = User(
tenant_id=1,
username="boss",
email="boss@example.com",
password_hash="hash",
name="老板",
role="BOSS",
assigned_domains=[],
workload=0,
status="active",
is_active=True,
)
db.add(boss)
await db.flush()
self.boss_id = int(boss.id)
domain = ProjectDomain(
tenant_id=1,
name="医药",
slug="pharma",
description="医药产业情报域",
config={},
is_active=True,
is_default=False,
signal_count=0,
context_node_count=0,
experiment_count=0,
owner_id=str(boss.id),
)
db.add(domain)
await db.flush()
self.domain_id = int(domain.id)
project = Project(
tenant_id=1,
name="医药行业十五五规划布局研究",
title="医药行业十五五规划布局研究",
description="跟踪医保支付改革、器械招采与创新药方向。",
owner_id=int(boss.id),
status="active",
domain_id=int(domain.id),
assigned_role="pm",
signal_count=0,
)
db.add(project)
await db.flush()
self.base_project_id = int(project.id)
charter = TruthNode(
tenant_id=1,
node_type="charter",
project_id=int(project.id),
signal_id=None,
created_by=int(boss.id),
title="医药研究项目宪章",
content={"success_criteria": ["两周内形成华东医保支付改革观察报告"]},
version=1,
status="active",
)
db.add(charter)
await db.flush()
self.base_charter_id = int(charter.id)
db.add(
Budget(
tenant_id=1,
budget_type="project",
project_id=int(project.id),
period="monthly",
budget_amount_usd=1500.0,
alert_threshold_percent=80.0,
hard_limit_enabled=True,
current_usage_usd=620.0,
current_period_start=datetime(2026, 3, 1),
current_period_end=datetime(2026, 3, 31),
status="active",
override_history={},
)
)
db.add(UserProject(tenant_id=1, user_id=int(boss.id), project_id=int(project.id)))
db.add(
UserWorkload(
tenant_id=1,
user_id=int(boss.id),
current_assigned_hours=18.0,
max_available_hours_per_week=40.0,
workload_ratio=0.0,
)
)
await self._bind_skill(db, user_id=int(boss.id), skill_tag_id=self.policy_skill_id, level="advanced", proficiency=86)
await self._bind_skill(db, user_id=int(boss.id), skill_tag_id=self.research_skill_id, level="expert", proficiency=92)
db.add(
Task(
tenant_id=1,
truth_node_id=int(charter.id),
project_id=int(project.id),
assignee_id=int(boss.id),
ai_recommended_assignees=None,
title="拆解华东医保支付改革样本",
description="需要比对江苏/浙江 DRG-DIP 改革口径和器械招采变化。",
required_skill_tag_ids=[self.policy_skill_id, self.web_skill_id],
status="blocked",
priority="high",
estimated_hours=12.0,
actual_hours=0.0,
quality_score=None,
due_date=datetime(2026, 3, 12),
completed_at=None,
)
)
await workload_service.recalculate_user_workload_ratio(db, tenant_id=1, user_id=int(boss.id), commit=False)
await db.commit()
async def asyncTearDown(self) -> None:
await self.engine.dispose()
if getattr(self, "db_path", None):
try:
os.remove(self.db_path)
except FileNotFoundError:
pass
async def _create_skill_tag(self, db: AsyncSession, *, name: str, slug: str) -> int:
row = SkillTag(
tenant_id=1,
name=name,
slug=slug,
category="skill",
description=f"{name}能力",
is_system=False,
sort_order=0,
)
db.add(row)
await db.flush()
return int(row.id)
async def _bind_skill(
self,
db: AsyncSession,
*,
user_id: int,
skill_tag_id: int,
level: str,
proficiency: int,
) -> None:
db.add(
UserSkillBinding(
tenant_id=1,
user_id=int(user_id),
skill_tag_id=int(skill_tag_id),
proficiency_level=level,
proficiency=int(proficiency),
level=level,
years_experience=5.0,
quality_score=8.5,
efficiency_score=8.0,
)
)
await db.flush()
async def _set_workload_hours(self, db: AsyncSession, *, user_id: int, assigned_hours: float) -> None:
workload = (
await db.execute(
select(UserWorkload).where(
and_(
UserWorkload.tenant_id == 1,
UserWorkload.user_id == int(user_id),
)
)
)
).scalar_one()
workload.current_assigned_hours = float(assigned_hours)
await workload_service.recalculate_user_workload_ratio(db, tenant_id=1, user_id=int(user_id), commit=False)
async def test_full_difficult_scenario_covers_feishu_context_project_signal_workflow_and_workload(self) -> None:
async with self.session_factory() as db:
strategy_user = await _create_user_with_defaults(
db,
tenant_id=1,
user_data=UserCreate(
username="hanqing",
password="password-1",
name="韩青",
email="hanqing@example.com",
role="ADMIN",
assigned_domains=["pharma"],
project_ids=[self.base_project_id],
),
actor_id=self.boss_id,
commit=False,
)
crawler_user = await _create_user_with_defaults(
db,
tenant_id=1,
user_data=UserCreate(
username="zhouzhan",
password="password-2",
name="周湛",
email="zhouzhan@example.com",
role="BE",
assigned_domains=["pharma"],
project_ids=[],
),
actor_id=self.boss_id,
commit=False,
)
graph_user = await _create_user_with_defaults(
db,
tenant_id=1,
user_data=UserCreate(
username="guyan",
password="password-3",
name="顾言",
email="guyan@example.com",
role="AI",
assigned_domains=["pharma"],
project_ids=[],
),
actor_id=self.boss_id,
commit=False,
)
await self._bind_skill(db, user_id=int(strategy_user.id), skill_tag_id=self.policy_skill_id, level="expert", proficiency=93)
await self._bind_skill(db, user_id=int(strategy_user.id), skill_tag_id=self.research_skill_id, level="expert", proficiency=90)
await self._bind_skill(db, user_id=int(crawler_user.id), skill_tag_id=self.web_skill_id, level="expert", proficiency=95)
await self._bind_skill(db, user_id=int(crawler_user.id), skill_tag_id=self.policy_skill_id, level="advanced", proficiency=84)
await self._bind_skill(db, user_id=int(crawler_user.id), skill_tag_id=self.pipeline_skill_id, level="advanced", proficiency=88)
await self._bind_skill(db, user_id=int(graph_user.id), skill_tag_id=self.graph_skill_id, level="expert", proficiency=91)
await self._bind_skill(db, user_id=int(graph_user.id), skill_tag_id=self.pipeline_skill_id, level="advanced", proficiency=82)
await self._set_workload_hours(db, user_id=int(strategy_user.id), assigned_hours=28.0)
await self._set_workload_hours(db, user_id=int(crawler_user.id), assigned_hours=8.0)
await self._set_workload_hours(db, user_id=int(graph_user.id), assigned_hours=10.0)
await db.commit()
kernel = AIPMOKernelService(db, tenant_id=1)
capture_service = ProactiveSignalCaptureService(db)
hard_question = (
"项目ID: 1 如果把样本窗口收紧到 2025Q4-2026Q1,只看江苏/浙江的 DRG-DIP 支付、"
"三甲院器械招采与设备更新预算,当前状态下最可能拖慢结论闭环的到底是预算、"
"blocked task 还是技能缺口?为什么?别给套话,按证据优先级直接判断。"
)
with (
patch(
"app.services.ai_pmo_kernel_service.AdaptiveRAGService.answer_question",
AsyncMock(
return_value={
"answer": "优先级 1 是 blocked task,其次是技能缺口,预算暂未触顶。",
"sources": [],
}
),
),
patch(
"app.services.signal_capture.context_cache_service.get_session_messages",
AsyncMock(return_value=[]),
),
patch(
"app.services.signal_capture.context_cache_service.append_session_messages",
AsyncMock(),
) as cache_append_mock,
patch(
"app.services.signal_capture.AdaptiveRAGService.ingest_feishu_message_async",
AsyncMock(),
),
):
qa_result = await capture_service.capture_from_feishu(
group_chat_id="oc_complex_case",
message_content=hard_question,
sender="老板",
tenant_id=1,
proposer_open_id="ou_boss",
captured_at=datetime(2026, 3, 9, 9, 0, 0),
feishu_message_id="om-qa-hard-1",
chat_type="group",
)
async def create_project_side_effect(*, input_text: str, creator_id: int, tenant_id: int) -> dict:
project = Project(
tenant_id=int(tenant_id),
name="华东医保控费与器械国产替代观察台",
title="华东医保控费与器械国产替代观察台",
description="聚焦华东 DRG-DIP、器械带量采购和国产替代节奏。",
owner_id=int(creator_id),
status="active",
domain_id=self.domain_id,
assigned_role="ai_engineer",
signal_count=0,
)
db.add(project)
await db.flush()
charter = TruthNode(
tenant_id=1,
node_type="charter",
project_id=int(project.id),
signal_id=None,
created_by=int(creator_id),
title="华东医保观察项目宪章",
content={"success_criteria": ["10 天内交付策略 memo、政策雷达和竞品跟踪框架"]},
version=1,
status="active",
)
db.add(charter)
await db.flush()
bootstrap_signal = Signal(
tenant_id=1,
source="PMFlow",
title="医保控费与器械国产替代立项信号",
content=input_text,
summary="立项信号",
signal_type="新需求",
priority="high",
impact_score=0.85,
matched_project_id=int(project.id),
reasoning=json.dumps(["来自高强度情景立项指令"], ensure_ascii=False),
confidence=0.92,
recommended_action="enter_only",
processed=True,
tasks_generated=json.dumps([], ensure_ascii=False),
status="已处理",
)
db.add(bootstrap_signal)
await db.flush()
db.add(
ProjectSignalLink(
tenant_id=1,
project_id=int(project.id),
signal_id=int(bootstrap_signal.id),
link_type="related",
relevance_score=1.0,
)
)
db.add(
Budget(
tenant_id=1,
budget_type="project",
project_id=int(project.id),
period="monthly",
budget_amount_usd=3200.0,
alert_threshold_percent=80.0,
hard_limit_enabled=True,
current_usage_usd=240.0,
current_period_start=datetime(2026, 3, 1),
current_period_end=datetime(2026, 3, 31),
status="active",
override_history={},
)
)
db.add(UserProject(tenant_id=1, user_id=int(creator_id), project_id=int(project.id)))
task_one = Task(
tenant_id=1,
truth_node_id=int(charter.id),
project_id=int(project.id),
assignee_id=None,
ai_recommended_assignees=None,
title="搭建政策抓取与公告解析管道",
description="每天跟踪江苏/浙江/安徽 医保支付改革、招采公告与价格体系拐点。",
required_skill_tag_ids=[self.policy_skill_id, self.web_skill_id],
status="todo",
priority="high",
estimated_hours=18.0,
actual_hours=0.0,
quality_score=None,
due_date=datetime(2026, 3, 19),
completed_at=None,
)
task_two = Task(
tenant_id=1,
truth_node_id=int(charter.id),
project_id=int(project.id),
assignee_id=None,
ai_recommended_assignees=None,
title="沉淀器械中标结构知识图谱",
description="把区域政策、招采公告和竞品动作串成结构化图谱。",
required_skill_tag_ids=[self.graph_skill_id],
status="todo",
priority="medium",
estimated_hours=12.0,
actual_hours=0.0,
quality_score=None,
due_date=datetime(2026, 3, 20),
completed_at=None,
)
db.add_all([task_one, task_two])
await db.flush()
return {
"project": {"id": int(project.id), "title": project.title, "name": project.name},
"signal": {"id": int(bootstrap_signal.id), "title": bootstrap_signal.title},
"tasks": [
{"id": int(task_one.id), "title": task_one.title},
{"id": int(task_two.id), "title": task_two.title},
],
}
with patch(
"app.services.ai_pmo_kernel_service.AIStewardService.create_project_from_natural_language",
AsyncMock(side_effect=create_project_side_effect),
):
project_result = await kernel.process_ingress(
source_system="feishu",
source_message_id="om-project-hard-1",
message_text=(
"不要复用现在这个宏观研究盘子,单独立一个项目,名字就叫"
"“华东医保控费与器械国产替代观察台”,目标不是写周报,"
"而是 10 天内拿出策略 memo、政策雷达和竞品跟踪框架。"
),
sender="老板",
channel_ref="oc_complex_case",
source_sender_open_id="ou_boss",
)
new_project_id = int(project_result["created_project_id"])
signal_result = await kernel.process_ingress(
source_system="feishu",
source_message_id="om-signal-hard-1",
message_text=(
f"给项目ID: {new_project_id} 新增一条信号:持续盯江苏、浙江、安徽 2026 年器械带量采购、"
"DRG-DIP 支付改革和国产替代中标结构,只要价格体系或准入口径出现拐点就预警,"
"不要只抓新闻标题。"
),
sender="老板",
channel_ref="oc_complex_case",
source_sender_open_id="ou_boss",
project_hint_id=new_project_id,
)
source_result = await kernel.process_ingress(
source_system="feishu",
source_message_id="om-source-hard-1",
message_text=(
f"给项目ID: {new_project_id} 接一个复合信号源,不是简单 RSS:每天自动抓医保支付改革、"
"器械招采公告、上市公司电话会和 GitHub 开源竞品动态,优先走自动接入。"
),
sender="老板",
channel_ref="oc_complex_case",
source_sender_open_id="ou_boss",
project_hint_id=new_project_id,
)
db.add_all(
[
UserProject(tenant_id=1, user_id=int(strategy_user.id), project_id=new_project_id),
UserProject(tenant_id=1, user_id=int(crawler_user.id), project_id=new_project_id),
UserProject(tenant_id=1, user_id=int(graph_user.id), project_id=new_project_id),
]
)
await db.flush()
with patch(
"app.services.ai_pmo_kernel_service.SignalSourceDomainWorkflow.run",
AsyncMock(
return_value={
"status": "completed",
"registry_id": 9,
"registry_status": "active",
"onboarding_status": "completed",
"openclaw_status": "completed",
"typed_actions": [{"operation_code": "task.create", "status": "applied"}],
"material_action_count": 1,
"worker_executions": [{"used_fallback": False}],
"pending_workers": [],
"next_step": "继续自动执行后续任务",
"fallback_used": False,
"raw_result": {},
}
),
) as onboarding_run_mock:
resume_result = await kernel.process_ingress(
source_system="ai_pmo_ui",
source_message_id="resume-work-item-hard-1",
message_text=f"继续处理 work_item #{int(source_result['work_item_id'])}",
sender="老板",
channel_ref="ai_pmo_ui",
project_hint_id=new_project_id,
)
target_work_item = (
await db.execute(
select(WorkItem).where(
and_(
WorkItem.tenant_id == 1,
WorkItem.id == int(source_result["work_item_id"]),
)
)
)
).scalar_one()
new_project_tasks = (
await db.execute(
select(Task)
.where(and_(Task.tenant_id == 1, Task.project_id == new_project_id))
.order_by(Task.id.asc())
)
).scalars().all()
task_policy_stream = next(task for task in new_project_tasks if "抓取" in task.title)
task_graph = next(task for task in new_project_tasks if "知识图谱" in task.title)
recommendation_result = await recommend_personnel(
db=db,
tenant_id=1,
required_skills=["policy-analysis", "web-crawling"],
task_description=task_policy_stream.description,
task_id=str(task_policy_stream.id),
limit=5,
)
with patch(
"app.services.workload_service.llm_service.generate",
AsyncMock(
return_value=json.dumps(
{
"proposals": [
{
"task_id": int(task_policy_stream.id),
"new_assignee_id": int(crawler_user.id),
"reason": "crawler engineer has exact skill mix and lower load",
},
{
"task_id": int(task_graph.id),
"new_assignee_id": int(graph_user.id),
"reason": "graph engineer owns ontology and extraction pipeline",
},
]
},
ensure_ascii=False,
)
),
):
optimize_preview = await workload_service.build_ai_optimize_preview(
db,
tenant_id=1,
project_id=new_project_id,
max_moves=5,
unassigned_only=True,
)
optimize_result = await workload_service.apply_ai_optimize_plan(
db,
tenant_id=1,
proposals=optimize_preview,
)
board_rows = await workload_service.get_team_workload_board(db, tenant_id=1, project_id=new_project_id)
cadence_service = AIPMOCadenceService(db, tenant_id=1)
program_view = await cadence_service.build_program_view()
self.assertEqual(qa_result["status"], "answered")
self.assertIn("blocked task", qa_result["direct_reply_text"])
cache_append_mock.assert_awaited_once()
self.assertEqual(cache_append_mock.await_args.kwargs["question"], hard_question)
self.assertEqual(project_result["status"], "captured")
self.assertIsNotNone(project_result["created_project_id"])
self.assertIn("已立项", project_result["direct_reply_text"])
self.assertEqual(signal_result["status"], "captured")
self.assertEqual(signal_result["linked_project_ids"], [new_project_id])
self.assertIsNotNone(signal_result["signal_id"])
self.assertEqual(source_result["status"], "captured")
self.assertIn("技能缺口", source_result["direct_reply_text"])
self.assertFalse(bool(source_result["review_summary"]["verified"]))
self.assertEqual(onboarding_run_mock.await_count, 1)
self.assertEqual(resume_result["status"], "answered")
self.assertIn("已继续处理", resume_result["direct_reply_text"])
self.assertEqual(target_work_item.state, "succeeded")
self.assertEqual(target_work_item.execution_result["signal_source_workflow"]["status"], "completed")
self.assertGreaterEqual(recommendation_result["total_count"], 1)
self.assertEqual(recommendation_result["recommendations"][0]["user_name"], "周湛")
self.assertGreaterEqual(len(optimize_preview), 2)
self.assertEqual(optimize_result["applied_count"], 2)
task_assignee_map = {int(task.id): int(task.assignee_id or 0) for task in new_project_tasks}
self.assertEqual(task_assignee_map[int(task_policy_stream.id)], int(crawler_user.id))
self.assertEqual(task_assignee_map[int(task_graph.id)], int(graph_user.id))
board_by_name = {str(row["user_name"]): row for row in board_rows}
self.assertIn("周湛", board_by_name)
self.assertIn("顾言", board_by_name)
self.assertGreater(float(board_by_name["周湛"]["current_assigned_hours"]), 8.0)
self.assertGreaterEqual(float(board_by_name["顾言"]["current_assigned_hours"]), 9.5)
self.assertGreaterEqual(program_view["summary"]["program_count"], 1)
self.assertTrue(
any(new_project_id in program["project_ids"] for program in program_view["programs"])
)