conector-scardua/tests/test_conector.py

"""
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)