Files
agent_contas/tests/migration/test_multi_intent_contas.py

111 lines
4.7 KiB
Python

from agent_framework.routing.config_loader import load_intents, load_multi_intent_config
import pytest
from agent_framework.routing.multi_intent import MultiIntentPlanner
from agent_framework.routing.pending_topics import drain_pending_topics
from agent_framework.mcp.tool_policy import ToolPolicyRegistry
def _planner():
path = "config/routing.yaml"
registry = ToolPolicyRegistry("config/tool_policies.yaml")
transactional = {name for name, policy in registry.policies.items() if policy.operation_type == "transactional"}
return MultiIntentPlanner(load_intents(path), load_multi_intent_config(path), transactional_tools=transactional)
def test_cancelamento_e_segunda_via_preserva_cancelamento_como_primario():
plan = _planner().plan("Cancela o tamboro e manda a segunda via para mim")
assert plan is not None
assert plan.operations[0].intent == "contas_vas_cancel"
assert plan.operations[0].agent == "contestacao_agent"
assert plan.operations[1].intent == "contas_invoice_query"
assert plan.operations[1].disposition == "execute"
def test_sim_composto_preserva_confirmacao_e_topico_secundario():
plan = _planner().plan_confirmation_secondary(
"Sim e me manda a segunda via",
primary_intent="contas_vas_cancel",
primary_agent="contestacao_agent",
)
assert plan is not None
assert plan.operations[0].intent == "contas_vas_cancel"
assert plan.operations[1].intent == "contas_invoice_query"
def test_pedido_secundario_realmente_fora_do_escopo_vira_pending_topic():
plan = _planner().plan("Cancela o tamboro e altera o CNPJ da conta")
assert plan is not None
assert plan.operations[0].intent == "contas_vas_cancel"
assert plan.operations[1].intent == "off_context"
assert plan.operations[1].disposition == "unsupported"
def test_duas_mais_caras_sao_calculadas_sobre_evidencia_autorizada():
from contas_mcp.servers.contas_mcp_server.main import _select_relative_invoice_charges
args = {
"focused_subject": "Tamboro Mensal",
"invoice_detail": {"119": {"SVA Detalhe Total": [
{"desc": "Tamboro Mensal", "value": 9.90, "date": "01/09/26", "classe": "avulso"},
{"desc": "Tamboro Mensal", "value": 29.90, "date": "02/09/26", "classe": "avulso"},
{"desc": "Tamboro Mensal", "value": 14.90, "date": "03/09/26", "classe": "avulso"},
{"desc": "Tamboro Mensal", "value": 24.90, "date": "04/09/26", "classe": "avulso"},
]}},
}
selected = _select_relative_invoice_charges("as duas mais caras", args)
assert [row["value"] for row in selected] == [29.90, 24.90]
assert [row["charge_date"] for row in selected] == ["02/09/26", "04/09/26"]
@pytest.mark.asyncio
async def test_promocao_transacional_substitui_snapshot_de_continuidade_anterior():
async def orders_handler(state):
assert state["intent"] == "retail_order_cancel"
return {
"answer": "Informe o número do pedido que deseja cancelar.",
"transaction_status": "COLLECTING_PARAMETERS",
"active_transaction": {
"tool_name": "cancelar_pedido",
"status": "COLLECTING_PARAMETERS",
},
"missing_parameters": ["order_id"],
"pending_tool_call": {
"tool_name": "cancelar_pedido",
"arguments": {},
},
"next_state": "COLLECTING_PARAMETERS",
}
drained = await drain_pending_topics(
{
"answer": "Devolução concluída.",
"route": "support_agent",
"active_agent": "support_agent",
"intent": "retail_support_exchange_return",
"transaction_status": "COMPLETED",
"route_decision": {
"route": "support_agent",
"agent": "support_agent",
"intent": "state:WAITING_SUPPORT_CONFIRMATION",
},
"pending_topics": [{
"operation_id": "op-3",
"agent": "orders_agent",
"intent": "retail_order_cancel",
"domain": "retail",
"tools": ["consultar_pedido", "cancelar_pedido"],
"source_text": "quero cancelar pedido",
"disposition": "defer",
}],
},
"Devolução concluída.",
{"orders_agent": orders_handler},
)
assert drained.state_patch["route"] == "orders_agent"
assert drained.state_patch["transaction_status"] == "COLLECTING_PARAMETERS"
assert drained.state_patch["route_decision"]["agent"] == "orders_agent"
assert drained.state_patch["route_decision"]["intent"] == "retail_order_cancel"
assert drained.state_patch["route_decision"]["metadata"]["promoted_from_pending_topics"] is True