Ce module ajoute Payplug comme +fournisseur de paiement au cadre de paiement d’Odoo (module +payment).
+Le client est redirigé vers la page de paiement hébergée par Payplug, +puis revient sur Odoo. Le module gère :
+Payplug n’accepte que l’euro : le fournisseur n’est proposé que pour les +paiements en EUR.
+Le module appelle directement l’API REST de Payplug avec requests ; +il ne demande aucune bibliothèque Python supplémentaire.
+Avertissement : ce module est un développement indépendant. Il n’est +ni affilié à Payplug, ni approuvé par Payplug. « Payplug » est une +marque de son propriétaire.
+Table of contents
+Le module dépend des modules standard payment et account_payment +(installés avec la Facturation ou le site e-commerce). Il ne demande +aucune bibliothèque Python supplémentaire.
+Le fournisseur Payplug est créé désactivé, pour chaque société.
+À la désinstallation, comme pour les fournisseurs standard, Odoo refuse +de continuer si des paiements comptables ont été faits avec Payplug. Un +fournisseur qui n’a que des transactions sans paiement est conservé +(désactivé, sans code) pour garder ces transactions lisibles ; sinon il +est supprimé.
+Clés Payplug
+Le module accepte deux méthodes d’authentification, au choix dans le +champ Authentification de l’onglet Identifiants :
+Dans le portail Payplug, à la rubrique des clés d’API, générez une +clé en environnement Test et, plus tard, une clé en environnement +Live :
+La clé publique n’est pas utilisée : le module ne passe que par la +page de paiement hébergée.
+Dans Odoo, ouvrez Facturation (ou Comptabilité) > Configuration > +Fournisseurs de paiement et choisissez Payplug.
+Dans l’onglet Identifiants, choisissez la méthode +d’authentification, puis collez la clé secrète (Clé API) ou le +client ID et le client secret (OAuth2). Seuls les +administrateurs (groupe Paramètres) peuvent les lire ou les +modifier.
+Réglez l’état :
+Pour une ancienne clé sk_..., Odoo refuse une clé de production +en mode test et une clé de test sur un fournisseur activé. Pour une +clé JWT, l’accord entre le mode et la clé est vérifié à chaque +paiement : un paiement dont le mode ne correspond pas est refusé. En +OAuth2, l’environnement du jeton (scope test) est en plus +vérifié à chaque requête : en mode test, seul un jeton de +l’environnement Test est accepté ; sur un fournisseur activé, un +jeton de test est refusé.
+Pour passer de Test à Live en OAuth2, remplacez le client ID et le +client secret par ceux de la clé Live en même temps que vous +activez le fournisseur.
+Dans l’onglet Configuration, vérifiez le journal de paiement et les +moyens de paiement. Par défaut, seul Carte (CB, Visa, Mastercard) +est activé. Bancontact, iDEAL et MyBank sont proposés, mais ne les +activez qu’après les avoir fait ouvrir sur votre compte Payplug.
+Publiez le fournisseur pour qu’il apparaisse sur le site web.
+Adresse de notification (webhook)
+Le module donne à Payplug, pour chaque paiement, l’adresse +https://<votre-domaine>/payment/payplug/webhook. Il n’y a rien à +saisir dans le portail Payplug, mais cette adresse doit être joignable +depuis Internet en HTTPS :
+Les retours du client passent par /payment/payplug/return et +/payment/payplug/cancel.
+Jetons d’accès OAuth2
+Un jeton dure 5 minutes ; Odoo le garde en mémoire et en demande un +nouveau une minute avant son expiration. Les jetons ne sont jamais +écrits en base ni dans les journaux ; chaque processus (worker) d’Odoo +demande les siens. Si Payplug refuse un jeton (HTTP 401), Odoo l’oublie, +en demande un autre et renvoie la requête une seule fois. Après une +modification des identifiants, chaque processus demande un nouveau jeton +dès sa requête suivante ; après un changement d’état, le jeton en +mémoire est de nouveau confronté au mode du fournisseur.
+Paramètres système réservés aux tests
+Ils ne servent qu’à pointer vers un faux serveur de test ; ne les créez +pas en production.
+Paiement sur le site web ou depuis le portail
+Le client choisit Carte (ou un autre moyen activé), est redirigé vers +la page Payplug, paie, puis revient sur la page d’état du paiement +d’Odoo. La commande ou la facture est confirmée dès qu’Odoo a relu le +paiement auprès de Payplug, soit au retour du client, soit à la +réception de la notification.
+Si le client clique sur Annuler sur la page Payplug, le paiement est +abandonné chez Payplug et la transaction passe à l’état Annulée. Un +paiement refusé passe à l’état Erreur avec le motif donné par Payplug +; un paiement jamais terminé est annulé quand Payplug signale son +expiration (environ 15 minutes).
+Remboursement
+Depuis le paiement (Comptabilité > Clients > Paiements), ou depuis la +facture via son paiement, cliquez sur Rembourser et indiquez le +montant (total ou partiel). Odoo demande le remboursement à Payplug, +crée une transaction de remboursement liée à la transaction d’origine, +puis le paiement comptable négatif correspondant.
+Un remboursement fait directement dans le portail Payplug est aussi +enregistré dans Odoo, à la réception de sa notification.
+Payplug n’accepte pas de remboursement inférieur à 0,10 €.
+Vérifier une transaction
+Sur une transaction Payplug en attente ou en erreur, le bouton Vérifier +le statut sur Payplug relit le paiement auprès de Payplug et met la +transaction à jour (utile si une notification n’a pas pu arriver).
+Les tests automatiques (--test-tags /payment_payplug) n’appellent +jamais l’API Payplug : ils démarrent un faux serveur Payplug local +(tests/fake_payplug.py) et y font pointer le module par le paramètre +système payment_payplug.api_base_url.
+Le même faux serveur sert aux essais manuels de bout en bout, avec une +page de paiement factice (boutons Payer, Refuser, Annuler) et +l’envoi des notifications :
++python3 payment_payplug/tests/fake_payplug.py --host 127.0.0.1 --port 8765 ++
Sur une base de test, créez alors le paramètre système +payment_payplug.api_base_url avec la valeur +http://127.0.0.1:8765/v1/, et saisissez une clé quelconque +commençant par sk_test_. POST /_control/refund/<id du paiement> +(corps {"amount": 1000}) simule un remboursement fait depuis le +portail Payplug.
+Pour essayer la méthode OAuth2, le faux serveur répond aussi sur +POST /oauth2/token (authentification Basic, +grant_type=client_credentials, audience https://www.payplug.com) +avec deux clients factices : fake_client_test / +fake_client_secret_test (jetons de test) et fake_client_live / +fake_client_secret_live (jetons Live). L’adresse du point de jeton +se déduit de payment_payplug.api_base_url ; il n’y a rien d’autre à +régler.
+À vérifier avec de vraies clés de test Payplug (non vérifié, faute de +compte) :
+Bugs are tracked on GitHub Issues. +In case of trouble, please check there if your issue has already been reported. +If you spotted it first, help us to smash it by providing a detailed and welcomed +feedback.
+Do not contact contributors directly about support or help with technical issues.
+This module is maintained by the OCA.
+ +
+
+OCA, or the Odoo Community Association, is a nonprofit organization whose +mission is to support the collaborative development of Odoo features and +promote its widespread use.
+Current maintainer:
+
This module is part of the OCA/e-commerce project on GitHub.
+You are welcome to contribute. To learn how please visit https://odoo-community.org/page/Contribute.
+Paiement {title} : {amount}
+""" + body = page.encode() + self.send_response(200) + self.send_header("Content-Type", "text/html; charset=utf-8") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + return None + action = parse_qs(raw.decode()).get("action", [""])[0] + target = payment["hosted_payment"]["return_url"] + if action == "pay": + self.fake.mark_paid(payment["id"]) + elif action == "refuse": + self.fake.mark_failed(payment["id"]) + else: + target = payment["hosted_payment"]["cancel_url"] + if action in ("pay", "refuse") and self.fake.notify: + self.fake.send_notification( + { + "id": payment["id"], + "object": "payment", + "is_live": payment["is_live"], + } + ) + self.send_response(303) + self.send_header("Location", target) + self.end_headers() + return None + + +def main(): + parser = argparse.ArgumentParser(description=__doc__.split("\n", 1)[0]) + parser.add_argument("--host", default="127.0.0.1") + parser.add_argument("--port", type=int, default=8765) + parser.add_argument( + "--public-url", help="URL of the hosted pages seen by the browser" + ) + args = parser.parse_args() + logging.basicConfig(level=logging.DEBUG) + server = FakePayplugServer(args.host, args.port, args.public_url, notify=True) + _logger.info( + "Fake Payplug API on %s (pages on %s)", server.api_url, server.public_url + ) + try: + server.httpd.serve_forever() + except KeyboardInterrupt: + server.stop() + + +if __name__ == "__main__": + main() diff --git a/payment_payplug/tests/test_payplug.py b/payment_payplug/tests/test_payplug.py new file mode 100644 index 0000000000..f1cda9954f --- /dev/null +++ b/payment_payplug/tests/test_payplug.py @@ -0,0 +1,495 @@ +# Copyright 2026 Sébastien Berger (sbinfo67) +# License LGPL-3.0 or later (https://www.gnu.org/licenses/lgpl-3.0). + +import json +from unittest.mock import patch +from urllib.parse import parse_qs, urlsplit + +import requests + +from odoo.exceptions import AccessError, ValidationError +from odoo.tests import tagged +from odoo.tools import mute_logger + +from .. import const +from .common import PayplugCommon + +TX_LOGGER = "odoo.addons.payment_payplug.models.payment_transaction" +LOGGERS = ( + TX_LOGGER, + "odoo.addons.payment.models.payment_provider", + "odoo.addons.payment.models.payment_transaction", +) + + +def _make_response(status_code, body): + """Return a `requests` response with the given status and body (bytes or JSON).""" + response = requests.models.Response() + response.status_code = status_code + response.encoding = "utf-8" + response._content = body if isinstance(body, bytes) else json.dumps(body).encode() + return response + + +@tagged("post_install", "-at_install", "payment_payplug") +class TestPayplugProvider(PayplugCommon): + def test_supported_currencies_are_eur_only(self): + self.assertEqual(self.payplug.available_currency_ids, self.currency_euro) + compatible = self.env["payment.provider"]._get_compatible_providers( + self.company.id, + self.partner.id, + self.amount, + currency_id=self.currency_euro.id, + ) + self.assertIn(self.payplug, compatible) + incompatible = self.env["payment.provider"]._get_compatible_providers( + self.company.id, + self.partner.id, + self.amount, + currency_id=self.currency_usd.id, + ) + self.assertNotIn(self.payplug, incompatible) + + def test_feature_support(self): + self.assertEqual(self.payplug.support_refund, "partial") + self.assertFalse(self.payplug.support_tokenization) + self.assertFalse(self.payplug.support_manual_capture) + + def test_secret_key_restricted_to_system_group(self): + field = self.env["payment.provider"]._fields["payplug_secret_key"] + self.assertEqual(field.groups, "base.group_system") + with self.assertRaises(AccessError): + self.payplug.with_user(self.internal_user).read(["payplug_secret_key"]) + + def test_secret_key_must_match_state(self): + with self.assertRaises(ValidationError): + self.payplug.write({"payplug_secret_key": "sk_live_whatever"}) + with self.assertRaises(ValidationError): + self.payplug.write({"state": "enabled"}) + self.payplug.write( + {"state": "enabled", "payplug_secret_key": "sk_live_whatever"} + ) + self.assertEqual(self.payplug.state, "enabled") + + def test_reset_credentials_clears_secret_key(self): + self.payplug.action_reset_credentials() + self.assertFalse(self.payplug.payplug_secret_key) + self.assertEqual(self.payplug.state, "disabled") + + def test_default_api_url(self): + self.env["ir.config_parameter"].sudo().set_param( + const.API_BASE_URL_PARAM, False + ) + self.assertEqual( + self.payplug._build_request_url("payments"), + "https://api.payplug.com/v1/payments", + ) + + def test_other_providers_are_not_affected(self): + other = self.dummy_provider + self.assertEqual(other._get_default_payment_method_codes(), set()) + self.assertEqual(other._build_request_url("payments"), "") + # The Payplug credentials are never sent to another provider. + self.assertEqual(other._build_request_headers("GET", "payments", None), {}) + response = _make_response( + 400, {"message": "Invalid parameters.", "details": {"amount": "Too low."}} + ) + self.assertEqual(other._parse_response_error(response), response.text) + self.assertEqual( + self.payplug._parse_response_error(response), + "Invalid parameters. {'amount': 'Too low.'}", + ) + + @mute_logger(*LOGGERS) + def test_error_response_without_json_object(self): + # E.g. the HTML page of a proxy, or an unexpected JSON value. + for body, expected in ( + (b"502 Bad Gateway", "502 Bad Gateway"), + (["unexpected"], '["unexpected"]'), + ({"object": "error"}, '{"object": "error"}'), + ): + with self.subTest(body=body): + response = _make_response(502, body) + self.assertEqual(self.payplug._parse_response_error(response), expected) + response = _make_response(502, b"502 Bad Gateway") + with ( + patch.object(requests, "request", return_value=response), + self.assertRaisesRegex(ValidationError, "502 Bad Gateway"), + ): + self.payplug._send_api_request("GET", "payments/pay_unknown") + + +@tagged("post_install", "-at_install", "payment_payplug") +class TestPayplugPayment(PayplugCommon): + def _last_request(self): + method, path, headers, body = self.fake_payplug.state.requests[-1] + return method, path, headers, json.loads(body) if body else {} + + def test_payment_creation_and_redirect_form(self): + tx, processing_values = self._create_payplug_payment() + + method, path, headers, payload = self._last_request() + self.assertEqual((method, path), ("POST", "/v1/payments")) + self.assertEqual( + headers["Authorization"], "Bearer sk_test_odoo_payplug_fake_key" + ) + self.assertEqual(headers["PayPlug-Version"], const.API_VERSION) + self.assertEqual(payload["amount"], 4990) + self.assertEqual(payload["currency"], "EUR") + self.assertEqual(payload["metadata"], {"odoo_reference": tx.reference}) + self.assertNotIn( + "payment_method", payload + ) # Card: the default of the payment page. + self.assertTrue( + payload["notification_url"].endswith("/payment/payplug/webhook") + ) + for key, route in (("return_url", "return"), ("cancel_url", "cancel")): + url = urlsplit(payload["hosted_payment"][key]) + self.assertEqual(url.path, f"/payment/payplug/{route}") + params = parse_qs(url.query) + self.assertEqual(params["reference"], [tx.reference]) + self.assertEqual( + params["access_token"], [self._generate_test_access_token(tx.reference)] + ) + self.assertEqual(payload["billing"]["email"], "norbert.buyer@example.com") + self.assertEqual(payload["billing"]["country"], "BE") + self.assertEqual(payload["billing"]["language"], "en") + self.assertEqual(payload["shipping"]["delivery_type"], "BILLING") + # "0032 12 34 56 78" is international (00 prefix), not a French mobile. + self.assertEqual(payload["billing"]["landline_phone_number"], "+3212345678") + self.assertNotIn("mobile_phone_number", payload["billing"]) + + payment = self.fake_payplug.get_payment(tx.provider_reference) + form = self._extract_values_from_html_form( + processing_values["redirect_form_html"] + ) + self.assertEqual(form["action"], payment["hosted_payment"]["payment_url"]) + self.assertEqual(form["method"], "get") + self.assertEqual(tx.state, "draft") + + def test_alternative_payment_method_is_sent(self): + bancontact = self.env.ref("payment.payment_method_bancontact") + bancontact.active = True + self._create_payplug_payment(payment_method_id=bancontact.id) + payload = self._last_request()[3] + self.assertEqual(payload["payment_method"], "bancontact") + + def test_phone_formatting(self): + tx_model = self.env["payment.transaction"] + self.assertEqual( + tx_model._payplug_format_phone("+33 6 12 34 56 78"), "+33612345678" + ) + self.assertEqual( + tx_model._payplug_format_phone("0033 3 88 12 34 56"), "+33388123456" + ) + # Local numbers are not sent: their country cannot be guessed safely. + self.assertIsNone(tx_model._payplug_format_phone("06 12 34 56 78")) + self.partner.phone = "+33 6 12 34 56 78" + self._create_payplug_payment() + billing = self._last_request()[3]["billing"] + self.assertEqual(billing["mobile_phone_number"], "+33612345678") + + @mute_logger(*LOGGERS) + def test_amount_below_minimum_is_refused_before_any_request(self): + tx, _values = self._create_payplug_payment(amount=0.29) + self.assertEqual(tx.state, "error") + self.assertFalse(self.fake_payplug.state.requests) + + @mute_logger(*LOGGERS) + def test_api_rejection_sets_error(self): + bancontact = self.env.ref("payment.payment_method_bancontact") + bancontact.active = True + # No local check for alternative methods: the (fake) API refuses the amount. + tx, _values = self._create_payplug_payment( + amount=30000.00, payment_method_id=bancontact.id + ) + self.assertEqual(tx.state, "error") + self.assertIn("Invalid parameters", tx.state_message) + + @mute_logger(*LOGGERS) + def test_paid_payment_confirms_transaction(self): + tx, _values = self._create_payplug_payment() + self.fake_payplug.mark_paid(tx.provider_reference, brand="Visa") + tx._payplug_fetch_and_process() + self.assertEqual(tx.state, "done") + self.assertEqual(tx.payment_method_id.code, "visa") + self.assertEqual( + self.fake_payplug.state.requests[-1][:2], + ("GET", f"/v1/payments/{tx.provider_reference}"), + ) + + @mute_logger(*LOGGERS) + def test_unpaid_payment_sets_pending(self): + tx, _values = self._create_payplug_payment() + tx._payplug_fetch_and_process() + self.assertEqual(tx.state, "pending") + + @mute_logger(*LOGGERS) + def test_failed_payment_sets_error(self): + tx, _values = self._create_payplug_payment() + self.fake_payplug.mark_failed( + tx.provider_reference, "card_declined", "Declined." + ) + tx._payplug_fetch_and_process() + self.assertEqual(tx.state, "error") + self.assertIn("card_declined", tx.state_message) + + @mute_logger(*LOGGERS) + def test_timed_out_payment_cancels_transaction(self): + tx, _values = self._create_payplug_payment() + self.fake_payplug.mark_failed(tx.provider_reference, "timeout", "Timeout.") + tx._payplug_fetch_and_process() + self.assertEqual(tx.state, "cancel") + + @mute_logger(*LOGGERS) + def test_customer_cancellation_aborts_payment(self): + tx, _values = self._create_payplug_payment() + tx._payplug_abort_and_process() + self.assertEqual(tx.state, "cancel") + payment = self.fake_payplug.get_payment(tx.provider_reference) + self.assertEqual(payment["failure"]["code"], "aborted") + + @mute_logger(*LOGGERS) + def test_cancellation_of_paid_payment_keeps_it_paid(self): + tx, _values = self._create_payplug_payment() + self.fake_payplug.mark_paid(tx.provider_reference) + tx._payplug_abort_and_process() + self.assertEqual(tx.state, "done") + + @mute_logger(*LOGGERS) + def test_amount_mismatch_sets_error(self): + tx, _values = self._create_payplug_payment() + payment = self.fake_payplug.mark_paid(tx.provider_reference) + payment["amount"] = 100 + tx._payplug_fetch_and_process() + self.assertEqual(tx.state, "error") + + @mute_logger(*LOGGERS) + def test_currency_mismatch_sets_error(self): + tx, _values = self._create_payplug_payment() + payment = self.fake_payplug.mark_paid(tx.provider_reference) + payment["currency"] = "USD" + tx._payplug_fetch_and_process() + self.assertEqual(tx.state, "error") + + @mute_logger(*LOGGERS) + def test_reference_mismatch_sets_error(self): + tx, _values = self._create_payplug_payment() + payment = self.fake_payplug.mark_paid(tx.provider_reference) + payment["metadata"] = {"odoo_reference": "another-reference"} + tx._payplug_fetch_and_process() + self.assertEqual(tx.state, "error") + + @mute_logger(*LOGGERS) + def test_live_payment_on_test_provider_sets_error(self): + tx, _values = self._create_payplug_payment() + payment = self.fake_payplug.mark_paid(tx.provider_reference) + # The fake API hides live payments from test keys: process the data directly. + tx._process("payplug", dict(payment, is_live=True)) + self.assertEqual(tx.state, "error") + + @mute_logger(*LOGGERS) + def test_other_object_than_payment_sets_error(self): + tx, _values = self._create_payplug_payment() + payment = self.fake_payplug.mark_paid(tx.provider_reference) + tx._process("payplug", dict(payment, object="refund")) + self.assertEqual(tx.state, "error") + self.assertIn("not a payment", tx.state_message) + + @mute_logger(*LOGGERS) + def test_payment_id_mismatch_sets_error(self): + tx, _values = self._create_payplug_payment() + payment = self.fake_payplug.mark_paid(tx.provider_reference) + tx._process("payplug", dict(payment, id="pay_another_payment")) + self.assertEqual(tx.state, "error") + self.assertIn("payment id does not match", tx.state_message) + + @mute_logger(*LOGGERS) + def test_missing_payment_page_url_sets_error(self): + payment_data = {"id": "pay_without_url", "object": "payment"} + with patch.object( + type(self.payplug), "_send_api_request", return_value=payment_data + ): + tx, _values = self._create_payplug_payment() + self.assertEqual(tx.state, "error") + self.assertIn("did not return any payment page URL", tx.state_message) + # The payment id is kept, to match the later notifications of the payment. + self.assertEqual(tx.provider_reference, "pay_without_url") + + def test_empty_customer_values_are_not_sent(self): + self.partner.write({"phone": False, "street2": False, "city": "X" * 150}) + self._create_payplug_payment() + billing = self._last_request()[3]["billing"] + self.assertNotIn("mobile_phone_number", billing) + self.assertNotIn("landline_phone_number", billing) + self.assertNotIn("address2", billing) + self.assertEqual(billing["city"], "X" * const.FIELD_MAX_LENGTHS["city"]) + self.assertIsNone(self.env["payment.transaction"]._payplug_format_phone(False)) + + def test_fetch_without_payplug_id_sends_no_request(self): + tx = self._create_transaction("redirect") + with self.assertLogs(TX_LOGGER, "WARNING") as logs: + tx._payplug_fetch_and_process() + self.assertIn("has no Payplug payment id", logs.output[0]) + self.assertEqual(tx.state, "draft") + self.assertFalse(self.fake_payplug.state.requests) + + @mute_logger(*LOGGERS) + def test_cancellation_without_payment_or_once_processed_sends_no_request(self): + draft_tx = self._create_transaction("redirect") # No payment on Payplug. + draft_tx._payplug_abort_and_process() + self.assertEqual(draft_tx.state, "draft") + self.assertFalse(self.fake_payplug.state.requests) + + tx, _values = self._create_payplug_payment(reference="paid-reference") + self.fake_payplug.mark_paid(tx.provider_reference) + tx._payplug_fetch_and_process() + self.assertEqual(tx.state, "done") + requests_count = len(self.fake_payplug.state.requests) + # E.g. the customer opens the cancel URL after paying. + tx._payplug_abort_and_process() + self.assertEqual(tx.state, "done") + self.assertEqual(len(self.fake_payplug.state.requests), requests_count) + + def test_other_providers_transactions_are_not_affected(self): + payplug_like_data = { + "reference": "dummy-reference", + "object": "payment", + "is_paid": True, + "amount": 4990, + "currency": "EUR", + "metadata": {"odoo_reference": "another-reference"}, + } + tx_model = self.env["payment.transaction"] + self.assertEqual( + tx_model._extract_reference("none", payplug_like_data), "dummy-reference" + ) + dummy_tx = self._create_transaction( + "redirect", + provider_id=self.dummy_provider.id, + payment_method_id=self.dummy_provider.payment_method_ids[:1].id, + reference="dummy-reference", + ) + self.assertEqual(dummy_tx._get_specific_rendering_values({}), {}) + self.assertEqual(dummy_tx._extract_amount_data(payplug_like_data), {}) + dummy_tx._apply_updates(payplug_like_data) + self.assertEqual(dummy_tx.state, "draft") + dummy_tx._send_refund_request() + self.assertFalse(self.fake_payplug.state.requests) + + +@tagged("post_install", "-at_install", "payment_payplug") +class TestPayplugRefreshStatus(PayplugCommon): + @mute_logger(*LOGGERS) + def test_refresh_status(self): + tx, _values = self._create_payplug_payment() + self.fake_payplug.mark_paid(tx.provider_reference) + dummy_tx = self._create_transaction( + "redirect", + provider_id=self.dummy_provider.id, + payment_method_id=self.dummy_provider.payment_method_ids[:1].id, + reference="dummy-reference", + ) + action = (tx + dummy_tx).action_payplug_refresh_status() + self.assertEqual(action, {"type": "ir.actions.client", "tag": "soft_reload"}) + self.assertEqual(tx.state, "done") + self.assertEqual(dummy_tx.state, "draft") + # The creation and one reading, for the Payplug transaction only. + self.assertEqual(len(self.fake_payplug.state.requests), 2) + + @mute_logger(*LOGGERS) + def test_refresh_status_reports_api_errors(self): + tx, _values = self._create_payplug_payment() + del self.fake_payplug.state.payments[tx.provider_reference] + action = tx.action_payplug_refresh_status() + self.assertEqual(action["tag"], "display_notification") + self.assertEqual(action["params"]["type"], "danger") + self.assertIn("could not be found", action["params"]["message"]) + self.assertEqual(tx.state, "draft") + + def test_refresh_status_requires_write_access(self): + tx, _values = self._create_payplug_payment() + with self.assertRaises(AccessError): + tx.with_user(self.internal_user).action_payplug_refresh_status() + self.assertEqual(len(self.fake_payplug.state.requests), 1) + + +@tagged("post_install", "-at_install", "payment_payplug") +class TestPayplugRefund(PayplugCommon): + def _create_paid_transaction(self): + tx, _values = self._create_payplug_payment() + self.fake_payplug.mark_paid(tx.provider_reference) + tx._payplug_fetch_and_process() + self.assertEqual(tx.state, "done") + return tx + + @mute_logger(*LOGGERS) + def test_full_refund(self): + tx = self._create_paid_transaction() + tx.action_refund() + refund_tx = tx.child_transaction_ids + self.assertEqual(len(refund_tx), 1) + self.assertEqual(refund_tx.operation, "refund") + self.assertEqual(refund_tx.state, "done") + self.assertAlmostEqual(refund_tx.amount, -self.amount) + self.assertTrue(refund_tx.provider_reference.startswith("re_")) + payment = self.fake_payplug.get_payment(tx.provider_reference) + self.assertTrue(payment["is_refunded"]) + refund = self.fake_payplug.state.refunds[refund_tx.provider_reference] + self.assertEqual(refund["amount"], 4990) + self.assertEqual(refund["metadata"], {"odoo_reference": refund_tx.reference}) + + @mute_logger(*LOGGERS) + def test_partial_refunds(self): + tx = self._create_paid_transaction() + tx.action_refund(amount_to_refund=10.00) + tx.action_refund(amount_to_refund=15.50) + refund_txs = tx.child_transaction_ids + self.assertEqual(len(refund_txs), 2) + self.assertEqual(set(refund_txs.mapped("state")), {"done"}) + self.assertAlmostEqual(sum(refund_txs.mapped("amount")), -25.50) + payment = self.fake_payplug.get_payment(tx.provider_reference) + self.assertEqual(payment["amount_refunded"], 2550) + self.assertFalse(payment["is_refunded"]) + + @mute_logger(*LOGGERS) + def test_refund_above_remaining_amount_sets_error(self): + tx = self._create_paid_transaction() + tx.action_refund(amount_to_refund=40.00) + tx.action_refund(amount_to_refund=20.00) + failed = tx.child_transaction_ids.filtered(lambda t: t.state == "error") + self.assertEqual(len(failed), 1) + self.assertIn("refundable amount", failed.state_message) + + @mute_logger(*LOGGERS) + def test_refund_below_minimum_sets_error(self): + tx = self._create_paid_transaction() + tx.action_refund(amount_to_refund=0.05) + refund_tx = tx.child_transaction_ids + self.assertEqual(refund_tx.state, "error") + self.assertFalse(self.fake_payplug.state.refunds) + + @mute_logger(*LOGGERS) + def test_refund_data_not_matching_the_refund_sets_error(self): + tx = self._create_paid_transaction() + for name, changes in ( + ("object", {"object": "payment"}), + ("payment", {"payment_id": "pay_another_payment"}), + ("id", {"id": "re_another_refund"}), + ("environment", {"is_live": True}), + ): + with self.subTest(name=name): + refund_tx = tx._create_child_transaction(10.0, is_refund=True) + refund = self.fake_payplug.create_refund( + tx.provider_reference, + 1000, + metadata={"odoo_reference": refund_tx.reference}, + ) + refund_tx.provider_reference = refund["id"] + refund_tx._process("payplug", dict(refund, **changes)) + self.assertEqual(refund_tx.state, "error") + self.assertIn( + "environment" if name == "environment" else "does not match", + refund_tx.state_message, + ) + self.assertEqual(refund_tx.provider_reference, refund["id"]) diff --git a/payment_payplug/tests/test_payplug_hooks.py b/payment_payplug/tests/test_payplug_hooks.py new file mode 100644 index 0000000000..bfdce1b0a8 --- /dev/null +++ b/payment_payplug/tests/test_payplug_hooks.py @@ -0,0 +1,35 @@ +# Copyright 2026 Sébastien Berger (sbinfo67) +# License LGPL-3.0 or later (https://www.gnu.org/licenses/lgpl-3.0). + +from odoo.tests import tagged + +from ..hooks import uninstall_hook +from .common import PayplugCommon + + +@tagged("post_install", "-at_install", "payment_payplug") +class TestPayplugUninstallHook(PayplugCommon): + def test_uninstall_hook(self): + unused_provider = self.payplug.copy( + {"name": "Payplug (unused)", "state": "disabled"} + ) + tx = self._create_transaction("redirect") + self.assertTrue(self.payplug.get_external_id()[self.payplug.id]) + + uninstall_hook(self.env) + + # The provider without transactions is deleted with the module. + self.assertFalse(unused_provider.exists()) + # The provider with transactions is reset and kept, detached from the module, + # so that its transactions stay readable. + self.assertTrue(self.payplug.exists()) + self.assertEqual(self.payplug.code, "none") + self.assertEqual(self.payplug.state, "disabled") + self.assertFalse(self.payplug.is_published) + self.assertFalse(self.payplug.get_external_id()[self.payplug.id]) + self.assertEqual(tx.provider_id, self.payplug) + self.assertFalse( + self.env["payment.provider"] + .with_context(active_test=False) + .search([("code", "=", "payplug")]) + ) diff --git a/payment_payplug/tests/test_payplug_http.py b/payment_payplug/tests/test_payplug_http.py new file mode 100644 index 0000000000..dafba2755b --- /dev/null +++ b/payment_payplug/tests/test_payplug_http.py @@ -0,0 +1,359 @@ +# Copyright 2026 Sébastien Berger (sbinfo67) +# License LGPL-3.0 or later (https://www.gnu.org/licenses/lgpl-3.0). + +from unittest.mock import patch +from urllib.parse import urlsplit + +from odoo.tests import tagged +from odoo.tools import mute_logger + +from odoo.addons.payment.tests.http_common import PaymentHttpCommon + +from ..controllers.main import PayplugController +from .common import PayplugCommon, PayplugOAuth2Common + +CONTROLLER_LOGGER = "odoo.addons.payment_payplug.controllers.main" +TX_LOGGER = "odoo.addons.payment_payplug.models.payment_transaction" +OTHER_LOGGERS = ( + "odoo.addons.payment.models.payment_provider", + "odoo.addons.payment.models.payment_transaction", +) +LOGGERS = (CONTROLLER_LOGGER, TX_LOGGER, *OTHER_LOGGERS) + + +@tagged("post_install", "-at_install", "payment_payplug") +class TestPayplugHttp(PayplugCommon, PaymentHttpCommon): + def _post_notification(self, data): + url = self._build_url(PayplugController._webhook_url) + return self.url_open(url, json=data) + + def _open_hosted_url(self, tx, key): + """Open the return or cancel URL given to Payplug, without the redirection.""" + payment = self.fake_payplug.get_payment(tx.provider_reference) + url = urlsplit(payment["hosted_payment"][key]) + return self.url_open(f"{url.path}?{url.query}", allow_redirects=False) + + def _paid_tx(self): + tx, _values = self._create_payplug_payment() + self.fake_payplug.mark_paid(tx.provider_reference) + return tx + + # === Payment notifications === # + + @mute_logger(*LOGGERS) + def test_webhook_confirms_payment(self): + tx = self._paid_tx() + payment = self.fake_payplug.get_payment(tx.provider_reference) + response = self._post_notification(self._notification(payment)) + self.assertEqual(response.status_code, 200) + self.assertEqual(tx.state, "done") + + @mute_logger(*LOGGERS) + def test_webhook_forged_body_is_not_trusted(self): + tx, _values = self._create_payplug_payment() + payment = dict(self.fake_payplug.get_payment(tx.provider_reference)) + # The forged body claims the payment is paid; the API says otherwise. + payment.update(is_paid=True, paid_at=1700000000) + response = self._post_notification(payment) + self.assertEqual(response.status_code, 200) + self.assertEqual(tx.state, "pending") + + @mute_logger(*LOGGERS) + def test_webhook_unknown_payment_is_ignored(self): + tx = self._paid_tx() + response = self._post_notification( + {"id": "pay_unknown", "object": "payment", "is_live": False} + ) + self.assertEqual(response.status_code, 200) + self.assertEqual(tx.state, "draft") + # No API request was sent for the unknown payment. + self.assertNotIn( + "/v1/payments/pay_unknown", [r[1] for r in self.fake_payplug.state.requests] + ) + + @mute_logger(*LOGGERS) + def test_webhook_replay_is_idempotent(self): + tx = self._paid_tx() + notification = self._notification( + self.fake_payplug.get_payment(tx.provider_reference) + ) + self._post_notification(notification) + self.assertEqual(tx.state, "done") + last_state_change = tx.last_state_change + messages = self.env["mail.message"].search_count([]) + response = self._post_notification(notification) + self.assertEqual(response.status_code, 200) + self.assertEqual(tx.state, "done") + self.assertEqual(tx.last_state_change, last_state_change) + self.assertEqual(self.env["mail.message"].search_count([]), messages) + + @mute_logger(*LOGGERS) + def test_webhook_invalid_body(self): + url = self._build_url(PayplugController._webhook_url) + response = self.url_open( + url, data="not json", headers={"Content-Type": "application/json"} + ) + self.assertEqual(response.status_code, 400) + + @mute_logger(*LOGGERS) + def test_webhook_api_failure_asks_for_retry(self): + tx = self._paid_tx() + notification = self._notification( + self.fake_payplug.get_payment(tx.provider_reference) + ) + del self.fake_payplug.state.payments[tx.provider_reference] + response = self._post_notification(notification) + self.assertEqual(response.status_code, 503) + self.assertEqual(tx.state, "draft") + + @mute_logger(*LOGGERS) + def test_webhook_body_not_an_object(self): + tx = self._paid_tx() + response = self._post_notification( + [self._notification(self.fake_payplug.get_payment(tx.provider_reference))] + ) + self.assertEqual(response.status_code, 400) + self.assertEqual(tx.state, "draft") + self.assertEqual(len(self.fake_payplug.state.requests), 1) # The creation. + + @mute_logger(CONTROLLER_LOGGER, *OTHER_LOGGERS) + def test_webhook_without_object_id_is_ignored(self): + for notification in ( + {"object": "payment"}, + {"object": "payment", "id": ""}, + {"object": "payment", "id": 42}, + ): + with ( + self.subTest(notification=notification), + self.assertLogs(TX_LOGGER, "WARNING") as logs, + ): + response = self._post_notification(notification) + self.assertEqual(response.status_code, 200) + self.assertIn("without object id", logs.output[0]) + self.assertFalse(self.fake_payplug.state.requests) + + @mute_logger(CONTROLLER_LOGGER, *OTHER_LOGGERS) + def test_webhook_of_unknown_object_type_is_ignored(self): + tx = self._paid_tx() + with self.assertLogs(TX_LOGGER, "INFO") as logs: + response = self._post_notification( + {"id": tx.provider_reference, "object": "installment_plan"} + ) + self.assertEqual(response.status_code, 200) + self.assertIn( + "Ignored Payplug notification of type installment_plan", logs.output[0] + ) + self.assertEqual(tx.state, "draft") + self.assertEqual(len(self.fake_payplug.state.requests), 1) # The creation. + + # === Customer redirections === # + + @mute_logger(*LOGGERS) + def test_return_url_processes_payment(self): + tx = self._paid_tx() + response = self._open_hosted_url(tx, "return_url") + self.assertEqual(response.status_code, 303) + self.assertTrue(response.headers["Location"].endswith("/payment/status")) + self.assertEqual(tx.state, "done") + + @mute_logger(*LOGGERS) + def test_return_url_with_invalid_token_is_ignored(self): + tx = self._paid_tx() + url = self._build_url(PayplugController._return_url) + response = self.url_open( + f"{url}?reference={tx.reference}&access_token=forged", allow_redirects=False + ) + self.assertEqual(response.status_code, 303) + self.assertEqual(tx.state, "draft") + + @mute_logger(*LOGGERS) + def test_cancel_url_aborts_payment(self): + tx, _values = self._create_payplug_payment() + response = self._open_hosted_url(tx, "cancel_url") + self.assertEqual(response.status_code, 303) + self.assertEqual(tx.state, "cancel") + payment = self.fake_payplug.get_payment(tx.provider_reference) + self.assertEqual(payment["failure"]["code"], "aborted") + + def _open_hosted_url_without_payment(self, tx, key): + """Open the return or cancel URL after the payment vanished from Payplug, so + that every API request about it fails.""" + url = urlsplit( + self.fake_payplug.get_payment(tx.provider_reference)["hosted_payment"][key] + ) + del self.fake_payplug.state.payments[tx.provider_reference] + with self.assertLogs(CONTROLLER_LOGGER, "ERROR") as logs: + response = self.url_open(f"{url.path}?{url.query}", allow_redirects=False) + # The customer is still redirected to the status page. + self.assertEqual(response.status_code, 303) + self.assertTrue(response.headers["Location"].endswith("/payment/status")) + self.assertEqual(tx.state, "draft") + return logs.output + + @mute_logger(TX_LOGGER, *OTHER_LOGGERS) + def test_return_url_api_failure_still_redirects(self): + tx = self._paid_tx() + logs = self._open_hosted_url_without_payment(tx, "return_url") + self.assertIn(f"Unable to fetch the Payplug payment of {tx.reference}", logs[0]) + + @mute_logger(TX_LOGGER, *OTHER_LOGGERS) + def test_cancel_url_api_failure_still_redirects(self): + tx, _values = self._create_payplug_payment() + logs = self._open_hosted_url_without_payment(tx, "cancel_url") + self.assertIn( + f"Unable to cancel the Payplug payment of {tx.reference}", logs[0] + ) + # The abortion failed, then the reading of the payment. + self.assertEqual( + [request[0] for request in self.fake_payplug.state.requests], + ["POST", "PATCH", "GET"], + ) + + # === Refund notifications === # + + def _done_tx(self): + tx = self._paid_tx() + tx._payplug_fetch_and_process() + self.assertEqual(tx.state, "done") + return tx + + @mute_logger(*LOGGERS) + def test_refund_from_payplug_portal_creates_refund_transaction(self): + tx = self._done_tx() + refund = self.fake_payplug.create_refund( + tx.provider_reference, 1000 + ) # No metadata. + response = self._post_notification(self._notification(refund)) + self.assertEqual(response.status_code, 200) + refund_tx = tx.child_transaction_ids + self.assertEqual(len(refund_tx), 1) + self.assertEqual(refund_tx.operation, "refund") + self.assertEqual(refund_tx.state, "done") + self.assertAlmostEqual(refund_tx.amount, -10.0) + self.assertEqual(refund_tx.provider_reference, refund["id"]) + + # A replayed notification does not create a second refund. + self._post_notification(self._notification(refund)) + self.assertEqual(len(tx.child_transaction_ids), 1) + + @mute_logger(*LOGGERS) + def test_refund_notification_of_odoo_refund_is_not_duplicated(self): + tx = self._done_tx() + tx.action_refund(amount_to_refund=5.0) + refund_tx = tx.child_transaction_ids + refund = self.fake_payplug.state.refunds[refund_tx.provider_reference] + self._post_notification(self._notification(refund)) + self.assertEqual(tx.child_transaction_ids, refund_tx) + self.assertEqual(refund_tx.state, "done") + + @mute_logger(*LOGGERS) + def test_refund_notification_with_forged_payment_id_is_ignored(self): + tx = self._done_tx() + other_tx = self._done_tx_with_reference("other-reference") + refund = self.fake_payplug.create_refund(tx.provider_reference, 1000) + notification = self._notification(refund) + notification["payment_id"] = ( + other_tx.provider_reference + ) # Points to the wrong payment. + self._post_notification(notification) + self.assertFalse(other_tx.child_transaction_ids) + self.assertFalse(tx.child_transaction_ids) + + def _done_tx_with_reference(self, reference): + tx, _values = self._create_payplug_payment(reference=reference) + self.fake_payplug.mark_paid(tx.provider_reference) + tx._payplug_fetch_and_process() + return tx + + @mute_logger(CONTROLLER_LOGGER, *OTHER_LOGGERS) + def test_refund_notification_of_unknown_payment_is_ignored(self): + for notification in ( + {"id": "re_unknown", "object": "refund", "payment_id": "pay_unknown"}, + {"id": "re_unknown", "object": "refund", "payment_id": 42}, + {"id": "re_unknown", "object": "refund"}, + ): + with ( + self.subTest(notification=notification), + self.assertLogs(TX_LOGGER, "WARNING") as logs, + ): + response = self._post_notification(notification) + self.assertEqual(response.status_code, 200) + self.assertIn("No Payplug transaction found for refund", logs.output[0]) + self.assertFalse(self.fake_payplug.state.requests) + + @mute_logger(CONTROLLER_LOGGER, *OTHER_LOGGERS) + def test_refund_notification_not_matching_its_payment_is_ignored(self): + tx = self._done_tx() + refund = self.fake_payplug.create_refund(tx.provider_reference, 1000) + # Defensive check: the API returns a refund of another payment. + inconsistent_refund = dict(refund, payment_id="pay_another_payment") + with ( + patch.object( + type(tx), "_payplug_get_refund", return_value=inconsistent_refund + ), + self.assertLogs(TX_LOGGER, "WARNING") as logs, + ): + response = self._post_notification(self._notification(refund)) + self.assertEqual(response.status_code, 200) + self.assertIn("does not match its payment", logs.output[0]) + self.assertFalse(tx.child_transaction_ids) + + @mute_logger(*LOGGERS) + def test_refund_notification_completes_odoo_refund_with_lost_response(self): + tx = self._done_tx() + # The refund request reached Payplug, but its response was lost: the refund + # transaction has no Payplug id. + refund_tx = tx._create_child_transaction(10.0, is_refund=True) + refund = self.fake_payplug.create_refund( + tx.provider_reference, + 1000, + metadata={"odoo_reference": refund_tx.reference}, + ) + response = self._post_notification(self._notification(refund)) + self.assertEqual(response.status_code, 200) + self.assertEqual(tx.child_transaction_ids, refund_tx) + self.assertEqual(refund_tx.state, "done") + self.assertEqual(refund_tx.provider_reference, refund["id"]) + + @mute_logger(CONTROLLER_LOGGER, *OTHER_LOGGERS) + def test_refund_notification_of_odoo_refund_in_progress_is_left_to_odoo(self): + tx = self._done_tx() + # Refund requested by Odoo, whose transaction is not committed yet. + refund = self.fake_payplug.create_refund( + tx.provider_reference, + 1000, + metadata={"odoo_reference": f"R-{tx.reference}-in-progress"}, + ) + with self.assertLogs(TX_LOGGER, "INFO") as logs: + response = self._post_notification(self._notification(refund)) + self.assertEqual(response.status_code, 200) + self.assertTrue( + any("is being processed by Odoo" in line for line in logs.output) + ) + self.assertFalse(tx.child_transaction_ids) + + @mute_logger(CONTROLLER_LOGGER, *OTHER_LOGGERS) + def test_refund_notification_of_unconfirmed_payment_is_ignored(self): + tx = self._paid_tx() # Paid on Payplug, not confirmed in Odoo yet. + refund = self.fake_payplug.create_refund(tx.provider_reference, 1000) + with self.assertLogs(TX_LOGGER, "WARNING") as logs: + response = self._post_notification(self._notification(refund)) + self.assertEqual(response.status_code, 200) + self.assertIn(f"unconfirmed transaction {tx.reference}", logs.output[0]) + self.assertFalse(tx.child_transaction_ids) + self.assertEqual(tx.state, "draft") + + +@tagged("post_install", "-at_install", "payment_payplug") +class TestPayplugOAuth2Http(PayplugOAuth2Common, PaymentHttpCommon): + @mute_logger(*LOGGERS) + def test_webhook_with_oauth2(self): + tx, _values = self._create_payplug_payment() + payment = self.fake_payplug.mark_paid(tx.provider_reference) + url = self._build_url(PayplugController._webhook_url) + response = self.url_open(url, json=self._notification(payment)) + self.assertEqual(response.status_code, 200) + self.assertEqual(tx.state, "done") + # The request served by the HTTP worker thread reused the cached token. + self.assertEqual(len(self.fake_payplug.state.token_requests), 1) + self.assertEqual(len(self.fake_payplug.state.requests), 2) diff --git a/payment_payplug/tests/test_payplug_invoice.py b/payment_payplug/tests/test_payplug_invoice.py new file mode 100644 index 0000000000..5fdc2e6ab3 --- /dev/null +++ b/payment_payplug/tests/test_payplug_invoice.py @@ -0,0 +1,64 @@ +# Copyright 2026 Sébastien Berger (sbinfo67) +# License LGPL-3.0 or later (https://www.gnu.org/licenses/lgpl-3.0). + +from odoo import Command +from odoo.tests import tagged +from odoo.tools import mute_logger + +from odoo.addons.account_payment.tests.common import AccountPaymentCommon + +from .common import PayplugCommon + +LOGGERS = ( + "odoo.addons.payment_payplug.models.payment_transaction", + "odoo.addons.payment.models.payment_provider", + "odoo.addons.payment.models.payment_transaction", +) + + +@tagged("post_install", "-at_install", "payment_payplug") +class TestPayplugInvoice(PayplugCommon, AccountPaymentCommon): + @classmethod + def setUpClass(cls): + super().setUpClass() + cls.payplug.journal_id = cls.company_data["default_journal_bank"] + cls.invoice = cls.init_invoice( + "out_invoice", partner=cls.partner, amounts=[120.0], taxes=[], post=True + ) + + @mute_logger(*LOGGERS) + def test_invoice_payment_and_partial_refund(self): + self.assertEqual(self.invoice.amount_total, 120.0) + tx, _values = self._create_payplug_payment( + amount=self.invoice.amount_total, + reference=None, + invoice_ids=[Command.set(self.invoice.ids)], + ) + self.assertIn(self.invoice.name, tx.reference) + self.fake_payplug.mark_paid(tx.provider_reference) + tx._payplug_fetch_and_process() + self.assertEqual(tx.state, "done") + + tx._post_process() + self.assertTrue(tx.payment_id) + self.assertEqual(tx.payment_id.amount, 120.0) + self.assertIn(self.invoice.payment_state, ("in_payment", "paid")) + + # Partial refund from the payment, as with the "Refund" button of the payment. + wizard = ( + self.env["payment.refund.wizard"] + .with_context(active_model="account.payment", active_id=tx.payment_id.id) + .create({}) + ) + self.assertEqual(wizard.support_refund, "partial") + wizard.amount_to_refund = 20.0 + wizard.action_refund() + refund_tx = tx.child_transaction_ids + self.assertEqual(refund_tx.state, "done") + self.assertAlmostEqual(refund_tx.amount, -20.0) + refund_tx._post_process() + self.assertTrue(refund_tx.payment_id) + tx.payment_id.invalidate_recordset(["amount_available_for_refund"]) + self.assertAlmostEqual(tx.payment_id.amount_available_for_refund, 100.0) + payment = self.fake_payplug.get_payment(tx.provider_reference) + self.assertEqual(payment["amount_refunded"], 2000) diff --git a/payment_payplug/tests/test_payplug_oauth.py b/payment_payplug/tests/test_payplug_oauth.py new file mode 100644 index 0000000000..099ed0ffd6 --- /dev/null +++ b/payment_payplug/tests/test_payplug_oauth.py @@ -0,0 +1,508 @@ +# Copyright 2026 Sébastien Berger (sbinfo67) +# License LGPL-3.0 or later (https://www.gnu.org/licenses/lgpl-3.0). + +import base64 +import threading +from unittest.mock import Mock, patch +from urllib.parse import parse_qs + +import requests + +from odoo.exceptions import AccessError, ValidationError +from odoo.tests import tagged +from odoo.tools import mute_logger + +from .. import const, oauth +from .common import ( + LIVE_CLIENT_ID, + LIVE_CLIENT_SECRET, + TEST_CLIENT_ID, + TEST_CLIENT_SECRET, + TEST_SECRET_KEY, + PayplugOAuth2Common, +) + +LOGGERS = ( + "odoo.addons.payment_payplug.models.payment_provider", + "odoo.addons.payment_payplug.models.payment_transaction", + "odoo.addons.payment.models.payment_provider", + "odoo.addons.payment.models.payment_transaction", +) + +REQUESTS_POST = "odoo.addons.payment_payplug.models.payment_provider.requests.post" + + +@tagged("post_install", "-at_install", "payment_payplug") +class TestPayplugOAuth2(PayplugOAuth2Common): + # === Helpers === # + + def _token_requests(self): + return self.fake_payplug.state.token_requests + + def _api_requests(self): + return self.fake_payplug.state.requests + + def _api_tokens(self): + """Return the bearer tokens of the API requests, in order.""" + return [ + headers["Authorization"].removeprefix("Bearer ") + for _method, _path, headers, _body in self._api_requests() + ] + + def _paid_tx(self): + tx, _values = self._create_payplug_payment() + self.fake_payplug.mark_paid(tx.provider_reference) + return tx + + # === Token request === # + + def test_token_request(self): + tx, _values = self._create_payplug_payment() + self.assertEqual(tx.state, "draft") + self.assertTrue(tx.provider_reference) + + self.assertEqual(len(self._token_requests()), 1) + headers, body = self._token_requests()[0] + credentials = base64.b64encode( + f"{TEST_CLIENT_ID}:{TEST_CLIENT_SECRET}".encode() + ).decode() + self.assertEqual(headers["Authorization"], f"Basic {credentials}") + self.assertEqual(headers["Content-Type"], "application/x-www-form-urlencoded") + self.assertEqual( + parse_qs(body.decode()), + { + "grant_type": ["client_credentials"], + "audience": [const.DEFAULT_TOKEN_AUDIENCE], + }, + ) + + token = self.fake_payplug.issued_tokens()[0] + method, path, api_headers, _body = self._api_requests()[-1] + self.assertEqual((method, path), ("POST", "/v1/payments")) + self.assertEqual(api_headers["Authorization"], f"Bearer {token}") + self.assertEqual(api_headers["PayPlug-Version"], const.API_VERSION) + + def test_default_token_url_and_audience(self): + self.assertEqual( + self.payplug._payplug_get_token_url(), + self.fake_payplug.api_url.replace("/v1/", "/oauth2/token"), + ) + self.env["ir.config_parameter"].sudo().set_param( + const.API_BASE_URL_PARAM, False + ) + self.assertEqual( + self.payplug._payplug_get_token_url(), + "https://api.payplug.com/oauth2/token", + ) + self.assertEqual( + self.payplug._payplug_get_token_audience(), "https://www.payplug.com" + ) + self.env["ir.config_parameter"].sudo().set_param( + const.TOKEN_URL_PARAM, "http://127.0.0.1:1/token" + ) + self.assertEqual( + self.payplug._payplug_get_token_url(), "http://127.0.0.1:1/token" + ) + + @mute_logger(*LOGGERS) + def test_whole_flow_with_oauth2(self): + tx = self._paid_tx() + tx._payplug_fetch_and_process() + self.assertEqual(tx.state, "done") + tx.action_refund(amount_to_refund=10.0) + refund_tx = tx.child_transaction_ids + self.assertEqual(refund_tx.state, "done") + other_tx, _values = self._create_payplug_payment(reference="other-ref") + other_tx._payplug_abort_and_process() + self.assertEqual(other_tx.state, "cancel") + # One token for all the requests: creations, reading, refund, abortion. + self.assertEqual(len(self._token_requests()), 1) + self.assertEqual(len(self._api_requests()), 5) + self.assertEqual( + set(self._api_tokens()), set(self.fake_payplug.issued_tokens()) + ) + + # === Cache === # + + @mute_logger(*LOGGERS) + def test_token_is_cached(self): + tx = self._paid_tx() + tx._payplug_fetch_and_process() + tx._payplug_fetch_and_process() + self._create_payplug_payment(reference="second-ref") + self.assertEqual(len(self._token_requests()), 1) + self.assertEqual(len(set(self._api_tokens())), 1) + + def test_cache_key_does_not_hold_secrets(self): + self.payplug._payplug_get_access_token() + (key,) = oauth.TOKEN_CACHE.keys() + self.assertEqual(key[:2], (self.env.cr.dbname, self.payplug.id)) + self.assertNotIn(TEST_CLIENT_SECRET, repr(key)) + self.assertNotIn(TEST_CLIENT_ID, repr(key)) + + @mute_logger(*LOGGERS) + def test_token_renewed_before_expiry(self): + start = oauth.now() + with patch.object(oauth, "now", return_value=start): + tx = self._paid_tx() + # The token lives 299 s and is renewed 60 s before its expiry. + with patch.object(oauth, "now", return_value=start + 238): + tx._payplug_fetch_and_process() + self.assertEqual(len(self._token_requests()), 1) + with patch.object(oauth, "now", return_value=start + 240): + tx._payplug_fetch_and_process() + self.assertEqual(len(self._token_requests()), 2) + first, second = self.fake_payplug.issued_tokens() + self.assertEqual(self._api_tokens(), [first, first, second]) + + @mute_logger(*LOGGERS) + def test_short_token_renewed_at_half_life(self): + self.fake_payplug.state.token_lifetime = 30 + start = oauth.now() + with patch.object(oauth, "now", return_value=start): + self.payplug._payplug_get_access_token() + with patch.object(oauth, "now", return_value=start + 14): + self.payplug._payplug_get_access_token() + self.assertEqual(len(self._token_requests()), 1) + with patch.object(oauth, "now", return_value=start + 16): + self.payplug._payplug_get_access_token() + self.assertEqual(len(self._token_requests()), 2) + + def test_cache_invalidated_on_credentials_change(self): + self.payplug._payplug_get_access_token() + self.payplug.name = "Payplug (renamed)" + self.payplug._payplug_get_access_token() + self.assertEqual(len(self._token_requests()), 1) + self.payplug.payplug_client_secret = TEST_CLIENT_SECRET + self.payplug._payplug_get_access_token() + self.assertEqual(len(self._token_requests()), 2) + + def test_concurrent_requests_fetch_one_token(self): + cache = oauth.TokenCache(renewal_margin=60) + calls = [] + barrier = threading.Barrier(5) + + def fetch(): + calls.append(1) + threading.Event().wait(0.1) + return f"token-{len(calls)}", 299, oauth.MODE_TEST, "test" + + results = [] + + def worker(): + barrier.wait() + results.append(cache.get_or_fetch(("db", 1, "h"), b"fp", fetch).value) + + threads = [threading.Thread(target=worker) for _i in range(5)] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + self.assertEqual(len(calls), 1) + self.assertEqual(results, ["token-1"] * 5) + # Another secret (other fingerprint) gets another token. + self.assertEqual( + cache.get_or_fetch(("db", 1, "h"), b"other", fetch).value, "token-2" + ) + + # === Rejected tokens === # + + @mute_logger(*LOGGERS) + def test_revoked_token_is_renewed_and_request_replayed(self): + tx = self._paid_tx() + self.fake_payplug.revoke_tokens() + tx._payplug_fetch_and_process() + self.assertEqual(tx.state, "done") + self.assertEqual(len(self._token_requests()), 2) + first, second = self.fake_payplug.issued_tokens() + # Creation, rejected reading, replayed reading. + self.assertEqual(self._api_tokens(), [first, first, second]) + self.assertEqual( + [request[1] for request in self._api_requests()[1:]], + [f"/v1/payments/{tx.provider_reference}"] * 2, + ) + + @mute_logger(*LOGGERS) + def test_request_replayed_only_once(self): + self.fake_payplug.state.reject_all_tokens = True + tx, _values = self._create_payplug_payment() + self.assertEqual(tx.state, "error") + self.assertEqual(len(self._token_requests()), 2) + self.assertEqual(len(self._api_requests()), 2) + self.assertFalse(self.fake_payplug.state.payments) + + @mute_logger(*LOGGERS) + def test_api_key_errors_are_not_replayed(self): + self.payplug.write( + {"payplug_auth_method": "api_key", "payplug_secret_key": "sk_test_x"} + ) + self.payplug.payplug_secret_key = "invalid-key" + with self.assertRaises(ValidationError): + self.payplug._send_api_request("GET", "payments/pay_unknown") + self.assertEqual(len(self._api_requests()), 1) + self.assertFalse(self._token_requests()) + + # === Token endpoint errors === # + + @mute_logger(*LOGGERS) + def test_invalid_credentials(self): + wrong_secret = "fake_wrong_secret_8f3a" + self.payplug.payplug_client_secret = wrong_secret + tx, _values = self._create_payplug_payment() + self.assertEqual(tx.state, "error") + self.assertIn("client ID or secret", tx.state_message) + self.assertNotIn(wrong_secret, tx.state_message) + self.assertFalse(self._api_requests()) + self.assertFalse(oauth.TOKEN_CACHE.keys()) + # Nothing is cached after a failure: the next attempt asks again. + with self.assertRaises(ValidationError): + self.payplug._payplug_get_access_token() + self.assertEqual(len(self._token_requests()), 2) + + def test_blank_credentials_are_refused_before_any_request(self): + self.payplug.payplug_client_secret = " " + with self.assertRaisesRegex( + ValidationError, "client ID and secret are not set" + ): + self.payplug._payplug_get_access_token() + self.assertFalse(self._token_requests()) + + @mute_logger(*LOGGERS) + def test_token_request_refused(self): + # The fake server refuses an unknown audience with an `invalid_request` error. + self.env["ir.config_parameter"].sudo().set_param( + const.TOKEN_AUDIENCE_PARAM, "https://audience.example.com" + ) + with self.assertRaisesRegex( + ValidationError, r"refused the access token request \(HTTP 400 invalid_req" + ): + self.payplug._payplug_get_access_token() + self.assertEqual(len(self._token_requests()), 1) + self.assertFalse(oauth.TOKEN_CACHE.keys()) + + @mute_logger(*LOGGERS) + def test_unexpected_token_error_code_is_not_shown(self): + response = Mock(ok=False, status_code=400) + for content in ( + {"error": "Forged error"}, + {"error": 42}, + ["invalid_request"], + ValueError, # Not JSON. + ): + if content is ValueError: + response.json.side_effect = ValueError + else: + response.json.return_value = content + with ( + self.subTest(content=content), + patch(REQUESTS_POST, return_value=response), + self.assertRaises(ValidationError) as error, + ): + self.payplug._payplug_get_access_token() + message = str(error.exception) + self.assertIn("refused the access token request (HTTP 400", message) + self.assertNotIn("Forged", message) + self.assertNotIn("42", message) + self.assertNotIn("invalid_request", message) + + @mute_logger(*LOGGERS) + def test_token_endpoint_unavailable(self): + self.fake_payplug.state.token_failure_status = 503 + tx, _values = self._create_payplug_payment() + self.assertEqual(tx.state, "error") + self.assertIn("unavailable (HTTP 503)", tx.state_message) + self.assertFalse(self._api_requests()) + + @mute_logger(*LOGGERS) + def test_token_endpoint_timeout(self): + with ( + patch(REQUESTS_POST, side_effect=requests.exceptions.Timeout), + self.assertRaisesRegex(ValidationError, "Could not reach"), + ): + self.payplug._payplug_get_access_token() + self.assertFalse(oauth.TOKEN_CACHE.keys()) + + @mute_logger(*LOGGERS) + def test_invalid_token_response(self): + response = Mock(ok=True, status_code=200) + for content in ( + {"token_type": "bearer", "expires_in": 299}, + {"access_token": "abc def", "token_type": "bearer"}, + {"access_token": "abc", "token_type": "mac"}, + ): + response.json.return_value = content + with ( + patch(REQUESTS_POST, return_value=response), + self.assertRaisesRegex(ValidationError, "invalid access token"), + ): + self.payplug._payplug_get_access_token() + response.json.side_effect = ValueError + with ( + patch(REQUESTS_POST, return_value=response), + self.assertRaisesRegex(ValidationError, "invalid access token"), + ): + self.payplug._payplug_get_access_token() + + def test_token_without_lifetime_is_not_reused(self): + response = Mock(ok=True, status_code=200) + response.json.return_value = {"access_token": "opaque", "scope": "test"} + with patch(REQUESTS_POST, return_value=response) as post: + self.assertEqual(self.payplug._payplug_get_access_token(), "opaque") + self.payplug._payplug_get_access_token() + self.assertEqual(post.call_count, 2) + + # === Environment (scope) === # + + @mute_logger(*LOGGERS) + def test_live_key_refused_in_test_mode(self): + self.payplug.write( + { + "payplug_client_id": LIVE_CLIENT_ID, + "payplug_client_secret": LIVE_CLIENT_SECRET, + } + ) + tx, _values = self._create_payplug_payment() + self.assertEqual(tx.state, "error") + self.assertIn("Test environment", tx.state_message) + self.assertFalse(self._api_requests()) + self.assertFalse(oauth.TOKEN_CACHE.keys()) + + @mute_logger(*LOGGERS) + def test_unknown_scope_refused_in_test_mode(self): + self.fake_payplug.state.token_scope = "sandbox" + with self.assertRaisesRegex(ValidationError, "scope: sandbox"): + self.payplug._payplug_get_access_token() + + @mute_logger(*LOGGERS) + def test_test_key_refused_when_enabled(self): + self.payplug.state = "enabled" + with self.assertRaisesRegex(ValidationError, "Live environment"): + self.payplug._payplug_get_access_token() + + @mute_logger(*LOGGERS) + def test_live_key_when_enabled(self): + self.payplug.write( + { + "state": "enabled", + "payplug_client_id": LIVE_CLIENT_ID, + "payplug_client_secret": LIVE_CLIENT_SECRET, + } + ) + tx, _values = self._create_payplug_payment() + self.assertTrue(tx.is_live) + payment = self.fake_payplug.get_payment(tx.provider_reference) + self.assertTrue(payment["is_live"]) + tx._payplug_fetch_and_process() + self.assertEqual(tx.state, "pending") + + @mute_logger(*LOGGERS) + def test_unknown_scope_accepted_when_enabled(self): + # The scope of the Live tokens is not documented: an unknown one is accepted, + # every payment being checked with `is_live`. + self.fake_payplug.state.token_scope = "production" + self.payplug.write( + { + "state": "enabled", + "payplug_client_id": LIVE_CLIENT_ID, + "payplug_client_secret": LIVE_CLIENT_SECRET, + } + ) + self.assertTrue(self.payplug._payplug_get_access_token()) + + @mute_logger(*LOGGERS) + def test_cached_token_checked_against_current_state(self): + self.payplug._payplug_get_access_token() + # As if another worker enabled the provider: this cache is not invalidated. + with patch.object(oauth.TOKEN_CACHE, "invalidate"): + self.payplug.state = "enabled" + self.assertTrue(oauth.TOKEN_CACHE.keys()) + with self.assertRaisesRegex(ValidationError, "Live environment"): + self.payplug._payplug_get_access_token() + self.assertEqual(len(self._token_requests()), 1) + + def test_token_mode(self): + self.assertEqual(oauth.get_token_mode({"test"}), oauth.MODE_TEST) + self.assertEqual(oauth.get_token_mode({"live", "x"}), oauth.MODE_LIVE) + self.assertEqual(oauth.get_token_mode({"test", "live"}), oauth.MODE_MIXED) + self.assertEqual(oauth.get_token_mode(set()), oauth.MODE_UNKNOWN) + token = self.payplug._payplug_get_access_token() + self.assertEqual(oauth.get_token_scopes(token), {"test"}) + self.assertEqual(oauth.get_token_scopes("not-a-jwt", "a b"), {"a", "b"}) + self.assertEqual(oauth.decode_jwt_claims("a.%%%.c"), {}) + + # === Configuration === # + + def test_credentials_required_by_method(self): + with self.assertRaisesRegex(ValidationError, "must be filled"): + self.payplug.payplug_client_secret = False + with self.assertRaisesRegex(ValidationError, "must be filled"): + self.payplug.payplug_auth_method = "api_key" # No secret key. + # A disabled provider needs no credentials. + self.payplug.write({"state": "disabled", "payplug_client_id": False}) + self.payplug.write( + { + "state": "test", + "payplug_auth_method": "api_key", + "payplug_secret_key": TEST_SECRET_KEY, + } + ) + self.assertEqual(self.payplug.state, "test") + + def test_secret_key_prefix_not_checked_with_oauth2(self): + # A legacy live key left in the field does not block the OAuth2 test mode. + self.payplug.payplug_secret_key = "sk_live_whatever" + self.assertEqual(self.payplug.state, "test") + + def test_client_secret_restricted_to_system_group(self): + for field_name in ("payplug_client_id", "payplug_client_secret"): + field = self.env["payment.provider"]._fields[field_name] + self.assertEqual(field.groups, "base.group_system") + with self.assertRaises(AccessError): + self.payplug.with_user(self.internal_user).read([field_name]) + + def test_reset_credentials_clears_oauth2_client(self): + self.payplug.action_reset_credentials() + self.assertFalse(self.payplug.payplug_client_id) + self.assertFalse(self.payplug.payplug_client_secret) + self.assertEqual(self.payplug.state, "disabled") + + @mute_logger(*LOGGERS) + def test_api_key_method_still_works(self): + self.payplug.write( + {"payplug_auth_method": "api_key", "payplug_secret_key": TEST_SECRET_KEY} + ) + tx, _values = self._create_payplug_payment() + self.assertEqual(tx.state, "draft") + self.assertEqual(self._api_tokens(), [TEST_SECRET_KEY]) + self.assertFalse(self._token_requests()) + + # === Logs === # + + def test_no_secret_in_logs(self): + with self.assertLogs(level="DEBUG") as logs: + tx = self._paid_tx() + tx._payplug_fetch_and_process() + tx.action_refund(amount_to_refund=5.0) + self.fake_payplug.revoke_tokens() + tx._payplug_fetch_and_process() + self.payplug.payplug_client_secret = "fake_wrong_secret_8f3a" + with self.assertLogs(level="DEBUG") as error_logs: + failed_tx, _values = self._create_payplug_payment(reference="failed-ref") + self.assertEqual(tx.state, "done") + self.assertEqual(failed_tx.state, "error") + output = "\n".join(logs.output + error_logs.output) + # The logs were captured: the requests and the token renewal are there. + self.assertIn("Obtained a Payplug access token", output) + self.assertIn("Payplug rejected the access token", output) + self.assertIn("/v1/payments", output) + tokens = self.fake_payplug.issued_tokens() + self.assertEqual(len(tokens), 2) + basic = base64.b64encode( + f"{TEST_CLIENT_ID}:{TEST_CLIENT_SECRET}".encode() + ).decode() + for secret in (TEST_CLIENT_SECRET, "fake_wrong_secret_8f3a", basic, *tokens): + self.assertNotIn(secret, output) + for token in tokens: # Not even the signature alone. + self.assertNotIn(token.split(".")[2], output) + self.assertNotIn("Bearer", output) + self.assertNotIn("Basic", output) diff --git a/payment_payplug/views/payment_payplug_templates.xml b/payment_payplug/views/payment_payplug_templates.xml new file mode 100644 index 0000000000..ccd9532c47 --- /dev/null +++ b/payment_payplug/views/payment_payplug_templates.xml @@ -0,0 +1,17 @@ + + +