feat: CRUD de CLIENTE_CONTATO

app/db/contatos.py

@@ -0,0 +1,75 @@
+"""
+CRUD de CLIENTE_CONTATO — relacionado a CLIENTE por CLIENTE_ID (FK real,
+ON DELETE RESTRICT ON UPDATE CASCADE). Substitui CLIENTE_CONTATO_CHAVE
+(superseded — ver licenciamento-n8n/docs/CANDIDATO_CLIENTE_CONTATO.md).
+"""
+from .connection import get_connection
+
+
+def listar_contatos_por_cliente(cliente_id: int) -> list[dict]:
+    conn = get_connection()
+    try:
+        with conn.cursor() as cur:
+            cur.execute(
+                "SELECT ID, NOME, TELEFONE, EMAIL FROM CLIENTE_CONTATO "
+                "WHERE CLIENTE_ID = %s ORDER BY ID",
+                (cliente_id,),
+            )
+            return list(cur.fetchall())
+    finally:
+        conn.close()
+
+
+def contato_duplicado(cliente_id: int, telefone: str, ignorar_id: int | None = None) -> bool:
+    conn = get_connection()
+    try:
+        with conn.cursor() as cur:
+            sql = "SELECT ID FROM CLIENTE_CONTATO WHERE CLIENTE_ID = %s AND TELEFONE = %s"
+            params = [cliente_id, telefone]
+            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 inserir_contato(cliente_id: int, nome: str | None, telefone: str, email: str | None = None) -> int:
+    conn = get_connection()
+    try:
+        with conn.cursor() as cur:
+            cur.execute(
+                "INSERT INTO CLIENTE_CONTATO (CLIENTE_ID, NOME, TELEFONE, EMAIL) VALUES (%s, %s, %s, %s)",
+                (cliente_id, nome, telefone, email),
+            )
+            return cur.lastrowid
+    finally:
+        conn.close()
+
+
+def atualizar_contato(
+    contato_id: int, cliente_id: int, nome: str | None, telefone: str, email: str | None = None
+) -> None:
+    conn = get_connection()
+    try:
+        with conn.cursor() as cur:
+            cur.execute(
+                "UPDATE CLIENTE_CONTATO SET NOME = %s, TELEFONE = %s, EMAIL = %s "
+                "WHERE ID = %s AND CLIENTE_ID = %s",
+                (nome, telefone, email, contato_id, cliente_id),
+            )
+    finally:
+        conn.close()
+
+
+def excluir_contato(contato_id: int, cliente_id: int) -> None:
+    conn = get_connection()
+    try:
+        with conn.cursor() as cur:
+            cur.execute(
+                "DELETE FROM CLIENTE_CONTATO WHERE ID = %s AND CLIENTE_ID = %s",
+                (contato_id, cliente_id),
+            )
+    finally:
+        conn.close()

tests/test_db_contatos.py

@@ -0,0 +1,59 @@
+"""
+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")
+    db_contatos.atualizar_contato(contato_id, cliente_id, "Fulano Editado", "75999090003")
+    contatos = db_contatos.listar_contatos_por_cliente(cliente_id)
+    assert contatos[0]["NOME"] == "Fulano Editado"
+    assert contatos[0]["TELEFONE"] == "75999090003"
+
+
+def test_excluir_contato(cliente_id):
+    contato_id = db_contatos.inserir_contato(cliente_id, "Fulano", "75999090001")
+    db_contatos.excluir_contato(contato_id, cliente_id)
+    assert db_contatos.listar_contatos_por_cliente(cliente_id) == []