aec5b1d054
All checks were successful
Build and deploy / Validate source (push) Successful in 5s
Build and deploy / Integration suite on a real stack (push) Successful in 2m5s
Build and deploy / Secret scan and release gate (push) Successful in 5s
Build and deploy / Publish images (push) Successful in 1m45s
The client already sends WhatsApp notices from Tiny's order situação (Tiny webhook -> middleware -> n8n). With TINY_STATUS_UPDATES on, a paid order is set to "Aprovada" once and a finished pickup order to "Pronto para envio"; pickup orders carry the client's pickup forma de envio (TINY_FORMA_ENVIO_RETIRADA). The ready event now carries the order and the Tiny id from the sale's receipt. Off by default until go-live, when n8n stops sending the DTFIMP designer message. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
297 lines
16 KiB
Python
297 lines
16 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', 'TINY_FORMA_ENVIO_RETIRADA': '77'}
|
|
PICKUP = {**PAID, 'order': {**PAID['order'], 'destination': None, 'freight': {'service': 'pickup', 'total_cents': 0}}}
|
|
READY = {'order_id': PAID['order_id'], 'number': 42, 'event': 'ready', 'reason': '', 'order': PICKUP['order']}
|
|
|
|
|
|
class FakeAuth:
|
|
def access_token(self):
|
|
return 'access-1'
|
|
|
|
|
|
class FakeTinyCase(unittest.TestCase):
|
|
"""A Tiny account in memory: contacts, orders, products and formas de envio."""
|
|
|
|
def setUp(self):
|
|
self.contacts = []
|
|
self.orders = []
|
|
self.calls = []
|
|
self.refuse_status = 0
|
|
self.methods = {'77': {'id': 77, 'nome': 'Retirar pessoalmente', 'tipo': '6'},
|
|
'78': {'id': 78, 'nome': 'Correios', 'tipo': '1'}}
|
|
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.startswith('/formas-envio/'):
|
|
wanted = path.rsplit('/', 1)[-1]
|
|
if wanted not in self.methods:
|
|
return httpx.Response(404, json={'mensagem': 'não encontrado'})
|
|
return httpx.Response(200, json=self.methods[wanted])
|
|
if request.method == 'GET' and path == '/formas-envio':
|
|
return httpx.Response(200, json={'itens': list(self.methods.values())})
|
|
if request.method == 'PUT' and path.startswith('/pedidos/') and path.endswith('/situacao'):
|
|
if self.refuse_status:
|
|
self.refuse_status -= 1
|
|
return httpx.Response(503, text='indisponível')
|
|
wanted = int(path.split('/')[2])
|
|
next(o for o in self.orders if o['id'] == wanted)['situacao'] = json.loads(request.content)['situacao']
|
|
return httpx.Response(204)
|
|
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 = {'situacao': 0, **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), 7)
|
|
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 TinyStatusTests(FakeTinyCase):
|
|
"""Situações that the client's Tiny -> n8n notices turn into WhatsApp messages."""
|
|
|
|
def puts(self):
|
|
return [call for call in self.calls if call[0] == 'PUT']
|
|
|
|
def test_nothing_changes_while_status_updates_are_off(self):
|
|
self.assertEqual(self.tiny.deliver('k1', PICKUP)['status'], 'created')
|
|
self.assertEqual(self.orders[0]['situacao'], 0)
|
|
self.assertEqual(self.tiny.deliver('k2', READY)['status'], 'not-applicable')
|
|
self.assertEqual(self.puts(), [])
|
|
|
|
@mock.patch.dict(os.environ, {'TINY_STATUS_UPDATES': 'true'})
|
|
def test_paid_order_is_approved_once(self):
|
|
first = self.tiny.deliver('k1', PICKUP)
|
|
self.assertEqual((first['status'], first['situacao']), ('created', 'Aprovada'))
|
|
self.assertEqual(self.orders[0]['situacao'], 3)
|
|
second = self.tiny.deliver('k1', PICKUP)
|
|
self.assertEqual(second['status'], 'already-created')
|
|
self.assertNotIn('situacao', second)
|
|
self.assertEqual(len(self.puts()), 1)
|
|
|
|
@mock.patch.dict(os.environ, {'TINY_STATUS_UPDATES': 'true'})
|
|
def test_retry_after_a_failed_approval_finishes_it(self):
|
|
self.refuse_status = 1
|
|
with self.assertRaises(TinyError):
|
|
self.tiny.deliver('k1', PICKUP)
|
|
self.assertEqual((len(self.orders), self.orders[0]['situacao']), (1, 0))
|
|
retry = self.tiny.deliver('k1', PICKUP)
|
|
self.assertEqual((retry['status'], retry['situacao']), ('already-created', 'Aprovada'))
|
|
self.assertEqual(self.calls.count(('POST', '/pedidos')), 1)
|
|
|
|
@mock.patch.dict(os.environ, {'TINY_STATUS_UPDATES': 'true'})
|
|
def test_an_order_already_moved_on_is_not_approved_again(self):
|
|
self.tiny.deliver('k1', PICKUP)
|
|
self.orders[0]['situacao'] = 7
|
|
self.tiny.deliver('k1', PICKUP)
|
|
self.assertEqual(self.orders[0]['situacao'], 7)
|
|
|
|
@mock.patch.dict(os.environ, {'TINY_STATUS_UPDATES': 'true'})
|
|
def test_ready_pickup_order_is_set_ready_once_by_its_known_id(self):
|
|
sale = self.tiny.deliver('k1', PICKUP)
|
|
self.calls.clear()
|
|
ready = self.tiny.deliver('k2', {**READY, 'tiny_id': sale['tiny_id']})
|
|
self.assertEqual((ready['status'], ready['situacao'], ready['tiny_number']),
|
|
('status-updated', 'Pronto para envio', sale['tiny_number']))
|
|
self.assertEqual(self.orders[0]['situacao'], 7)
|
|
self.assertNotIn(('GET', '/pedidos'), self.calls, 'a known id needs no search')
|
|
again = self.tiny.deliver('k2', {**READY, 'tiny_id': sale['tiny_id']})
|
|
self.assertEqual(again['status'], 'status-unchanged')
|
|
self.assertEqual(len(self.puts()), 1)
|
|
|
|
@mock.patch.dict(os.environ, {'TINY_STATUS_UPDATES': 'true'})
|
|
def test_ready_without_an_id_finds_the_order(self):
|
|
self.tiny.deliver('k1', PICKUP)
|
|
self.assertEqual(self.tiny.deliver('k2', READY)['status'], 'status-updated')
|
|
self.assertEqual(self.orders[0]['situacao'], 7)
|
|
|
|
@mock.patch.dict(os.environ, {'TINY_STATUS_UPDATES': 'true'})
|
|
def test_ready_before_the_sale_reached_tiny_is_retried(self):
|
|
with self.assertRaisesRegex(TinyError, 'not in Tiny yet'):
|
|
self.tiny.deliver('k2', READY)
|
|
self.assertEqual(self.puts(), [])
|
|
|
|
@mock.patch.dict(os.environ, {'TINY_STATUS_UPDATES': 'true'})
|
|
def test_ready_delivery_order_waits_for_freight(self):
|
|
self.tiny.deliver('k1', PAID)
|
|
self.calls.clear()
|
|
ready = self.tiny.deliver('k2', {**READY, 'order': PAID['order']})
|
|
self.assertEqual(ready['status'], 'not-applicable')
|
|
self.assertEqual(self.calls, [])
|
|
|
|
def test_pickup_orders_carry_the_pickup_forma_de_envio(self):
|
|
self.assertEqual(order_payload(PICKUP, 1)['transportador'], {'formaEnvio': {'id': 77}})
|
|
self.assertNotIn('transportador', order_payload(PAID, 1))
|
|
with mock.patch.dict(os.environ, {'TINY_FORMA_ENVIO_RETIRADA': ''}):
|
|
self.assertNotIn('transportador', order_payload(PICKUP, 1))
|
|
|
|
|
|
@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.assertIn('TINY_FORMA_ENVIO_RETIRADA (retirada): ok Retirar pessoalmente', text)
|
|
code, text = self.probe('formas-envio')
|
|
self.assertIn('77\t6\tRetirar pessoalmente', 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()
|