from __future__ import annotations import pytest def _result(name: str, status: str, output: dict, *, error: str | None = None, error_details=None): return { "execution_id": f"exec-{name}", "workflow_name": name, "workflow_version": 1, "status": status, "output": output, "state": {}, "trace": [], "error": error, "error_details": error_details, } class Service: def consultar_faturas(self, *, msisdn: str): return { "billingProfile": {"customer": {"document": "12345678901", "customerId": "344575482"}}, "paymentItems": [{"invoiceId": "5536806781", "dueDate": "2025-08-11"}], } @pytest.mark.asyncio async def test_cancelamento_item_ja_contestado_preserva_sucesso_do_cancelamento(monkeypatch): import contas_mcp.servers.contas_mcp_server.main as main class Runtime: def __init__(self): self.calls=[] async def arun(self, name, payload, execution_id=None): self.calls.append((name,payload,execution_id)) if name == "cancelamento_vas_avulso": return _result(name, "COMPLETED", {"cancelar_vas_avulso": { "results": [{"success": True, "subject": "TIM Fashion Mensal", "msisdn": payload["msisdn"]}], "contestation_candidates": [{"success": True, "subject": "TIM Fashion Mensal", "msisdn": payload["msisdn"]}], "protocolos_por_linha": [{"msisdn": payload["msisdn"], "protocolo_id": "PRT-CANCEL-1"}], }}) return _result(name, "FAILED", {"registrar_protocolo": {"protocolo_id": "PRT-102"}}, error="Itens já contestados ou conflito de itens da fatura", error_details={"status_code": 409, "body": {"provider": {"errorCode": "ITEM_JA_CONTESTADO"}}}) runtime=Runtime() monkeypatch.setattr(main, "get_workflow_runtime", lambda: runtime) monkeypatch.setattr(main, "service", Service()) result = await main._run_cancelamento_com_contestacao({"msisdn":"5511999999999","subject":"TIM Fashion Mensal","valor":"14,99"}) assert result["success"] is True assert result["erro_sistemico"] is False assert result["auto_finalize_on_failure"] is False assert result["protocol_closed"] is False assert result["contestacao_protocol"] == "PRT-102" assert result["cancelamento_vas_protocol"] == "PRT-CANCEL-1" assert result["protocols_for_response"] == ["PRT-CANCEL-1", "PRT-102"] assert "já contestados" in result["mensagem"] @pytest.mark.asyncio async def test_cancelamento_falha_tecnica_na_contestacao_recomenda_auto_finalizacao(monkeypatch): import contas_mcp.servers.contas_mcp_server.main as main class Runtime: async def arun(self, name, payload, execution_id=None): if name == "cancelamento_vas_avulso": return _result(name, "COMPLETED", {"cancelar_vas_avulso": { "results": [{"success": True, "subject": "TIM Fashion Mensal", "msisdn": payload["msisdn"]}], "contestation_candidates": [{"success": True, "subject": "TIM Fashion Mensal", "msisdn": payload["msisdn"]}], }}) return _result(name, "FAILED", {}, error="timeout no provider", error_details={"status_code": 504}) monkeypatch.setattr(main, "get_workflow_runtime", lambda: Runtime()) monkeypatch.setattr(main, "service", Service()) result = await main._run_cancelamento_com_contestacao({"msisdn":"5511999999999","subject":"TIM Fashion Mensal","valor":"14,99"}) assert result["success"] is False assert result["erro_sistemico"] is True assert result["auto_finalize_on_failure"] is True assert result["mensagem"] == "timeout no provider" @pytest.mark.asyncio async def test_cancelamento_sem_contestacao_expoe_protocolo_de_cancelamento(monkeypatch): import contas_mcp.servers.contas_mcp_server.main as main class Runtime: async def arun(self, name, payload, execution_id=None): return _result(name, "COMPLETED", {"cancelar_vas_avulso": { "results": [{"success": True, "subject": "TIM Fashion Mensal", "msisdn": payload["msisdn"]}], "contestation_candidates": [], "protocolos_por_linha": [{"msisdn": payload["msisdn"], "protocolo_id": "PRT-CANCEL-1"}], }}) monkeypatch.setattr(main, "get_workflow_runtime", lambda: Runtime()) result = await main._run_cancelamento_com_contestacao({"msisdn":"5511999999999","subject":"TIM Fashion Mensal"}) assert result["cancelamento_vas_protocol"] == "PRT-CANCEL-1" assert result["protocols_for_response"] == ["PRT-CANCEL-1"] assert result["requires_protocol_in_response"] is True @pytest.mark.asyncio async def test_familia_contesta_titular_e_dependente_em_uma_unica_chamada(monkeypatch): import contas_mcp.servers.contas_mcp_server.main as main class Runtime: def __init__(self): self.calls=[] async def arun(self, name, payload, execution_id=None): self.calls.append((name,payload,execution_id)) if name == "cancelamento_vas_avulso": return _result(name, "COMPLETED", {"cancelar_vas_avulso": { "results": [ {"success": True, "subject": "TIM Fashion Mensal", "msisdn": "5511888880000"}, {"success": True, "subject": "TIM Games", "msisdn": "5511999999999"}, ], "contestation_candidates": [ {"success": True, "subject": "TIM Fashion Mensal", "msisdn": "5511888880000"}, {"success": True, "subject": "TIM Games", "msisdn": "5511999999999"}, ], }}) return _result(name, "COMPLETED", {"registrar_protocolo":{"protocolo_id":"PRT-FAM"}, "abrir_contestacao_cliente":{"success":True}}) runtime=Runtime() monkeypatch.setattr(main,"get_workflow_runtime",lambda:runtime) monkeypatch.setattr(main,"service",Service()) await main._run_cancelamento_com_contestacao({ "msisdn":"5511888880000", "items":[ {"msisdn":"5511888880000","name":"TIM Fashion Mensal","value":"14,99"}, {"msisdn":"5511999999999","name":"TIM Games","value":"9,90"}, ], }) assert [x[0] for x in runtime.calls] == ["cancelamento_vas_avulso","contestacao_tool"] contest=runtime.calls[1][1] assert contest["msisdn"] == "5511888880000" assert contest["item_msisdn"] == "5511888880000" assert contest["dependent_invoice_item"] is True assert {x["itemName"] for x in contest["items"]} == {"TIM Fashion Mensal","TIM Games"} @pytest.mark.asyncio async def test_lote_trinta_contesta_somente_unico_candidato(monkeypatch): import contas_mcp.servers.contas_mcp_server.main as main class Runtime: def __init__(self): self.calls=[] async def arun(self,name,payload,execution_id=None): self.calls.append((name,payload,execution_id)) if name == "cancelamento_vas_avulso": failures=[{"success":False,"subject":f"VAS não encontrado {i}","msisdn":payload["msisdn"],"reason":"service_not_found","error":"não encontrado"} for i in range(1,30)] ok={"success":True,"subject":"TIM Fashion Mensal","msisdn":payload["msisdn"]} return _result(name,"COMPLETED",{"cancelar_vas_avulso":{"results":[ok,*failures],"contestation_candidates":[ok]}}) return _result(name,"COMPLETED",{"registrar_protocolo":{"protocolo_id":"PRT-PARCIAL"},"abrir_contestacao_cliente":{"success":True}}) runtime=Runtime() monkeypatch.setattr(main,"get_workflow_runtime",lambda:runtime) monkeypatch.setattr(main,"service",Service()) items=[{"msisdn":"5511999999999","name":"TIM Fashion Mensal","value":"14,99"}, *[ {"msisdn":"5511999999999","name":f"VAS não encontrado {i}","value":"22,00"} for i in range(1,30) ]] result=await main._run_cancelamento_com_contestacao({"msisdn":"5511999999999","items":items}) assert len(runtime.calls)==2 assert [x["itemName"] for x in runtime.calls[1][1]["items"]] == ["TIM Fashion Mensal"] assert result["success"] is True assert result["contestacao_protocol"] == "PRT-PARCIAL" @pytest.mark.asyncio async def test_wrapper_itens_ja_contestados_preserva_mensagem_deterministica(monkeypatch): from contas_mcp.servers.contas_mcp_server import main cancel = _result("cancelamento_vas_avulso", "COMPLETED", {"cancelar_vas_avulso": { "results": [{"success": True, "msisdn": "11999999999", "subject": "TIM Fashion Mensal"}], "contestation_candidates": [{"success": True, "msisdn": "11999999999", "subject": "TIM Fashion Mensal"}], }}) contest = _result("contestacao_tool", "COMPLETED", { "registrar_protocolo": {"protocolo_id": "PRT-100"}, "abrir_contestacao_cliente": { "success": True, "itens_ja_contestados": ["TIM Fashion Mensal"], "items_response": [], "contested_invoice_amount_open": "0", "contested_invoice_amount": "0", }, }) class Runtime: def __init__(self): self.i=0 async def arun(self, name, payload, execution_id=None): self.i += 1 return cancel if self.i == 1 else contest runtime = Runtime() monkeypatch.setattr(main, "get_workflow_runtime", lambda: runtime) monkeypatch.setattr(main, "service", Service()) result = await main._run_cancelamento_com_contestacao({ "msisdn": "11999999999", "subject": "TIM Fashion Mensal", "valor": 14.99, "items": [{"msisdn": "11999999999", "name": "TIM Fashion Mensal", "value": 14.99}], }) assert result["success"] is True assert result["mensagem"] == ( "Identifiquei que o item TIM Fashion Mensal já havia sido contestado anteriormente e, " "por esse motivo, não é possível registrar uma nova contestação para ele. " "Seu número de protocolo é PRT-100." ) @pytest.mark.asyncio async def test_wrapper_contestacao_zerada_usa_total_cancelado_na_mensagem(monkeypatch): from contas_mcp.servers.contas_mcp_server import main cancel = _result("cancelamento_vas_avulso", "COMPLETED", {"cancelar_vas_avulso": { "results": [{"success": True, "msisdn": "11999999999", "subject": "TIM Fashion Mensal"}], "contestation_candidates": [{"success": True, "msisdn": "11999999999", "subject": "TIM Fashion Mensal"}], }}) contest = _result("contestacao_tool", "COMPLETED", { "registrar_protocolo": {"protocolo_id": "PRT-100"}, "abrir_contestacao_cliente": { "success": True, "items_response": [{"status": "CRIAR", "itemName": "TIM Fashion Mensal"}], "contested_invoice_amount_open": "0", "contested_invoice_amount": "0", }, }) class Runtime: def __init__(self): self.i=0 async def arun(self, name, payload, execution_id=None): self.i += 1 return cancel if self.i == 1 else contest runtime = Runtime() monkeypatch.setattr(main, "get_workflow_runtime", lambda: runtime) monkeypatch.setattr(main, "service", Service()) result = await main._run_cancelamento_com_contestacao({ "msisdn": "11999999999", "subject": "TIM Fashion Mensal", "valor": 14.99, "items": [{"msisdn": "11999999999", "name": "TIM Fashion Mensal", "value": 14.99}], }) assert "R$ 14,99" in result["mensagem"] @pytest.mark.asyncio async def test_wrapper_next_subject_nao_afeta_mensagem_deterministica(monkeypatch): from contas_mcp.servers.contas_mcp_server import main cancel = _result("cancelamento_vas_avulso", "COMPLETED", {"cancelar_vas_avulso": { "results": [{"success": True, "msisdn": "11999999999", "subject": "TIM Fashion Mensal"}], "contestation_candidates": [{"success": True, "msisdn": "11999999999", "subject": "TIM Fashion Mensal"}], }}) contest = _result("contestacao_tool", "COMPLETED", { "registrar_protocolo": {"protocolo_id": "PRT-100"}, "abrir_contestacao_cliente": { "success": True, "items_response": [{"status": "CRIAR", "itemName": "TIM Fashion Mensal"}], "contested_invoice_amount_open": "14,99", }, }) class Runtime: def __init__(self): self.i=0 async def arun(self, name, payload, execution_id=None): self.i += 1 return cancel if self.i == 1 else contest runtime = Runtime() monkeypatch.setattr(main, "get_workflow_runtime", lambda: runtime) monkeypatch.setattr(main, "service", Service()) result = await main._run_cancelamento_com_contestacao({ "msisdn": "11999999999", "subject": "TIM Fashion Mensal", "valor": 14.99, "next_subject": "contestação de fatura", "items": [{"msisdn": "11999999999", "name": "TIM Fashion Mensal", "value": 14.99}], }) assert "contestação de fatura" not in result["mensagem"]