feat: log de acoes/envios, ordenacao/filtro, codigo sequencial, campos opcionais
- Tabela LOG_PAINEL (nova, ja criada no banco de dev) + app/db/log.py: toda criacao/edicao/exclusao de cliente e contato fica registrada (usuario, acao, sucesso/erro, detalhe). Tela /log mostra esse historico. - app/n8n_client.py: leitura opcional (N8N_API_URL/N8N_API_KEY) da Data Table pagamentos_processados do n8n, mostrando os envios automaticos de chave na mesma tela /log. Degrada de forma segura (lista vazia) se nao configurado ou se a chamada falhar. - Lista de clientes ganha ordenacao configuravel (codigo/razao/ fantasia/CNPJ) via whitelist, alem da busca que ja existia. - CODIGO duplicado agora BLOQUEIA de verdade na criacao/edicao (sem bypass) — antes so avisava com opcao de confirmar mesmo assim. - Formulario de novo cliente sugere o proximo CODIGO sequencial (editavel), ignorando a faixa reservada para dados de teste (>= 900000) para nao ser contaminado por sobra de teste no banco. - IMPLANTACAO/MENSALIDADE/JUROS_MORA/JUROS_MULTA/DATA_CONTRATO/ VENCIMENTO_CONTRATO deixam de ser obrigatorios — em branco, gravam default (0,00 / hoje) em vez de bloquear o cadastro. Só CODIGO/CNPJ/ RAZAO/FANTASIA continuam obrigatorios. - Limpeza: removidas 10 linhas de teste vazadas (CODIGO 999902) que ficaram no banco de dev por sessoes de teste interrompidas. 69/69 testes passando (16 novos).
This commit is contained in:
Gustavo
@@ -12,3 +12,10 @@ DB_PORT=3307
|
||||
DB_NAME=gtisoftw_clientes_dev
|
||||
DB_USER=root
|
||||
DB_PASSWORD=devpassword123
|
||||
|
||||
# Opcional: integração de LEITURA com a Data Table "pagamentos_processados"
|
||||
# do n8n, para mostrar os envios automáticos de chave na tela /log. Gere a
|
||||
# chave em Configurações > API dentro do n8n (devn8n.gtigrupo.com.br).
|
||||
# Se deixar em branco, a tela /log só mostra o log de ações do painel.
|
||||
N8N_API_URL=
|
||||
N8N_API_KEY=
|
||||
|
||||
@@ -2,7 +2,11 @@
|
||||
|
||||
Painel administrativo (FastAPI + Jinja2) para gestão de `CLIENTE`, `CLIENTE_CONTATO` e geração manual de chave de licenciamento no MySQL do sistema legado.
|
||||
|
||||
Proteções: login HTTP Basic em toda rota, CSRF via cookie de duplo envio (`app/csrf.py` — todo formulário que muda estado embute o mesmo token do cookie `csrf_token`, definido pelo middleware em `app/main.py`) e a lista de clientes é paginada (100 por página).
|
||||
Proteções: login HTTP Basic em toda rota, CSRF via cookie de duplo envio (`app/csrf.py` — todo formulário que muda estado embute o mesmo token do cookie `csrf_token`, definido pelo middleware em `app/main.py`) e a lista de clientes é paginada (100 por página), com busca e ordenação por código/razão/fantasia/CNPJ.
|
||||
|
||||
`CODIGO` de cliente não aceita duplicado (bloqueio duro, sem bypass) e o formulário de novo cliente já sugere o próximo código sequencial (editável). Só `CODIGO`, `CNPJ`, `RAZAO` e `FANTASIA` são obrigatórios — os campos de contrato (implantação, mensalidade, juros, datas) podem ficar em branco e recebem default (0,00 / data de hoje).
|
||||
|
||||
A rota `/log` mostra o histórico de ações do painel (tabela `LOG_PAINEL`, criada no banco de dev — **precisa ser criada manualmente em produção também**, DDL no início de `app/db/log.py`) e, se `N8N_API_URL`/`N8N_API_KEY` estiverem configuradas (`.env.example`), os últimos envios automáticos de chave via WhatsApp (lidos da Data Table `pagamentos_processados` do n8n, somente leitura).
|
||||
|
||||
## Rodando localmente (sem Docker)
|
||||
|
||||
|
||||
+41
-3
@@ -16,12 +16,25 @@ CAMPOS_CLIENTE = [
|
||||
|
||||
_COLUNAS_SELECT = ", ".join(["ID", *CAMPOS_CLIENTE])
|
||||
|
||||
# Whitelist de colunas aceitas para ordenar a listagem — nunca interpolar
|
||||
# o valor de "ordenar_por" vindo da URL direto no SQL sem passar por aqui.
|
||||
ORDENACOES_VALIDAS = {
|
||||
"codigo": "CODIGO",
|
||||
"razao": "RAZAO",
|
||||
"fantasia": "FANTASIA",
|
||||
"cnpj": "CNPJ",
|
||||
}
|
||||
ORDENACAO_PADRAO = "razao"
|
||||
|
||||
|
||||
class ClienteComDependenciasError(Exception):
|
||||
"""Levantada ao tentar excluir um CLIENTE que ainda tem CLIENTE_CONTATO."""
|
||||
|
||||
|
||||
def listar_clientes(busca: str | None = None, limite: int = 100, offset: int = 0) -> list[dict]:
|
||||
def listar_clientes(
|
||||
busca: str | None = None, limite: int = 100, offset: int = 0, ordenar_por: str = ORDENACAO_PADRAO
|
||||
) -> list[dict]:
|
||||
coluna_ordenacao = ORDENACOES_VALIDAS.get(ordenar_por, ORDENACOES_VALIDAS[ORDENACAO_PADRAO])
|
||||
conn = get_connection()
|
||||
try:
|
||||
with conn.cursor() as cur:
|
||||
@@ -29,19 +42,44 @@ def listar_clientes(busca: str | None = None, limite: int = 100, offset: int = 0
|
||||
sql = (
|
||||
f"SELECT {_COLUNAS_SELECT} FROM CLIENTE "
|
||||
"WHERE CODIGO LIKE %s OR CNPJ LIKE %s OR RAZAO LIKE %s OR FANTASIA LIKE %s "
|
||||
"ORDER BY RAZAO LIMIT %s OFFSET %s"
|
||||
f"ORDER BY {coluna_ordenacao} LIMIT %s OFFSET %s"
|
||||
)
|
||||
termo = f"%{busca}%"
|
||||
cur.execute(sql, (termo, termo, termo, termo, limite, offset))
|
||||
else:
|
||||
cur.execute(
|
||||
f"SELECT {_COLUNAS_SELECT} FROM CLIENTE ORDER BY RAZAO LIMIT %s OFFSET %s", (limite, offset)
|
||||
f"SELECT {_COLUNAS_SELECT} FROM CLIENTE ORDER BY {coluna_ordenacao} LIMIT %s OFFSET %s",
|
||||
(limite, offset),
|
||||
)
|
||||
return cur.fetchall()
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
def proximo_codigo_sugerido() -> str:
|
||||
"""
|
||||
Sugere o próximo CODIGO sequencial (maior CODIGO numérico + 1, com zeros
|
||||
à esquerda até 6 dígitos). É só uma sugestão pré-preenchida no formulário
|
||||
de novo cliente — o campo continua editável, e codigo_em_uso() ainda
|
||||
confere unicidade de verdade antes de gravar.
|
||||
"""
|
||||
conn = get_connection()
|
||||
try:
|
||||
with conn.cursor() as cur:
|
||||
# CODIGO >= 900000 é a faixa reservada para dados de teste deste
|
||||
# painel (ver tests/test_db_clientes.py) — excluída aqui para a
|
||||
# sugestão não disparar lá em cima se sobrar alguma linha de teste.
|
||||
cur.execute(
|
||||
"SELECT MAX(CAST(CODIGO AS UNSIGNED)) AS maior FROM CLIENTE "
|
||||
"WHERE CODIGO REGEXP '^[0-9]+$' AND CAST(CODIGO AS UNSIGNED) < 900000"
|
||||
)
|
||||
maior = cur.fetchone()["maior"]
|
||||
finally:
|
||||
conn.close()
|
||||
proximo = (maior or 0) + 1
|
||||
return str(proximo).zfill(6)
|
||||
|
||||
|
||||
def contar_clientes(busca: str | None = None) -> int:
|
||||
conn = get_connection()
|
||||
try:
|
||||
|
||||
@@ -0,0 +1,67 @@
|
||||
"""
|
||||
Log de ações do painel (LOG_PAINEL) — uma linha por ação que muda estado
|
||||
(criar/editar/excluir cliente, adicionar/editar/excluir contato), sucesso
|
||||
ou erro. Escrita feita pelos routers via `registrar(...)`, nunca
|
||||
sobrescrita/editada depois — é histórico, não estado.
|
||||
|
||||
A tabela já existe no banco de dev. Em qualquer outro ambiente (produção),
|
||||
crie-a primeiro com:
|
||||
|
||||
CREATE TABLE LOG_PAINEL (
|
||||
ID INT UNSIGNED NOT NULL AUTO_INCREMENT,
|
||||
CRIADO_EM DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||||
USUARIO VARCHAR(60) NOT NULL,
|
||||
ACAO VARCHAR(40) NOT NULL,
|
||||
ENTIDADE VARCHAR(20) NOT NULL,
|
||||
ENTIDADE_ID VARCHAR(20) DEFAULT NULL,
|
||||
SUCESSO TINYINT(1) NOT NULL,
|
||||
DETALHE VARCHAR(500) DEFAULT NULL,
|
||||
PRIMARY KEY (ID),
|
||||
KEY IDX_LOG_PAINEL_CRIADO_EM (CRIADO_EM)
|
||||
) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_unicode_ci;
|
||||
"""
|
||||
from .connection import get_connection
|
||||
|
||||
|
||||
def registrar(
|
||||
usuario: str,
|
||||
acao: str,
|
||||
entidade: str,
|
||||
entidade_id: str | None,
|
||||
sucesso: bool,
|
||||
detalhe: str | None = None,
|
||||
) -> None:
|
||||
conn = get_connection()
|
||||
try:
|
||||
with conn.cursor() as cur:
|
||||
cur.execute(
|
||||
"INSERT INTO LOG_PAINEL (USUARIO, ACAO, ENTIDADE, ENTIDADE_ID, SUCESSO, DETALHE) "
|
||||
"VALUES (%s, %s, %s, %s, %s, %s)",
|
||||
(usuario, acao, entidade, entidade_id, 1 if sucesso else 0, detalhe),
|
||||
)
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
def listar(limite: int = 100, offset: int = 0) -> list[dict]:
|
||||
conn = get_connection()
|
||||
try:
|
||||
with conn.cursor() as cur:
|
||||
cur.execute(
|
||||
"SELECT ID, CRIADO_EM, USUARIO, ACAO, ENTIDADE, ENTIDADE_ID, SUCESSO, DETALHE "
|
||||
"FROM LOG_PAINEL ORDER BY CRIADO_EM DESC, ID DESC LIMIT %s OFFSET %s",
|
||||
(limite, offset),
|
||||
)
|
||||
return cur.fetchall()
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
def contar() -> int:
|
||||
conn = get_connection()
|
||||
try:
|
||||
with conn.cursor() as cur:
|
||||
cur.execute("SELECT COUNT(*) AS total FROM LOG_PAINEL")
|
||||
return cur.fetchone()["total"]
|
||||
finally:
|
||||
conn.close()
|
||||
+2
-1
@@ -8,13 +8,14 @@ from fastapi.staticfiles import StaticFiles
|
||||
|
||||
from .auth import exigir_login
|
||||
from .csrf import CSRF_COOKIE_NAME, gerar_token
|
||||
from .routers import chave_manual, clientes, contatos
|
||||
from .routers import chave_manual, clientes, contatos, log
|
||||
|
||||
app = FastAPI(title="Painel de Clientes", docs_url=None, redoc_url=None, openapi_url=None)
|
||||
app.mount("/static", StaticFiles(directory="app/static"), name="static")
|
||||
app.include_router(clientes.router)
|
||||
app.include_router(contatos.router)
|
||||
app.include_router(chave_manual.router)
|
||||
app.include_router(log.router)
|
||||
|
||||
|
||||
@app.middleware("http")
|
||||
|
||||
@@ -0,0 +1,52 @@
|
||||
"""
|
||||
Cliente HTTP mínimo para ler a Data Table "pagamentos_processados" do n8n —
|
||||
o histórico de envios automáticos de chave de reativação via WhatsApp,
|
||||
mantido pelo workflow "Pagamento Confirmado Asaas..." (ver
|
||||
licenciamento-n8n/docs/). Cada linha é um pagamento processado, nunca
|
||||
sobrescrita — é o histórico que o sistema legado nunca teve.
|
||||
|
||||
Requer N8N_API_URL e N8N_API_KEY no ambiente (uma API key gerada em
|
||||
Configurações > API do n8n, com o escopo de leitura de linhas de Data
|
||||
Table). Se não estiverem configuradas, ou se a chamada falhar por
|
||||
qualquer motivo (rede, chave errada, endpoint diferente do esperado),
|
||||
listar_envios() devolve [] — a tela de log mostra um aviso em vez de
|
||||
quebrar. Este módulo NUNCA escreve na Data Table, só lê.
|
||||
"""
|
||||
import os
|
||||
|
||||
import httpx
|
||||
|
||||
DATA_TABLE_ID = "bJHhFUDwFJ30YJVt" # pagamentos_processados
|
||||
|
||||
|
||||
def configurado() -> bool:
|
||||
return bool(os.environ.get("N8N_API_URL")) and bool(os.environ.get("N8N_API_KEY"))
|
||||
|
||||
|
||||
def listar_envios(limite: int = 50) -> list[dict]:
|
||||
"""
|
||||
Últimas linhas de pagamentos_processados, mais recentes primeiro.
|
||||
Lista vazia se a integração não estiver configurada ou a chamada falhar.
|
||||
"""
|
||||
if not configurado():
|
||||
return []
|
||||
|
||||
base = os.environ["N8N_API_URL"].rstrip("/")
|
||||
url = f"{base}/api/v1/data-tables/{DATA_TABLE_ID}/rows"
|
||||
try:
|
||||
resposta = httpx.get(
|
||||
url,
|
||||
headers={"X-N8N-API-KEY": os.environ["N8N_API_KEY"], "accept": "application/json"},
|
||||
params={"take": limite, "sortBy": "data_processamento:desc"},
|
||||
timeout=5.0,
|
||||
)
|
||||
resposta.raise_for_status()
|
||||
corpo = resposta.json()
|
||||
except (httpx.HTTPError, ValueError):
|
||||
return []
|
||||
|
||||
if isinstance(corpo, dict):
|
||||
return corpo.get("data") or []
|
||||
if isinstance(corpo, list):
|
||||
return corpo
|
||||
return []
|
||||
+87
-49
@@ -1,8 +1,9 @@
|
||||
# app/routers/clientes.py
|
||||
"""
|
||||
Rotas de gestão de CLIENTE: listar/buscar (paginado), criar, editar, ver
|
||||
detalhe (contatos + status de Clientes_Chaves), excluir.
|
||||
Rotas de gestão de CLIENTE: listar/buscar/ordenar (paginado), criar,
|
||||
editar, ver detalhe (contatos + status de Clientes_Chaves), excluir.
|
||||
"""
|
||||
import datetime
|
||||
import math
|
||||
from typing import Optional
|
||||
from urllib.parse import urlencode
|
||||
@@ -17,6 +18,7 @@ from ..csrf import exigir_csrf, token_do_contexto
|
||||
from ..db import chaves as db_chaves
|
||||
from ..db import clientes as db_clientes
|
||||
from ..db import contatos as db_contatos
|
||||
from ..db import log as db_log
|
||||
from ..normalize import normalizar_documento, normalizar_telefone, validar_dv_cpf, validar_dv_cnpj
|
||||
|
||||
router = APIRouter(prefix="/clientes")
|
||||
@@ -24,6 +26,13 @@ templates = Jinja2Templates(directory="app/templates")
|
||||
|
||||
TAMANHO_PAGINA = 100
|
||||
|
||||
# Únicos campos que o formulário exige preenchidos — os demais (contrato,
|
||||
# valores, datas) podem ficar em branco e recebem um default sensato
|
||||
# (ver _aplicar_defaults_opcionais) em vez de bloquear o cadastro.
|
||||
CAMPOS_OBRIGATORIOS = ["CODIGO", "CNPJ", "RAZAO", "FANTASIA"]
|
||||
_CAMPOS_DECIMAIS_OPCIONAIS = ["IMPLANTACAO", "MENSALIDADE", "JUROS_MORA", "JUROS_MULTA"]
|
||||
_CAMPOS_DATA_OPCIONAIS = ["DATA_CONTRATO", "VENCIMENTO_CONTRATO"]
|
||||
|
||||
|
||||
def _redirect_com_mensagem(url: str, mensagem: str, tipo: str = "success") -> RedirectResponse:
|
||||
separador = "&" if "?" in url else "?"
|
||||
@@ -38,37 +47,63 @@ def _normalizar_documento_e_telefones(dados: dict) -> None:
|
||||
"""
|
||||
Grava CNPJ/CPF e telefones sempre sem máscara (só dígitos/letras),
|
||||
igual ao que CLIENTE_CONTATO.TELEFONE já faz — evita que CLIENTE fique
|
||||
com dois formatos diferentes convivendo (achado da revisão final:
|
||||
docs/superpowers/plans/2026-08-13-painel-clientes-web.md). Só afeta
|
||||
escritas novas feitas por este painel; registros legados só passam a
|
||||
ficar sem máscara quando forem reeditados e salvos aqui.
|
||||
com dois formatos diferentes convivendo. Só afeta escritas novas feitas
|
||||
por este painel; registros legados só passam a ficar sem máscara quando
|
||||
forem reeditados e salvos aqui.
|
||||
"""
|
||||
dados["CNPJ"] = normalizar_documento(dados["CNPJ"])[0]
|
||||
dados["TELEFONE"] = normalizar_telefone(dados["TELEFONE"])
|
||||
dados["CELULAR"] = normalizar_telefone(dados["CELULAR"])
|
||||
|
||||
|
||||
def _aplicar_defaults_opcionais(dados: dict) -> None:
|
||||
"""
|
||||
IMPLANTACAO/MENSALIDADE/JUROS_MORA/JUROS_MULTA/DATA_CONTRATO/
|
||||
VENCIMENTO_CONTRATO não são mais exigidos no formulário — se vierem em
|
||||
branco, gravamos um valor neutro (0,00 / data de hoje) em vez de
|
||||
bloquear o cadastro. Quem cadastra pode ajustar depois.
|
||||
"""
|
||||
for campo in _CAMPOS_DECIMAIS_OPCIONAIS:
|
||||
if not dados.get(campo):
|
||||
dados[campo] = "0.00"
|
||||
hoje = datetime.date.today().isoformat()
|
||||
for campo in _CAMPOS_DATA_OPCIONAIS:
|
||||
if not dados.get(campo):
|
||||
dados[campo] = hoje
|
||||
|
||||
|
||||
def _campos_obrigatorios_faltando(dados: dict) -> list[str]:
|
||||
return [campo for campo in CAMPOS_OBRIGATORIOS if not dados.get(campo)]
|
||||
|
||||
|
||||
@router.get("", response_class=HTMLResponse)
|
||||
def listar(
|
||||
request: Request,
|
||||
busca: Optional[str] = None,
|
||||
ordenar_por: str = db_clientes.ORDENACAO_PADRAO,
|
||||
pagina: int = 1,
|
||||
mensagem: Optional[str] = None,
|
||||
mensagem_tipo: Optional[str] = None,
|
||||
usuario: str = Depends(exigir_login),
|
||||
):
|
||||
if ordenar_por not in db_clientes.ORDENACOES_VALIDAS:
|
||||
ordenar_por = db_clientes.ORDENACAO_PADRAO
|
||||
|
||||
pagina = max(pagina, 1)
|
||||
total = db_clientes.contar_clientes(busca=busca or None)
|
||||
total_paginas = max(math.ceil(total / TAMANHO_PAGINA), 1)
|
||||
pagina = min(pagina, total_paginas)
|
||||
offset = (pagina - 1) * TAMANHO_PAGINA
|
||||
|
||||
clientes = db_clientes.listar_clientes(busca=busca or None, limite=TAMANHO_PAGINA, offset=offset)
|
||||
clientes = db_clientes.listar_clientes(
|
||||
busca=busca or None, limite=TAMANHO_PAGINA, offset=offset, ordenar_por=ordenar_por
|
||||
)
|
||||
return templates.TemplateResponse(
|
||||
request,
|
||||
"clientes_lista.html",
|
||||
{
|
||||
"clientes": clientes, "busca": busca, "mensagem": mensagem, "mensagem_tipo": mensagem_tipo,
|
||||
"clientes": clientes, "busca": busca, "ordenar_por": ordenar_por,
|
||||
"mensagem": mensagem, "mensagem_tipo": mensagem_tipo,
|
||||
"pagina": pagina, "total_paginas": total_paginas, "total": total,
|
||||
},
|
||||
)
|
||||
@@ -76,12 +111,14 @@ def listar(
|
||||
|
||||
@router.get("/novo", response_class=HTMLResponse)
|
||||
def form_novo(request: Request, usuario: str = Depends(exigir_login)):
|
||||
dados = {c: "" for c in db_clientes.CAMPOS_CLIENTE}
|
||||
dados["CODIGO"] = db_clientes.proximo_codigo_sugerido()
|
||||
return templates.TemplateResponse(
|
||||
request,
|
||||
"cliente_form.html",
|
||||
{
|
||||
"cliente": None, "dados": {c: "" for c in db_clientes.CAMPOS_CLIENTE}, "campos": db_clientes.CAMPOS_CLIENTE,
|
||||
"csrf_token": token_do_contexto(request),
|
||||
"cliente": None, "dados": dados, "campos": db_clientes.CAMPOS_CLIENTE,
|
||||
"campos_obrigatorios": CAMPOS_OBRIGATORIOS, "csrf_token": token_do_contexto(request),
|
||||
},
|
||||
)
|
||||
|
||||
@@ -91,19 +128,23 @@ async def criar(request: Request, usuario: str = Depends(exigir_login)):
|
||||
form = dict((await request.form()))
|
||||
dados = _ler_form_cliente(form)
|
||||
_normalizar_documento_e_telefones(dados)
|
||||
confirmar_duplicado = form.get("confirmar_codigo_duplicado") == "1"
|
||||
confirmar_dv = form.get("confirmar_dv_invalido") == "1"
|
||||
|
||||
if not confirmar_duplicado and db_clientes.codigo_em_uso(dados["CODIGO"]):
|
||||
contexto_base = {
|
||||
"cliente": None, "dados": dados, "campos": db_clientes.CAMPOS_CLIENTE,
|
||||
"campos_obrigatorios": CAMPOS_OBRIGATORIOS, "csrf_token": token_do_contexto(request),
|
||||
}
|
||||
|
||||
faltando = _campos_obrigatorios_faltando(dados)
|
||||
if faltando:
|
||||
return templates.TemplateResponse(
|
||||
request,
|
||||
"cliente_form.html",
|
||||
{
|
||||
"cliente": None, "dados": dados, "campos": db_clientes.CAMPOS_CLIENTE,
|
||||
"aviso_codigo_duplicado": True,
|
||||
"confirmar_codigo_duplicado": confirmar_duplicado, "confirmar_dv_invalido": confirmar_dv,
|
||||
"csrf_token": token_do_contexto(request),
|
||||
},
|
||||
request, "cliente_form.html", {**contexto_base, "erro_campos_obrigatorios": faltando}
|
||||
)
|
||||
|
||||
if db_clientes.codigo_em_uso(dados["CODIGO"]):
|
||||
db_log.registrar(usuario, "criar", "cliente", dados["CODIGO"], False, "código já em uso")
|
||||
return templates.TemplateResponse(
|
||||
request, "cliente_form.html", {**contexto_base, "erro_codigo_duplicado": True}
|
||||
)
|
||||
|
||||
documento_normalizado, formato_valido, tipo = normalizar_documento(dados["CNPJ"])
|
||||
@@ -114,27 +155,18 @@ async def criar(request: Request, usuario: str = Depends(exigir_login)):
|
||||
)
|
||||
if dv_invalido and not confirmar_dv:
|
||||
return templates.TemplateResponse(
|
||||
request,
|
||||
"cliente_form.html",
|
||||
{
|
||||
"cliente": None, "dados": dados, "campos": db_clientes.CAMPOS_CLIENTE,
|
||||
"aviso_dv_invalido": True,
|
||||
"confirmar_codigo_duplicado": confirmar_duplicado, "confirmar_dv_invalido": confirmar_dv,
|
||||
"csrf_token": token_do_contexto(request),
|
||||
},
|
||||
request, "cliente_form.html", {**contexto_base, "aviso_dv_invalido": True}
|
||||
)
|
||||
|
||||
_aplicar_defaults_opcionais(dados)
|
||||
try:
|
||||
novo_id = db_clientes.criar_cliente(dados)
|
||||
except (pymysql.err.DataError, pymysql.err.IntegrityError) as exc:
|
||||
db_log.registrar(usuario, "criar", "cliente", dados["CODIGO"], False, str(exc))
|
||||
return templates.TemplateResponse(
|
||||
request,
|
||||
"cliente_form.html",
|
||||
{
|
||||
"cliente": None, "dados": dados, "campos": db_clientes.CAMPOS_CLIENTE, "erro_gravacao": str(exc),
|
||||
"csrf_token": token_do_contexto(request),
|
||||
},
|
||||
request, "cliente_form.html", {**contexto_base, "dados": dados, "erro_gravacao": str(exc)}
|
||||
)
|
||||
db_log.registrar(usuario, "criar", "cliente", dados["CODIGO"], True)
|
||||
return _redirect_com_mensagem(f"/clientes/{novo_id}", "Cliente criado.")
|
||||
|
||||
|
||||
@@ -173,7 +205,7 @@ def form_editar(request: Request, cliente_id: int, usuario: str = Depends(exigir
|
||||
"cliente_form.html",
|
||||
{
|
||||
"cliente": cliente, "dados": cliente, "campos": db_clientes.CAMPOS_CLIENTE,
|
||||
"csrf_token": token_do_contexto(request),
|
||||
"campos_obrigatorios": CAMPOS_OBRIGATORIOS, "csrf_token": token_do_contexto(request),
|
||||
},
|
||||
)
|
||||
|
||||
@@ -183,29 +215,33 @@ async def editar(request: Request, cliente_id: int, usuario: str = Depends(exigi
|
||||
form = dict((await request.form()))
|
||||
dados = _ler_form_cliente(form)
|
||||
_normalizar_documento_e_telefones(dados)
|
||||
confirmar_duplicado = form.get("confirmar_codigo_duplicado") == "1"
|
||||
|
||||
if not confirmar_duplicado and db_clientes.codigo_em_uso(dados["CODIGO"], ignorar_id=cliente_id):
|
||||
contexto_base = {
|
||||
"cliente": {"ID": cliente_id}, "dados": dados, "campos": db_clientes.CAMPOS_CLIENTE,
|
||||
"campos_obrigatorios": CAMPOS_OBRIGATORIOS, "csrf_token": token_do_contexto(request),
|
||||
}
|
||||
|
||||
faltando = _campos_obrigatorios_faltando(dados)
|
||||
if faltando:
|
||||
return templates.TemplateResponse(
|
||||
request,
|
||||
"cliente_form.html",
|
||||
{
|
||||
"cliente": {"ID": cliente_id}, "dados": dados, "campos": db_clientes.CAMPOS_CLIENTE,
|
||||
"aviso_codigo_duplicado": True, "csrf_token": token_do_contexto(request),
|
||||
},
|
||||
request, "cliente_form.html", {**contexto_base, "erro_campos_obrigatorios": faltando}
|
||||
)
|
||||
|
||||
if db_clientes.codigo_em_uso(dados["CODIGO"], ignorar_id=cliente_id):
|
||||
db_log.registrar(usuario, "editar", "cliente", str(cliente_id), False, "código já em uso")
|
||||
return templates.TemplateResponse(
|
||||
request, "cliente_form.html", {**contexto_base, "erro_codigo_duplicado": True}
|
||||
)
|
||||
|
||||
_aplicar_defaults_opcionais(dados)
|
||||
try:
|
||||
db_clientes.atualizar_cliente(cliente_id, dados)
|
||||
except (pymysql.err.DataError, pymysql.err.IntegrityError) as exc:
|
||||
db_log.registrar(usuario, "editar", "cliente", str(cliente_id), False, str(exc))
|
||||
return templates.TemplateResponse(
|
||||
request,
|
||||
"cliente_form.html",
|
||||
{
|
||||
"cliente": {"ID": cliente_id}, "dados": dados, "campos": db_clientes.CAMPOS_CLIENTE,
|
||||
"erro_gravacao": str(exc), "csrf_token": token_do_contexto(request),
|
||||
},
|
||||
request, "cliente_form.html", {**contexto_base, "dados": dados, "erro_gravacao": str(exc)}
|
||||
)
|
||||
db_log.registrar(usuario, "editar", "cliente", str(cliente_id), True)
|
||||
return _redirect_com_mensagem(f"/clientes/{cliente_id}", "Cliente atualizado.")
|
||||
|
||||
|
||||
@@ -214,5 +250,7 @@ def excluir(cliente_id: int, usuario: str = Depends(exigir_login)):
|
||||
try:
|
||||
db_clientes.excluir_cliente(cliente_id)
|
||||
except db_clientes.ClienteComDependenciasError as exc:
|
||||
db_log.registrar(usuario, "excluir", "cliente", str(cliente_id), False, str(exc))
|
||||
return _redirect_com_mensagem(f"/clientes/{cliente_id}", str(exc), "danger")
|
||||
db_log.registrar(usuario, "excluir", "cliente", str(cliente_id), True)
|
||||
return _redirect_com_mensagem("/clientes", "Cliente excluído.")
|
||||
|
||||
@@ -12,6 +12,7 @@ from ..auth import exigir_login
|
||||
from ..csrf import exigir_csrf
|
||||
from ..db import clientes as db_clientes
|
||||
from ..db import contatos as db_contatos
|
||||
from ..db import log as db_log
|
||||
from ..normalize import normalizar_telefone
|
||||
|
||||
router = APIRouter(prefix="/clientes")
|
||||
@@ -38,9 +39,11 @@ def adicionar(
|
||||
return _redirect_com_mensagem(cliente_id, "Telefone obrigatório.", "danger")
|
||||
|
||||
if db_contatos.contato_duplicado(cliente_id, telefone_normalizado):
|
||||
db_log.registrar(usuario, "adicionar", "contato", str(cliente_id), False, "telefone duplicado")
|
||||
return _redirect_com_mensagem(cliente_id, f"O telefone {telefone_normalizado} já está cadastrado para este cliente.", "warning")
|
||||
|
||||
db_contatos.inserir_contato(cliente_id, nome.strip() or None, telefone_normalizado, email.strip() or None)
|
||||
db_log.registrar(usuario, "adicionar", "contato", str(cliente_id), True)
|
||||
return _redirect_com_mensagem(cliente_id, "Contato adicionado.")
|
||||
|
||||
|
||||
@@ -58,13 +61,16 @@ def editar(
|
||||
return _redirect_com_mensagem(cliente_id, "Telefone obrigatório.", "danger")
|
||||
|
||||
if db_contatos.contato_duplicado(cliente_id, telefone_normalizado, ignorar_id=contato_id):
|
||||
db_log.registrar(usuario, "editar", "contato", str(contato_id), False, "telefone duplicado")
|
||||
return _redirect_com_mensagem(cliente_id, f"O telefone {telefone_normalizado} já está cadastrado para este cliente.", "warning")
|
||||
|
||||
db_contatos.atualizar_contato(contato_id, cliente_id, nome.strip() or None, telefone_normalizado, email.strip() or None)
|
||||
db_log.registrar(usuario, "editar", "contato", str(contato_id), True)
|
||||
return _redirect_com_mensagem(cliente_id, "Contato atualizado.")
|
||||
|
||||
|
||||
@router.post("/{cliente_id}/contatos/{contato_id}/excluir", dependencies=[Depends(exigir_csrf)])
|
||||
def excluir(cliente_id: int, contato_id: int, usuario: str = Depends(exigir_login)):
|
||||
db_contatos.excluir_contato(contato_id, cliente_id)
|
||||
db_log.registrar(usuario, "excluir", "contato", str(contato_id), True)
|
||||
return _redirect_com_mensagem(cliente_id, "Contato excluído.")
|
||||
|
||||
@@ -0,0 +1,26 @@
|
||||
"""
|
||||
Rota /log: histórico de ações do painel (LOG_PAINEL, app/db/log.py) e,
|
||||
quando configurado, os últimos envios automáticos de chave via WhatsApp
|
||||
(Data Table pagamentos_processados do n8n, app/n8n_client.py).
|
||||
"""
|
||||
from fastapi import APIRouter, Depends, Request
|
||||
from fastapi.responses import HTMLResponse
|
||||
from fastapi.templating import Jinja2Templates
|
||||
|
||||
from .. import n8n_client
|
||||
from ..auth import exigir_login
|
||||
from ..db import log as db_log
|
||||
|
||||
router = APIRouter()
|
||||
templates = Jinja2Templates(directory="app/templates")
|
||||
|
||||
|
||||
@router.get("/log", response_class=HTMLResponse)
|
||||
def pagina(request: Request, usuario: str = Depends(exigir_login)):
|
||||
acoes = db_log.listar(limite=100)
|
||||
envios = n8n_client.listar_envios(limite=50)
|
||||
return templates.TemplateResponse(
|
||||
request,
|
||||
"log.html",
|
||||
{"acoes": acoes, "envios": envios, "envios_configurado": n8n_client.configurado()},
|
||||
)
|
||||
@@ -26,6 +26,7 @@
|
||||
<nav>
|
||||
<a href="/clientes">Clientes</a>
|
||||
<a href="/gerador-chave">Gerador de chave</a>
|
||||
<a href="/log">Log</a>
|
||||
</nav>
|
||||
<main>
|
||||
{% if mensagem %}<div class="mensagem {{ mensagem_tipo or 'success' }}">{{ mensagem }}</div>{% endif %}
|
||||
|
||||
@@ -3,16 +3,9 @@
|
||||
{% block titulo %}{{ "Editar" if cliente else "Novo" }} cliente{% endblock %}
|
||||
{% block conteudo %}
|
||||
<h1>{{ "Editar" if cliente else "Novo" }} cliente</h1>
|
||||
{% if aviso_codigo_duplicado %}
|
||||
<div class="mensagem warning">
|
||||
O código {{ dados.CODIGO }} já está em uso por outro cliente.
|
||||
<form method="post">
|
||||
<input type="hidden" name="csrf_token" value="{{ csrf_token }}">
|
||||
{% for campo, valor in dados.items() %}<input type="hidden" name="{{ campo }}" value="{{ valor }}">{% endfor %}
|
||||
<input type="hidden" name="confirmar_codigo_duplicado" value="1">
|
||||
<input type="hidden" name="confirmar_dv_invalido" value="{{ '1' if (aviso_dv_invalido or confirmar_dv_invalido) else '' }}">
|
||||
<button type="submit">Cadastrar mesmo assim</button>
|
||||
</form>
|
||||
{% if erro_codigo_duplicado %}
|
||||
<div class="mensagem danger">
|
||||
O código {{ dados.CODIGO }} já está em uso por outro cliente — escolha outro código antes de salvar.
|
||||
</div>
|
||||
{% endif %}
|
||||
{% if aviso_dv_invalido %}
|
||||
@@ -22,7 +15,6 @@
|
||||
<input type="hidden" name="csrf_token" value="{{ csrf_token }}">
|
||||
{% for campo, valor in dados.items() %}<input type="hidden" name="{{ campo }}" value="{{ valor }}">{% endfor %}
|
||||
<input type="hidden" name="confirmar_dv_invalido" value="1">
|
||||
<input type="hidden" name="confirmar_codigo_duplicado" value="{{ '1' if (aviso_codigo_duplicado or confirmar_codigo_duplicado) else '' }}">
|
||||
<button type="submit">Cadastrar mesmo assim</button>
|
||||
</form>
|
||||
</div>
|
||||
@@ -35,15 +27,16 @@
|
||||
<form method="post">
|
||||
<input type="hidden" name="csrf_token" value="{{ csrf_token }}">
|
||||
{% for campo in campos %}
|
||||
<label for="{{ campo }}">{{ campo }}</label>
|
||||
<label for="{{ campo }}">{{ campo }}{% if campo in campos_obrigatorios %} *{% endif %}</label>
|
||||
{% if campo in ["DATA_CONTRATO", "VENCIMENTO_CONTRATO"] %}
|
||||
<input type="date" id="{{ campo }}" name="{{ campo }}" value="{{ dados[campo] if dados else '' }}" required>
|
||||
<input type="date" id="{{ campo }}" name="{{ campo }}" value="{{ dados[campo] if dados else '' }}" {% if campo in campos_obrigatorios %}required{% endif %}>
|
||||
{% elif campo in ["IMPLANTACAO", "MENSALIDADE", "JUROS_MORA", "JUROS_MULTA"] %}
|
||||
<input type="number" step="0.01" id="{{ campo }}" name="{{ campo }}" value="{{ dados[campo] if dados else '' }}" required>
|
||||
<input type="number" step="0.01" id="{{ campo }}" name="{{ campo }}" value="{{ dados[campo] if dados else '' }}" {% if campo in campos_obrigatorios %}required{% endif %}>
|
||||
{% else %}
|
||||
<input type="text" id="{{ campo }}" name="{{ campo }}" value="{{ dados[campo] if dados else '' }}" required>
|
||||
<input type="text" id="{{ campo }}" name="{{ campo }}" value="{{ dados[campo] if dados else '' }}" {% if campo in campos_obrigatorios %}required{% endif %}>
|
||||
{% endif %}
|
||||
{% endfor %}
|
||||
<p><small>* campos obrigatórios. Os demais podem ficar em branco — implantação/mensalidade/juros ficam 0,00 e as datas de contrato ficam com a data de hoje até serem ajustadas.</small></p>
|
||||
<button type="submit">Salvar</button>
|
||||
</form>
|
||||
{% endblock %}
|
||||
|
||||
@@ -5,6 +5,13 @@
|
||||
<h1>Clientes</h1>
|
||||
<form method="get" action="/clientes">
|
||||
<input type="text" name="busca" value="{{ busca or '' }}" placeholder="Buscar por código, CNPJ, razão ou fantasia">
|
||||
<label for="ordenar_por">Ordenar por</label>
|
||||
<select id="ordenar_por" name="ordenar_por">
|
||||
<option value="razao" {% if ordenar_por == "razao" %}selected{% endif %}>Razão social</option>
|
||||
<option value="codigo" {% if ordenar_por == "codigo" %}selected{% endif %}>Código</option>
|
||||
<option value="fantasia" {% if ordenar_por == "fantasia" %}selected{% endif %}>Fantasia</option>
|
||||
<option value="cnpj" {% if ordenar_por == "cnpj" %}selected{% endif %}>CNPJ</option>
|
||||
</select>
|
||||
<button type="submit">Buscar</button>
|
||||
</form>
|
||||
<p><a class="botao" href="/clientes/novo">+ Novo cliente</a></p>
|
||||
@@ -23,11 +30,11 @@
|
||||
{% if total_paginas and total_paginas > 1 %}
|
||||
<p>
|
||||
{% if pagina > 1 %}
|
||||
<a class="botao" href="/clientes?pagina={{ pagina - 1 }}{{ '&busca=' + busca if busca else '' }}">← Anterior</a>
|
||||
<a class="botao" href="/clientes?pagina={{ pagina - 1 }}{{ '&busca=' + busca if busca else '' }}&ordenar_por={{ ordenar_por }}">← Anterior</a>
|
||||
{% endif %}
|
||||
Página {{ pagina }} de {{ total_paginas }} ({{ total }} clientes)
|
||||
{% if pagina < total_paginas %}
|
||||
<a class="botao" href="/clientes?pagina={{ pagina + 1 }}{{ '&busca=' + busca if busca else '' }}">Próxima →</a>
|
||||
<a class="botao" href="/clientes?pagina={{ pagina + 1 }}{{ '&busca=' + busca if busca else '' }}&ordenar_por={{ ordenar_por }}">Próxima →</a>
|
||||
{% endif %}
|
||||
</p>
|
||||
{% endif %}
|
||||
|
||||
@@ -0,0 +1,55 @@
|
||||
<!-- app/templates/log.html -->
|
||||
{% extends "base.html" %}
|
||||
{% block titulo %}Log{% endblock %}
|
||||
{% block conteudo %}
|
||||
<h1>Log</h1>
|
||||
<p><a class="botao" href="/clientes">← Clientes</a></p>
|
||||
|
||||
<h2>Ações no painel</h2>
|
||||
{% if acoes %}
|
||||
<table>
|
||||
<tr><th>Quando</th><th>Usuário</th><th>Ação</th><th>Entidade</th><th>Status</th><th>Detalhe</th></tr>
|
||||
{% for a in acoes %}
|
||||
<tr>
|
||||
<td>{{ a.CRIADO_EM }}</td>
|
||||
<td>{{ a.USUARIO }}</td>
|
||||
<td>{{ a.ACAO }}</td>
|
||||
<td>{{ a.ENTIDADE }}{% if a.ENTIDADE_ID %} #{{ a.ENTIDADE_ID }}{% endif %}</td>
|
||||
<td>{{ "OK" if a.SUCESSO else "ERRO" }}</td>
|
||||
<td>{{ a.DETALHE or "" }}</td>
|
||||
</tr>
|
||||
{% endfor %}
|
||||
</table>
|
||||
{% else %}
|
||||
<p>Nenhuma ação registrada ainda.</p>
|
||||
{% endif %}
|
||||
|
||||
<h2>Envios automáticos de chave (n8n)</h2>
|
||||
{% if not envios_configurado %}
|
||||
<p>
|
||||
Integração com o n8n não configurada — defina <code>N8N_API_URL</code> e
|
||||
<code>N8N_API_KEY</code> no ambiente (ver <code>.env.example</code>) para ver
|
||||
aqui os envios automáticos de chave feitos pelo workflow do n8n.
|
||||
</p>
|
||||
{% elif not envios %}
|
||||
<p>Nenhum envio encontrado (ou não foi possível consultar o n8n agora).</p>
|
||||
{% else %}
|
||||
<table>
|
||||
<tr>
|
||||
<th>Data</th><th>Cliente</th><th>Código</th><th>Competência</th><th>Status</th>
|
||||
<th>Telefones enviados</th><th>Telefones com erro</th>
|
||||
</tr>
|
||||
{% for e in envios %}
|
||||
<tr>
|
||||
<td>{{ e.data_processamento }}</td>
|
||||
<td>{{ e.nome_cliente }}</td>
|
||||
<td>{{ e.codigo_cliente }}</td>
|
||||
<td>{{ e.competencia }}</td>
|
||||
<td>{{ e.status }}</td>
|
||||
<td>{{ e.telefones_enviados }}</td>
|
||||
<td>{{ e.telefones_com_erro or "" }}</td>
|
||||
</tr>
|
||||
{% endfor %}
|
||||
</table>
|
||||
{% endif %}
|
||||
{% endblock %}
|
||||
@@ -4,3 +4,4 @@ jinja2==3.1.4
|
||||
python-multipart==0.0.9
|
||||
PyMySQL==1.1.1
|
||||
python-dotenv==1.0.1
|
||||
httpx==0.27.2
|
||||
|
||||
@@ -87,6 +87,28 @@ def test_excluir_cliente_sem_dependencias(cliente_id):
|
||||
assert db_clientes.obter_cliente(cliente_id) is None
|
||||
|
||||
|
||||
def test_listar_clientes_ordena_por_codigo(cliente_id):
|
||||
resultado = db_clientes.listar_clientes(busca="TESTE PAINEL RAZAO", 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_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_excluir_cliente_com_contato_bloqueia(cliente_id):
|
||||
conn = get_connection()
|
||||
try:
|
||||
|
||||
@@ -0,0 +1,42 @@
|
||||
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
|
||||
@@ -0,0 +1,56 @@
|
||||
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
|
||||
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"
|
||||
+142
-99
@@ -3,6 +3,7 @@ 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
|
||||
|
||||
@@ -21,11 +22,42 @@ 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)
|
||||
|
||||
|
||||
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",
|
||||
"csrf_token": CSRF_TESTE,
|
||||
}
|
||||
)
|
||||
return dados
|
||||
|
||||
|
||||
def test_listar_clientes(monkeypatch):
|
||||
monkeypatch.setattr(
|
||||
router_clientes.db_clientes,
|
||||
"listar_clientes",
|
||||
lambda busca=None, limite=100, offset=0: [
|
||||
lambda busca=None, limite=100, offset=0, ordenar_por="razao": [
|
||||
{"ID": 1, "CODIGO": "000001", "RAZAO": "Cliente Um", "FANTASIA": "Um", "CNPJ": "11222333000181"}
|
||||
],
|
||||
)
|
||||
@@ -39,7 +71,7 @@ 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: [
|
||||
lambda busca=None, limite=100, offset=0, ordenar_por="razao": [
|
||||
{"ID": i, "CODIGO": f"{i:06d}", "RAZAO": f"Cliente {i}", "FANTASIA": f"C{i}", "CNPJ": "11222333000181"}
|
||||
for i in range(1, 6)
|
||||
],
|
||||
@@ -54,6 +86,30 @@ def test_listar_clientes_pagina_com_controles_de_paginacao(monkeypatch):
|
||||
assert "Próxima" 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"):
|
||||
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: 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_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_detalhe_cliente_mostra_formulario_de_edicao_de_contato(monkeypatch):
|
||||
monkeypatch.setattr(
|
||||
router_clientes.db_clientes,
|
||||
@@ -74,39 +130,71 @@ def test_detalhe_cliente_mostra_formulario_de_edicao_de_contato(monkeypatch):
|
||||
assert 'value="75999090001"' in resposta.text
|
||||
|
||||
|
||||
def test_criar_cliente_codigo_duplicado_exige_confirmacao(monkeypatch):
|
||||
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 = {campo: "000001" if campo == "CODIGO" else "x" for campo in router_clientes.db_clientes.CAMPOS_CLIENTE}
|
||||
dados_form.update(
|
||||
{
|
||||
"IMPLANTACAO": "0",
|
||||
"MENSALIDADE": "100",
|
||||
"JUROS_MORA": "0",
|
||||
"JUROS_MULTA": "0",
|
||||
"DATA_CONTRATO": "2026-01-01",
|
||||
"VENCIMENTO_CONTRATO": "2027-01-01",
|
||||
"DIA_VENCIMENTO": "10",
|
||||
"CNPJ": "11222333000181",
|
||||
"csrf_token": CSRF_TESTE,
|
||||
}
|
||||
)
|
||||
dados_form = _form_valido()
|
||||
|
||||
# Sem "confirmar_codigo_duplicado": não deve criar, deve pedir confirmação.
|
||||
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
|
||||
|
||||
# Com confirmação: cria.
|
||||
dados_form["confirmar_codigo_duplicado"] = "1"
|
||||
# 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)
|
||||
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)
|
||||
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
|
||||
assert criado.get("ok") is True
|
||||
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):
|
||||
@@ -116,22 +204,12 @@ def test_criar_cliente_normaliza_cnpj_e_telefones(monkeypatch):
|
||||
router_clientes.db_clientes, "criar_cliente", lambda dados: capturado.setdefault("dados", dados) and 42
|
||||
)
|
||||
|
||||
dados_form = {campo: "x" for campo in router_clientes.db_clientes.CAMPOS_CLIENTE}
|
||||
dados_form = _form_valido()
|
||||
dados_form.update(
|
||||
{
|
||||
"CODIGO": "000001",
|
||||
"CNPJ": "11.222.333/0001-81",
|
||||
"TELEFONE": "(75) 3333-4444",
|
||||
"CELULAR": "(75) 99909-1780",
|
||||
"IMPLANTACAO": "0",
|
||||
"MENSALIDADE": "100",
|
||||
"JUROS_MORA": "0",
|
||||
"JUROS_MULTA": "0",
|
||||
"DATA_CONTRATO": "2026-01-01",
|
||||
"VENCIMENTO_CONTRATO": "2027-01-01",
|
||||
"DIA_VENCIMENTO": "10",
|
||||
"confirmar_dv_invalido": "1",
|
||||
"csrf_token": CSRF_TESTE,
|
||||
}
|
||||
)
|
||||
|
||||
@@ -151,21 +229,12 @@ def test_editar_cliente_normaliza_cnpj_e_telefones(monkeypatch):
|
||||
lambda cliente_id, dados: capturado.setdefault("dados", dados),
|
||||
)
|
||||
|
||||
dados_form = {campo: "x" for campo in router_clientes.db_clientes.CAMPOS_CLIENTE}
|
||||
dados_form = _form_valido()
|
||||
dados_form.update(
|
||||
{
|
||||
"CODIGO": "000001",
|
||||
"CNPJ": "111.444.777-35",
|
||||
"TELEFONE": "(75) 3333-4444",
|
||||
"CELULAR": "(75) 99909-1780",
|
||||
"IMPLANTACAO": "0",
|
||||
"MENSALIDADE": "100",
|
||||
"JUROS_MORA": "0",
|
||||
"JUROS_MULTA": "0",
|
||||
"DATA_CONTRATO": "2026-01-01",
|
||||
"VENCIMENTO_CONTRATO": "2027-01-01",
|
||||
"DIA_VENCIMENTO": "10",
|
||||
"csrf_token": CSRF_TESTE,
|
||||
}
|
||||
)
|
||||
|
||||
@@ -176,6 +245,22 @@ def test_editar_cliente_normaliza_cnpj_e_telefones(monkeypatch):
|
||||
assert capturado["dados"]["CELULAR"] == "75999091780"
|
||||
|
||||
|
||||
def test_editar_cliente_codigo_duplicado_bloqueia_sem_bypass(monkeypatch):
|
||||
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_excluir_cliente_com_dependencia_mostra_erro(monkeypatch):
|
||||
monkeypatch.setattr(
|
||||
router_clientes.db_clientes,
|
||||
@@ -204,83 +289,41 @@ def _limpar_codigo(codigo: str) -> None:
|
||||
conn.close()
|
||||
|
||||
|
||||
def test_criar_cliente_com_data_e_decimais_vazios_nao_gera_500():
|
||||
def test_criar_cliente_com_data_e_decimais_vazios_cria_com_defaults():
|
||||
"""
|
||||
Integração com o banco real de dev: DATA_CONTRATO/decimais são NOT NULL
|
||||
e o dev roda em STRICT_TRANS_TABLES — submeter strings vazias deve ser
|
||||
capturado pelo router (pymysql.err.DataError) e reexibir o formulário
|
||||
com erro_gravacao, nunca estourar como 500.
|
||||
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 = {campo: "x" for campo in router_clientes.db_clientes.CAMPOS_CLIENTE}
|
||||
dados_form.update(
|
||||
{
|
||||
"CODIGO": codigo,
|
||||
"CNPJ": "11222333000181",
|
||||
"IMPLANTACAO": "",
|
||||
"MENSALIDADE": "",
|
||||
"JUROS_MORA": "",
|
||||
"JUROS_MULTA": "",
|
||||
"DATA_CONTRATO": "",
|
||||
"VENCIMENTO_CONTRATO": "",
|
||||
"DIA_VENCIMENTO": "10",
|
||||
"csrf_token": CSRF_TESTE,
|
||||
}
|
||||
)
|
||||
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 == 200
|
||||
assert "Não foi possível salvar" in resposta.text
|
||||
assert resposta.status_code == 303
|
||||
finally:
|
||||
_limpar_codigo(codigo)
|
||||
|
||||
|
||||
def test_criar_cliente_com_ambos_avisos_confirma_ambos_de_uma_vez(monkeypatch):
|
||||
"""
|
||||
Reproduz Fix 3: um cliente com CODIGO duplicado E DV de documento
|
||||
inválido não pode ficar preso em loop entre os dois avisos — confirmar
|
||||
os dois de uma vez (hidden fields carregados de um aviso para o outro)
|
||||
deve efetivamente criar o cliente.
|
||||
"""
|
||||
monkeypatch.setattr(router_clientes.db_clientes, "codigo_em_uso", lambda codigo, ignorar_id=None: True)
|
||||
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)
|
||||
criado = {}
|
||||
monkeypatch.setattr(
|
||||
router_clientes.db_clientes, "criar_cliente", lambda dados: criado.setdefault("ok", True) and 42
|
||||
)
|
||||
|
||||
dados_form = {campo: "000002" if campo == "CODIGO" else "x" for campo in router_clientes.db_clientes.CAMPOS_CLIENTE}
|
||||
dados_form.update(
|
||||
{
|
||||
"IMPLANTACAO": "0",
|
||||
"MENSALIDADE": "100",
|
||||
"JUROS_MORA": "0",
|
||||
"JUROS_MULTA": "0",
|
||||
"DATA_CONTRATO": "2026-01-01",
|
||||
"VENCIMENTO_CONTRATO": "2027-01-01",
|
||||
"DIA_VENCIMENTO": "10",
|
||||
"CNPJ": "11144477734", # CPF com DV inválido (1 dígito diferente do válido "...735")
|
||||
"csrf_token": CSRF_TESTE,
|
||||
}
|
||||
)
|
||||
dados_form = _form_valido()
|
||||
dados_form["CNPJ"] = "11144477734" # CPF com DV inválido (1 dígito diferente do válido "...735")
|
||||
|
||||
# Round 1: nem codigo nem dv confirmados -> deve pedir confirmação de código duplicado.
|
||||
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
|
||||
|
||||
# Round 2: confirma o código duplicado -> agora deve pedir confirmação do DV inválido.
|
||||
dados_form["confirmar_codigo_duplicado"] = "1"
|
||||
# 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
|
||||
# A raiz do Fix 3: o form do aviso de DV precisa carregar
|
||||
# confirmar_codigo_duplicado=1 adiante (senão confirmar só o DV a seguir
|
||||
# reabriria o aviso de código duplicado, em loop infinito).
|
||||
assert 'name="confirmar_codigo_duplicado" value="1"' in resposta.text
|
||||
|
||||
# Round 3: confirma ambos de uma vez -> deve criar (303), sem cair em outro loop.
|
||||
# Confirmando: cria.
|
||||
dados_form["confirmar_dv_invalido"] = "1"
|
||||
resposta = client.post("/clientes/novo", data=dados_form)
|
||||
assert resposta.status_code == 303
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import pytest
|
||||
from fastapi import FastAPI
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
@@ -17,6 +18,13 @@ 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)
|
||||
|
||||
@@ -0,0 +1,52 @@
|
||||
from fastapi import FastAPI
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from app.auth import 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: "teste"
|
||||
client = TestClient(app_teste, follow_redirects=False)
|
||||
|
||||
|
||||
def test_pagina_log_mostra_acoes_do_painel(monkeypatch):
|
||||
monkeypatch.setattr(
|
||||
router_log.db_log,
|
||||
"listar",
|
||||
lambda limite=100: [
|
||||
{
|
||||
"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
|
||||
|
||||
|
||||
def test_pagina_log_mostra_envios_quando_configurado(monkeypatch):
|
||||
monkeypatch.setattr(router_log.db_log, "listar", lambda limite=100: [])
|
||||
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
|
||||
Reference in New Issue
Block a user