feat(tests): adicionar testes para ExcelService e PDFService
ExcelService (41 testes): - Geração de Excel com consultores - Parse de JSON, formatação de filtros - Extração de coordenações, consultoria, prêmios, titulação PDFService (44 testes): - Formatação de datas (completa e curta) - Ordenação por data com múltiplos formatos - Wrappers (ConsultorWrapper, DictWrapper) - Geração de ficha e PDF de equipe (mocked) Correções: - Ajuste nos testes Oracle para acessar params corretamente Cobertura: 54% → 66%
This commit is contained in:
184
backend/tests/infrastructure/oracle/test_ranking_repository.py
Normal file
184
backend/tests/infrastructure/oracle/test_ranking_repository.py
Normal file
@@ -0,0 +1,184 @@
|
||||
from datetime import datetime
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import pytest
|
||||
|
||||
from src.infrastructure.oracle.ranking_repository import RankingOracleRepository
|
||||
|
||||
|
||||
class DummyCLOB:
|
||||
def __init__(self, value: str) -> None:
|
||||
self._value = value
|
||||
|
||||
def read(self) -> str:
|
||||
return self._value
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def oracle_client():
|
||||
client = MagicMock()
|
||||
client.executar_query.return_value = []
|
||||
return client
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def repo(oracle_client):
|
||||
return RankingOracleRepository(oracle_client)
|
||||
|
||||
|
||||
def test_buscar_paginado_monta_params_e_parseia_json(repo, oracle_client):
|
||||
oracle_client.executar_query.return_value = [
|
||||
{
|
||||
"ID_PESSOA": 1,
|
||||
"NOME": "Teste",
|
||||
"POSICAO": 10,
|
||||
"PONTUACAO_TOTAL": 200,
|
||||
"COMPONENTE_A": 10,
|
||||
"COMPONENTE_B": 20,
|
||||
"COMPONENTE_C": 30,
|
||||
"COMPONENTE_D": 40,
|
||||
"COMPONENTE_E": 0,
|
||||
"ATIVO": "S",
|
||||
"ANOS_ATUACAO": 5,
|
||||
"DT_CALCULO": datetime(2025, 1, 1),
|
||||
"JSON_DETALHES": DummyCLOB("{\"k\": 1}"),
|
||||
}
|
||||
]
|
||||
|
||||
resultado = repo.buscar_paginado(page=2, size=20, filtro_ativo=True, filtro_selos=["CA", "EVENTO"])
|
||||
|
||||
assert len(resultado) == 1
|
||||
consultor = resultado[0]
|
||||
assert consultor.id_pessoa == 1
|
||||
assert consultor.ativo is True
|
||||
assert consultor.json_detalhes == "{\"k\": 1}"
|
||||
|
||||
call_args = oracle_client.executar_query.call_args
|
||||
params = call_args[0][1]
|
||||
assert params["offset"] == 20
|
||||
assert params["limit_end"] == 40
|
||||
assert params["ativo"] == "S"
|
||||
assert params["selo_0"] == "CA"
|
||||
assert params["selo_1"] == "EVENTO"
|
||||
|
||||
|
||||
def test_contar_total_com_filtros(repo, oracle_client):
|
||||
oracle_client.executar_query.return_value = [{"TOTAL": 123}]
|
||||
total = repo.contar_total(filtro_ativo=False, filtro_selos=["CONS_ATIVO"])
|
||||
assert total == 123
|
||||
call_args = oracle_client.executar_query.call_args
|
||||
params = call_args[0][1]
|
||||
assert params["ativo"] == "N"
|
||||
assert params["selo_0"] == "CONS_ATIVO"
|
||||
|
||||
|
||||
def test_buscar_por_nome_monta_query(repo, oracle_client):
|
||||
oracle_client.executar_query.return_value = [{"ID_PESSOA": 1, "NOME": "JOAO", "POSICAO": 1, "PONTUACAO_TOTAL": 10}]
|
||||
resultado = repo.buscar_por_nome("Joao Silva", limit=3)
|
||||
assert resultado[0]["ID_PESSOA"] == 1
|
||||
call_args = oracle_client.executar_query.call_args
|
||||
params = call_args[0][1]
|
||||
assert params["limit"] == 3
|
||||
assert params["p0"] == "%JOAO%"
|
||||
assert params["p1"] == "%SILVA%"
|
||||
|
||||
|
||||
def test_buscar_por_id_parseia_json(repo, oracle_client):
|
||||
oracle_client.executar_query.return_value = [
|
||||
{
|
||||
"ID_PESSOA": 5,
|
||||
"NOME": "Teste",
|
||||
"POSICAO": 20,
|
||||
"PONTUACAO_TOTAL": 300,
|
||||
"COMPONENTE_A": 10,
|
||||
"COMPONENTE_B": 20,
|
||||
"COMPONENTE_C": 30,
|
||||
"COMPONENTE_D": 40,
|
||||
"COMPONENTE_E": 0,
|
||||
"ATIVO": "N",
|
||||
"ANOS_ATUACAO": 2,
|
||||
"DT_CALCULO": datetime(2024, 12, 31),
|
||||
"JSON_DETALHES": DummyCLOB("{\"x\": true}"),
|
||||
}
|
||||
]
|
||||
|
||||
consultor = repo.buscar_por_id(5)
|
||||
assert consultor is not None
|
||||
assert consultor.id_pessoa == 5
|
||||
assert consultor.ativo is False
|
||||
assert consultor.json_detalhes == "{\"x\": true}"
|
||||
|
||||
|
||||
def test_atualizar_posicoes_chama_procedure(repo, oracle_client):
|
||||
cursor = MagicMock()
|
||||
conn = MagicMock()
|
||||
conn.cursor.return_value = cursor
|
||||
oracle_client.get_connection.return_value.__enter__.return_value = conn
|
||||
|
||||
repo.atualizar_posicoes()
|
||||
|
||||
cursor.callproc.assert_called_once_with("SP_ATUALIZAR_POSICOES")
|
||||
conn.commit.assert_called_once()
|
||||
|
||||
|
||||
def test_obter_estatisticas(repo, oracle_client):
|
||||
oracle_client.executar_query.return_value = [
|
||||
{
|
||||
"TOTAL_CONSULTORES": 10,
|
||||
"TOTAL_ATIVOS": 7,
|
||||
"TOTAL_INATIVOS": 3,
|
||||
"ULTIMA_ATUALIZACAO": datetime(2025, 1, 2, 10, 0, 0),
|
||||
"PONTUACAO_MEDIA": 100,
|
||||
"PONTUACAO_MAXIMA": 200,
|
||||
"PONTUACAO_MINIMA": 10,
|
||||
"MEDIA_COMP_A": 10,
|
||||
"MEDIA_COMP_B": 20,
|
||||
"MEDIA_COMP_C": 30,
|
||||
"MEDIA_COMP_D": 40,
|
||||
"MEDIA_COMP_E": 0,
|
||||
}
|
||||
]
|
||||
|
||||
stats = repo.obter_estatisticas()
|
||||
assert stats["total_consultores"] == 10
|
||||
assert stats["total_ativos"] == 7
|
||||
assert stats["ultima_atualizacao"].startswith("2025-01-02")
|
||||
assert stats["media_componentes"]["a"] == 10
|
||||
|
||||
|
||||
def test_obter_distribuicao(repo, oracle_client):
|
||||
oracle_client.executar_query.return_value = [
|
||||
{"FAIXA": "0-199", "QUANTIDADE": 5, "PERCENTUAL": 50},
|
||||
{"FAIXA": "200-399", "QUANTIDADE": 5, "PERCENTUAL": 50},
|
||||
]
|
||||
dist = repo.obter_distribuicao()
|
||||
assert dist[0]["faixa"] == "0-199"
|
||||
assert dist[1]["percentual"] == 50.0
|
||||
|
||||
|
||||
def test_buscar_para_exportacao_usa_cursor(repo, oracle_client):
|
||||
cursor = MagicMock()
|
||||
cursor.description = [("ID_PESSOA",), ("NOME",), ("JSON_DETALHES",)]
|
||||
cursor.fetchmany.side_effect = [
|
||||
[(1, "Teste", DummyCLOB("{\"k\":1}"))],
|
||||
[],
|
||||
]
|
||||
conn = MagicMock()
|
||||
conn.cursor.return_value = cursor
|
||||
oracle_client.get_connection.return_value.__enter__.return_value = conn
|
||||
|
||||
registros = repo.buscar_para_exportacao(filtro_ativo=True, filtro_selos=["CA"], batch_size=1)
|
||||
assert registros[0]["ID_PESSOA"] == 1
|
||||
assert registros[0]["JSON_DETALHES"] == "{\"k\":1}"
|
||||
|
||||
|
||||
def test_contar_para_exportacao_com_filtros(repo, oracle_client):
|
||||
oracle_client.executar_query.return_value = [{"TOTAL": 42}]
|
||||
total = repo.contar_para_exportacao(filtro_ativo=True, filtro_selos=["EVENTO", "PROJ"])
|
||||
assert total == 42
|
||||
call_args = oracle_client.executar_query.call_args
|
||||
params = call_args[0][1]
|
||||
assert params["ativo"] == "S"
|
||||
assert params["selo_0"] == "EVENTO"
|
||||
assert params["selo_1"] == "PROJ"
|
||||
Reference in New Issue
Block a user