74 lines
2.5 KiB
Python
74 lines
2.5 KiB
Python
import pytest
|
|
|
|
from app.services.ai_pmo.ai_pmo_decision.tactical import TacticalBrain
|
|
from app.services.ai_pmo.ai_pmo_intent import (
|
|
INTENT_FOLLOWUP_CONTINUE,
|
|
INTENT_QUERY,
|
|
INTENT_SIGNAL_CREATE,
|
|
ROUTE_PROJECT_SIGNAL_CANCEL,
|
|
ROUTE_PROJECT_SIGNAL_COMMIT,
|
|
ROUTE_PROJECT_SIGNAL_DISCOVER,
|
|
ROUTE_PROJECT_SIGNAL_REFRESH,
|
|
CanonicalIntent,
|
|
)
|
|
|
|
|
|
def _build_intent(name: str, *, metadata: dict | None = None) -> CanonicalIntent:
|
|
return CanonicalIntent(
|
|
name=name,
|
|
confidence=0.9,
|
|
project_id=11,
|
|
signal_title="",
|
|
signal_summary="",
|
|
metadata=metadata or {},
|
|
)
|
|
|
|
|
|
def test_tactical_brain_query_intent_maps_to_read_only_answer_action() -> None:
|
|
_strategic_plan, execution_plan = TacticalBrain.plan(
|
|
intent=_build_intent(INTENT_QUERY),
|
|
world_model={},
|
|
decision={},
|
|
)
|
|
assert execution_plan["intent"] == INTENT_QUERY
|
|
assert execution_plan["actions"] == [{"operation_code": "query.answer", "mode": "read_only"}]
|
|
|
|
|
|
def test_tactical_brain_followup_intent_maps_to_resume_action() -> None:
|
|
_strategic_plan, execution_plan = TacticalBrain.plan(
|
|
intent=_build_intent(INTENT_FOLLOWUP_CONTINUE),
|
|
world_model={},
|
|
decision={},
|
|
)
|
|
assert execution_plan["actions"] == [{"operation_code": "work_item.resume", "mode": "write"}]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("workflow_route", "expected_operation_code"),
|
|
[
|
|
(ROUTE_PROJECT_SIGNAL_DISCOVER, ROUTE_PROJECT_SIGNAL_DISCOVER),
|
|
(ROUTE_PROJECT_SIGNAL_COMMIT, ROUTE_PROJECT_SIGNAL_COMMIT),
|
|
(ROUTE_PROJECT_SIGNAL_REFRESH, ROUTE_PROJECT_SIGNAL_REFRESH),
|
|
(ROUTE_PROJECT_SIGNAL_CANCEL, ROUTE_PROJECT_SIGNAL_CANCEL),
|
|
],
|
|
)
|
|
def test_tactical_brain_signal_create_route_specific_action_mapping(
|
|
workflow_route: str,
|
|
expected_operation_code: str,
|
|
) -> None:
|
|
_strategic_plan, execution_plan = TacticalBrain.plan(
|
|
intent=_build_intent(INTENT_SIGNAL_CREATE, metadata={"workflow_route": workflow_route}),
|
|
world_model={},
|
|
decision={},
|
|
)
|
|
assert execution_plan["actions"] == [{"operation_code": expected_operation_code, "mode": "write"}]
|
|
|
|
|
|
def test_tactical_brain_signal_create_defaults_to_signal_enter() -> None:
|
|
_strategic_plan, execution_plan = TacticalBrain.plan(
|
|
intent=_build_intent(INTENT_SIGNAL_CREATE, metadata={"workflow_route": "project_signal.unknown"}),
|
|
world_model={},
|
|
decision={},
|
|
)
|
|
assert execution_plan["actions"] == [{"operation_code": "signal.enter", "mode": "write"}]
|