"""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 from datetime import datetime, timezone 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-1') 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')) # The code expires in 30 minutes, in the format Mercado Pago documents. expires = datetime.fromisoformat(body['date_of_expiration']) self.assertRegex(body['date_of_expiration'], r'^\d{4}-\d\d-\d\dT\d\d:\d\d:\d\d\.\d{3}-03:00$') self.assertAlmostEqual((expires - datetime.now(timezone.utc)).total_seconds(), 1800, delta=60) self.assertEqual(created['expires_at'], body['date_of_expiration']) # A new code after the last expired is the next attempt, not the same payment. self.mp.create('11111111-2222-3333-4444-555555555555', 12345, {'mail': 'a@example.test', 'cnpj': '11222333000181'}, {'type': 'pix', 'attempt': 2}) self.assertTrue(self.requests[-1].headers['x-idempotency-key'].endswith('-pix-2')) 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-')) # 3-D Secure is asked of debit only; the cardholder is the payer. self.assertNotIn('three_d_secure_mode', body) self.mp.create('q', 1000, {'mail': 'a@example.test', 'cnpj': '11222333000181'}, {'type': 'card', 'token': 'tok_deb', 'payment_method_id': 'debmaster', 'payer_document_type': 'CPF', 'payer_document': '12345678909'}) debit = json.loads(self.requests[-1].content) self.assertEqual(debit['three_d_secure_mode'], 'optional') self.assertEqual(debit['payer']['identification'], {'type': 'CPF', 'number': '12345678909'}) self.assertEqual(body['payer']['identification'], {'type': 'CNPJ', 'number': '11222333000181'}) self.mp.create('q', 1000, {'mail': 'a@example.test', 'cnpj': '11222333000181'}, {'type': 'card', 'token': 'tok_iss', 'payment_method_id': 'master', 'issuer_id': '24'}) self.assertNotIn('issuer_id', json.loads(self.requests[-1].content)) 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()