Add 'gerenciador/' from commit '34cc9c75a57dedbf61a6e6a1ba1f35f76e72434b'

git-subtree-dir: gerenciador
git-subtree-mainline: f4f546c928
git-subtree-split: 34cc9c75a5
This commit is contained in:
gti
2026-09-29 14:43:12 -03:00
84 changed files with 8906 additions and 0 deletions
View File
+16
View File
@@ -0,0 +1,16 @@
"""
Configuração compartilhada dos testes. Define defaults de conexão iguais ao
.env.example ANTES de qualquer módulo do app ser importado, para os testes
rodarem direto contra o container de dev (licenciamento-mysql-dev, porta
3307) sem exigir setup manual — quem rodar `pytest` só precisa ter esse
container de pé (docker-compose.yml de licenciamento-n8n/ ou Banco para Prod/).
"""
import os
os.environ.setdefault("ADMIN_USER", "admin")
os.environ.setdefault("ADMIN_PASSWORD", "teste123")
os.environ.setdefault("DB_HOST", "127.0.0.1")
os.environ.setdefault("DB_PORT", "3307")
os.environ.setdefault("DB_NAME", "gtisoftw_clientes_dev")
os.environ.setdefault("DB_USER", "root")
os.environ.setdefault("DB_PASSWORD", "devpassword123")
+164
View File
@@ -0,0 +1,164 @@
import pytest
from fastapi import Depends, FastAPI, HTTPException
from fastapi.testclient import TestClient
from app import session
from app.auth import NaoAutenticado, UsuarioLogado, autenticar_credenciais, exigir_admin, exigir_login
from app.db import funcionarios as db_funcionarios
from app.db.connection import get_connection
app_teste = FastAPI()
@app_teste.get("/protegido")
def rota_protegida(usuario: UsuarioLogado = Depends(exigir_login)):
return {"usuario": str(usuario), "perfil": usuario.perfil}
@app_teste.get("/so-admin")
def rota_admin(usuario: UsuarioLogado = Depends(exigir_admin)):
return {"usuario": str(usuario)}
@app_teste.exception_handler(NaoAutenticado)
def _handler_nao_autenticado(request, exc):
raise HTTPException(status_code=401, detail="não autenticado")
client = TestClient(app_teste, follow_redirects=False)
def _cookie_sessao(login: str, nome: str, perfil: str) -> dict:
return {session.SESSION_COOKIE_NAME: session.criar_token(login, nome, perfil)}
def test_sem_cookie_retorna_401():
resposta = client.get("/protegido")
assert resposta.status_code == 401
def test_cookie_invalido_retorna_401():
resposta = client.get("/protegido", cookies={session.SESSION_COOKIE_NAME: "lixo"})
assert resposta.status_code == 401
def test_cookie_valido_retorna_dados_do_usuario():
resposta = client.get("/protegido", cookies=_cookie_sessao("admin", "Admin", "ADMIN"))
assert resposta.status_code == 200
assert resposta.json() == {"usuario": "Admin", "perfil": "ADMIN"}
def test_usuario_logado_str_retorna_nome():
usuario = UsuarioLogado(login="fulano", nome="Fulano da Silva", perfil="PADRAO")
assert str(usuario) == "Fulano da Silva"
def test_usuario_logado_eh_admin():
assert UsuarioLogado(login="a", nome="A", perfil="ADMIN").eh_admin is True
assert UsuarioLogado(login="a", nome="A", perfil="PADRAO").eh_admin is False
def test_exigir_admin_aceita_admin():
resposta = client.get("/so-admin", cookies=_cookie_sessao("admin", "Admin", "ADMIN"))
assert resposta.status_code == 200
def test_exigir_admin_rejeita_padrao(funcionario_teste):
# exigir_login agora reconfirma o perfil contra o banco, então o login
# precisa existir de fato em FUNCIONARIO_PAINEL (funcionario_teste é
# cadastrado com PERFIL="PADRAO" — ver fixture abaixo).
resposta = client.get("/so-admin", cookies=_cookie_sessao(funcionario_teste, "Fulano", "PADRAO"))
assert resposta.status_code == 403
def test_autenticar_credenciais_admin_env():
# ADMIN_USER/ADMIN_PASSWORD vêm de tests/conftest.py ("admin"/"teste123")
usuario = autenticar_credenciais("admin", "teste123")
assert usuario is not None
assert usuario.perfil == "ADMIN"
assert usuario.login == "admin"
def test_autenticar_credenciais_admin_env_senha_errada():
assert autenticar_credenciais("admin", "senha_errada") is None
def test_autenticar_credenciais_admin_env_vazio_nao_autentica_vazio(monkeypatch):
monkeypatch.setenv("ADMIN_USER", "")
monkeypatch.setenv("ADMIN_PASSWORD", "")
assert autenticar_credenciais("", "") is None
def _limpar_login(login: str) -> None:
conn = get_connection()
try:
with conn.cursor() as cur:
cur.execute("DELETE FROM FUNCIONARIO_PAINEL WHERE LOGIN = %s", (login,))
finally:
conn.close()
@pytest.fixture
def funcionario_teste():
login = "teste.login.auth"
_limpar_login(login)
funcionario_id = db_funcionarios.criar_funcionario("Fulano de Teste", login, "senhaForte123", "PADRAO")
yield login
_limpar_login(login)
def test_autenticar_credenciais_funcionario_ativo(funcionario_teste):
usuario = autenticar_credenciais(funcionario_teste, "senhaForte123")
assert usuario is not None
assert usuario.nome == "Fulano de Teste"
assert usuario.perfil == "PADRAO"
def test_autenticar_credenciais_funcionario_senha_errada(funcionario_teste):
assert autenticar_credenciais(funcionario_teste, "senhaErrada") is None
def test_autenticar_credenciais_funcionario_inativo(funcionario_teste):
funcionario = db_funcionarios.obter_por_login(funcionario_teste)
db_funcionarios.alternar_ativo(funcionario["ID"]) # desativa
assert autenticar_credenciais(funcionario_teste, "senhaForte123") is None
def test_autenticar_credenciais_login_inexistente():
assert autenticar_credenciais("nao-existe-esse-login", "qualquer") is None
def test_exigir_login_autentica_funcionario_ativo(funcionario_teste):
resposta = client.get("/protegido", cookies=_cookie_sessao(funcionario_teste, "Fulano de Teste", "PADRAO"))
assert resposta.status_code == 200
assert resposta.json() == {"usuario": "Fulano de Teste", "perfil": "PADRAO"}
def test_exigir_login_revoga_sessao_ao_desativar_funcionario(funcionario_teste):
# Cookie "válido" (assinatura e validade OK) gerado antes de desativar.
cookies = _cookie_sessao(funcionario_teste, "Fulano de Teste", "PADRAO")
assert client.get("/protegido", cookies=cookies).status_code == 200
funcionario = db_funcionarios.obter_por_login(funcionario_teste)
db_funcionarios.alternar_ativo(funcionario["ID"]) # desativa
resposta = client.get("/protegido", cookies=cookies)
assert resposta.status_code == 401
def test_exigir_admin_usa_perfil_atual_do_banco_nao_do_cookie(funcionario_teste):
# Cookie diz ADMIN, mas o funcionário no banco é PADRAO (ex: foi
# rebaixado depois de logar) — exigir_admin precisa rejeitar.
cookies = _cookie_sessao(funcionario_teste, "Fulano de Teste", "ADMIN")
resposta = client.get("/so-admin", cookies=cookies)
assert resposta.status_code == 403
def test_exigir_login_conta_bootstrap_nao_consulta_banco(monkeypatch):
def _falha_se_chamado(login):
raise AssertionError("exigir_login não deveria consultar o banco para a conta de bootstrap")
monkeypatch.setattr(db_funcionarios, "obter_por_login", _falha_se_chamado)
resposta = client.get("/protegido", cookies=_cookie_sessao("admin", "Admin", "ADMIN"))
assert resposta.status_code == 200
assert resposta.json() == {"usuario": "Admin", "perfil": "ADMIN"}
+75
View File
@@ -0,0 +1,75 @@
import httpx
from app import cnpj_lookup
def test_documento_com_tamanho_errado_nao_consulta(monkeypatch):
chamou = {}
monkeypatch.setattr(cnpj_lookup.httpx, "get", lambda *a, **k: chamou.setdefault("ok", True))
assert cnpj_lookup.consultar_cnpj("123") is None
assert "ok" not in chamou
def test_consulta_com_sucesso_mapeia_campos(monkeypatch):
class _RespostaFalsa:
status_code = 200
def json(self):
return {
"razao_social": "EMPRESA TESTE LTDA",
"nome_fantasia": "EMPRESA TESTE",
"logradouro": "RUA TESTE",
"numero": "123",
"bairro": "CENTRO",
"municipio": "FEIRA DE SANTANA",
"uf": "BA",
"cep": "44000000",
"ddd_telefone_1": "7533334444",
"email": "contato@empresateste.com",
}
monkeypatch.setattr(cnpj_lookup.httpx, "get", lambda url, timeout=None: _RespostaFalsa())
resultado = cnpj_lookup.consultar_cnpj("11222333000181")
assert resultado == {
"RAZAO": "EMPRESA TESTE LTDA",
"FANTASIA": "EMPRESA TESTE",
"ENDERECO": "RUA TESTE, 123",
"CIDADE": "FEIRA DE SANTANA",
"BAIRRO": "CENTRO",
"UF": "BA",
"CEP": "44000000",
"TELEFONE": "7533334444",
"EMAIL": "contato@empresateste.com",
}
def test_consulta_nao_encontrada_retorna_none(monkeypatch):
class _RespostaFalsa:
status_code = 404
def json(self):
return {}
monkeypatch.setattr(cnpj_lookup.httpx, "get", lambda url, timeout=None: _RespostaFalsa())
assert cnpj_lookup.consultar_cnpj("11222333000181") is None
def test_falha_de_rede_retorna_none(monkeypatch):
def _get_com_erro(url, timeout=None):
raise httpx.ConnectError("falhou")
monkeypatch.setattr(cnpj_lookup.httpx, "get", _get_com_erro)
assert cnpj_lookup.consultar_cnpj("11222333000181") is None
def test_sem_nome_fantasia_usa_razao_social(monkeypatch):
class _RespostaFalsa:
status_code = 200
def json(self):
return {"razao_social": "EMPRESA TESTE LTDA", "nome_fantasia": None}
monkeypatch.setattr(cnpj_lookup.httpx, "get", lambda url, timeout=None: _RespostaFalsa())
resultado = cnpj_lookup.consultar_cnpj("11222333000181")
assert resultado["FANTASIA"] == "EMPRESA TESTE LTDA"
+57
View File
@@ -0,0 +1,57 @@
"""
Testes de integração. Clientes_Chaves é MyISAM, sem FK com CLIENTE — o
teste cria e limpa a linha diretamente por SQL.
"""
import pytest
from app.db import chaves as db_chaves
from app.db.connection import get_connection
CODIGO_TESTE = "999903"
def _linha_vazia(codigo: str, cliente: str) -> dict:
linha = {f"M{i}": "" for i in range(1, 13)}
linha["ID"] = codigo
linha["Cliente"] = cliente
# "0000-00-00" (o valor real usado no legado) é rejeitado pelo MySQL em modo estrito
# (NO_ZERO_DATE) — usamos uma data valida qualquer, só para satisfazer a coluna NOT NULL.
linha["Prorrogacao"] = "2099-12-31"
return linha
@pytest.fixture
def chave_teste():
conn = get_connection()
try:
with conn.cursor() as cur:
linha = _linha_vazia(CODIGO_TESTE, "TESTE PAINEL")
linha["M8"] = "0B62CF51CDCF3328107092"
colunas = ", ".join(linha.keys())
marcadores = ", ".join(["%s"] * len(linha))
cur.execute(
f"INSERT INTO Clientes_Chaves ({colunas}) VALUES ({marcadores})",
list(linha.values()),
)
finally:
conn.close()
yield CODIGO_TESTE
conn = get_connection()
try:
with conn.cursor() as cur:
cur.execute("DELETE FROM Clientes_Chaves WHERE ID = %s", (CODIGO_TESTE,))
finally:
conn.close()
def test_status_chaves_existente(chave_teste):
status = db_chaves.status_chaves(CODIGO_TESTE)
assert status is not None
assert status["codigo"] == CODIGO_TESTE
assert status["cliente"] == "TESTE PAINEL"
assert status["meses"]["M8"] == "0B62CF51CDCF3328107092"
assert status["meses"]["M1"] is None
def test_status_chaves_inexistente():
assert db_chaves.status_chaves("999999") is None
+247
View File
@@ -0,0 +1,247 @@
"""
Testes de integração contra o banco de dev (ver test_db_connection.py).
Usa CODIGO na faixa "999900"-"999999", reservada para dados de teste deste
painel (fora da faixa real de clientes, que é sequencial e baixa — ver
licenciamento-n8n/docs/MAPEAMENTO_BANCO.md). Cada teste limpa o que criou.
"""
import pytest
from app.db import clientes as db_clientes
from app.db.connection import get_connection
CLIENTE_TESTE = {
"CODIGO": "999901",
"CNPJ": "11222333000181",
"RAZAO": "TESTE PAINEL RAZAO",
"FANTASIA": "TESTE PAINEL FANTASIA",
"IE": "ISENTO",
"ENDERECO": "RUA TESTE, 1",
"CIDADE": "FEIRA DE SANTANA",
"BAIRRO": "CENTRO",
"UF": "BA",
"CONTATO": "FULANO TESTE",
"TELEFONE": "75999090000",
"CELULAR": "75999090000",
"CEP": "44000000",
"IMPLANTACAO": "0.00",
"MENSALIDADE": "100.00",
"JUROS_MORA": "0.00",
"JUROS_MULTA": "0.00",
"DATA_CONTRATO": "2026-01-01",
"EMAIL": "teste@example.com",
"OBS": "Criado por teste automatizado — apagar se sobrar.",
"VENCIMENTO_CONTRATO": "2027-01-01",
"DIA_VENCIMENTO": "10",
"ATIVO": "1",
}
def _limpar(codigo: str) -> None:
conn = get_connection()
try:
with conn.cursor() as cur:
cur.execute("DELETE FROM CLIENTE WHERE CODIGO = %s", (codigo,))
finally:
conn.close()
@pytest.fixture
def cliente_id():
novo_id = db_clientes.criar_cliente(CLIENTE_TESTE)
yield novo_id
_limpar(CLIENTE_TESTE["CODIGO"])
def test_criar_e_obter_cliente(cliente_id):
cliente = db_clientes.obter_cliente(cliente_id)
assert cliente is not None
assert cliente["CODIGO"] == "999901"
assert cliente["RAZAO"] == "TESTE PAINEL RAZAO"
def test_obter_cliente_inexistente():
assert db_clientes.obter_cliente(999999999) is None
def test_listar_clientes_encontra_por_busca(cliente_id):
resultado = db_clientes.listar_clientes(busca="TESTE PAINEL RAZAO")
codigos = [c["CODIGO"] for c in resultado]
assert "999901" in codigos
def test_codigo_em_uso(cliente_id):
assert db_clientes.codigo_em_uso("999901") is True
assert db_clientes.codigo_em_uso("999901", ignorar_id=cliente_id) is False
assert db_clientes.codigo_em_uso("000000") is False
def test_cnpj_em_uso(cliente_id):
assert db_clientes.cnpj_em_uso("11222333000181") is True
assert db_clientes.cnpj_em_uso("11222333000181", ignorar_id=cliente_id) is False
assert db_clientes.cnpj_em_uso("99988877000166") is False
def test_cnpj_em_uso_ignora_pontuacao_de_registros_legados(cliente_id):
# CLIENTE.CNPJ tem formatação inconsistente em registros legados (ver
# licenciamento-n8n/docs/ANALISE_CLIENTE_IPREAL.md) — cnpj_em_uso
# precisa achar o duplicado mesmo se o legado ainda tem máscara.
conn = get_connection()
try:
with conn.cursor() as cur:
cur.execute("UPDATE CLIENTE SET CNPJ = %s WHERE ID = %s", ("11.222.333/0001-81", cliente_id))
finally:
conn.close()
assert db_clientes.cnpj_em_uso("11222333000181") is True
def test_atualizar_cliente(cliente_id):
dados = dict(CLIENTE_TESTE)
dados["FANTASIA"] = "TESTE PAINEL FANTASIA EDITADA"
db_clientes.atualizar_cliente(cliente_id, dados)
cliente = db_clientes.obter_cliente(cliente_id)
assert cliente["FANTASIA"] == "TESTE PAINEL FANTASIA EDITADA"
def test_excluir_cliente_sem_dependencias(cliente_id):
assert db_clientes.excluir_cliente(cliente_id) is True
assert db_clientes.obter_cliente(cliente_id) is None
def test_excluir_cliente_inexistente_retorna_false():
assert db_clientes.excluir_cliente(999999999) is False
def test_listar_clientes_busca_e_restrita_a_coluna_escolhida(cliente_id):
# ordenar_por="codigo" restringe a busca à coluna CODIGO — o termo
# "TESTE PAINEL RAZAO" existe em RAZAO/FANTASIA, mas não deve aparecer
# aqui porque a busca não olha para essas colunas quando é "codigo".
resultado = db_clientes.listar_clientes(busca="TESTE PAINEL RAZAO", ordenar_por="codigo")
assert len(resultado) == 0
# Buscando pelo próprio código, com ordenar_por="codigo", encontra.
resultado = db_clientes.listar_clientes(busca="999901", ordenar_por="codigo")
assert resultado[0]["CODIGO"] == "999901"
def test_listar_clientes_ordenacao_invalida_usa_padrao(cliente_id):
# "ordenar_por" fora da whitelist não deve quebrar nem virar SQL arbitrário
# — cai no padrão (RAZAO).
resultado = db_clientes.listar_clientes(busca="TESTE PAINEL RAZAO", ordenar_por="ID; DROP TABLE CLIENTE")
assert any(c["CODIGO"] == "999901" for c in resultado)
def test_listar_clientes_direcao_asc_e_desc(cliente_id):
# Dois clientes de teste com códigos diferentes, mesma busca por RAZAO,
# pra comparar a ordem entre asc/desc de forma determinística.
outro_id = db_clientes.criar_cliente(dict(CLIENTE_TESTE, CODIGO="999905"))
try:
asc = db_clientes.listar_clientes(busca="TESTE PAINEL RAZAO", ordenar_por="codigo", direcao="asc")
desc = db_clientes.listar_clientes(busca="TESTE PAINEL RAZAO", ordenar_por="codigo", direcao="desc")
codigos_asc = [c["CODIGO"] for c in asc]
codigos_desc = [c["CODIGO"] for c in desc]
assert codigos_asc == sorted(codigos_asc)
assert codigos_desc == sorted(codigos_desc, reverse=True)
finally:
_limpar("999905")
def test_listar_clientes_direcao_invalida_usa_padrao_asc(cliente_id):
# Não deve quebrar nem virar SQL arbitrário — cai em ASC.
resultado = db_clientes.listar_clientes(
busca="999901", ordenar_por="codigo", direcao="ASC; DROP TABLE CLIENTE"
)
assert resultado[0]["CODIGO"] == "999901"
def test_proximo_codigo_sugerido_ignora_faixa_de_teste(cliente_id):
# CLIENTE_TESTE usa CODIGO "999901" (faixa reservada para teste, >= 900000)
# — a sugestão não pode considerar isso, senão qualquer sobra de dado de
# teste no banco faria a sugestão pular para a casa dos 999xxx.
sugerido = db_clientes.proximo_codigo_sugerido()
assert int(sugerido) < 900000
assert len(sugerido) == 6
assert db_clientes.codigo_em_uso(sugerido) is False
def test_listar_clientes_filtro_ativo_so_ativos(cliente_id):
inativo_id = db_clientes.criar_cliente(dict(CLIENTE_TESTE, CODIGO="999902", ATIVO="0"))
try:
resultado = db_clientes.listar_clientes(busca="TESTE PAINEL", filtro_ativo=1)
codigos = [c["CODIGO"] for c in resultado]
assert "999901" in codigos
assert "999902" not in codigos
finally:
_limpar("999902")
def test_listar_clientes_filtro_inativo_so_inativos(cliente_id):
inativo_id = db_clientes.criar_cliente(dict(CLIENTE_TESTE, CODIGO="999903", ATIVO="0"))
try:
resultado = db_clientes.listar_clientes(busca="TESTE PAINEL", filtro_ativo=0)
codigos = [c["CODIGO"] for c in resultado]
assert "999901" not in codigos
assert "999903" in codigos
finally:
_limpar("999903")
def test_listar_clientes_filtro_none_traz_todos(cliente_id):
inativo_id = db_clientes.criar_cliente(dict(CLIENTE_TESTE, CODIGO="999904", ATIVO="0"))
try:
resultado = db_clientes.listar_clientes(busca="TESTE PAINEL", filtro_ativo=None)
codigos = [c["CODIGO"] for c in resultado]
assert "999901" in codigos
assert "999904" in codigos
finally:
_limpar("999904")
def test_contar_clientes_respeita_filtro_ativo(cliente_id):
inativo_id = db_clientes.criar_cliente(dict(CLIENTE_TESTE, CODIGO="999907", ATIVO="0"))
try:
assert db_clientes.contar_clientes(busca="TESTE PAINEL", filtro_ativo=1) == 1
assert db_clientes.contar_clientes(busca="TESTE PAINEL", filtro_ativo=0) == 1
assert db_clientes.contar_clientes(busca="TESTE PAINEL", filtro_ativo=None) == 2
finally:
_limpar("999907")
def test_criar_cliente_grava_ativo(cliente_id):
cliente = db_clientes.obter_cliente(cliente_id)
assert cliente["ATIVO"] == 1
def test_alternar_ativo_inverte_o_valor(cliente_id):
assert db_clientes.obter_cliente(cliente_id)["ATIVO"] == 1
assert db_clientes.alternar_ativo(cliente_id) is True
assert db_clientes.obter_cliente(cliente_id)["ATIVO"] == 0
assert db_clientes.alternar_ativo(cliente_id) is True
assert db_clientes.obter_cliente(cliente_id)["ATIVO"] == 1
def test_alternar_ativo_id_inexistente_retorna_false():
assert db_clientes.alternar_ativo(999999999) is False
def test_excluir_cliente_com_contato_bloqueia(cliente_id):
conn = get_connection()
try:
with conn.cursor() as cur:
cur.execute(
"INSERT INTO CLIENTE_CONTATO (CLIENTE_ID, NOME, TELEFONE) VALUES (%s, %s, %s)",
(cliente_id, "Contato Teste", "75999090001"),
)
finally:
conn.close()
try:
with pytest.raises(db_clientes.ClienteComDependenciasError):
db_clientes.excluir_cliente(cliente_id)
finally:
conn = get_connection()
try:
with conn.cursor() as cur:
cur.execute("DELETE FROM CLIENTE_CONTATO WHERE CLIENTE_ID = %s", (cliente_id,))
finally:
conn.close()
+17
View File
@@ -0,0 +1,17 @@
"""
Requer o container de dev rodando (licenciamento-mysql-dev, porta 3307 —
ver licenciamento-n8n/docs/configurar-banco-sistema.md). tests/conftest.py
já configura as env vars de conexão com os mesmos defaults do .env.example.
"""
from app.db.connection import get_connection
def test_conexao_abre_e_consulta():
conn = get_connection()
try:
with conn.cursor() as cur:
cur.execute("SELECT 1 AS um")
row = cur.fetchone()
assert row == {"um": 1}
finally:
conn.close()
+67
View File
@@ -0,0 +1,67 @@
"""
Testes de integração. Cria um CLIENTE de teste (CODIGO 999902) só para
pendurar os contatos, e limpa tudo ao final.
"""
import pytest
from app.db import contatos as db_contatos
from app.db import clientes as db_clientes
from tests.test_db_clientes import CLIENTE_TESTE, _limpar
@pytest.fixture
def cliente_id():
dados = dict(CLIENTE_TESTE, CODIGO="999902")
novo_id = db_clientes.criar_cliente(dados)
yield novo_id
# Clean up contacts first due to FK constraint, then client
from app.db.connection import get_connection
conn = get_connection()
try:
with conn.cursor() as cur:
cur.execute("DELETE FROM CLIENTE_CONTATO WHERE CLIENTE_ID = %s", (novo_id,))
cur.execute("DELETE FROM CLIENTE WHERE ID = %s", (novo_id,))
finally:
conn.close()
def test_inserir_e_listar_contato(cliente_id):
contato_id = db_contatos.inserir_contato(cliente_id, "Fulano", "75999090001")
contatos = db_contatos.listar_contatos_por_cliente(cliente_id)
assert len(contatos) == 1
assert contatos[0]["ID"] == contato_id
assert contatos[0]["NOME"] == "Fulano"
assert contatos[0]["TELEFONE"] == "75999090001"
def test_contato_duplicado(cliente_id):
db_contatos.inserir_contato(cliente_id, "Fulano", "75999090001")
assert db_contatos.contato_duplicado(cliente_id, "75999090001") is True
assert db_contatos.contato_duplicado(cliente_id, "75999090002") is False
def test_contato_duplicado_ignora_proprio_id(cliente_id):
contato_id = db_contatos.inserir_contato(cliente_id, "Fulano", "75999090001")
assert db_contatos.contato_duplicado(cliente_id, "75999090001", ignorar_id=contato_id) is False
def test_atualizar_contato(cliente_id):
contato_id = db_contatos.inserir_contato(cliente_id, "Fulano", "75999090001")
assert db_contatos.atualizar_contato(contato_id, cliente_id, "Fulano Editado", "75999090003") is True
contatos = db_contatos.listar_contatos_por_cliente(cliente_id)
assert contatos[0]["NOME"] == "Fulano Editado"
assert contatos[0]["TELEFONE"] == "75999090003"
def test_atualizar_contato_inexistente_retorna_false(cliente_id):
assert db_contatos.atualizar_contato(999999, cliente_id, "Fulano", "75999090001") is False
def test_excluir_contato(cliente_id):
contato_id = db_contatos.inserir_contato(cliente_id, "Fulano", "75999090001")
assert db_contatos.excluir_contato(contato_id, cliente_id) is True
assert db_contatos.listar_contatos_por_cliente(cliente_id) == []
def test_excluir_contato_inexistente_retorna_false(cliente_id):
assert db_contatos.excluir_contato(999999, cliente_id) is False
+98
View File
@@ -0,0 +1,98 @@
"""Testes de integração contra o banco de dev (ver test_db_connection.py)."""
import pytest
from app.db import funcionarios as db_funcionarios
from app.db.connection import get_connection
LOGIN_TESTE = "teste.db.funcionarios"
def _limpar():
conn = get_connection()
try:
with conn.cursor() as cur:
cur.execute("DELETE FROM FUNCIONARIO_PAINEL WHERE LOGIN LIKE 'teste.db.funcionarios%'")
finally:
conn.close()
@pytest.fixture
def funcionario_id():
_limpar()
novo_id = db_funcionarios.criar_funcionario("Fulano Teste", LOGIN_TESTE, "senhaForte123", "PADRAO")
yield novo_id
_limpar()
def test_criar_e_obter_funcionario(funcionario_id):
f = db_funcionarios.obter_funcionario(funcionario_id)
assert f is not None
assert f["NOME"] == "Fulano Teste"
assert f["LOGIN"] == LOGIN_TESTE
assert f["ATIVO"] == 1
assert f["PERFIL"] == "PADRAO"
assert "SENHA_HASH" not in f # obter_funcionario não expõe o hash
def test_obter_por_login_inclui_senha_hash_e_perfil(funcionario_id):
f = db_funcionarios.obter_por_login(LOGIN_TESTE)
assert f is not None
assert "SENHA_HASH" in f
assert f["SENHA_HASH"] != "senhaForte123" # nunca em texto puro
assert f["PERFIL"] == "PADRAO"
def test_obter_por_login_inexistente(funcionario_id):
assert db_funcionarios.obter_por_login("nao-existe-esse-login-xyz") is None
def test_login_em_uso(funcionario_id):
assert db_funcionarios.login_em_uso(LOGIN_TESTE) is True
assert db_funcionarios.login_em_uso(LOGIN_TESTE, ignorar_id=funcionario_id) is False
assert db_funcionarios.login_em_uso("login-que-nao-existe") is False
def test_listar_funcionarios_inclui_o_criado(funcionario_id):
encontrados = {f["LOGIN"]: f for f in db_funcionarios.listar_funcionarios()}
assert LOGIN_TESTE in encontrados
assert encontrados[LOGIN_TESTE]["PERFIL"] == "PADRAO"
def test_atualizar_funcionario(funcionario_id):
novo_login = "teste.db.funcionarios.editado"
db_funcionarios.atualizar_funcionario(funcionario_id, "Fulano Editado", novo_login, "ADMIN")
f = db_funcionarios.obter_funcionario(funcionario_id)
assert f["NOME"] == "Fulano Editado"
assert f["LOGIN"] == novo_login
assert f["PERFIL"] == "ADMIN"
def test_redefinir_senha(funcionario_id):
hash_antes = db_funcionarios.obter_por_login(LOGIN_TESTE)["SENHA_HASH"]
db_funcionarios.redefinir_senha(funcionario_id, "outraSenha456")
hash_depois = db_funcionarios.obter_por_login(LOGIN_TESTE)["SENHA_HASH"]
assert hash_antes != hash_depois
def test_alternar_ativo(funcionario_id):
assert db_funcionarios.obter_funcionario(funcionario_id)["ATIVO"] == 1
db_funcionarios.alternar_ativo(funcionario_id)
assert db_funcionarios.obter_funcionario(funcionario_id)["ATIVO"] == 0
db_funcionarios.alternar_ativo(funcionario_id)
assert db_funcionarios.obter_funcionario(funcionario_id)["ATIVO"] == 1
def test_contar_admins_ativos(funcionario_id):
# funcionario_id é PADRAO (fixture acima) — não conta.
antes = db_funcionarios.contar_admins_ativos()
admin_id = db_funcionarios.criar_funcionario(
"Admin Teste", "teste.db.funcionarios.admin", "senhaForte123", "ADMIN"
)
try:
assert db_funcionarios.contar_admins_ativos() == antes + 1
db_funcionarios.alternar_ativo(admin_id) # desativa
assert db_funcionarios.contar_admins_ativos() == antes
finally:
_limpar()
+51
View File
@@ -0,0 +1,51 @@
import pytest
from app.db import log as db_log
from app.db.connection import get_connection
def _limpar_teste():
conn = get_connection()
try:
with conn.cursor() as cur:
cur.execute("DELETE FROM LOG_PAINEL WHERE USUARIO = 'teste-log'")
finally:
conn.close()
@pytest.fixture
def limpar():
_limpar_teste()
yield
_limpar_teste()
def test_registrar_e_listar(limpar):
db_log.registrar("teste-log", "criar", "cliente", "1", True)
db_log.registrar("teste-log", "excluir", "cliente", "1", False, "tem contato")
linhas = db_log.listar(limite=10)
linhas_teste = [l for l in linhas if l["USUARIO"] == "teste-log"]
assert len(linhas_teste) == 2
# ORDER BY CRIADO_EM DESC, ID DESC -> a mais recente (excluir) vem primeiro.
assert linhas_teste[0]["ACAO"] == "excluir"
assert linhas_teste[0]["SUCESSO"] == 0
assert linhas_teste[0]["DETALHE"] == "tem contato"
assert linhas_teste[1]["ACAO"] == "criar"
assert linhas_teste[1]["SUCESSO"] == 1
def test_contar(limpar):
antes = db_log.contar()
db_log.registrar("teste-log", "criar", "cliente", "1", True)
depois = db_log.contar()
assert depois == antes + 1
def test_limpar_apaga_tudo(limpar):
db_log.registrar("teste-log", "criar", "cliente", "1", True)
assert db_log.contar() > 0
db_log.limpar()
assert db_log.contar() == 0
+41
View File
@@ -0,0 +1,41 @@
from app.gerador_chave import gerar_instalacao, gerar_reativacao
def test_gerar_instalacao_bate_com_exemplo_real_byte_dc():
assert gerar_instalacao("000001", "ALUGUEL", byte_inicial=0xDC) == "DC45092C"
def test_gerar_instalacao_bate_com_exemplo_real_byte_2e():
assert gerar_instalacao("000001", "ALUGUEL", byte_inicial=0x2E) == "2E35083D"
def test_gerar_instalacao_bate_com_exemplo_real_byte_3d():
assert gerar_instalacao("000001", "ALUGUEL", byte_inicial=0x3D) == "3D25090C"
def test_gerar_instalacao_compra_usa_modalidade_00():
# Mesmo byte inicial, modalidade diferente -> cifrado diferente do ALUGUEL.
aluguel = gerar_instalacao("000001", "ALUGUEL", byte_inicial=0xDC)
compra = gerar_instalacao("000001", "COMPRA", byte_inicial=0xDC)
assert aluguel != compra
assert compra.startswith("DC") # byte inicial some no começo, sem cifrar
def test_gerar_instalacao_sem_byte_inicial_gera_aleatorio_valido():
chave = gerar_instalacao("000001", "ALUGUEL")
assert len(chave) == 8
assert chave[3:5].isdigit() # valor_cliente embutido no meio
def test_gerar_reativacao_e_deterministico_com_byte_fixo():
chave1 = gerar_reativacao("000123", 8, 2026, byte_inicial=0x10)
chave2 = gerar_reativacao("000123", 8, 2026, byte_inicial=0x10)
assert chave1 == chave2
assert len(chave1) == 22 # 11 bytes (1 aleatorio + 10 do texto) em hex
def test_gerar_reativacao_sem_byte_inicial_varia():
# Estatisticamente, gerar várias vezes sem byte fixo não deve sempre
# bater igual (chance de colisão é 1/256).
chaves = {gerar_reativacao("000123", 8, 2026) for _ in range(20)}
assert len(chaves) > 1
+81
View File
@@ -0,0 +1,81 @@
"""GET /internal/fiscal/clients — endpoint read-only server-to-server para o
GTI Fiscal espelhar o cadastro de clientes. Usa CODIGO na faixa de teste
"999950"-"999959" (ver tests/test_db_clientes.py) e limpa o que criou."""
import pytest
from fastapi.testclient import TestClient
from app.db import clientes as db_clientes
from app.db.connection import get_connection
from app.main import app
TOKEN = "teste-token-fiscal-interno-suficientemente-longo"
BASE = {
"CNPJ": "11222333000181", "RAZAO": "TESTE INTERNO RAZAO", "IE": "ISENTO",
"ENDERECO": "RUA TESTE, 1", "CIDADE": "FEIRA DE SANTANA", "BAIRRO": "CENTRO", "UF": "BA",
"CONTATO": "FULANO", "TELEFONE": "75999090000", "CELULAR": "75999090000", "CEP": "44000000",
"IMPLANTACAO": "0.00", "MENSALIDADE": "100.00", "JUROS_MORA": "0.00", "JUROS_MULTA": "0.00",
"DATA_CONTRATO": "2026-01-01", "EMAIL": "teste@example.com", "OBS": "teste automatizado",
"VENCIMENTO_CONTRATO": "2027-01-01", "DIA_VENCIMENTO": "10",
}
def _limpar(codigos):
conn = get_connection()
try:
with conn.cursor() as cur:
cur.execute(f"DELETE FROM CLIENTE WHERE CODIGO IN ({','.join(['%s'] * len(codigos))})", codigos)
finally:
conn.close()
@pytest.fixture
def clientes_teste():
ativo_id = db_clientes.criar_cliente({**BASE, "CODIGO": "999950", "FANTASIA": "Ativo Fiscal Teste", "ATIVO": "1"})
inativo_id = db_clientes.criar_cliente({**BASE, "CODIGO": "999951", "FANTASIA": "", "RAZAO": "Inativo Fiscal Teste Razao", "ATIVO": "0"})
yield ativo_id, inativo_id
_limpar(["999950", "999951"])
@pytest.fixture
def client(monkeypatch):
monkeypatch.setenv("FISCAL_INTERNAL_API_TOKEN", TOKEN)
from app import internal_fiscal
internal_fiscal._falhas.clear()
return TestClient(app)
def test_returns_active_and_inactive_clients_with_minimal_fields(client, clientes_teste):
ativo_id, inativo_id = clientes_teste
response = client.get("/internal/fiscal/clients", headers={"Authorization": f"Bearer {TOKEN}"})
assert response.status_code == 200
by_id = {row["external_client_id"]: row for row in response.json()["clients"]}
assert by_id[ativo_id] == {"external_client_id": ativo_id, "code": "999950", "display_name": "Ativo Fiscal Teste", "cnpj": "11222333000181", "active": True}
assert by_id[inativo_id] == {"external_client_id": inativo_id, "code": "999951", "display_name": "Inativo Fiscal Teste Razao", "cnpj": "11222333000181", "active": False}
# never leaks fields the Fiscal doesn't need
assert set(by_id[ativo_id]) == {"external_client_id", "code", "display_name", "cnpj", "active"}
def test_missing_token_rejected(client, clientes_teste):
assert client.get("/internal/fiscal/clients").status_code == 401
def test_wrong_token_rejected(client, clientes_teste):
response = client.get("/internal/fiscal/clients", headers={"Authorization": "Bearer wrong-token"})
assert response.status_code == 401
def test_endpoint_disabled_when_token_not_configured(monkeypatch):
monkeypatch.delenv("FISCAL_INTERNAL_API_TOKEN", raising=False)
from app import internal_fiscal
internal_fiscal._falhas.clear()
client = TestClient(app)
response = client.get("/internal/fiscal/clients", headers={"Authorization": "Bearer anything"})
assert response.status_code == 503
def test_rate_limited_after_repeated_invalid_attempts(client):
from app import internal_fiscal
for _ in range(internal_fiscal.MAX_TENTATIVAS):
assert client.get("/internal/fiscal/clients", headers={"Authorization": "Bearer wrong"}).status_code == 401
assert client.get("/internal/fiscal/clients", headers={"Authorization": "Bearer wrong"}).status_code == 429
+130
View File
@@ -0,0 +1,130 @@
from fastapi.testclient import TestClient
from app import session
from app.main import app
client = TestClient(app)
def _cookie_sessao_admin() -> dict:
return {session.SESSION_COOKIE_NAME: session.criar_token("admin", "admin", "ADMIN")}
def test_raiz_sem_login_redireciona_para_login():
resposta = client.get("/", follow_redirects=False)
assert resposta.status_code == 303
assert resposta.headers["location"] == "/login?next=/"
def test_docs_desabilitado_retorna_404():
assert client.get("/docs").status_code == 404
assert client.get("/redoc").status_code == 404
assert client.get("/openapi.json").status_code == 404
def test_raiz_redireciona_para_clientes_com_login(monkeypatch):
from app.routers import clientes as router_clientes
monkeypatch.setattr(router_clientes.db_clientes, "listar_clientes", lambda busca=None, limite=100, offset=0, ordenar_por="razao", direcao="asc", filtro_ativo=1: [])
monkeypatch.setattr(router_clientes.db_clientes, "contar_clientes", lambda busca=None, ordenar_por="razao", filtro_ativo=1: 0)
resposta = client.get("/", cookies=_cookie_sessao_admin(), follow_redirects=False)
assert resposta.status_code == 307
assert resposta.headers["location"] == "/clientes"
def test_csrf_middleware_define_cookie_em_qualquer_resposta():
client.cookies.clear()
resposta = client.get("/docs")
assert resposta.status_code == 404
assert "csrf_token" in resposta.cookies
def test_csrf_bloqueia_post_com_token_que_nao_bate_com_cookie():
client.cookies.clear()
resposta = client.post(
"/clientes/1/excluir", cookies=_cookie_sessao_admin(), data={"csrf_token": "token-forjado-sem-cookie"}
)
assert resposta.status_code == 403
def test_csrf_libera_post_quando_token_bate_com_cookie(monkeypatch):
from app.routers import clientes as router_clientes
monkeypatch.setattr(router_clientes.db_clientes, "excluir_cliente", lambda cliente_id: None)
client.cookies.clear()
client.get("/docs") # só para o middleware definir o cookie csrf_token no client de teste
token_cookie = client.cookies.get("csrf_token")
assert token_cookie
cookies = {**_cookie_sessao_admin(), "csrf_token": token_cookie}
resposta = client.post(
"/clientes/1/excluir", cookies=cookies, data={"csrf_token": token_cookie}, follow_redirects=False
)
assert resposta.status_code == 303
def test_pagina_clientes_tem_csrf_token_preenchido_no_botao_sair(monkeypatch):
"""
Regressão: GET /clientes (a página onde todo login cai) não passava
csrf_token no contexto do template, então o formulário de logout em
base.html renderizava com value="" e POST /logout era rejeitado com
403 por app/csrf.py. base.html agora usa
"{{ csrf_token or request.state.csrf_token }}" como fallback.
"""
from app.routers import clientes as router_clientes
monkeypatch.setattr(
router_clientes.db_clientes,
"listar_clientes",
lambda busca=None, limite=100, offset=0, ordenar_por="razao", direcao="asc", filtro_ativo=1: [],
)
monkeypatch.setattr(router_clientes.db_clientes, "contar_clientes", lambda busca=None, ordenar_por="razao", filtro_ativo=1: 0)
client.cookies.clear()
resposta = client.get("/clientes", cookies=_cookie_sessao_admin())
assert resposta.status_code == 200
assert 'name="csrf_token" value=""' not in resposta.text
import re
match = re.search(r'name="csrf_token" value="([^"]*)"', resposta.text)
assert match is not None
assert len(match.group(1)) > 0
def test_logout_funciona_depois_de_visitar_clientes(monkeypatch):
"""
Regressão complementar: simula o fluxo real — visitar /clientes (onde
o cookie csrf_token é definido pelo middleware) e então enviar
POST /logout usando o csrf_token real do cookie do navegador, sem
passar csrf_token manualmente no contexto do template.
"""
from app.routers import clientes as router_clientes
monkeypatch.setattr(
router_clientes.db_clientes,
"listar_clientes",
lambda busca=None, limite=100, offset=0, ordenar_por="razao", direcao="asc", filtro_ativo=1: [],
)
monkeypatch.setattr(router_clientes.db_clientes, "contar_clientes", lambda busca=None, ordenar_por="razao", filtro_ativo=1: 0)
client.cookies.clear()
client.cookies.update(_cookie_sessao_admin())
resposta_clientes = client.get("/clientes")
assert resposta_clientes.status_code == 200
token_cookie = client.cookies.get("csrf_token")
assert token_cookie
resposta_logout = client.post("/logout", data={"csrf_token": token_cookie}, follow_redirects=False)
assert resposta_logout.status_code == 303
def test_estilo_css_tem_versao_de_cache_busting():
resposta = client.get("/login")
assert resposta.status_code == 200
import re
m = re.search(r'/static/estilo\.css\?v=(\d+)', resposta.text)
assert m is not None, "link do estilo.css deveria ter ?v=<versao>"
assert m.group(1) != "0"
+60
View File
@@ -0,0 +1,60 @@
import httpx
import pytest
from app import n8n_client
def test_nao_configurado_retorna_vazio(monkeypatch):
monkeypatch.delenv("N8N_API_URL", raising=False)
monkeypatch.delenv("N8N_API_KEY", raising=False)
assert n8n_client.configurado() is False
assert n8n_client.listar_envios() == []
def test_configurado_com_ambas_variaveis(monkeypatch):
monkeypatch.setenv("N8N_API_URL", "https://devn8n.exemplo.com")
monkeypatch.setenv("N8N_API_KEY", "chave-teste")
assert n8n_client.configurado() is True
def test_listar_envios_falha_de_rede_retorna_vazio(monkeypatch):
monkeypatch.setenv("N8N_API_URL", "https://devn8n.exemplo.com")
monkeypatch.setenv("N8N_API_KEY", "chave-teste")
def _get_com_erro(*args, **kwargs):
raise httpx.ConnectError("falhou")
monkeypatch.setattr(n8n_client.httpx, "get", _get_com_erro)
assert n8n_client.listar_envios() == []
def test_listar_envios_sucesso_retorna_linhas(monkeypatch):
monkeypatch.setenv("N8N_API_URL", "https://devn8n.exemplo.com/")
monkeypatch.setenv("N8N_API_KEY", "chave-teste")
linhas = [{"codigo_cliente": "000001", "status": "enviado"}]
class _RespostaFalsa:
def raise_for_status(self):
pass
def json(self):
return {"data": linhas}
capturado = {}
def _get_falso(url, headers=None, params=None, timeout=None):
capturado["url"] = url
capturado["headers"] = headers
capturado["params"] = params
return _RespostaFalsa()
monkeypatch.setattr(n8n_client.httpx, "get", _get_falso)
resultado = n8n_client.listar_envios(limite=10)
assert resultado == linhas
assert capturado["url"] == "https://devn8n.exemplo.com/api/v1/data-tables/bJHhFUDwFJ30YJVt/rows"
assert capturado["headers"]["X-N8N-API-KEY"] == "chave-teste"
# Regression guard: a API do n8n usa "limit", não "take" (achado real
# testando contra a instância de verdade — "take" retorna 400).
assert capturado["params"] == {"limit": 10, "sortBy": "data_processamento:desc"}
+124
View File
@@ -0,0 +1,124 @@
"""
Fixtures de CPF/CNPJ verificadas manualmente:
- CPF "11144477735": DV clássico módulo 11, exemplo público amplamente usado
em validadores de CPF.
- CNPJ numérico "11222333000181": exemplo público amplamente usado em
validadores de CNPJ.
- CNPJ alfanumérico "00000000E08G12": primeiro CNPJ alfanumérico realmente
emitido no Brasil (Banco do Bradil, 02/08/2026) — confirma que o algoritmo
bate com um caso real, não só um exemplo didático.
"""
from app.normalize import (
normalizar_documento,
normalizar_telefone,
validar_dv_cpf,
validar_dv_cnpj,
)
def test_normaliza_cpf_com_mascara():
doc, valido, tipo = normalizar_documento("111.444.777-35")
assert doc == "11144477735"
assert valido is True
assert tipo == "CPF"
def test_normaliza_cnpj_numerico_com_mascara():
doc, valido, tipo = normalizar_documento("11.222.333/0001-81")
assert doc == "11222333000181"
assert valido is True
assert tipo == "CNPJ"
def test_normaliza_cnpj_alfanumerico_com_mascara():
doc, valido, tipo = normalizar_documento("00.000.000/E08G-12")
assert doc == "00000000E08G12"
assert valido is True
assert tipo == "CNPJ"
def test_documento_invalido_tamanho_errado():
doc, valido, tipo = normalizar_documento("123")
assert valido is False
assert tipo is None
def test_documento_none():
doc, valido, tipo = normalizar_documento(None)
assert doc == ""
assert valido is False
assert tipo is None
def test_normaliza_telefone():
assert normalizar_telefone("(75) 99909-1780") == "75999091780"
assert normalizar_telefone(None) == ""
def test_validar_dv_cpf_correto():
assert validar_dv_cpf("11144477735") is True
def test_validar_dv_cpf_incorreto():
assert validar_dv_cpf("11144477734") is False
def test_validar_dv_cnpj_numerico_correto():
assert validar_dv_cnpj("11222333000181") is True
def test_validar_dv_cnpj_numerico_incorreto():
assert validar_dv_cnpj("11222333000180") is False
def test_validar_dv_cnpj_alfanumerico_correto():
# Primeiro CNPJ alfanumérico real emitido no Brasil.
assert validar_dv_cnpj("00000000E08G12") is True
def test_validar_dv_cnpj_alfanumerico_incorreto():
assert validar_dv_cnpj("00000000E08G13") is False
from app.normalize import normalizar_telefone_com_ddi, normalizar_uf, normalizar_cep
def test_normaliza_telefone_com_ddi_prefixa_55_quando_ausente():
assert normalizar_telefone_com_ddi("(75) 3333-4444") == "557533334444"
assert normalizar_telefone_com_ddi("(75) 99909-1780") == "5575999091780"
def test_normaliza_telefone_com_ddi_nao_duplica_55_ja_presente():
assert normalizar_telefone_com_ddi("55 75 99909-1780") == "5575999091780"
assert normalizar_telefone_com_ddi("5575333344440") != "555575333344440"
def test_normaliza_telefone_com_ddi_vazio_continua_vazio():
assert normalizar_telefone_com_ddi("") == ""
assert normalizar_telefone_com_ddi(None) == ""
def test_normaliza_telefone_com_ddi_e_idempotente():
"""Aplicar a função duas vezes seguidas nunca deve dobrar o DDI."""
casos = ["(75) 3333-4444", "(75) 99909-1780", "5575999091780", "3333-4444", ""]
for bruto in casos:
uma_vez = normalizar_telefone_com_ddi(bruto)
duas_vezes = normalizar_telefone_com_ddi(uma_vez)
assert duas_vezes == uma_vez, f"não é idempotente para {bruto!r}: {uma_vez!r} -> {duas_vezes!r}"
def test_normaliza_uf_maiusculas_e_trunca():
assert normalizar_uf("ba") == "BA"
assert normalizar_uf(" sp ") == "SP"
assert normalizar_uf("SAO") == "SA"
def test_normaliza_uf_remove_numeros_e_simbolos():
assert normalizar_uf("B4!") == "B"
assert normalizar_uf(None) == ""
def test_normaliza_cep_so_digitos_e_trunca():
assert normalizar_cep("44430-174") == "44430174"
assert normalizar_cep("44.430-174-extra") == "44430174"
assert normalizar_cep(None) == ""
+58
View File
@@ -0,0 +1,58 @@
from app import rate_limit
def _origem_unica() -> str:
"""Cada teste usa uma chave própria — evita que o dict em memória do
módulo (compartilhado entre testes) vaze estado de um teste pro outro."""
import uuid
return f"teste-{uuid.uuid4()}"
def test_nao_bloqueado_sem_nenhuma_falha():
assert rate_limit.bloqueado(_origem_unica()) is False
def test_nao_bloqueado_abaixo_do_limite():
origem = _origem_unica()
agora = 1_000_000.0
for _ in range(rate_limit.MAX_TENTATIVAS - 1):
rate_limit.registrar_falha(origem, agora=agora)
assert rate_limit.bloqueado(origem, agora=agora) is False
def test_bloqueado_ao_atingir_o_limite():
origem = _origem_unica()
agora = 1_000_000.0
for _ in range(rate_limit.MAX_TENTATIVAS):
rate_limit.registrar_falha(origem, agora=agora)
assert rate_limit.bloqueado(origem, agora=agora) is True
def test_falhas_fora_da_janela_nao_contam():
origem = _origem_unica()
agora = 1_000_000.0
fora_da_janela = agora - rate_limit.JANELA_SEGUNDOS - 1
for _ in range(rate_limit.MAX_TENTATIVAS):
rate_limit.registrar_falha(origem, agora=fora_da_janela)
assert rate_limit.bloqueado(origem, agora=agora) is False
def test_limpar_reseta_o_historico():
origem = _origem_unica()
agora = 1_000_000.0
for _ in range(rate_limit.MAX_TENTATIVAS):
rate_limit.registrar_falha(origem, agora=agora)
assert rate_limit.bloqueado(origem, agora=agora) is True
rate_limit.limpar(origem)
assert rate_limit.bloqueado(origem, agora=agora) is False
def test_origens_diferentes_nao_se_afetam():
origem_a, origem_b = _origem_unica(), _origem_unica()
agora = 1_000_000.0
for _ in range(rate_limit.MAX_TENTATIVAS):
rate_limit.registrar_falha(origem_a, agora=agora)
assert rate_limit.bloqueado(origem_a, agora=agora) is True
assert rate_limit.bloqueado(origem_b, agora=agora) is False
+132
View File
@@ -0,0 +1,132 @@
from fastapi import Depends, FastAPI
from fastapi.testclient import TestClient
from app import rate_limit, session
from app.auth import NaoAutenticado, UsuarioLogado, exigir_login
from app.routers import auth as router_auth
app_teste = FastAPI()
app_teste.include_router(router_auth.router)
@app_teste.exception_handler(NaoAutenticado)
def _handler(request, exc):
from fastapi.responses import RedirectResponse
return RedirectResponse(url=f"/login?next={exc.caminho_original}", status_code=303)
@app_teste.get("/rota-protegida-teste")
def _rota_protegida(usuario: UsuarioLogado = Depends(exigir_login)):
return {"ok": True}
client = TestClient(app_teste, follow_redirects=False)
CSRF_TESTE = "teste-csrf-token"
client.cookies.set("csrf_token", CSRF_TESTE)
def test_get_login_mostra_formulario():
resposta = client.get("/login")
assert resposta.status_code == 200
assert "senha" in resposta.text.lower()
def test_post_login_credenciais_corretas_grava_cookie_e_redireciona():
resposta = client.post(
"/login", data={"usuario": "admin", "senha": "teste123", "next": "/clientes", "csrf_token": CSRF_TESTE}
)
assert resposta.status_code == 303
assert resposta.headers["location"] == "/clientes"
assert session.SESSION_COOKIE_NAME in resposta.cookies
def test_post_login_sem_next_redireciona_para_clientes():
resposta = client.post("/login", data={"usuario": "admin", "senha": "teste123", "csrf_token": CSRF_TESTE})
assert resposta.status_code == 303
assert resposta.headers["location"] == "/clientes"
def test_post_login_next_externo_e_ignorado_por_seguranca():
resposta = client.post(
"/login",
data={"usuario": "admin", "senha": "teste123", "next": "https://site-malicioso.com", "csrf_token": CSRF_TESTE},
)
assert resposta.status_code == 303
assert resposta.headers["location"] == "/clientes"
def test_post_login_next_com_barra_invertida_e_ignorado_por_seguranca():
# "/\evil.com" é tratado por navegadores modernos como equivalente a
# "//evil.com" (URL protocol-relative) — precisa ser bloqueado também.
resposta = client.post(
"/login",
data={"usuario": "admin", "senha": "teste123", "next": "/\\evil.com", "csrf_token": CSRF_TESTE},
)
assert resposta.status_code == 303
assert resposta.headers["location"] == "/clientes"
def test_post_login_credenciais_erradas_nao_grava_cookie():
resposta = client.post(
"/login", data={"usuario": "admin", "senha": "errada", "csrf_token": CSRF_TESTE}
)
assert resposta.status_code == 200
assert "inválid" in resposta.text.lower()
assert session.SESSION_COOKIE_NAME not in resposta.cookies
def test_get_rota_protegida_sem_login_redireciona_para_login():
# Testes anteriores deixaram um painel_session válido no cookie jar do
# client (persistente, como um navegador real) — limpa antes de simular
# a requisição de um visitante não autenticado.
client.cookies.delete(session.SESSION_COOKIE_NAME)
resposta = client.get("/rota-protegida-teste")
assert resposta.status_code == 303
assert resposta.headers["location"] == "/login?next=/rota-protegida-teste"
def test_post_login_apos_muitas_falhas_e_bloqueado_mesmo_com_senha_certa():
# TestClient reporta sempre a mesma origem ("testclient") — limpa
# antes pra não herdar a falha registrada em
# test_post_login_credenciais_erradas_nao_grava_cookie() acima, e
# limpa depois pra não vazar bloqueio pros testes seguintes deste
# arquivo (que também usam essa mesma origem "testclient").
rate_limit._falhas.pop("testclient", None)
try:
for _ in range(rate_limit.MAX_TENTATIVAS):
client.post("/login", data={"usuario": "admin", "senha": "errada", "csrf_token": CSRF_TESTE})
resposta = client.post(
"/login", data={"usuario": "admin", "senha": "teste123", "csrf_token": CSRF_TESTE}
)
assert resposta.status_code == 429
assert "muitas tentativas" in resposta.text.lower()
assert session.SESSION_COOKIE_NAME not in resposta.cookies
finally:
rate_limit._falhas.pop("testclient", None)
def test_post_logout_apaga_cookie_e_redireciona():
cookies = {session.SESSION_COOKIE_NAME: session.criar_token("admin", "Admin", "ADMIN")}
resposta = client.post("/logout", data={"csrf_token": CSRF_TESTE}, cookies=cookies)
assert resposta.status_code == 303
assert resposta.headers["location"] == "/login"
# cookie apagado: o Set-Cookie da resposta zera o valor
set_cookie = resposta.headers.get("set-cookie", "")
assert session.SESSION_COOKIE_NAME in set_cookie
def test_post_logout_apaga_cookie_de_preferencia_de_paginacao():
from app.routers.clientes import POR_PAGINA_COOKIE
cookies = {session.SESSION_COOKIE_NAME: session.criar_token("admin", "Admin", "ADMIN")}
resposta = client.post("/logout", data={"csrf_token": CSRF_TESTE}, cookies=cookies)
assert resposta.status_code == 303
set_cookie = resposta.headers.get("set-cookie", "")
assert POR_PAGINA_COOKIE in set_cookie
# confirma que o cookie foi de fato apagado (Max-Age=0), não só regravado
trecho_cookie = next(
parte for parte in resposta.headers.get_list("set-cookie") if POR_PAGINA_COOKIE in parte
)
assert "Max-Age=0" in trecho_cookie
@@ -0,0 +1,102 @@
from fastapi import FastAPI
from fastapi.testclient import TestClient
from app.auth import exigir_login
from app.routers import chave_manual as router_chave_manual
app_teste = FastAPI()
app_teste.include_router(router_chave_manual.router)
app_teste.dependency_overrides[exigir_login] = lambda: "teste"
client = TestClient(app_teste)
CSRF_TESTE = "teste-csrf-token"
client.cookies.set("csrf_token", CSRF_TESTE)
def test_pagina_gerador_carrega():
resposta = client.get("/gerador-chave")
assert resposta.status_code == 200
assert "Gerador de Chave" in resposta.text
# O algoritmo NUNCA pode voltar a aparecer no HTML/JS da página — ele
# roda só no backend (app/gerador_chave.py) desde que essa exposição
# foi corrigida.
assert "CHAVE_INTERNA" not in resposta.text
assert "function gerarReativacao" not in resposta.text
assert "function gerarInstalacao" not in resposta.text
def test_pagina_gerador_nao_grava_nada():
# GET /gerador-chave não recebe csrf_token no cookie/form combinando
# com nada que grave — só há dois POSTs possíveis nesse router:
# /gerador-chave (não existe, 405) e /gerador-chave/gerar (não grava
# em Clientes_Chaves, só calcula e devolve).
resposta = client.post("/gerador-chave")
assert resposta.status_code == 405
def test_gerar_reativacao_com_sucesso():
resposta = client.post(
"/gerador-chave/gerar",
data={"csrf_token": CSRF_TESTE, "tipo": "reativacao", "codigo": "123", "referencia": "2026-08"},
)
assert resposta.status_code == 200
corpo = resposta.json()
assert len(corpo["chave"]) == 22
assert all(c in "0123456789ABCDEF" for c in corpo["chave"])
def test_gerar_instalacao_com_sucesso_bate_com_exemplo_conhecido():
# Não dá pra fixar o byte aleatório via HTTP, mas confere que o valor do
# cliente (1*19+31=50) aparece embutido no meio da chave, como o
# algoritmo real exige.
resposta = client.post(
"/gerador-chave/gerar",
data={"csrf_token": CSRF_TESTE, "tipo": "instalacao", "codigo": "000001", "modalidade": "ALUGUEL"},
)
assert resposta.status_code == 200
corpo = resposta.json()
assert len(corpo["chave"]) == 8
assert corpo["chave"][3:5] == "50"
def test_gerar_codigo_invalido_bloqueia():
resposta = client.post(
"/gerador-chave/gerar",
data={"csrf_token": CSRF_TESTE, "tipo": "instalacao", "codigo": "abc123", "modalidade": "ALUGUEL"},
)
assert resposta.status_code == 400
assert "6 dígitos" in resposta.json()["erro"]
def test_gerar_reativacao_sem_referencia_bloqueia():
resposta = client.post(
"/gerador-chave/gerar",
data={"csrf_token": CSRF_TESTE, "tipo": "reativacao", "codigo": "123", "referencia": ""},
)
assert resposta.status_code == 400
assert "mês/ano" in resposta.json()["erro"]
def test_gerar_instalacao_modalidade_invalida_bloqueia():
resposta = client.post(
"/gerador-chave/gerar",
data={"csrf_token": CSRF_TESTE, "tipo": "instalacao", "codigo": "123", "modalidade": "X"},
)
assert resposta.status_code == 400
assert "Modalidade" in resposta.json()["erro"]
def test_gerar_tipo_invalido_bloqueia():
resposta = client.post(
"/gerador-chave/gerar",
data={"csrf_token": CSRF_TESTE, "tipo": "outracoisa", "codigo": "123"},
)
assert resposta.status_code == 400
assert "Tipo de chave" in resposta.json()["erro"]
def test_gerar_sem_csrf_bloqueia():
resposta = client.post(
"/gerador-chave/gerar",
data={"tipo": "instalacao", "codigo": "123", "modalidade": "ALUGUEL"},
)
assert resposta.status_code == 403
+962
View File
@@ -0,0 +1,962 @@
"""
Testa só a camada HTTP: monta um FastAPI() com apenas este router, sobrescreve
exigir_login (sem exigir Basic Auth real) e faz monkeypatch das funções de
app.db.clientes usadas pelo router — não toca no banco de verdade.
"""
import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
from app.auth import exigir_login
from app.db.connection import get_connection
from app.routers import clientes as router_clientes
app_teste = FastAPI()
app_teste.include_router(router_clientes.router)
app_teste.dependency_overrides[exigir_login] = lambda: "teste"
client = TestClient(app_teste, follow_redirects=False)
# Este app de teste não tem o middleware de CSRF (só o main.py real o monta).
# Simula um navegador que já tem o cookie definido: os testes de POST
# precisam mandar o mesmo valor no campo oculto "csrf_token".
CSRF_TESTE = "teste-csrf-token"
client.cookies.set("csrf_token", CSRF_TESTE)
@pytest.fixture(autouse=True)
def _sem_log_real(monkeypatch):
"""
O router registra cada ação em LOG_PAINEL (app/db/log.py) — por padrão
isso bateria no banco real. Substitui por um no-op nos testes que não
são especificamente sobre o log (test_router_clientes não testa
app/db/log.py, isso já é coberto em tests/test_db_log.py).
"""
monkeypatch.setattr(router_clientes.db_log, "registrar", lambda *args, **kwargs: None)
@pytest.fixture(autouse=True)
def _limpar_cookies():
"""Limpa cookies entre testes para evitar compartilhamento de estado."""
client.cookies.clear()
client.cookies.set("csrf_token", CSRF_TESTE)
yield
# Limpa novamente após o teste
client.cookies.clear()
client.cookies.set("csrf_token", CSRF_TESTE)
def _form_valido(codigo: str = "000001") -> dict:
"""Monta um dict de formulário válido (todos os campos preenchidos)."""
dados = {campo: "x" for campo in router_clientes.db_clientes.CAMPOS_CLIENTE}
dados.update(
{
"CODIGO": codigo,
"CNPJ": "11222333000181",
"IMPLANTACAO": "0",
"MENSALIDADE": "100",
"JUROS_MORA": "0",
"JUROS_MULTA": "0",
"DATA_CONTRATO": "2026-01-01",
"VENCIMENTO_CONTRATO": "2027-01-01",
"DIA_VENCIMENTO": "10",
"ATIVO": "1",
"csrf_token": CSRF_TESTE,
}
)
return dados
def test_resolver_por_pagina_usa_query_quando_valido():
assert router_clientes._resolver_por_pagina(50, None) == 50
def test_resolver_por_pagina_ignora_query_invalida_usa_cookie():
assert router_clientes._resolver_por_pagina(999, "30") == 30
def test_resolver_por_pagina_ignora_cookie_invalido_usa_padrao():
assert router_clientes._resolver_por_pagina(None, "abc") == 15
assert router_clientes._resolver_por_pagina(None, "999") == 15
def test_resolver_por_pagina_sem_query_nem_cookie_usa_padrao():
assert router_clientes._resolver_por_pagina(None, None) == 15
def test_janela_paginas_poucas_paginas_mostra_todas():
assert router_clientes._janela_paginas(1, 5) == [1, 2, 3, 4, 5]
assert router_clientes._janela_paginas(3, 7) == [1, 2, 3, 4, 5, 6, 7]
def test_janela_paginas_no_meio_com_reticencias_dos_dois_lados():
assert router_clientes._janela_paginas(5, 20) == [1, None, 4, 5, 6, None, 20]
def test_janela_paginas_primeira_pagina():
assert router_clientes._janela_paginas(1, 20) == [1, 2, None, 20]
def test_janela_paginas_ultima_pagina():
assert router_clientes._janela_paginas(20, 20) == [1, None, 19, 20]
def test_listar_clientes_por_pagina_padrao_e_15(monkeypatch):
capturado = {}
monkeypatch.setattr(
router_clientes.db_clientes,
"listar_clientes",
lambda busca=None, limite=100, offset=0, ordenar_por="razao", direcao="asc", filtro_ativo=1: capturado.setdefault("limite", limite) and [],
)
monkeypatch.setattr(router_clientes.db_clientes, "contar_clientes", lambda busca=None, ordenar_por="razao", filtro_ativo=1:3)
resposta = client.get("/clientes")
assert resposta.status_code == 200
assert capturado["limite"] == 15
def test_listar_clientes_por_pagina_da_query_e_usada_e_grava_cookie(monkeypatch):
capturado = {}
monkeypatch.setattr(
router_clientes.db_clientes,
"listar_clientes",
lambda busca=None, limite=100, offset=0, ordenar_por="razao", direcao="asc", filtro_ativo=1: capturado.setdefault("limite", limite) and [],
)
monkeypatch.setattr(router_clientes.db_clientes, "contar_clientes", lambda busca=None, ordenar_por="razao", filtro_ativo=1:3)
resposta = client.get("/clientes?por_pagina=50")
assert resposta.status_code == 200
assert capturado["limite"] == 50
assert resposta.cookies.get(router_clientes.POR_PAGINA_COOKIE) == "50"
def test_listar_clientes_usa_cookie_quando_sem_por_pagina_na_query(monkeypatch):
capturado = {}
monkeypatch.setattr(
router_clientes.db_clientes,
"listar_clientes",
lambda busca=None, limite=100, offset=0, ordenar_por="razao", direcao="asc", filtro_ativo=1: capturado.setdefault("limite", limite) and [],
)
monkeypatch.setattr(router_clientes.db_clientes, "contar_clientes", lambda busca=None, ordenar_por="razao", filtro_ativo=1:3)
resposta = client.get("/clientes", cookies={router_clientes.POR_PAGINA_COOKIE: "30"})
assert resposta.status_code == 200
assert capturado["limite"] == 30
def test_listar_clientes_por_pagina_invalida_na_query_cai_no_padrao_e_nao_grava_cookie(monkeypatch):
capturado = {}
monkeypatch.setattr(
router_clientes.db_clientes,
"listar_clientes",
lambda busca=None, limite=100, offset=0, ordenar_por="razao", direcao="asc", filtro_ativo=1: capturado.setdefault("limite", limite) and [],
)
monkeypatch.setattr(router_clientes.db_clientes, "contar_clientes", lambda busca=None, ordenar_por="razao", filtro_ativo=1:3)
resposta = client.get("/clientes?por_pagina=999")
assert resposta.status_code == 200
assert capturado["limite"] == 15
assert router_clientes.POR_PAGINA_COOKIE not in resposta.cookies
def test_listar_clientes_status_padrao_e_ativos(monkeypatch):
capturado = {}
monkeypatch.setattr(
router_clientes.db_clientes,
"listar_clientes",
lambda busca=None, limite=100, offset=0, ordenar_por="razao", direcao="asc", filtro_ativo=1: (capturado.__setitem__("filtro_ativo", filtro_ativo), [])[1],
)
monkeypatch.setattr(
router_clientes.db_clientes,
"contar_clientes",
lambda busca=None, ordenar_por="razao", filtro_ativo=1: 0,
)
resposta = client.get("/clientes")
assert resposta.status_code == 200
assert capturado["filtro_ativo"] == 1
def test_listar_clientes_status_inativos_usa_filtro_zero(monkeypatch):
capturado = {}
monkeypatch.setattr(
router_clientes.db_clientes,
"listar_clientes",
lambda busca=None, limite=100, offset=0, ordenar_por="razao", direcao="asc", filtro_ativo=1: (capturado.__setitem__("filtro_ativo", filtro_ativo), [])[1],
)
monkeypatch.setattr(router_clientes.db_clientes, "contar_clientes", lambda busca=None, ordenar_por="razao", filtro_ativo=1: 0)
resposta = client.get("/clientes?status=inativos")
assert resposta.status_code == 200
assert capturado["filtro_ativo"] == 0
def test_listar_clientes_status_todos_sem_filtro(monkeypatch):
capturado = {}
monkeypatch.setattr(
router_clientes.db_clientes,
"listar_clientes",
lambda busca=None, limite=100, offset=0, ordenar_por="razao", direcao="asc", filtro_ativo=1: (capturado.__setitem__("filtro_ativo", filtro_ativo), [])[1],
)
monkeypatch.setattr(router_clientes.db_clientes, "contar_clientes", lambda busca=None, ordenar_por="razao", filtro_ativo=1: 0)
resposta = client.get("/clientes?status=todos")
assert resposta.status_code == 200
assert capturado["filtro_ativo"] is None
def test_listar_clientes_status_invalido_cai_no_padrao(monkeypatch):
capturado = {}
monkeypatch.setattr(
router_clientes.db_clientes,
"listar_clientes",
lambda busca=None, limite=100, offset=0, ordenar_por="razao", direcao="asc", filtro_ativo=1: (capturado.__setitem__("filtro_ativo", filtro_ativo), [])[1],
)
monkeypatch.setattr(router_clientes.db_clientes, "contar_clientes", lambda busca=None, ordenar_por="razao", filtro_ativo=1: 0)
resposta = client.get("/clientes?status=lixo_qualquer")
assert resposta.status_code == 200
assert capturado["filtro_ativo"] == 1
def test_listar_clientes(monkeypatch):
monkeypatch.setattr(
router_clientes.db_clientes,
"listar_clientes",
lambda busca=None, limite=100, offset=0, ordenar_por="razao", direcao="asc", filtro_ativo=1: [
{"ID": 1, "CODIGO": "000001", "RAZAO": "Cliente Um", "FANTASIA": "Um", "CNPJ": "11222333000181"}
],
)
monkeypatch.setattr(router_clientes.db_clientes, "contar_clientes", lambda busca=None, ordenar_por="razao", filtro_ativo=1:1)
resposta = client.get("/clientes")
assert resposta.status_code == 200
assert "Cliente Um" in resposta.text
def test_listar_clientes_inclui_csrf_token_no_contexto(monkeypatch):
monkeypatch.setattr(
router_clientes.db_clientes,
"listar_clientes",
lambda busca=None, limite=100, offset=0, ordenar_por="razao", direcao="asc", filtro_ativo=1: [],
)
monkeypatch.setattr(router_clientes.db_clientes, "contar_clientes", lambda busca=None, ordenar_por="razao", filtro_ativo=1:0)
resposta = client.get("/clientes")
assert resposta.status_code == 200
# Este app de teste não monta o middleware de CSRF (ver comentário no
# topo do arquivo), então token_do_contexto() retorna "" — o que
# importa aqui é que o template usa a variável sem quebrar e ela
# aparece no script embutido da página.
assert 'const CSRF_TOKEN = "";' in resposta.text
def test_listar_clientes_tem_botao_de_gerar_chave_por_linha(monkeypatch):
monkeypatch.setattr(
router_clientes.db_clientes,
"listar_clientes",
lambda busca=None, limite=100, offset=0, ordenar_por="razao", direcao="asc", filtro_ativo=1: [
{"ID": 1, "CODIGO": "000042", "RAZAO": "Cliente Um", "FANTASIA": "Um", "CNPJ": "11222333000181"}
],
)
monkeypatch.setattr(router_clientes.db_clientes, "contar_clientes", lambda busca=None, ordenar_por="razao", filtro_ativo=1:1)
resposta = client.get("/clientes")
assert resposta.status_code == 200
assert 'data-codigo="000042"' in resposta.text
assert "gerarChaveReativacao(this)" in resposta.text
assert "/gerador-chave/gerar" in resposta.text
def test_listar_clientes_pagina_com_controles_de_paginacao(monkeypatch):
monkeypatch.setattr(
router_clientes.db_clientes,
"listar_clientes",
lambda busca=None, limite=100, offset=0, ordenar_por="razao", direcao="asc", filtro_ativo=1: [
{"ID": i, "CODIGO": f"{i:06d}", "RAZAO": f"Cliente {i}", "FANTASIA": f"C{i}", "CNPJ": "11222333000181"}
for i in range(1, 6)
],
)
monkeypatch.setattr(router_clientes.db_clientes, "contar_clientes", lambda busca=None, ordenar_por="razao", filtro_ativo=1:250)
resposta = client.get("/clientes?pagina=2&por_pagina=100")
assert resposta.status_code == 200
assert "Exibindo 101 a 200 de 250 clientes" in resposta.text
assert "pagina=1&ordenar_por=razao&direcao=asc&por_pagina=100" in resposta.text
assert "pagina=3&ordenar_por=razao&direcao=asc&por_pagina=100" in resposta.text
def test_listar_clientes_footer_com_selecao_de_por_pagina(monkeypatch):
monkeypatch.setattr(
router_clientes.db_clientes,
"listar_clientes",
lambda busca=None, limite=100, offset=0, ordenar_por="razao", direcao="asc", filtro_ativo=1: [
{"ID": i, "CODIGO": f"{i:06d}", "RAZAO": f"Cliente {i}", "FANTASIA": f"C{i}", "CNPJ": "1"}
for i in range(1, 16)
],
)
monkeypatch.setattr(router_clientes.db_clientes, "contar_clientes", lambda busca=None, ordenar_por="razao", filtro_ativo=1:98)
resposta = client.get("/clientes?pagina=1")
assert resposta.status_code == 200
assert 'value="15" selected' in resposta.text
assert "Exibindo 1 a 15 de 98 clientes" in resposta.text
# 98 / 15 = 7 paginas -> <=7, sem reticencias, todas visiveis
assert "pagina=7&ordenar_por=razao&direcao=asc&por_pagina=15" in resposta.text
def test_listar_clientes_paginacao_com_reticencias_quando_muitas_paginas(monkeypatch):
monkeypatch.setattr(router_clientes.db_clientes, "listar_clientes", lambda **kwargs: [])
monkeypatch.setattr(router_clientes.db_clientes, "contar_clientes", lambda busca=None, ordenar_por="razao", filtro_ativo=1:500)
# 500 / 5 = 100 paginas, pedindo a pagina 10 (no meio) -> tem reticencias
resposta = client.get("/clientes?pagina=10&por_pagina=5")
assert resposta.status_code == 200
assert "&hellip;" in resposta.text
def test_listar_clientes_ordenar_por_invalido_cai_no_padrao(monkeypatch):
capturado = {}
def _listar_clientes(busca=None, limite=100, offset=0, ordenar_por="razao", direcao="asc", filtro_ativo=1):
capturado["ordenar_por"] = ordenar_por
return []
monkeypatch.setattr(router_clientes.db_clientes, "listar_clientes", _listar_clientes)
monkeypatch.setattr(router_clientes.db_clientes, "contar_clientes", lambda busca=None, ordenar_por="razao", filtro_ativo=1:0)
client.get("/clientes?ordenar_por=coisa_invalida_qualquer")
assert capturado["ordenar_por"] == "razao"
client.get("/clientes?ordenar_por=codigo")
assert capturado["ordenar_por"] == "codigo"
def test_cabecalhos_da_lista_alternam_direcao(monkeypatch):
monkeypatch.setattr(
router_clientes.db_clientes, "listar_clientes", lambda **kwargs: []
)
monkeypatch.setattr(router_clientes.db_clientes, "contar_clientes", lambda busca=None, ordenar_por="razao", filtro_ativo=1:0)
# Sem ordenar_por/direcao na URL: padrão é razao/asc. Clicar de novo em
# "Razão" deve levar pra desc; clicar em "Código" (coluna diferente da
# atual) deve levar pra asc (não continua o desc de outra coluna).
resposta = client.get("/clientes")
assert 'ordenar_por=razao&direcao=desc' in resposta.text
assert 'ordenar_por=codigo&direcao=asc' in resposta.text
# Já ordenado por codigo/desc: clicar em "Código" de novo deve voltar pra asc.
resposta = client.get("/clientes?ordenar_por=codigo&direcao=desc")
assert 'ordenar_por=codigo&direcao=asc' in resposta.text
# E as outras colunas oferecem asc (primeira vez que seriam clicadas).
assert 'ordenar_por=razao&direcao=asc' in resposta.text
def test_busca_com_e_comercial_nao_quebra_links_de_paginacao_e_ordenacao(monkeypatch):
# "&" cru dentro de busca, sem urlencode, corta o resto da query string
# do link (ex.: "busca=A&B&pagina=2" vira busca="A", parâmetro solto
# "B", perdendo "pagina=2") — reprodução do bug corrigido em
# clientes_lista.html.
monkeypatch.setattr(
router_clientes.db_clientes,
"listar_clientes",
lambda busca=None, limite=100, offset=0, ordenar_por="razao", direcao="asc", filtro_ativo=1: [
{"ID": i, "CODIGO": f"{i:06d}", "RAZAO": f"Cliente {i}", "FANTASIA": f"C{i}", "CNPJ": "1"}
for i in range(1, 6)
],
)
monkeypatch.setattr(router_clientes.db_clientes, "contar_clientes", lambda busca=None, ordenar_por="razao", filtro_ativo=1:250)
resposta = client.get("/clientes?pagina=2&busca=A%26B")
assert resposta.status_code == 200
assert "busca=A%26B" in resposta.text
assert "busca=A&B" not in resposta.text
def test_listar_clientes_direcao_invalida_cai_no_padrao(monkeypatch):
capturado = {}
def _listar_clientes(busca=None, limite=100, offset=0, ordenar_por="razao", direcao="asc", filtro_ativo=1):
capturado["direcao"] = direcao
return []
monkeypatch.setattr(router_clientes.db_clientes, "listar_clientes", _listar_clientes)
monkeypatch.setattr(router_clientes.db_clientes, "contar_clientes", lambda busca=None, ordenar_por="razao", filtro_ativo=1:0)
client.get("/clientes?direcao=coisa_invalida")
assert capturado["direcao"] == "asc"
client.get("/clientes?direcao=desc")
assert capturado["direcao"] == "desc"
def test_consulta_cnpj_encontrado(monkeypatch):
monkeypatch.setattr(
router_clientes.cnpj_lookup,
"consultar_cnpj",
lambda cnpj: {"RAZAO": "EMPRESA TESTE LTDA", "FANTASIA": "EMPRESA TESTE"},
)
resposta = client.get("/clientes/consulta-cnpj/11222333000181")
assert resposta.status_code == 200
corpo = resposta.json()
assert corpo["encontrado"] is True
assert corpo["dados"]["RAZAO"] == "EMPRESA TESTE LTDA"
def test_consulta_cnpj_nao_encontrado(monkeypatch):
monkeypatch.setattr(router_clientes.cnpj_lookup, "consultar_cnpj", lambda cnpj: None)
resposta = client.get("/clientes/consulta-cnpj/11222333000181")
assert resposta.status_code == 404
assert resposta.json()["encontrado"] is False
def test_consulta_cnpj_rejeita_cpf(monkeypatch):
chamou = {}
monkeypatch.setattr(
router_clientes.cnpj_lookup, "consultar_cnpj", lambda cnpj: chamou.setdefault("ok", True)
)
# CPF válido (mesma fixture usada em test_normalize.py) — nunca deve
# chamar a consulta de CNPJ, é dado pessoal.
resposta = client.get("/clientes/consulta-cnpj/11144477735")
assert resposta.status_code == 404
assert "ok" not in chamou
def test_consulta_cnpj_rejeita_formato_invalido(monkeypatch):
chamou = {}
monkeypatch.setattr(
router_clientes.cnpj_lookup, "consultar_cnpj", lambda cnpj: chamou.setdefault("ok", True)
)
resposta = client.get("/clientes/consulta-cnpj/123")
assert resposta.status_code == 404
assert "ok" not in chamou
def test_form_novo_preenche_codigo_sugerido(monkeypatch):
monkeypatch.setattr(router_clientes.db_clientes, "proximo_codigo_sugerido", lambda: "000243")
resposta = client.get("/clientes/novo")
assert resposta.status_code == 200
assert 'value="000243"' in resposta.text
def test_form_novo_ativo_pre_selecionado(monkeypatch):
monkeypatch.setattr(router_clientes.db_clientes, "proximo_codigo_sugerido", lambda: "000243")
resposta = client.get("/clientes/novo")
assert resposta.status_code == 200
assert '<option value="1" selected>Ativo</option>' in resposta.text
def test_editar_cliente_ativo_zero_fica_selecionado(monkeypatch):
monkeypatch.setattr(
router_clientes.db_clientes,
"obter_cliente",
lambda cid: {**{c: "" for c in router_clientes.db_clientes.CAMPOS_CLIENTE}, "ID": cid, "ATIVO": 0},
)
resposta = client.get("/clientes/1/editar")
assert resposta.status_code == 200
assert '<option value="0" selected>Inativo</option>' in resposta.text
def test_listar_clientes_tem_select_de_status(monkeypatch):
monkeypatch.setattr(router_clientes.db_clientes, "listar_clientes", lambda **kwargs: [])
monkeypatch.setattr(router_clientes.db_clientes, "contar_clientes", lambda **kwargs: 0)
resposta = client.get("/clientes")
assert resposta.status_code == 200
assert '<option value="ativos" selected>Ativos</option>' in resposta.text
assert '<option value="inativos" >Desativados ou inativos</option>' in resposta.text
assert '<option value="todos" >Todos</option>' in resposta.text
def test_listar_clientes_status_preservado_nos_links_de_paginacao(monkeypatch):
monkeypatch.setattr(
router_clientes.db_clientes,
"listar_clientes",
lambda **kwargs: [
{"ID": i, "CODIGO": f"{i:06d}", "RAZAO": f"Cliente {i}", "FANTASIA": f"C{i}", "CNPJ": "1"}
for i in range(1, 6)
],
)
monkeypatch.setattr(router_clientes.db_clientes, "contar_clientes", lambda **kwargs: 250)
resposta = client.get("/clientes?pagina=2&por_pagina=100&status=todos")
assert resposta.status_code == 200
assert "status=todos" in resposta.text
assert '<option value="todos" selected>Todos</option>' in resposta.text
def test_detalhe_cliente_mostra_formulario_de_edicao_de_contato(monkeypatch):
monkeypatch.setattr(
router_clientes.db_clientes,
"obter_cliente",
lambda cid: {"ID": cid, "CODIGO": "000001", "RAZAO": "Cliente Um", "FANTASIA": "Um", "CNPJ": "11222333000181"},
)
monkeypatch.setattr(
router_clientes.db_contatos,
"listar_contatos_por_cliente",
lambda cid: [{"ID": 7, "NOME": "Fulano", "TELEFONE": "75999090001", "EMAIL": None}],
)
monkeypatch.setattr(router_clientes.db_chaves, "status_chaves", lambda codigo: None)
resposta = client.get("/clientes/1")
assert resposta.status_code == 200
assert '/clientes/1/contatos/7/editar' in resposta.text
assert 'value="Fulano"' in resposta.text
assert 'value="75999090001"' in resposta.text
def test_criar_cliente_codigo_duplicado_bloqueia_sem_bypass(monkeypatch):
monkeypatch.setattr(router_clientes.db_clientes, "codigo_em_uso", lambda codigo, ignorar_id=None: True)
criado = {}
monkeypatch.setattr(
router_clientes.db_clientes, "criar_cliente", lambda dados: criado.setdefault("ok", True) and 42
)
dados_form = _form_valido()
resposta = client.post("/clientes/novo", data=dados_form)
assert resposta.status_code == 200
assert "já está em uso" in resposta.text
assert "ok" not in criado
# Não existe mais bypass: mesmo insistindo, sem trocar o código, continua bloqueado.
resposta = client.post("/clientes/novo", data=dados_form)
assert resposta.status_code == 200
assert "já está em uso" in resposta.text
assert "ok" not in criado
def test_criar_cliente_campo_obrigatorio_faltando_bloqueia(monkeypatch):
monkeypatch.setattr(router_clientes.db_clientes, "codigo_em_uso", lambda codigo, ignorar_id=None: False)
monkeypatch.setattr(router_clientes.db_clientes, "cnpj_em_uso", lambda documento, ignorar_id=None: False)
criado = {}
monkeypatch.setattr(
router_clientes.db_clientes, "criar_cliente", lambda dados: criado.setdefault("ok", True) and 42
)
dados_form = _form_valido()
dados_form["RAZAO"] = ""
resposta = client.post("/clientes/novo", data=dados_form)
assert resposta.status_code == 200
assert "RAZAO" in resposta.text
assert "ok" not in criado
def test_criar_cliente_campos_administrativos_em_branco_usa_defaults(monkeypatch):
"""
IMPLANTACAO/MENSALIDADE/JUROS_MORA/JUROS_MULTA/DATA_CONTRATO/
VENCIMENTO_CONTRATO não são mais obrigatórios — em branco, o cliente é
criado mesmo assim, com defaults (0,00 / hoje).
"""
import datetime
monkeypatch.setattr(router_clientes.db_clientes, "codigo_em_uso", lambda codigo, ignorar_id=None: False)
monkeypatch.setattr(router_clientes.db_clientes, "cnpj_em_uso", lambda documento, ignorar_id=None: False)
capturado = {}
monkeypatch.setattr(
router_clientes.db_clientes, "criar_cliente", lambda dados: capturado.setdefault("dados", dados) and 42
)
dados_form = _form_valido()
for campo in ("IMPLANTACAO", "MENSALIDADE", "JUROS_MORA", "JUROS_MULTA", "DATA_CONTRATO", "VENCIMENTO_CONTRATO"):
dados_form[campo] = ""
resposta = client.post("/clientes/novo", data=dados_form)
assert resposta.status_code == 303
dados = capturado["dados"]
assert dados["IMPLANTACAO"] == "0.00"
assert dados["MENSALIDADE"] == "0.00"
assert dados["JUROS_MORA"] == "0.00"
assert dados["JUROS_MULTA"] == "0.00"
hoje = datetime.date.today().isoformat()
assert dados["DATA_CONTRATO"] == hoje
assert dados["VENCIMENTO_CONTRATO"] == hoje
def test_criar_cliente_normaliza_cnpj_e_telefones(monkeypatch):
monkeypatch.setattr(router_clientes.db_clientes, "codigo_em_uso", lambda codigo, ignorar_id=None: False)
monkeypatch.setattr(router_clientes.db_clientes, "cnpj_em_uso", lambda documento, ignorar_id=None: False)
capturado = {}
monkeypatch.setattr(
router_clientes.db_clientes, "criar_cliente", lambda dados: capturado.setdefault("dados", dados) and 42
)
dados_form = _form_valido()
dados_form.update(
{
"CNPJ": "11.222.333/0001-81",
"TELEFONE": "(75) 3333-4444",
"CELULAR": "(75) 99909-1780",
}
)
resposta = client.post("/clientes/novo", data=dados_form)
assert resposta.status_code == 303
assert capturado["dados"]["CNPJ"] == "11222333000181"
assert capturado["dados"]["TELEFONE"] == "557533334444"
assert capturado["dados"]["CELULAR"] == "5575999091780"
def test_editar_cliente_normaliza_cnpj_e_telefones(monkeypatch):
monkeypatch.setattr(router_clientes.db_clientes, "obter_cliente", lambda cliente_id: {"ID": cliente_id})
monkeypatch.setattr(router_clientes.db_clientes, "codigo_em_uso", lambda codigo, ignorar_id=None: False)
monkeypatch.setattr(router_clientes.db_clientes, "cnpj_em_uso", lambda documento, ignorar_id=None: False)
capturado = {}
monkeypatch.setattr(
router_clientes.db_clientes,
"atualizar_cliente",
lambda cliente_id, dados: capturado.setdefault("dados", dados),
)
dados_form = _form_valido()
dados_form.update(
{
"CNPJ": "111.444.777-35",
"TELEFONE": "(75) 3333-4444",
"CELULAR": "(75) 99909-1780",
}
)
resposta = client.post("/clientes/1/editar", data=dados_form)
assert resposta.status_code == 303
assert capturado["dados"]["CNPJ"] == "11144477735"
assert capturado["dados"]["TELEFONE"] == "557533334444"
assert capturado["dados"]["CELULAR"] == "5575999091780"
def test_editar_cliente_codigo_duplicado_bloqueia_sem_bypass(monkeypatch):
monkeypatch.setattr(router_clientes.db_clientes, "obter_cliente", lambda cliente_id: {"ID": cliente_id})
monkeypatch.setattr(router_clientes.db_clientes, "codigo_em_uso", lambda codigo, ignorar_id=None: True)
atualizado = {}
monkeypatch.setattr(
router_clientes.db_clientes,
"atualizar_cliente",
lambda cliente_id, dados: atualizado.setdefault("ok", True),
)
dados_form = _form_valido()
resposta = client.post("/clientes/1/editar", data=dados_form)
assert resposta.status_code == 200
assert "já está em uso" in resposta.text
assert "ok" not in atualizado
def test_criar_cliente_cnpj_duplicado_bloqueia(monkeypatch):
monkeypatch.setattr(router_clientes.db_clientes, "codigo_em_uso", lambda codigo, ignorar_id=None: False)
monkeypatch.setattr(router_clientes.db_clientes, "cnpj_em_uso", lambda documento, ignorar_id=None: True)
criado = {}
monkeypatch.setattr(
router_clientes.db_clientes, "criar_cliente", lambda dados: criado.setdefault("ok", True) and 42
)
resposta = client.post("/clientes/novo", data=_form_valido())
assert resposta.status_code == 200
assert "já está cadastrado" in resposta.text
assert "ok" not in criado
def test_editar_cliente_cnpj_duplicado_bloqueia(monkeypatch):
monkeypatch.setattr(router_clientes.db_clientes, "obter_cliente", lambda cliente_id: {"ID": cliente_id})
monkeypatch.setattr(router_clientes.db_clientes, "codigo_em_uso", lambda codigo, ignorar_id=None: False)
monkeypatch.setattr(router_clientes.db_clientes, "cnpj_em_uso", lambda documento, ignorar_id=None: True)
atualizado = {}
monkeypatch.setattr(
router_clientes.db_clientes,
"atualizar_cliente",
lambda cliente_id, dados: atualizado.setdefault("ok", True),
)
resposta = client.post("/clientes/1/editar", data=_form_valido())
assert resposta.status_code == 200
assert "já está cadastrado" in resposta.text
assert "ok" not in atualizado
def test_editar_cliente_inexistente_nao_finge_sucesso(monkeypatch):
# Sem a checagem de existência, isso caía direto no UPDATE (0 linhas
# afetadas, sem erro do MySQL) e ainda registrava "sucesso" no log.
monkeypatch.setattr(router_clientes.db_clientes, "obter_cliente", lambda cliente_id: None)
atualizado = {}
monkeypatch.setattr(
router_clientes.db_clientes,
"atualizar_cliente",
lambda cliente_id, dados: atualizado.setdefault("ok", True),
)
resposta = client.post("/clientes/999/editar", data=_form_valido())
assert resposta.status_code == 303
assert resposta.headers["location"].startswith("/clientes?")
assert "ok" not in atualizado
def test_excluir_cliente_inexistente_nao_finge_sucesso(monkeypatch):
monkeypatch.setattr(router_clientes.db_clientes, "excluir_cliente", lambda cliente_id: False)
resposta = client.post("/clientes/999/excluir", data={"csrf_token": CSRF_TESTE})
assert resposta.status_code == 303
assert resposta.headers["location"].startswith("/clientes?")
assert "danger" in resposta.headers["location"]
def test_excluir_cliente_com_dependencia_mostra_erro(monkeypatch):
monkeypatch.setattr(
router_clientes.db_clientes,
"obter_cliente",
lambda cliente_id: {"ID": cliente_id, "CODIGO": "000001", "RAZAO": "R", "FANTASIA": "F"},
)
monkeypatch.setattr(router_clientes.db_contatos, "listar_contatos_por_cliente", lambda cliente_id: [])
monkeypatch.setattr(router_clientes.db_chaves, "status_chaves", lambda codigo: None)
def _excluir(cliente_id):
raise router_clientes.db_clientes.ClienteComDependenciasError("tem contato")
monkeypatch.setattr(router_clientes.db_clientes, "excluir_cliente", _excluir)
resposta = client.post("/clientes/1/excluir", data={"csrf_token": CSRF_TESTE}, follow_redirects=True)
assert resposta.status_code == 200
assert "tem contato" in resposta.text
def test_alternar_ativo_inverte_e_redireciona_para_url_informada(monkeypatch):
monkeypatch.setattr(router_clientes.db_clientes, "alternar_ativo", lambda cliente_id: True)
resposta = client.post(
"/clientes/1/alternar-ativo",
data={"csrf_token": CSRF_TESTE, "redirecionar_para": "/clientes?status=todos&pagina=3"},
)
assert resposta.status_code == 303
assert resposta.headers["location"].startswith("/clientes?status=todos&pagina=3")
def test_alternar_ativo_redirecionar_para_invalido_cai_no_clientes_generico(monkeypatch):
monkeypatch.setattr(router_clientes.db_clientes, "alternar_ativo", lambda cliente_id: True)
resposta = client.post(
"/clientes/1/alternar-ativo",
data={"csrf_token": CSRF_TESTE, "redirecionar_para": "https://malicioso.example/"},
)
assert resposta.status_code == 303
assert resposta.headers["location"].startswith("/clientes?")
assert "malicioso" not in resposta.headers["location"]
def test_alternar_ativo_sem_redirecionar_para_cai_no_clientes_generico(monkeypatch):
monkeypatch.setattr(router_clientes.db_clientes, "alternar_ativo", lambda cliente_id: True)
resposta = client.post("/clientes/1/alternar-ativo", data={"csrf_token": CSRF_TESTE})
assert resposta.status_code == 303
assert resposta.headers["location"].startswith("/clientes?")
def test_alternar_ativo_id_inexistente_nao_finge_sucesso(monkeypatch):
monkeypatch.setattr(router_clientes.db_clientes, "alternar_ativo", lambda cliente_id: False)
resposta = client.post("/clientes/999/alternar-ativo", data={"csrf_token": CSRF_TESTE})
assert resposta.status_code == 303
assert "n%C3%A3o+encontrado" in resposta.headers["location"] or "nao encontrado" in resposta.headers["location"]
def test_listar_clientes_mostra_bolinha_de_status(monkeypatch):
monkeypatch.setattr(
router_clientes.db_clientes,
"listar_clientes",
lambda **kwargs: [
{"ID": 1, "CODIGO": "000001", "RAZAO": "Cliente Ativo", "FANTASIA": "A", "CNPJ": "1", "ATIVO": 1},
{"ID": 2, "CODIGO": "000002", "RAZAO": "Cliente Inativo", "FANTASIA": "I", "CNPJ": "2", "ATIVO": 0},
],
)
monkeypatch.setattr(router_clientes.db_clientes, "contar_clientes", lambda **kwargs: 2)
resposta = client.get("/clientes?status=todos")
assert resposta.status_code == 200
assert "status-dot-ativo" in resposta.text
assert "status-dot-inativo" in resposta.text
assert '/clientes/1/alternar-ativo' in resposta.text
assert '/clientes/2/alternar-ativo' in resposta.text
def test_listar_clientes_redirecionar_para_reflete_url_atual(monkeypatch):
monkeypatch.setattr(
router_clientes.db_clientes,
"listar_clientes",
lambda **kwargs: [{"ID": 1, "CODIGO": "000001", "RAZAO": "Cliente Um", "FANTASIA": "A", "CNPJ": "1", "ATIVO": 1}],
)
monkeypatch.setattr(router_clientes.db_clientes, "contar_clientes", lambda **kwargs: 1)
resposta = client.get("/clientes?status=inativos&pagina=2&por_pagina=50")
assert resposta.status_code == 200
assert 'value="/clientes?status=inativos&amp;pagina=2&amp;por_pagina=50"' in resposta.text
def _limpar_codigo(codigo: str) -> None:
conn = get_connection()
try:
with conn.cursor() as cur:
cur.execute("DELETE FROM CLIENTE WHERE CODIGO = %s", (codigo,))
finally:
conn.close()
def test_criar_cliente_com_data_e_decimais_vazios_cria_com_defaults():
"""
Integração com o banco real de dev: DATA_CONTRATO/decimais deixaram de
ser obrigatórios — em branco, o router aplica defaults (0,00 / hoje)
antes de gravar, então o cliente é criado normalmente (303), não gera
erro nem 500.
"""
codigo = "999906"
dados_form = _form_valido(codigo)
for campo in ("IMPLANTACAO", "MENSALIDADE", "JUROS_MORA", "JUROS_MULTA", "DATA_CONTRATO", "VENCIMENTO_CONTRATO"):
dados_form[campo] = ""
try:
resposta = client.post("/clientes/novo", data=dados_form)
assert resposta.status_code == 303
finally:
_limpar_codigo(codigo)
def test_criar_cliente_com_dv_invalido_exige_confirmacao(monkeypatch):
monkeypatch.setattr(router_clientes.db_clientes, "codigo_em_uso", lambda codigo, ignorar_id=None: False)
monkeypatch.setattr(router_clientes.db_clientes, "cnpj_em_uso", lambda documento, ignorar_id=None: False)
criado = {}
monkeypatch.setattr(
router_clientes.db_clientes, "criar_cliente", lambda dados: criado.setdefault("ok", True) and 42
)
dados_form = _form_valido()
dados_form["CNPJ"] = "11144477734" # CPF com DV inválido (1 dígito diferente do válido "...735")
# Sem confirmar: pede confirmação, não cria.
resposta = client.post("/clientes/novo", data=dados_form)
assert resposta.status_code == 200
assert "não confere" in resposta.text
assert "ok" not in criado
# Confirmando: cria.
dados_form["confirmar_dv_invalido"] = "1"
resposta = client.post("/clientes/novo", data=dados_form)
assert resposta.status_code == 303
assert criado.get("ok") is True
def test_criar_cliente_com_cnpj_dv_invalido_bloqueia_sempre(monkeypatch):
"""
Diferente de CPF: CNPJ com DV inválido bloqueia sempre, sem opção de
"cadastrar mesmo assim" — nem com confirmar_dv_invalido=1.
"""
monkeypatch.setattr(router_clientes.db_clientes, "codigo_em_uso", lambda codigo, ignorar_id=None: False)
monkeypatch.setattr(router_clientes.db_clientes, "cnpj_em_uso", lambda documento, ignorar_id=None: False)
criado = {}
monkeypatch.setattr(
router_clientes.db_clientes, "criar_cliente", lambda dados: criado.setdefault("ok", True) and 42
)
dados_form = _form_valido()
dados_form["CNPJ"] = "11222333000180" # último dígito trocado: DV inválido
resposta = client.post("/clientes/novo", data=dados_form)
assert resposta.status_code == 200
assert "não confere" in resposta.text
assert "ok" not in criado
# Mesmo confirmando, continua bloqueando (diferente do fluxo de CPF).
dados_form["confirmar_dv_invalido"] = "1"
resposta = client.post("/clientes/novo", data=dados_form)
assert resposta.status_code == 200
assert "ok" not in criado
def test_editar_cliente_com_cnpj_dv_invalido_bloqueia(monkeypatch):
monkeypatch.setattr(router_clientes.db_clientes, "obter_cliente", lambda cliente_id: {"ID": cliente_id})
monkeypatch.setattr(router_clientes.db_clientes, "codigo_em_uso", lambda codigo, ignorar_id=None: False)
monkeypatch.setattr(router_clientes.db_clientes, "cnpj_em_uso", lambda documento, ignorar_id=None: False)
atualizado = {}
monkeypatch.setattr(
router_clientes.db_clientes,
"atualizar_cliente",
lambda cliente_id, dados: atualizado.setdefault("ok", True),
)
dados_form = _form_valido()
dados_form["CNPJ"] = "11222333000180"
resposta = client.post("/clientes/1/editar", data=dados_form)
assert resposta.status_code == 200
assert "não confere" in resposta.text
assert "ok" not in atualizado
def test_criar_cliente_com_cpf_dv_invalido_continua_permitindo_confirmar(monkeypatch):
"""Regressão: CPF com DV inválido não muda de comportamento."""
monkeypatch.setattr(router_clientes.db_clientes, "codigo_em_uso", lambda codigo, ignorar_id=None: False)
monkeypatch.setattr(router_clientes.db_clientes, "cnpj_em_uso", lambda documento, ignorar_id=None: False)
criado = {}
monkeypatch.setattr(
router_clientes.db_clientes, "criar_cliente", lambda dados: criado.setdefault("ok", True) and 42
)
dados_form = _form_valido()
dados_form["CNPJ"] = "11144477734" # CPF com DV inválido
resposta = client.post("/clientes/novo", data=dados_form)
assert resposta.status_code == 200
assert "não confere" in resposta.text
assert "ok" not in criado
dados_form["confirmar_dv_invalido"] = "1"
resposta = client.post("/clientes/novo", data=dados_form)
assert resposta.status_code == 303
assert criado.get("ok") is True
def test_criar_cliente_grava_telefone_com_ddi_55(monkeypatch):
monkeypatch.setattr(router_clientes.db_clientes, "codigo_em_uso", lambda codigo, ignorar_id=None: False)
monkeypatch.setattr(router_clientes.db_clientes, "cnpj_em_uso", lambda documento, ignorar_id=None: False)
capturado = {}
monkeypatch.setattr(
router_clientes.db_clientes, "criar_cliente", lambda dados: capturado.setdefault("dados", dados) and 42
)
dados_form = _form_valido()
dados_form.update({"TELEFONE": "(75) 3333-4444", "CELULAR": "(75) 99909-1780"})
resposta = client.post("/clientes/novo", data=dados_form)
assert resposta.status_code == 303
assert capturado["dados"]["TELEFONE"] == "557533334444"
assert capturado["dados"]["CELULAR"] == "5575999091780"
def test_criar_cliente_telefone_ja_com_55_nao_duplica(monkeypatch):
monkeypatch.setattr(router_clientes.db_clientes, "codigo_em_uso", lambda codigo, ignorar_id=None: False)
monkeypatch.setattr(router_clientes.db_clientes, "cnpj_em_uso", lambda documento, ignorar_id=None: False)
capturado = {}
monkeypatch.setattr(
router_clientes.db_clientes, "criar_cliente", lambda dados: capturado.setdefault("dados", dados) and 42
)
dados_form = _form_valido()
dados_form["CELULAR"] = "5575999091780"
resposta = client.post("/clientes/novo", data=dados_form)
assert resposta.status_code == 303
assert capturado["dados"]["CELULAR"] == "5575999091780"
def test_criar_cliente_grava_uf_maiuscula_e_cep_sem_traco(monkeypatch):
monkeypatch.setattr(router_clientes.db_clientes, "codigo_em_uso", lambda codigo, ignorar_id=None: False)
monkeypatch.setattr(router_clientes.db_clientes, "cnpj_em_uso", lambda documento, ignorar_id=None: False)
capturado = {}
monkeypatch.setattr(
router_clientes.db_clientes, "criar_cliente", lambda dados: capturado.setdefault("dados", dados) and 42
)
dados_form = _form_valido()
dados_form.update({"UF": "ba", "CEP": "44430-174"})
resposta = client.post("/clientes/novo", data=dados_form)
assert resposta.status_code == 303
assert capturado["dados"]["UF"] == "BA"
assert capturado["dados"]["CEP"] == "44430174"
def test_listar_clientes_tabela_tem_classe_de_escopo(monkeypatch):
monkeypatch.setattr(router_clientes.db_clientes, "listar_clientes", lambda busca=None, limite=100, offset=0, ordenar_por="razao", direcao="asc", filtro_ativo=1: [])
monkeypatch.setattr(router_clientes.db_clientes, "contar_clientes", lambda busca=None, ordenar_por="razao", filtro_ativo=1: 0)
resposta = client.get("/clientes")
assert resposta.status_code == 200
assert '<table class="table-clientes">' in resposta.text
+142
View File
@@ -0,0 +1,142 @@
import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
from app.auth import exigir_login
from app.routers import contatos as router_contatos
from app.routers import clientes as router_clientes
app_teste = FastAPI()
app_teste.include_router(router_clientes.router)
app_teste.include_router(router_contatos.router)
app_teste.dependency_overrides[exigir_login] = lambda: "teste"
client = TestClient(app_teste, follow_redirects=False)
# Este app de teste não tem o middleware de CSRF (só o main.py real o monta).
# Simula um navegador que já tem o cookie definido: os testes de POST
# precisam mandar o mesmo valor no campo oculto "csrf_token".
CSRF_TESTE = "teste-csrf-token"
client.cookies.set("csrf_token", CSRF_TESTE)
@pytest.fixture(autouse=True)
def _sem_log_real(monkeypatch):
"""Os routers registram cada ação em LOG_PAINEL — sem no-op, bateria no banco real."""
monkeypatch.setattr(router_contatos.db_log, "registrar", lambda *args, **kwargs: None)
monkeypatch.setattr(router_clientes.db_log, "registrar", lambda *args, **kwargs: None)
def test_adicionar_contato_com_sucesso(monkeypatch):
monkeypatch.setattr(router_contatos.db_clientes, "obter_cliente", lambda cid: {"ID": cid, "CODIGO": "000001"})
monkeypatch.setattr(router_contatos.db_contatos, "contato_duplicado", lambda cid, tel, ignorar_id=None: False)
chamadas = {}
monkeypatch.setattr(
router_contatos.db_contatos,
"inserir_contato",
lambda cid, nome, tel, email=None: chamadas.setdefault("dados", (cid, nome, tel, email)),
)
resposta = client.post(
"/clientes/1/contatos", data={"nome": "Fulano", "telefone": "(75) 99909-0001", "csrf_token": CSRF_TESTE}
)
assert resposta.status_code == 303
# DDI 55 sempre embutido — é esse valor que o workflow n8n de envio de
# chave por WhatsApp lê direto de CLIENTE_CONTATO.TELEFONE, sem
# normalizar nada (ver comentário no topo de app/routers/contatos.py).
assert chamadas["dados"] == (1, "Fulano", "5575999090001", None)
def test_adicionar_contato_duplicado_avisa(monkeypatch):
monkeypatch.setattr(router_contatos.db_clientes, "obter_cliente", lambda cid: {"ID": cid, "CODIGO": "000001"})
monkeypatch.setattr(router_contatos.db_contatos, "contato_duplicado", lambda cid, tel, ignorar_id=None: True)
# Also mock the clientes router's db functions for the redirect
monkeypatch.setattr(router_clientes.db_clientes, "obter_cliente", lambda cid: {"ID": cid, "CODIGO": "000001", "RAZAO": "Cliente", "FANTASIA": "Cliente"})
monkeypatch.setattr(router_clientes.db_contatos, "listar_contatos_por_cliente", lambda cid: [])
monkeypatch.setattr(router_clientes.db_chaves, "status_chaves", lambda codigo: None)
resposta = client.post(
"/clientes/1/contatos",
data={"nome": "Fulano", "telefone": "75999090001", "csrf_token": CSRF_TESTE},
follow_redirects=True,
)
assert "já está cadastrado" in resposta.text
def test_editar_contato_com_sucesso(monkeypatch):
monkeypatch.setattr(router_contatos.db_contatos, "contato_duplicado", lambda cid, tel, ignorar_id=None: False)
chamadas = {}
monkeypatch.setattr(
router_contatos.db_contatos,
"atualizar_contato",
lambda contato_id, cid, nome, tel, email=None: chamadas.setdefault(
"dados", (contato_id, cid, nome, tel, email)
),
)
resposta = client.post(
"/clientes/1/contatos/5/editar",
data={
"nome": "Fulano Editado", "telefone": "(75) 99909-0002", "email": "fulano@example.com",
"csrf_token": CSRF_TESTE,
},
)
assert resposta.status_code == 303
assert chamadas["dados"] == (5, 1, "Fulano Editado", "5575999090002", "fulano@example.com")
def test_editar_contato_inexistente_avisa_e_nao_finge_sucesso(monkeypatch):
# atualizar_contato retorna False quando o WHERE (ID + CLIENTE_ID) não
# bate com nenhuma linha — ID errado, ou de outro cliente.
monkeypatch.setattr(router_contatos.db_contatos, "contato_duplicado", lambda cid, tel, ignorar_id=None: False)
monkeypatch.setattr(router_contatos.db_contatos, "atualizar_contato", lambda *a, **k: False)
registros = []
monkeypatch.setattr(
router_contatos.db_log, "registrar", lambda *args, **kwargs: registros.append(args)
)
resposta = client.post(
"/clientes/1/contatos/999/editar",
data={"nome": "Fulano", "telefone": "75999090002", "csrf_token": CSRF_TESTE},
follow_redirects=False,
)
assert resposta.status_code == 303
assert "mensagem_tipo=danger" in resposta.headers["location"]
assert registros[-1][4] is False # sucesso=False no log (args: usuario, acao, entidade, entidade_id, sucesso, detalhe)
def test_editar_contato_duplicado_avisa(monkeypatch):
monkeypatch.setattr(router_contatos.db_contatos, "contato_duplicado", lambda cid, tel, ignorar_id=None: True)
monkeypatch.setattr(router_clientes.db_clientes, "obter_cliente", lambda cid: {"ID": cid, "CODIGO": "000001", "RAZAO": "Cliente", "FANTASIA": "Cliente"})
monkeypatch.setattr(router_clientes.db_contatos, "listar_contatos_por_cliente", lambda cid: [])
monkeypatch.setattr(router_clientes.db_chaves, "status_chaves", lambda codigo: None)
resposta = client.post(
"/clientes/1/contatos/5/editar",
data={"nome": "Fulano", "telefone": "75999090001", "csrf_token": CSRF_TESTE},
follow_redirects=True,
)
assert "já está cadastrado" in resposta.text
def test_excluir_contato(monkeypatch):
monkeypatch.setattr(router_contatos.db_clientes, "obter_cliente", lambda cid: {"ID": cid, "CODIGO": "000001"})
chamadas = {}
monkeypatch.setattr(
router_contatos.db_contatos, "excluir_contato", lambda contato_id, cid: chamadas.setdefault("dados", (contato_id, cid))
)
resposta = client.post("/clientes/1/contatos/5/excluir", data={"csrf_token": CSRF_TESTE})
assert resposta.status_code == 303
assert chamadas["dados"] == (5, 1)
def test_excluir_contato_inexistente_avisa_e_nao_finge_sucesso(monkeypatch):
monkeypatch.setattr(router_contatos.db_contatos, "excluir_contato", lambda *a, **k: False)
registros = []
monkeypatch.setattr(
router_contatos.db_log, "registrar", lambda *args, **kwargs: registros.append(args)
)
resposta = client.post("/clientes/1/contatos/999/excluir", data={"csrf_token": CSRF_TESTE}, follow_redirects=False)
assert resposta.status_code == 303
assert "mensagem_tipo=danger" in resposta.headers["location"]
assert registros[-1][4] is False
@@ -0,0 +1,328 @@
import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
from app.auth import UsuarioLogado, exigir_login
from app.routers import funcionarios as router_funcionarios
app_teste = FastAPI()
app_teste.include_router(router_funcionarios.router)
app_teste.dependency_overrides[exigir_login] = lambda: UsuarioLogado(login="admin", nome="Admin", perfil="ADMIN")
client = TestClient(app_teste, follow_redirects=False)
CSRF_TESTE = "teste-csrf-token"
client.cookies.set("csrf_token", CSRF_TESTE)
@pytest.fixture(autouse=True)
def _sem_log_real(monkeypatch):
monkeypatch.setattr(router_funcionarios.db_log, "registrar", lambda *args, **kwargs: None)
@pytest.fixture(autouse=True)
def _admin_como_padrao(monkeypatch):
"""Todo teste roda como admin por padrão (override acima). Testes que
querem simular usuário padrão sobrescrevem de novo dentro do próprio
teste e restauram no fim via yield."""
yield
app_teste.dependency_overrides[exigir_login] = lambda: UsuarioLogado(login="admin", nome="Admin", perfil="ADMIN")
def test_usuario_padrao_toma_403_em_qualquer_rota():
app_teste.dependency_overrides[exigir_login] = lambda: UsuarioLogado(login="fulano", nome="Fulano", perfil="PADRAO")
resposta = client.get("/funcionarios")
assert resposta.status_code == 403
def test_listar_funcionarios(monkeypatch):
monkeypatch.setattr(
router_funcionarios.db_funcionarios,
"listar_funcionarios",
lambda: [{"ID": 1, "NOME": "Fulano", "LOGIN": "fulano", "PERFIL": "PADRAO", "ATIVO": 1, "CRIADO_EM": "2026-01-01"}],
)
resposta = client.get("/funcionarios")
assert resposta.status_code == 200
assert "Fulano" in resposta.text
def test_criar_funcionario_com_sucesso(monkeypatch):
monkeypatch.setattr(router_funcionarios.db_funcionarios, "login_em_uso", lambda login, ignorar_id=None: False)
criado = {}
monkeypatch.setattr(
router_funcionarios.db_funcionarios,
"criar_funcionario",
lambda nome, login, senha, perfil: criado.setdefault("dados", (nome, login, senha, perfil)) and 1,
)
resposta = client.post(
"/funcionarios/novo",
data={
"nome": "Fulano", "login": "fulano", "senha": "senhaForte123", "confirmar_senha": "senhaForte123",
"perfil": "PADRAO", "csrf_token": CSRF_TESTE,
},
)
assert resposta.status_code == 303
assert criado["dados"] == ("Fulano", "fulano", "senhaForte123", "PADRAO")
def test_criar_funcionario_senhas_diferentes_bloqueia(monkeypatch):
monkeypatch.setattr(router_funcionarios.db_funcionarios, "login_em_uso", lambda login, ignorar_id=None: False)
chamou = {}
monkeypatch.setattr(
router_funcionarios.db_funcionarios,
"criar_funcionario",
lambda nome, login, senha, perfil: chamou.setdefault("ok", True),
)
resposta = client.post(
"/funcionarios/novo",
data={
"nome": "Fulano", "login": "fulano", "senha": "senhaForte123", "confirmar_senha": "outraCoisa",
"perfil": "PADRAO", "csrf_token": CSRF_TESTE,
},
)
assert resposta.status_code == 200
assert "não coincidem" in resposta.text
assert "ok" not in chamou
def test_criar_funcionario_login_duplicado_bloqueia(monkeypatch):
monkeypatch.setattr(router_funcionarios.db_funcionarios, "login_em_uso", lambda login, ignorar_id=None: True)
chamou = {}
monkeypatch.setattr(
router_funcionarios.db_funcionarios,
"criar_funcionario",
lambda nome, login, senha, perfil: chamou.setdefault("ok", True),
)
resposta = client.post(
"/funcionarios/novo",
data={
"nome": "Fulano", "login": "fulano", "senha": "senhaForte123", "confirmar_senha": "senhaForte123",
"perfil": "PADRAO", "csrf_token": CSRF_TESTE,
},
)
assert resposta.status_code == 200
assert "já está em uso" in resposta.text
assert "ok" not in chamou
def test_criar_funcionario_perfil_invalido_bloqueia(monkeypatch):
monkeypatch.setattr(router_funcionarios.db_funcionarios, "login_em_uso", lambda login, ignorar_id=None: False)
chamou = {}
monkeypatch.setattr(
router_funcionarios.db_funcionarios,
"criar_funcionario",
lambda nome, login, senha, perfil: chamou.setdefault("ok", True),
)
resposta = client.post(
"/funcionarios/novo",
data={
"nome": "Fulano", "login": "fulano", "senha": "senhaForte123", "confirmar_senha": "senhaForte123",
"perfil": "SUPERUSUARIO", "csrf_token": CSRF_TESTE,
},
)
assert resposta.status_code == 200
assert "ok" not in chamou
def test_editar_funcionario_com_sucesso(monkeypatch):
monkeypatch.setattr(router_funcionarios.db_funcionarios, "login_em_uso", lambda login, ignorar_id=None: False)
monkeypatch.setattr(
router_funcionarios.db_funcionarios,
"obter_funcionario",
lambda fid: {"ID": fid, "NOME": "Fulano", "LOGIN": "fulano", "PERFIL": "PADRAO", "ATIVO": 1},
)
capturado = {}
monkeypatch.setattr(
router_funcionarios.db_funcionarios,
"atualizar_funcionario",
lambda fid, nome, login, perfil: capturado.setdefault("dados", (fid, nome, login, perfil)),
)
resposta = client.post(
"/funcionarios/1/editar",
data={"nome": "Fulano Editado", "login": "fulano.editado", "perfil": "PADRAO", "csrf_token": CSRF_TESTE},
)
assert resposta.status_code == 303
assert capturado["dados"] == (1, "Fulano Editado", "fulano.editado", "PADRAO")
def test_editar_ultimo_admin_para_padrao_bloqueia(monkeypatch):
monkeypatch.setattr(router_funcionarios.db_funcionarios, "login_em_uso", lambda login, ignorar_id=None: False)
monkeypatch.setattr(
router_funcionarios.db_funcionarios,
"obter_funcionario",
lambda fid: {"ID": fid, "NOME": "Admin", "LOGIN": "admin.db", "PERFIL": "ADMIN", "ATIVO": 1},
)
monkeypatch.setattr(router_funcionarios.db_funcionarios, "contar_admins_ativos", lambda: 1)
chamou = {}
monkeypatch.setattr(
router_funcionarios.db_funcionarios,
"atualizar_funcionario",
lambda fid, nome, login, perfil: chamou.setdefault("ok", True),
)
resposta = client.post(
"/funcionarios/1/editar",
data={"nome": "Admin", "login": "admin.db", "perfil": "PADRAO", "csrf_token": CSRF_TESTE},
)
assert resposta.status_code == 200
assert "único administrador" in resposta.text
assert "ok" not in chamou
def test_editar_admin_para_padrao_com_outros_admins_permite(monkeypatch):
monkeypatch.setattr(router_funcionarios.db_funcionarios, "login_em_uso", lambda login, ignorar_id=None: False)
monkeypatch.setattr(
router_funcionarios.db_funcionarios,
"obter_funcionario",
lambda fid: {"ID": fid, "NOME": "Admin", "LOGIN": "admin.db", "PERFIL": "ADMIN", "ATIVO": 1},
)
monkeypatch.setattr(router_funcionarios.db_funcionarios, "contar_admins_ativos", lambda: 2)
capturado = {}
monkeypatch.setattr(
router_funcionarios.db_funcionarios,
"atualizar_funcionario",
lambda fid, nome, login, perfil: capturado.setdefault("dados", (fid, nome, login, perfil)),
)
resposta = client.post(
"/funcionarios/1/editar",
data={"nome": "Admin", "login": "admin.db", "perfil": "PADRAO", "csrf_token": CSRF_TESTE},
)
assert resposta.status_code == 303
assert capturado["dados"] == (1, "Admin", "admin.db", "PADRAO")
def test_editar_funcionario_inexistente_nao_finge_sucesso(monkeypatch):
monkeypatch.setattr(router_funcionarios.db_funcionarios, "login_em_uso", lambda login, ignorar_id=None: False)
monkeypatch.setattr(router_funcionarios.db_funcionarios, "obter_funcionario", lambda fid: None)
chamou = {}
monkeypatch.setattr(
router_funcionarios.db_funcionarios,
"atualizar_funcionario",
lambda fid, nome, login, perfil: chamou.setdefault("ok", True),
)
resposta = client.post(
"/funcionarios/999/editar",
data={"nome": "Fulano", "login": "fulano", "perfil": "PADRAO", "csrf_token": CSRF_TESTE},
)
assert resposta.status_code == 200
assert "não encontrado" in resposta.text
assert "ok" not in chamou
def test_redefinir_senha_com_sucesso(monkeypatch):
capturado = {}
monkeypatch.setattr(
router_funcionarios.db_funcionarios,
"redefinir_senha",
lambda fid, senha: capturado.setdefault("dados", (fid, senha)),
)
resposta = client.post(
"/funcionarios/1/redefinir-senha",
data={"nova_senha": "novaSenha123", "confirmar_nova_senha": "novaSenha123", "csrf_token": CSRF_TESTE},
)
assert resposta.status_code == 303
assert capturado["dados"] == (1, "novaSenha123")
def test_redefinir_senha_nao_coincide_bloqueia(monkeypatch):
chamou = {}
monkeypatch.setattr(
router_funcionarios.db_funcionarios, "redefinir_senha", lambda fid, senha: chamou.setdefault("ok", True)
)
resposta = client.post(
"/funcionarios/1/redefinir-senha",
data={"nova_senha": "a", "confirmar_nova_senha": "b", "csrf_token": CSRF_TESTE},
follow_redirects=True,
)
assert "não coincidem" in resposta.text
assert "ok" not in chamou
def test_redefinir_senha_funcionario_inexistente_nao_finge_sucesso(monkeypatch):
monkeypatch.setattr(router_funcionarios.db_funcionarios, "redefinir_senha", lambda fid, senha: False)
resposta = client.post(
"/funcionarios/999/redefinir-senha",
data={"nova_senha": "novaSenha123", "confirmar_nova_senha": "novaSenha123", "csrf_token": CSRF_TESTE},
follow_redirects=True,
)
assert "não encontrado" in resposta.text
def test_alternar_ativo_funcionario_inexistente_nao_finge_sucesso(monkeypatch):
monkeypatch.setattr(router_funcionarios.db_funcionarios, "obter_funcionario", lambda fid: None)
chamou = {}
monkeypatch.setattr(
router_funcionarios.db_funcionarios, "alternar_ativo", lambda fid: chamou.setdefault("ok", True)
)
resposta = client.post(
"/funcionarios/999/alternar-ativo", data={"csrf_token": CSRF_TESTE}, follow_redirects=True
)
assert "não encontrado" in resposta.text
assert "ok" not in chamou
def test_alternar_ativo(monkeypatch):
monkeypatch.setattr(
router_funcionarios.db_funcionarios,
"obter_funcionario",
lambda fid: {"ID": fid, "NOME": "Fulano", "LOGIN": "fulano", "PERFIL": "PADRAO", "ATIVO": 1},
)
capturado = {}
monkeypatch.setattr(
router_funcionarios.db_funcionarios, "alternar_ativo", lambda fid: capturado.setdefault("id", fid)
)
resposta = client.post("/funcionarios/1/alternar-ativo", data={"csrf_token": CSRF_TESTE})
assert resposta.status_code == 303
assert capturado["id"] == 1
def test_desativar_ultimo_admin_ativo_bloqueia(monkeypatch):
monkeypatch.setattr(
router_funcionarios.db_funcionarios,
"obter_funcionario",
lambda fid: {"ID": fid, "NOME": "Admin", "LOGIN": "admin.db", "PERFIL": "ADMIN", "ATIVO": 1},
)
monkeypatch.setattr(router_funcionarios.db_funcionarios, "contar_admins_ativos", lambda: 1)
chamou = {}
monkeypatch.setattr(
router_funcionarios.db_funcionarios, "alternar_ativo", lambda fid: chamou.setdefault("ok", True)
)
resposta = client.post(
"/funcionarios/1/alternar-ativo", data={"csrf_token": CSRF_TESTE}, follow_redirects=True
)
assert "único administrador" in resposta.text
assert "ok" not in chamou
def test_reativar_funcionario_nao_passa_pela_checagem_de_ultimo_admin(monkeypatch):
# ATIVO indo de 0 para 1 (reativação) nunca reduz a contagem de admins
# ativos — não precisa nem chamar contar_admins_ativos().
monkeypatch.setattr(
router_funcionarios.db_funcionarios,
"obter_funcionario",
lambda fid: {"ID": fid, "NOME": "Admin", "LOGIN": "admin.db", "PERFIL": "ADMIN", "ATIVO": 0},
)
def _falha_se_chamado():
raise AssertionError("contar_admins_ativos não deveria ser chamado ao reativar")
monkeypatch.setattr(router_funcionarios.db_funcionarios, "contar_admins_ativos", _falha_se_chamado)
capturado = {}
monkeypatch.setattr(
router_funcionarios.db_funcionarios, "alternar_ativo", lambda fid: capturado.setdefault("id", fid)
)
resposta = client.post("/funcionarios/1/alternar-ativo", data={"csrf_token": CSRF_TESTE})
assert resposta.status_code == 303
assert capturado["id"] == 1
+134
View File
@@ -0,0 +1,134 @@
import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
from app.auth import UsuarioLogado, exigir_login
from app.routers import log as router_log
app_teste = FastAPI()
app_teste.include_router(router_log.router)
app_teste.dependency_overrides[exigir_login] = lambda: UsuarioLogado(login="admin", nome="Admin", perfil="ADMIN")
client = TestClient(app_teste, follow_redirects=False)
CSRF_TESTE = "teste-csrf-token"
client.cookies.set("csrf_token", CSRF_TESTE)
@pytest.fixture(autouse=True)
def _admin_como_padrao():
"""Todo teste roda como admin por padrão (override acima). Testes que
querem simular usuário padrão sobrescrevem de novo dentro do próprio
teste e restauram no fim via yield."""
yield
app_teste.dependency_overrides[exigir_login] = lambda: UsuarioLogado(login="admin", nome="Admin", perfil="ADMIN")
@pytest.fixture(autouse=True)
def _sem_log_real(monkeypatch):
monkeypatch.setattr(router_log.db_log, "registrar", lambda *args, **kwargs: None)
def test_pagina_log_mostra_acoes_do_painel(monkeypatch):
capturado = {}
monkeypatch.setattr(
router_log.db_log,
"listar",
lambda limite=100: capturado.setdefault("limite", limite) and [
{
"ID": 1, "CRIADO_EM": "2026-08-13 10:00:00", "USUARIO": "admin", "ACAO": "criar",
"ENTIDADE": "cliente", "ENTIDADE_ID": "000001", "SUCESSO": 1, "DETALHE": None,
}
],
)
monkeypatch.setattr(router_log.n8n_client, "listar_envios", lambda limite=50: [])
monkeypatch.setattr(router_log.n8n_client, "configurado", lambda: False)
resposta = client.get("/log")
assert resposta.status_code == 200
assert "admin" in resposta.text
assert "criar" in resposta.text
assert "N8N_API_URL" in resposta.text # aviso de integração não configurada
assert capturado["limite"] == 20
def test_pagina_log_mostra_envios_quando_configurado(monkeypatch):
monkeypatch.setattr(router_log.db_log, "listar", lambda limite=20: [])
monkeypatch.setattr(
router_log.n8n_client,
"listar_envios",
lambda limite=50: [
{
"data_processamento": "2026-08-13T10:00:00", "nome_cliente": "Cliente Teste",
"codigo_cliente": "000001", "competencia": "2026-08", "status": "enviado",
"telefones_enviados": "5575999090000", "telefones_com_erro": "",
}
],
)
monkeypatch.setattr(router_log.n8n_client, "configurado", lambda: True)
resposta = client.get("/log")
assert resposta.status_code == 200
assert "Cliente Teste" in resposta.text
assert "enviado" in resposta.text
def test_pagina_log_admin_ve_botao_de_limpar(monkeypatch):
monkeypatch.setattr(router_log.db_log, "listar", lambda limite=20: [])
monkeypatch.setattr(router_log.n8n_client, "listar_envios", lambda limite=50: [])
monkeypatch.setattr(router_log.n8n_client, "configurado", lambda: False)
resposta = client.get("/log")
assert resposta.status_code == 200
assert "/log/limpar" in resposta.text
def test_pagina_log_usuario_padrao_nao_ve_botao_de_limpar(monkeypatch):
app_teste.dependency_overrides[exigir_login] = lambda: UsuarioLogado(login="fulano", nome="Fulano", perfil="PADRAO")
monkeypatch.setattr(router_log.db_log, "listar", lambda limite=20: [])
monkeypatch.setattr(router_log.n8n_client, "listar_envios", lambda limite=50: [])
monkeypatch.setattr(router_log.n8n_client, "configurado", lambda: False)
resposta = client.get("/log")
assert resposta.status_code == 200
assert "/log/limpar" not in resposta.text
def test_limpar_log_admin_funciona(monkeypatch):
chamou = {}
monkeypatch.setattr(router_log.db_log, "limpar", lambda: chamou.setdefault("limpou", True))
resposta = client.post("/log/limpar", data={"csrf_token": CSRF_TESTE})
assert resposta.status_code == 303
assert resposta.headers["location"].startswith("/log?")
assert chamou.get("limpou") is True
def test_limpar_log_usuario_padrao_toma_403():
app_teste.dependency_overrides[exigir_login] = lambda: UsuarioLogado(login="fulano", nome="Fulano", perfil="PADRAO")
resposta = client.post("/log/limpar", data={"csrf_token": CSRF_TESTE})
assert resposta.status_code == 403
def test_pagina_log_secoes_comecam_fechadas(monkeypatch):
monkeypatch.setattr(router_log.db_log, "listar", lambda limite=20: [])
monkeypatch.setattr(router_log.n8n_client, "listar_envios", lambda limite=50: [])
monkeypatch.setattr(router_log.n8n_client, "configurado", lambda: False)
resposta = client.get("/log")
assert resposta.status_code == 200
texto = resposta.text
assert texto.count('<details class="secao-log">') == 2
assert "<details class=\"secao-log\" open>" not in texto
assert "<details class='secao-log' open>" not in texto
def test_pagina_log_botao_toggle_tem_handler_de_clique_proprio(monkeypatch):
"""Regressão: um <button> dentro de <summary> não propaga o clique para
o toggle nativo do <details> — precisa de um listener de clique
explícito que abre/fecha a seção manualmente (ver log.html)."""
monkeypatch.setattr(router_log.db_log, "listar", lambda limite=20: [])
monkeypatch.setattr(router_log.n8n_client, "listar_envios", lambda limite=50: [])
monkeypatch.setattr(router_log.n8n_client, "configurado", lambda: False)
resposta = client.get("/log")
assert resposta.status_code == 200
assert 'botao.addEventListener("click"' in resposta.text
assert "secao.open = !secao.open" in resposta.text
@@ -0,0 +1,142 @@
import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
from app.auth import UsuarioLogado, exigir_login
from app.routers import minha_conta as router_minha_conta
app_teste = FastAPI()
app_teste.include_router(router_minha_conta.router)
client = TestClient(app_teste, follow_redirects=False)
CSRF_TESTE = "teste-csrf-token"
client.cookies.set("csrf_token", CSRF_TESTE)
@pytest.fixture(autouse=True)
def _sem_log_real(monkeypatch):
monkeypatch.setattr(router_minha_conta.db_log, "registrar", lambda *args, **kwargs: None)
def _logar_como(login: str, nome: str, perfil: str = "PADRAO"):
app_teste.dependency_overrides[exigir_login] = lambda: UsuarioLogado(login=login, nome=nome, perfil=perfil)
def test_get_minha_conta_mostra_nome_e_login(monkeypatch):
_logar_como("fulano", "Fulano da Silva")
monkeypatch.setattr(
router_minha_conta.db_funcionarios,
"obter_por_login",
lambda login: {"ID": 1, "NOME": "Fulano da Silva", "LOGIN": "fulano", "SENHA_HASH": "x", "PERFIL": "PADRAO", "ATIVO": 1},
)
resposta = client.get("/minha-conta")
assert resposta.status_code == 200
assert "Fulano da Silva" in resposta.text
assert "fulano" in resposta.text
def test_get_minha_conta_admin_env_mostra_aviso(monkeypatch):
import os
monkeypatch.setenv("ADMIN_USER", "admin")
_logar_como("admin", "admin")
monkeypatch.setattr(router_minha_conta.db_funcionarios, "obter_por_login", lambda login: None)
resposta = client.get("/minha-conta")
assert resposta.status_code == 200
assert ".env" in resposta.text
def test_post_trocar_senha_com_sucesso(monkeypatch):
_logar_como("fulano", "Fulano da Silva")
monkeypatch.setattr(
router_minha_conta.db_funcionarios,
"obter_por_login",
lambda login: {"ID": 1, "NOME": "Fulano da Silva", "LOGIN": "fulano", "SENHA_HASH": "hash-atual", "PERFIL": "PADRAO", "ATIVO": 1},
)
monkeypatch.setattr(router_minha_conta, "verificar_senha", lambda senha, hash_: True)
capturado = {}
monkeypatch.setattr(
router_minha_conta.db_funcionarios,
"redefinir_senha",
lambda fid, nova: capturado.setdefault("dados", (fid, nova)),
)
resposta = client.post(
"/minha-conta/senha",
data={
"senha_atual": "senhaAtual123", "nova_senha": "senhaNova456", "confirmar_nova_senha": "senhaNova456",
"csrf_token": CSRF_TESTE,
},
follow_redirects=True,
)
assert resposta.status_code == 200
assert capturado["dados"] == (1, "senhaNova456")
def test_post_trocar_senha_atual_errada_bloqueia(monkeypatch):
_logar_como("fulano", "Fulano da Silva")
monkeypatch.setattr(
router_minha_conta.db_funcionarios,
"obter_por_login",
lambda login: {"ID": 1, "NOME": "Fulano da Silva", "LOGIN": "fulano", "SENHA_HASH": "hash-atual", "PERFIL": "PADRAO", "ATIVO": 1},
)
monkeypatch.setattr(router_minha_conta, "verificar_senha", lambda senha, hash_: False)
chamou = {}
monkeypatch.setattr(
router_minha_conta.db_funcionarios, "redefinir_senha", lambda fid, nova: chamou.setdefault("ok", True)
)
resposta = client.post(
"/minha-conta/senha",
data={
"senha_atual": "senhaErrada", "nova_senha": "senhaNova456", "confirmar_nova_senha": "senhaNova456",
"csrf_token": CSRF_TESTE,
},
)
assert resposta.status_code == 200
assert "senha atual" in resposta.text.lower()
assert "ok" not in chamou
def test_post_trocar_senha_confirmacao_nao_bate_bloqueia(monkeypatch):
_logar_como("fulano", "Fulano da Silva")
monkeypatch.setattr(
router_minha_conta.db_funcionarios,
"obter_por_login",
lambda login: {"ID": 1, "NOME": "Fulano da Silva", "LOGIN": "fulano", "SENHA_HASH": "hash-atual", "PERFIL": "PADRAO", "ATIVO": 1},
)
monkeypatch.setattr(router_minha_conta, "verificar_senha", lambda senha, hash_: True)
chamou = {}
monkeypatch.setattr(
router_minha_conta.db_funcionarios, "redefinir_senha", lambda fid, nova: chamou.setdefault("ok", True)
)
resposta = client.post(
"/minha-conta/senha",
data={
"senha_atual": "senhaAtual123", "nova_senha": "senhaNova456", "confirmar_nova_senha": "outraCoisa",
"csrf_token": CSRF_TESTE,
},
)
assert resposta.status_code == 200
assert "não coincidem" in resposta.text
assert "ok" not in chamou
def test_post_trocar_senha_admin_env_bloqueia(monkeypatch):
_logar_como("admin", "admin")
monkeypatch.setattr(router_minha_conta.db_funcionarios, "obter_por_login", lambda login: None)
chamou = {}
monkeypatch.setattr(
router_minha_conta.db_funcionarios, "redefinir_senha", lambda fid, nova: chamou.setdefault("ok", True)
)
resposta = client.post(
"/minha-conta/senha",
data={
"senha_atual": "qualquer", "nova_senha": "senhaNova456", "confirmar_nova_senha": "senhaNova456",
"csrf_token": CSRF_TESTE,
},
)
assert resposta.status_code == 200
assert ".env" in resposta.text
assert "ok" not in chamou
+26
View File
@@ -0,0 +1,26 @@
from app.senha import gerar_hash, verificar_senha
def test_hash_confere_com_senha_correta():
h = gerar_hash("minhaSenha123")
assert verificar_senha("minhaSenha123", h) is True
def test_hash_rejeita_senha_errada():
h = gerar_hash("minhaSenha123")
assert verificar_senha("senhaErrada", h) is False
def test_hashes_de_senhas_iguais_sao_diferentes():
# Salt aleatório por chamada — dois hashes da mesma senha não são iguais.
h1 = gerar_hash("mesmaSenha")
h2 = gerar_hash("mesmaSenha")
assert h1 != h2
assert verificar_senha("mesmaSenha", h1) is True
assert verificar_senha("mesmaSenha", h2) is True
def test_hash_malformado_nao_lanca():
assert verificar_senha("qualquer", "isso-nao-e-um-hash-valido") is False
assert verificar_senha("qualquer", "") is False
assert verificar_senha("qualquer", "md5$abc$def") is False
+47
View File
@@ -0,0 +1,47 @@
import time
from app import session
def test_criar_e_ler_token_valido():
token = session.criar_token("fulano", "Fulano da Silva", "ADMIN")
dados = session.ler_token(token)
assert dados == {"login": "fulano", "nome": "Fulano da Silva", "perfil": "ADMIN"}
def test_ler_token_none_retorna_none():
assert session.ler_token(None) is None
def test_ler_token_vazio_retorna_none():
assert session.ler_token("") is None
def test_ler_token_sem_ponto_retorna_none():
assert session.ler_token("lixo-sem-formato") is None
def test_ler_token_assinatura_adulterada_retorna_none():
token = session.criar_token("fulano", "Fulano", "PADRAO")
payload, assinatura = token.rsplit(".", 1)
token_adulterado = payload + "." + ("0" if assinatura[0] != "0" else "1") + assinatura[1:]
assert session.ler_token(token_adulterado) is None
def test_ler_token_payload_adulterado_retorna_none():
token = session.criar_token("fulano", "Fulano", "PADRAO")
payload, assinatura = token.rsplit(".", 1)
token_adulterado = payload + "X." + assinatura
assert session.ler_token(token_adulterado) is None
def test_ler_token_expirado_retorna_none():
agora = time.time()
token = session.criar_token("fulano", "Fulano", "PADRAO", agora=agora - session.VALIDADE_SEGUNDOS - 1)
assert session.ler_token(token, agora=agora) is None
def test_ler_token_no_limite_da_validade_ainda_e_valido():
agora = time.time()
token = session.criar_token("fulano", "Fulano", "PADRAO", agora=agora - session.VALIDADE_SEGUNDOS + 5)
assert session.ler_token(token, agora=agora) is not None
+123
View File
@@ -0,0 +1,123 @@
import base64
import re
from urllib.parse import parse_qs, urlsplit
import jwt
import pytest
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from cryptography.hazmat.primitives.serialization import Encoding, NoEncryption, PrivateFormat
from fastapi.testclient import TestClient
from app import session
from app.db import funcionarios as db_funcionarios
from app.db import log as db_log
from app.main import app
CALLBACK = "https://gtifiscal.gtigrupo.com.br/auth/sso/callback"
STATE = "estadoDoNavegador_0123456789"
ATIVO = {"ID": 7, "NOME": "Maria Teste", "LOGIN": "maria", "SENHA_HASH": "x", "PERFIL": "PADRAO", "ATIVO": 1}
@pytest.fixture
def chave(monkeypatch):
privada = Ed25519PrivateKey.generate()
semente = privada.private_bytes(Encoding.Raw, PrivateFormat.Raw, NoEncryption())
monkeypatch.setenv("SSO_FISCAL_CHAVE_PRIVADA", base64.urlsafe_b64encode(semente).decode().rstrip("="))
monkeypatch.setenv("SSO_FISCAL_CALLBACK_URL", CALLBACK)
monkeypatch.setenv("SSO_GERENCIADOR_URL_PUBLICA", "https://gtigerenciador.gtigrupo.com.br")
monkeypatch.setattr(db_log, "registrar", lambda *a, **k: None)
return privada.public_key()
def _cliente(funcionario=ATIVO, logado=True, https=True, monkeypatch=None):
monkeypatch.setattr(db_funcionarios, "obter_por_login", lambda login: funcionario if funcionario and login == funcionario["LOGIN"] else None)
cliente = TestClient(app, base_url="https://testserver" if https else "http://testserver")
if logado:
login = funcionario["LOGIN"] if funcionario else "admin"
cliente.cookies.set(session.SESSION_COOKIE_NAME, session.criar_token(login, "Maria Teste", "PADRAO"))
return cliente
def _assercao(resposta):
return re.search(r'name="assertion" value="([^"]+)"', resposta.text).group(1)
def test_funcionario_ativo_recebe_formulario_com_assercao_valida(chave, monkeypatch):
resposta = _cliente(monkeypatch=monkeypatch).get("/sso/fiscal", params={"next": "/agents", "state": STATE})
assert resposta.status_code == 200
assert f'action="{CALLBACK}"' in resposta.text and 'name="next" value="/agents"' in resposta.text
claims = jwt.decode(_assercao(resposta), chave, algorithms=["EdDSA"], audience="gti-fiscal", issuer="gti-gerenciador", options={"require": ["exp", "iat", "nbf", "iss", "aud", "sub", "jti", "nonce"]})
assert claims["nonce"] == STATE
assert claims["sub"] == "funcionario:7" and claims["login"] == "maria" and claims["perfil"] == "PADRAO"
assert claims["exp"] - claims["iat"] == 60
assert "password" not in claims and "SENHA_HASH" not in str(claims)
csp = resposta.headers["content-security-policy"]
assert "form-action https://gtifiscal.gtigrupo.com.br;" in csp and "'self'" not in csp.split("form-action")[1].split(";")[0]
assert resposta.headers["cache-control"] == "no-store"
# A asserção nunca aparece na URL
assert "assertion" not in str(resposta.url)
def test_cada_emissao_tem_jti_diferente(chave, monkeypatch):
cliente = _cliente(monkeypatch=monkeypatch)
a, b = (jwt.decode(_assercao(cliente.get("/sso/fiscal", params={"state": STATE})), chave, algorithms=["EdDSA"], audience="gti-fiscal") for _ in range(2))
assert a["jti"] != b["jti"]
def test_sem_sessao_vai_para_login_e_volta_ao_sso(chave, monkeypatch):
resposta = _cliente(logado=False, monkeypatch=monkeypatch).get("/sso/fiscal", params={"next": "/jobs", "state": STATE}, follow_redirects=False)
assert resposta.status_code == 303
destino = urlsplit(resposta.headers["location"])
assert destino.path == "/login"
assert parse_qs(destino.query)["next"] == [f"/sso/fiscal?state={STATE}&next=%2Fjobs"]
def test_funcionario_inativo_nao_recebe_assercao(chave, monkeypatch):
resposta = _cliente(funcionario={**ATIVO, "ATIVO": 0}, monkeypatch=monkeypatch).get("/sso/fiscal", params={"state": STATE}, follow_redirects=False)
assert resposta.status_code == 303 and resposta.headers["location"].startswith("/login")
assert "assertion" not in resposta.text
def test_conta_de_bootstrap_nao_usa_sso(chave, monkeypatch):
monkeypatch.setenv("ADMIN_USER", "admin")
resposta = _cliente(funcionario=None, monkeypatch=monkeypatch).get("/sso/fiscal", params={"state": STATE}, follow_redirects=False)
assert resposta.status_code == 403 and "assertion" not in resposta.text
def test_http_redireciona_para_endereco_https_publico(chave, monkeypatch):
resposta = _cliente(https=False, monkeypatch=monkeypatch).get("/sso/fiscal", params={"next": "/blockers", "state": STATE}, follow_redirects=False)
assert resposta.status_code == 303
assert resposta.headers["location"] == f"https://gtigerenciador.gtigrupo.com.br/sso/fiscal?state={STATE}&next=%2Fblockers"
@pytest.mark.parametrize("malicioso", ["//evil.com/x", "https://evil.com", "javascript:alert(1)", "/\\evil.com", "data:text/html,x", "agents", "/a\nb"])
def test_next_malicioso_vira_raiz(chave, monkeypatch, malicioso):
resposta = _cliente(monkeypatch=monkeypatch).get("/sso/fiscal", params={"next": malicioso, "state": STATE})
assert 'name="next" value="/"' in resposta.text
def test_callback_so_https_configurado(chave, monkeypatch):
monkeypatch.setenv("SSO_FISCAL_CALLBACK_URL", "http://gtifiscal.gtigrupo.com.br/auth/sso/callback")
with pytest.raises(RuntimeError):
_cliente(monkeypatch=monkeypatch).get("/sso/fiscal", params={"state": STATE})
def test_sem_configuracao_nao_emite(monkeypatch):
monkeypatch.delenv("SSO_FISCAL_CHAVE_PRIVADA", raising=False)
monkeypatch.setattr(db_log, "registrar", lambda *a, **k: None)
resposta = _cliente(monkeypatch=monkeypatch).get("/sso/fiscal", params={"state": STATE})
assert resposta.status_code == 503
def test_outras_paginas_mantem_csp_padrao(chave, monkeypatch):
resposta = _cliente(monkeypatch=monkeypatch).get("/login")
assert "form-action 'self'" in resposta.headers["content-security-policy"]
@pytest.mark.parametrize("estado", [None, "curto", "com espaço e símbolos!!!!!!", "x" * 200])
def test_sem_state_valido_manda_para_o_login_do_fiscal(chave, monkeypatch, estado):
params = {"next": "/agents"} | ({"state": estado} if estado is not None else {})
resposta = _cliente(monkeypatch=monkeypatch).get("/sso/fiscal", params=params, follow_redirects=False)
assert resposta.status_code == 303
assert resposta.headers["location"] == "https://gtifiscal.gtigrupo.com.br/login?next=%2Fagents"
assert "assertion" not in resposta.text