e3d5558198
All checks were successful
Build and deploy / Validate source (push) Successful in 9s
Build and deploy / Integration suite on a real stack (push) Successful in 2m49s
Build and deploy / Secret scan and release gate (push) Successful in 9s
Build and deploy / Publish images and notify Portainer (push) Has been skipped
Tiny v3 replaces the v2 token adapter. An operator connects Tiny once from the Kanban; the callback is authorised by a single-use state, because Tiny's cross-site redirect does not carry the SameSite=Strict operator cookie. Tokens are kept in provider_tokens, the refresh token rotates under a row lock, and the worker keeps the connection alive while order creation is off. Orders find or create the customer's contact by CNPJ, then POST /pedidos with product ids from TINY_PRODUCT_TEXTIL_FOLHA, _TEXTIL_AVULSA, _UV_FOLHA and _UV_AVULSA and numeroOrdemCompra DTF-<number>; a retry searches the customer's recent orders for that number first. The product settings avoid a _FILE suffix, which the secrets loader reads as a secret file path. Production passes the application credentials through but keeps TINY_ADAPTER fake: Tiny has no sandbox, so creating real orders waits for a supervised test. compose.providers.yaml gives the local API and worker an internet route for provider testing; the default local stack still has none. Verified with the full CI integration sequence locally, including the new tiny_oauth_test against the real database. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
106 lines
4.6 KiB
Python
106 lines
4.6 KiB
Python
"""The Tiny OAuth connection against the real database, with a fake token server.
|
|
|
|
Run inside the API container: python -m tests.tiny_oauth_test
|
|
It saves any existing Tiny connection first and restores it afterwards.
|
|
"""
|
|
import os
|
|
from datetime import datetime, timedelta, timezone
|
|
from urllib.parse import parse_qs, urlparse
|
|
|
|
import httpx
|
|
|
|
from app.core.db import connect
|
|
from app.tiny import TinyAuth, TinyError, TinyNotConnected
|
|
|
|
os.environ.update(TINY_CLIENT_ID='test-client', TINY_CLIENT_SECRET='test-secret',
|
|
TINY_REDIRECT_URI='http://localhost:8081/api/operator/tiny/callback')
|
|
|
|
|
|
def run():
|
|
with connect() as c:
|
|
saved = c.execute("SELECT * FROM dtf_local.provider_tokens WHERE provider='tiny'").fetchone()
|
|
c.execute("DELETE FROM dtf_local.provider_tokens WHERE provider='tiny'")
|
|
try:
|
|
exercise()
|
|
finally:
|
|
with connect() as c:
|
|
c.execute("DELETE FROM dtf_local.provider_tokens WHERE provider='tiny'")
|
|
if saved:
|
|
columns = ','.join(saved)
|
|
c.execute(f'INSERT INTO dtf_local.provider_tokens({columns}) VALUES({",".join(["%s"] * len(saved))})',
|
|
tuple(saved.values()))
|
|
|
|
|
|
def exercise():
|
|
issued = []
|
|
|
|
def token_server(request):
|
|
form = {k: v[0] for k, v in parse_qs(request.content.decode()).items()}
|
|
assert (form['client_id'], form['client_secret']) == ('test-client', 'test-secret')
|
|
if form['grant_type'] == 'authorization_code':
|
|
assert form['code'] == 'good-code' and form['redirect_uri'].endswith('/tiny/callback')
|
|
elif form['grant_type'] == 'refresh_token':
|
|
if form['refresh_token'] != issued[-1]:
|
|
return httpx.Response(400, json={'error': 'invalid_grant'})
|
|
n = len(issued) + 1
|
|
issued.append(f'refresh-{n}')
|
|
return httpx.Response(200, json={'access_token': f'access-{n}', 'expires_in': 14400,
|
|
'refresh_token': f'refresh-{n}', 'refresh_expires_in': 86400})
|
|
|
|
auth = TinyAuth(transport=httpx.MockTransport(token_server))
|
|
assert auth.status() == {'connected': False}
|
|
try:
|
|
auth.access_token()
|
|
raise AssertionError('an unconnected Tiny must not yield a token')
|
|
except TinyNotConnected:
|
|
pass
|
|
|
|
url = urlparse(auth.authorize_url('operator@example.test'))
|
|
query = {k: v[0] for k, v in parse_qs(url.query).items()}
|
|
assert url.netloc == 'accounts.tiny.com.br' and query['client_id'] == 'test-client'
|
|
assert query['redirect_uri'].endswith('/api/operator/tiny/callback') and len(query['state']) > 30
|
|
try:
|
|
auth.complete('good-code', 'forged-state')
|
|
raise AssertionError('a state no operator created must be refused')
|
|
except TinyError:
|
|
pass
|
|
assert auth.complete('good-code', query['state']) == 'operator@example.test'
|
|
try:
|
|
auth.complete('good-code', query['state'])
|
|
raise AssertionError('a state must be single-use')
|
|
except TinyError:
|
|
pass
|
|
status = auth.status()
|
|
assert status['connected'] and status['connected_by'] == 'operator@example.test'
|
|
assert auth.access_token() == 'access-1'
|
|
print('PASS: operator-started state is required, single-use, and stores the connection')
|
|
|
|
# An access token about to expire is refreshed, and the refresh token rotates.
|
|
with connect() as c:
|
|
c.execute("UPDATE dtf_local.provider_tokens SET access_expires_at=now() WHERE provider='tiny'")
|
|
assert auth.access_token() == 'access-2'
|
|
with connect() as c:
|
|
row = c.execute("SELECT refresh_token,connected_by FROM dtf_local.provider_tokens WHERE provider='tiny'").fetchone()
|
|
assert row == {'refresh_token': 'refresh-2', 'connected_by': 'operator@example.test'}
|
|
assert auth.access_token() == 'access-2'
|
|
print('PASS: expiring access token refreshed once, refresh token rotated and stored')
|
|
|
|
# A refused refresh (revoked in Tiny) asks for a new connection, not a retry loop.
|
|
with connect() as c:
|
|
c.execute("""UPDATE dtf_local.provider_tokens SET access_expires_at=now(), refresh_token='revoked'
|
|
WHERE provider='tiny'""")
|
|
try:
|
|
auth.access_token()
|
|
raise AssertionError('a refused refresh must report the connection as lost')
|
|
except TinyNotConnected:
|
|
pass
|
|
with connect() as c:
|
|
c.execute("UPDATE dtf_local.provider_tokens SET refresh_expires_at=%s WHERE provider='tiny'",
|
|
(datetime.now(timezone.utc) - timedelta(seconds=1),))
|
|
assert not auth.status()['connected']
|
|
print('PASS: revoked or expired connections report that Tiny must be connected again')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
run()
|