93 lines
2.8 KiB
Python
93 lines
2.8 KiB
Python
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from app.extensions.tim_guardrails import TimProactiveOfferRail
|
|
|
|
|
|
class _FailIfCalledLLM:
|
|
async def ainvoke(self, *args, **kwargs):
|
|
raise AssertionError('TIM_AOFERTA LLM must not run for transaction continuation')
|
|
|
|
|
|
class _BlockingLLM:
|
|
def __init__(self):
|
|
self.calls = 0
|
|
|
|
async def ainvoke(self, *args, **kwargs):
|
|
self.calls += 1
|
|
return '{"allowed": false, "reason": "oferta proativa"}'
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize('status', ['COLLECTING_PARAMETERS', 'AWAITING_CONFIRMATION'])
|
|
async def test_tim_aoferta_bypasses_native_transaction_continuation_states(status):
|
|
decision = await TimProactiveOfferRail().evaluate(
|
|
'Você confirma o cancelamento do serviço TIM Fashion?',
|
|
{
|
|
'transaction_status': status,
|
|
'guardrail_llm': _FailIfCalledLLM(),
|
|
},
|
|
)
|
|
|
|
assert decision.allowed is True
|
|
assert decision.code == 'TIM_AOFERTA'
|
|
assert decision.reason == f'continuidade_transacional:{status}'
|
|
assert decision.metadata['mechanism'] == 'deterministic_transaction_bypass'
|
|
assert decision.metadata['transaction_status'] == status
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_tim_aoferta_detects_awaiting_confirmation_from_mcp_results():
|
|
decision = await TimProactiveOfferRail().evaluate(
|
|
'Você confirma o cancelamento do serviço TIM Fashion?',
|
|
{
|
|
'mcp_results': [
|
|
{
|
|
'tool_name': 'cancelar_vas_avulso',
|
|
'awaiting_confirmation': True,
|
|
'transaction_status': 'AWAITING_CONFIRMATION',
|
|
}
|
|
],
|
|
'guardrail_llm': _FailIfCalledLLM(),
|
|
},
|
|
)
|
|
|
|
assert decision.allowed is True
|
|
assert decision.metadata['transaction_status'] == 'AWAITING_CONFIRMATION'
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_tim_aoferta_detects_parameter_collection_from_tool_result():
|
|
decision = await TimProactiveOfferRail().evaluate(
|
|
'Para prosseguir, informe valor.',
|
|
{
|
|
'tool_result': [
|
|
{
|
|
'tool_name': 'contestar_cobranca',
|
|
'transaction_status': 'COLLECTING_PARAMETERS',
|
|
}
|
|
],
|
|
'guardrail_llm': _FailIfCalledLLM(),
|
|
},
|
|
)
|
|
|
|
assert decision.allowed is True
|
|
assert decision.metadata['transaction_status'] == 'COLLECTING_PARAMETERS'
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_tim_aoferta_still_uses_llm_outside_transaction_continuation():
|
|
llm = _BlockingLLM()
|
|
decision = await TimProactiveOfferRail().evaluate(
|
|
'Caso queira, posso cancelar outro serviço.',
|
|
{
|
|
'transaction_status': 'COMPLETED',
|
|
'guardrail_llm': llm,
|
|
},
|
|
)
|
|
|
|
assert llm.calls == 1
|
|
assert decision.allowed is False
|
|
assert decision.reason == 'oferta proativa'
|