eae3dad306
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 2m3s
Build and deploy / Secret scan and release gate (push) Successful in 5s
Build and deploy / Publish images (push) Successful in 1m31s
The payment page lists credit card (preselected), debit card and PIX. Each card option limits Mercado Pago's form to its kind; debit is paid at once. Card payments ask for 3-D Secure when the issuer requires it, and a challenge opens the bank's page in a frame, which needs PAYMENT_CHALLENGE_SOURCES=https: (frames and form posts only). A card left waiting for that confirmation stops blocking a new attempt after ten minutes, and a refusal reported by the notification returns the customer to the payment choice. Written from the documentation; not yet run with a real debit card. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
150 lines
8.7 KiB
Python
150 lines
8.7 KiB
Python
"""The Mercado Pago adapter against a fake HTTP transport.
|
|
|
|
This proves the adapter follows the documented contract. It does not prove the
|
|
integration: that needs the sandbox flows with the client's own account.
|
|
Runs where httpx is installed (the API image, or a local virtualenv).
|
|
"""
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
import unittest
|
|
|
|
import httpx
|
|
|
|
from app.mercadopago import MercadoPagoPayment, event_from_payment
|
|
|
|
SECRET = 'test-webhook-secret'
|
|
NOW = 1_790_000_000
|
|
|
|
|
|
def signature(data_id, request_id, ts, secret=SECRET):
|
|
manifest = ''
|
|
if data_id:
|
|
manifest += f'id:{data_id};'
|
|
if request_id:
|
|
manifest += f'request-id:{request_id};'
|
|
manifest += f'ts:{ts};'
|
|
return f'ts={ts},v1=' + hmac.new(secret.encode(), manifest.encode(), hashlib.sha256).hexdigest()
|
|
|
|
|
|
class MercadoPagoTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.requests = []
|
|
self.payments = {}
|
|
|
|
def handler(request):
|
|
self.requests.append(request)
|
|
if request.method == 'GET':
|
|
payment_id = request.url.path.rsplit('/', 1)[-1]
|
|
if payment_id == '500':
|
|
return httpx.Response(500, json={'message': 'internal_error'})
|
|
if payment_id not in self.payments:
|
|
return httpx.Response(404, json={'message': 'Payment not found'})
|
|
return httpx.Response(200, json=self.payments[payment_id])
|
|
body = json.loads(request.content)
|
|
payment = {'id': 555, 'status': 'pending', 'status_detail': 'pending_waiting_transfer',
|
|
'point_of_interaction': {'transaction_data': {
|
|
'qr_code': '000201PIX', 'qr_code_base64': 'aW1n', 'ticket_url': 'https://mp/t'}},
|
|
**{k: body[k] for k in ('transaction_amount', 'external_reference')}}
|
|
return httpx.Response(201, json=payment)
|
|
|
|
self.mp = MercadoPagoPayment('TEST-token', SECRET, 'https://dtf.example/api/payments/webhook',
|
|
transport=httpx.MockTransport(handler), clock=lambda: NOW)
|
|
|
|
def test_signature_follows_the_documented_manifest(self):
|
|
headers = {'x-signature': signature('123456', 'req-1', NOW), 'x-request-id': 'req-1'}
|
|
self.assertTrue(self.mp.verify(headers, b'{}', {'data.id': '123456'}))
|
|
# Any change to the signed values breaks it.
|
|
self.assertFalse(self.mp.verify(headers, b'{}', {'data.id': '123457'}))
|
|
self.assertFalse(self.mp.verify({**headers, 'x-request-id': 'req-2'}, b'{}', {'data.id': '123456'}))
|
|
self.assertFalse(self.mp.verify({'x-signature': signature('123456', 'req-1', NOW, 'other'),
|
|
'x-request-id': 'req-1'}, b'{}', {'data.id': '123456'}))
|
|
self.assertFalse(self.mp.verify({}, b'{}', {'data.id': '123456'}))
|
|
|
|
def test_absent_values_are_left_out_and_alphanumeric_ids_lowercased(self):
|
|
self.assertTrue(self.mp.verify({'x-signature': signature('abc123', None, NOW)}, b'{}',
|
|
{'data.id': 'ABC123'}))
|
|
|
|
def test_old_signatures_are_refused(self):
|
|
old = NOW - 3600
|
|
self.assertFalse(self.mp.verify({'x-signature': signature('1', 'r', old), 'x-request-id': 'r'},
|
|
b'{}', {'data.id': '1'}))
|
|
|
|
def test_notification_is_only_a_pointer(self):
|
|
# The body claims nothing about amount or status; the API is asked.
|
|
self.payments['999'] = {'id': 999, 'status': 'approved', 'currency_id': 'BRL',
|
|
'transaction_amount': 123.45, 'external_reference': 'quote-1'}
|
|
body = json.dumps({'id': 42, 'type': 'payment', 'action': 'payment.updated',
|
|
'data': {'id': '999'}}).encode()
|
|
event = self.mp.parse(body, {'data.id': '999', 'type': 'payment'})
|
|
self.assertEqual((event.status, event.amount_cents, event.reference), ('approved', 12345, 'quote-1'))
|
|
self.assertEqual(event.event_id, '999:approved')
|
|
self.assertEqual(self.requests[-1].headers['authorization'], 'Bearer TEST-token')
|
|
self.assertIsNone(self.mp.parse(json.dumps({'type': 'merchant_order', 'data': {'id': '1'}}).encode()))
|
|
|
|
def test_notification_for_an_unknown_payment_is_acknowledged(self):
|
|
# The panel's "Simular notificação" sends a payment id that does not
|
|
# exist. Raising would answer 500 and Mercado Pago would retry for ever.
|
|
body = json.dumps({'id': 43, 'type': 'payment', 'data': {'id': '123456'}}).encode()
|
|
self.assertIsNone(self.mp.parse(body, {'data.id': '123456', 'type': 'payment'}))
|
|
# Any other failure still raises, so a real notification is retried.
|
|
with self.assertRaises(httpx.HTTPStatusError):
|
|
self.mp.parse(json.dumps({'type': 'payment', 'data': {'id': '500'}}).encode(), {'data.id': '500'})
|
|
|
|
def test_amounts_outside_brl_centavos_are_not_trusted(self):
|
|
for payment in ({'currency_id': 'USD', 'transaction_amount': 10},
|
|
{'currency_id': 'BRL', 'transaction_amount': 10.001},
|
|
{'currency_id': 'BRL', 'transaction_amount': None}):
|
|
self.assertIsNone(event_from_payment({'id': 1, 'status': 'approved', **payment}).amount_cents)
|
|
self.assertEqual(event_from_payment({'id': 1, 'status': 'approved', 'currency_id': 'BRL',
|
|
'transaction_amount': 0.1}).amount_cents, 10)
|
|
|
|
def test_statuses_map_to_the_service_vocabulary(self):
|
|
for provider, ours in (('in_process', 'pending'), ('charged_back', 'refunded'),
|
|
('cancelled', 'cancelled'), ('rejected', 'rejected')):
|
|
self.assertEqual(event_from_payment({'id': 1, 'status': provider}).status, ours)
|
|
|
|
def test_pix_payment_is_idempotent_on_the_quote(self):
|
|
created = self.mp.create('11111111-2222-3333-4444-555555555555', 12345,
|
|
{'mail': 'a@example.test', 'cnpj': '11222333000181'})
|
|
request = self.requests[-1]
|
|
body = json.loads(request.content)
|
|
self.assertEqual(request.headers['x-idempotency-key'],
|
|
'dtf-quote-11111111-2222-3333-4444-555555555555-pix')
|
|
self.assertEqual((body['payment_method_id'], body['transaction_amount']), ('pix', 123.45))
|
|
self.assertEqual(body['external_reference'], '11111111-2222-3333-4444-555555555555')
|
|
self.assertEqual(body['notification_url'], 'https://dtf.example/api/payments/webhook')
|
|
self.assertEqual((created['pix_qr_code'], created['status']), ('000201PIX', 'pending'))
|
|
|
|
def test_card_payment_uses_the_browser_token_only(self):
|
|
self.mp.create('q', 1000, {'mail': 'a@example.test', 'cnpj': '11222333000181'},
|
|
{'type': 'card', 'token': 'tok_abc', 'payment_method_id': 'visa', 'installments': 3})
|
|
request = self.requests[-1]
|
|
body = json.loads(request.content)
|
|
self.assertEqual((body['token'], body['payment_method_id'], body['installments']), ('tok_abc', 'visa', 3))
|
|
self.assertNotIn('card_number', json.dumps(body))
|
|
# A new card attempt after a decline must not collide with the first.
|
|
first_key = request.headers['x-idempotency-key']
|
|
self.mp.create('q', 1000, {'mail': 'a@example.test', 'cnpj': '11222333000181'},
|
|
{'type': 'card', 'token': 'tok_def', 'payment_method_id': 'visa'})
|
|
self.assertNotEqual(self.requests[-1].headers['x-idempotency-key'], first_key)
|
|
self.assertTrue(first_key.startswith('dtf-quote-q-card-'))
|
|
self.assertEqual(body['three_d_secure_mode'], 'optional')
|
|
self.assertIsNone(self.mp.create('q', 1000, {'mail': 'a@example.test', 'cnpj': '11222333000181'},
|
|
{'type': 'card', 'token': 'tok_ghi', 'payment_method_id': 'visa'})['challenge'])
|
|
|
|
def test_a_card_the_bank_must_confirm_returns_its_challenge(self):
|
|
def handler(request):
|
|
return httpx.Response(201, json={'id': 777, 'status': 'pending', 'status_detail': 'pending_challenge',
|
|
'three_ds_info': {'external_resource_url': 'https://acs.bank.example/challenge',
|
|
'creq': 'eyJjcmVxIjoiMSJ9'}})
|
|
mp = MercadoPagoPayment('TEST-token', SECRET, transport=httpx.MockTransport(handler), clock=lambda: NOW)
|
|
created = mp.create('q', 1000, {'mail': 'a@example.test', 'cnpj': '11222333000181'},
|
|
{'type': 'card', 'token': 'tok_debit', 'payment_method_id': 'debvisa'})
|
|
self.assertEqual((created['status'], created['status_detail']), ('pending', 'pending_challenge'))
|
|
self.assertEqual(created['challenge'], {'url': 'https://acs.bank.example/challenge', 'creq': 'eyJjcmVxIjoiMSJ9'})
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|