feat: CRUD de CLIENTE

2026-08-13 09:47:52 -03:00
parent 880329bb8e
commit 2ecf866678
2 changed files with 213 additions and 0 deletions
@@ -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()
@@ -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()