diff --git a/backend/tests/application/services/__init__.py b/backend/tests/application/services/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/backend/tests/application/services/test_excel_service.py b/backend/tests/application/services/test_excel_service.py new file mode 100644 index 0000000..08cec78 --- /dev/null +++ b/backend/tests/application/services/test_excel_service.py @@ -0,0 +1,384 @@ +import pytest +from io import BytesIO +import json + +from openpyxl import load_workbook + +from src.application.services.excel_service import ExcelService + + +@pytest.fixture +def excel_service(): + return ExcelService() + + +@pytest.fixture +def consultor_basico(): + return { + "POSICAO": 1, + "ID_PESSOA": 12345, + "NOME": "MARIA SILVA SANTOS", + "PONTUACAO_TOTAL": 450.0, + "COMPONENTE_A": 200.0, + "COMPONENTE_C": 150.0, + "COMPONENTE_D": 50.0, + "COMPONENTE_E": 50.0, + "ATIVO": "S", + "ANOS_ATUACAO": 10.5, + "SELOS": "CA,CONS_ATIVO,AUTOR_GP", + "JSON_DETALHES": json.dumps({ + "coordenacoes_capes": [ + {"tipo": "CA", "area": "CIÊNCIAS DA COMPUTAÇÃO"} + ], + "consultoria": {"situacao": "Atividade Contínua"}, + "premiacoes": [ + {"premio": "PCT", "tipo": "Grande Prêmio", "ano": 2023} + ], + "titulacao": "Doutorado - USP (2010)" + }) + } + + +@pytest.fixture +def consultor_inativo(): + return { + "POSICAO": 100, + "ID_PESSOA": 99999, + "NOME": "JOÃO PEREIRA", + "PONTUACAO_TOTAL": 100.0, + "COMPONENTE_A": 0.0, + "COMPONENTE_C": 100.0, + "COMPONENTE_D": 0.0, + "COMPONENTE_E": 0.0, + "ATIVO": "N", + "ANOS_ATUACAO": 3.0, + "SELOS": "", + "JSON_DETALHES": None + } + + +class TestGerarRankingExcel: + + def test_gerar_excel_lista_vazia(self, excel_service): + resultado = excel_service.gerar_ranking_excel([]) + + assert isinstance(resultado, bytes) + assert len(resultado) > 0 + + wb = load_workbook(BytesIO(resultado)) + ws = wb.active + assert ws.title == "Ranking Consultores" + + def test_gerar_excel_um_consultor(self, excel_service, consultor_basico): + resultado = excel_service.gerar_ranking_excel([consultor_basico]) + + wb = load_workbook(BytesIO(resultado)) + ws = wb.active + + assert ws["A1"].value == "RANKING DE CONSULTORES CAPES" + assert ws["A5"].value == 1 + assert ws["B5"].value == 12345 + assert ws["C5"].value == "MARIA SILVA SANTOS" + assert ws["D5"].value == 450.0 + + def test_gerar_excel_multiplos_consultores(self, excel_service, consultor_basico, consultor_inativo): + consultores = [consultor_basico, consultor_inativo] + resultado = excel_service.gerar_ranking_excel(consultores) + + wb = load_workbook(BytesIO(resultado)) + ws = wb.active + + assert ws["C5"].value == "MARIA SILVA SANTOS" + assert ws["C6"].value == "JOÃO PEREIRA" + + def test_gerar_excel_com_filtros(self, excel_service, consultor_basico): + filtros = {"ativo": True, "selos": ["CA", "CONS_ATIVO"]} + resultado = excel_service.gerar_ranking_excel([consultor_basico], filtros_aplicados=filtros) + + wb = load_workbook(BytesIO(resultado)) + ws = wb.active + + subtitulo = ws["A2"].value + assert "Ativos" in subtitulo + assert "CA" in subtitulo + + def test_gerar_excel_consultor_ativo(self, excel_service, consultor_basico): + resultado = excel_service.gerar_ranking_excel([consultor_basico]) + + wb = load_workbook(BytesIO(resultado)) + ws = wb.active + + assert ws["I5"].value == "Ativo" + + def test_gerar_excel_consultor_inativo(self, excel_service, consultor_inativo): + resultado = excel_service.gerar_ranking_excel([consultor_inativo]) + + wb = load_workbook(BytesIO(resultado)) + ws = wb.active + + assert ws["I5"].value == "Inativo" + + def test_gerar_excel_extrai_coordenacoes(self, excel_service, consultor_basico): + resultado = excel_service.gerar_ranking_excel([consultor_basico]) + + wb = load_workbook(BytesIO(resultado)) + ws = wb.active + + coord_cell = ws["L5"].value + assert "CA" in coord_cell + assert "CIÊNCIAS DA COMPUTAÇÃO" in coord_cell + + def test_gerar_excel_extrai_consultoria(self, excel_service, consultor_basico): + resultado = excel_service.gerar_ranking_excel([consultor_basico]) + + wb = load_workbook(BytesIO(resultado)) + ws = wb.active + + assert ws["M5"].value == "Atividade Contínua" + + def test_gerar_excel_extrai_premios(self, excel_service, consultor_basico): + resultado = excel_service.gerar_ranking_excel([consultor_basico]) + + wb = load_workbook(BytesIO(resultado)) + ws = wb.active + + premios_cell = ws["N5"].value + assert "PCT" in premios_cell + assert "2023" in premios_cell + + def test_gerar_excel_extrai_titulacao(self, excel_service, consultor_basico): + resultado = excel_service.gerar_ranking_excel([consultor_basico]) + + wb = load_workbook(BytesIO(resultado)) + ws = wb.active + + assert "Doutorado" in ws["O5"].value + + def test_gerar_excel_selos_formatados(self, excel_service, consultor_basico): + resultado = excel_service.gerar_ranking_excel([consultor_basico]) + + wb = load_workbook(BytesIO(resultado)) + ws = wb.active + + selos_cell = ws["K5"].value + assert "CA" in selos_cell + assert ", " in selos_cell + + +class TestParseJsonDetalhes: + + def test_parse_json_string_valido(self, excel_service): + json_str = '{"coordenacoes_capes": [{"tipo": "CA"}]}' + resultado = excel_service._parse_json_detalhes(json_str) + + assert resultado == {"coordenacoes_capes": [{"tipo": "CA"}]} + + def test_parse_json_none(self, excel_service): + resultado = excel_service._parse_json_detalhes(None) + assert resultado == {} + + def test_parse_json_string_vazia(self, excel_service): + resultado = excel_service._parse_json_detalhes("") + assert resultado == {} + + def test_parse_json_invalido(self, excel_service): + resultado = excel_service._parse_json_detalhes("{invalid json") + assert resultado == {} + + def test_parse_json_dict_direto(self, excel_service): + dados = {"coordenacoes_capes": []} + resultado = excel_service._parse_json_detalhes(dados) + assert resultado == dados + + def test_parse_json_file_like_object(self, excel_service): + class MockLob: + def read(self): + return '{"test": "value"}' + + resultado = excel_service._parse_json_detalhes(MockLob()) + assert resultado == {"test": "value"} + + +class TestFormatarFiltros: + + def test_formatar_filtros_ativo_true(self, excel_service): + filtros = {"ativo": True} + resultado = excel_service._formatar_filtros(filtros) + assert resultado == "Ativos" + + def test_formatar_filtros_ativo_false(self, excel_service): + filtros = {"ativo": False} + resultado = excel_service._formatar_filtros(filtros) + assert resultado == "Inativos" + + def test_formatar_filtros_com_selos(self, excel_service): + filtros = {"selos": ["CA", "CAJ"]} + resultado = excel_service._formatar_filtros(filtros) + assert "Selos: CA, CAJ" in resultado + + def test_formatar_filtros_combinados(self, excel_service): + filtros = {"ativo": True, "selos": ["CA"]} + resultado = excel_service._formatar_filtros(filtros) + assert "Ativos" in resultado + assert "Selos: CA" in resultado + + def test_formatar_filtros_vazio(self, excel_service): + filtros = {} + resultado = excel_service._formatar_filtros(filtros) + assert resultado == "" + + +class TestExtrairCoordenacoesResumo: + + def test_extrair_coordenacoes_vazio(self, excel_service): + resultado = excel_service._extrair_coordenacoes_resumo({}) + assert resultado == "" + + def test_extrair_coordenacoes_lista_vazia(self, excel_service): + resultado = excel_service._extrair_coordenacoes_resumo({"coordenacoes_capes": []}) + assert resultado == "" + + def test_extrair_coordenacoes_com_tipo_e_area(self, excel_service): + detalhes = { + "coordenacoes_capes": [ + {"tipo": "CA", "area": "MATEMÁTICA"} + ] + } + resultado = excel_service._extrair_coordenacoes_resumo(detalhes) + assert resultado == "CA: MATEMÁTICA" + + def test_extrair_coordenacoes_apenas_tipo(self, excel_service): + detalhes = { + "coordenacoes_capes": [ + {"tipo": "CAJ"} + ] + } + resultado = excel_service._extrair_coordenacoes_resumo(detalhes) + assert resultado == "CAJ" + + def test_extrair_coordenacoes_multiplas(self, excel_service): + detalhes = { + "coordenacoes_capes": [ + {"tipo": "CA", "area": "FÍSICA"}, + {"tipo": "CAJ", "area": "QUÍMICA"} + ] + } + resultado = excel_service._extrair_coordenacoes_resumo(detalhes) + assert "CA: FÍSICA" in resultado + assert "CAJ: QUÍMICA" in resultado + assert ";" in resultado + + def test_extrair_coordenacoes_maximo_tres(self, excel_service): + detalhes = { + "coordenacoes_capes": [ + {"tipo": "CA", "area": "A"}, + {"tipo": "CAJ", "area": "B"}, + {"tipo": "CAM", "area": "C"}, + {"tipo": "CAJ_MP", "area": "D"}, + ] + } + resultado = excel_service._extrair_coordenacoes_resumo(detalhes) + assert "CA: A" in resultado + assert "CAJ: B" in resultado + assert "CAM: C" in resultado + assert "CAJ_MP" not in resultado + + +class TestExtrairSituacaoConsultoria: + + def test_extrair_situacao_vazio(self, excel_service): + resultado = excel_service._extrair_situacao_consultoria({}) + assert resultado == "" + + def test_extrair_situacao_sem_consultoria(self, excel_service): + resultado = excel_service._extrair_situacao_consultoria({"consultoria": None}) + assert resultado == "" + + def test_extrair_situacao_ativa(self, excel_service): + detalhes = {"consultoria": {"situacao": "Atividade Contínua"}} + resultado = excel_service._extrair_situacao_consultoria(detalhes) + assert resultado == "Atividade Contínua" + + +class TestExtrairPremiosResumo: + + def test_extrair_premios_vazio(self, excel_service): + resultado = excel_service._extrair_premios_resumo({}) + assert resultado == "" + + def test_extrair_premios_lista_vazia(self, excel_service): + resultado = excel_service._extrair_premios_resumo({"premiacoes": []}) + assert resultado == "" + + def test_extrair_premio_completo(self, excel_service): + detalhes = { + "premiacoes": [ + {"premio": "PCT", "tipo": "Grande Prêmio", "ano": 2023} + ] + } + resultado = excel_service._extrair_premios_resumo(detalhes) + assert "PCT" in resultado + assert "Grande Prêmio" in resultado + assert "2023" in resultado + + def test_extrair_premio_apenas_nome(self, excel_service): + detalhes = { + "premiacoes": [{"premio": "PCT"}] + } + resultado = excel_service._extrair_premios_resumo(detalhes) + assert resultado == "PCT" + + def test_extrair_premios_mais_de_tres(self, excel_service): + detalhes = { + "premiacoes": [ + {"premio": "A"}, + {"premio": "B"}, + {"premio": "C"}, + {"premio": "D"}, + {"premio": "E"}, + ] + } + resultado = excel_service._extrair_premios_resumo(detalhes) + assert "A" in resultado + assert "B" in resultado + assert "C" in resultado + assert "+2 outros" in resultado + + +class TestExtrairTitulacao: + + def test_extrair_titulacao_vazio(self, excel_service): + resultado = excel_service._extrair_titulacao({}) + assert resultado == "" + + def test_extrair_titulacao_string_direta(self, excel_service): + detalhes = {"titulacao": "Doutorado - UNICAMP (2015)"} + resultado = excel_service._extrair_titulacao(detalhes) + assert resultado == "Doutorado - UNICAMP (2015)" + + def test_extrair_titulacao_do_lattes(self, excel_service): + detalhes = { + "lattes": { + "titulacoes": [ + {"grau": "Doutorado", "ies_sigla": "USP", "ano": 2010} + ] + } + } + resultado = excel_service._extrair_titulacao(detalhes) + assert "Doutorado" in resultado + assert "USP" in resultado + assert "2010" in resultado + + def test_extrair_titulacao_lattes_apenas_grau(self, excel_service): + detalhes = { + "lattes": { + "titulacoes": [{"grau": "Mestrado"}] + } + } + resultado = excel_service._extrair_titulacao(detalhes) + assert resultado == "Mestrado" + + def test_extrair_titulacao_lattes_vazio(self, excel_service): + detalhes = {"lattes": {"titulacoes": []}} + resultado = excel_service._extrair_titulacao(detalhes) + assert resultado == "" diff --git a/backend/tests/application/services/test_pdf_service.py b/backend/tests/application/services/test_pdf_service.py new file mode 100644 index 0000000..7c0d0c7 --- /dev/null +++ b/backend/tests/application/services/test_pdf_service.py @@ -0,0 +1,383 @@ +import pytest +from unittest.mock import MagicMock, patch +from datetime import datetime +from dataclasses import dataclass + +from src.application.services.pdf_service import PDFService, ConsultorWrapper, DictWrapper + + +class TestFormatDate: + + def test_format_date_vazio(self): + service = PDFService.__new__(PDFService) + resultado = service._format_date("") + assert resultado == "-" + + def test_format_date_none(self): + service = PDFService.__new__(PDFService) + resultado = service._format_date(None) + assert resultado == "-" + + def test_format_date_formato_brasileiro(self): + service = PDFService.__new__(PDFService) + resultado = service._format_date("25/12/2023") + assert resultado == "25/12/2023" + + def test_format_date_iso(self): + service = PDFService.__new__(PDFService) + resultado = service._format_date("2023-12-25T10:30:00") + assert resultado == "25/12/2023" + + def test_format_date_iso_com_z(self): + service = PDFService.__new__(PDFService) + resultado = service._format_date("2023-12-25T10:30:00Z") + assert resultado == "25/12/2023" + + def test_format_date_invalido(self): + service = PDFService.__new__(PDFService) + resultado = service._format_date("data-invalida") + assert resultado == "data-invalida" + + +class TestFormatDateShort: + + def test_format_date_short_vazio(self): + service = PDFService.__new__(PDFService) + resultado = service._format_date_short("") + assert resultado == "-" + + def test_format_date_short_none(self): + service = PDFService.__new__(PDFService) + resultado = service._format_date_short(None) + assert resultado == "-" + + def test_format_date_short_brasileiro_3_partes(self): + service = PDFService.__new__(PDFService) + resultado = service._format_date_short("25/12/2023") + assert resultado == "Dez/2023" + + def test_format_date_short_brasileiro_2_partes(self): + service = PDFService.__new__(PDFService) + resultado = service._format_date_short("12/2023") + assert resultado == "Dez/2023" + + def test_format_date_short_iso(self): + service = PDFService.__new__(PDFService) + resultado = service._format_date_short("2023-06-15T10:30:00") + assert resultado == "Jun/2023" + + def test_format_date_short_janeiro(self): + service = PDFService.__new__(PDFService) + resultado = service._format_date_short("15/01/2024") + assert resultado == "Jan/2024" + + +class TestSortByDate: + + def test_sort_by_date_lista_vazia(self): + service = PDFService.__new__(PDFService) + resultado = service._sort_by_date([], "data") + assert resultado == [] + + def test_sort_by_date_sem_campo(self): + service = PDFService.__new__(PDFService) + items = [{"nome": "A"}, {"nome": "B"}] + resultado = service._sort_by_date(items) + assert resultado == items + + def test_sort_by_date_ordena_desc(self): + service = PDFService.__new__(PDFService) + items = [ + {"nome": "A", "data": "2020-01-01"}, + {"nome": "B", "data": "2023-01-01"}, + {"nome": "C", "data": "2021-01-01"}, + ] + resultado = service._sort_by_date(items, "data") + assert resultado[0]["nome"] == "B" + assert resultado[1]["nome"] == "C" + assert resultado[2]["nome"] == "A" + + def test_sort_by_date_formato_brasileiro(self): + service = PDFService.__new__(PDFService) + items = [ + {"nome": "A", "data": "01/01/2020"}, + {"nome": "B", "data": "01/01/2023"}, + ] + resultado = service._sort_by_date(items, "data") + assert resultado[0]["nome"] == "B" + + def test_sort_by_date_com_ano_numerico(self): + service = PDFService.__new__(PDFService) + items = [ + {"nome": "A", "ano": 2020}, + {"nome": "B", "ano": 2023}, + ] + resultado = service._sort_by_date(items, "ano") + assert resultado[0]["nome"] == "B" + + def test_sort_by_date_com_ano_string(self): + service = PDFService.__new__(PDFService) + items = [ + {"nome": "A", "ano": "2020"}, + {"nome": "B", "ano": "2023"}, + ] + resultado = service._sort_by_date(items, "ano") + assert resultado[0]["nome"] == "B" + + def test_sort_by_date_valores_nulos_no_final(self): + service = PDFService.__new__(PDFService) + items = [ + {"nome": "A", "data": None}, + {"nome": "B", "data": "2023-01-01"}, + {"nome": "C", "data": ""}, + ] + resultado = service._sort_by_date(items, "data") + assert resultado[0]["nome"] == "B" + + def test_sort_by_date_fallback_campos(self): + service = PDFService.__new__(PDFService) + items = [ + {"nome": "A", "inicio": "2020-01-01"}, + {"nome": "B", "data": "2023-01-01"}, + ] + resultado = service._sort_by_date(items, "data", "inicio") + assert resultado[0]["nome"] == "B" + assert resultado[1]["nome"] == "A" + + +class TestConsultorToDict: + + def test_consultor_to_dict_dict(self): + service = PDFService.__new__(PDFService) + dados = {"nome": "Teste", "id": 123} + resultado = service._consultor_to_dict(dados) + assert resultado == dados + + def test_consultor_to_dict_dataclass(self): + @dataclass + class ConsultorTest: + nome: str + id: int + + service = PDFService.__new__(PDFService) + consultor = ConsultorTest(nome="Teste", id=123) + resultado = service._consultor_to_dict(consultor) + assert resultado == {"nome": "Teste", "id": 123} + + def test_consultor_to_dict_pydantic_model_dump(self): + service = PDFService.__new__(PDFService) + mock_obj = MagicMock() + mock_obj.model_dump.return_value = {"nome": "Teste"} + del mock_obj.dict + resultado = service._consultor_to_dict(mock_obj) + assert resultado == {"nome": "Teste"} + + def test_consultor_to_dict_pydantic_dict(self): + service = PDFService.__new__(PDFService) + mock_obj = MagicMock(spec=['dict']) + mock_obj.dict.return_value = {"nome": "Teste"} + resultado = service._consultor_to_dict(mock_obj) + assert resultado == {"nome": "Teste"} + + +class TestExtrairPontuacaoCoord: + + def test_extrair_pontuacao_coord_vazio(self): + service = PDFService.__new__(PDFService) + resultado = service._extrair_pontuacao_coord({}) + assert resultado == {} + + def test_extrair_pontuacao_coord_sem_pontuacao(self): + service = PDFService.__new__(PDFService) + resultado = service._extrair_pontuacao_coord({"nome": "Teste"}) + assert resultado == {} + + def test_extrair_pontuacao_coord_com_atuacoes(self): + service = PDFService.__new__(PDFService) + dados = { + "pontuacao": { + "bloco_a": { + "atuacoes": [ + {"codigo": "CA", "base": 200, "tempo": 50, "bonus": 30, "total": 280}, + {"codigo": "CAJ", "base": 150, "tempo": 40, "bonus": 20, "total": 210}, + ] + } + } + } + resultado = service._extrair_pontuacao_coord(dados) + assert "CA" in resultado + assert resultado["CA"]["base"] == 200 + assert resultado["CA"]["tempo"] == 50 + assert resultado["CA"]["total"] == 280 + assert "CAJ" in resultado + + +class TestConsultorWrapper: + + def test_wrapper_acesso_atributo_simples(self): + wrapper = ConsultorWrapper({"nome": "Teste", "id": 123}) + assert wrapper.nome == "Teste" + assert wrapper.id == 123 + + def test_wrapper_atributo_inexistente(self): + wrapper = ConsultorWrapper({"nome": "Teste"}) + assert wrapper.id is None + + def test_wrapper_atributo_dict_retorna_dictwrapper(self): + wrapper = ConsultorWrapper({ + "dados": {"campo": "valor"} + }) + assert isinstance(wrapper.dados, DictWrapper) + assert wrapper.dados.campo == "valor" + + def test_wrapper_atributo_lista_de_dicts(self): + wrapper = ConsultorWrapper({ + "itens": [{"nome": "A"}, {"nome": "B"}] + }) + assert len(wrapper.itens) == 2 + assert isinstance(wrapper.itens[0], DictWrapper) + assert wrapper.itens[0].nome == "A" + + def test_wrapper_bool_true(self): + wrapper = ConsultorWrapper({"nome": "Teste"}) + assert bool(wrapper) is True + + def test_wrapper_bool_false(self): + wrapper = ConsultorWrapper({}) + assert bool(wrapper) is False + + +class TestDictWrapper: + + def test_dictwrapper_acesso_atributo(self): + wrapper = DictWrapper({"campo": "valor"}) + assert wrapper.campo == "valor" + + def test_dictwrapper_atributo_inexistente(self): + wrapper = DictWrapper({"campo": "valor"}) + assert wrapper.outro is None + + def test_dictwrapper_aninhado(self): + wrapper = DictWrapper({ + "nivel1": {"nivel2": {"nivel3": "valor"}} + }) + assert wrapper.nivel1.nivel2.nivel3 == "valor" + + def test_dictwrapper_lista(self): + wrapper = DictWrapper({ + "itens": [{"a": 1}, {"b": 2}] + }) + assert len(wrapper.itens) == 2 + assert wrapper.itens[0].a == 1 + + def test_dictwrapper_get(self): + wrapper = DictWrapper({"campo": "valor"}) + assert wrapper.get("campo") == "valor" + assert wrapper.get("inexistente", "default") == "default" + + def test_dictwrapper_str(self): + wrapper = DictWrapper({"campo": "valor"}) + assert "campo" in str(wrapper) + + def test_dictwrapper_bool_true(self): + wrapper = DictWrapper({"campo": "valor"}) + assert bool(wrapper) is True + + def test_dictwrapper_bool_false(self): + wrapper = DictWrapper({}) + assert bool(wrapper) is False + + +class TestGerarFichaConsultor: + + def test_gerar_ficha_consultor_mock(self): + import sys + mock_weasyprint = MagicMock() + mock_weasyprint.HTML.return_value.write_pdf.return_value = b"PDF_CONTENT" + mock_weasyprint.CSS.return_value = MagicMock() + sys.modules['weasyprint'] = mock_weasyprint + + try: + with patch.object(PDFService, '__init__', lambda self: None): + service = PDFService() + service.template = MagicMock() + service.template.render.return_value = "" + from pathlib import Path + service.template_dir = Path("/tmp") + + consultor = {"nome": "Teste", "id": 123} + resultado = service.gerar_ficha_consultor(consultor) + + assert resultado == b"PDF_CONTENT" + service.template.render.assert_called_once() + finally: + del sys.modules['weasyprint'] + + +class TestGerarPdfEquipe: + + def test_gerar_pdf_equipe_mock(self): + import sys + mock_weasyprint = MagicMock() + mock_weasyprint.HTML.return_value.write_pdf.return_value = b"PDF_EQUIPE" + mock_weasyprint.CSS.return_value = MagicMock() + sys.modules['weasyprint'] = mock_weasyprint + + try: + with patch.object(PDFService, '__init__', lambda self: None): + service = PDFService() + service.env = MagicMock() + mock_template = MagicMock() + mock_template.render.return_value = "" + service.env.get_template.return_value = mock_template + from pathlib import Path + service.template_dir = Path("/tmp") + + consultores = [ + {"nome": "A", "ies": "USP", "areas_avaliacao": ["FÍSICA"], "situacao": "Atividade Contínua", "foi_coordenador": True, "foi_premiado": False}, + {"nome": "B", "ies": "UNICAMP", "areas_avaliacao": ["QUÍMICA"], "situacao": "Inativo", "foi_coordenador": False, "foi_premiado": True}, + ] + + resultado = service.gerar_pdf_equipe("Tema X", "FÍSICA", consultores) + + assert resultado == b"PDF_EQUIPE" + mock_template.render.assert_called_once() + call_kwargs = mock_template.render.call_args[1] + assert call_kwargs["tema"] == "Tema X" + assert call_kwargs["area_avaliacao"] == "FÍSICA" + assert call_kwargs["estatisticas"]["total"] == 2 + assert call_kwargs["estatisticas"]["coordenadores"] == 1 + assert call_kwargs["estatisticas"]["premiados"] == 1 + assert call_kwargs["estatisticas"]["ies_distintas"] == 2 + finally: + del sys.modules['weasyprint'] + + def test_estatisticas_equipe_ativos(self): + import sys + mock_weasyprint = MagicMock() + mock_weasyprint.HTML.return_value.write_pdf.return_value = b"PDF" + mock_weasyprint.CSS.return_value = MagicMock() + sys.modules['weasyprint'] = mock_weasyprint + + try: + with patch.object(PDFService, '__init__', lambda self: None): + service = PDFService() + service.env = MagicMock() + mock_template = MagicMock() + mock_template.render.return_value = "" + service.env.get_template.return_value = mock_template + from pathlib import Path + service.template_dir = Path("/tmp") + + consultores = [ + {"nome": "A", "situacao": "Atividade Contínua"}, + {"nome": "B", "situacao": "Atividade Contínua"}, + {"nome": "C", "situacao": "Inativo"}, + ] + + service.gerar_pdf_equipe("Tema", "ÁREA", consultores) + + call_kwargs = mock_template.render.call_args[1] + assert call_kwargs["estatisticas"]["ativos"] == 2 + finally: + del sys.modules['weasyprint'] diff --git a/backend/tests/infrastructure/oracle/test_ranking_repository.py b/backend/tests/infrastructure/oracle/test_ranking_repository.py new file mode 100644 index 0000000..f07f3e7 --- /dev/null +++ b/backend/tests/infrastructure/oracle/test_ranking_repository.py @@ -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"