988b252f9d
All checks were successful
Build and deploy / Validate source (push) Successful in 11s
Build and deploy / Integration suite on a real stack (push) Successful in 2m39s
Build and deploy / Secret scan and release gate (push) Successful in 7s
Build and deploy / Publish images (push) Successful in 1m43s
"Testar conexão" now also reads the four configured Tiny products and requires each to be active. app/tiny_probe.py runs from the worker console to list products, confirm the configured ids, and create one marked test order through the worker's own delivery path, proving the duplicate guard by search before a second delivery. Nothing is sent without --confirmar. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
195 lines
10 KiB
Python
195 lines
10 KiB
Python
"""The Tiny v3 adapter against a fake HTTP transport: payload and idempotency.
|
|
|
|
This proves the documented contract only; the client's account must still
|
|
confirm products, contacts and order fields. Runs where httpx is installed.
|
|
The OAuth connection against the real database is tests/tiny_oauth_test.py.
|
|
"""
|
|
import json
|
|
import os
|
|
import unittest
|
|
from datetime import date
|
|
from unittest import mock
|
|
|
|
import httpx
|
|
|
|
from app.tiny import TinyError, TinyOrders, contact_payload, order_payload
|
|
|
|
PAID = {'order_id': 'b6f1c0de-0000-4000-8000-000000000001', 'number': 42, 'event': 'payment_approved',
|
|
'order': {'customer': {'cnpj': '11222333000181', 'zap': '16999999999', 'mail': 'loja@example.test'},
|
|
'items': [{'mode': 'avulsa', 'grade': 90, 'billed_metres': '2.8', 'unit_cents': 2490}],
|
|
'freight': {'service': 'mock-standard', 'total_cents': 1500},
|
|
'destination': {'recipient': 'Loja Teste', 'street': 'Rua A', 'number': '10',
|
|
'complement': '', 'district': 'Centro', 'city': 'Franca',
|
|
'state': 'SP', 'postal_code': '14400000'},
|
|
'total_cents': 8472}}
|
|
PRODUCTS = {'TINY_PRODUCT_TEXTIL_FOLHA': '101', 'TINY_PRODUCT_TEXTIL_AVULSA': '102',
|
|
'TINY_PRODUCT_UV_FOLHA': '103', 'TINY_PRODUCT_UV_AVULSA': '104'}
|
|
|
|
|
|
class FakeAuth:
|
|
def access_token(self):
|
|
return 'access-1'
|
|
|
|
|
|
class FakeTinyCase(unittest.TestCase):
|
|
"""A Tiny account in memory: contacts, orders and the four products."""
|
|
|
|
def setUp(self):
|
|
self.contacts = []
|
|
self.orders = []
|
|
self.calls = []
|
|
self.products = {value: {'id': int(value), 'sku': f'DTF-{value}', 'descricao': f'Produto {value}',
|
|
'situacao': 'A', 'precos': {'preco': 14.9}} for value in PRODUCTS.values()}
|
|
|
|
def handler(request):
|
|
path = request.url.path.removeprefix('/public-api/v3')
|
|
self.calls.append((request.method, path))
|
|
assert request.headers['authorization'] == 'Bearer access-1'
|
|
if request.method == 'GET' and path == '/contatos':
|
|
cnpj = request.url.params.get('cpfCnpj')
|
|
return httpx.Response(200, json={'itens': [c for c in self.contacts if cnpj in (None, c['cpfCnpj'])]})
|
|
if request.method == 'POST' and path == '/contatos':
|
|
contact = {**json.loads(request.content), 'id': 500 + len(self.contacts)}
|
|
self.contacts.append(contact)
|
|
return httpx.Response(200, json={'id': contact['id']})
|
|
if request.method == 'GET' and path.startswith('/produtos/'):
|
|
wanted = path.rsplit('/', 1)[-1]
|
|
if wanted not in self.products:
|
|
return httpx.Response(404, json={'mensagem': 'não encontrado'})
|
|
return httpx.Response(200, json=self.products[wanted])
|
|
if request.method == 'GET' and path == '/produtos':
|
|
return httpx.Response(200, json={'itens': list(self.products.values())})
|
|
if request.method == 'GET' and path == '/pedidos':
|
|
return httpx.Response(200, json={'itens': [{'id': o['id']} for o in self.orders]})
|
|
if request.method == 'GET' and path.startswith('/pedidos/'):
|
|
wanted = int(path.rsplit('/', 1)[-1])
|
|
return httpx.Response(200, json=next(o for o in self.orders if o['id'] == wanted))
|
|
if request.method == 'POST' and path == '/pedidos':
|
|
order = {**json.loads(request.content), 'id': 9000 + len(self.orders),
|
|
'numeroPedido': str(100 + len(self.orders))}
|
|
self.orders.append(order)
|
|
return httpx.Response(200, json={'id': order['id'], 'numeroPedido': order['numeroPedido']})
|
|
return httpx.Response(404)
|
|
|
|
self.handler = handler
|
|
self.tiny = TinyOrders(auth=FakeAuth(), transport=httpx.MockTransport(handler),
|
|
today=date(2026, 9, 24))
|
|
|
|
|
|
@mock.patch.dict(os.environ, PRODUCTS)
|
|
class TinyTests(FakeTinyCase):
|
|
def test_payload_carries_contact_products_address_and_freight(self):
|
|
pedido = order_payload(PAID, 777, date(2026, 9, 24))
|
|
self.assertEqual((pedido['idContato'], pedido['numeroOrdemCompra'], pedido['data']),
|
|
(777, 'DTF-42', '2026-09-24'))
|
|
item = pedido['itens'][0]
|
|
self.assertEqual((item['produto'], item['quantidade'], item['valorUnitario']),
|
|
({'id': 102}, 2.8, 24.9))
|
|
self.assertEqual(pedido['valorFrete'], 15.0)
|
|
self.assertEqual((pedido['enderecoEntrega']['cep'], pedido['enderecoEntrega']['enderecoNro'],
|
|
pedido['enderecoEntrega']['nomeDestinatario']), ('14400000', '10', 'Loja Teste'))
|
|
contact = contact_payload(PAID['order'])
|
|
self.assertEqual((contact['tipoPessoa'], contact['cpfCnpj'], contact['endereco']['uf']),
|
|
('J', '11222333000181', 'SP'))
|
|
pickup = order_payload({**PAID, 'order': {**PAID['order'], 'destination': None,
|
|
'freight': {'service': 'pickup', 'total_cents': 0}}}, 1)
|
|
self.assertNotIn('enderecoEntrega', pickup)
|
|
self.assertIn('retirada', pickup['observacoes'])
|
|
|
|
def test_second_delivery_finds_the_first_order(self):
|
|
first = self.tiny.deliver('k1', PAID)
|
|
second = self.tiny.deliver('k1', PAID)
|
|
self.assertEqual((first['status'], second['status']), ('created', 'already-created'))
|
|
self.assertEqual(first['tiny_id'], second['tiny_id'])
|
|
self.assertEqual(self.calls.count(('POST', '/pedidos')), 1)
|
|
self.assertEqual(self.calls.count(('POST', '/contatos')), 1)
|
|
|
|
def test_existing_contact_is_reused(self):
|
|
self.contacts.append({'id': 321, 'cpfCnpj': '11222333000181'})
|
|
self.tiny.deliver('k1', PAID)
|
|
self.assertNotIn(('POST', '/contatos'), self.calls)
|
|
self.assertEqual(self.orders[0]['idContato'], 321)
|
|
|
|
def test_production_events_are_not_sent(self):
|
|
self.assertEqual(self.tiny.deliver('k2', {**PAID, 'event': 'ready'})['status'], 'not-applicable')
|
|
self.assertEqual(self.calls, [])
|
|
|
|
def test_missing_product_mapping_fails_rather_than_guessing(self):
|
|
with mock.patch.dict(os.environ, {'TINY_PRODUCT_TEXTIL_AVULSA': ''}):
|
|
with self.assertRaises(TinyError):
|
|
self.tiny.deliver('k3', PAID)
|
|
self.assertNotIn(('POST', '/pedidos'), self.calls)
|
|
|
|
def test_connection_check_only_reads(self):
|
|
from app.tiny import check
|
|
results = check(auth=FakeAuth(), transport=httpx.MockTransport(self.handler))
|
|
self.assertEqual(set(results.values()), {'ok'})
|
|
self.assertEqual(len(results), 6)
|
|
self.assertTrue(all(method == 'GET' for method, _ in self.calls))
|
|
self.products['103']['situacao'] = 'I'
|
|
with mock.patch.dict(os.environ, {'TINY_PRODUCT_UV_AVULSA': ''}):
|
|
results = check(auth=FakeAuth(), transport=httpx.MockTransport(self.handler))
|
|
self.assertIn('não está ativo', results['DTF UV 28,5 cm · folha montada'])
|
|
self.assertEqual(results['DTF UV 28,5 cm · artes avulsas'], 'TINY_PRODUCT_UV_AVULSA sem id')
|
|
self.assertEqual(results['pedidos'], 'ok')
|
|
denied = check(auth=FakeAuth(), transport=httpx.MockTransport(lambda r: httpx.Response(403, text='forbidden')))
|
|
self.assertTrue(denied['pedidos'].startswith('GET /pedidos: 403'))
|
|
|
|
def test_rate_limit_is_a_retryable_failure(self):
|
|
tiny = TinyOrders(auth=FakeAuth(), transport=httpx.MockTransport(lambda r: httpx.Response(429)))
|
|
with self.assertRaisesRegex(TinyError, 'rate limit'):
|
|
tiny.deliver('k4', PAID)
|
|
|
|
|
|
@mock.patch.dict(os.environ, PRODUCTS)
|
|
class TinyProbeTests(FakeTinyCase):
|
|
"""The supervised console tool run against the fake Tiny."""
|
|
|
|
def probe(self, *argv):
|
|
from app import tiny_probe
|
|
lines = []
|
|
code = tiny_probe.main(list(argv), orders=self.tiny, out=lines.append)
|
|
return code, '\n'.join(lines)
|
|
|
|
def test_products_and_settings_are_listed_without_writing(self):
|
|
code, text = self.probe('produtos', 'DTF')
|
|
self.assertEqual(code, 0)
|
|
self.assertIn('101\tDTF-101\tProduto 101', text)
|
|
code, text = self.probe('conferir')
|
|
self.assertEqual(code, 0)
|
|
self.assertIn('TINY_PRODUCT_UV_AVULSA (DTF UV 28,5 cm · artes avulsas): ok DTF-104', text)
|
|
self.assertTrue(all(method == 'GET' for method, _ in self.calls))
|
|
|
|
def test_order_is_shown_and_not_created_without_confirmation(self):
|
|
code, text = self.probe('pedido', '--cnpj', '11.222.333/0001-81', '--email', 'a@example.test',
|
|
'--celular', '16999999999')
|
|
self.assertEqual(code, 0)
|
|
self.assertIn('"numeroOrdemCompra": "DTF-TESTE-', text)
|
|
self.assertIn('Nada foi criado', text)
|
|
self.assertEqual(self.orders, [])
|
|
self.assertTrue(all(method == 'GET' for method, _ in self.calls))
|
|
|
|
def test_confirmed_order_is_created_once_and_found_on_resend(self):
|
|
code, text = self.probe('pedido', '--cnpj', '11222333000181', '--email', 'a@example.test',
|
|
'--celular', '16999999999', '--modo', 'uv', '--confirmar')
|
|
self.assertEqual(code, 0, text)
|
|
self.assertEqual(len(self.orders), 1)
|
|
self.assertEqual(self.orders[0]['itens'][0]['produto'], {'id': 104})
|
|
self.assertIn('2º envio: already-created', text)
|
|
|
|
def test_resend_is_skipped_when_the_search_cannot_find_the_order(self):
|
|
from app import tiny_probe
|
|
blind = lambda request: (httpx.Response(200, json={'itens': []})
|
|
if request.method == 'GET' and request.url.path.endswith('/pedidos')
|
|
else self.handler(request))
|
|
self.tiny = TinyOrders(auth=FakeAuth(), transport=httpx.MockTransport(blind), today=date(2026, 9, 24))
|
|
payload = tiny_probe.test_order('11222333000181', 'a@example.test', '16999999999', 'file')
|
|
lines = []
|
|
self.assertFalse(tiny_probe.pedido(self.tiny, payload, True, lines.append, wait=lambda s: None))
|
|
self.assertEqual(len(self.orders), 1)
|
|
self.assertIn('o 2º envio não foi feito', lines[-1])
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|