126 lines
4.0 KiB
Python
126 lines
4.0 KiB
Python
import pytest
|
|
|
|
from app.services.ai_pmo.ai_pmo_intent import (
|
|
INTENT_FOLLOWUP_CONTINUE,
|
|
INTENT_QUERY,
|
|
INTENT_TASK_CREATE,
|
|
ROUTE_ACTION_ANSWER,
|
|
ROUTE_ACTION_CANCEL,
|
|
ROUTE_ACTION_CREATE,
|
|
ROUTE_FAMILY_QUERY,
|
|
ROUTE_FAMILY_TASK,
|
|
ROUTE_QUERY_ANSWER,
|
|
ROUTE_TASK_CREATE,
|
|
CanonicalIntent,
|
|
IntentResolver,
|
|
RouteResolver,
|
|
)
|
|
from app.services.ai_pmo.ai_pmo_kernel_common_core.imports import WORKFLOW_ROUTE_KEYS
|
|
from app.services.ai_pmo.ai_pmo_typed_executor_execute.service import TypedExecutorExecuteMixin
|
|
|
|
|
|
def test_intent_resolver_recognizes_task_create() -> None:
|
|
intent = IntentResolver.resolve(
|
|
text="请在项目#12 新建任务：修复登录超时",
|
|
project_id=12,
|
|
)
|
|
assert intent.name == INTENT_TASK_CREATE
|
|
assert intent.project_id == 12
|
|
assert str((intent.metadata or {}).get("task_title") or "").startswith("修复登录超时")
|
|
|
|
|
|
def test_route_resolver_builds_task_create_route() -> None:
|
|
route = RouteResolver.from_canonical(
|
|
canonical_intent=CanonicalIntent(
|
|
name=INTENT_TASK_CREATE,
|
|
confidence=0.9,
|
|
project_id=3,
|
|
signal_title="修复登录超时",
|
|
signal_summary="修复线上登录超时问题",
|
|
metadata={},
|
|
),
|
|
message_text="新增任务：修复登录超时",
|
|
)
|
|
assert route.family == ROUTE_FAMILY_TASK
|
|
assert route.action == ROUTE_ACTION_CREATE
|
|
assert route.key == ROUTE_TASK_CREATE
|
|
|
|
|
|
def test_intent_resolver_negated_write_statement_falls_back_to_query_read_only() -> None:
|
|
intent = IntentResolver.resolve(
|
|
text="先不要创建任务，告诉我当前进展和风险",
|
|
project_id=7,
|
|
)
|
|
assert intent.name == INTENT_QUERY
|
|
assert str((intent.metadata or {}).get("reason") or "") == "negated_write_statement"
|
|
|
|
|
|
def test_route_resolver_builds_query_answer_route() -> None:
|
|
route = RouteResolver.from_canonical(
|
|
canonical_intent=CanonicalIntent(
|
|
name=INTENT_QUERY,
|
|
confidence=0.92,
|
|
project_id=7,
|
|
signal_title="",
|
|
signal_summary="当前进展和风险",
|
|
metadata={},
|
|
),
|
|
message_text="当前进展和风险是什么？",
|
|
)
|
|
assert route.family == ROUTE_FAMILY_QUERY
|
|
assert route.action == ROUTE_ACTION_ANSWER
|
|
assert route.key == ROUTE_QUERY_ANSWER
|
|
|
|
|
|
def test_route_resolver_followup_action_supports_message_text_fallback() -> None:
|
|
action = RouteResolver.parse_followup_action(
|
|
canonical_intent=INTENT_FOLLOWUP_CONTINUE,
|
|
message_text="先取消这个工作项",
|
|
)
|
|
assert action == ROUTE_ACTION_CANCEL
|
|
|
|
|
|
def test_kernel_common_imports_exposes_workflow_route_keys() -> None:
|
|
assert "query.answer" in WORKFLOW_ROUTE_KEYS
|
|
assert "task.create" in WORKFLOW_ROUTE_KEYS
|
|
|
|
|
|
class _DummyTypedExecutor(TypedExecutorExecuteMixin):
|
|
def __init__(self) -> None:
|
|
self.called: dict[str, object] = {}
|
|
|
|
async def _execute_task_create_intent(self, *, intent, message_text: str, operator_user_id):
|
|
self.called = {
|
|
"intent": intent.name,
|
|
"message_text": message_text,
|
|
"operator_user_id": operator_user_id,
|
|
}
|
|
return {"status": "succeeded", "task_id": 77}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_typed_executor_dispatches_task_create() -> None:
|
|
executor = _DummyTypedExecutor()
|
|
intent = CanonicalIntent(
|
|
name=INTENT_TASK_CREATE,
|
|
confidence=0.88,
|
|
project_id=6,
|
|
signal_title="补充埋点",
|
|
signal_summary="补充行为埋点以支持漏斗分析",
|
|
metadata={"task_title": "补充埋点"},
|
|
)
|
|
|
|
result = await executor.execute(
|
|
intent=intent,
|
|
message_text="请创建任务：补充埋点",
|
|
sender="alice",
|
|
operator_user_id=101,
|
|
source_system="feishu",
|
|
source_message_id="msg-1",
|
|
)
|
|
|
|
assert result["status"] == "succeeded"
|
|
assert result["task_id"] == 77
|
|
assert executor.called["intent"] == INTENT_TASK_CREATE
|
|
assert executor.called["operator_user_id"] == 101
|