dtf-system/tests/test_mercadopago.py

"""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]
                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_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})
        body = json.loads(self.requests[-1].content)
        self.assertEqual((body['token'], body['payment_method_id'], body['installments']), ('tok_abc', 'visa', 3))
        self.assertNotIn('card_number', json.dumps(body))


if __name__ == '__main__':
    unittest.main()