"""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"