393 lines
19 KiB
Python
393 lines
19 KiB
Python
from app.domain.contas import ContasDomainService
|
|
|
|
|
|
def test_read_only_integrations_with_fixtures(monkeypatch):
|
|
monkeypatch.setenv("TIM_USE_MOCK_GATEWAY", "true")
|
|
svc = ContasDomainService()
|
|
assert svc.consultar_faturas(msisdn="11999999999").get("paymentItems")
|
|
assert svc.consultar_vas(msisdn="11999999999").get("products")
|
|
assert svc.consultar_historico_vas(msisdn="11999999999").get("services")
|
|
|
|
|
|
def test_cancel_vas_is_domain_operation_not_conversation_engine(monkeypatch):
|
|
monkeypatch.setenv("TIM_USE_MOCK_GATEWAY", "true")
|
|
svc = ContasDomainService()
|
|
out = svc.cancelar_vas_avulso(msisdn="11999999999", subject="TIM Fashion Mensal")
|
|
assert out["success"] is True
|
|
assert out["block"]["status"] == 200
|
|
assert out["cancellation"]["status"] == 200
|
|
|
|
|
|
def test_contestation_mock(monkeypatch):
|
|
monkeypatch.setenv("TIM_USE_MOCK_GATEWAY", "true")
|
|
svc = ContasDomainService()
|
|
out = svc.contestar_cobranca(msisdn="11999999999", subject="Tamboro Mensal", valor=14.99, motivo="não reconheço")
|
|
assert out["success"] is True
|
|
assert out["protocol"]
|
|
assert out["contestation"]
|
|
|
|
|
|
def test_cancelamento_usa_historico_como_fallback_quando_ativo_nao_encontra():
|
|
from app.domain.contas.service import ContasDomainService
|
|
|
|
class Client:
|
|
def __init__(self):
|
|
self.blocked = []
|
|
self.cancelled = []
|
|
def consultar_vas(self, msisdn):
|
|
return {"products": []}
|
|
def historico_vas(self, msisdn):
|
|
return {"services": [{"name": "Max mensal fatura", "appId": 14395, "cspId": 825, "canCancel": True}]}
|
|
def bloquear_vas(self, msisdn, service):
|
|
self.blocked.append((msisdn, service))
|
|
return {"ok": True}
|
|
def cancelar_vas(self, msisdn, service, protocol=""):
|
|
self.cancelled.append((msisdn, service, protocol))
|
|
return {"ok": True}
|
|
|
|
client = Client()
|
|
result = ContasDomainService(client=client).cancelar_vas_avulso(
|
|
msisdn="11999999999", subject="Max mensal fatura", protocol="PRT-1"
|
|
)
|
|
assert result["success"] is True
|
|
assert result["evidence_source"] == "history"
|
|
assert client.blocked[0][1]["appId"] == 14395
|
|
assert client.cancelled[0][2] == "PRT-1"
|
|
|
|
|
|
def test_cancelamento_nao_recancela_servico_inativo_do_historico():
|
|
from app.domain.contas.service import ContasDomainService
|
|
|
|
class Client:
|
|
def consultar_vas(self, msisdn):
|
|
return {"products": []}
|
|
def historico_vas(self, msisdn):
|
|
return {"services": [{"name": "Max mensal fatura", "appId": 14395, "cspId": 825, "canCancel": False}]}
|
|
def bloquear_vas(self, *args, **kwargs):
|
|
raise AssertionError("não deve bloquear")
|
|
def cancelar_vas(self, *args, **kwargs):
|
|
raise AssertionError("não deve cancelar")
|
|
|
|
result = ContasDomainService(client=Client()).cancelar_vas_avulso(
|
|
msisdn="11999999999", subject="Max mensal fatura"
|
|
)
|
|
assert result["success"] is False
|
|
assert result["reason"] == "already_inactive_or_blocked"
|
|
|
|
|
|
def test_finalizacao_normaliza_status_invalido_e_summary():
|
|
from app.domain.contas.service import ContasDomainService
|
|
|
|
class Client: pass
|
|
result = ContasDomainService(client=Client()).finalizar_atendimento(
|
|
status="qualquer_coisa", summary=""
|
|
)
|
|
assert result["status"] == "erro_falha_sistema"
|
|
assert result["summary"] == "Encerramento realizado pelo agente."
|
|
|
|
|
|
def test_finalizacao_preserva_status_valido_e_prefixa_summary():
|
|
from app.domain.contas.service import ContasDomainService
|
|
|
|
class Client: pass
|
|
result = ContasDomainService(client=Client()).finalizar_atendimento(
|
|
status="resolvido_outros_assuntos",
|
|
summary="VAS cancelado e cliente quer falar de plano.",
|
|
)
|
|
assert result["status"] == "resolvido_outros_assuntos"
|
|
assert result["summary"] == (
|
|
"Encerramento realizado pelo agente. VAS cancelado e cliente quer falar de plano."
|
|
)
|
|
|
|
|
|
def test_finalizacao_informacional_cria_protocolo_fechado_sem_duplicar_status():
|
|
from app.domain.contas.workflow_actions import build_contas_workflow_actions
|
|
|
|
class Client:
|
|
def __init__(self): self.protocol_calls = []; self.status_calls = []
|
|
def abrir_protocolo(self, payload):
|
|
self.protocol_calls.append(payload)
|
|
return {"protocolNumber": "INFO-123"}
|
|
def status_sr(self, payload):
|
|
self.status_calls.append(payload)
|
|
return {"ok": True}
|
|
|
|
from app.domain.contas.service import ContasDomainService
|
|
client = Client()
|
|
reg = build_contas_workflow_actions(ContasDomainService(client=client))
|
|
action = reg.get("finalizar_atendimento_action")
|
|
result = action({
|
|
"status": "resolvido",
|
|
"summary": "Explicação aceita.",
|
|
"msisdn": "11999999999",
|
|
"social_sec_no": "12345678901",
|
|
"informational_vas_types": ["bundle", "estrategico"],
|
|
}, {"input": {}})
|
|
assert result["protocol_number"] == "INFO-123"
|
|
assert result["informational_protocol_notes"] == "Explicação de VAS Bundle, VAS Estratégico"
|
|
assert client.protocol_calls[0]["requestStatus"] == "Fechado"
|
|
assert client.protocol_calls[0]["status"] == "CLOSED"
|
|
assert client.protocol_calls[0]["serviceRequestNotes"] == "Explicação de VAS Bundle, VAS Estratégico"
|
|
assert client.status_calls == []
|
|
|
|
|
|
def test_finalizacao_com_protocolo_existente_nao_cria_informacional():
|
|
from app.domain.contas.workflow_actions import build_contas_workflow_actions
|
|
from app.domain.contas.service import ContasDomainService
|
|
|
|
class Client:
|
|
def __init__(self): self.protocol_calls = []; self.status_calls = []
|
|
def abrir_protocolo(self, payload):
|
|
self.protocol_calls.append(payload)
|
|
return {"protocolNumber": "NOVO"}
|
|
def status_sr(self, payload):
|
|
self.status_calls.append(payload)
|
|
return {"ok": True}
|
|
|
|
client = Client()
|
|
action = build_contas_workflow_actions(ContasDomainService(client=client)).get("finalizar_atendimento_action")
|
|
result = action({
|
|
"status": "resolvido",
|
|
"summary": "ok",
|
|
"msisdn": "11999999999",
|
|
"protocol_number": "PRT-EXISTENTE",
|
|
"informational_vas_types": ["avulso"],
|
|
}, {"input": {}})
|
|
assert client.protocol_calls == []
|
|
assert client.status_calls[0]["protocolNumber"] == "PRT-EXISTENTE"
|
|
assert "informational_protocol_notes" not in result
|
|
|
|
|
|
import pytest
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("types", "expected"),
|
|
[
|
|
({"bundle"}, "Explicação de VAS Bundle"),
|
|
({"estrategico"}, "Explicação de VAS Estratégico"),
|
|
({"avulso"}, "Explicação de VAS Avulso"),
|
|
({"bundle", "estrategico"}, "Explicação de VAS Bundle, VAS Estratégico"),
|
|
({"bundle", "avulso"}, "Explicação de VAS Bundle, VAS Avulso"),
|
|
({"estrategico", "avulso"}, "Explicação de VAS Estratégico, VAS Avulso"),
|
|
({"bundle", "estrategico", "avulso"}, "Explicação de VAS Bundle, VAS Estratégico, VAS Avulso"),
|
|
],
|
|
)
|
|
def test_finalizacao_informacional_notas_canonicas(types, expected):
|
|
from app.domain.contas.workflow_actions import _format_informational_vas_notes
|
|
|
|
assert _format_informational_vas_notes(types) == expected
|
|
|
|
|
|
def test_finalizacao_cria_protocolo_informacional_quando_invoice_explanation_executou():
|
|
from app.domain.contas.workflow_actions import build_contas_workflow_actions
|
|
|
|
class Client:
|
|
def __init__(self):
|
|
self.protocol_calls = []
|
|
def abrir_protocolo(self, payload):
|
|
self.protocol_calls.append(payload)
|
|
return {"interactionProtocol": "INFO-123"}
|
|
def status_sr(self, payload):
|
|
raise AssertionError("protocolo informacional fechado nao deve ser fechado novamente")
|
|
class Service:
|
|
def __init__(self): self.client = Client()
|
|
def finalizar_atendimento(self, *, status, summary, protocol="", msisdn="", **kwargs):
|
|
return {"success": True, "status": status, "summary": summary}
|
|
|
|
service = Service()
|
|
reg = build_contas_workflow_actions(service) # type: ignore[arg-type]
|
|
action = reg.get("finalizar_atendimento_action")
|
|
result = action({
|
|
"status": "resolvido",
|
|
"summary": "Cliente entendeu a explicação.",
|
|
"msisdn": "11999999999",
|
|
"social_sec_no": "12345678901",
|
|
"message_id": "msg-123",
|
|
"invoice_explanation_base": "Explicação dos valores da fatura.",
|
|
"business_workflows_executed": ["invoice_explanation"],
|
|
}, {"input": {}})
|
|
assert result["protocol_number"] == "INFO-123"
|
|
assert service.client.protocol_calls[0]["serviceRequestNotes"] == "Explicação dos valores da fatura"
|
|
|
|
|
|
def test_finalizacao_nao_cria_protocolo_apenas_por_prefetch_invoice_explanation():
|
|
from app.domain.contas.workflow_actions import build_contas_workflow_actions
|
|
class Client:
|
|
def __init__(self): self.protocol_calls = []
|
|
def abrir_protocolo(self, payload): self.protocol_calls.append(payload); return {"interactionProtocol": "BAD"}
|
|
class Service:
|
|
def __init__(self): self.client = Client()
|
|
def finalizar_atendimento(self, *, status, summary, protocol="", msisdn="", **kwargs): return {"success": True}
|
|
service = Service()
|
|
action = build_contas_workflow_actions(service).get("finalizar_atendimento_action") # type: ignore[arg-type]
|
|
result = action({"status": "resolvido", "msisdn": "119", "invoice_explanation_base": "prefetch"}, {"input": {}})
|
|
assert "protocol_number" not in result
|
|
assert service.client.protocol_calls == []
|
|
|
|
|
|
def test_finalizacao_nao_duplica_protocolo_quando_contestacao_ja_existe():
|
|
from app.domain.contas.workflow_actions import build_contas_workflow_actions
|
|
class Client:
|
|
def __init__(self): self.protocol_calls = []; self.status_calls=[]
|
|
def abrir_protocolo(self, payload): self.protocol_calls.append(payload); return {"interactionProtocol": "BAD"}
|
|
def status_sr(self, payload): self.status_calls.append(payload); return {"ok": True}
|
|
class Service:
|
|
def __init__(self): self.client = Client()
|
|
def finalizar_atendimento(self, *, status, summary, protocol="", msisdn="", **kwargs):
|
|
if protocol: self.client.status_sr({"protocolNumber": protocol, "status": "Fechado"})
|
|
return {"success": True}
|
|
service = Service()
|
|
action = build_contas_workflow_actions(service).get("finalizar_atendimento_action") # type: ignore[arg-type]
|
|
result = action({
|
|
"status": "resolvido", "msisdn": "119", "informational_vas_types": ["avulso"],
|
|
"contestacao_protocol": "2026000083899",
|
|
}, {"input": {}})
|
|
assert "protocol_number" not in result
|
|
assert service.client.protocol_calls == []
|
|
|
|
|
|
def test_finalizacao_reusa_protocolo_veb_ja_fechado_sem_abrir_outro():
|
|
from app.domain.contas.workflow_actions import build_contas_workflow_actions
|
|
class Client:
|
|
def __init__(self): self.protocol_calls=[]; self.status_calls=[]
|
|
def abrir_protocolo(self, payload): self.protocol_calls.append(payload); return {"interactionProtocol":"BAD"}
|
|
def status_sr(self, payload): self.status_calls.append(payload); return {"ok":True}
|
|
class Service:
|
|
def __init__(self): self.client=Client()
|
|
def finalizar_atendimento(self, *, status, summary, protocol="", msisdn="", **kwargs):
|
|
if protocol: self.client.status_sr({"protocolNumber":protocol})
|
|
return {"success":True}
|
|
svc=Service(); action=build_contas_workflow_actions(svc).get("finalizar_atendimento_action") # type: ignore[arg-type]
|
|
result=action({
|
|
"status":"resolvido", "msisdn":"119", "protocol_closed":True,
|
|
"protocolos_por_linha":[{"msisdn":"119","protocolo_id":"VEB-123"}],
|
|
"informational_vas_types":["estrategico"],
|
|
}, {"input":{}})
|
|
assert result["protocol_number"] == "VEB-123"
|
|
assert svc.client.protocol_calls == []
|
|
assert svc.client.status_calls == []
|
|
|
|
|
|
def test_finalizacao_force_rt15_cria_novo_protocolo_mesmo_com_veb_fechado():
|
|
from app.domain.contas.workflow_actions import build_contas_workflow_actions
|
|
class Client:
|
|
def __init__(self): self.protocol_calls=[]
|
|
def abrir_protocolo(self,payload): self.protocol_calls.append(payload); return {"interactionProtocol":"INFO-RT15"}
|
|
class Service:
|
|
def __init__(self): self.client=Client()
|
|
def finalizar_atendimento(self, **kwargs): return {"success":True}
|
|
svc=Service(); action=build_contas_workflow_actions(svc).get("finalizar_atendimento_action") # type: ignore[arg-type]
|
|
result=action({
|
|
"status":"resolvido", "msisdn":"119", "protocol_closed":True,
|
|
"protocolos_por_linha":[{"protocolo_id":"VEB-123"}],
|
|
"informational_vas_types":["bundle"], "force_rt15_finalization_protocol":True,
|
|
}, {"input":{}})
|
|
assert result["protocol_number"] == "INFO-RT15"
|
|
assert len(svc.client.protocol_calls) == 1
|
|
assert svc.client.protocol_calls[0]["serviceRequestNotes"].endswith("VAS Bundle")
|
|
|
|
|
|
def test_finalizacao_veb_deferido_suprime_evento_cvn_de_protocolo():
|
|
from app.domain.contas.workflow_actions import build_contas_workflow_actions
|
|
class Client:
|
|
def abrir_protocolo(self,payload): return {"interactionProtocol":"VEB-NEW"}
|
|
class Service:
|
|
def __init__(self): self.client=Client()
|
|
def finalizar_atendimento(self, **kwargs): return {"success":True}
|
|
action=build_contas_workflow_actions(Service()).get("finalizar_atendimento_action") # type: ignore[arg-type]
|
|
result=action({
|
|
"status":"resolvido", "msisdn":"119", "informational_vas_types":["estrategico"],
|
|
"vas_estrategico_protocol_deferred_to_finalization":True, "suppress_cvn_protocol_ic":True,
|
|
}, {"input":{}})
|
|
codes=[e.get("code") if isinstance(e,dict) else str(e) for e in result["business_events"]]
|
|
assert result["protocol_number"] == "VEB-NEW"
|
|
assert "CVN.010" not in codes and "CVN.011" not in codes
|
|
|
|
|
|
def test_invoice_explanation_converte_retry_transport_em_rct_por_tentativa():
|
|
from app.domain.contas.workflow_actions import build_contas_workflow_actions
|
|
class Client: pass
|
|
class Service:
|
|
client = Client()
|
|
def invoice_explanation(self, *, msisdn, **params):
|
|
return {"billing_analysis": {
|
|
"invoiceExplanation": "Sua fatura mudou por um serviço adicional.",
|
|
"_transport": {"rct_operation": "base_conhecimento", "attempts": [
|
|
{"attempt": 1, "success": False, "status_code": 500, "latency_ms": 12},
|
|
{"attempt": 2, "success": True, "status_code": 200, "latency_ms": 7},
|
|
]},
|
|
}}
|
|
action=build_contas_workflow_actions(Service()).get("preparar_invoice_explanation") # type: ignore[arg-type]
|
|
result=action({"msisdn":"119"}, {"input":{"msisdn":"119"}})
|
|
codes=[e.get("code") for e in result["business_events"] if isinstance(e,dict)]
|
|
assert "RCT.080" in codes
|
|
assert "RCT.081" in codes
|
|
assert result["success"] is True
|
|
|
|
|
|
def test_invoice_explanation_falha_preserva_rct_das_tres_tentativas():
|
|
from app.domain.contas.client import TimApiError
|
|
from app.domain.contas.workflow_actions import build_contas_workflow_actions
|
|
class Client: pass
|
|
class Service:
|
|
client=Client()
|
|
def invoice_explanation(self, *, msisdn, **params):
|
|
raise TimApiError("gateway indisponivel", status_code=550, body={"message":"erro"}, attempts=[
|
|
{"attempt":1,"success":False,"status_code":500,"latency_ms":3},
|
|
{"attempt":2,"success":False,"status_code":502,"latency_ms":4},
|
|
{"attempt":3,"success":False,"status_code":550,"latency_ms":5},
|
|
])
|
|
action=build_contas_workflow_actions(Service()).get("preparar_invoice_explanation") # type: ignore[arg-type]
|
|
result=action({"msisdn":"119"}, {"input":{"msisdn":"119"}})
|
|
codes=[e.get("code") for e in result["business_events"] if isinstance(e,dict)]
|
|
assert [c for c in codes if str(c).startswith("RCT.")] == ["RCT.080","RCT.082","RCT.084"]
|
|
assert result["api_status_code"] == 550
|
|
assert result["service_failed"] is True
|
|
|
|
|
|
def _event_codes(value):
|
|
return [x.get("code") if isinstance(x, dict) else str(x) for x in (value.get("business_events") or [])]
|
|
|
|
|
|
def test_cancelamento_batch_emite_vaa_caminho_feliz_uma_vez():
|
|
from app.domain.contas.workflow_actions import build_contas_workflow_actions
|
|
class Client:
|
|
def consultar_vas(self, msisdn):
|
|
return {"services": [{"bill_description": "TIM Fashion Mensal", "app_id": "A1", "can": {"cancel": True}}]}
|
|
def line_info(self, msisdn): return {"socialSecNo": "123"}
|
|
def abrir_protocolo(self, payload): return {"protocol": "P-1"}
|
|
class Service:
|
|
client = Client()
|
|
def cancelar_vas_avulso(self, **kwargs):
|
|
return {"success": True, "msisdn": kwargs["msisdn"], "subject": kwargs["subject"], "protocol": kwargs.get("protocol", "")}
|
|
from agent_framework.idempotency import InMemoryIdempotencyStore
|
|
action = build_contas_workflow_actions(Service(), idempotency_store=InMemoryIdempotencyStore()).get("cancelamento_vas_avulso_batch") # type: ignore[arg-type]
|
|
import asyncio
|
|
result = asyncio.run(action({"msisdn": "5511999999999", "items": [{"msisdn": "5511999999999", "service": "TIM Fashion Mensal"}]}, {"input": {}}))
|
|
codes = _event_codes(result)
|
|
assert codes.count("VAA.001") == 1
|
|
assert codes.count("VAA.002") == 1
|
|
assert codes.count("VAA.003") == 1
|
|
assert "VAA.004" not in codes
|
|
vaa = [x for x in result["business_events"] if isinstance(x, dict) and str(x.get("code", "")).startswith("VAA.")]
|
|
assert all(x["payload"]["gsm"] == "5511999999999" for x in vaa)
|
|
|
|
|
|
def test_cancelamento_batch_emite_vaa004_quando_falha_operacional():
|
|
from app.domain.contas.workflow_actions import build_contas_workflow_actions
|
|
class Service:
|
|
class Client:
|
|
def line_info(self, msisdn): return {"socialSecNo": "123"}
|
|
def abrir_protocolo(self, payload): return {"protocol": "P-1"}
|
|
client = Client()
|
|
def cancelar_vas_avulso(self, **kwargs):
|
|
return {"success": False, "msisdn": kwargs["msisdn"], "subject": kwargs["subject"], "reason": "block_vas_failed", "error": "timeout"}
|
|
from agent_framework.idempotency import InMemoryIdempotencyStore
|
|
action = build_contas_workflow_actions(Service(), idempotency_store=InMemoryIdempotencyStore()).get("cancelamento_vas_avulso_batch") # type: ignore[arg-type]
|
|
import asyncio
|
|
result = asyncio.run(action({"msisdn": "5511999999999", "items": [{"msisdn": "5511999999999", "service": "TIM Fashion Mensal"}]}, {"input": {}}))
|
|
codes = _event_codes(result)
|
|
assert "VAA.001" in codes and "VAA.002" in codes and "VAA.004" in codes
|
|
assert "VAA.003" not in codes
|