From 2ecf86667883af1fcce75fe055725cdabdd77779 Mon Sep 17 00:00:00 2001 From: Gustavo Date: Thu, 13 Aug 2026 09:47:52 -0300 Subject: [PATCH] feat: CRUD de CLIENTE --- app/db/clientes.py | 104 ++++++++++++++++++++++++++++++++++++ tests/test_db_clientes.py | 109 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 213 insertions(+) create mode 100644 app/db/clientes.py create mode 100644 tests/test_db_clientes.py diff --git a/app/db/clientes.py b/app/db/clientes.py new file mode 100644 index 0000000..c291904 --- /dev/null +++ b/app/db/clientes.py @@ -0,0 +1,104 @@ +""" +CRUD completo de CLIENTE. Única tabela cadastral (ID autoincrement é a PK +técnica; CODIGO é o identificador de negócio usado pelo resto do sistema — +ver licenciamento-n8n/docs/MAPEAMENTO_BANCO.md). +""" +import pymysql + +from .connection import get_connection + +CAMPOS_CLIENTE = [ + "CODIGO", "CNPJ", "RAZAO", "FANTASIA", "IE", "ENDERECO", "CIDADE", + "BAIRRO", "UF", "CONTATO", "TELEFONE", "CELULAR", "CEP", "IMPLANTACAO", + "MENSALIDADE", "JUROS_MORA", "JUROS_MULTA", "DATA_CONTRATO", "EMAIL", + "OBS", "VENCIMENTO_CONTRATO", "DIA_VENCIMENTO", +] + +_COLUNAS_SELECT = ", ".join(["ID", *CAMPOS_CLIENTE]) + + +class ClienteComDependenciasError(Exception): + """Levantada ao tentar excluir um CLIENTE que ainda tem CLIENTE_CONTATO.""" + + +def listar_clientes(busca: str | None = None, limite: int = 100) -> list[dict]: + conn = get_connection() + try: + with conn.cursor() as cur: + if busca: + 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" + ) + termo = f"%{busca}%" + cur.execute(sql, (termo, termo, termo, termo, limite)) + else: + cur.execute(f"SELECT {_COLUNAS_SELECT} FROM CLIENTE ORDER BY RAZAO LIMIT %s", (limite,)) + return cur.fetchall() + finally: + conn.close() + + +def obter_cliente(cliente_id: int) -> dict | None: + conn = get_connection() + try: + with conn.cursor() as cur: + cur.execute(f"SELECT {_COLUNAS_SELECT} FROM CLIENTE WHERE ID = %s", (cliente_id,)) + return cur.fetchone() + finally: + conn.close() + + +def codigo_em_uso(codigo: str, ignorar_id: int | None = None) -> bool: + conn = get_connection() + try: + with conn.cursor() as cur: + sql = "SELECT ID FROM CLIENTE WHERE CODIGO = %s" + params = [codigo] + if ignorar_id is not None: + sql += " AND ID != %s" + params.append(ignorar_id) + cur.execute(sql, params) + return cur.fetchone() is not None + finally: + conn.close() + + +def criar_cliente(dados: dict) -> int: + colunas = ", ".join(CAMPOS_CLIENTE) + marcadores = ", ".join(["%s"] * len(CAMPOS_CLIENTE)) + valores = [dados[campo] for campo in CAMPOS_CLIENTE] + conn = get_connection() + try: + with conn.cursor() as cur: + cur.execute(f"INSERT INTO CLIENTE ({colunas}) VALUES ({marcadores})", valores) + return cur.lastrowid + finally: + conn.close() + + +def atualizar_cliente(cliente_id: int, dados: dict) -> None: + atribuicoes = ", ".join(f"{campo} = %s" for campo in CAMPOS_CLIENTE) + valores = [dados[campo] for campo in CAMPOS_CLIENTE] + valores.append(cliente_id) + conn = get_connection() + try: + with conn.cursor() as cur: + cur.execute(f"UPDATE CLIENTE SET {atribuicoes} WHERE ID = %s", valores) + finally: + conn.close() + + +def excluir_cliente(cliente_id: int) -> None: + conn = get_connection() + try: + with conn.cursor() as cur: + try: + cur.execute("DELETE FROM CLIENTE WHERE ID = %s", (cliente_id,)) + except pymysql.err.IntegrityError as exc: + raise ClienteComDependenciasError( + "Este cliente tem contatos cadastrados (CLIENTE_CONTATO) — remova-os antes de excluir." + ) from exc + finally: + conn.close() diff --git a/tests/test_db_clientes.py b/tests/test_db_clientes.py new file mode 100644 index 0000000..456178e --- /dev/null +++ b/tests/test_db_clientes.py @@ -0,0 +1,109 @@ +""" +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", +} + + +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_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): + db_clientes.excluir_cliente(cliente_id) + assert db_clientes.obter_cliente(cliente_id) is None + + +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() + + with pytest.raises(db_clientes.ClienteComDependenciasError): + db_clientes.excluir_cliente(cliente_id) + + conn = get_connection() + try: + with conn.cursor() as cur: + cur.execute("DELETE FROM CLIENTE_CONTATO WHERE CLIENTE_ID = %s", (cliente_id,)) + finally: + conn.close()