From 7a9b0f8aa512f0ca7131cc43a786c1ddc258ed12 Mon Sep 17 00:00:00 2001 From: Ricardo Date: Tue, 18 Aug 2026 06:13:13 -0300 Subject: [PATCH] Adiciona testes do conector com fixtures offline MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) --- .gitignore | 4 + app/services/holmes.py | 704 ------------------------- app/utils/cesar.py | 41 ++ app/v1/compras.py | 2 +- tests/fixtures/chave_teste_privada.pem | 28 + tests/fixtures/chave_teste_publica.pem | 9 + tests/fixtures/holmes_compras.json | 1 + tests/fixtures/payload_tratado.json | 21 + tests/test_conector.py | 178 +++++++ 9 files changed, 283 insertions(+), 705 deletions(-) delete mode 100644 app/services/holmes.py create mode 100644 app/utils/cesar.py create mode 100644 tests/fixtures/chave_teste_privada.pem create mode 100644 tests/fixtures/chave_teste_publica.pem create mode 100644 tests/fixtures/holmes_compras.json create mode 100644 tests/fixtures/payload_tratado.json create mode 100644 tests/test_conector.py diff --git a/.gitignore b/.gitignore index 5381105..73b42a2 100644 --- a/.gitignore +++ b/.gitignore @@ -12,3 +12,7 @@ configs.json.* *.bak-* *.pem *.key + +# Excecao: chaves descartaveis de teste. Geradas so pro pytest, nunca +# usadas em runtime. As reais continuam barradas pela regra acima. +!tests/fixtures/*.pem diff --git a/app/services/holmes.py b/app/services/holmes.py deleted file mode 100644 index b4020ea..0000000 --- a/app/services/holmes.py +++ /dev/null @@ -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()) diff --git a/app/utils/cesar.py b/app/utils/cesar.py new file mode 100644 index 0000000..cf9f4d8 --- /dev/null +++ b/app/utils/cesar.py @@ -0,0 +1,41 @@ +"""Cifra de Cesar para criptografar/descriptografar senhas.""" + +import sys + + +def caesar_encrypt(text: str, shift: int) -> str: + result = [] + for char in text: + if char.isalpha(): + base = ord("A") if char.isupper() else ord("a") + result.append(chr((ord(char) - base + shift) % 26 + base)) + elif char.isdigit(): + result.append(str((int(char) + shift) % 10)) + else: + result.append(char) + return "".join(result) + + +def caesar_decrypt(text: str, shift: int) -> str: + return caesar_encrypt(text, -shift) + + +if __name__ == "__main__": + if len(sys.argv) < 4: + print("Uso: python cezar.py [texto2 ...]") + print("Exemplo: python cezar.py encript 12 minha_senha outra_senha") + sys.exit(1) + + modo = sys.argv[1].lower() + shift = int(sys.argv[2]) + textos = sys.argv[3:] + + if modo == "encript": + for i, texto in enumerate(textos, start=1): + print(f"texto_criptografado{i}: {caesar_encrypt(texto, shift)}") + elif modo == "decript": + for i, texto in enumerate(textos, start=1): + print(f"texto_descriptografado{i}: {caesar_decrypt(texto, shift)}") + else: + print("Modo inválido. Use 'encript' ou 'decript'.") + sys.exit(1) diff --git a/app/v1/compras.py b/app/v1/compras.py index 719ba8d..96d9862 100644 --- a/app/v1/compras.py +++ b/app/v1/compras.py @@ -37,7 +37,7 @@ async def receber_compras(envelope: Envelope) -> dict: # Decifrou mas o formato mudou: os dois lados sairam de sincronia. logging.error(f"Payload fora do contrato: {e}") raise HTTPException( - status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, + status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail="payload nao bate com o contrato esperado", ) diff --git a/tests/fixtures/chave_teste_privada.pem b/tests/fixtures/chave_teste_privada.pem new file mode 100644 index 0000000..a9c1aff --- /dev/null +++ b/tests/fixtures/chave_teste_privada.pem @@ -0,0 +1,28 @@ +-----BEGIN PRIVATE KEY----- +MIIEvQIBADANBgkqhkiG9w0BAQEFAASCBKcwggSjAgEAAoIBAQC/B6IxhXHEgkmR +INcZoR2py+NlOU0AihMvoPGcZcgD5YsDOLI2WuNlPn4d4bPLN3hyfSsduROUEpwn +78q/ZuEpHEW4YBE8wWrhPOTcMSQ4gwnJ/vDlXhXu1I5iwauOGn4qd8tUtqPavqZ0 ++H7UCwo/JXmiPSB4z95rY8ZU+AhRY+YAYa2KP/Omjypb2bKurlglY3NId0862lgK +gfAMe6fCu16Cbw74A4lU1xtv3v98zb8QtRR19qzz4UuSTGPGHWBsHbNnmBmq/sHv +IL0BDSFuWxUdtiT61f+mzmCI0bqdvi9EzgyMjcLChzKUJzQnYiJXUd9anS3poHX9 +L0uVQxSRAgMBAAECgf84DKS6hzLyZvqd9jUTVNunsq1rJ6OvBz585ob40HoKx6EL +tlz0Vwiup0HvgpIq/5knRu5MVl0fx87UcDAp0gD2E85NyExoQ4Nqr+cuD+r/1UKR +RZzEoUWKjvncTVd8otCSNjz52BWXUv1AgENacXiLW93IW+NzonHopjccCUngQCdF +CSgxCNUZYJN8+DqcaPos4DyBhyQa7jIlGINaAQjvXfy0OzPeTRkEYKUd/UvjArpe +CYk0JEQSh5GIdFy5ljMqRmMzKGfbvlSXASfpH8Rv1UP/Z9meLd241rhvvSicOHKe +W7t6jgjberbgYon+VY6nioTWrFKP7htdvRNwQ/ECgYEA7lSEaeS+yx9RWwRLqHnX +z8JWe8RUD2k7fgtIr9dADqxtBjpSZD5Ito6yZkrENlKKdtmrChOkO+6lQdwMShcd +TuWg/83UxzAdSzmobPGw4nehZ3FvO0HhVV5GCXjFqLdlIFLC/q4CTA+Kr9Zq8eJw +jxVVjT8e+utGPNwSL3Hx0n0CgYEAzTFcW07PqG0wxP9VIQyn1fAO+NnjgP0rO61l +0pX8e7bbX/LezT1RaeZ6GXKNcpTthoZbM79QMq+yrwVU+IiG2gocGJ0ta5F1PLoN +Ndr2MpH4XIR/w3Hc5QJqqo7FXQpaVwBmtxkMcncJzXkayU6iuX5rTMMgNuK9i5aM +M7aSMqUCgYEAz6bp/A2mwvnVe5ThirgmdclgatPjXc4VXLveJ/9Gu8I198Am+1pd +JlpsS74G+UGvjOAYK15zsRg7+ocMWh2LlqtyPI8NPkPIsjtZaRqoQJl1Etj9zkaX +WzcZlLUamua3gJM7fIUGUVkVQCF06q2d3rz63mdJydvmRa6FVbtUtE0CgYEAloH5 +U4Q4bztARY9gGvDUfukpokD4ThnlR03F3TRk0T3sJbHY2UR73ijPMLFErWIt47nT +Vd6jbbpQX26SyAYkm/REbY2Ezl0QWG3D2Nf2NFu7h+ksaeiv9U7TfK1ieP9qClzh ++rWl2qQUDaIiErzaQSNIgzKxA3FHRQc1aY0mKX0CgYEAxFOdYHbi08BdGvigs4jG +BaXDYhKo4+B+N8Rim0DcfEmMTl03mIhcHPiprOspKntcryUirkizXr/KYJsAoegS +e/BdvEAfl3Nv/gabCMLHch3As5plCa0MJ1lFaAJVzbo7h7YnpdCsrxBxJ4xUTZKG +JO8fje7X8wfQ4DUxXZHkbkY= +-----END PRIVATE KEY----- diff --git a/tests/fixtures/chave_teste_publica.pem b/tests/fixtures/chave_teste_publica.pem new file mode 100644 index 0000000..7362574 --- /dev/null +++ b/tests/fixtures/chave_teste_publica.pem @@ -0,0 +1,9 @@ +-----BEGIN PUBLIC KEY----- +MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAvweiMYVxxIJJkSDXGaEd +qcvjZTlNAIoTL6DxnGXIA+WLAziyNlrjZT5+HeGzyzd4cn0rHbkTlBKcJ+/Kv2bh +KRxFuGARPMFq4Tzk3DEkOIMJyf7w5V4V7tSOYsGrjhp+KnfLVLaj2r6mdPh+1AsK +PyV5oj0geM/ea2PGVPgIUWPmAGGtij/zpo8qW9myrq5YJWNzSHdPOtpYCoHwDHun +wrtegm8O+AOJVNcbb97/fM2/ELUUdfas8+FLkkxjxh1gbB2zZ5gZqv7B7yC9AQ0h +blsVHbYk+tX/ps5giNG6nb4vRM4MjI3CwocylCc0J2IiV1HfWp0t6aB1/S9LlUMU +kQIDAQAB +-----END PUBLIC KEY----- diff --git a/tests/fixtures/holmes_compras.json b/tests/fixtures/holmes_compras.json new file mode 100644 index 0000000..c9ac10e --- /dev/null +++ b/tests/fixtures/holmes_compras.json @@ -0,0 +1 @@ +{"id":"6a6a59266f8555a6725125a7","author":{"id":"69b00f9897ee2e2f33fc7dc8","name":"Acrisio Dos Santos","email":"acrisio@comercialscardua.com.br"},"created_at":null,"properties":{"protocolo":"protocolo","Tipo de Compra":"COMPRA INTERNA - DESPESAS","Fornecedores":"ACIMAQ EQUIPAMENTOS INDUSTRIAIS E COMERCIAIS LTDA","Prioridade Alta - Motivo":"","Observações":"","Possui Adiantamento do Cliente?":"Não","Fornecedor - Exclusividade?":false,"Cond. Pagamento - 1 (Em dias ex. 30/60/90)":"3060","Forma de Pagamento - 1 ":"Boleto","Descrição da Compra":"ertyhrewerth","ORIGEM CENTRO CUSTO":"32456","PEDIDO_LINX":345,"Pagamento Antecipado Fornecedor":"Sim","Valor Adiantamento fornecedor":null,"parcela1":100.0,"parcela2":100.0,"parcela3":100.0,"parcela4":100.0,"parcela5":100.0,"parcela6":100.0,"parcela7":100.0,"parcela8":100.0,"parcela9":100.0,"vencimento parcela3":"2026-07-10T00:00:00.000Z","vencimento parcela4":"2026-07-10T00:00:00.000Z","vencimento parcela5":"2026-07-22T00:00:00.000Z","vencimento parcela6":"2026-07-17T00:00:00.000Z","vencimento parcela7":"2026-07-11T00:00:00.000Z","vencimento parcela8":"2026-07-23T00:00:00.000Z","vencimento parcela9":"2026-07-17T00:00:00.000Z","vencimento parcela11":"2026-07-23T00:00:00.000Z","vencimento parcela12":"2026-07-03T00:00:00.000Z","parcela11":100720.26,"parcela12":100720.26,"num. parcelas":11,"CNPJ":"28482230000153"},"documents":[]} diff --git a/tests/fixtures/payload_tratado.json b/tests/fixtures/payload_tratado.json new file mode 100644 index 0000000..a8addbd --- /dev/null +++ b/tests/fixtures/payload_tratado.json @@ -0,0 +1,21 @@ +{ + "id_processo": "6a6a59266f8555a6725125a7", + "protocolo": "protocolo", + "cnpj": "28482230000153", + "pedido_linx": "345", + "fornecedor": "ACIMAQ EQUIPAMENTOS INDUSTRIAIS E COMERCIAIS LTDA", + "tipo": "COMPRA INTERNA - DESPESAS", + "nf_entrada": null, + "aprovador": "Acrisio Dos Santos", + "valor_total": 202340.52, + "parcelas": { + "1": { + "valor": 100.0, + "vencimento": null + }, + "2": { + "valor": 100.0, + "vencimento": null + } + } +} \ No newline at end of file diff --git a/tests/test_conector.py b/tests/test_conector.py new file mode 100644 index 0000000..183eab1 --- /dev/null +++ b/tests/test_conector.py @@ -0,0 +1,178 @@ +""" +Testes do caminho completo do conector, sem depender da VPS estar no ar. + +A chave em tests/fixtures/ e descartavel, gerada so pra isto — nao tem +relacao com a do configs.json. O teste cifra com a publica de teste e +troca a chave do app pela privada de teste, entao roda offline. +""" + +import json +from datetime import datetime, timezone +from pathlib import Path + +import pytest +from app.schemas import PayloadCompras +from app.services import cripto +from cryptography.hazmat.primitives import hashes, serialization +from cryptography.hazmat.primitives.asymmetric import padding +from cryptography.hazmat.primitives.ciphers.aead import AESGCM +from fastapi.testclient import TestClient + +FIXTURES = Path(__file__).parent / "fixtures" +ROTA = "/v1/compras/dados" + +_OAEP = padding.OAEP( + mgf=padding.MGF1(algorithm=hashes.SHA256()), + algorithm=hashes.SHA256(), + label=None, +) + + +def _cifrar(dados: dict) -> dict: + """Faz o papel da API da VPS: cifra com a publica de teste.""" + pem = (FIXTURES / "chave_teste_publica.pem").read_bytes() + publica = serialization.load_pem_public_key(pem) + + chave_aes = AESGCM.generate_key(bit_length=256) + nonce = b"\x00" * 12 # fixo no teste: nao precisa de aleatoriedade aqui + corpo = json.dumps(dados, ensure_ascii=False, separators=(",", ":")).encode() + + import base64 + + return { + "alg": cripto.ALGORITMO, + "chave": base64.b64encode(publica.encrypt(chave_aes, _OAEP)).decode(), + "nonce": base64.b64encode(nonce).decode(), + "dados": base64.b64encode(AESGCM(chave_aes).encrypt(nonce, corpo, None)).decode(), + } + + +@pytest.fixture +def payload() -> dict: + return json.loads((FIXTURES / "payload_tratado.json").read_text(encoding="utf-8")) + + +@pytest.fixture +def cliente(monkeypatch) -> TestClient: + """App com a chave de teste no lugar da real.""" + from app.v1 import compras + + privada = cripto.carregar_chave_privada( + (FIXTURES / "chave_teste_privada.pem").read_text() + ) + monkeypatch.setattr(compras, "_CHAVE", privada) + + from main import app + + return TestClient(app, raise_server_exceptions=False) + + +@pytest.fixture +def token() -> str: + from app.config import settings + + return settings.vps.token + + +# --- autenticacao --------------------------------------------------------- + + +def test_sem_token_recusa(cliente, payload): + r = cliente.post(ROTA, json=_cifrar(payload)) + assert r.status_code == 401 + + +def test_token_errado_recusa(cliente, payload): + r = cliente.post(ROTA, json=_cifrar(payload), headers={"X-Token": "chute"}) + assert r.status_code == 401 + + +# --- caminho feliz -------------------------------------------------------- + + +def test_recebe_e_valida(cliente, payload, token): + r = cliente.post(ROTA, json=_cifrar(payload), headers={"X-Token": token}) + assert r.status_code == 200 + + corpo = r.json() + assert corpo["id_processo"] == payload["id_processo"] + assert corpo["parcelas"] == len(payload["parcelas"]) + # Enquanto nao houver Oracle isto tem que continuar False — se virar + # True sem camada de banco, alguem mentiu na resposta. + assert corpo["persistido"] is False + + +# --- rejeicoes ------------------------------------------------------------ + + +def test_envelope_adulterado_recusa(cliente, payload, token): + """O GCM autentica: mexer no corpo tem que estourar, nao passar.""" + envelope = _cifrar(payload) + envelope["dados"] = "A" + envelope["dados"][1:] + + r = cliente.post(ROTA, json=envelope, headers={"X-Token": token}) + assert r.status_code == 400 + + +def test_chave_errada_recusa(cliente, payload, token): + """Cifrado com outra publica: a privada do conector nao abre.""" + from cryptography.hazmat.primitives.asymmetric import rsa + + outra = rsa.generate_private_key(public_exponent=65537, key_size=2048).public_key() + envelope = _cifrar(payload) + import base64 + + envelope["chave"] = base64.b64encode( + outra.encrypt(AESGCM.generate_key(bit_length=256), _OAEP) + ).decode() + + r = cliente.post(ROTA, json=envelope, headers={"X-Token": token}) + assert r.status_code == 400 + + +def test_payload_fora_do_contrato_recusa(cliente, payload, token): + """ + Decifra mas o formato mudou: os dois lados sairam de sincronia. + + Tem que dar 422 (nao 200 nem 500) — falhar explicito e melhor do que + gravar dado torto no Oracle depois. + """ + torto = dict(payload) + del torto["cnpj"] + + r = cliente.post(ROTA, json=_cifrar(torto), headers={"X-Token": token}) + assert r.status_code == 422 + + +# --- contrato ------------------------------------------------------------- + + +def test_contrato_aceita_o_payload_da_api(payload): + """ + Trava o contrato com a API da VPS. + + Se este teste quebrar, o PayloadCompras daqui e o de la divergiram — + conferir os dois antes de qualquer deploy. + """ + dados = PayloadCompras.model_validate(payload) + + assert dados.id_processo + assert dados.valor_total > 0 + assert dados.parcelas + for numero, parcela in dados.parcelas.items(): + assert isinstance(numero, int) + assert parcela.valor > 0 + assert parcela.vencimento is None or isinstance(parcela.vencimento, datetime) + + +def test_vencimento_com_z_vira_datetime_utc(): + """O Holmes manda '2026-07-10T00:00:00.000Z'; tem que virar datetime aware.""" + dados = PayloadCompras.model_validate( + { + "id_processo": "x", "protocolo": "p", "cnpj": "1", "pedido_linx": "2", + "fornecedor": "F", "tipo": "T", "aprovador": "A", "valor_total": 10.0, + "parcelas": {"3": {"valor": 10.0, "vencimento": "2026-07-10T00:00:00.000Z"}}, + } + ) + vencimento = dados.parcelas[3].vencimento + assert vencimento == datetime(2026, 7, 10, tzinfo=timezone.utc)