81 lines
4.3 KiB
Python
81 lines
4.3 KiB
Python
from __future__ import annotations
|
|
import asyncio
|
|
from types import SimpleNamespace
|
|
from typing import Any
|
|
from contas_mcp.servers.contas_mcp_server import main as m
|
|
|
|
|
|
def R(status='COMPLETED', output=None, trace=None, error=None, error_details=None):
|
|
d={'execution_id':'x','status':status,'output':output or {},'trace':trace or [],'state':{},'error':error,'error_details':error_details}
|
|
return SimpleNamespace(**d, model_dump=lambda: dict(d))
|
|
|
|
class Runtime:
|
|
def __init__(self,*rs): self.rs=list(rs); self.calls=[]
|
|
async def arun(self,name,payload,execution_id=None):
|
|
self.calls.append((name,dict(payload),execution_id)); return self.rs.pop(0)
|
|
|
|
|
|
def test_invoice_explanation_preserva_fatura_e_analise_variacao_no_input(monkeypatch):
|
|
rt=Runtime(R(output={}))
|
|
monkeypatch.setattr(m,'get_workflow_runtime',lambda:rt)
|
|
args={'msisdn':'119','invoice_detail':{'119':{'SVA':[]}},'billing_analysis':{'currentInvoice':{'invoiceNumber':'I1'}},'vas_variation':{'added':['X']}}
|
|
asyncio.run(m._run_workflow('invoice_explanation',args))
|
|
p=rt.calls[0][1]
|
|
assert p['invoice_detail']==args['invoice_detail']
|
|
assert p['billing_analysis']==args['billing_analysis']
|
|
assert p['vas_variation']==args['vas_variation']
|
|
|
|
|
|
def test_pro_rata_devolucao_forca_composicao_llm_framework():
|
|
from app.domain.contas.workflow_actions import build_contas_workflow_actions
|
|
class S: pass
|
|
reg=build_contas_workflow_actions(S())
|
|
a=reg.get('orientar_pagamento_pro_rata')
|
|
out=a({'refund_option':'proxima_fatura','valor_ajuste':'10,00'}, {'input':{}})
|
|
assert out.get('requires_llm_composition') is True
|
|
assert str(out.get('response_instruction') or '').strip()
|
|
|
|
|
|
def test_cancelamento_boleto_zerado_ou_nao_nao_vaza_sms_sent(monkeypatch):
|
|
# Independentemente do barcode, o contrato externo não expõe o detalhe interno sms_sent.
|
|
for barcode in ('0'*44,'34191790010104351004791020150008291070026000'):
|
|
cancel=R(output={'cancelar_vas_avulso':{'success':True,'cancelados':[{'servico':'VAS','valor':'9,90','codigo_boleto':barcode,'msisdn':'119'}], 'contestation_candidates':[], 'protocolos_por_linha':[{'protocolo_id':'P1'}]}})
|
|
rt=Runtime(cancel); monkeypatch.setattr(m,'get_workflow_runtime',lambda rt=rt:rt)
|
|
out=asyncio.run(m._run_cancelamento_com_contestacao({'msisdn':'119','subject':'VAS','valor':'9,90'}))
|
|
assert 'sms_sent' not in out
|
|
assert 'prompt_signals' not in out
|
|
|
|
|
|
def test_cancelamento_corrige_linha_item_por_invoice_detail():
|
|
args={'items':[{'msisdn':'111','name':'Filme 2'},{'msisdn':'111','name':'Filme 3'}], 'invoice_detail':{'111':{'SVA':[{'desc':'Filme 2','msisdn':'111'}]},'222':{'SVA':[{'desc':'Filme 3','msisdn':'222'}]}}}
|
|
got=m._normalize_cancel_items(args,'111')
|
|
assert [x['msisdn'] for x in got]==['111','222']
|
|
|
|
|
|
def test_cancelamento_compositor_e_deterministico_sem_diretiva_llm():
|
|
from app.domain.contas.vas_cancellation_message import compose_vas_cancellation_message
|
|
payload={'cancelados':[{'servico':'TIM Fashion Mensal','valor':'14,99'}], 'contested_items':['TIM Fashion Mensal'], 'contested_invoice_amount_open':'14,99', 'contestacao_protocol':'P100'}
|
|
text=compose_vas_cancellation_message(payload)
|
|
assert 'TIM Fashion Mensal' in text and '14,99' in text and 'P100' in text
|
|
|
|
|
|
def test_vas_estrategico_normaliza_nome_e_agrupa_linhas():
|
|
from app.domain.contas.workflow_actions import build_contas_workflow_actions
|
|
class S: pass
|
|
a=build_contas_workflow_actions(S()).get('preparar_vas_estrategico')
|
|
p={'items':[{'msisdn':'111','name':'Paramount plus','type':'estrategico'},{'msisdn':'222','name':'YouTube Premium','type':'estrategico'}]}
|
|
out=a(p,{'input':p})
|
|
assert len(out['linhas'])==2
|
|
names=[n for l in out['linhas'] for n in l.get('estrategico_names',[])]
|
|
assert any('Paramount' in n for n in names)
|
|
|
|
|
|
def test_rag_dominio_apenas_declara_queries_e_framework_executa():
|
|
from app.domain.contas.workflow_actions import build_contas_workflow_actions
|
|
class S: pass
|
|
a=build_contas_workflow_actions(S()).get('buscar_informacao_rag')
|
|
out=a({'query':'como funciona pro rata','queries':['como funciona pro rata'],'top_k':4,'segment':'POS'}, {'input':{}})
|
|
assert out['delegate_to_framework_rag'] is True
|
|
assert out['queries']==['como funciona pro rata']
|
|
assert out['documents']==[]
|