From 094c8ed1ab3b4bea8dfa0284cdb3eae718d3f2bc Mon Sep 17 00:00:00 2001 From: BrzGab Date: Tue, 26 May 2026 00:14:53 -0300 Subject: [PATCH 1/2] testes: adiciona cobertura e docstrings --- api/db/registro.py | 26 +- tests/integration/test_academy.py | 1 + tests/integration/test_health.py | 1 + tests/integration/test_integracoes.py | 2 + tests/integration/test_leads.py | 3 + tests/integration/test_servicos.py | 3 + tests/integration/test_users.py | 9 + tests/test_app.py | 2 + tests/unit/test_agenda_service.py | 1 + tests/unit/test_base_repository.py | 156 ++++ tests/unit/test_config.py | 46 ++ tests/unit/test_dashboard_service.py | 200 +++++ tests/unit/test_database.py | 37 + tests/unit/test_db_registro.py | 67 ++ tests/unit/test_google_calendar.py | 4 + tests/unit/test_migrations.py | 24 + tests/unit/test_password.py | 2 + tests/unit/test_schemas.py | 102 ++- tests/unit/test_security.py | 80 +- tests/unit/test_services.py | 1078 +++++++++++++++++++++++++ 20 files changed, 1827 insertions(+), 17 deletions(-) create mode 100644 tests/unit/test_base_repository.py create mode 100644 tests/unit/test_config.py create mode 100644 tests/unit/test_dashboard_service.py create mode 100644 tests/unit/test_database.py create mode 100644 tests/unit/test_db_registro.py create mode 100644 tests/unit/test_migrations.py create mode 100644 tests/unit/test_services.py diff --git a/api/db/registro.py b/api/db/registro.py index 75ba4fd..d5b4948 100644 --- a/api/db/registro.py +++ b/api/db/registro.py @@ -1,17 +1,17 @@ -from modules.academy import modelo as modelo_academia -from modules.agenda import modelo as modelo_agenda -from modules.comunicados import modelo as modelo_comunicados -from modules.gamificacao import modelo as modelo_gamificacao -from modules.integracoes import modelo as modelo_integracoes -from modules.leads import modelo as modelo_leads -from modules.lgpd import modelo as modelo_lgpd -from modules.portfolio import modelo as modelo_portfolio -from modules.projetos import modelo as modelo_projetos -from modules.servicos import modelo as modelo_servicos -from modules.tarefas import modelo as modelo_tarefas -from modules.users import modelo as modelo_usuarios +from modules.academy import model as modelo_academia +from modules.agenda import model as modelo_agenda +from modules.comunicados import model as modelo_comunicados +from modules.gamificacao import model as modelo_gamificacao +from modules.integracoes import model as modelo_integracoes +from modules.leads import model as modelo_leads +from modules.lgpd import model as modelo_lgpd +from modules.portfolio import model as modelo_portfolio +from modules.projetos import model as modelo_projetos +from modules.servicos import model as modelo_servicos +from modules.tarefas import model as modelo_tarefas +from modules.users import model as modelo_usuarios -# utilizado na env de migração do allembic +# utilizado na env de migração do Alembic TODOS_MODELOS = ( modelo_academia, modelo_agenda, diff --git a/tests/integration/test_academy.py b/tests/integration/test_academy.py index 015d630..4afd26e 100644 --- a/tests/integration/test_academy.py +++ b/tests/integration/test_academy.py @@ -23,6 +23,7 @@ def _dados(tipo: str) -> dict[str, object]: async def test_listar_academia_filtra_body_json( cliente_admin: AsyncClient, cliente_http: AsyncClient ): + """Valida que listar academia filtra body json.""" ebook = (await cliente_admin.post('/api/v1/academy', json=_dados('ebook'))).json() await cliente_admin.post('/api/v1/academy', json=_dados('curso')) diff --git a/tests/integration/test_health.py b/tests/integration/test_health.py index 07e66b7..ea600fa 100644 --- a/tests/integration/test_health.py +++ b/tests/integration/test_health.py @@ -7,6 +7,7 @@ @pytest.mark.asyncio @exige_banco async def test_saude_retorna_ok(cliente_http: AsyncClient): + """Valida que saude retorna ok.""" resposta = await cliente_http.get('/health/db') assert resposta.status_code == 200 assert resposta.json() == {'status': 'ok'} diff --git a/tests/integration/test_integracoes.py b/tests/integration/test_integracoes.py index 325e7dd..36ee93b 100644 --- a/tests/integration/test_integracoes.py +++ b/tests/integration/test_integracoes.py @@ -13,6 +13,7 @@ async def test_criar_integracao_status( cliente_admin: AsyncClient, situacao_entrada: str ): + """Valida que criar integracao status.""" dados = { 'nome': 'google_calendar', 'status': situacao_entrada, @@ -26,6 +27,7 @@ async def test_criar_integracao_status( @pytest.mark.asyncio @exige_banco async def test_listar_integracoes(cliente_admin: AsyncClient): + """Valida que listar integracoes.""" resposta = await cliente_admin.get('/api/v1/integracoes') assert resposta.status_code == 200 assert isinstance(resposta.json(), list) diff --git a/tests/integration/test_leads.py b/tests/integration/test_leads.py index 10b2f33..779077d 100644 --- a/tests/integration/test_leads.py +++ b/tests/integration/test_leads.py @@ -24,6 +24,7 @@ def _dados(**substituicoes): @exige_banco @pytest.mark.parametrize('prioridade', ['baixa', 'media', 'alta']) async def test_criar_lead_com_prioridades(cliente_admin: AsyncClient, prioridade: str): + """Valida que criar lead com prioridades.""" resposta = await cliente_admin.post( '/api/v1/leads', json=_dados(prioridade=prioridade) ) @@ -44,6 +45,7 @@ async def test_criar_lead_com_prioridades(cliente_admin: AsyncClient, prioridade async def test_atualizar_status_lead( cliente_admin: AsyncClient, situacao_inicial: str, nova_situacao: str ): + """Valida que atualizar status lead.""" criado = ( await cliente_admin.post('/api/v1/leads', json=_dados(status=situacao_inicial)) ).json() @@ -57,5 +59,6 @@ async def test_atualizar_status_lead( @pytest.mark.asyncio @exige_banco async def test_lead_nao_encontrado_retorna_404(cliente_admin: AsyncClient): + """Valida que lead nao encontrado retorna 404.""" resposta = await cliente_admin.get('/api/v1/leads/99999999') assert resposta.status_code == 404 diff --git a/tests/integration/test_servicos.py b/tests/integration/test_servicos.py index 9714f9b..5baa7bc 100644 --- a/tests/integration/test_servicos.py +++ b/tests/integration/test_servicos.py @@ -22,6 +22,7 @@ def _dados_servico(**substituicoes): @exige_banco @pytest.mark.parametrize('status', ['ativo', 'inativo']) async def test_criar_servico_com_status(cliente_admin: AsyncClient, status: str): + """Valida que criar servico com status.""" resposta = await cliente_admin.post( '/api/v1/servicos', json=_dados_servico(status=status) ) @@ -33,6 +34,7 @@ async def test_criar_servico_com_status(cliente_admin: AsyncClient, status: str) @pytest.mark.asyncio @exige_banco async def test_adicionar_listar_e_atualizar_entregavel(cliente_admin: AsyncClient): + """Valida que adicionar listar e atualizar entregavel.""" servico = ( await cliente_admin.post('/api/v1/servicos', json=_dados_servico()) ).json() @@ -59,5 +61,6 @@ async def test_adicionar_listar_e_atualizar_entregavel(cliente_admin: AsyncClien @pytest.mark.asyncio @exige_banco async def test_servico_nao_encontrado_retorna_404(cliente_admin: AsyncClient): + """Valida que servico nao encontrado retorna 404.""" resposta = await cliente_admin.get('/api/v1/servicos/99999999') assert resposta.status_code == 404 diff --git a/tests/integration/test_users.py b/tests/integration/test_users.py index 2fa12b1..3012f81 100644 --- a/tests/integration/test_users.py +++ b/tests/integration/test_users.py @@ -38,6 +38,7 @@ def _dados_funcionario(): async def test_admin_cria_usuarios_de_tipos_diversos( cliente_http: AsyncClient, cabecalhos_admin: dict[str, str], construtor_dados ): + """Valida que admin cria usuarios de tipos diversos.""" dados = construtor_dados() resposta = await cliente_http.post( '/api/v1/usuarios', json=dados, headers=cabecalhos_admin @@ -52,6 +53,7 @@ async def test_admin_cria_usuarios_de_tipos_diversos( @pytest.mark.asyncio @exige_banco async def test_criar_usuario_sem_token_retorna_401(cliente_http: AsyncClient): + """Valida que criar usuario sem token retorna 401.""" resposta = await cliente_http.post('/api/v1/usuarios', json=_dados_cliente()) assert resposta.status_code == 401 @@ -61,6 +63,7 @@ async def test_criar_usuario_sem_token_retorna_401(cliente_http: AsyncClient): async def test_criar_usuario_com_role_nao_admin_retorna_403( cliente_http: AsyncClient, cabecalhos_cliente: dict[str, str] ): + """Valida que criar usuario com role nao admin retorna 403.""" resposta = await cliente_http.post( '/api/v1/usuarios', json=_dados_cliente(), headers=cabecalhos_cliente ) @@ -72,6 +75,7 @@ async def test_criar_usuario_com_role_nao_admin_retorna_403( async def test_email_duplicado_retorna_409( cliente_http: AsyncClient, cabecalhos_admin: dict[str, str] ): + """Valida que email duplicado retorna 409.""" dados = _dados_cliente() primeira_resposta = await cliente_http.post( '/api/v1/usuarios', json=dados, headers=cabecalhos_admin @@ -86,6 +90,7 @@ async def test_email_duplicado_retorna_409( @pytest.mark.asyncio @exige_banco async def test_registro_publico_cria_usuario_pendente(cliente_http: AsyncClient): + """Valida que registro publico cria usuario pendente.""" dados = _dados_cliente() resposta = await cliente_http.post('/api/v1/usuarios/registro', json=dados) assert resposta.status_code == 201, resposta.text @@ -96,6 +101,7 @@ async def test_registro_publico_cria_usuario_pendente(cliente_http: AsyncClient) @pytest.mark.asyncio @exige_banco async def test_registro_publico_nao_permite_admin(cliente_http: AsyncClient): + """Valida que registro publico nao permite admin.""" dados = _dados_cliente() dados['role'] = 'admin' resposta = await cliente_http.post('/api/v1/usuarios/registro', json=dados) @@ -107,6 +113,7 @@ async def test_registro_publico_nao_permite_admin(cliente_http: AsyncClient): async def test_admin_aprova_cadastro_pendente( cliente_http: AsyncClient, cabecalhos_admin: dict[str, str] ): + """Valida que admin aprova cadastro pendente.""" registro = await cliente_http.post( '/api/v1/usuarios/registro', json=_dados_funcionario() ) @@ -124,6 +131,7 @@ async def test_admin_aprova_cadastro_pendente( async def test_aprovar_sem_admin_retorna_403( cliente_http: AsyncClient, cabecalhos_cliente: dict[str, str] ): + """Valida que aprovar sem admin retorna 403.""" registro = await cliente_http.post( '/api/v1/usuarios/registro', json=_dados_funcionario() ) @@ -141,6 +149,7 @@ async def test_editar_usuario_requer_admin( cabecalhos_admin: dict[str, str], cabecalhos_cliente: dict[str, str], ): + """Valida que editar usuario requer admin.""" criado = await cliente_http.post( '/api/v1/usuarios', json=_dados_cliente(), headers=cabecalhos_admin ) diff --git a/tests/test_app.py b/tests/test_app.py index a0088c6..0a35a14 100644 --- a/tests/test_app.py +++ b/tests/test_app.py @@ -7,12 +7,14 @@ def test_saude(): + """Valida que saude.""" resposta = cliente_http.get('/health') assert resposta.status_code == HTTPStatus.OK assert resposta.json() == {'message': 'API is running...'} def test_swagger_authorize_usa_bearer_sem_chamada_de_login(): + """Valida que swagger authorize usa bearer sem chamada de login.""" openapi = cliente_http.get('/openapi.json') esquema_seguranca = openapi.json()['components']['securitySchemes']['HTTPBearer'] assert openapi.status_code == HTTPStatus.OK diff --git a/tests/unit/test_agenda_service.py b/tests/unit/test_agenda_service.py index d0d3598..cbccd40 100644 --- a/tests/unit/test_agenda_service.py +++ b/tests/unit/test_agenda_service.py @@ -6,6 +6,7 @@ def test_evento_local_para_site_usa_fuso_horario_configurado(): + """Valida que evento local para site usa fuso horario configurado.""" evento = EventoAgenda( id=1, usuario_id=1, diff --git a/tests/unit/test_base_repository.py b/tests/unit/test_base_repository.py new file mode 100644 index 0000000..6a36bfd --- /dev/null +++ b/tests/unit/test_base_repository.py @@ -0,0 +1,156 @@ +from typing import Any + +import pytest +from db.base_repository import RepositorioBase +from modules.servicos.model import Servico +from sqlalchemy import select + + +class RepositorioServicoTeste(RepositorioBase[Servico]): + modelo = Servico + + +class ResultadoRepositorioFake: + def __init__(self, valores: list[Any] | None = None, escalar: Any = None): + self.valores = valores or [] + self.escalar = escalar + + def scalars(self): + return self + + def all(self): + return self.valores + + def scalar_one_or_none(self): + return self.escalar + + +class SessaoRepositorioFake: + def __init__(self): + self.adicionados: list[Any] = [] + self.flushes = 0 + self.refreshes: list[Any] = [] + self.execucoes: list[Any] = [] + self.resultado = ResultadoRepositorioFake() + self.entidades: dict[tuple[type[Any], int], Any] = {} + + def add(self, entidade: Any): + self.adicionados.append(entidade) + + async def flush(self): + self.flushes += 1 + + async def refresh(self, entidade: Any): + self.refreshes.append(entidade) + + async def get(self, modelo: type[Any], entidade_id: int): + return self.entidades.get((modelo, entidade_id)) + + async def execute(self, instrucao: Any): + self.execucoes.append(instrucao) + return self.resultado + + +@pytest.mark.asyncio +async def test_repositorio_base_adiciona_e_atualiza_estado_da_sessao(): + """Valida que repositorio base adiciona e atualiza estado da sessao.""" + sessao = SessaoRepositorioFake() + repositorio = RepositorioServicoTeste(sessao) + servico = Servico(id=1, nome='Site', slug='site') + + resposta = await repositorio.adicionar(servico) + + assert resposta is servico + assert sessao.adicionados == [servico] + assert sessao.flushes == 1 + assert sessao.refreshes == [servico] + + +@pytest.mark.asyncio +async def test_repositorio_base_obtem_por_id_usando_modelo_configurado(): + """Valida que repositorio base obtem por id usando modelo configurado.""" + sessao = SessaoRepositorioFake() + servico = Servico(id=1, nome='Site', slug='site') + sessao.entidades[(Servico, 1)] = servico + repositorio = RepositorioServicoTeste(sessao) + + assert await repositorio.obter(1) is servico + + +@pytest.mark.asyncio +async def test_repositorio_base_lista_com_paginacao_e_filtros_validos(): + """Valida que repositorio base lista com paginacao e filtros validos.""" + servico = Servico(id=1, nome='Site', slug='site') + sessao = SessaoRepositorioFake() + sessao.resultado = ResultadoRepositorioFake(valores=[servico]) + repositorio = RepositorioServicoTeste(sessao) + + resposta = await repositorio.listar_todos( + offset=5, + limit=10, + filtros={'status': None, 'slug': 'site', 'campo_inexistente': 'x'}, + ) + + assert resposta == [servico] + assert len(sessao.execucoes) == 1 + + +@pytest.mark.asyncio +async def test_repositorio_base_atualiza_ignorando_valores_none(): + """Valida que repositorio base atualiza ignorando valores none.""" + servico = Servico(id=1, nome='Site atualizado', slug='site') + sessao = SessaoRepositorioFake() + sessao.resultado = ResultadoRepositorioFake(escalar=servico) + repositorio = RepositorioServicoTeste(sessao) + + resposta = await repositorio.atualizar(1, {'nome': 'Site atualizado', 'slug': None}) + + assert resposta is servico + assert sessao.flushes == 1 + assert len(sessao.execucoes) == 1 + + +@pytest.mark.asyncio +async def test_repositorio_base_atualizacao_vazia_retorna_entidade_atual(): + """Valida que repositorio base atualizacao vazia retorna entidade atual.""" + servico = Servico(id=1, nome='Site', slug='site') + sessao = SessaoRepositorioFake() + sessao.entidades[(Servico, 1)] = servico + repositorio = RepositorioServicoTeste(sessao) + + resposta = await repositorio.atualizar(1, {'nome': None}) + + assert resposta is servico + assert sessao.execucoes == [] + + +@pytest.mark.asyncio +@pytest.mark.parametrize(('escalar', 'esperado'), [(1, True), (None, False)]) +async def test_repositorio_base_deleta_retornando_se_linha_existia(escalar, esperado): + """Valida que repositorio base deleta retornando se linha existia.""" + sessao = SessaoRepositorioFake() + sessao.resultado = ResultadoRepositorioFake(escalar=escalar) + repositorio = RepositorioServicoTeste(sessao) + + assert await repositorio.deletar(1) is esperado + assert sessao.flushes == 1 + assert len(sessao.execucoes) == 1 + + +def test_repositorio_base_remove_valores_vazios(): + """Valida que repositorio base remove valores vazios.""" + assert RepositorioBase._remover_valores_vazios({'a': 1, 'b': None}) == {'a': 1} + + +def test_repositorio_base_aplicar_filtros_ignora_valores_nulos_e_campos_invalidos(): + """Valida filtros nulos e campos invalidos no repositorio base.""" + repositorio = RepositorioServicoTeste(SessaoRepositorioFake()) + consulta = select(Servico) + + filtrada = repositorio._aplicar_filtros( + consulta, + {'slug': 'site', 'status': None, 'inexistente': 'x'}, + ) + + assert filtrada is not consulta + assert repositorio._aplicar_filtros(consulta, None) is consulta diff --git a/tests/unit/test_config.py b/tests/unit/test_config.py new file mode 100644 index 0000000..a682a25 --- /dev/null +++ b/tests/unit/test_config.py @@ -0,0 +1,46 @@ +from pathlib import Path + +import pytest +from core.config import Configuracoes + + +@pytest.mark.parametrize('valor_debug', ['release', 'production', 'prod']) +def test_configuracoes_interpreta_ambientes_de_release_como_debug_falso( + valor_debug: str, +): + """Valida que configuracoes interpreta ambientes de release como debug falso.""" + configuracoes = Configuracoes(debug=valor_debug) + + assert configuracoes.debug is False + + +@pytest.mark.parametrize('valor_debug', ['true', '1', True]) +def test_configuracoes_mantem_valores_verdadeiros_de_debug(valor_debug: str | bool): + """Valida que configuracoes mantem valores verdadeiros de debug.""" + configuracoes = Configuracoes(debug=valor_debug) + + assert configuracoes.debug is True + + +def test_caminho_conta_servico_google_prioriza_arquivo_do_host(tmp_path: Path): + """Valida que caminho conta servico google prioriza arquivo do host.""" + arquivo_container = tmp_path / 'container.json' + arquivo_host = tmp_path / 'host.json' + arquivo_container.write_text('{}') + arquivo_host.write_text('{}') + configuracoes = Configuracoes( + google_calendar_service_account_file=str(arquivo_container), + google_calendar_service_account_host_file=str(arquivo_host), + ) + + assert configuracoes.caminho_conta_servico_google() == arquivo_host.resolve() + + +def test_caminho_conta_servico_google_retorna_none_sem_arquivo_existente(): + """Valida que caminho conta servico google retorna none sem arquivo existente.""" + configuracoes = Configuracoes( + google_calendar_service_account_file='arquivo-inexistente.json', + google_calendar_service_account_host_file='outro-inexistente.json', + ) + + assert configuracoes.caminho_conta_servico_google() is None diff --git a/tests/unit/test_dashboard_service.py b/tests/unit/test_dashboard_service.py new file mode 100644 index 0000000..05a157b --- /dev/null +++ b/tests/unit/test_dashboard_service.py @@ -0,0 +1,200 @@ +from datetime import date +from typing import Any +from unittest.mock import AsyncMock + +import pytest +from core.enums import ( + Complexidade, + PapelUsuario, + Prioridade, + SituacaoProjeto, + SituacaoTarefa, +) +from fastapi import HTTPException +from modules.dashboard.service import ServicoPainel +from modules.projetos.model import Projeto +from modules.tarefas.model import Tarefa +from modules.users.model import Cliente, Funcionario + + +class ResultadoEscalarFake: + def __init__(self, valores: list[Any]): + self.valores = valores + + def scalars(self): + return self + + def all(self): + return self.valores + + +class SessaoPainelFake: + def __init__(self): + self.entidades: dict[tuple[type[Any], int], Any] = {} + self.resultado_execute: Any = ResultadoEscalarFake([]) + + async def get(self, modelo: type[Any], entidade_id: int): + return self.entidades.get((modelo, entidade_id)) + + async def execute(self, _consulta: Any): + return self.resultado_execute + + +@pytest.mark.asyncio +async def test_painel_admin_monta_indicadores_principais(): + """Valida que painel admin monta indicadores principais.""" + servico = ServicoPainel(SessaoPainelFake()) + servico._contar = AsyncMock(side_effect=[3, 2, 7, 5]) + servico._series_por_periodo = AsyncMock( + side_effect=[ + [{'periodo': 'semana', 'total': 3}], + [{'periodo': 'dia', 'total': 7}], + ] + ) + servico._leads_recentes = AsyncMock(return_value=[{'id': 1}]) + servico._ranking = AsyncMock(return_value=[{'funcionario_id': 2}]) + + resposta = await servico.admin() + + assert resposta['cards'] == { + 'leads_no_mes': 3, + 'projetos_ativos': 2, + 'tarefas_concluidas': 7, + 'clientes_ativos': 5, + } + assert resposta['leads_por_semana'] == [{'periodo': 'semana', 'total': 3}] + assert resposta['ranking_xp_semanal'] == [{'funcionario_id': 2}] + + +@pytest.mark.asyncio +async def test_painel_cliente_monta_dados_do_cliente_autorizado(): + """Valida que painel cliente monta dados do cliente autorizado.""" + servico = ServicoPainel(SessaoPainelFake()) + servico._garantir_acesso_cliente = AsyncMock(return_value=Cliente(id=10)) + servico._contar = AsyncMock(side_effect=[2, 4, 1]) + servico._projetos_do_cliente = AsyncMock(return_value=[{'id': 1}]) + servico._tarefas_proximas = AsyncMock(return_value=[{'id': 2}]) + servico._comunicados_recentes = AsyncMock(return_value=[{'id': 3}]) + servico._eventos_do_usuario = AsyncMock(return_value=[{'id': 4}]) + + resposta = await servico.cliente( + cliente_id=10, + usuario_id=10, + papel=PapelUsuario.CLIENTE.value, + ) + + assert resposta['cards']['projetos_ativos'] == 2 + assert resposta['cards']['tarefas_em_andamento'] == 4 + assert resposta['proximas_entregas'] == [{'id': 2}] + assert resposta['eventos'] == [{'id': 4}] + + +@pytest.mark.asyncio +async def test_painel_funcionario_monta_perfil_cards_e_listas(): + """Valida que painel funcionario monta perfil cards e listas.""" + funcionario = Funcionario(id=20, cargo='Dev', xp_total=750, nivel=2) + servico = ServicoPainel(SessaoPainelFake()) + servico._garantir_acesso_funcionario = AsyncMock(return_value=funcionario) + servico._contar = AsyncMock(side_effect=[8, 6]) + servico._xp_do_funcionario = AsyncMock(return_value=150) + servico._tarefas_do_funcionario = AsyncMock(return_value=[{'id': 1}]) + servico._historico_xp = AsyncMock(return_value=[{'id': 2}]) + servico._conquistas_do_funcionario = AsyncMock(return_value=[{'id': 3}]) + servico._ranking = AsyncMock(return_value=[{'funcionario_id': 20}]) + + resposta = await servico.funcionario( + funcionario_id=20, + usuario_id=20, + papel=PapelUsuario.FUNCIONARIO.value, + ) + + assert resposta['perfil'] == { + 'id': 20, + 'cargo': 'Dev', + 'xp_total': 750, + 'nivel': 2, + } + assert resposta['cards']['xp_no_mes'] == 150 + assert resposta['conquistas'] == [{'id': 3}] + + +@pytest.mark.asyncio +async def test_painel_projeto_kanban_agrupa_tarefas_por_status(): + """Valida que painel projeto kanban agrupa tarefas por status.""" + sessao = SessaoPainelFake() + projeto = Projeto( + id=1, + cliente_id=10, + nome='Site', + status=SituacaoProjeto.EM_ANDAMENTO, + progresso=40, + data_fim=date(2026, 6, 30), + ) + tarefa = Tarefa( + id=1, + projeto_id=1, + responsavel_id=20, + titulo='Layout', + descricao='Criar layout', + status=SituacaoTarefa.EM_PROGRESSO, + complexidade=Complexidade.MEDIA, + prioridade=Prioridade.ALTA, + categoria='design', + prazo=date(2026, 6, 1), + ordem=2, + ) + sessao.entidades[(Projeto, 1)] = projeto + sessao.resultado_execute = ResultadoEscalarFake([tarefa]) + servico = ServicoPainel(sessao) + + resposta = await servico.projeto_kanban( + projeto_id=1, + usuario_id=99, + papel=PapelUsuario.ADMIN.value, + ) + + assert resposta['projeto']['nome'] == 'Site' + assert resposta['colunas']['em_progresso'][0]['titulo'] == 'Layout' + assert resposta['columns'] is resposta['colunas'] + + +@pytest.mark.asyncio +async def test_painel_bloqueia_cliente_e_funcionario_sem_permissao(): + """Valida que painel bloqueia cliente e funcionario sem permissao.""" + sessao = SessaoPainelFake() + sessao.entidades[(Cliente, 10)] = Cliente(id=10) + sessao.entidades[(Funcionario, 20)] = Funcionario(id=20, cargo='Dev') + servico = ServicoPainel(sessao) + + with pytest.raises(HTTPException) as cliente: + await servico._garantir_acesso_cliente( + 10, + usuario_id=99, + papel=PapelUsuario.CLIENTE.value, + ) + with pytest.raises(HTTPException) as funcionario: + await servico._garantir_acesso_funcionario( + 20, + usuario_id=99, + papel=PapelUsuario.FUNCIONARIO.value, + ) + + assert cliente.value.status_code == 403 + assert funcionario.value.status_code == 403 + + +@pytest.mark.asyncio +async def test_painel_retorna_404_para_entidades_inexistentes(): + """Valida que painel retorna 404 para entidades inexistentes.""" + servico = ServicoPainel(SessaoPainelFake()) + + with pytest.raises(HTTPException) as cliente: + await servico._garantir_cliente(404) + with pytest.raises(HTTPException) as funcionario: + await servico._garantir_funcionario(404) + with pytest.raises(HTTPException) as projeto: + await servico.projeto_kanban(404) + + assert cliente.value.status_code == 404 + assert funcionario.value.status_code == 404 + assert projeto.value.status_code == 404 diff --git a/tests/unit/test_database.py b/tests/unit/test_database.py new file mode 100644 index 0000000..ce18d99 --- /dev/null +++ b/tests/unit/test_database.py @@ -0,0 +1,37 @@ +import pytest +from core.database import normalizar_url_banco_assincrono + + +@pytest.mark.parametrize( + ('url_entrada', 'url_esperada', 'argumentos_esperados'), + [ + ( + 'postgresql://usuario:senha@localhost:5432/korus', + 'postgresql+asyncpg://', + {'dsn': 'postgresql://usuario:senha@localhost:5432/korus'}, + ), + ( + 'postgres://usuario:senha@localhost/korus?sslmode=require', + 'postgresql+asyncpg://', + {'dsn': 'postgresql://usuario:senha@localhost/korus?sslmode=require'}, + ), + ( + 'postgresql+asyncpg://usuario:senha@localhost/korus', + 'postgresql+asyncpg://', + {'dsn': 'postgresql://usuario:senha@localhost/korus'}, + ), + ( + 'sqlite+aiosqlite:///tmp/korus.db', + 'sqlite+aiosqlite:///tmp/korus.db', + {}, + ), + ], +) +def test_normalizar_url_banco_assincrono( + url_entrada: str, url_esperada: str, argumentos_esperados: dict[str, str] +): + """Valida que normalizar url banco assincrono.""" + url_normalizada, argumentos = normalizar_url_banco_assincrono(url_entrada) + + assert url_normalizada == url_esperada + assert argumentos == argumentos_esperados diff --git a/tests/unit/test_db_registro.py b/tests/unit/test_db_registro.py new file mode 100644 index 0000000..e6f5ae6 --- /dev/null +++ b/tests/unit/test_db_registro.py @@ -0,0 +1,67 @@ +import importlib +from types import ModuleType + +import pytest +from db.base import Base +from db.registro import TODOS_MODELOS + +MODULOS_MODELO = ( + 'modules.academy.model', + 'modules.agenda.model', + 'modules.comunicados.model', + 'modules.gamificacao.model', + 'modules.integracoes.model', + 'modules.leads.model', + 'modules.lgpd.model', + 'modules.portfolio.model', + 'modules.projetos.model', + 'modules.servicos.model', + 'modules.tarefas.model', + 'modules.users.model', +) + +TABELAS_ESPERADAS = { + 'academy', + 'admin', + 'anexo', + 'cliente', + 'comentario', + 'comunicado', + 'comunicado_leitura', + 'conquista', + 'consentimento_lgpd', + 'entregavel', + 'evento_agenda', + 'funcionario', + 'funcionario_conquista', + 'historico_xp', + 'integracao', + 'lead', + 'portfolio', + 'projeto', + 'projeto_funcionario', + 'regra_xp', + 'servico', + 'solicitacao_reuniao', + 'tarefa', + 'usuario', +} + + +def test_registro_carrega_um_modulo_por_contexto_de_dominio(): + """Valida que registro carrega um modulo por contexto de dominio.""" + assert len(TODOS_MODELOS) == len(MODULOS_MODELO) + assert all(isinstance(modulo, ModuleType) for modulo in TODOS_MODELOS) + + +@pytest.mark.parametrize('nome_modulo', MODULOS_MODELO) +def test_modulos_de_modelo_registrados_sao_importaveis(nome_modulo: str): + """Valida que modulos de modelo registrados sao importaveis.""" + modulo = importlib.import_module(nome_modulo) + + assert modulo in TODOS_MODELOS + + +def test_registro_popula_metadata_usada_pelo_alembic(): + """Valida que registro popula metadata usada pelo alembic.""" + assert TABELAS_ESPERADAS <= set(Base.metadata.tables) diff --git a/tests/unit/test_google_calendar.py b/tests/unit/test_google_calendar.py index b7280ea..92eb60d 100644 --- a/tests/unit/test_google_calendar.py +++ b/tests/unit/test_google_calendar.py @@ -10,6 +10,7 @@ def test_caminho_conta_servico_prioriza_arquivo_local(tmp_path: Path): + """Valida que caminho conta servico prioriza arquivo local.""" arquivo_local = tmp_path / 'conta-servico.json' arquivo_local.write_text('{"client_email":"x","private_key":"y"}', encoding='utf-8') arquivo_docker = tmp_path / 'docker.json' @@ -23,6 +24,7 @@ def test_caminho_conta_servico_prioriza_arquivo_local(tmp_path: Path): def test_criar_evento_ignora_google_quando_arquivo_credencial_nao_existe(): + """Valida que criar evento ignora google quando arquivo credencial nao existe.""" cliente = ClienteGoogleCalendar( Configuracoes( google_calendar_enabled=True, @@ -36,6 +38,7 @@ def test_criar_evento_ignora_google_quando_arquivo_credencial_nao_existe(): @pytest.mark.asyncio async def test_criar_evento_retorna_none_sem_credenciais_validas(): + """Valida que criar evento retorna none sem credenciais validas.""" cliente = ClienteGoogleCalendar( Configuracoes( google_calendar_enabled=True, @@ -55,6 +58,7 @@ async def test_criar_evento_retorna_none_sem_credenciais_validas(): def test_evento_local_para_site_usa_fuso_horario_configurado(): + """Valida que evento local para site usa fuso horario configurado.""" evento = EventoAgenda( id=1, usuario_id=1, diff --git a/tests/unit/test_migrations.py b/tests/unit/test_migrations.py new file mode 100644 index 0000000..cbcc768 --- /dev/null +++ b/tests/unit/test_migrations.py @@ -0,0 +1,24 @@ +import os +import subprocess +import sys + + +def test_alembic_gera_sql_offline_com_modelos_registrados(): + """Valida que alembic gera sql offline com modelos registrados.""" + ambiente = { + **os.environ, + 'DATABASE_URL': 'postgresql+asyncpg://korus:korus@localhost:5432/korus_test', + 'JWT_SECRET_KEY': 'test-secret', + } + + resultado = subprocess.run( + [sys.executable, '-m', 'alembic', 'upgrade', 'head', '--sql'], + check=False, + capture_output=True, + env=ambiente, + text=True, + ) + + assert resultado.returncode == 0, resultado.stderr + assert 'CREATE TABLE usuario' in resultado.stdout + assert 'google_event_id' in resultado.stdout diff --git a/tests/unit/test_password.py b/tests/unit/test_password.py index 40ff18a..0e2dc4e 100644 --- a/tests/unit/test_password.py +++ b/tests/unit/test_password.py @@ -6,6 +6,7 @@ 'senha_plana', ['secreta-123', 'outra-coisa-segura!', 'çãoéàü'] ) def test_hash_senha_e_verifica(senha_plana: str): + """Valida que hash senha e verifica.""" hash_gerado = gerar_hash_senha(senha_plana) assert hash_gerado != senha_plana assert verificar_senha(senha_plana, hash_gerado) is True @@ -13,5 +14,6 @@ def test_hash_senha_e_verifica(senha_plana: str): def test_hash_senha_gera_hashes_distintos_para_mesma_senha(): + """Valida que hash senha gera hashes distintos para mesma senha.""" senha_plana = 'mesma-senha' assert gerar_hash_senha(senha_plana) != gerar_hash_senha(senha_plana) diff --git a/tests/unit/test_schemas.py b/tests/unit/test_schemas.py index cddc05d..1a9a58b 100644 --- a/tests/unit/test_schemas.py +++ b/tests/unit/test_schemas.py @@ -1,11 +1,26 @@ import pytest -from core.enums import LeadPrioridade, SituacaoLead, SituacaoServico -from modules.leads.schema import LeadCriar -from modules.servicos.schema import ServicoAtualizar, ServicoCriar +from core.enums import ( + LeadPrioridade, + SituacaoLead, + SituacaoProjeto, + SituacaoServico, + SituacaoSolicitacao, +) +from modules.academy.schema import AcademiaAtualizar +from modules.agenda.schema import ( + EventoAgendaAtualizar, + SolicitacaoReuniaoCriar, +) +from modules.integracoes.schema import IntegracaoAtualizar, IntegracaoCriar +from modules.leads.schema import LeadAtualizar, LeadCriar +from modules.projetos.schema import ProjetoAtualizar, ProjetoCriar +from modules.servicos.schema import EntregavelAtualizar, ServicoAtualizar, ServicoCriar +from modules.users.schema import UsuarioAtualizar from pydantic import ValidationError def test_servico_criar_define_status_ativo_por_padrao(): + """Valida que servico criar define status ativo por padrao.""" dados = ServicoCriar( nome='Identidade Visual', slug='identidade-visual', @@ -16,6 +31,7 @@ def test_servico_criar_define_status_ativo_por_padrao(): def test_servico_atualizar_permite_patch_parcial(): + """Valida que servico atualizar permite patch parcial.""" dados = ServicoAtualizar(descricao='Nova descricao') assert dados.nome is None @@ -23,6 +39,7 @@ def test_servico_atualizar_permite_patch_parcial(): def test_lead_criar_define_status_e_prioridade_padrao(): + """Valida que lead criar define status e prioridade padrao.""" dados = LeadCriar( nome='Cliente Teste', email='cliente@example.com', @@ -35,5 +52,84 @@ def test_lead_criar_define_status_e_prioridade_padrao(): def test_lead_criar_valida_email(): + """Valida que lead criar valida email.""" with pytest.raises(ValidationError): LeadCriar(nome='Cliente Teste', email='email-invalido') + + +@pytest.mark.parametrize( + 'classe_schema', + [ + AcademiaAtualizar, + EntregavelAtualizar, + EventoAgendaAtualizar, + IntegracaoAtualizar, + LeadAtualizar, + ProjetoAtualizar, + ServicoAtualizar, + UsuarioAtualizar, + ], +) +def test_schemas_de_atualizacao_permitem_patch_vazio(classe_schema): + """Valida que schemas de atualizacao permitem patch vazio.""" + dados = classe_schema() + + assert dados.model_dump(exclude_none=True) == {} + + +@pytest.mark.parametrize( + ('progresso', 'deve_validar'), + [(-1, False), (0, True), (45, True), (100, True), (101, False)], +) +def test_projeto_criar_valida_limites_de_progresso(progresso: int, deve_validar: bool): + """Valida que projeto criar valida limites de progresso.""" + dados = { + 'nome': 'Redesign do site', + 'cliente_id': 1, + 'status': SituacaoProjeto.PLANEJAMENTO, + 'progresso': progresso, + } + + if deve_validar: + assert ProjetoCriar(**dados).progresso == progresso + else: + with pytest.raises(ValidationError): + ProjetoCriar(**dados) + + +@pytest.mark.parametrize( + ('remetente_id', 'destinatario_id', 'deve_validar'), + [(1, 2, True), (2, 1, True), (7, 7, False)], +) +def test_solicitacao_reuniao_exige_usuarios_diferentes( + remetente_id: int, destinatario_id: int, deve_validar: bool +): + """Valida que solicitacao reuniao exige usuarios diferentes.""" + dados = { + 'titulo': 'Revisao de layout', + 'data': '2026-05-25', + 'hora': '15:30:00', + 'remetente_id': remetente_id, + 'destinatario_id': destinatario_id, + } + + if deve_validar: + assert SolicitacaoReuniaoCriar(**dados).status == SituacaoSolicitacao.PENDENTE + else: + with pytest.raises(ValidationError): + SolicitacaoReuniaoCriar(**dados) + + +@pytest.mark.parametrize('status', ['conectado', 'desconectado']) +def test_integracao_criar_aceita_status_validos(status: str): + """Valida que integracao criar aceita status validos.""" + dados = IntegracaoCriar(status=status) + + assert dados.nome == 'google_calendar' + assert dados.status.value == status + + +def test_integracao_criar_rejeita_nome_fora_da_lista_permitida(): + """Valida que integracao criar rejeita nome fora da lista permitida.""" + with pytest.raises(ValidationError): + IntegracaoCriar(nome='slack') diff --git a/tests/unit/test_security.py b/tests/unit/test_security.py index 61359b7..c171d15 100644 --- a/tests/unit/test_security.py +++ b/tests/unit/test_security.py @@ -1,12 +1,21 @@ import pytest -from core.security import criar_token_acesso, decodificar_token +from core.security import ( + criar_token_acesso, + criar_token_atualizacao, + decodificar_token, + exigir_papel, + obter_usuario_atual, + obter_usuario_atual_id, +) from fastapi import HTTPException +from fastapi.security import HTTPAuthorizationCredentials @pytest.mark.parametrize( ('sujeito', 'papel'), [(1, 'admin'), (42, 'cliente'), (99, 'funcionario')] ) def test_criar_e_decodificar_token_acesso(sujeito: int, papel: 'str'): + """Valida que criar e decodificar token acesso.""" token = criar_token_acesso(sujeito, dados_extras={'role': papel}) dados = decodificar_token(token) assert dados['sub'] == str(sujeito) @@ -15,6 +24,75 @@ def test_criar_e_decodificar_token_acesso(sujeito: int, papel: 'str'): def test_decodificar_token_invalido_lanca_excecao(): + """Valida que decodificar token invalido lanca excecao.""" with pytest.raises(HTTPException) as exc: decodificar_token('token-invalido') assert exc.value.status_code == 401 + + +def _credenciais(token: str) -> HTTPAuthorizationCredentials: + return HTTPAuthorizationCredentials(scheme='Bearer', credentials=token) + + +def test_criar_token_atualizacao_define_tipo_refresh(): + """Valida que criar token atualizacao define tipo refresh.""" + token = criar_token_atualizacao(123) + dados = decodificar_token(token) + + assert dados['sub'] == '123' + assert dados['type'] == 'refresh' + + +def test_obter_usuario_atual_id_extrai_sub_do_token(): + """Valida que obter usuario atual id extrai sub do token.""" + token = criar_token_acesso(77, dados_extras={'role': 'admin'}) + + assert obter_usuario_atual_id(_credenciais(token)) == 77 + + +def test_obter_usuario_atual_extrai_id_e_papel(): + """Valida que obter usuario atual extrai id e papel.""" + token = criar_token_acesso(88, dados_extras={'role': 'funcionario'}) + + usuario = obter_usuario_atual(_credenciais(token)) + + assert usuario.id == 88 + assert usuario.papel == 'funcionario' + + +@pytest.mark.parametrize('dependencia', [obter_usuario_atual_id, obter_usuario_atual]) +def test_dependencias_de_usuario_rejeitam_credenciais_ausentes(dependencia): + """Valida que dependencias de usuario rejeitam credenciais ausentes.""" + with pytest.raises(HTTPException) as exc: + dependencia(None) + + assert exc.value.status_code == 401 + + +def test_obter_usuario_atual_rejeita_token_sem_role(): + """Valida que obter usuario atual rejeita token sem role.""" + token = criar_token_acesso(88) + + with pytest.raises(HTTPException) as exc: + obter_usuario_atual(_credenciais(token)) + + assert exc.value.status_code == 401 + + +def test_exigir_papel_permite_papel_autorizado(): + """Valida que exigir papel permite papel autorizado.""" + token = criar_token_acesso(99, dados_extras={'role': 'admin'}) + verificar_admin = exigir_papel('admin') + + assert verificar_admin(_credenciais(token)) == 99 + + +def test_exigir_papel_bloqueia_papel_nao_autorizado(): + """Valida que exigir papel bloqueia papel nao autorizado.""" + token = criar_token_acesso(99, dados_extras={'role': 'cliente'}) + verificar_admin = exigir_papel('admin') + + with pytest.raises(HTTPException) as exc: + verificar_admin(_credenciais(token)) + + assert exc.value.status_code == 403 diff --git a/tests/unit/test_services.py b/tests/unit/test_services.py new file mode 100644 index 0000000..c861302 --- /dev/null +++ b/tests/unit/test_services.py @@ -0,0 +1,1078 @@ +from datetime import date, datetime, time, timezone +from decimal import Decimal +from types import SimpleNamespace +from typing import Any + +import pytest +from core.enums import ( + Complexidade, + ComunicadoAlvo, + ConsentimentoTipo, + EventoTipo, + LeadPrioridade, + PapelUsuario, + Prioridade, + SituacaoIntegracao, + SituacaoLead, + SituacaoProjeto, + SituacaoTarefa, + SituacaoUsuario, + TipoAcademia, +) +from fastapi import HTTPException +from modules.academy.model import Academia +from modules.academy.schema import AcademiaAtualizar, AcademiaCriar +from modules.academy.service import ServicoAcademia +from modules.agenda.google_calendar import EventoGoogleCalendar +from modules.agenda.model import EventoAgenda, SolicitacaoReuniao +from modules.agenda.schema import ( + EventoAgendaAtualizar, + EventoAgendaCriar, + SolicitacaoReuniaoAtualizar, + SolicitacaoReuniaoCriar, +) +from modules.agenda.service import ServicoAgenda +from modules.comunicados.model import Comunicado, ComunicadoLeitura +from modules.comunicados.schema import ComunicadoAtualizar, ComunicadoCriar +from modules.comunicados.service import ServicoComunicado +from modules.gamificacao.model import ( + Conquista, + FuncionarioConquista, + HistoricoXp, + RegraXp, +) +from modules.gamificacao.schema import ( + ConquistaAtualizar, + ConquistaCriar, + HistoricoXpCriar, + RegraXpAtualizar, + RegraXpCriar, +) +from modules.gamificacao.service import ServicoGamificacao +from modules.integracoes.model import Integracao +from modules.integracoes.schema import IntegracaoAtualizar, IntegracaoCriar +from modules.integracoes.service import ServicoIntegracao +from modules.leads.model import Lead +from modules.leads.schema import LeadAtualizar, LeadCriar +from modules.leads.service import ServicoLead +from modules.lgpd.model import ConsentimentoLgpd +from modules.lgpd.schema import ConsentimentoLgpdCriar +from modules.lgpd.service import ServicoLgpd +from modules.portfolio.model import Portfolio +from modules.portfolio.schema import PortfolioAtualizar, PortfolioCriar +from modules.portfolio.service import ServicoPortfolio +from modules.projetos.model import Projeto, ProjetoFuncionario +from modules.projetos.schema import ProjetoFuncionarioCriar +from modules.projetos.service import ServicoProjeto +from modules.servicos.model import Entregavel, Servico +from modules.servicos.schema import ( + EntregavelAtualizar, + EntregavelCriar, + ServicoCriar, +) +from modules.servicos.service import ServicoServico +from modules.tarefas.model import Anexo, Comentario, Tarefa +from modules.tarefas.schema import ( + AnexoCriar, + ComentarioCriar, + TarefaAtualizar, +) +from modules.tarefas.service import ServicoTarefa +from modules.users.model import Funcionario, Usuario +from modules.users.schema import ( + DadosCliente, + DadosFuncionario, + UsuarioCriar, + UsuarioRegistrar, +) +from modules.users.service import ServicoUsuario + + +class SessaoFake: + def __init__(self): + self.commits = 0 + self.refreshes: list[Any] = [] + self.entidades: dict[tuple[type[Any], int], Any] = {} + self.execute_result = None + self.flushes = 0 + + async def commit(self): + self.commits += 1 + + async def flush(self): + self.flushes += 1 + + async def refresh(self, entidade: Any): + self.refreshes.append(entidade) + + async def get(self, modelo: type[Any], entidade_id: int): + return self.entidades.get((modelo, entidade_id)) + + async def execute(self, _consulta: Any): + return ResultadoFake(self.execute_result) + + +class ResultadoFake: + def __init__(self, valor: Any): + self.valor = valor + + def scalar_one_or_none(self): + return self.valor + + +class RepositorioFake: # noqa: PLR0904 + def __init__(self, itens: dict[int, Any] | None = None): + self.itens = itens or {} + self.adicionados: list[Any] = [] + self.atualizacoes: list[tuple[int, dict[str, Any]]] = [] + self.delecoes: list[int] = [] + self.filtros_recebidos: list[dict[str, Any]] = [] + self.slug_existente = False + self.nome_existente = False + self.email_existente = False + + async def adicionar(self, entidade: Any): + if getattr(entidade, 'id', None) is None: + entidade.id = len(self.itens) + len(self.adicionados) + 1 + self.adicionados.append(entidade) + self.itens[entidade.id] = entidade + return entidade + + async def obter(self, entidade_id: int): + return self.itens.get(entidade_id) + + async def listar_todos(self, offset: int = 0, limit: int = 100, filtros=None): + self.filtros_recebidos.append( + {'offset': offset, 'limit': limit, 'filtros': filtros} + ) + return list(self.itens.values())[offset : offset + limit] + + async def atualizar(self, entidade_id: int, dados: dict[str, Any]): + self.atualizacoes.append((entidade_id, dados)) + entidade = self.itens.get(entidade_id) + if entidade is None: + return None + for campo, valor in dados.items(): + setattr(entidade, campo, valor) + return entidade + + async def deletar(self, entidade_id: int): + self.delecoes.append(entidade_id) + return self.itens.pop(entidade_id, None) is not None + + async def obter_por_slug(self, _slug: str): + return SimpleNamespace(id=99) if self.slug_existente else None + + async def obter_por_nome(self, _nome: str): + return SimpleNamespace(id=99) if self.nome_existente else None + + async def obter_por_email(self, _email: str): + return SimpleNamespace(id=99) if self.email_existente else None + + async def obter_google_calendar(self, integracao_id: int): + return await self.obter(integracao_id) + + async def listar_google_calendar(self, offset: int, limit: int): + return await self.listar_todos(offset=offset, limit=limit) + + async def listar_filtrados(self, **filtros: Any): + self.filtros_recebidos.append(filtros) + return list(self.itens.values()) + + async def listar_por_servico(self, servico_id: int): + self.filtros_recebidos.append({'servico_id': servico_id}) + return [ + item for item in self.itens.values() if item.servico_id == servico_id + ] + + async def listar_por_projeto(self, projeto_id: int): + self.filtros_recebidos.append({'projeto_id': projeto_id}) + return [ + item for item in self.itens.values() if item.projeto_id == projeto_id + ] + + async def listar_por_tarefa(self, tarefa_id: int): + self.filtros_recebidos.append({'tarefa_id': tarefa_id}) + return [ + item for item in self.itens.values() if item.tarefa_id == tarefa_id + ] + + async def listar_por_comunicado(self, comunicado_id: int): + self.filtros_recebidos.append({'comunicado_id': comunicado_id}) + return [ + item + for item in self.itens.values() + if item.comunicado_id == comunicado_id + ] + + async def listar_por_funcionario(self, funcionario_id: int): + self.filtros_recebidos.append({'funcionario_id': funcionario_id}) + return [ + item + for item in self.itens.values() + if item.funcionario_id == funcionario_id + ] + + async def listar_por_usuario(self, usuario_id: int): + self.filtros_recebidos.append({'usuario_id': usuario_id}) + return [ + item for item in self.itens.values() if item.usuario_id == usuario_id + ] + + async def listar_recebidas(self, destinatario_id: int): + self.filtros_recebidos.append({'destinatario_id': destinatario_id}) + return [ + item + for item in self.itens.values() + if item.destinatario_id == destinatario_id + ] + + async def marcar_lido(self, comunicado_id: int, usuario_id: int): + leitura = ComunicadoLeitura( + comunicado_id=comunicado_id, + usuario_id=usuario_id, + ) + return await self.adicionar(leitura) + + async def desbloquear(self, funcionario_id: int, conquista_id: int): + registro = FuncionarioConquista( + funcionario_id=funcionario_id, + conquista_id=conquista_id, + ) + return await self.adicionar(registro) + + async def remover(self, projeto_id: int, funcionario_id: int): + chave = next( + ( + item_id + for item_id, item in self.itens.items() + if ( + item.projeto_id == projeto_id + and item.funcionario_id == funcionario_id + ) + ), + None, + ) + if chave is None: + return False + self.itens.pop(chave) + return True + + async def contem_membro(self, projeto_id: int, funcionario_id: int): + return any( + item.projeto_id == projeto_id and item.funcionario_id == funcionario_id + for item in self.itens.values() + ) + + async def listar_para_funcionario(self, **filtros: Any): + self.filtros_recebidos.append(filtros) + return list(self.itens.values()) + + async def listar_visiveis(self, **filtros: Any): + self.filtros_recebidos.append(filtros) + return list(self.itens.values()) + + +class CalendarioGoogleFake: + def __init__(self): + self.eventos: list[EventoGoogleCalendar] = [] + self.evento_criado = EventoGoogleCalendar( + id='google-1', + titulo='Evento Google', + descricao=None, + inicio=datetime(2026, 5, 25, 13, 0, tzinfo=timezone.utc), + fim=datetime(2026, 5, 25, 14, 0, tzinfo=timezone.utc), + link='https://calendar.test/google-1', + ) + self.criados: list[dict[str, Any]] = [] + self.atualizados: list[dict[str, Any]] = [] + self.deletados: list[str] = [] + + async def criar_evento(self, **dados: Any): + self.criados.append(dados) + return self.evento_criado + + async def atualizar_evento(self, **dados: Any): + self.atualizados.append(dados) + return self.evento_criado + + async def deletar_evento(self, google_event_id: str): + self.deletados.append(google_event_id) + + async def listar_eventos(self, data_inicio=None, data_fim=None): + return [ + evento + for evento in self.eventos + if (data_inicio is None or evento.inicio.date() >= data_inicio) + and (data_fim is None or evento.inicio.date() <= data_fim) + ] + + +def servico_academia(sessao: SessaoFake): + servico = ServicoAcademia(sessao) + servico.repository = RepositorioFake( + { + 1: Academia( + id=1, + titulo='Marketing', + tipo=TipoAcademia.CURSO, + preco=Decimal('10.00'), + url_externa='https://korus.test/marketing', + ) + } + ) + return servico + + +def servico_lead(sessao: SessaoFake): + servico = ServicoLead(sessao) + servico.repository = RepositorioFake( + {1: Lead(id=1, nome='Maria', email='maria@example.com')} + ) + return servico + + +def servico_portfolio(sessao: SessaoFake): + servico = ServicoPortfolio(sessao) + servico.repository = RepositorioFake({1: Portfolio(id=1, nome='Case')}) + return servico + + +def servico_agenda(sessao: SessaoFake): + servico = ServicoAgenda(sessao) + servico.eventos = RepositorioFake( + { + 1: EventoAgenda( + id=1, + usuario_id=1, + titulo='Reunião local', + tipo=EventoTipo.REUNIAO, + data=date(2026, 5, 25), + hora=time(9, 0), + duracao_min=60, + google_event_id='sincronizado', + ), + 2: EventoAgenda( + id=2, + usuario_id=1, + titulo='Fora do periodo', + tipo=EventoTipo.TAREFA, + data=date(2026, 6, 10), + hora=time(9, 0), + duracao_min=30, + ), + } + ) + servico.solicitacoes = RepositorioFake( + { + 1: SolicitacaoReuniao( + id=1, + remetente_id=2, + destinatario_id=1, + titulo='Revisão', + data=date(2026, 5, 26), + hora=time(10, 0), + ) + } + ) + servico.calendario_google = CalendarioGoogleFake() + return servico + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ('fabrica_servico', 'dados_criacao', 'tipo_modelo'), + [ + ( + servico_academia, + AcademiaCriar( + titulo='Curso', + tipo=TipoAcademia.CURSO, + url_externa='https://korus.test/curso', + ), + Academia, + ), + ( + servico_lead, + LeadCriar(nome='Cliente', email='cliente@example.com'), + Lead, + ), + (servico_portfolio, PortfolioCriar(nome='Identidade Aurora'), Portfolio), + ], +) +async def test_services_crud_criam_entidade_e_comitam( + fabrica_servico, dados_criacao, tipo_modelo +): + """Valida que services crud criam entidade e comitam.""" + sessao = SessaoFake() + servico = fabrica_servico(sessao) + + entidade = await servico.criar(dados_criacao) + + assert isinstance(entidade, tipo_modelo) + assert sessao.commits == 1 + assert servico.repository.adicionados == [entidade] + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ('fabrica_servico', 'dados_atualizacao', 'campo', 'valor'), + [ + (servico_academia, AcademiaAtualizar(titulo='Novo'), 'titulo', 'Novo'), + ( + servico_lead, + LeadAtualizar(status=SituacaoLead.QUALIFICADO), + 'status', + SituacaoLead.QUALIFICADO, + ), + (servico_portfolio, PortfolioAtualizar(destaque=True), 'destaque', True), + ], +) +async def test_services_crud_atualizam_e_comitam( + fabrica_servico, dados_atualizacao, campo: str, valor: Any +): + """Valida que services crud atualizam e comitam.""" + sessao = SessaoFake() + servico = fabrica_servico(sessao) + + entidade = await servico.atualizar(1, dados_atualizacao) + + assert getattr(entidade, campo) == valor + assert sessao.commits == 1 + assert servico.repository.atualizacoes[-1][0] == 1 + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ('fabrica_servico', 'dados_atualizacao'), + [ + (servico_academia, AcademiaAtualizar(titulo='Novo')), + (servico_lead, LeadAtualizar(status=SituacaoLead.PERDIDO)), + (servico_portfolio, PortfolioAtualizar(destaque=False)), + ], +) +async def test_services_crud_retorna_404_quando_atualizacao_nao_encontra_registro( + fabrica_servico, dados_atualizacao +): + """Valida que services crud retorna 404 quando atualizacao nao encontra registro.""" + sessao = SessaoFake() + servico = fabrica_servico(sessao) + + with pytest.raises(HTTPException) as exc: + await servico.atualizar(404, dados_atualizacao) + + assert exc.value.status_code == 404 + assert sessao.commits == 0 + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + 'fabrica_servico', [servico_academia, servico_lead, servico_portfolio] +) +async def test_services_crud_deletam_e_comitam(fabrica_servico): + """Valida que services crud deletam e comitam.""" + sessao = SessaoFake() + servico = fabrica_servico(sessao) + + await servico.deletar(1) + + assert sessao.commits == 1 + assert servico.repository.delecoes == [1] + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + 'fabrica_servico', [servico_academia, servico_lead, servico_portfolio] +) +async def test_services_crud_retorna_404_quando_delecao_nao_encontra_registro( + fabrica_servico, +): + """Valida que services crud retorna 404 quando delecao nao encontra registro.""" + sessao = SessaoFake() + servico = fabrica_servico(sessao) + + with pytest.raises(HTTPException) as exc: + await servico.deletar(404) + + assert exc.value.status_code == 404 + assert sessao.commits == 0 + + +@pytest.mark.asyncio +async def test_servico_academia_lista_com_filtros(): + """Valida que servico academia lista com filtros.""" + sessao = SessaoFake() + servico = servico_academia(sessao) + + await servico.listar_filtrados( + offset=2, limit=10, tipo=TipoAcademia.CURSO, publicado=True + ) + + assert servico.repository.filtros_recebidos[-1] == { + 'offset': 2, + 'limit': 10, + 'filtros': {'tipo': TipoAcademia.CURSO, 'publicado': True}, + } + + +@pytest.mark.asyncio +async def test_servico_agenda_cria_evento_e_sincroniza_google_calendar(): + """Valida que servico agenda cria evento e sincroniza google calendar.""" + sessao = SessaoFake() + servico = servico_agenda(sessao) + + evento = await servico.criar_evento( + EventoAgendaCriar( + usuario_id=1, + titulo='Kickoff', + tipo=EventoTipo.REUNIAO, + data=date(2026, 5, 25), + hora=time(11, 0), + duracao_min=45, + ) + ) + + assert evento.google_event_id == 'google-1' + assert evento.google_link == 'https://calendar.test/google-1' + assert servico.calendario_google.criados[0]['titulo'] == 'Kickoff' + assert sessao.flushes == 1 + assert sessao.commits == 1 + + +@pytest.mark.asyncio +async def test_servico_agenda_lista_site_filtra_ordena_e_remove_google_duplicado(): + """Valida que servico agenda lista site filtra ordena e remove google duplicado.""" + sessao = SessaoFake() + servico = servico_agenda(sessao) + servico.calendario_google.eventos = [ + EventoGoogleCalendar( + id='sincronizado', + titulo='Duplicado', + descricao=None, + inicio=datetime(2026, 5, 25, 8, 0, tzinfo=timezone.utc), + fim=None, + link=None, + ), + EventoGoogleCalendar( + id='externo', + titulo='Google externo', + descricao='Descrição', + inicio=datetime(2026, 5, 25, 7, 0, tzinfo=timezone.utc), + fim=datetime(2026, 5, 25, 8, 0, tzinfo=timezone.utc), + link='https://calendar.test/externo', + ), + ] + + eventos = await servico.listar_eventos_site( + usuario_id=1, + data_inicio=date(2026, 5, 25), + data_fim=date(2026, 5, 25), + ) + + assert [evento.id for evento in eventos] == [ + 'google_calendar:externo', + 'local:1', + ] + assert eventos[0].duracao_min == 60 + assert eventos[1].google_event_id == 'sincronizado' + + +@pytest.mark.asyncio +async def test_servico_agenda_atualiza_evento_sincronizado_e_deleta_no_google(): + """Valida que servico agenda atualiza evento sincronizado e deleta no google.""" + sessao = SessaoFake() + servico = servico_agenda(sessao) + + evento = await servico.atualizar_evento( + 1, + EventoAgendaAtualizar(titulo='Kickoff atualizado'), + ) + await servico.deletar_evento(1) + + assert evento.titulo == 'Kickoff atualizado' + assert ( + servico.calendario_google.atualizados[0]['id_evento_google'] + == 'sincronizado' + ) + assert servico.calendario_google.deletados == ['google-1'] + assert sessao.flushes == 1 + assert sessao.commits == 2 + + +@pytest.mark.asyncio +async def test_servico_agenda_gerencia_solicitacoes(): + """Valida que servico agenda gerencia solicitacoes.""" + sessao = SessaoFake() + servico = servico_agenda(sessao) + + solicitacao = await servico.criar_solicitacao( + SolicitacaoReuniaoCriar( + remetente_id=2, + destinatario_id=1, + titulo='Reunião', + data=date(2026, 5, 27), + hora=time(14, 0), + ) + ) + recebidas = await servico.listar_solicitacoes_recebidas(1) + atualizada = await servico.atualizar_solicitacao( + 1, SolicitacaoReuniaoAtualizar(titulo='Reunião remarcada') + ) + await servico.deletar_solicitacao(1) + + assert solicitacao.destinatario_id == 1 + assert {item.destinatario_id for item in recebidas} == {1} + assert atualizada.titulo == 'Reunião remarcada' + assert sessao.commits == 3 + + +@pytest.mark.asyncio +async def test_servico_lead_repassa_filtros_para_repositorio(): + """Valida que servico lead repassa filtros para repositorio.""" + sessao = SessaoFake() + servico = servico_lead(sessao) + + await servico.listar_filtrados( + offset=1, + limit=5, + status=SituacaoLead.NOVO, + prioridade=LeadPrioridade.ALTA, + servico_id=3, + busca='maria', + ) + + assert servico.repository.filtros_recebidos[-1] == { + 'offset': 1, + 'limit': 5, + 'status': SituacaoLead.NOVO, + 'prioridade': LeadPrioridade.ALTA, + 'servico_id': 3, + 'busca': 'maria', + } + + +@pytest.mark.asyncio +async def test_servico_servico_rejeita_slug_duplicado(): + """Valida que servico servico rejeita slug duplicado.""" + sessao = SessaoFake() + servico = ServicoServico(sessao) + servico.repository = RepositorioFake() + servico.repository.slug_existente = True + servico.entregaveis = RepositorioFake() + + with pytest.raises(HTTPException) as exc: + await servico.criar(ServicoCriar(nome='Site', slug='site')) + + assert exc.value.status_code == 409 + assert sessao.commits == 0 + + +@pytest.mark.asyncio +async def test_servico_servico_gerencia_entregaveis(): + """Valida que servico servico gerencia entregaveis.""" + sessao = SessaoFake() + servico = ServicoServico(sessao) + servico.repository = RepositorioFake({1: Servico(id=1, nome='Site', slug='site')}) + servico.entregaveis = RepositorioFake( + {1: Entregavel(id=1, servico_id=1, descricao='Wireframe', ordem=1)} + ) + + criado = await servico.criar_entregavel( + EntregavelCriar(servico_id=1, descricao='Layout', ordem=2) + ) + listados = await servico.listar_entregaveis(1) + atualizado = await servico.atualizar_entregavel( + 1, EntregavelAtualizar(descricao='Wireframe revisado') + ) + await servico.deletar_entregavel(1) + + assert criado.servico_id == 1 + assert listados[0].servico_id == 1 + assert atualizado.descricao == 'Wireframe revisado' + assert sessao.commits == 3 + + +@pytest.mark.asyncio +async def test_servico_integracao_cria_atualiza_e_deleta_google_calendar(): + """Valida que servico integracao cria atualiza e deleta google calendar.""" + sessao = SessaoFake() + servico = ServicoIntegracao(sessao) + servico.repository = RepositorioFake( + { + 1: Integracao( + id=1, + nome='google_calendar', + status=SituacaoIntegracao.DESCONECTADO, + ) + } + ) + + criado = await servico.criar(IntegracaoCriar(status=SituacaoIntegracao.CONECTADO)) + atualizado = await servico.atualizar( + 1, IntegracaoAtualizar(status=SituacaoIntegracao.CONECTADO) + ) + await servico.deletar(1) + + assert criado.nome == 'google_calendar' + assert atualizado.status == SituacaoIntegracao.CONECTADO + assert sessao.commits == 3 + + +@pytest.mark.asyncio +async def test_servico_integracao_rejeita_google_calendar_duplicado(): + """Valida que servico integracao rejeita google calendar duplicado.""" + sessao = SessaoFake() + servico = ServicoIntegracao(sessao) + servico.repository = RepositorioFake() + servico.repository.nome_existente = True + + with pytest.raises(HTTPException) as exc: + await servico.criar(IntegracaoCriar()) + + assert exc.value.status_code == 409 + + +@pytest.mark.asyncio +async def test_servico_comunicado_marca_e_lista_leituras(): + """Valida que servico comunicado marca e lista leituras.""" + sessao = SessaoFake() + servico = ServicoComunicado(sessao) + servico.repository = RepositorioFake( + { + 1: Comunicado( + id=1, + autor_id=1, + titulo='Aviso', + conteudo='Conteudo', + alvo=ComunicadoAlvo.TODOS, + ) + } + ) + servico.leituras = RepositorioFake() + + criado = await servico.criar( + ComunicadoCriar(autor_id=1, titulo='Novo', conteudo='Texto') + ) + atualizado = await servico.atualizar( + 1, ComunicadoAtualizar(alvo=ComunicadoAlvo.CLIENTES) + ) + leitura = await servico.marcar_lido(1, usuario_id=7) + leituras = await servico.listar_leituras(1) + + assert criado.autor_id == 1 + assert atualizado.alvo == ComunicadoAlvo.CLIENTES + assert leitura.usuario_id == 7 + assert leituras == [leitura] + assert sessao.commits == 3 + + +@pytest.mark.asyncio +async def test_servico_lgpd_registra_e_lista_consentimentos(): + """Valida que servico lgpd registra e lista consentimentos.""" + sessao = SessaoFake() + servico = ServicoLgpd(sessao) + servico.repository = RepositorioFake() + + consentimento = await servico.registrar( + ConsentimentoLgpdCriar( + usuario_id=1, + tipo=ConsentimentoTipo.ESSENCIAL, + aceito=True, + ) + ) + listados = await servico.listar(offset=0, limit=10) + + assert isinstance(consentimento, ConsentimentoLgpd) + assert listados == [consentimento] + assert sessao.commits == 1 + + +@pytest.mark.asyncio +async def test_servico_gamificacao_atualiza_xp_e_nivel_do_funcionario(): + """Valida que servico gamificacao atualiza xp e nivel do funcionario.""" + sessao = SessaoFake() + funcionario = Funcionario(id=2, cargo='Dev', xp_total=450, nivel=1) + sessao.entidades[(Funcionario, 2)] = funcionario + servico = ServicoGamificacao(sessao) + servico.historicos = RepositorioFake() + + historico = await servico.registrar_xp( + HistoricoXpCriar(funcionario_id=2, acao='Entrega', xp=100) + ) + + assert isinstance(historico, HistoricoXp) + assert funcionario.xp_total == 550 + assert funcionario.nivel == 2 + assert sessao.commits == 1 + + +@pytest.mark.asyncio +async def test_servico_gamificacao_retorna_404_quando_funcionario_nao_existe(): + """Valida que servico gamificacao retorna 404 quando funcionario nao existe.""" + sessao = SessaoFake() + servico = ServicoGamificacao(sessao) + servico.historicos = RepositorioFake() + + with pytest.raises(HTTPException) as exc: + await servico.registrar_xp( + HistoricoXpCriar(funcionario_id=404, acao='Entrega', xp=10) + ) + + assert exc.value.status_code == 404 + assert sessao.commits == 0 + + +@pytest.mark.asyncio +async def test_servico_gamificacao_crud_regras_conquistas_e_desbloqueio(): + """Valida que servico gamificacao crud regras conquistas e desbloqueio.""" + sessao = SessaoFake() + servico = ServicoGamificacao(sessao) + servico.regras = RepositorioFake( + {1: RegraXp(id=1, tarefa='Entrega', complexidade=Complexidade.MEDIA, xp=50)} + ) + servico.conquistas = RepositorioFake( + {1: Conquista(id=1, nome='Primeira entrega', xp_bonus=100)} + ) + servico.fc = RepositorioFake() + + regra = await servico.criar_regra( + RegraXpCriar(tarefa='Review', complexidade=Complexidade.BAIXA, xp=5) + ) + regra_atualizada = await servico.atualizar_regra(1, RegraXpAtualizar(xp=80)) + conquista = await servico.criar_conquista(ConquistaCriar(nome='Bravo')) + conquista_atualizada = await servico.atualizar_conquista( + 1, ConquistaAtualizar(xp_bonus=150) + ) + desbloqueio = await servico.desbloquear_conquista(2, 1) + + assert regra.tarefa == 'Review' + assert regra_atualizada.xp == 80 + assert conquista.nome == 'Bravo' + assert conquista_atualizada.xp_bonus == 150 + assert desbloqueio.funcionario_id == 2 + assert sessao.commits == 5 + + +@pytest.mark.asyncio +async def test_servico_projeto_aplica_permissoes_de_visibilidade(): + """Valida que servico projeto aplica permissoes de visibilidade.""" + sessao = SessaoFake() + projeto = Projeto( + id=1, + cliente_id=10, + nome='Site', + status=SituacaoProjeto.PLANEJAMENTO, + progresso=0, + ) + servico = ServicoProjeto(sessao) + servico.repository = RepositorioFake({1: projeto}) + servico.equipe = RepositorioFake( + { + 1: ProjetoFuncionario( + projeto_id=1, + funcionario_id=20, + papel='Designer', + ) + } + ) + + assert await servico.obter_visivel(1, 99, PapelUsuario.ADMIN.value) is projeto + assert await servico.obter_visivel(1, 10, PapelUsuario.CLIENTE.value) is projeto + assert await servico.obter_visivel(1, 20, PapelUsuario.FUNCIONARIO.value) is projeto + with pytest.raises(HTTPException) as exc: + await servico.obter_visivel(1, 30, PapelUsuario.FUNCIONARIO.value) + + assert exc.value.status_code == 403 + + +@pytest.mark.asyncio +async def test_servico_projeto_gerencia_equipe(): + """Valida que servico projeto gerencia equipe.""" + sessao = SessaoFake() + servico = ServicoProjeto(sessao) + servico.repository = RepositorioFake( + { + 1: Projeto( + id=1, + cliente_id=10, + nome='Site', + status=SituacaoProjeto.PLANEJAMENTO, + progresso=0, + ) + } + ) + servico.equipe = RepositorioFake( + {1: ProjetoFuncionario(projeto_id=1, funcionario_id=7, papel='Dev')} + ) + + membro = await servico.adicionar_membro( + 1, ProjetoFuncionarioCriar(funcionario_id=8, papel='Design') + ) + equipe = await servico.listar_equipe(1) + await servico.remover_membro(1, 7) + + assert membro.funcionario_id == 8 + assert {item.funcionario_id for item in equipe} == {7, 8} + assert sessao.commits == 2 + + +@pytest.mark.asyncio +async def test_servico_tarefa_conclui_com_data_e_gerencia_comentarios_anexos(): + """Valida que servico tarefa conclui com data e gerencia comentarios anexos.""" + sessao = SessaoFake() + servico = ServicoTarefa(sessao) + servico.repository = RepositorioFake( + { + 1: Tarefa( + id=1, + projeto_id=1, + titulo='Layout', + status=SituacaoTarefa.A_FAZER, + complexidade=Complexidade.MEDIA, + prioridade=Prioridade.MEDIA, + ) + } + ) + servico.comentarios = RepositorioFake() + servico.anexos = RepositorioFake() + + tarefa = await servico.atualizar( + 1, TarefaAtualizar(status=SituacaoTarefa.CONCLUIDO) + ) + comentario = await servico.adicionar_comentario( + ComentarioCriar(tarefa_id=1, conteudo='feito'), autor_id=2 + ) + comentarios = await servico.listar_comentarios(1) + anexo = await servico.adicionar_anexo( + AnexoCriar(tarefa_id=1, nome='arquivo.pdf', url='https://arquivo.test/pdf') + ) + anexos = await servico.listar_anexos(1) + + assert tarefa.concluido_em is not None + assert isinstance(comentario, Comentario) + assert comentarios == [comentario] + assert isinstance(anexo, Anexo) + assert anexos == [anexo] + assert sessao.commits == 3 + + +@pytest.mark.asyncio +async def test_servico_tarefa_aplica_permissoes_de_acesso_e_gerenciamento(): + """Valida que servico tarefa aplica permissoes de acesso e gerenciamento.""" + sessao = SessaoFake() + tarefa = Tarefa( + id=1, + projeto_id=1, + responsavel_id=20, + titulo='Layout', + status=SituacaoTarefa.A_FAZER, + complexidade=Complexidade.MEDIA, + prioridade=Prioridade.MEDIA, + ) + sessao.entidades[(Projeto, 1)] = Projeto( + id=1, + cliente_id=10, + nome='Site', + status=SituacaoProjeto.PLANEJAMENTO, + progresso=0, + ) + servico = ServicoTarefa(sessao) + servico.repository = RepositorioFake({1: tarefa}) + + assert await servico.obter_visivel(1, 99, PapelUsuario.ADMIN.value) is tarefa + assert await servico.obter_visivel(1, 10, PapelUsuario.CLIENTE.value) is tarefa + assert ( + await servico.garantir_permissao_gerenciar_tarefa( + 1, 20, PapelUsuario.FUNCIONARIO.value + ) + is tarefa + ) + with pytest.raises(HTTPException) as exc: + await servico.garantir_permissao_gerenciar_tarefa( + 1, 30, PapelUsuario.CLIENTE.value + ) + + assert exc.value.status_code == 403 + + +@pytest.mark.asyncio +async def test_servico_usuario_cria_perfis_e_aprova_cadastro(): + """Valida que servico usuario cria perfis e aprova cadastro.""" + sessao = SessaoFake() + servico = ServicoUsuario(sessao) + servico.usuarios = RepositorioFake( + { + 1: Usuario( + id=1, + nome='Ana', + email='ana@example.com', + senha_hash='x', + role=PapelUsuario.CLIENTE, + status=SituacaoUsuario.PENDENTE, + ) + } + ) + servico.clientes = RepositorioFake() + servico.funcionarios = RepositorioFake() + servico.admins = RepositorioFake() + + usuario = await servico.criar( + UsuarioCriar( + nome='Cliente', + email='cliente@example.com', + senha='senha-segura', + role=PapelUsuario.CLIENTE, + cliente=DadosCliente(razao_social='Cliente LTDA', cnpj_cpf='123'), + ) + ) + funcionario = await servico.criar( + UsuarioCriar( + nome='Func', + email='func@example.com', + senha='senha-segura', + role=PapelUsuario.FUNCIONARIO, + funcionario=DadosFuncionario(cargo='Dev'), + ) + ) + aprovado = await servico.aprovar(1) + + assert usuario.role == PapelUsuario.CLIENTE + assert funcionario.role == PapelUsuario.FUNCIONARIO + assert servico.clientes.adicionados[0].id == usuario.id + assert servico.funcionarios.adicionados[0].id == funcionario.id + assert aprovado.status == SituacaoUsuario.ATIVO + assert sessao.commits == 3 + assert sessao.refreshes == [usuario, funcionario] + + +@pytest.mark.asyncio +async def test_servico_usuario_rejeita_email_duplicado_e_autocadastro_admin(): + """Valida que servico usuario rejeita email duplicado e autocadastro admin.""" + sessao = SessaoFake() + servico = ServicoUsuario(sessao) + servico.usuarios = RepositorioFake() + servico.usuarios.email_existente = True + + with pytest.raises(HTTPException) as conflito: + await servico.criar( + UsuarioCriar( + nome='Duplicado', + email='duplicado@example.com', + senha='senha-segura', + role=PapelUsuario.CLIENTE, + ) + ) + + servico.usuarios.email_existente = False + with pytest.raises(HTTPException) as invalido: + await servico.registrar( + UsuarioRegistrar( + nome='Admin', + email='admin@example.com', + senha='senha-segura', + role=PapelUsuario.ADMIN, + ) + ) + + assert conflito.value.status_code == 409 + assert invalido.value.status_code == 400 From e57a9c29b03bb6ef3813e85cf2766b8adc484fdb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Artur=20Mendon=C3=A7a=20Arruda?= Date: Tue, 26 May 2026 09:15:27 -0300 Subject: [PATCH 2/2] feat: adiciona rota de registro publico no controller --- api/modules/users/controller.py | 13 +++++++++++++ 1 file changed, 13 insertions(+) diff --git a/api/modules/users/controller.py b/api/modules/users/controller.py index 00c5365..a2968c5 100644 --- a/api/modules/users/controller.py +++ b/api/modules/users/controller.py @@ -8,6 +8,7 @@ from modules.users.schema import ( UsuarioAtualizar, UsuarioCriar, + UsuarioRegistrar, UsuarioResposta, ) from modules.users.service import ServicoUsuario @@ -112,3 +113,15 @@ async def aprovar(usuario_id: int, servico: DependenciaServico): async def deletar(usuario_id: int, servico: DependenciaServico): """Função para excluir um registro pelo ID.""" await servico.deletar(usuario_id) + + +@router.post( + '/registro', + response_model=UsuarioResposta, + status_code=status.HTTP_201_CREATED, + summary='Registra um novo usuário pendente de aprovação', +) +async def registrar(dados: UsuarioRegistrar, servico: DependenciaServico): + """Função para registrar um novo usuário pendente de aprovação.""" + return await servico.registrar(dados) +