56 lines
2.3 KiB
Python
56 lines
2.3 KiB
Python
import pytest
|
|
|
|
from agent_framework.workflows.models import WorkflowRunResult
|
|
import contas_mcp.servers.contas_mcp_server.main as mcp_main
|
|
|
|
|
|
class Runtime:
|
|
def __init__(self, result):
|
|
self.result = result
|
|
async def arun(self, name, payload, execution_id=None):
|
|
return self.result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_contestacao_failed_preserva_provider_message_e_protocolo(monkeypatch):
|
|
result = WorkflowRunResult(
|
|
execution_id="e1",
|
|
workflow_name="contestacao_tool",
|
|
workflow_version=2,
|
|
status="FAILED",
|
|
error="500 provider",
|
|
error_details={
|
|
"status_code": 550,
|
|
"body": {
|
|
"description": "Itens já contestados",
|
|
"provider": {"errorMessage": "Impossível prosseguir, pois os itens já foram contestados."},
|
|
},
|
|
"attempts": [{"attempt": 1, "success": False}],
|
|
},
|
|
output={"registrar_protocolo": {"protocolo_id": "PRT-123"}},
|
|
state={}, trace=[],
|
|
)
|
|
monkeypatch.setattr(mcp_main, "get_workflow_runtime", lambda: Runtime(result))
|
|
payload = await mcp_main._run_workflow("contestacao_tool", {"msisdn": "119", "subject": "VAS", "valor": 10})
|
|
assert payload["success"] is False
|
|
assert payload["contestation_error_description"].startswith("Impossível prosseguir")
|
|
assert payload["mensagem"] == payload["contestation_error_description"]
|
|
assert payload["contestacao_protocol"] == "PRT-123"
|
|
assert payload["erro"] == "contestacao_nao_realizada"
|
|
assert payload["erro_sistemico"] is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_contestacao_failed_sem_mensagem_provider_vira_falha_sistemica(monkeypatch):
|
|
result = WorkflowRunResult(
|
|
execution_id="e2", workflow_name="contestacao_tool", workflow_version=2,
|
|
status="FAILED", error="connection reset",
|
|
error_details={"type": "ConnectionError"}, output={}, state={}, trace=[],
|
|
)
|
|
monkeypatch.setattr(mcp_main, "get_workflow_runtime", lambda: Runtime(result))
|
|
payload = await mcp_main._run_workflow("contestacao_tool", {"msisdn": "119", "subject": "VAS", "valor": 10})
|
|
assert payload["success"] is False
|
|
assert payload["erro"] == "erro_falha_sistema"
|
|
assert payload["erro_sistemico"] is True
|
|
assert "contestation_error_description" not in payload
|