diff --git a/app/db/chaves.py b/app/db/chaves.py new file mode 100644 index 0000000..6577304 --- /dev/null +++ b/app/db/chaves.py @@ -0,0 +1,31 @@ +""" +Leitura somente-leitura de Clientes_Chaves (tabela legada — ver +licenciamento-n8n/docs/MAPEAMENTO_BANCO.md seção 3). O painel NUNCA escreve +aqui; só mostra status para conferência. +""" +from .connection import get_connection + +MESES = [f"M{i}" for i in range(1, 13)] + + +def status_chaves(codigo: str) -> dict | None: + conn = get_connection() + try: + with conn.cursor() as cur: + colunas = ", ".join(MESES) + cur.execute( + f"SELECT ID, Cliente, {colunas} FROM Clientes_Chaves WHERE ID = %s", + (codigo,), + ) + linha = cur.fetchone() + finally: + conn.close() + + if linha is None: + return None + + return { + "codigo": linha["ID"], + "cliente": linha["Cliente"], + "meses": {mes: (linha[mes] or None) for mes in MESES}, + } diff --git a/tests/test_db_chaves.py b/tests/test_db_chaves.py new file mode 100644 index 0000000..4a455a1 --- /dev/null +++ b/tests/test_db_chaves.py @@ -0,0 +1,55 @@ +""" +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 + 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