mirror of
https://github.com/hoshikawa2/compass_backoffice.git
synced 2026-07-09 13:54:20 +00:00
121 lines
4.4 KiB
Python
121 lines
4.4 KiB
Python
import pytest
|
|
from datetime import datetime
|
|
from src.agent.logic.validation import validate_required_fields
|
|
from src.api.schemas.anatel_schemas import (
|
|
DataSourceEvent, ReasonCode
|
|
)
|
|
|
|
@pytest.fixture
|
|
def valid_ticket_payload():
|
|
return {
|
|
"source_system": "test",
|
|
"ingested_at": datetime.now(),
|
|
"correlation_id": "123",
|
|
"data_source_ticket": {
|
|
"data_source_ticket_id": "TICKET-123",
|
|
"status": "OPEN",
|
|
"action_type": "TEST",
|
|
"area_fechamento": "AREA",
|
|
"site_fechamento": "SITE",
|
|
"grupo_skill_responsavel": "SKILL",
|
|
"canal_origem": "WEB"
|
|
},
|
|
"anatel": {
|
|
"anatel_complaint_id": "COMP-1",
|
|
"protocol": "PROT-123",
|
|
"opened_at": datetime.now(),
|
|
"due_at": datetime.now(),
|
|
"title": "Title",
|
|
"motive": "Motive",
|
|
"submotive": "Submotive",
|
|
"description": "Desc",
|
|
"service": "Service",
|
|
"first_service": "FS",
|
|
"modality": "Modality"
|
|
},
|
|
"crm": {
|
|
"siebel_sr_number": "SR-1",
|
|
"customer_segment": "SEG",
|
|
"odc_customer": False,
|
|
"contumaz_customer": False
|
|
},
|
|
"customer": {
|
|
"customer_name": "Test Name",
|
|
"cpf_cnpj": "12345678909",
|
|
"phones": ["11999999999"],
|
|
"address": {
|
|
"cep": "00000000",
|
|
"street": "Rua",
|
|
"number": "1",
|
|
"neighborhood": "Bairro",
|
|
"city": "City",
|
|
"state": "SP"
|
|
}
|
|
},
|
|
"routing_inputs": {
|
|
"acionamentos_count": 0,
|
|
"routing_group_skill": "G",
|
|
"routing_modalidade": "M",
|
|
"routing_priority": "P",
|
|
"speech_expected": True
|
|
},
|
|
"attachments": {"consumer_attachments": [], "data_source_attachments": []},
|
|
"subscriber": {
|
|
"cpf_cnpj": "12345678909",
|
|
"customer_name": "Subscriber Name",
|
|
"contact_phone": "11999999999",
|
|
"contact_name": "Contact Name"
|
|
},
|
|
"history_ids": [],
|
|
"raw_print_fields": {}
|
|
}
|
|
|
|
def test_validate_required_fields_success(valid_ticket_payload):
|
|
ticket = DataSourceEvent(**valid_ticket_payload)
|
|
result = validate_required_fields(ticket)
|
|
assert result.is_valid is True
|
|
assert result.error_response is None
|
|
|
|
def test_validate_missing_cpf(valid_ticket_payload):
|
|
valid_ticket_payload["customer"]["cpf_cnpj"] = ""
|
|
valid_ticket_payload["subscriber"]["cpf_cnpj"] = ""
|
|
ticket = DataSourceEvent(**valid_ticket_payload)
|
|
result = validate_required_fields(ticket)
|
|
|
|
assert result.is_valid is False
|
|
assert result.error_response["status"] == 400
|
|
messages = result.error_response["detail"]["messages"]
|
|
assert any(m["code"] == ReasonCode.MISSING_CPF.value for m in messages)
|
|
|
|
def test_validate_missing_multiple_fields(valid_ticket_payload):
|
|
valid_ticket_payload["customer"]["cpf_cnpj"] = ""
|
|
valid_ticket_payload["subscriber"]["cpf_cnpj"] = ""
|
|
valid_ticket_payload["anatel"]["title"] = ""
|
|
valid_ticket_payload["data_source_ticket"]["data_source_ticket_id"] = ""
|
|
|
|
ticket = DataSourceEvent(**valid_ticket_payload)
|
|
result = validate_required_fields(ticket)
|
|
|
|
assert result.is_valid is False
|
|
messages = result.error_response["detail"]["messages"]
|
|
codes = [m["code"] for m in messages]
|
|
|
|
assert ReasonCode.MISSING_CPF.value in codes
|
|
assert ReasonCode.MISSING_TITLE.value in codes
|
|
assert ReasonCode.MISSING_IDS.value in codes
|
|
|
|
|
|
def test_validate_missing_anatel_motive(valid_ticket_payload):
|
|
"""Valida rejeição por falta de motivo da Anatel."""
|
|
valid_ticket_payload["anatel"]["motive"] = " "
|
|
ticket = DataSourceEvent(**valid_ticket_payload)
|
|
result = validate_required_fields(ticket)
|
|
assert result.error_response["detail"]["messages"][0]["code"] == ReasonCode.MISSING_MOTIVE.value
|
|
|
|
# def test_validate_missing_opened_at(valid_ticket_payload):
|
|
# """Valida rejeição por falta de data de abertura."""
|
|
# valid_ticket_payload["anatel"]["opened_at"] = None
|
|
# ticket = DataSourceEvent(**valid_ticket_payload)
|
|
# result = validate_required_fields(ticket)
|
|
# assert result.error_response["detail"]["messages"][0]["code"] == ReasonCode.MISSING_OPENED_AT.value
|