From f940ff576820d2db3317640df2ef8c376d7cf50e Mon Sep 17 00:00:00 2001 From: Gustavo Date: Wed, 19 Aug 2026 09:13:43 -0300 Subject: [PATCH] feat: bloqueia CNPJ/CPF duplicado ao cadastrar ou editar cliente MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Mesmo padrão já usado pro CODIGO (db_clientes.codigo_em_uso): nova db_clientes.cnpj_em_uso(documento, ignorar_id=None) checada em criar()/editar() antes de gravar. A comparação ignora pontuação dos dois lados (não só normaliza o valor recebido) porque CLIENTE.CNPJ tem formatação inconsistente em registros legados (alguns com máscara, outros sem) — sem isso, um cadastro novo com o mesmo documento de um legado ainda mascarado passaria batido. app/templates/cliente_form.html: nova mensagem de erro 'erro_cnpj_duplicado', mesmo estilo visual do erro de código duplicado. 6 testes novos (unitário do db + integração do router pros dois fluxos), suíte completa (196 testes) passando. Co-Authored-By: Claude Sonnet 5 --- app/db/clientes.py | 30 +++++++++++++++++++++++ app/routers/clientes.py | 12 ++++++++++ app/templates/cliente_form.html | 5 ++++ tests/test_db_clientes.py | 20 ++++++++++++++++ tests/test_router_clientes.py | 42 +++++++++++++++++++++++++++++++++ 5 files changed, 109 insertions(+) diff --git a/app/db/clientes.py b/app/db/clientes.py index 2fe7af8..139e221 100644 --- a/app/db/clientes.py +++ b/app/db/clientes.py @@ -115,6 +115,36 @@ def obter_cliente(cliente_id: int) -> dict | None: conn.close() +def cnpj_em_uso(documento_normalizado: str, ignorar_id: int | None = None) -> bool: + """ + True se já existe outro CLIENTE com esse mesmo CNPJ/CPF (o mesmo campo + CLIENTE.CNPJ aceita os dois — ver app/normalize.py). Compara ignorando + pontuação em vez de igualdade exata porque CLIENTE.CNPJ tem + formatação inconsistente em registros legados (alguns com máscara, + outros sem — ver licenciamento-n8n/docs/ANALISE_CLIENTE_IPREAL.md); + sem normalizar dos dois lados, um cadastro novo (sempre gravado sem + máscara) não bateria com um legado que ainda tem pontos/traços, + deixando passar o duplicado. `documento_normalizado` já chega sem + máscara (normalizar_documento); a normalização abaixo só precisa + tratar o lado do banco. + """ + conn = get_connection() + try: + with conn.cursor() as cur: + sql = ( + "SELECT ID FROM CLIENTE WHERE " + "UPPER(REPLACE(REPLACE(REPLACE(REPLACE(CNPJ, '.', ''), '/', ''), '-', ''), ' ', '')) = %s" + ) + params = [documento_normalizado] + 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 codigo_em_uso(codigo: str, ignorar_id: int | None = None) -> bool: conn = get_connection() try: diff --git a/app/routers/clientes.py b/app/routers/clientes.py index aae8757..d445706 100644 --- a/app/routers/clientes.py +++ b/app/routers/clientes.py @@ -183,6 +183,12 @@ async def criar(request: Request, usuario: str = Depends(exigir_login)): request, "cliente_form.html", {**contexto_base, "erro_codigo_duplicado": True} ) + if dados["CNPJ"] and db_clientes.cnpj_em_uso(dados["CNPJ"]): + db_log.registrar(usuario, "criar", "cliente", dados["CODIGO"], False, "CNPJ/CPF já cadastrado") + return templates.TemplateResponse( + request, "cliente_form.html", {**contexto_base, "erro_cnpj_duplicado": True} + ) + extra_erro = _validar_documento(dados, confirmar_dv) if extra_erro: return templates.TemplateResponse(request, "cliente_form.html", {**contexto_base, **extra_erro}) @@ -290,6 +296,12 @@ async def editar(request: Request, cliente_id: int, usuario: str = Depends(exigi request, "cliente_form.html", {**contexto_base, "erro_codigo_duplicado": True} ) + if dados["CNPJ"] and db_clientes.cnpj_em_uso(dados["CNPJ"], ignorar_id=cliente_id): + db_log.registrar(usuario, "editar", "cliente", str(cliente_id), False, "CNPJ/CPF já cadastrado") + return templates.TemplateResponse( + request, "cliente_form.html", {**contexto_base, "erro_cnpj_duplicado": True} + ) + extra_erro = _validar_documento(dados, confirmar_dv) if extra_erro: return templates.TemplateResponse(request, "cliente_form.html", {**contexto_base, **extra_erro}) diff --git a/app/templates/cliente_form.html b/app/templates/cliente_form.html index e291d0f..35d3bdd 100644 --- a/app/templates/cliente_form.html +++ b/app/templates/cliente_form.html @@ -8,6 +8,11 @@ O código {{ dados.CODIGO }} já está em uso por outro cliente — escolha outro código antes de salvar. {% endif %} + {% if erro_cnpj_duplicado %} +
+ O CNPJ/CPF {{ dados.CNPJ }} já está cadastrado para outro cliente. +
+ {% endif %} {% if aviso_dv_invalido %}
O dígito verificador do documento ({{ dados.CNPJ }}) não confere — confirme se digitou certo antes de salvar. diff --git a/tests/test_db_clientes.py b/tests/test_db_clientes.py index 26ce736..b042b79 100644 --- a/tests/test_db_clientes.py +++ b/tests/test_db_clientes.py @@ -74,6 +74,26 @@ def test_codigo_em_uso(cliente_id): assert db_clientes.codigo_em_uso("000000") is False +def test_cnpj_em_uso(cliente_id): + assert db_clientes.cnpj_em_uso("11222333000181") is True + assert db_clientes.cnpj_em_uso("11222333000181", ignorar_id=cliente_id) is False + assert db_clientes.cnpj_em_uso("99988877000166") is False + + +def test_cnpj_em_uso_ignora_pontuacao_de_registros_legados(cliente_id): + # CLIENTE.CNPJ tem formatação inconsistente em registros legados (ver + # licenciamento-n8n/docs/ANALISE_CLIENTE_IPREAL.md) — cnpj_em_uso + # precisa achar o duplicado mesmo se o legado ainda tem máscara. + conn = get_connection() + try: + with conn.cursor() as cur: + cur.execute("UPDATE CLIENTE SET CNPJ = %s WHERE ID = %s", ("11.222.333/0001-81", cliente_id)) + finally: + conn.close() + + assert db_clientes.cnpj_em_uso("11222333000181") is True + + def test_atualizar_cliente(cliente_id): dados = dict(CLIENTE_TESTE) dados["FANTASIA"] = "TESTE PAINEL FANTASIA EDITADA" diff --git a/tests/test_router_clientes.py b/tests/test_router_clientes.py index 68069ae..9b008d2 100644 --- a/tests/test_router_clientes.py +++ b/tests/test_router_clientes.py @@ -253,6 +253,7 @@ def test_criar_cliente_codigo_duplicado_bloqueia_sem_bypass(monkeypatch): def test_criar_cliente_campo_obrigatorio_faltando_bloqueia(monkeypatch): monkeypatch.setattr(router_clientes.db_clientes, "codigo_em_uso", lambda codigo, ignorar_id=None: False) + monkeypatch.setattr(router_clientes.db_clientes, "cnpj_em_uso", lambda documento, ignorar_id=None: False) criado = {} monkeypatch.setattr( router_clientes.db_clientes, "criar_cliente", lambda dados: criado.setdefault("ok", True) and 42 @@ -276,6 +277,7 @@ def test_criar_cliente_campos_administrativos_em_branco_usa_defaults(monkeypatch import datetime monkeypatch.setattr(router_clientes.db_clientes, "codigo_em_uso", lambda codigo, ignorar_id=None: False) + monkeypatch.setattr(router_clientes.db_clientes, "cnpj_em_uso", lambda documento, ignorar_id=None: False) capturado = {} monkeypatch.setattr( router_clientes.db_clientes, "criar_cliente", lambda dados: capturado.setdefault("dados", dados) and 42 @@ -299,6 +301,7 @@ def test_criar_cliente_campos_administrativos_em_branco_usa_defaults(monkeypatch def test_criar_cliente_normaliza_cnpj_e_telefones(monkeypatch): monkeypatch.setattr(router_clientes.db_clientes, "codigo_em_uso", lambda codigo, ignorar_id=None: False) + monkeypatch.setattr(router_clientes.db_clientes, "cnpj_em_uso", lambda documento, ignorar_id=None: False) capturado = {} monkeypatch.setattr( router_clientes.db_clientes, "criar_cliente", lambda dados: capturado.setdefault("dados", dados) and 42 @@ -323,6 +326,7 @@ def test_criar_cliente_normaliza_cnpj_e_telefones(monkeypatch): def test_editar_cliente_normaliza_cnpj_e_telefones(monkeypatch): monkeypatch.setattr(router_clientes.db_clientes, "obter_cliente", lambda cliente_id: {"ID": cliente_id}) monkeypatch.setattr(router_clientes.db_clientes, "codigo_em_uso", lambda codigo, ignorar_id=None: False) + monkeypatch.setattr(router_clientes.db_clientes, "cnpj_em_uso", lambda documento, ignorar_id=None: False) capturado = {} monkeypatch.setattr( router_clientes.db_clientes, @@ -363,6 +367,37 @@ def test_editar_cliente_codigo_duplicado_bloqueia_sem_bypass(monkeypatch): assert "ok" not in atualizado +def test_criar_cliente_cnpj_duplicado_bloqueia(monkeypatch): + monkeypatch.setattr(router_clientes.db_clientes, "codigo_em_uso", lambda codigo, ignorar_id=None: False) + monkeypatch.setattr(router_clientes.db_clientes, "cnpj_em_uso", lambda documento, ignorar_id=None: True) + criado = {} + monkeypatch.setattr( + router_clientes.db_clientes, "criar_cliente", lambda dados: criado.setdefault("ok", True) and 42 + ) + + resposta = client.post("/clientes/novo", data=_form_valido()) + assert resposta.status_code == 200 + assert "já está cadastrado" in resposta.text + assert "ok" not in criado + + +def test_editar_cliente_cnpj_duplicado_bloqueia(monkeypatch): + monkeypatch.setattr(router_clientes.db_clientes, "obter_cliente", lambda cliente_id: {"ID": cliente_id}) + monkeypatch.setattr(router_clientes.db_clientes, "codigo_em_uso", lambda codigo, ignorar_id=None: False) + monkeypatch.setattr(router_clientes.db_clientes, "cnpj_em_uso", lambda documento, ignorar_id=None: True) + atualizado = {} + monkeypatch.setattr( + router_clientes.db_clientes, + "atualizar_cliente", + lambda cliente_id, dados: atualizado.setdefault("ok", True), + ) + + resposta = client.post("/clientes/1/editar", data=_form_valido()) + assert resposta.status_code == 200 + assert "já está cadastrado" in resposta.text + assert "ok" not in atualizado + + def test_editar_cliente_inexistente_nao_finge_sucesso(monkeypatch): # Sem a checagem de existência, isso caía direto no UPDATE (0 linhas # afetadas, sem erro do MySQL) e ainda registrava "sucesso" no log. @@ -437,6 +472,7 @@ def test_criar_cliente_com_data_e_decimais_vazios_cria_com_defaults(): def test_criar_cliente_com_dv_invalido_exige_confirmacao(monkeypatch): monkeypatch.setattr(router_clientes.db_clientes, "codigo_em_uso", lambda codigo, ignorar_id=None: False) + monkeypatch.setattr(router_clientes.db_clientes, "cnpj_em_uso", lambda documento, ignorar_id=None: False) criado = {} monkeypatch.setattr( router_clientes.db_clientes, "criar_cliente", lambda dados: criado.setdefault("ok", True) and 42 @@ -464,6 +500,7 @@ def test_criar_cliente_com_cnpj_dv_invalido_bloqueia_sempre(monkeypatch): "cadastrar mesmo assim" — nem com confirmar_dv_invalido=1. """ monkeypatch.setattr(router_clientes.db_clientes, "codigo_em_uso", lambda codigo, ignorar_id=None: False) + monkeypatch.setattr(router_clientes.db_clientes, "cnpj_em_uso", lambda documento, ignorar_id=None: False) criado = {} monkeypatch.setattr( router_clientes.db_clientes, "criar_cliente", lambda dados: criado.setdefault("ok", True) and 42 @@ -487,6 +524,7 @@ def test_criar_cliente_com_cnpj_dv_invalido_bloqueia_sempre(monkeypatch): def test_editar_cliente_com_cnpj_dv_invalido_bloqueia(monkeypatch): monkeypatch.setattr(router_clientes.db_clientes, "obter_cliente", lambda cliente_id: {"ID": cliente_id}) monkeypatch.setattr(router_clientes.db_clientes, "codigo_em_uso", lambda codigo, ignorar_id=None: False) + monkeypatch.setattr(router_clientes.db_clientes, "cnpj_em_uso", lambda documento, ignorar_id=None: False) atualizado = {} monkeypatch.setattr( router_clientes.db_clientes, @@ -506,6 +544,7 @@ def test_editar_cliente_com_cnpj_dv_invalido_bloqueia(monkeypatch): def test_criar_cliente_com_cpf_dv_invalido_continua_permitindo_confirmar(monkeypatch): """Regressão: CPF com DV inválido não muda de comportamento.""" monkeypatch.setattr(router_clientes.db_clientes, "codigo_em_uso", lambda codigo, ignorar_id=None: False) + monkeypatch.setattr(router_clientes.db_clientes, "cnpj_em_uso", lambda documento, ignorar_id=None: False) criado = {} monkeypatch.setattr( router_clientes.db_clientes, "criar_cliente", lambda dados: criado.setdefault("ok", True) and 42 @@ -527,6 +566,7 @@ def test_criar_cliente_com_cpf_dv_invalido_continua_permitindo_confirmar(monkeyp def test_criar_cliente_grava_telefone_com_ddi_55(monkeypatch): monkeypatch.setattr(router_clientes.db_clientes, "codigo_em_uso", lambda codigo, ignorar_id=None: False) + monkeypatch.setattr(router_clientes.db_clientes, "cnpj_em_uso", lambda documento, ignorar_id=None: False) capturado = {} monkeypatch.setattr( router_clientes.db_clientes, "criar_cliente", lambda dados: capturado.setdefault("dados", dados) and 42 @@ -543,6 +583,7 @@ def test_criar_cliente_grava_telefone_com_ddi_55(monkeypatch): def test_criar_cliente_telefone_ja_com_55_nao_duplica(monkeypatch): monkeypatch.setattr(router_clientes.db_clientes, "codigo_em_uso", lambda codigo, ignorar_id=None: False) + monkeypatch.setattr(router_clientes.db_clientes, "cnpj_em_uso", lambda documento, ignorar_id=None: False) capturado = {} monkeypatch.setattr( router_clientes.db_clientes, "criar_cliente", lambda dados: capturado.setdefault("dados", dados) and 42 @@ -558,6 +599,7 @@ def test_criar_cliente_telefone_ja_com_55_nao_duplica(monkeypatch): def test_criar_cliente_grava_uf_maiuscula_e_cep_sem_traco(monkeypatch): monkeypatch.setattr(router_clientes.db_clientes, "codigo_em_uso", lambda codigo, ignorar_id=None: False) + monkeypatch.setattr(router_clientes.db_clientes, "cnpj_em_uso", lambda documento, ignorar_id=None: False) capturado = {} monkeypatch.setattr( router_clientes.db_clientes, "criar_cliente", lambda dados: capturado.setdefault("dados", dados) and 42