Files
dtf-system/tests/test_jadlog.py
Cauê Faleiros 9bea187ea4
All checks were successful
Build and deploy / Validate source (push) Successful in 1m20s
Build and deploy / Integration suite on a real stack (push) Successful in 2m49s
Build and deploy / Secret scan and release gate (push) Successful in 10s
Build and deploy / Publish images (push) Successful in 1m51s
feat: add a read-only Jadlog price probe
app/jadlog.py prices one package through Jadlog's Simulador de Frete as the
API manual v2.3 describes it; app.jadlog_probe prices test weights to six
regions on the client's account to confirm token, account and contract.
Tested against a fake transport. The roadmap records the Jadlog data and the
Mercado Pago account setup.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-09-28 10:41:18 -03:00

101 lines
4.2 KiB
Python

"""The Jadlog quote client against a fake HTTP transport: payload and errors.
This proves the manual's contract only; app.jadlog_probe must still confirm
it on the client's account. Runs where httpx is installed.
"""
import json
import unittest
import httpx
from app.jadlog import QUOTE_URL, JadlogError, JadlogQuotes
from app.jadlog_probe import run
CNPJ = '11.222.333/0001-81'
def client(handler, **settings):
return JadlogQuotes(token=settings.pop('token', 'tok-1'), cnpj=CNPJ, conta='123456',
transport=httpx.MockTransport(handler), **settings)
class JadlogQuoteTest(unittest.TestCase):
def test_payload_follows_the_manual_and_price_becomes_centavos(self):
seen = []
def handler(request):
seen.append(request)
item = json.loads(request.content)['frete'][0]
return httpx.Response(200, json={'frete': [{**item, 'vltotal': 23.455, 'prazo': 4}]})
result = client(handler).quote('01310-100', 1.5, 12990)
request = seen[0]
self.assertEqual(str(request.url), QUOTE_URL)
self.assertEqual(request.headers['authorization'], 'tok-1')
item = json.loads(request.content)['frete'][0]
self.assertEqual(item, {'cepori': '14402310', 'cepdes': '01310100', 'frap': 'N', 'peso': 1.5,
'cnpj': '11222333000181', 'conta': '123456', 'contrato': None,
'modalidade': 3, 'tpentrega': 'D', 'tpseguro': 'N',
'vldeclarado': 129.9, 'vlcoleta': 0})
self.assertEqual((result['total_cents'], result['days']), (2346, 4))
def test_contract_is_sent_only_when_configured(self):
def handler(request):
item = json.loads(request.content)['frete'][0]
self.assertEqual(item['contrato'], '042')
return httpx.Response(200, json={'frete': [{'vltotal': 10, 'prazo': 2}]})
client(handler, contrato='042').quote('01310100', 1, 100)
def test_account_and_item_errors_are_raised_with_jadlog_text(self):
replies = [
httpx.Response(200, json={'frete': [{}], 'error': {'id': -1, 'descricao': 'frete[0].contrato Numero de contrato invalido'}}),
httpx.Response(200, json={'frete': [{'erro': {'id': 2, 'descricao': 'CEP destino invalido'}}]}),
httpx.Response(200, json={'frete': [{'prazo': 3}]}),
httpx.Response(502, text='<html>bad gateway</html>'),
]
messages = ['contrato invalido', 'CEP destino invalido', 'no freight value', 'not JSON']
for reply, message in zip(replies, messages):
with self.subTest(message=message), self.assertRaisesRegex(JadlogError, message):
client(lambda request, reply=reply: reply).quote('01310100', 1, 100)
def test_missing_credentials_refuse_to_start(self):
with self.assertRaises(RuntimeError):
JadlogQuotes(token='', cnpj=CNPJ)
with self.assertRaises(RuntimeError):
JadlogQuotes(token='tok', cnpj='123')
class JadlogProbeTest(unittest.TestCase):
def test_bearer_is_tried_once_after_a_401_and_kept(self):
headers = []
def handler(request):
headers.append(request.headers['authorization'])
if not request.headers['authorization'].startswith('Bearer '):
return httpx.Response(401, json={'descricao': 'TOKEN INVALIDO', 'id': 1})
return httpx.Response(200, json={'frete': [{'vltotal': 12.5, 'prazo': 3}]})
lines = []
status = run(client(handler), ['01310100', '20040002'], ['1'], 10000, lines.append)
self.assertEqual(status, 0)
self.assertEqual(headers, ['tok-1', 'Bearer tok-1', 'Bearer tok-1'])
self.assertIn('Bearer', lines[0])
self.assertIn('R$ 12,50', lines[1])
def test_first_failure_stops_the_run(self):
calls = []
def handler(request):
calls.append(request)
return httpx.Response(200, json={'frete': [{}], 'error': {'id': -1, 'descricao': 'CNPJ invalido'}})
lines = []
self.assertEqual(run(client(handler), ['01310100', '20040002'], ['1', '2'], 10000, lines.append), 1)
self.assertEqual(len(calls), 1)
self.assertIn('CNPJ invalido', lines[0])
if __name__ == '__main__':
unittest.main()