1437 lines
53 KiB
Python
1437 lines
53 KiB
Python
"""Testes de ``vas_variation``: quais VAS avulsos variaram entre as duas faturas.
|
||
|
||
É a base do gatilho e do escopo da retenção de VAS no pedido de humano (SPEC §9),
|
||
então um erro aqui cancela o serviço errado do cliente. Cobre as duas formas de
|
||
payload (ISO-8601 + ``value`` string, como o runtime entrega; e ``dd/mm/yyyy`` +
|
||
float, da serialização alternativa), a partição por vencimento, o desempate de data
|
||
única, a exclusão do bucket ``streaming``, a contagem por cobrança (multiset) e todas
|
||
as portas de falha fechada."""
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
from decimal import Decimal
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
import yaml
|
||
|
||
from app.domain.contas.invoice_resolver import InvoiceResolver
|
||
from app.domain.contas.vas_variation import (
|
||
VariedCharge,
|
||
charge_match_key,
|
||
varied_avulso_items,
|
||
varied_current_charges,
|
||
varied_vas_charges,
|
||
variation_source,
|
||
)
|
||
|
||
PAST = "2026-03-10T00:00:00.000Z"
|
||
CURRENT = "2026-04-07T00:00:00.000Z"
|
||
|
||
_REPO_ROOT = Path(__file__).resolve().parents[2]
|
||
|
||
|
||
# ----- factories ------------------------------------------------------------
|
||
|
||
|
||
def _item(desc: str, value, invoice: str, **extra) -> dict:
|
||
return {"desc": desc, "value": value, "invoice": invoice, **extra}
|
||
|
||
|
||
def _group(*items: dict, type_: str = "servicos_contratados_de_parceiros") -> dict:
|
||
return {
|
||
"type": type_,
|
||
"desc": "Serviços de valor adicionado",
|
||
"nItems": str(len(items)),
|
||
"items": list(items),
|
||
}
|
||
|
||
|
||
def _payload(*groups: dict, current: list | None = None) -> dict:
|
||
payload: dict = {"invoiceVariation": list(groups)}
|
||
if current is not None:
|
||
payload["currentInvoice"] = current
|
||
return payload
|
||
|
||
|
||
def _current_group(*descs_values) -> list:
|
||
return [
|
||
{
|
||
"type": "servicos_contratados_de_parceiros",
|
||
"desc": "Serviços de valor adicionado",
|
||
"items": [
|
||
{"desc": d, "value": v, "contestable": True} for d, v in descs_values
|
||
],
|
||
}
|
||
]
|
||
|
||
|
||
def _names(charges: tuple[VariedCharge, ...]) -> list[str]:
|
||
return sorted(c.desc for c in charges)
|
||
|
||
|
||
# ----- caminho principal ----------------------------------------------------
|
||
|
||
|
||
def test_entrou_na_atual_e_detectado():
|
||
"""Serviço presente só na fatura atual → variou (entrou)."""
|
||
payload = _payload(
|
||
_group(
|
||
_item("Aluguel de Filme 1", "19.9", PAST),
|
||
_item("Aluguel de Filme 1", "19.9", CURRENT),
|
||
_item("Aluguel de Filme 3", "19.9", CURRENT),
|
||
)
|
||
)
|
||
assert varied_current_charges(payload) == (
|
||
VariedCharge(desc="Aluguel de Filme 3", value=Decimal("19.90")),
|
||
)
|
||
|
||
|
||
def test_inalterados_nao_entram():
|
||
"""Itens idênticos nas duas faturas não variaram — e são a MAIORIA do grupo
|
||
(o payload lista as duas faturas inteiras, não só o diff)."""
|
||
payload = _payload(
|
||
_group(
|
||
_item("Carnal Mensal", "16.99", PAST),
|
||
_item("MasterChef Mensal", "19.99", PAST),
|
||
_item("Carnal Mensal", "16.99", CURRENT),
|
||
_item("MasterChef Mensal", "19.99", CURRENT),
|
||
)
|
||
)
|
||
assert varied_current_charges(payload) == ()
|
||
|
||
|
||
def test_mudanca_de_valor_conta_o_valor_novo():
|
||
"""Mesmo serviço com valor diferente: entra a cobrança da fatura ATUAL."""
|
||
payload = _payload(
|
||
_group(
|
||
_item("Games Station Mensal", "14.9", PAST),
|
||
_item("Games Station Mensal", "16.99", CURRENT),
|
||
)
|
||
)
|
||
assert varied_current_charges(payload) == (
|
||
VariedCharge(desc="Games Station Mensal", value=Decimal("16.99")),
|
||
)
|
||
|
||
|
||
def test_removido_nao_oferta():
|
||
"""Variação causada só por REMOÇÃO: o avulso saiu e a conta caiu. Não há o que
|
||
cancelar → nada a ofertar (decisão de produto, SPEC §9)."""
|
||
payload = _payload(
|
||
_group(
|
||
_item("Monicaverso Semanal", "5.99", PAST),
|
||
_item("Carnal Mensal", "16.99", PAST),
|
||
_item("Carnal Mensal", "16.99", CURRENT),
|
||
)
|
||
)
|
||
assert varied_current_charges(payload) == ()
|
||
|
||
|
||
def test_cobranca_duplicada_conta_as_duas():
|
||
"""O mesmo serviço cobrado 2× no ciclo são DUAS cobranças reais: a segunda
|
||
entra como variação (multiset, não conjunto)."""
|
||
payload = _payload(
|
||
_group(
|
||
_item("Qualifica Mensal", "19.99", PAST),
|
||
_item("Qualifica Mensal", "19.99", CURRENT),
|
||
_item("Qualifica Mensal", "19.99", CURRENT),
|
||
)
|
||
)
|
||
assert varied_current_charges(payload) == (
|
||
VariedCharge(desc="Qualifica Mensal", value=Decimal("19.99")),
|
||
)
|
||
|
||
|
||
def test_duas_cobrancas_identicas_novas_contam_as_duas():
|
||
"""Duas cobranças IDÊNTICAS (mesmo nome E mesmo valor) que entraram na fatura
|
||
atual devolvem DUAS ``VariedCharge`` — um ``set`` colapsaria em uma e metade da
|
||
variação ficaria sem cancelar."""
|
||
payload = _payload(
|
||
_group(
|
||
_item("Aluguel de Filme 1", "19.9", CURRENT),
|
||
_item("Aluguel de Filme 1", "19.9", CURRENT),
|
||
),
|
||
current=_current_group(
|
||
("Aluguel de Filme 1", "19.9"), ("Aluguel de Filme 1", "19.9")
|
||
),
|
||
)
|
||
assert varied_current_charges(payload) == (
|
||
VariedCharge(desc="Aluguel de Filme 1", value=Decimal("19.90")),
|
||
VariedCharge(desc="Aluguel de Filme 1", value=Decimal("19.90")),
|
||
)
|
||
|
||
|
||
def test_duas_identicas_com_uma_na_passada_conta_apenas_uma():
|
||
"""Três cobranças iguais no total (1 na passada, 2 na atual): só UMA entrou. O
|
||
multiset cancela o par e sobra uma — cancelar as duas cobraria do cliente um
|
||
cancelamento que ele não pediu."""
|
||
payload = _payload(
|
||
_group(
|
||
_item("Aluguel de Filme 1", "19.9", PAST),
|
||
_item("Aluguel de Filme 1", "19.9", CURRENT),
|
||
_item("Aluguel de Filme 1", "19.9", CURRENT),
|
||
)
|
||
)
|
||
assert varied_current_charges(payload) == (
|
||
VariedCharge(desc="Aluguel de Filme 1", value=Decimal("19.90")),
|
||
)
|
||
|
||
|
||
# ----- classes que NÃO disparam ---------------------------------------------
|
||
|
||
|
||
def test_streaming_e_ignorado():
|
||
"""O bucket ``streaming`` é o VAS ESTRATÉGICO e nunca dispara retenção."""
|
||
payload = _payload(
|
||
_group(
|
||
_item("Netflix Padrão", "20.9", CURRENT),
|
||
_item("Disney+ Premium", "62.9", CURRENT),
|
||
type_="streaming",
|
||
)
|
||
)
|
||
assert varied_current_charges(payload) == ()
|
||
|
||
|
||
def test_grupo_avulso_ausente():
|
||
"""Só grupos de outras categorias (plano, juros) → nada de avulso."""
|
||
payload = _payload(
|
||
_group(_item("TIM Black Premium 2 0", "239.99", CURRENT), type_="plano")
|
||
)
|
||
assert varied_current_charges(payload) == ()
|
||
|
||
|
||
# ----- desempate de data única ---------------------------------------------
|
||
|
||
|
||
def test_data_unica_casando_a_atual_conta_como_entrada():
|
||
"""Uma só data no grupo (a outra fatura não tinha avulso): se as cobranças
|
||
aparecem em ``currentInvoice``, o bloco é o lado ATUAL → tudo entrou."""
|
||
payload = _payload(
|
||
_group(
|
||
_item("Gamedom Mensal", "14.99", CURRENT),
|
||
_item("Carnal Mensal", "16.99", CURRENT),
|
||
),
|
||
current=_current_group(("Gamedom Mensal", "14.99"), ("Carnal Mensal", "16.99")),
|
||
)
|
||
assert _names(varied_current_charges(payload)) == [
|
||
"Carnal Mensal",
|
||
"Gamedom Mensal",
|
||
]
|
||
|
||
|
||
def test_data_unica_sem_casar_a_atual_conta_como_saida():
|
||
"""Uma só data e as cobranças NÃO estão na fatura atual: o bloco é o lado
|
||
PASSADO (tudo saiu) → nada a cancelar."""
|
||
payload = _payload(
|
||
_group(
|
||
_item("Gamedom Mensal", "14.99", PAST),
|
||
_item("Carnal Mensal", "16.99", PAST),
|
||
),
|
||
current=_current_group(("Netflix Padrão", "20.9")),
|
||
)
|
||
assert varied_current_charges(payload) == ()
|
||
|
||
|
||
def test_data_unica_sem_current_invoice_nao_oferta():
|
||
"""Sem ``currentInvoice`` não há como desempatar a direção → falha fechada."""
|
||
payload = _payload(_group(_item("Gamedom Mensal", "14.99", CURRENT)))
|
||
assert varied_current_charges(payload) == ()
|
||
|
||
|
||
def test_data_unica_casa_item_em_outro_bucket_da_fatura_atual():
|
||
"""Os dois campos vêm de geradores distintos e DISCORDAM da classe do mesmo
|
||
item (caso real: avulso na variação, ``streaming`` na fatura atual). O
|
||
desempate pergunta "está na fatura atual?", não "de que classe é" — a classe é
|
||
do resolver. Restringir ao bucket avulso inverteria a direção e perderia a
|
||
cobrança."""
|
||
payload = _payload(
|
||
_group(_item("Truecaller Family Mensal", "13.99", CURRENT)),
|
||
current=[
|
||
{
|
||
"type": "streaming",
|
||
"desc": "Streamings",
|
||
"items": [{"desc": "Truecaller Family Mensal", "value": "13.99"}],
|
||
}
|
||
],
|
||
)
|
||
assert varied_current_charges(payload) == (
|
||
VariedCharge(desc="Truecaller Family Mensal", value=Decimal("13.99")),
|
||
)
|
||
|
||
|
||
# ----- formas de payload ----------------------------------------------------
|
||
|
||
|
||
def test_forma_alternativa_ddmmyyyy_e_float():
|
||
"""Serialização alternativa do mesmo pipeline: ``dd/mm/yyyy`` e ``value``
|
||
float. Ordena por data real, não lexicograficamente."""
|
||
payload = _payload(
|
||
_group(
|
||
_item("Fluid Mensal", 14.99, "10/07/2026"),
|
||
_item("Fluid Mensal", 14.99, "10/08/2026"),
|
||
_item("Neymar Jr.Experience-", 14.99, "10/08/2026"),
|
||
)
|
||
)
|
||
assert varied_current_charges(payload) == (
|
||
VariedCharge(desc="Neymar Jr.Experience-", value=Decimal("14.99")),
|
||
)
|
||
|
||
|
||
def test_ordenacao_por_data_nao_lexicografica():
|
||
"""``dd/mm/yyyy``: 09/12/2025 é ANTERIOR a 09/01/2026 — ordenar como string
|
||
inverteria a direção e cancelaria o item errado."""
|
||
payload = _payload(
|
||
_group(
|
||
_item("Apps Club Mensal", 12.99, "09/12/2025"),
|
||
_item("Apps Club Mensal", 12.99, "09/01/2026"),
|
||
_item("Pocoyo Mensal", 19.99, "09/01/2026"),
|
||
)
|
||
)
|
||
assert varied_current_charges(payload) == (
|
||
VariedCharge(desc="Pocoyo Mensal", value=Decimal("19.99")),
|
||
)
|
||
|
||
|
||
def test_bucket_por_msisdn():
|
||
"""``invoiceVariation`` dentro de um bucket por linha (fatura família) é lido
|
||
igual ao do top-level."""
|
||
payload = {
|
||
"Fatura Resumo": {"total": "100"},
|
||
"2196460-1930": {
|
||
"invoiceVariation": [
|
||
_group(
|
||
_item("Fit Me App", "16.99", PAST),
|
||
_item("Fit Me App", "16.99", CURRENT),
|
||
_item("Budz Premium mensal", "14.99", CURRENT),
|
||
)
|
||
]
|
||
},
|
||
}
|
||
assert varied_current_charges(payload) == (
|
||
VariedCharge(desc="Budz Premium mensal", value=Decimal("14.99")),
|
||
)
|
||
|
||
|
||
def test_desc_devolvido_e_o_cru_da_fatura():
|
||
"""O ``desc`` devolvido é o texto CRU (com acento/pontuação), nunca o
|
||
normalizado usado para casar."""
|
||
payload = _payload(
|
||
_group(_item("VOD +Canais Abertos", "19.9", CURRENT)),
|
||
current=_current_group(("VOD +Canais Abertos", "19.9")),
|
||
)
|
||
assert varied_current_charges(payload)[0].desc == "VOD +Canais Abertos"
|
||
|
||
|
||
# ----- falha fechada --------------------------------------------------------
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"payload",
|
||
[
|
||
pytest.param(None, id="none"),
|
||
pytest.param({}, id="vazio"),
|
||
pytest.param({"invoiceVariation": []}, id="lista-vazia"),
|
||
pytest.param({"invoiceVariation": "nao-e-lista"}, id="tipo-errado"),
|
||
pytest.param({"invoiceVariation": [None, 42]}, id="grupos-invalidos"),
|
||
pytest.param(
|
||
{"invoiceVariation": [{"type": "servicos_contratados_de_parceiros"}]},
|
||
id="grupo-sem-items",
|
||
),
|
||
],
|
||
)
|
||
def test_payload_invalido_devolve_vazio(payload):
|
||
"""Sem dado utilizável não oferta retenção — nunca promete cancelar o que não
|
||
se sabe ter variado."""
|
||
assert varied_current_charges(payload) == ()
|
||
|
||
|
||
def test_vencimento_ilegivel_aborta():
|
||
"""Item que não pode ser atribuído a nenhuma das duas faturas corromperia o
|
||
diff — e um diff errado cancela o serviço errado. Aborta tudo."""
|
||
payload = _payload(
|
||
_group(
|
||
_item("Carnal Mensal", "16.99", PAST),
|
||
_item("Gamedom Mensal", "14.99", "sem-data"),
|
||
_item("Gamedom Mensal", "14.99", CURRENT),
|
||
)
|
||
)
|
||
assert varied_current_charges(payload) == ()
|
||
|
||
|
||
def test_item_sem_valor_nao_entra():
|
||
"""Cobrança sem identidade (valor não parseável) fica fora do diff."""
|
||
payload = _payload(
|
||
_group(
|
||
_item("Carnal Mensal", "16.99", PAST),
|
||
_item("Carnal Mensal", "16.99", CURRENT),
|
||
_item("Servico Sem Valor", "n/a", CURRENT),
|
||
)
|
||
)
|
||
assert varied_current_charges(payload) == ()
|
||
|
||
|
||
# ----- charge_match_key -----------------------------------------------------
|
||
|
||
|
||
def test_charge_match_key_normaliza_nome_e_valor():
|
||
"""A chave é a MESMA dos dois lados do casamento (cobrança variada × item do
|
||
resolver): acento/caixa/``+`` normalizados, valor com 2 casas."""
|
||
assert charge_match_key("VOD + Canais Abertos", "19.9") == charge_match_key(
|
||
"vod +canais abertos", Decimal("19.90")
|
||
)
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"desc,value",
|
||
[("", "19.9"), (None, "19.9"), ("Carnal Mensal", None), ("Carnal Mensal", "n/a")],
|
||
)
|
||
def test_charge_match_key_sem_identidade(desc, value):
|
||
assert charge_match_key(desc, value) is None
|
||
|
||
|
||
# ----- varied_avulso_items: cruzamento com a classificação da fatura ---------
|
||
#
|
||
# ``varied_current_charges`` (acima) sabe o que VARIOU; só o PDF parseado sabe o que é
|
||
# CANCELÁVEL. ``varied_avulso_items`` cruza os dois, e é a função que os DOIS
|
||
# consumidores chamam — a retenção no pedido de humano (SPEC §9) e o ramo NÃO do
|
||
# ``invoice_explanation``. Os dois guards contra estratégico mis-bucketado pela API
|
||
# vivem aqui, e antes só eram exercitados de through-the-runtime.
|
||
|
||
|
||
def _detail(sections: dict) -> dict:
|
||
"""``invoice_detail`` de uma linha, no formato que o resolver espera."""
|
||
return {"11999999999": sections}
|
||
|
||
|
||
def _entrou(*descs_values) -> dict:
|
||
"""Análise em que TODAS as cobranças entraram na fatura atual: um único
|
||
vencimento no bloco de variação, desempatado por ``currentInvoice`` (a forma que o
|
||
runtime entrega quando a fatura passada não tinha avulso)."""
|
||
return _payload(
|
||
_group(*(_item(d, v, CURRENT) for d, v in descs_values)),
|
||
current=_current_group(*descs_values),
|
||
)
|
||
|
||
|
||
def _varied(invoice_detail: dict, variation: dict | None = None, *, by_charge=False):
|
||
return varied_avulso_items(
|
||
invoice_variation=variation,
|
||
invoice_detail=invoice_detail,
|
||
resolver=InvoiceResolver(),
|
||
by_charge=by_charge,
|
||
)
|
||
|
||
|
||
def test_avulso_variado_casa_nome_e_valor():
|
||
"""O caminho feliz: o serviço variou na análise E está na fatura como avulso."""
|
||
detail = _detail(
|
||
{"SVA Detalhe Total": [{"desc": "TIM Fashion Mensal", "value": 10.0}]}
|
||
)
|
||
|
||
items = _varied(detail, _entrou(("TIM Fashion Mensal", "10.0")))
|
||
|
||
assert [it.canonical_name for it in items] == ["TIM Fashion Mensal"]
|
||
|
||
|
||
def test_guard_nome_estrategico_no_bucket_avulso_da_api():
|
||
"""Guard 1: a API bucketa streaming como avulso (visto em fatura real com
|
||
``YouTube Premium Mensal``) e a seção classifica por ``contestable``, então o item
|
||
sairia ``avulso``. A lista fixa §12.A vence — estratégico não é cancelável."""
|
||
detail = _detail(
|
||
{"SVA Detalhe Total": [{"desc": "YouTube Premium Mensal", "value": 24.9}]}
|
||
)
|
||
|
||
assert _varied(detail, _entrou(("YouTube Premium Mensal", "24.9"))) == []
|
||
|
||
|
||
def test_guard_classe_mista_na_mesma_fatura():
|
||
"""Guard 2: o MESMO nome aparece como avulso numa seção e como bundle noutra —
|
||
discordância entre os dois geradores da API. A não-cancelável vence."""
|
||
detail = _detail(
|
||
{
|
||
"SVA Detalhe Total": [{"desc": "Tamboro Mensal", "value": 14.99}],
|
||
"Serviços Bundle Inclusos": [{"desc": "Tamboro Mensal", "value": 14.99}],
|
||
}
|
||
)
|
||
|
||
assert _varied(detail, _entrou(("Tamboro Mensal", "14.99"))) == []
|
||
|
||
|
||
def test_valor_divergente_entre_as_fontes_nao_casa(caplog):
|
||
"""Casar só por nome pegaria a cobrança ERRADA do mesmo serviço (2 cobranças, só
|
||
uma variou). Fora da lista — e logado, para a taxa não sumir."""
|
||
detail = _detail(
|
||
{"SVA Detalhe Total": [{"desc": "Tamboro Mensal", "value": 14.99}]}
|
||
)
|
||
|
||
with caplog.at_level(logging.INFO):
|
||
assert _varied(detail, _entrou(("Tamboro Mensal", "19.99"))) == []
|
||
|
||
assert "value_mismatch" in caplog.text
|
||
|
||
|
||
def test_variado_ausente_da_fatura_e_logado(caplog):
|
||
"""A análise diz que variou, mas o serviço não está na fatura atual."""
|
||
detail = _detail(
|
||
{"SVA Detalhe Total": [{"desc": "Tamboro Mensal", "value": 14.99}]}
|
||
)
|
||
|
||
with caplog.at_level(logging.INFO):
|
||
assert _varied(detail, _entrou(("Serviço Fantasma", "5.0"))) == []
|
||
|
||
assert "name_absent" in caplog.text
|
||
|
||
|
||
def test_sem_variacao_devolve_lista_vazia():
|
||
"""Distinto de ``None``: sabemos que nada variou."""
|
||
detail = _detail(
|
||
{"SVA Detalhe Total": [{"desc": "Tamboro Mensal", "value": 14.99}]}
|
||
)
|
||
|
||
assert _varied(detail, _payload(_group())) == []
|
||
|
||
|
||
def test_sem_fatura_devolve_none():
|
||
"""``None`` = não deu para decidir. Cada consumidor escolhe o que fazer com a
|
||
dúvida (a retenção não oferta; o ramo NÃO devolve ao orquestrador)."""
|
||
assert _varied({}, _entrou(("X", "1.0"))) is None
|
||
|
||
|
||
def test_variation_source_prefere_a_analise():
|
||
payload = _payload(_group(_item("X", "1.0", CURRENT)))
|
||
assert variation_source(payload, {"outro": 1}) is payload
|
||
|
||
|
||
def test_variation_source_cai_no_invoice_detail():
|
||
"""Entradas em que o próprio ``invoice_detail`` já é o payload de análise."""
|
||
detail = _payload(_group(_item("X", "1.0", CURRENT)))
|
||
assert variation_source({}, detail) is detail
|
||
assert variation_source(None, detail) is detail
|
||
|
||
|
||
# ----- varied_vas_charges: recorte AVULSO + ESTRATÉGICO ----------------------
|
||
#
|
||
# O ramo NÃO do ``invoice_explanation`` não pergunta "o que eu cancelo?" (aí só avulso
|
||
# serve, e é o recorte da retenção acima) — pergunta "existe assunto que eu trate?". O
|
||
# estratégico o agente trata pela tool ``vas_estrategico``, então ele conta.
|
||
#
|
||
# E, ao contrário da retenção, esta função NÃO cruza com o PDF: o ``type`` do grupo já é
|
||
# a classe. Os testes abaixo provam as duas metades disso — o grupo ``streaming`` entra,
|
||
# e a fatura não é consultada em nenhum caminho.
|
||
|
||
|
||
def _entrou_vas(*, avulsos=(), estrategicos=()) -> dict:
|
||
"""Análise em que TODAS as cobranças entraram na fatura atual, nos dois grupos de
|
||
VAS (avulso e ``streaming``) — a forma que o runtime entrega quando a fatura passada
|
||
não tinha VAS."""
|
||
groups = []
|
||
if avulsos:
|
||
groups.append(_group(*(_item(d, v, CURRENT) for d, v in avulsos)))
|
||
if estrategicos:
|
||
groups.append(
|
||
_group(
|
||
*(_item(d, v, CURRENT) for d, v in estrategicos), type_="streaming"
|
||
)
|
||
)
|
||
return _payload(*groups, current=_current_group(*avulsos, *estrategicos))
|
||
|
||
|
||
def _varied_vas(variation: dict | None, invoice_detail: dict | None = None):
|
||
return varied_vas_charges(
|
||
invoice_variation=variation, invoice_detail=invoice_detail
|
||
)
|
||
|
||
|
||
def test_vas_variado_pega_o_estrategico_que_variou():
|
||
"""Variação causada só por streaming: o recorte da retenção não vê nada (não é
|
||
cancelável), o do ramo NÃO vê — e é a diferença que faz o atendimento voltar ao
|
||
orquestrador em vez de finalizar ``nao_resolvido``."""
|
||
variation = _entrou_vas(estrategicos=[("Max Mensal", "34.9")])
|
||
detail = _detail({"Streamings": [{"desc": "Max Mensal", "value": 34.9}]})
|
||
|
||
assert _names(_varied_vas(variation)) == ["Max Mensal"]
|
||
assert _varied(detail, variation) == []
|
||
|
||
|
||
def test_vas_variado_pega_o_avulso_tambem():
|
||
"""O recorte é mais largo, não outro: o avulso continua contando."""
|
||
charges = _varied_vas(_entrou_vas(avulsos=[("TIM Fashion Mensal", "10.0")]))
|
||
|
||
assert _names(charges) == ["TIM Fashion Mensal"]
|
||
|
||
|
||
def test_vas_variado_soma_os_dois_grupos():
|
||
variation = _entrou_vas(
|
||
avulsos=[("Tamboro Mensal", "14.99")], estrategicos=[("Max Mensal", "34.9")]
|
||
)
|
||
|
||
assert _names(_varied_vas(variation)) == ["Max Mensal", "Tamboro Mensal"]
|
||
|
||
|
||
def test_vas_variado_particiona_pela_uniao_dos_grupos():
|
||
"""A partição passada × atual é feita sobre a UNIÃO dos dois grupos. Aqui só o
|
||
avulso tem as duas faturas e o streaming existe só na atual: lido sozinho, o
|
||
streaming cairia no desempate de data única (que exige ``currentInvoice``); na
|
||
união, os dois vencimentos do avulso dão a direção."""
|
||
variation = _payload(
|
||
_group(
|
||
_item("Tamboro Mensal", "19.99", PAST),
|
||
_item("Tamboro Mensal", "19.99", CURRENT),
|
||
),
|
||
_group(_item("Max Mensal", "34.9", CURRENT), type_="streaming"),
|
||
)
|
||
|
||
assert _names(_varied_vas(variation)) == ["Max Mensal"]
|
||
|
||
|
||
def test_vas_variado_nao_consulta_a_fatura():
|
||
"""O contrato que separa este recorte do da retenção: a decisão sai só da análise.
|
||
|
||
O mesmo payload de variação devolve o MESMO resultado com fatura ausente, com
|
||
fatura que classifica o item como ``bundle`` e com fatura ilegível — três entradas
|
||
que, no cruzamento antigo, davam três respostas diferentes e ENCERRAVAM a ligação."""
|
||
variation = _entrou_vas(avulsos=[("Tamboro Mensal", "14.99")])
|
||
faturas = [
|
||
None,
|
||
{},
|
||
_detail({"Serviços Bundle Inclusos": [{"desc": "Tamboro Mensal", "value": 14.99}]}),
|
||
{"vocalized_msisdn": {}}, # PDF não processado (incidente Neymar Jr.Experience)
|
||
_detail({"SVA Detalhe Total": [{"desc": "Tamboro Mensal Premium", "value": 17.00}]}),
|
||
]
|
||
|
||
assert all(_names(_varied_vas(variation, f)) == ["Tamboro Mensal"] for f in faturas)
|
||
|
||
|
||
def test_vas_variado_sem_variacao_devolve_vazio():
|
||
"""Nada variou nos dois grupos → finaliza ``nao_resolvido``."""
|
||
assert _varied_vas(_payload(_group())) == ()
|
||
|
||
|
||
def test_vas_variado_sem_analise_devolve_vazio():
|
||
"""Análise ausente é indistinguível de "nada variou" — falha fechada de
|
||
``varied_current_charges``, herdada por este recorte (decisão de 2026-08-10)."""
|
||
assert _varied_vas({"invoiceVariation": []}) == ()
|
||
assert _varied_vas(None) == ()
|
||
|
||
|
||
def test_vas_variado_ignora_grupo_que_nao_e_vas():
|
||
"""Plano, juros e roaming vivem em outros ``type`` e não entram no recorte."""
|
||
variation = _payload(
|
||
_group(_item("Torpedos Internacionais", "0.79", CURRENT), type_="pacote_do_tim_viagem_roaming"),
|
||
_group(_item("Multas", "2.59", CURRENT), type_="juros_e_multa_por_atraso_no_pagamento"),
|
||
current=_current_group(("Torpedos Internacionais", "0.79"), ("Multas", "2.59")),
|
||
)
|
||
|
||
assert _varied_vas(variation) == ()
|
||
|
||
|
||
def test_vas_variado_le_a_analise_do_invoice_detail_como_fallback():
|
||
"""``variation_source``: entradas em que o ``invoice_detail`` JÁ é o payload de
|
||
análise. É o único uso que sobrou do parâmetro."""
|
||
analise = _entrou_vas(avulsos=[("Tamboro Mensal", "14.99")])
|
||
|
||
assert _names(_varied_vas(None, analise)) == ["Tamboro Mensal"]
|
||
|
||
|
||
def test_grupo_streaming_fora_do_recorte_avulso():
|
||
"""A retenção lê SÓ o grupo avulso: streaming na variação não vira cobrança
|
||
variada (é o default de ``varied_current_charges``)."""
|
||
payload = _payload(
|
||
_group(_item("Max Mensal", "34.9", CURRENT), type_="streaming"),
|
||
current=_current_group(("Max Mensal", "34.9")),
|
||
)
|
||
|
||
assert varied_current_charges(payload) == ()
|
||
|
||
|
||
# ----- regressão contra dados reais -----------------------------------------
|
||
#
|
||
# A regra foi validada durante o desenvolvimento contra 88 pares de fatura reais
|
||
# (88/88, usando as faturas normalizadas como verdade). Esse corpus NÃO virou fixture:
|
||
# medido por mutação, ele pegava 6 de 8 defeitos injetados e nenhum que os testes acima
|
||
# já não peguem — era cego justamente ao multiset e ao parse de data. Somado a não ser
|
||
# regenerável (as CSVs de origem vivem fora do repo), virava asserção opaca. Os dois
|
||
# testes abaixo cobrem o que importa em dado real: a FORMA do payload de runtime.
|
||
#
|
||
# O dado vem COM o teste, não de mock do runtime: mock de gateway é editado à mão em
|
||
# teste manual, e um teste que o lê quebra por edição legítima. O cenário cy0001 é a
|
||
# exceção consciente — ele é a captura de um atendimento real e tem guarda de skip.
|
||
|
||
|
||
# Payload copiado do mock do gateway fake (`billingAnalysis` de `divergencia.json`),
|
||
# mas MANTIDO AQUI de propósito: aquele arquivo é mock de desenvolvimento e é editado à
|
||
# mão para exercitar ramos em teste manual (esvaziar `invoiceVariation` reproduz o ramo
|
||
# "sem VAS avulso"). Lendo-o, este teste ficava vermelho por causa de uma edição
|
||
# legítima — falha pelo motivo errado. O dado é do teste; o mock é do runtime local.
|
||
_FATURA_COM_AVULSOS_VARIADOS = {
|
||
"invoiceVariation": [
|
||
{
|
||
"value": "46.99",
|
||
"type": "servicos_contratados_de_parceiros",
|
||
"desc": "Serviços de valor adicionado",
|
||
"nItems": "10",
|
||
"items": [
|
||
# Fatura PASSADA (venc. 20/10) — 3 cobranças.
|
||
_item("Tamboro Mensal", "19.99", "2025-10-20T00:00:00.000Z"),
|
||
_item("VOD + Canais abertos", "10.0", "2025-10-20T00:00:00.000Z"),
|
||
_item("VOD + Canais abertos", "10.0", "2025-10-20T00:00:00.000Z"),
|
||
# Fatura ATUAL (venc. 20/11) — as 3 acima, inalteradas...
|
||
_item("Tamboro Mensal", "19.99", "2025-11-20T00:00:00.000Z"),
|
||
_item("VOD + Canais abertos", "10.0", "2025-11-20T00:00:00.000Z"),
|
||
_item("VOD + Canais abertos", "10.0", "2025-11-20T00:00:00.000Z"),
|
||
# ...mais 4 que ENTRARAM.
|
||
_item("Tamboro Mensal", "14.99", "2025-11-20T00:00:00.000Z"),
|
||
_item("TIM Fashion Mensal", "10.0", "2025-11-20T00:00:00.000Z"),
|
||
_item("Neymar Jr", "12.0", "2025-11-20T00:00:00.000Z"),
|
||
_item("Youtube Premium", "10.0", "2025-11-20T00:00:00.000Z"),
|
||
],
|
||
}
|
||
]
|
||
}
|
||
|
||
|
||
def test_fatura_realista_expoe_so_os_avulsos_que_entraram():
|
||
"""Payload no formato do runtime (ISO-8601 + ``value`` string), com as três
|
||
situações que convivem numa fatura real:
|
||
|
||
- **entraram:** ``Tamboro Mensal`` R$14,99, ``TIM Fashion Mensal`` R$10,00,
|
||
``Neymar Jr`` R$12,00 e ``Youtube Premium`` R$10,00;
|
||
- **não variaram:** ``Tamboro Mensal`` R$19,99 — mesmo NOME de um que variou, outra
|
||
cobrança: é a discriminação por VALOR que a feature existe para fazer — e
|
||
``VOD + Canais abertos`` R$10,00 ×2 (multiplicidade preservada dos dois lados);
|
||
- ``Youtube Premium`` vem no bucket avulso, reproduzindo o mis-bucket real da API;
|
||
barrá-lo é do ``varied_avulso_items`` (guard de estratégico), não daqui.
|
||
|
||
Soma da variação: R$46,99 = R$36,99 canceláveis + R$10,00 do estratégico."""
|
||
got = {
|
||
(c.desc, str(c.value))
|
||
for c in varied_current_charges(_FATURA_COM_AVULSOS_VARIADOS)
|
||
}
|
||
|
||
assert got == {
|
||
("Tamboro Mensal", "14.99"),
|
||
("TIM Fashion Mensal", "10.00"),
|
||
("Neymar Jr", "12.00"),
|
||
("Youtube Premium", "10.00"),
|
||
}
|
||
canceláveis = {v for d, v in got if d != "Youtube Premium"}
|
||
assert sum(Decimal(v) for v in canceláveis) == Decimal("36.99")
|
||
|
||
|
||
def test_cenario_real_cy0001_pega_so_o_item_que_variou():
|
||
"""Payload de runtime REAL (ISO-8601 + ``value`` string, `nItems` camelCase).
|
||
|
||
A fatura tem 5 avulsos cobrados, 4 deles idênticos ao ciclo anterior — só
|
||
`Aluguel de Filme 3` entrou. O critério antigo ("a fatura TEM avulso")
|
||
enfileirava os 5 para cancelamento."""
|
||
scenario = (
|
||
_REPO_ROOT
|
||
/ "automated_conversation_test/scenarios"
|
||
/ "cy0001_explicacao_de_fatura_planos_pos"
|
||
/ "cy0001_explicacao_de_fatura_planos_pos.yaml"
|
||
)
|
||
if not scenario.exists(): # pragma: no cover - cenário opcional no checkout
|
||
pytest.skip("cenário cy0001 ausente")
|
||
payloads: list[dict] = []
|
||
|
||
def walk(node):
|
||
if isinstance(node, dict):
|
||
if "invoiceVariation" in node:
|
||
payloads.append(node)
|
||
for value in node.values():
|
||
walk(value)
|
||
elif isinstance(node, list):
|
||
for value in node:
|
||
walk(value)
|
||
|
||
walk(yaml.safe_load(scenario.read_text(encoding="utf-8")))
|
||
assert payloads, "cenário sem invoiceVariation"
|
||
assert varied_current_charges(payloads[0]) == (
|
||
VariedCharge(desc="Aluguel de Filme 3", value=Decimal("19.90")),
|
||
)
|
||
|
||
|
||
# ----- nome curto na variação × nome qualificado na fatura -------------------
|
||
# Incidente e081812886a (URA 06/08): a análise de variação traz o nome CURTO do
|
||
# serviço ("FIT ME App") e a fatura traz o nome COMPLETO ("FIT ME App Premium
|
||
# Mensal"). A igualdade estrita de nome descartava o item, ``_has_retention_vas``
|
||
# devolvia False e o pedido de atendente encerrava ``nao_resolvido`` sem oferecer
|
||
# a retenção. O casamento é por PREFIXO DE TOKENS, com trava de valor ao centavo,
|
||
# item avulso pós-guards e candidato único — ambiguidade falha FECHADA (SPEC §9).
|
||
|
||
|
||
@pytest.fixture
|
||
def prefixo_de_tokens_ligado(monkeypatch):
|
||
"""Liga ``TIM_VAS_QUALIFIED_NAME_MATCH``.
|
||
|
||
A segunda passada é DESLIGADA por padrão (ver o comentário da flag em
|
||
``vas_variation``): o mesmo prefixo qualifica o mesmo produto em
|
||
"FIT ME App"→"FIT ME App Premium Mensal" e separa produtos DIFERENTES em
|
||
"Fluid Light"×"Fluid Premium". Os testes abaixo fixam o comportamento da
|
||
passada em si, então precisam ligá-la explicitamente."""
|
||
from app.domain.contas import vas_variation
|
||
|
||
monkeypatch.setattr(vas_variation, "_VAS_QUALIFIED_NAME_MATCH", True)
|
||
|
||
|
||
def test_prefixo_de_tokens_desligado_por_padrao():
|
||
"""Contrato do default: sem a flag, o caso do incidente NÃO casa.
|
||
|
||
É o custo aceito de eliminar o falso positivo "TIM Games"→"TIM Games
|
||
Premium", que tem forma idêntica e nenhum sinal disponível para separar."""
|
||
detail = _detail(
|
||
{"Itens Eventuais": [{"desc": "FIT ME App Premium Mensal", "value": 16.99}]}
|
||
)
|
||
|
||
assert _varied(detail, _entrou(("FIT ME App", "16.99"))) == []
|
||
|
||
|
||
def test_nome_curto_casa_com_nome_qualificado_da_fatura(prefixo_de_tokens_ligado):
|
||
"""Com a flag ligada: variação "FIT ME App" × fatura "FIT ME App Premium
|
||
Mensal", mesmo valor → casa."""
|
||
detail = _detail(
|
||
{"Itens Eventuais": [{"desc": "FIT ME App Premium Mensal", "value": 16.99}]}
|
||
)
|
||
|
||
items = _varied(detail, _entrou(("FIT ME App", "16.99")))
|
||
|
||
assert [it.canonical_name for it in items] == ["FIT ME App Premium Mensal"]
|
||
|
||
|
||
def test_nome_identico_continua_casando_pela_primeira_passada():
|
||
"""A segunda passada não pode roubar o que a igualdade estrita já resolveu:
|
||
nome idêntico casa uma única vez (sem duplicar o item)."""
|
||
detail = _detail(
|
||
{"Itens Eventuais": [{"desc": "FIT ME App", "value": 16.99}]}
|
||
)
|
||
|
||
items = _varied(detail, _entrou(("FIT ME App", "16.99")))
|
||
|
||
assert [it.canonical_name for it in items] == ["FIT ME App"]
|
||
|
||
|
||
def test_valor_divergente_nao_casa_mesmo_com_prefixo(prefixo_de_tokens_ligado):
|
||
"""Trava de valor: prefixo de nome bate, centavo não → fora. Cancelar por
|
||
nome parecido com valor diferente é cancelar o serviço errado."""
|
||
detail = _detail(
|
||
{"Itens Eventuais": [{"desc": "FIT ME App Premium Mensal", "value": 29.90}]}
|
||
)
|
||
|
||
assert _varied(detail, _entrou(("FIT ME App", "16.99"))) == []
|
||
|
||
|
||
def test_prefixo_no_meio_do_nome_nao_casa(prefixo_de_tokens_ligado):
|
||
"""Não é substring solta: os tokens da variação têm de ser o INÍCIO do nome
|
||
do item. "ME App" no meio de "FIT ME App Premium" não casa."""
|
||
detail = _detail(
|
||
{"Itens Eventuais": [{"desc": "FIT ME App Premium Mensal", "value": 16.99}]}
|
||
)
|
||
|
||
assert _varied(detail, _entrou(("ME App", "16.99"))) == []
|
||
|
||
|
||
def test_ambiguidade_falha_fechada(caplog, prefixo_de_tokens_ligado):
|
||
"""Dois itens com o mesmo prefixo e o mesmo valor → NENHUM é escolhido.
|
||
SPEC §9: nunca prometer cancelamento do que não se sabe ter variado."""
|
||
detail = _detail(
|
||
{
|
||
"Itens Eventuais": [
|
||
{"desc": "Gamedom Mensal", "value": 16.99},
|
||
{"desc": "Gamedom Semanal", "value": 16.99},
|
||
]
|
||
}
|
||
)
|
||
|
||
with caplog.at_level(logging.INFO):
|
||
items = _varied(detail, _entrou(("Gamedom", "16.99")))
|
||
|
||
assert items == []
|
||
assert "reason=multiple_candidates" in caplog.text
|
||
|
||
|
||
def test_nome_da_variacao_mais_longo_que_o_da_fatura_nao_casa(prefixo_de_tokens_ligado):
|
||
"""Direção única: a variação é o prefixo do item, nunca o contrário. Nome da
|
||
variação mais específico que o da fatura não casa (evita casar "X Premium"
|
||
com um "X" que pode ser outro serviço)."""
|
||
detail = _detail({"Itens Eventuais": [{"desc": "FIT ME App", "value": 16.99}]})
|
||
|
||
assert _varied(detail, _entrou(("FIT ME App Premium Mensal", "16.99"))) == []
|
||
|
||
|
||
def test_estrategico_com_nome_qualificado_continua_fora(prefixo_de_tokens_ligado):
|
||
"""Os guards não são contornáveis pela segunda passada: estratégico não é
|
||
cancelável (SPEC §12.A), mesmo casando por prefixo e valor."""
|
||
detail = _detail(
|
||
{"Itens Eventuais": [{"desc": "YouTube Premium Mensal", "value": 24.90}]}
|
||
)
|
||
|
||
assert _varied(detail, _entrou(("YouTube Premium", "24.90"))) == []
|
||
|
||
|
||
# ----- linha de destino: msisdn e ambiguidade entre linhas -------------------
|
||
# A fila desta rota é autorizada SEM confirmação pendente, e o ``msisdn`` vai cru
|
||
# para os args de ``cancelar_vas_avulso``. Cancelar na linha errada — ou sem
|
||
# linha — é dano irreversível no cliente, então os dois casos falham FECHADOS.
|
||
# Nenhum destes cenários tinha cobertura antes: os helpers do arquivo montam uma
|
||
# única linha (``_detail``), e a suíte inteira era cega a fatura multi-linha.
|
||
|
||
|
||
def _detail_multi(buckets: dict) -> dict:
|
||
"""``invoice_detail`` com VÁRIAS linhas: ``{msisdn: {seção: [entries]}}``."""
|
||
return dict(buckets)
|
||
|
||
|
||
def test_msisdn_ausente_nao_produz_item_cancelavel(caplog):
|
||
"""Bucket sem linha → nenhum item cancelável, com razão própria.
|
||
|
||
Sem ``msisdn`` não há para onde mandar o cancelamento. O descarte não pode
|
||
sair como ``name_absent``: o nome estava lá, faltava a linha."""
|
||
detail = _detail_multi({"": {"Itens Eventuais": [{"desc": "Fit Me App", "value": 16.99}]}})
|
||
|
||
with caplog.at_level(logging.INFO):
|
||
items = _varied(detail, _entrou(("Fit Me App", "16.99")))
|
||
|
||
assert items == []
|
||
assert "reason=missing_msisdn" in caplog.text
|
||
|
||
|
||
def test_mesma_cobranca_em_duas_linhas_falha_fechada(caplog):
|
||
"""Mesmo ``(nome, valor)`` em duas linhas e só UMA variou → não se sabe qual."""
|
||
detail = _detail_multi(
|
||
{
|
||
"11999999999": {"Itens Eventuais": [{"desc": "Fit Me App", "value": 16.99}]},
|
||
"11888888888": {"Itens Eventuais": [{"desc": "Fit Me App", "value": 16.99}]},
|
||
}
|
||
)
|
||
|
||
with caplog.at_level(logging.INFO):
|
||
items = _varied(detail, _entrou(("Fit Me App", "16.99")))
|
||
|
||
assert items == []
|
||
assert "reason=multiple_candidates" in caplog.text
|
||
|
||
|
||
def test_ordem_das_linhas_nao_altera_a_decisao():
|
||
"""A ordem do payload nunca pode mudar o resultado.
|
||
|
||
Antes, o loop era item-driven e consumia a cobrança no PRIMEIRO item casado —
|
||
o cancelamento saía na linha que o dict listasse antes."""
|
||
linha_a = {"Itens Eventuais": [{"desc": "Fit Me App", "value": 16.99}]}
|
||
linha_b = {"Itens Eventuais": [{"desc": "Fit Me App", "value": 16.99}]}
|
||
|
||
ab = _varied(_detail_multi({"11999999999": linha_a, "11888888888": linha_b}),
|
||
_entrou(("Fit Me App", "16.99")))
|
||
ba = _varied(_detail_multi({"11888888888": linha_b, "11999999999": linha_a}),
|
||
_entrou(("Fit Me App", "16.99")))
|
||
|
||
assert ab == ba == []
|
||
|
||
|
||
def test_cobranca_so_na_linha_secundaria_casa_naquela_linha():
|
||
"""Fatura multi-linha NÃO é motivo de recusa: o que barra é a chave ambígua.
|
||
|
||
Com o serviço em uma só das linhas há candidato único → casa, e o ``msisdn``
|
||
devolvido é o da linha CERTA (não o da primeira do payload)."""
|
||
detail = _detail_multi(
|
||
{
|
||
"11999999999": {"Itens Eventuais": [{"desc": "Fluid Light", "value": 12.90}]},
|
||
"11888888888": {"Itens Eventuais": [{"desc": "Fit Me App", "value": 16.99}]},
|
||
}
|
||
)
|
||
|
||
items = _varied(detail, _entrou(("Fit Me App", "16.99")))
|
||
|
||
assert [(it.canonical_name, it.msisdn) for it in items] == [
|
||
("Fit Me App", "11888888888")
|
||
]
|
||
|
||
|
||
def test_multiplicidade_preservada_quando_a_contagem_bate():
|
||
"""SPEC §9: o mesmo serviço cobrado 2× no ciclo são DUAS cobranças, e cada uma
|
||
entra na fila. A trava de candidato é por contagem, não "no máximo um" — duas
|
||
variações + duas cobranças na mesma linha casam as duas."""
|
||
detail = _detail(
|
||
{
|
||
"Itens Eventuais": [
|
||
{"desc": "Aluguel de Filme 1", "value": 19.90, "period": "05/04/26"},
|
||
{"desc": "Aluguel de Filme 1", "value": 19.90, "period": "18/04/26"},
|
||
]
|
||
}
|
||
)
|
||
|
||
items = _varied(
|
||
detail,
|
||
_entrou(("Aluguel de Filme 1", "19.90"), ("Aluguel de Filme 1", "19.90")),
|
||
by_charge=True,
|
||
)
|
||
|
||
assert [it.canonical_name for it in items] == [
|
||
"Aluguel de Filme 1",
|
||
"Aluguel de Filme 1",
|
||
]
|
||
|
||
|
||
# ----- ORDEM DOS GUARDS: identidade antes de executabilidade -----------------
|
||
# O caso que separa um matcher correto de um perigoso. Com uma cobrança variada
|
||
# e DUAS identidades possíveis — uma executável, outra não —, filtrar por
|
||
# ``msisdn`` primeiro COROARIA a executável: o descarte da outra viraria
|
||
# "evidência" de que sobrou a certa. Não é. A ambiguidade de identidade é
|
||
# anterior e independente da executabilidade.
|
||
|
||
|
||
def _duas_cobrancas_iguais(desc: str, value: str) -> dict:
|
||
"""Análise em que a MESMA cobrança variou duas vezes no ciclo."""
|
||
return _payload(
|
||
_group(_item(desc, value, CURRENT), _item(desc, value, CURRENT)),
|
||
current=_current_group((desc, value), (desc, value)),
|
||
)
|
||
|
||
|
||
def test_uma_variacao_e_duas_identidades_uma_sem_msisdn_falha_fechada(caplog):
|
||
"""CASO A: 1 cobrança variada, 2 itens de mesma identidade (um sem linha).
|
||
|
||
O item sem ``msisdn`` NÃO pode fazer o outro vencer por eliminação."""
|
||
detail = {
|
||
"11999999999": {"Itens Eventuais": [{"desc": "Fit Me App", "value": 16.99}]},
|
||
"": {"Itens Eventuais": [{"desc": "Fit Me App", "value": 16.99}]},
|
||
}
|
||
|
||
with caplog.at_level(logging.INFO):
|
||
items = _varied(detail, _entrou(("Fit Me App", "16.99")))
|
||
|
||
assert items == []
|
||
assert "reason=multiple_candidates" in caplog.text
|
||
|
||
|
||
def test_ordem_nao_altera_o_caso_a():
|
||
"""CASO A permutado: quem aparece primeiro no payload é irrelevante."""
|
||
com_linha = {"Itens Eventuais": [{"desc": "Fit Me App", "value": 16.99}]}
|
||
sem_linha = {"Itens Eventuais": [{"desc": "Fit Me App", "value": 16.99}]}
|
||
variacao = _entrou(("Fit Me App", "16.99"))
|
||
|
||
ab = _varied({"11999999999": com_linha, "": sem_linha}, variacao)
|
||
ba = _varied({"": sem_linha, "11999999999": com_linha}, variacao)
|
||
|
||
assert ab == ba == []
|
||
|
||
|
||
def test_duas_variacoes_e_duas_identidades_uma_sem_msisdn(caplog):
|
||
"""CASO B: 2 cobranças variadas para 2 identidades → não há ambiguidade.
|
||
|
||
A executável entra na fila; a sem linha é descartada com a razão própria e
|
||
NÃO herda o ``msisdn`` da outra."""
|
||
detail = {
|
||
"11999999999": {"Itens Eventuais": [{"desc": "Fit Me App", "value": 16.99}]},
|
||
"": {"Itens Eventuais": [{"desc": "Fit Me App", "value": 16.99}]},
|
||
}
|
||
|
||
with caplog.at_level(logging.INFO):
|
||
items = _varied(detail, _duas_cobrancas_iguais("Fit Me App", "16.99"))
|
||
|
||
assert [(it.canonical_name, it.msisdn) for it in items] == [
|
||
("Fit Me App", "11999999999")
|
||
]
|
||
assert "reason=missing_msisdn" in caplog.text
|
||
# a razão certa, e SÓ ela
|
||
assert "reason=name_absent" not in caplog.text
|
||
assert "reason=value_mismatch" not in caplog.text
|
||
|
||
|
||
def test_descarte_emite_exatamente_um_evento_por_cobranca(caplog):
|
||
"""Telemetria: uma cobrança descartada gera UM evento.
|
||
|
||
Reportar a mesma cobrança como ``multiple_candidates`` e depois como
|
||
``value_mismatch`` no resíduo corrompe a taxa de não-casamento."""
|
||
detail = {
|
||
"11999999999": {"Itens Eventuais": [{"desc": "Fit Me App", "value": 16.99}]},
|
||
"11888888888": {"Itens Eventuais": [{"desc": "Fit Me App", "value": 16.99}]},
|
||
}
|
||
|
||
with caplog.at_level(logging.INFO):
|
||
_varied(detail, _entrou(("Fit Me App", "16.99")))
|
||
|
||
eventos = [m for m in caplog.messages if "varied_charge_unmatched" in m]
|
||
assert len(eventos) == 1, eventos
|
||
assert "reason=multiple_candidates" in eventos[0]
|
||
|
||
|
||
# ----- desempate por DATA DE COBRANÇA ----------------------------------------
|
||
# A análise de variação não carrega data utilizável (o campo estruturado vem
|
||
# sentinela/ausente em produção) e não carrega msisdn. A única fonte de data é o
|
||
# texto CRU do ``invoiceExplanation`` (``invoice_explanation_base``):
|
||
#
|
||
# * Cobrança FIT ME App no valor de R$ 16.99 no dia 29/06/26.
|
||
#
|
||
# O PDF traz a outra ponta em ``ResolvedInvoiceItem.charge_date`` (do ``period``).
|
||
# A data só ESTREITA candidatos que hoje seriam descartados — nunca amplia, nunca
|
||
# elimina por ausência, e a cardinalidade tem de fechar exatamente.
|
||
|
||
from app.domain.contas.vas_variation import ( # noqa: E402
|
||
_charge_date_key,
|
||
_explained_charges,
|
||
)
|
||
|
||
|
||
def _det_linhas(*entries) -> dict:
|
||
"""``invoice_detail`` multi-linha: ``(desc, value, period, msisdn)``."""
|
||
buckets: dict = {}
|
||
for desc, value, period, msisdn in entries:
|
||
entry = {"desc": desc, "value": value}
|
||
if period is not None:
|
||
entry["period"] = period
|
||
buckets.setdefault(msisdn, {}).setdefault("SVA Detalhe Total", []).append(entry)
|
||
return buckets
|
||
|
||
|
||
def _varied_expl(detail, variation, explanation=None, *, by_charge=True):
|
||
return varied_avulso_items(
|
||
invoice_variation=variation,
|
||
invoice_detail=detail,
|
||
resolver=InvoiceResolver(),
|
||
by_charge=by_charge,
|
||
explanation_text=explanation,
|
||
)
|
||
|
||
|
||
def _bullet(desc, value, dia):
|
||
return f"* Cobrança {desc} no valor de R$ {value} no dia {dia}."
|
||
|
||
|
||
# ----- normalizador ----------------------------------------------------------
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"raw,esperado",
|
||
[
|
||
("2025-11-01T00:00:00.000Z", ("2025", "11", "01")),
|
||
("01/11/25", ("2025", "11", "01")),
|
||
("25/06/26", ("2026", "06", "25")),
|
||
("01/11/2025", ("2025", "11", "01")),
|
||
("0000-12-31T20:53:32.000Z", None), # sentinela do backend
|
||
("", None),
|
||
(None, None),
|
||
("lixo", None),
|
||
("32/13/25", None), # data impossível
|
||
],
|
||
ids=lambda v: str(v)[:26],
|
||
)
|
||
def test_charge_date_key_normaliza_ou_recusa(raw, esperado):
|
||
"""Sentinela e ilegível viram ``None``: "não sei" tem de ser indistinguível
|
||
de "não tem", senão vira desempate falso."""
|
||
assert _charge_date_key(raw) == esperado
|
||
|
||
|
||
def test_explained_charges_so_extrai_bullet_com_data():
|
||
"""Formatos sem dia não participam — sem data não há evidência."""
|
||
texto = (
|
||
"Houve a variação dos seguintes serviços de valor adicionado:\n"
|
||
"* Cobrança FIT ME App no valor de R$ 16.99 no dia 29/06/26.\n"
|
||
"* Gamedom Mensal no valor de R$ 14,99 no dia 25/06/26.\n"
|
||
"* MasterChef variou em R$ 19.99.\n"
|
||
"Foram cobrados os seguintes serviços de terceiros: PARAGUAI."
|
||
)
|
||
|
||
achados = _explained_charges(texto)
|
||
|
||
assert [(c.desc, str(c.value), c.date_key) for c in achados] == [
|
||
("FIT ME App", "16.99", ("2026", "06", "29")),
|
||
("Gamedom Mensal", "14.99", ("2026", "06", "25")),
|
||
]
|
||
|
||
|
||
@pytest.mark.parametrize("texto", ["", " ", None, "texto sem nada", "x" * 20000])
|
||
def test_explained_charges_degrada_sem_quebrar(texto):
|
||
assert _explained_charges(texto) == [] or all(
|
||
c.date_key for c in _explained_charges(texto)
|
||
)
|
||
|
||
|
||
# ----- Caso A: nome qualificado (o incidente e081812886a) --------------------
|
||
|
||
|
||
def test_nome_qualificado_casa_por_valor_e_data():
|
||
""""FIT ME App" na análise × "FIT ME App Premium Mensal" no PDF: os nomes
|
||
divergem, mas valor ao centavo + data identificam a MESMA cobrança."""
|
||
detail = _det_linhas(
|
||
("FIT ME App Premium Mensal", "16.99", "29/06/26", "11999999999")
|
||
)
|
||
variacao = _entrou(("FIT ME App", "16.99"))
|
||
|
||
assert _varied_expl(detail, variacao) == [] # sem explicação: fail closed
|
||
items = _varied_expl(detail, variacao, _bullet("FIT ME App", "16.99", "29/06/26"))
|
||
|
||
assert [(it.canonical_name, it.msisdn) for it in items] == [
|
||
("FIT ME App Premium Mensal", "11999999999")
|
||
]
|
||
|
||
|
||
def test_nome_qualificado_com_data_divergente_nao_casa():
|
||
"""A ponte é (valor, data). Data diferente → não é a mesma cobrança."""
|
||
detail = _det_linhas(
|
||
("FIT ME App Premium Mensal", "16.99", "29/06/26", "11999999999")
|
||
)
|
||
|
||
assert (
|
||
_varied_expl(
|
||
detail,
|
||
_entrou(("FIT ME App", "16.99")),
|
||
_bullet("FIT ME App", "16.99", "30/06/26"),
|
||
)
|
||
== []
|
||
)
|
||
|
||
|
||
# ----- Caso B: mesmo valor, produtos diferentes, datas diferentes ------------
|
||
|
||
|
||
def test_mesmo_valor_produtos_diferentes_datas_distintas():
|
||
"""Dois produtos ao mesmo preço não podem ser separados por valor — a data
|
||
separa, e os DOIS casam corretamente."""
|
||
detail = _det_linhas(
|
||
("MasterChef Mensal", "19.99", "14/06/26", "11999999999"),
|
||
("Bancah Jornais II", "19.99", "18/06/26", "11999999999"),
|
||
)
|
||
variacao = _entrou(("MasterChef Mensal", "19.99"), ("Bancah Jornais II", "19.99"))
|
||
explicacao = (
|
||
_bullet("MasterChef Mensal", "19.99", "14/06/26")
|
||
+ "\n"
|
||
+ _bullet("Bancah Jornais II", "19.99", "18/06/26")
|
||
)
|
||
|
||
items = _varied_expl(detail, variacao, explicacao)
|
||
|
||
assert sorted(it.canonical_name for it in items) == [
|
||
"Bancah Jornais II",
|
||
"MasterChef Mensal",
|
||
]
|
||
|
||
|
||
# ----- Caso C: titular × dependente (a colisão de linha) ---------------------
|
||
|
||
|
||
def test_titular_e_dependente_datas_diferentes_casa_a_linha_certa(caplog):
|
||
"""Mesmo nome, mesmo valor, DUAS linhas: a data diz qual variou."""
|
||
detail = _det_linhas(
|
||
("TIM Fashion Mensal", "10.00", "01/11/25", "1199999999"),
|
||
("TIM Fashion Mensal", "10.00", "30/04/25", "91998188119"),
|
||
)
|
||
|
||
with caplog.at_level(logging.INFO):
|
||
items = _varied_expl(
|
||
detail,
|
||
_entrou(("TIM Fashion Mensal", "10.00")),
|
||
_bullet("TIM Fashion Mensal", "10.00", "01/11/25"),
|
||
)
|
||
|
||
assert [(it.canonical_name, it.msisdn) for it in items] == [
|
||
("TIM Fashion Mensal", "1199999999")
|
||
]
|
||
assert "reason=charge_date" in caplog.text
|
||
|
||
|
||
def test_titular_e_dependente_mesma_data_falha_fechada(caplog):
|
||
"""Datas iguais não desempatam nada."""
|
||
detail = _det_linhas(
|
||
("TIM Fashion Mensal", "10.00", "01/11/25", "1199999999"),
|
||
("TIM Fashion Mensal", "10.00", "01/11/25", "91998188119"),
|
||
)
|
||
|
||
with caplog.at_level(logging.INFO):
|
||
items = _varied_expl(
|
||
detail,
|
||
_entrou(("TIM Fashion Mensal", "10.00")),
|
||
_bullet("TIM Fashion Mensal", "10.00", "01/11/25"),
|
||
)
|
||
|
||
assert items == []
|
||
assert "reason=multiple_candidates" in caplog.text
|
||
assert "date_hint=ambiguous" in caplog.text
|
||
|
||
|
||
def test_candidato_sem_data_nao_pode_ser_eliminado(caplog):
|
||
"""A REGRA CRÍTICA: ausência de data NÃO é evidência negativa.
|
||
|
||
Com um candidato datado e outro sem data, escolher o datado seria decidir
|
||
por eliminação — proibido."""
|
||
detail = _det_linhas(
|
||
("TIM Fashion Mensal", "10.00", "01/11/25", "1199999999"),
|
||
("TIM Fashion Mensal", "10.00", None, "91998188119"),
|
||
)
|
||
|
||
with caplog.at_level(logging.INFO):
|
||
items = _varied_expl(
|
||
detail,
|
||
_entrou(("TIM Fashion Mensal", "10.00")),
|
||
_bullet("TIM Fashion Mensal", "10.00", "01/11/25"),
|
||
)
|
||
|
||
assert items == []
|
||
assert "date_hint=undated" in caplog.text
|
||
|
||
|
||
def test_ambos_sem_data_falha_fechada():
|
||
detail = _det_linhas(
|
||
("TIM Fashion Mensal", "10.00", None, "1199999999"),
|
||
("TIM Fashion Mensal", "10.00", None, "91998188119"),
|
||
)
|
||
|
||
assert (
|
||
_varied_expl(
|
||
detail,
|
||
_entrou(("TIM Fashion Mensal", "10.00")),
|
||
_bullet("TIM Fashion Mensal", "10.00", "01/11/25"),
|
||
)
|
||
== []
|
||
)
|
||
|
||
|
||
def test_sem_hint_para_a_cobranca_falha_fechada(caplog):
|
||
"""Explicação existe mas não menciona ESTA cobrança."""
|
||
detail = _det_linhas(
|
||
("TIM Fashion Mensal", "10.00", "01/11/25", "1199999999"),
|
||
("TIM Fashion Mensal", "10.00", "30/04/25", "91998188119"),
|
||
)
|
||
|
||
with caplog.at_level(logging.INFO):
|
||
items = _varied_expl(
|
||
detail,
|
||
_entrou(("TIM Fashion Mensal", "10.00")),
|
||
_bullet("Outro Servico", "99.99", "01/11/25"),
|
||
)
|
||
|
||
assert items == []
|
||
assert "date_hint=absent" in caplog.text
|
||
|
||
|
||
def test_sentinela_na_explicacao_nao_vira_evidencia():
|
||
"""Bullet com data sentinela é descartado no parser."""
|
||
detail = _det_linhas(
|
||
("TIM Fashion Mensal", "10.00", "01/11/25", "1199999999"),
|
||
("TIM Fashion Mensal", "10.00", "30/04/25", "91998188119"),
|
||
)
|
||
|
||
assert (
|
||
_varied_expl(
|
||
detail,
|
||
_entrou(("TIM Fashion Mensal", "10.00")),
|
||
"* Cobrança TIM Fashion Mensal no valor de R$ 10.00 no dia 31/12/00.",
|
||
)
|
||
== []
|
||
)
|
||
|
||
|
||
# ----- cardinalidade e reuso -------------------------------------------------
|
||
|
||
|
||
def test_resolucao_parcial_e_proibida():
|
||
"""``missing=2`` e a data identifica só 1 → nada casa (tudo ou nada)."""
|
||
detail = _det_linhas(
|
||
("Aya Books Premium", "9.99", "10/06/26", "11999999999"),
|
||
("Aya Books Premium", "9.99", "20/06/26", "11999999999"),
|
||
("Aya Books Premium", "9.99", "30/06/26", "11999999999"),
|
||
)
|
||
variacao = _entrou(("Aya Books Premium", "9.99"), ("Aya Books Premium", "9.99"))
|
||
|
||
items = _varied_expl(
|
||
detail, variacao, _bullet("Aya Books Premium", "9.99", "10/06/26")
|
||
)
|
||
|
||
assert items == []
|
||
|
||
|
||
def test_multiplicidade_real_dois_hints_casam_dois_itens():
|
||
"""Cardinalidade fecha: 2 cobranças variadas, 2 hints, 2 itens datados."""
|
||
detail = _det_linhas(
|
||
("Aya Books Premium", "9.99", "10/06/26", "11999999999"),
|
||
("Aya Books Premium", "9.99", "20/06/26", "11999999999"),
|
||
)
|
||
variacao = _entrou(("Aya Books Premium", "9.99"), ("Aya Books Premium", "9.99"))
|
||
explicacao = (
|
||
_bullet("Aya Books Premium", "9.99", "10/06/26")
|
||
+ "\n"
|
||
+ _bullet("Aya Books Premium", "9.99", "20/06/26")
|
||
)
|
||
|
||
items = _varied_expl(detail, variacao, explicacao)
|
||
|
||
assert len(items) == 2
|
||
assert sorted(it.charge_date for it in items) == ["10/06/26", "20/06/26"]
|
||
|
||
|
||
def test_hint_nao_e_reutilizado_por_duas_cobrancas():
|
||
"""Um bullet identifica UMA cobrança.
|
||
|
||
Duas chaves distintas (nomes curtos diferentes) disputam o mesmo hint no
|
||
passe residual: a primeira o consome, a segunda fica sem evidência e falha
|
||
fechada — o hint NÃO é reaproveitado."""
|
||
detail = _det_linhas(
|
||
("Aya Books Premium", "9.99", "10/06/26", "11999999999"),
|
||
("Aya Ensinah Premium", "9.99", "20/06/26", "11999999999"),
|
||
)
|
||
variacao = _entrou(("Aya Books", "9.99"), ("Aya Ensinah", "9.99"))
|
||
|
||
items = _varied_expl(
|
||
detail, variacao, _bullet("Aya Books", "9.99", "10/06/26")
|
||
)
|
||
|
||
assert [it.canonical_name for it in items] == ["Aya Books Premium"]
|
||
|
||
|
||
def test_guards_valem_no_passe_por_data():
|
||
"""Estratégico não é resgatado pela data."""
|
||
detail = _det_linhas(
|
||
("YouTube Premium Mensal", "24.90", "10/06/26", "11999999999")
|
||
)
|
||
|
||
assert (
|
||
_varied_expl(
|
||
detail,
|
||
_entrou(("YouTube Premium", "24.90")),
|
||
_bullet("YouTube Premium", "24.90", "10/06/26"),
|
||
)
|
||
== []
|
||
)
|
||
|
||
|
||
def test_msisdn_ausente_continua_bloqueando_no_passe_por_data(caplog):
|
||
"""Guard 3 vale nos dois passes: identidade resolvida, mas sem linha."""
|
||
detail = _det_linhas(("FIT ME App Premium Mensal", "16.99", "29/06/26", ""))
|
||
|
||
with caplog.at_level(logging.INFO):
|
||
items = _varied_expl(
|
||
detail,
|
||
_entrou(("FIT ME App", "16.99")),
|
||
_bullet("FIT ME App", "16.99", "29/06/26"),
|
||
)
|
||
|
||
assert items == []
|
||
assert "reason=missing_msisdn" in caplog.text
|
||
|
||
|
||
def test_explicacao_nao_altera_casamento_ja_valido():
|
||
"""Nenhum match existente pode regredir por a explicação existir."""
|
||
detail = _det_linhas(("Tamboro Mensal", "19.99", "02/11/25", "11999999999"))
|
||
variacao = _entrou(("Tamboro Mensal", "19.99"))
|
||
|
||
sem = _varied_expl(detail, variacao)
|
||
com = _varied_expl(detail, variacao, _bullet("Tamboro Mensal", "19.99", "02/11/25"))
|
||
|
||
assert [it.canonical_name for it in sem] == ["Tamboro Mensal"]
|
||
assert [it.canonical_name for it in com] == ["Tamboro Mensal"]
|
||
|
||
|
||
def test_passe_residual_exige_cardinalidade_exata():
|
||
"""Duas cobranças datadas candidatas para UMA variada → nada casa.
|
||
|
||
Sem a igualdade estrita (``!= missing``), um ``< missing`` deixaria a fila
|
||
levar as DUAS — cancelando uma cobrança que ninguém disse ter variado."""
|
||
detail = _det_linhas(
|
||
("Aya Books Premium", "9.99", "10/06/26", "11999999999"),
|
||
("Aya Ensinah Premium", "9.99", "10/06/26", "11999999999"),
|
||
)
|
||
explicacao = (
|
||
_bullet("Aya", "9.99", "10/06/26") + "\n" + _bullet("Aya", "9.99", "10/06/26")
|
||
)
|
||
|
||
assert _varied_expl(detail, _entrou(("Aya", "9.99")), explicacao) == []
|