mirror of
https://github.com/hoshikawa2/compass_backoffice.git
synced 2026-07-09 13:54:20 +00:00
353 lines
13 KiB
Python
353 lines
13 KiB
Python
"""Unit tests for identity_verification_node."""
|
|
|
|
import pytest
|
|
|
|
from src.agent.nodes.identity_verification_node import (
|
|
IDENTITY_DECISION_CANCEL,
|
|
IDENTITY_DECISION_PROCEED,
|
|
IDENTITY_DECISION_SMART_HUMAN,
|
|
_authorized_via_attachment,
|
|
_authorized_via_gov_br_seal,
|
|
_is_third_party_complaint,
|
|
_normalize_cpf,
|
|
perform_identity_verification,
|
|
route_after_identity_verification,
|
|
)
|
|
from src.utils.decision_helpers import CANCEL_REASON_UNAUTHORIZED_THIRD_PARTY
|
|
|
|
CANCELATION_REASON_1 = "Processo Interno"
|
|
CANCELATION_REASON_2 = "Atendimento"
|
|
CANCELATION_REASON_3 = "Cancelamento ID Anatel"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
TITULAR_CPF = "12345678901"
|
|
TERCEIRO_CPF = "99988877766"
|
|
ASSINANTE_CPF = "11122233344"
|
|
|
|
|
|
def _make_state(
|
|
customer_cpf=TITULAR_CPF,
|
|
subscriber_cpf=TITULAR_CPF,
|
|
imdb_cpf=TITULAR_CPF,
|
|
imdb_status=200,
|
|
gov_br_seal=None,
|
|
has_attachment=None,
|
|
omit_context=False,
|
|
):
|
|
"""Build a minimal AgentState driving the identity_verification node."""
|
|
customer = {
|
|
"cpfCnpj": customer_cpf,
|
|
"subscriber": {"cpfCnpj": subscriber_cpf} if subscriber_cpf is not None else {},
|
|
}
|
|
if gov_br_seal is not None:
|
|
customer["govBrSeal"] = gov_br_seal
|
|
if has_attachment is not None:
|
|
customer["hasAttachment"] = has_attachment
|
|
|
|
request_context = {
|
|
"customer": customer,
|
|
# `cpf_cnpj` continua sendo o eco do customer.cpfCnpj (consumido pelo
|
|
# cache_check para invalidação). A comparação real do titular usa
|
|
# `socialSecNo` (campo retornado pela IMDB).
|
|
"imdb_access_data": {
|
|
"status_code": imdb_status,
|
|
"cpf_cnpj": customer_cpf,
|
|
"socialSecNo": imdb_cpf,
|
|
},
|
|
}
|
|
|
|
metadata = {} if omit_context else {"request_context": request_context}
|
|
return {
|
|
"messages": [],
|
|
"current_step": "imdb_enrichment_completed",
|
|
"iteration_count": 0,
|
|
"tool_results": None,
|
|
"final_response": None,
|
|
"session_id": "sess-idv-test",
|
|
"metadata": metadata,
|
|
"error": None,
|
|
}
|
|
|
|
|
|
def _idv(state):
|
|
"""Shortcut: read identity_verification snapshot from state."""
|
|
return state["metadata"]["request_context"]["identity_verification"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 1. Tabela de decisão — os 4 cenários principais
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.asyncio
|
|
class TestDecisionTable:
|
|
|
|
async def test_titular_proceeds(self):
|
|
"""Customer == Subscriber == IMDB → proceed (fluxo normal)."""
|
|
state = await perform_identity_verification(_make_state())
|
|
|
|
snap = _idv(state)
|
|
assert snap["decision"] == IDENTITY_DECISION_PROCEED
|
|
assert snap["third_party_complaint"] is False
|
|
assert snap["divergence_source"] is None
|
|
assert state["current_step"] == "identity_verification"
|
|
# PROCEED não deve ativar smart human nem cancelamento
|
|
ctx = state["metadata"]["request_context"]
|
|
assert "smart_human_handling" not in ctx
|
|
assert ctx.get("siebel_action") != "cancelar"
|
|
|
|
async def test_terceiro_com_gov_br_vai_para_smart_human(self):
|
|
"""Customer ≠ Subscriber + Selo GOV BR autenticado → smart_human."""
|
|
state = await perform_identity_verification(_make_state(
|
|
customer_cpf=TERCEIRO_CPF,
|
|
subscriber_cpf=ASSINANTE_CPF,
|
|
imdb_cpf=ASSINANTE_CPF,
|
|
gov_br_seal=True,
|
|
))
|
|
|
|
snap = _idv(state)
|
|
assert snap["decision"] == IDENTITY_DECISION_SMART_HUMAN
|
|
assert snap["third_party_complaint"] is True
|
|
assert snap["divergence_source"] == "subscriber"
|
|
assert snap["authorized_via_gov_br_seal"] is True
|
|
|
|
ctx = state["metadata"]["request_context"]
|
|
assert ctx["smart_human_handling"] is True
|
|
assert "Selo GOV BR" in ctx["smart_human_reason"]
|
|
assert ctx.get("siebel_action") != "cancelar"
|
|
|
|
async def test_terceiro_com_anexo_vai_para_smart_human(self):
|
|
"""Customer ≠ Subscriber + sem selo + anexo presente → smart_human."""
|
|
state = await perform_identity_verification(_make_state(
|
|
customer_cpf=TERCEIRO_CPF,
|
|
subscriber_cpf=ASSINANTE_CPF,
|
|
imdb_cpf=ASSINANTE_CPF,
|
|
gov_br_seal=False,
|
|
has_attachment=True,
|
|
))
|
|
|
|
snap = _idv(state)
|
|
assert snap["decision"] == IDENTITY_DECISION_SMART_HUMAN
|
|
assert snap["third_party_complaint"] is True
|
|
assert snap["authorized_via_gov_br_seal"] is False
|
|
assert snap["authorized_via_attachment"] is True
|
|
|
|
ctx = state["metadata"]["request_context"]
|
|
assert ctx["smart_human_handling"] is True
|
|
assert "anexo" in ctx["smart_human_reason"].lower()
|
|
|
|
async def test_terceiro_nao_autorizado_e_cancelado(self):
|
|
"""Customer ≠ Subscriber + sem selo + sem anexo → cancel + triplet Siebel."""
|
|
state = await perform_identity_verification(_make_state(
|
|
customer_cpf=TERCEIRO_CPF,
|
|
subscriber_cpf=ASSINANTE_CPF,
|
|
imdb_cpf=ASSINANTE_CPF,
|
|
gov_br_seal=False,
|
|
has_attachment=False,
|
|
))
|
|
|
|
snap = _idv(state)
|
|
assert snap["decision"] == IDENTITY_DECISION_CANCEL
|
|
assert snap["third_party_complaint"] is True
|
|
assert snap["authorized_via_gov_br_seal"] is False
|
|
assert snap["authorized_via_attachment"] is False
|
|
|
|
ctx = state["metadata"]["request_context"]
|
|
assert ctx["siebel_action"] == "cancelar"
|
|
assert ctx["reason1"] == CANCELATION_REASON_1
|
|
assert ctx["reason2"] == CANCELATION_REASON_2
|
|
assert ctx["reason3"] == CANCELATION_REASON_3
|
|
assert ctx["cancel_reason"] == CANCEL_REASON_UNAUTHORIZED_THIRD_PARTY
|
|
assert ctx["canceling_reasoning"] == snap["reasoning"]
|
|
# identity_verification cancela inline e move o step para o ramo de cancelamento
|
|
assert state["current_step"] == "canceling_analysis_cancel_ticket"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 2. Detecção de "reclamação por terceiro"
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@pytest.mark.unit
|
|
class TestThirdPartyDetection:
|
|
|
|
def test_subscriber_divergence_takes_priority(self):
|
|
customer = {"cpfCnpj": TERCEIRO_CPF, "subscriber": {"cpfCnpj": ASSINANTE_CPF}}
|
|
imdb = {"status_code": 200, "socialSecNo": TITULAR_CPF} # divergente também, mas nem é alcançado
|
|
third, src = _is_third_party_complaint(customer, imdb)
|
|
assert third is True
|
|
assert src == "subscriber"
|
|
|
|
def test_imdb_divergence_when_subscriber_matches(self):
|
|
customer = {"cpfCnpj": TITULAR_CPF, "subscriber": {"cpfCnpj": TITULAR_CPF}}
|
|
imdb = {"status_code": 200, "socialSecNo": TERCEIRO_CPF}
|
|
third, src = _is_third_party_complaint(customer, imdb)
|
|
assert third is True
|
|
assert src == "imdb"
|
|
|
|
def test_imdb_cpf_echo_field_is_not_used_for_comparison(self):
|
|
"""`cpf_cnpj` é eco do customer e não deve disparar divergência."""
|
|
customer = {"cpfCnpj": TITULAR_CPF, "subscriber": {"cpfCnpj": TITULAR_CPF}}
|
|
imdb = {"status_code": 200, "cpf_cnpj": TERCEIRO_CPF, "socialSecNo": TITULAR_CPF}
|
|
third, _ = _is_third_party_complaint(customer, imdb)
|
|
assert third is False
|
|
|
|
def test_imdb_204_is_ignored(self):
|
|
"""IMDB 204 (no content) — comparison must be skipped per the rule."""
|
|
customer = {"cpfCnpj": TITULAR_CPF, "subscriber": {"cpfCnpj": TITULAR_CPF}}
|
|
imdb = {"status_code": 204, "socialSecNo": TERCEIRO_CPF}
|
|
third, src = _is_third_party_complaint(customer, imdb)
|
|
assert third is False
|
|
assert src is None
|
|
|
|
def test_imdb_200_without_cpf_is_ignored(self):
|
|
"""IMDB respondeu 200 mas sem socialSecNo → não comparar."""
|
|
customer = {"cpfCnpj": TITULAR_CPF, "subscriber": {"cpfCnpj": TITULAR_CPF}}
|
|
imdb = {"status_code": 200, "socialSecNo": None}
|
|
third, _ = _is_third_party_complaint(customer, imdb)
|
|
assert third is False
|
|
|
|
def test_non_200_status_is_ignored(self):
|
|
"""Qualquer status != 200 (timeout, 5xx) → comparação IMDB silenciada."""
|
|
customer = {"cpfCnpj": TITULAR_CPF, "subscriber": {"cpfCnpj": TITULAR_CPF}}
|
|
imdb = {"status_code": 500, "socialSecNo": TERCEIRO_CPF}
|
|
third, _ = _is_third_party_complaint(customer, imdb)
|
|
assert third is False
|
|
|
|
def test_formatted_cpfs_are_normalized(self):
|
|
"""CPFs com pontos/traços devem ser comparados pelos dígitos."""
|
|
customer = {
|
|
"cpfCnpj": "123.456.789-01",
|
|
"subscriber": {"cpfCnpj": "12345678901"},
|
|
}
|
|
imdb = {"status_code": 200, "socialSecNo": "123-456-789/01"}
|
|
third, _ = _is_third_party_complaint(customer, imdb)
|
|
assert third is False
|
|
|
|
def test_missing_customer_cpf_skips_comparison(self):
|
|
customer = {"cpfCnpj": None, "subscriber": {"cpfCnpj": TITULAR_CPF}}
|
|
imdb = {"status_code": 200, "socialSecNo": TERCEIRO_CPF}
|
|
third, _ = _is_third_party_complaint(customer, imdb)
|
|
assert third is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 3. Selo GOV BR e Anexo
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@pytest.mark.unit
|
|
class TestGovBrSeal:
|
|
|
|
@pytest.mark.parametrize("value, expected", [
|
|
(True, True),
|
|
(False, False),
|
|
(None, False),
|
|
("", False),
|
|
(" ", False), # whitespace só
|
|
("autenticado", True),
|
|
])
|
|
def test_seal_variants(self, value, expected):
|
|
assert _authorized_via_gov_br_seal({"govBrSeal": value}) is expected
|
|
|
|
def test_seal_missing_key(self):
|
|
assert _authorized_via_gov_br_seal({}) is False
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestAttachmentAuth:
|
|
|
|
@pytest.mark.parametrize("value, expected", [
|
|
(True, True),
|
|
(False, False),
|
|
(None, False),
|
|
("true", False), # apenas bool literal True conta
|
|
(1, False),
|
|
])
|
|
def test_attachment_variants(self, value, expected):
|
|
assert _authorized_via_attachment({"hasAttachment": value}) is expected
|
|
|
|
def test_attachment_missing_key(self):
|
|
assert _authorized_via_attachment({}) is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 4. Normalização de CPF
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@pytest.mark.unit
|
|
class TestNormalizeCpf:
|
|
|
|
@pytest.mark.parametrize("raw, expected", [
|
|
("12345678901", "12345678901"),
|
|
("123.456.789-01", "12345678901"),
|
|
(" 123 456 789 01 ", "12345678901"),
|
|
("", None),
|
|
("--..", None),
|
|
(None, None),
|
|
])
|
|
def test_normalize(self, raw, expected):
|
|
assert _normalize_cpf(raw) == expected
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 5. Erros e contratos do node
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.asyncio
|
|
class TestNodeContract:
|
|
|
|
async def test_missing_request_context_sets_validation_error(self):
|
|
state = await perform_identity_verification(_make_state(omit_context=True))
|
|
assert state["error"]["type"] == "ValidationError"
|
|
assert state["error"]["step"] == "identity_verification"
|
|
|
|
async def test_state_snapshot_always_populated(self):
|
|
"""Toda decisão bem-sucedida grava um snapshot completo."""
|
|
state = await perform_identity_verification(_make_state(
|
|
customer_cpf=TERCEIRO_CPF,
|
|
subscriber_cpf=ASSINANTE_CPF,
|
|
gov_br_seal=True,
|
|
))
|
|
snap = _idv(state)
|
|
for key in (
|
|
"decision",
|
|
"third_party_complaint",
|
|
"divergence_source",
|
|
"authorized_via_gov_br_seal",
|
|
"authorized_via_attachment",
|
|
"reasoning",
|
|
):
|
|
assert key in snap
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 6. Router (mapeamento decisão → próximo node)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@pytest.mark.unit
|
|
class TestRouter:
|
|
|
|
@pytest.mark.parametrize("decision, expected_route", [
|
|
(IDENTITY_DECISION_PROCEED, "proceed"),
|
|
(IDENTITY_DECISION_SMART_HUMAN, "smart_human"),
|
|
(IDENTITY_DECISION_CANCEL, "cancel"),
|
|
("anything_else", "failed"),
|
|
(None, "failed"),
|
|
])
|
|
def test_routing(self, decision, expected_route):
|
|
state = {
|
|
"metadata": {
|
|
"request_context": {"identity_verification": {"decision": decision}}
|
|
}
|
|
}
|
|
assert route_after_identity_verification(state) == expected_route
|
|
|
|
def test_routing_missing_identity_section(self):
|
|
"""Sem snapshot de identity_verification → 'failed'."""
|
|
state = {"metadata": {"request_context": {}}}
|
|
assert route_after_identity_verification(state) == "failed"
|