feat: leitura somente-leitura de status em Clientes_Chaves
This commit is contained in:
@@ -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},
|
||||||
|
}
|
||||||
@@ -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
|
||||||
Reference in New Issue
Block a user