Comercial-Scardua/conector-scardua
2
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Adiciona testes do conector com fixtures offline

Browse source 
Cobre o caminho completo sem depender da VPS estar no ar: cifra com uma
chave de teste descartavel, troca a chave do app por monkeypatch e exercita
a rota.

8 casos: recusa sem token e com token errado; caminho feliz; envelope
adulterado (o GCM tem que estourar, nao passar); cifrado com outra chave;
payload fora do contrato (422, nao 200 nem 500); e dois travando o contrato
com a API da VPS — se o PayloadCompras dos dois lados divergir, quebra aqui
em vez de em producao.

Fixtures em tests/fixtures/: par de chaves so pra teste, o payload real do
Holmes e o payload ja tratado. O .gitignore abre excecao estreita pra
tests/fixtures/*.pem — chave real continua barrada.

Tambem:
- 422 passa a usar HTTP_422_UNPROCESSABLE_CONTENT (o _ENTITY deprecou)
- app/services/holmes.py removido: o conector nao fala com o Holmes
- app/utils/cesar.py entra como veio

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Ricardo 2026-08-18 06:13:13 -03:00
parent f902ccbecb
commit 7a9b0f8aa5
9 changed files with 283 additions and 705 deletions
Download patch file Download diff file Expand all files Collapse all files

704
app/services/holmes.py
View file

@ -1,704 +0,0 @@
import asyncio
import json
import logging
import os
import random
import re
import tempfile
import time
import httpx
from app.config import settings
from app.services.controle_api import registrar_contador
from oracledb import Connection
_TOKEN_FILE = os.path.join(tempfile.gettempdir(), "holmes_token_cache.json")
def _ler_token_arquivo() -> dict | None:
try:
with open(_TOKEN_FILE) as f:
return json.load(f)
except Exception:
return None
def _salvar_token_arquivo(token: dict):
try:
with open(_TOKEN_FILE, "w") as f:
json.dump(token, f)
except Exception:
pass
def _invalidar_token_usuario():
try:
os.remove(_TOKEN_FILE)
except Exception:
pass
async def token_usuario():
cached = _ler_token_arquivo()
if cached:
return cached
await asyncio.sleep(random.uniform(0, 0.5))
cached = _ler_token_arquivo()
if cached:
return cached
url = "https://app-api.holmesdoc.io/v1/session"
body = {"email": settings.holmes.usuario , "password": settings.holmes.senha}
headers = {"Content-Type": "application/json", "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/150.0.0.0 Safari/537.36"}
async with httpx.AsyncClient() as client:
response = await client.post(url, json=body, headers=headers)
response.raise_for_status()
token = {"authorization": response.json()["token"]}
_salvar_token_arquivo(token)
return token
async def get_header() -> tuple[dict, str]:
if random.randint(1, 1) >= 2:
return {"api_token": settings.holmes.token_api}, "holmes"
header = await token_usuario()
return header, "holmes_user"
async def _holmes_request(
method: str, url: str, **kwargs
) -> tuple[httpx.Response, str]:
api_nome: str = "holmes"
for tentativa in range(2):
header, api_nome = await get_header()
async with httpx.AsyncClient() as client:
response = await getattr(client, method)(url, headers=header, **kwargs)
if (
response.status_code == 401
and api_nome == "holmes_user"
and tentativa == 0
):
_invalidar_token_usuario()
continue
return response, api_nome
raise RuntimeError("Holmes: falha de autenticação após retry")
def extrair_origem(origem: str):
"""Extrair origem da string do Holmes
Args:
origem (str): 1548 Entrada de Freio
Returns:
str: 1548
"""
return str(origem[:4])
def extrair_empresa(unidade: str):
"""Extrai a empresa da string inteira do holmes
Args:
unidade (str): Ex 10.1 Hyundai Teix. Freitas
Returns:
empresa: 10 | None
revenda: 1 | None
"""
match = re.search(r"^(\d+)\.(\d+)", unidade)
if match:
return match.group(1), match.group(2)
return None, None
def extrair_transacao(transacao: str) -> str:
"""Extrai apenas a parte transação do texto do Holmes
Args:
transacao (str): D15 Entrada de Nota
Returns:
str: D15
"""
return transacao[:3]
async def get_holmes_process(id_processo: str, conn: Connection):
"""
Busca um processo no Holmes. Centralizado para Peças e Despesas.
"""
url = f"https://app-api.holmesdoc.io/v1/processes/{id_processo}"
inicio = time.perf_counter()
status = None
sucesso = False
erro = None
api_nome = "holmes"
try:
response, api_nome = await _holmes_request("get", url)
response.raise_for_status()
status = response.status_code
sucesso = True
return response.json()
except httpx.HTTPStatusError as e:
status = e.response.status_code
erro = str(e)
logging.error(f"Erro na API Holmes (ID {id_processo}): {e}")
return None
except Exception as e:
erro = str(e)
logging.error(f"Erro inesperado ao consultar Holmes: {e}")
return None
finally:
duracao = (time.perf_counter() - inicio) * 1000
registrar_contador(
conn=conn,
api=api_nome,
endpoint="/v1/processes/{id}",
metodo="GET",
status_code=status,
sucesso=sucesso,
duracao_ms=duracao,
erro=erro,
)
async def get_holmes_process_details(id_processo: str, conn: Connection):
"""
Busca os detalhes (properties) de um processo no Holmes.
"""
url = f"https://app-api.holmesdoc.io/v1/processes/{id_processo}/details"
inicio = time.perf_counter()
status = None
sucesso = False
erro = None
api_nome = "holmes"
try:
response, api_nome = await _holmes_request("get", url)
response.raise_for_status()
status = response.status_code
sucesso = True
return response.json()
except httpx.HTTPStatusError as e:
status = e.response.status_code
erro = str(e)
logging.error(f"Erro na API Holmes details (ID {id_processo}): {e}")
return None
except Exception as e:
erro = str(e)
logging.error(f"Erro inesperado ao consultar Holmes details: {e}")
return None
finally:
duracao = (time.perf_counter() - inicio) * 1000
registrar_contador(
conn=conn,
api=api_nome,
endpoint="/v1/processes/{id}/details",
metodo="GET",
status_code=status,
sucesso=sucesso,
duracao_ms=duracao,
erro=erro,
)
async def get_holmes_history(id_processo: str, conn: Connection):
"""
Busca o historico no holmes (importante para pegar o id da ultima task, para avançar futuramente)
Args:
id_processo (str): id do processo no holmes
"""
url = f"https://app-api.holmesdoc.io/v1/processes/{id_processo}/history"
payload = {
"filters": [],
"page": 1,
"per_page": 100,
"sortBy": ["created_at", "desc"],
}
inicio = time.perf_counter()
status = None
sucesso = False
erro = None
api_nome = "holmes"
try:
response, api_nome = await _holmes_request("post", url, json=payload)
response.raise_for_status()
status = response.status_code
sucesso = True
return response.json()
except httpx.HTTPStatusError as e:
status = e.response.status_code
erro = str(e)
logging.error(f"Erro na API Holmes (ID {id_processo}): {e}")
return None
except Exception as e:
erro = str(e)
logging.error(f"Erro inesperado ao consultar Holmes: {e}")
return None
finally:
duracao = (time.perf_counter() - inicio) * 1000
registrar_contador(
conn=conn,
api=api_nome,
endpoint="/v1/processes/{id}/history",
metodo="POST",
status_code=status,
sucesso=sucesso,
duracao_ms=duracao,
erro=erro,
)
async def get_holmes_rateio(id_processo: str, conn: Connection):
url = f"https://app-api.holmesdoc.io/v1/processes/{id_processo}/tables/e124b2d0-ee14-11ef-95b4-25dee32fe73f/table_items?page=1&per_page=800"
inicio = time.perf_counter()
status = None
sucesso = False
erro = None
api_nome = "holmes"
try:
response, api_nome = await _holmes_request("get", url)
response.raise_for_status()
status = response.status_code
sucesso = True
return response.json()
except httpx.HTTPStatusError as e:
status = e.response.status_code
erro = str(e)
logging.error(f"Erro na API Holmes (ID {id_processo}): {e}")
return None
except Exception as e:
erro = str(e)
logging.error(f"Erro inesperado ao consultar Holmes: {e}")
return None
finally:
duracao = (time.perf_counter() - inicio) * 1000
registrar_contador(
conn=conn,
api=api_nome,
endpoint="/v1/processes/{id}/table/(rateio)",
metodo="GET",
status_code=status,
sucesso=sucesso,
duracao_ms=duracao,
erro=erro,
)
async def task_id_recente(id_processo: str, conn: Connection):
dados_tasks = await get_holmes_history(id_processo, conn)
if (
not dados_tasks
or "histories" not in dados_tasks
or not dados_tasks["histories"]
):
return None
mais_recente = max(dados_tasks["histories"], key=lambda x: x["created_at"])
return mais_recente["properties"]["task_id"]
async def task_mais_recente(id_processo: str, conn: Connection):
dados_tasks = await get_holmes_history(id_processo, conn)
if not dados_tasks or "histories" not in dados_tasks:
return None # Tratamento se a API falhar
# print(dados_tasks)
mais_recente = max(dados_tasks["histories"], key=lambda x: x["created_at"])
return mais_recente
async def historicos_task(id_processo: str, conn: Connection) -> dict | None:
"""_summary_
Args:
id_processo (str): Id do processo no holmes
Returns:
dict | None : dicionario do historico | None
"""
dados_tasks = await get_holmes_history(id_processo, conn)
if not dados_tasks or "histories" not in dados_tasks:
return None # Tratamento se a API falhar
return dados_tasks
async def buscar_processo(
conn: Connection,
chave: str | None = None,
fluxos: list[str] | bool = False,
ativos: bool = True,
payload: dict | bool = False,
) -> dict:
"""Obtem os processos que existem com a sua chave
Args:
chave (str): Chave principal a ser procurada, preferencialmente unica pfvr, ajuda ae po.
ativos (bool) Defaults to True
fluxos (list[str] | bool, optional): _description_. Defaults to False. se quer pegar de um fluxo específico ou geral. Padrão: Geral
Returns:
dict: _description_
"""
if not payload and chave is None:
raise ValueError(
"Chave é obrigatória caso o payload não seja enviado chefia, fica alerta ae rapa"
)
inicio = time.perf_counter()
status = None
sucesso = False
erro = None
api_nome = "holmes"
url = "https://app-api.holmesdoc.io/v2/search"
if not payload:
payload = {
"query": {
"from": 0,
"size": 200,
"context": "process",
"sort": "updated_at",
"order": "desc",
"groups": [
{
"match_all": True,
"terms": [
{
"value": f"{chave}",
"type": "match_phrase",
"field": "_content",
}
],
}
],
},
"trash": False,
"deleted_by_me": False,
}
try:
response, api_nome = await _holmes_request("post", url, json=payload)
response.raise_for_status()
status = response.status_code
sucesso = True
dados = response.json()
docs = dados.get("docs", [])
if ativos:
docs = [d for d in docs if d.get("status") != "canceled"]
if fluxos:
docs = [d for d in docs if d.get("name") in fluxos]
return {"status": True, "dados": {**dados, "docs": docs, "total": len(docs)}}
except httpx.HTTPStatusError as e:
status = e.response.status_code
erro = str(e)
logging.error(f"Erro na API Holmes (ID {chave}): {e}")
return {"status": False, "error": e}
except Exception as e:
erro = str(e)
logging.error(f"Erro inesperado ao consultar Holmes: {e}")
return {"status": False, "error": e}
finally:
duracao = (time.perf_counter() - inicio) * 1000
registrar_contador(
conn=conn,
api=api_nome,
endpoint="/v1/processes/{id}/search/por-chave",
metodo="POST",
status_code=status,
sucesso=sucesso,
duracao_ms=duracao,
erro=erro,
)
async def buscar_processo_por_chaves(
conn: Connection,
combinacoes: list[list[str]],
fluxos: list[str] | bool = False,
ativos: bool = True,
) -> dict:
"""Busca processos no Holmes usando combinações de termos.
Cada item de combinacoes é uma lista de valores que juntos identificam
um processo único (ex: [cnpj, numero_nf]). Cada combinação vira um group
separado na query.
Args:
combinacoes: Ex: [["03657256000164", "19"], ["698cd1c570fd0f8f5f8436a4"]]
ativos: Ignora processos cancelados. Padrão: True.
fluxos: Filtra por nome de fluxo. Padrão: False (todos).
"""
url = "https://app-api.holmesdoc.io/v2/search"
inicio = time.perf_counter()
status = None
sucesso = False
erro = None
api_nome = "holmes"
payload = {
"query": {
"from": 0,
"size": 200,
"context": "process",
"sort": "updated_at",
"order": "desc",
"groups": [
{
"match_all": True,
"terms": [
{"value": termo, "type": "match_phrase", "field": "_content"}
for termo in combinacao
],
}
for combinacao in combinacoes
],
},
"trash": False,
"deleted_by_me": False,
}
try:
response, api_nome = await _holmes_request("post", url, json=payload)
response.raise_for_status()
status = response.status_code
sucesso = True
dados = response.json()
docs = dados.get("docs", [])
if ativos:
docs = [d for d in docs if d.get("status") != "canceled"]
if fluxos:
docs = [d for d in docs if d.get("name") in fluxos]
return {"status": True, "dados": {**dados, "docs": docs, "total": len(docs)}}
except httpx.HTTPStatusError as e:
status = e.response.status_code
erro = str(e)
logging.error(f"Erro na API Holmes (combinacoes {combinacoes}): {e}")
return {"status": False, "error": e}
except Exception as e:
erro = str(e)
logging.error(f"Erro inesperado ao consultar Holmes: {e}")
return {"status": False, "error": e}
finally:
duracao = (time.perf_counter() - inicio) * 1000
registrar_contador(
conn=conn,
api=api_nome,
endpoint="/v1/processes/{id}/search/por-chaves",
metodo="POST",
status_code=status,
sucesso=sucesso,
duracao_ms=duracao,
erro=erro,
)
async def action(payload: dict, id_task: str, id_processo: str, conn: Connection):
url = f"https://app-api.holmesdoc.io/v1/tasks/{id_task}/action"
inicio = time.perf_counter()
status = None
sucesso = False
erro = None
api_nome = "holmes"
try:
async with httpx.AsyncClient() as client:
response = await client.post(
url, headers={"api_token": settings.holmes.token_api}, json=payload
)
response.raise_for_status()
status = response.status_code
sucesso = True
return True, response.json()
except httpx.HTTPStatusError as e:
status = e.response.status_code
erro = str(e)
logging.error(f"Erro na API Holmes (ID {id_processo}): {e}")
return False, e
except Exception as e:
erro = str(e)
logging.error(f"Erro inesperado ao consultar Holmes: {e}")
return False, e
finally:
duracao = (time.perf_counter() - inicio) * 1000
registrar_contador(
conn=conn,
api=api_nome,
endpoint="/v1/processes/{id}/action",
metodo="POST",
status_code=status,
sucesso=sucesso,
duracao_ms=duracao,
erro=erro,
)
async def cria_processo(
id_start: str,
payload: dict,
conn: Connection
) -> tuple[bool, dict | str]:
url = f"https://app-api.holmesdoc.io/v1/workflows/{id_start}/start"
inicio = time.perf_counter()
status = None
sucesso = False
erro = None
api_nome = "holmes"
try:
response, api_nome = await _holmes_request("post", url, json=payload)
response.raise_for_status()
status = response.status_code
sucesso = True
return True, response.json()
except httpx.HTTPStatusError as e:
status = e.response.status_code
erro = str(e)
logging.error(f"Erro na API Holmes - Criar Processo ({payload}): {e}")
return False, str(e)
except Exception as e:
erro = str(e)
logging.error(f"Erro inesperado ao criar processo no Holmes: {e}")
return False, str(e)
finally:
duracao = (time.perf_counter() - inicio) * 1000
registrar_contador(
conn=conn,
api=api_nome,
endpoint="/v1/workflows/{id}/start",
metodo="POST",
status_code=status,
sucesso=sucesso,
duracao_ms=duracao,
erro=erro,
)
async def enviar_documento(
id_processo: str,
arquivo: bytes,
nome_arquivo: str,
id_documento: str,
conn: Connection,
) -> tuple[bool, str | dict]:
task_id = await task_id_recente(id_processo, conn)
if not task_id:
return False, "Não foi possível obter a task mais recente do Holmes"
url = f"https://app-api.holmesdoc.io/v1/tasks/{task_id}/documents/{id_documento}"
inicio = time.perf_counter()
status = None
sucesso = False
erro = None
api_nome = "holmes"
try:
files = {"file": (nome_arquivo, arquivo, "application/pdf")}
async with httpx.AsyncClient() as client:
response = await client.post(
url, headers={"api_token": settings.holmes.token_api}, files=files
)
response.raise_for_status()
status = response.status_code
sucesso = True
return True, {"task_id": task_id}
except httpx.HTTPStatusError as e:
status = e.response.status_code
erro = str(e)
logging.error(f"Erro ao enviar documento Holmes (ID {id_processo}): {e}")
return False, str(e)
except Exception as e:
erro = str(e)
logging.error(f"Erro inesperado ao enviar documento Holmes: {e}")
return False, str(e)
finally:
duracao = (time.perf_counter() - inicio) * 1000
registrar_contador(
conn=conn,
api=api_nome,
endpoint="/v1/tasks/{id}/documents/{id_documento}",
metodo="POST",
status_code=status,
sucesso=sucesso,
duracao_ms=duracao,
erro=erro,
)
# Para testar as funcoes
# async def main():
# from app.infra.database import db_instance
# db_instance.create_pool()
# conn = db_instance.pool.acquire()
# try:
# print(await get_holmes_history('69e65ea83fad950fad5715ed', conn))
# finally:
# db_instance.pool.release(conn)
# if __name__ == '__main__':
# import asyncio
# asyncio.run(main())
async def cancela_processo(
id_processo: str, conn: Connection
) -> tuple[bool, str | dict]:
url = f"https://app-api.holmesdoc.io/v1/processes/{id_processo}/cancel"
payload = {"reason": "Erro na emissão, data de vencimento. Problema na Disal."}
inicio = time.perf_counter()
status = None
sucesso = False
erro = None
api_nome = "holmes"
try:
response, api_nome = await _holmes_request("put", url, json=payload)
response.raise_for_status()
status = response.status_code
sucesso = True
return True, response.json() if response.content else {
"mensagem": "processo cancelado"
}
except httpx.HTTPStatusError as e:
status = e.response.status_code
erro = str(e)
logging.error(f"Erro ao cancelar processo Holmes (ID {id_processo}): {e}")
return False, str(e)
except Exception as e:
erro = str(e)
logging.error(f"Erro inesperado ao cancelar processo Holmes: {e}")
return False, str(e)
finally:
duracao = (time.perf_counter() - inicio) * 1000
registrar_contador(
conn=conn,
api=api_nome,
endpoint="/v1/processes/{id}/cancel",
metodo="PUT",
status_code=status,
sucesso=sucesso,
duracao_ms=duracao,
erro=erro,
)
async def main():
# print(aaaa())
print(await token_usuario())
if __name__ == "__main__":
import asyncio
asyncio.run(main())