525 lines
16 KiB
Python
525 lines
16 KiB
Python
from __future__ import annotations
|
|
|
|
from contextlib import contextmanager
|
|
from typing import Any
|
|
|
|
from app.domain.contas import contestation_validation
|
|
from app.domain.contas.contestation_validation import (
|
|
validate_contestation_items,
|
|
)
|
|
|
|
|
|
def _vas_invoice(*items: dict[str, Any]) -> dict[str, Any]:
|
|
return {"Servicos de Valor Adicionado": list(items)}
|
|
|
|
|
|
def test_validate_contestation_items_bloqueia_item_inexistente() -> None:
|
|
items = [
|
|
{
|
|
"item_name": "Pacote nao cobrado",
|
|
"claimed_amount": "10",
|
|
"validated_amount": "10",
|
|
}
|
|
]
|
|
|
|
_, validation_log, error = validate_contestation_items(
|
|
items,
|
|
_vas_invoice({"desc": "Pacote cobrado", "valor": "10,00"}),
|
|
)
|
|
|
|
assert error == "Item 'Pacote nao cobrado' nao encontrado no json da fatura."
|
|
assert validation_log[-1]["erro"] == "item_nao_encontrado_na_fatura"
|
|
|
|
|
|
def test_validate_contestation_items_bloqueia_item_fora_da_secao_vas() -> None:
|
|
items = [
|
|
{
|
|
"item_name": "Pacote Dados",
|
|
"claimed_amount": "10",
|
|
"validated_amount": "10",
|
|
}
|
|
]
|
|
|
|
_, validation_log, error = validate_contestation_items(
|
|
items,
|
|
{"Plano": [{"desc": "Pacote Dados", "valor": "10,00"}]},
|
|
)
|
|
|
|
assert error == "Item 'Pacote Dados' nao e do tipo VAS no json da fatura."
|
|
assert validation_log[-1]["erro"] == "item_fora_secao_vas"
|
|
|
|
|
|
def test_validate_contestation_items_bloqueia_vas_estrategico() -> None:
|
|
items = [
|
|
{
|
|
"item_name": "Netflix Premium",
|
|
"claimed_amount": "10",
|
|
"validated_amount": "10",
|
|
}
|
|
]
|
|
|
|
_, validation_log, error = validate_contestation_items(
|
|
items,
|
|
_vas_invoice({"desc": "Netflix Premium", "valor": "10,00"}),
|
|
)
|
|
|
|
assert (
|
|
error
|
|
== "Item 'Netflix Premium' identificado como VAS estrategico e nao pode ser ajustado."
|
|
)
|
|
assert validation_log[-1]["erro"] == "vas_estrategico_nao_permitido"
|
|
|
|
|
|
def test_validate_contestation_items_bloqueia_valor_maior_que_o_item() -> None:
|
|
items = [
|
|
{
|
|
"item_name": "Pacote Musica",
|
|
"claimed_amount": "12",
|
|
"validated_amount": "12",
|
|
}
|
|
]
|
|
|
|
_, validation_log, error = validate_contestation_items(
|
|
items,
|
|
_vas_invoice({"desc": "Pacote Musica", "valor": "10,00"}),
|
|
)
|
|
|
|
assert (
|
|
error
|
|
== "Valor de ajuste do item 'Pacote Musica' excede o valor cobrado na fatura."
|
|
)
|
|
assert validation_log[-1]["erro"] == "valor_ajuste_maior_que_item"
|
|
|
|
|
|
def _record_observation_factory(records: list[dict[str, Any]]):
|
|
@contextmanager
|
|
def _record_observation(
|
|
*,
|
|
name: str,
|
|
input: dict[str, Any] | None = None,
|
|
metadata: dict[str, Any] | None = None,
|
|
):
|
|
record: dict[str, Any] = {
|
|
"name": name,
|
|
"input": input,
|
|
"metadata": metadata,
|
|
"updates": [],
|
|
}
|
|
|
|
class _Observation:
|
|
def update(self, **kwargs: Any) -> None:
|
|
record["updates"].append(dict(kwargs))
|
|
|
|
records.append(record)
|
|
yield _Observation()
|
|
|
|
return _record_observation
|
|
|
|
|
|
def test_validate_contestation_items_emite_spans_quando_bloqueia(
|
|
monkeypatch,
|
|
) -> None:
|
|
records: list[dict[str, Any]] = []
|
|
monkeypatch.setattr(
|
|
contestation_validation,
|
|
"_start_guardrail_observation",
|
|
_record_observation_factory(records),
|
|
)
|
|
|
|
_, validation_log, error = validate_contestation_items(
|
|
[
|
|
{
|
|
"item_name": "Pacote nao cobrado",
|
|
"claimed_amount": "10",
|
|
"validated_amount": "10",
|
|
}
|
|
],
|
|
_vas_invoice({"desc": "Pacote cobrado", "valor": "10,00"}),
|
|
)
|
|
|
|
assert error
|
|
assert [r["name"] for r in records] == [
|
|
"guardrail.CVAL.evaluated",
|
|
"guardrail.CVAL.blocked",
|
|
]
|
|
|
|
parent, child = records
|
|
assert parent["metadata"] == {
|
|
"mechanism": "guardrail_action_validation",
|
|
"code": "CVAL",
|
|
"action": "abrir_contestacao_cliente",
|
|
}
|
|
assert parent["input"]["items_count"] == 1
|
|
assert parent["input"]["invoice_candidates_count"] == 1
|
|
assert parent["updates"][-1]["level"] == "WARNING"
|
|
assert parent["updates"][-1]["output"]["approved"] is False
|
|
assert parent["updates"][-1]["output"]["validation_log"] == validation_log
|
|
assert (
|
|
parent["updates"][-1]["output"]["reason"]
|
|
== "item_nao_encontrado_na_fatura"
|
|
)
|
|
|
|
assert child["metadata"]["reason"] == "item_nao_encontrado_na_fatura"
|
|
assert child["updates"][-1]["level"] == "WARNING"
|
|
assert child["updates"][-1]["output"]["blocked"] is True
|
|
|
|
|
|
def test_validate_contestation_items_emite_span_quando_aprova(
|
|
monkeypatch,
|
|
) -> None:
|
|
records: list[dict[str, Any]] = []
|
|
monkeypatch.setattr(
|
|
contestation_validation,
|
|
"_start_guardrail_observation",
|
|
_record_observation_factory(records),
|
|
)
|
|
|
|
items, validation_log, error = validate_contestation_items(
|
|
[
|
|
{
|
|
"item_name": "Pacote Musica",
|
|
"claimed_amount": "10",
|
|
"validated_amount": "10",
|
|
}
|
|
],
|
|
_vas_invoice({"desc": "Pacote Musica", "valor": "10,00"}),
|
|
)
|
|
|
|
assert error is None
|
|
assert validation_log[-1]["status"] == "aprovado"
|
|
assert items[0]["validated_amount"] == "10"
|
|
assert [r["name"] for r in records] == ["guardrail.CVAL.evaluated"]
|
|
|
|
update = records[0]["updates"][-1]
|
|
assert "level" not in update
|
|
assert update["output"]["approved"] is True
|
|
assert update["output"]["items_count"] == 1
|
|
assert update["output"]["items_validated_count"] == 1
|
|
assert update["output"]["items_approved_count"] == 1
|
|
assert update["output"]["items_rejected_count"] == 0
|
|
assert update["output"]["validation_log"] == validation_log
|
|
|
|
|
|
def test_validate_contestation_items_valida_todos_antes_de_retornar() -> None:
|
|
items = [
|
|
{
|
|
"item_name": "Pacote Musica",
|
|
"claimed_amount": "10",
|
|
"validated_amount": "10",
|
|
},
|
|
{
|
|
"item_name": "Pacote nao cobrado",
|
|
"claimed_amount": "10",
|
|
"validated_amount": "10",
|
|
},
|
|
{
|
|
"item_name": "Pacote Filme",
|
|
"claimed_amount": "5",
|
|
"validated_amount": "5",
|
|
},
|
|
]
|
|
|
|
_, validation_log, error = validate_contestation_items(
|
|
items,
|
|
_vas_invoice(
|
|
{"desc": "Pacote Musica", "valor": "10,00"},
|
|
{"desc": "Pacote Filme", "valor": "5,00"},
|
|
),
|
|
)
|
|
|
|
assert error == "Item 'Pacote nao cobrado' nao encontrado no json da fatura."
|
|
statuses = [(entry["item_name"], entry["status"]) for entry in validation_log]
|
|
assert statuses == [
|
|
("Pacote Musica", "aprovado"),
|
|
("Pacote nao cobrado", "reprovado"),
|
|
("Pacote Filme", "aprovado"),
|
|
]
|
|
|
|
|
|
def test_validate_contestation_items_usa_classe_avulso_canonica() -> None:
|
|
items = [
|
|
{
|
|
"item_name": "Tamboro Mensal",
|
|
"claimed_amount": "14.99",
|
|
"validated_amount": "14.99",
|
|
}
|
|
]
|
|
|
|
_, validation_log, error = validate_contestation_items(
|
|
items,
|
|
{
|
|
"Secao Qualquer": [
|
|
{
|
|
"desc": "Tamboro Mensal",
|
|
"value": 14.99,
|
|
"classe": "avulso",
|
|
"verb": "cancelar",
|
|
}
|
|
]
|
|
},
|
|
)
|
|
|
|
assert error is None
|
|
assert validation_log[-1]["status"] == "aprovado"
|
|
assert validation_log[-1]["secao_vas"] is True
|
|
|
|
|
|
def test_validate_contestation_items_bloqueia_classe_estrategico_canonica() -> None:
|
|
items = [
|
|
{
|
|
"item_name": "ServicoNovo",
|
|
"claimed_amount": "10",
|
|
"validated_amount": "10",
|
|
}
|
|
]
|
|
|
|
_, validation_log, error = validate_contestation_items(
|
|
items,
|
|
{
|
|
"Servicos de Valor Adicionado": [
|
|
{
|
|
"desc": "ServicoNovo",
|
|
"value": 10.0,
|
|
"classe": "estrategico",
|
|
"estrategico": True,
|
|
"verb": "falar sobre",
|
|
}
|
|
]
|
|
},
|
|
)
|
|
|
|
assert (
|
|
error
|
|
== "Item 'ServicoNovo' identificado como VAS estrategico e nao pode ser ajustado."
|
|
)
|
|
assert validation_log[-1]["erro"] == "vas_estrategico_nao_permitido"
|
|
assert validation_log[-1]["vas_estrategico"] is True
|
|
|
|
|
|
def test_validate_contestation_items_span_blocked_inclui_contagens(
|
|
monkeypatch,
|
|
) -> None:
|
|
records: list[dict[str, Any]] = []
|
|
monkeypatch.setattr(
|
|
contestation_validation,
|
|
"_start_guardrail_observation",
|
|
_record_observation_factory(records),
|
|
)
|
|
|
|
_, validation_log, error = validate_contestation_items(
|
|
[
|
|
{
|
|
"item_name": "Pacote Musica",
|
|
"claimed_amount": "10",
|
|
"validated_amount": "10",
|
|
},
|
|
{
|
|
"item_name": "Pacote nao cobrado",
|
|
"claimed_amount": "10",
|
|
"validated_amount": "10",
|
|
},
|
|
],
|
|
_vas_invoice({"desc": "Pacote Musica", "valor": "10,00"}),
|
|
)
|
|
|
|
assert error
|
|
assert [r["name"] for r in records] == [
|
|
"guardrail.CVAL.evaluated",
|
|
"guardrail.CVAL.blocked",
|
|
]
|
|
parent, child = records
|
|
parent_output = parent["updates"][-1]["output"]
|
|
assert parent_output["items_validated_count"] == 2
|
|
assert parent_output["items_approved_count"] == 1
|
|
assert parent_output["items_rejected_count"] == 1
|
|
assert parent_output["validation_log"] == validation_log
|
|
|
|
child_output = child["updates"][-1]["output"]
|
|
assert child_output["items_validated_count"] == 2
|
|
assert child_output["items_approved_count"] == 1
|
|
assert child_output["items_rejected_count"] == 1
|
|
|
|
|
|
def test_validate_contestation_items_bloqueia_quando_total_ajustes_excede_fatura() -> None:
|
|
items = [
|
|
{
|
|
"item_name": "VAS Musica",
|
|
"claimed_amount": "100",
|
|
"validated_amount": "100",
|
|
},
|
|
{
|
|
"item_name": "VAS Filme",
|
|
"claimed_amount": "100",
|
|
"validated_amount": "100",
|
|
},
|
|
]
|
|
|
|
_, validation_log, error = validate_contestation_items(
|
|
items,
|
|
{
|
|
"Fatura Resumo": [{"desc": "Total geral", "value": 150.0}],
|
|
"Servicos de Valor Adicionado": [
|
|
{"desc": "VAS Musica", "valor": "100,00"},
|
|
{"desc": "VAS Filme", "valor": "100,00"},
|
|
],
|
|
},
|
|
)
|
|
|
|
assert error == (
|
|
"Valor total de ajustes (200) excede o valor total da fatura (150)."
|
|
)
|
|
total_entry = validation_log[-1]
|
|
assert total_entry["erro"] == "total_ajustes_excede_fatura"
|
|
assert total_entry["valor_total_ajustes"] == "200"
|
|
assert total_entry["valor_total_fatura"] == "150"
|
|
|
|
|
|
def test_validate_contestation_items_aprova_quando_total_ajustes_dentro_do_limite() -> None:
|
|
items = [
|
|
{
|
|
"item_name": "VAS Musica",
|
|
"claimed_amount": "50",
|
|
"validated_amount": "50",
|
|
}
|
|
]
|
|
|
|
_, validation_log, error = validate_contestation_items(
|
|
items,
|
|
{
|
|
"Fatura Resumo": [{"desc": "Total geral", "value": 150.0}],
|
|
"Servicos de Valor Adicionado": [
|
|
{"desc": "VAS Musica", "valor": "50,00"},
|
|
],
|
|
},
|
|
)
|
|
|
|
assert error is None
|
|
assert validation_log[-1]["status"] == "aprovado"
|
|
|
|
|
|
def test_contestation_action_classifies_already_contested_and_missing_requested_item():
|
|
from app.domain.contas.service import ContasDomainService
|
|
from app.domain.contas.workflow_actions import build_contas_workflow_actions
|
|
|
|
class Client:
|
|
def billing_analysis(self, msisdn):
|
|
return {"currentInvoice": {"items": [{"description": "TIM Fashion Mensal", "value": 14.99}, {"description": "Outro VAS", "value": 5.0}]}}
|
|
def contestar(self, payload):
|
|
return {"itemsResponse": [
|
|
{"itemName": "TIM Fashion Mensal", "message": "Item ja foi contestado anteriormente", "status": "REJEITADA"}
|
|
]}
|
|
|
|
action = build_contas_workflow_actions(ContasDomainService(client=Client())).get("abrir_contestacao_cliente")
|
|
result = action({
|
|
"msisdn": "11999999999", "skip_invoice_item_validation": True,
|
|
"items": [
|
|
{"itemName": "TIM Fashion Mensal", "claimedAmount": "14,99", "validatedAmount": "14,99"},
|
|
{"itemName": "Outro VAS", "claimedAmount": "5,00", "validatedAmount": "5,00"},
|
|
],
|
|
}, {"input": {}})
|
|
assert result["itens_ja_contestados"] == ["TIM Fashion Mensal"]
|
|
assert result["contested_items"] == []
|
|
assert result["not_contested_items"][0]["itemName"] == "Outro VAS"
|
|
assert result["contested_invoice_amount_open"] == "0.00"
|
|
|
|
|
|
def test_contestation_action_calculates_totals_only_for_items_actually_contested():
|
|
from app.domain.contas.service import ContasDomainService
|
|
from app.domain.contas.workflow_actions import build_contas_workflow_actions
|
|
|
|
class Client:
|
|
def contestar(self, payload):
|
|
return {"itemsResponse": [
|
|
{"itemName": "TIM Fashion Mensal", "status": "INICIADA"},
|
|
{"itemName": "Outro VAS", "status": "REJEITADA", "message": "fora da alcada"},
|
|
]}
|
|
|
|
action = build_contas_workflow_actions(ContasDomainService(client=Client())).get("abrir_contestacao_cliente")
|
|
result = action({
|
|
"msisdn": "11999999999", "skip_invoice_item_validation": True,
|
|
"items": [
|
|
{"itemName": "TIM Fashion Mensal", "claimedAmount": "14,99", "validatedAmount": "14,99"},
|
|
{"itemName": "Outro VAS", "claimedAmount": "5,00", "validatedAmount": "5,00"},
|
|
],
|
|
}, {"input": {}})
|
|
assert [x["itemName"] for x in result["contested_items"]] == ["TIM Fashion Mensal"]
|
|
assert [x["itemName"] for x in result["not_contested_items"]] == ["Outro VAS"]
|
|
assert result["contested_invoice_amount_open"] == "14.99"
|
|
assert result["contested_invoice_amount"] == "14.99"
|
|
|
|
|
|
def test_validate_contestation_items_reconhece_vas_por_type_do_billing_analysis() -> None:
|
|
items = [
|
|
{
|
|
"item_name": "TIM Fashion Mensal",
|
|
"claimed_amount": "10",
|
|
"validated_amount": "10",
|
|
}
|
|
]
|
|
invoice = {
|
|
"currentInvoice": [
|
|
{
|
|
"type": "servicos_contratados_de_parceiros",
|
|
"desc": "Serviços de valor adicionado",
|
|
"value": "76.98",
|
|
"items": [
|
|
{
|
|
"type": "servicos_contratados_de_parceiros",
|
|
"desc": "TIM Fashion Mensal",
|
|
"value": "10.0",
|
|
"contestable": True,
|
|
}
|
|
],
|
|
}
|
|
]
|
|
}
|
|
|
|
validated, validation_log, error = validate_contestation_items(items, invoice)
|
|
|
|
assert error is None
|
|
assert validation_log[-1]["status"] == "aprovado"
|
|
assert validation_log[-1]["secao_vas"] is True
|
|
assert validation_log[-1]["item_fatura_resolvido"] == "TIM Fashion Mensal"
|
|
assert validation_log[-1]["tipo_fatura"] == "servicos_contratados_de_parceiros"
|
|
assert validated[0]["validated_amount"] == "10"
|
|
|
|
|
|
def test_validate_contestation_items_prefere_evidencia_vas_quando_item_aparece_em_multiplas_visoes() -> None:
|
|
items = [
|
|
{
|
|
"item_name": "TIM Fashion Mensal",
|
|
"claimed_amount": "10",
|
|
"validated_amount": "10",
|
|
}
|
|
]
|
|
invoice = {
|
|
"invoiceVariation": [
|
|
{
|
|
"desc": "TIM Fashion Mensal",
|
|
"value": "10.0",
|
|
}
|
|
],
|
|
"currentInvoice": [
|
|
{
|
|
"type": "servicos_contratados_de_parceiros",
|
|
"desc": "Serviços de valor adicionado",
|
|
"value": "76.98",
|
|
"items": [
|
|
{
|
|
"type": "servicos_contratados_de_parceiros",
|
|
"desc": "TIM Fashion Mensal",
|
|
"value": "10.0",
|
|
"contestable": True,
|
|
}
|
|
],
|
|
}
|
|
],
|
|
}
|
|
|
|
_, validation_log, error = validate_contestation_items(items, invoice)
|
|
|
|
assert error is None
|
|
assert validation_log[-1]["status"] == "aprovado"
|
|
assert validation_log[-1]["secao_vas"] is True
|
|
assert validation_log[-1]["tipo_fatura"] == "servicos_contratados_de_parceiros"
|