[IMP] *_online_ponto: test ponto_interface and buffer purge.

account_bank_statement_import_online_ponto/models/online_bank_statement_provider_ponto.py

@@ -63,11 +63,10 @@ class OnlineBankStatementProviderPonto(models.Model):
                 access_data,
                 latest_identifier
             )
-        self.ponto_last_identifier = latest_identifier
 
     def _obtain_statement_data(self, date_since, date_until):
         self.ensure_one()
-        if self.service != "ponto":
+        if self.service != "ponto":  # pragma: no cover
             return super()._obtain_statement_data(
                 date_since,
                 date_until,
@@ -159,6 +158,8 @@ class OnlineBankStatementProviderPonto(models.Model):
             ("active", "=", True),
         ])
         for provider in providers:
+            if provider.service != "ponto":
+                continue
             if not provider.ponto_buffer_retain_days:
                 continue
             cutoff_date = today - relativedelta(days=provider.ponto_buffer_retain_days)

account_bank_statement_import_online_ponto/tests/__init__.py

@@ -1,3 +1,4 @@
 # License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl).
 
+from . import test_ponto_interface
 from . import test_account_statement_import_online_ponto

account_bank_statement_import_online_ponto/tests/test_account_statement_import_online_ponto.py

@@ -15,72 +15,7 @@ _interface_class = (
     + ".PontoInterface"
 )
 
-
-class TestBankAccountStatementImportOnlinePonto(common.TransactionCase):
-    post_install = True
-
-    def setUp(self):
-        super().setUp()
-
-        self.now = fields.Datetime.now()
-        self.currency_eur = self.env.ref("base.EUR")
-        self.currency_usd = self.env.ref("base.USD")
-        self.AccountJournal = self.env["account.journal"]
-        self.ResPartnerBank = self.env["res.partner.bank"]
-        self.OnlineBankStatementProvider = self.env["online.bank.statement.provider"]
-        self.AccountAccount = self.env["account.account"]
-        self.AccountBankStatement = self.env["account.bank.statement"]
-        self.AccountBankStatementLine = self.env["account.bank.statement.line"]
-        self.AccountStatementPull = self.env["online.bank.statement.pull.wizard"]
-
-        self.bank_account = self.ResPartnerBank.create(
-            {
-                "acc_number": "FR0214508000302245362775K46",
-                "partner_id": self.env.user.company_id.partner_id.id,
-            }
-        )
-        self.journal = self.AccountJournal.create(
-            {
-                "name": "Bank",
-                "type": "bank",
-                "code": "BANK",
-                "currency_id": self.currency_eur.id,
-                "bank_statements_source": "online",
-                "online_bank_statement_provider": "ponto",
-                "bank_account_id": self.bank_account.id,
-            }
-        )
-        self.receivable_account = self.AccountAccount.create(
-            {
-                "code": "1325",
-                "name": "Test receivable account",
-                "user_type_id": self.env.ref("account.data_account_type_payable").id,
-                "reconcile": True,
-            }
-        )
-        self.provider = self.journal.online_bank_statement_provider_id
-
-        self.mock_login = lambda: mock.patch(
-            _interface_class + "._login",
-            return_value={
-                "username": "test_user",
-                "password": "very_secret",
-                "access_token": "abcd1234",
-                "token_expiration": datetime(2099, 12, 31, 23, 59, 59),
-            },
-        )
-        self.mock_set_access_account = lambda: mock.patch(
-            _interface_class + "._set_access_account",
-            return_value=None,
-        )
-        self.mock_synchronisation = lambda: mock.patch(
-            _interface_class + "._ponto_synchronisation",
-            return_value=None,
-        )
-        # return list of transactions on first call, empty list on second call.
-        self.mock_get_transactions = lambda: mock.patch(
-            _interface_class + "._get_transactions",
-            side_effect=[[
+THREE_TRANSACTIONS = [
     {
         "type": "transaction",
         "relationships": {
@@ -153,7 +88,76 @@ class TestBankAccountStatementImportOnlinePonto(common.TransactionCase):
             "amount": 5.83,
         },
     },
-            ], [], ]
+]
+
+EMPTY_TRANSACTIONS = []
+
+
+class TestBankAccountStatementImportOnlinePonto(common.TransactionCase):
+    post_install = True
+
+    def setUp(self):
+        super().setUp()
+
+        self.now = fields.Datetime.now()
+        self.currency_eur = self.env.ref("base.EUR")
+        self.currency_usd = self.env.ref("base.USD")
+        self.AccountJournal = self.env["account.journal"]
+        self.ResPartnerBank = self.env["res.partner.bank"]
+        self.OnlineBankStatementProvider = self.env["online.bank.statement.provider"]
+        self.AccountAccount = self.env["account.account"]
+        self.AccountBankStatement = self.env["account.bank.statement"]
+        self.AccountBankStatementLine = self.env["account.bank.statement.line"]
+        self.AccountStatementPull = self.env["online.bank.statement.pull.wizard"]
+
+        self.bank_account = self.ResPartnerBank.create(
+            {
+                "acc_number": "FR0214508000302245362775K46",
+                "partner_id": self.env.user.company_id.partner_id.id,
+            }
+        )
+        self.journal = self.AccountJournal.create(
+            {
+                "name": "Bank",
+                "type": "bank",
+                "code": "BANK",
+                "currency_id": self.currency_eur.id,
+                "bank_statements_source": "online",
+                "online_bank_statement_provider": "ponto",
+                "bank_account_id": self.bank_account.id,
+            }
+        )
+        self.receivable_account = self.AccountAccount.create(
+            {
+                "code": "1325",
+                "name": "Test receivable account",
+                "user_type_id": self.env.ref("account.data_account_type_payable").id,
+                "reconcile": True,
+            }
+        )
+        self.provider = self.journal.online_bank_statement_provider_id
+
+        self.mock_login = lambda: mock.patch(
+            _interface_class + "._login",
+            return_value={
+                "username": "test_user",
+                "password": "very_secret",
+                "access_token": "abcd1234",
+                "token_expiration": datetime(2099, 12, 31, 23, 59, 59),
+            },
+        )
+        self.mock_set_access_account = lambda: mock.patch(
+            _interface_class + "._set_access_account",
+            return_value=None,
+        )
+        self.mock_synchronisation = lambda: mock.patch(
+            _interface_class + "._ponto_synchronisation",
+            return_value=None,
+        )
+        # return list of transactions on first call, empty list on second call.
+        self.mock_get_transactions = lambda: mock.patch(
+            _interface_class + "._get_transactions",
+            side_effect=[THREE_TRANSACTIONS, EMPTY_TRANSACTIONS, ]
         )
 
     def test_balance_start(self):
@@ -222,3 +226,20 @@ class TestBankAccountStatementImportOnlinePonto(common.TransactionCase):
             self.assertEqual(statement.balance_end, 17.39)
             # Ponto does not give balance info in transactions.
             # self.assertEqual(statement.balance_end_real, 17.39)
+
+    def test_ponto_buffer_purge(self):
+        """Create some old buffer records and test purging them."""
+        self.provider.write(
+            {
+                "ponto_buffer_retain_days": 31,
+                "active": True,
+            }
+        )
+        buffer_model = self.env["ponto.buffer"]
+        buffer_model.sudo()._store_transactions(self.provider, THREE_TRANSACTIONS)
+        buffers = buffer_model.search([("provider_id", "=", self.provider.id)])
+        # As all transactions have a different date, they will be in separate buffers.
+        self.assertEqual(len(buffers), 3)
+        self.provider._ponto_buffer_purge()
+        buffers = buffer_model.search([("provider_id", "=", self.provider.id)])
+        self.assertFalse(bool(buffers))

account_bank_statement_import_online_ponto/tests/test_ponto_interface.py

@@ -0,0 +1,124 @@
+# Copyright 2022 Therp BV <https://therp.nl>.
+# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl).
+from dateutil.relativedelta import relativedelta
+import json
+
+from unittest.mock import MagicMock, patch
+
+from odoo import fields
+from odoo.tests import common
+
+from .test_account_statement_import_online_ponto import THREE_TRANSACTIONS
+
+
+class TestPontoInterface(common.TransactionCase):
+    post_install = True
+
+    @patch("requests.post")
+    def test_login(self, requests_post):
+        """Check Ponto API login."""
+        mock_response = MagicMock()
+        mock_response.status_code = 200
+        mock_response.text = json.dumps(
+            {
+                "access_token": "live_the_token",
+                "expires_in": 1799,
+                "scope": "ai",
+                "token_type": "bearer",
+            }
+        )
+        requests_post.return_value = mock_response
+        interface_model = self.env["ponto.interface"]
+        access_data = interface_model._login("uncle_john", "secret")
+        self.assertEqual(access_data["access_token"], "live_the_token")
+        self.assertIn("token_expiration", access_data)
+
+    @patch("requests.get")
+    def test_set_access_account(self, requests_get):
+        """Test getting account data for Ponto access."""
+        mock_response = MagicMock()
+        mock_response.status_code = 200
+        mock_response.text = json.dumps(
+            {
+                "data": [
+                    {
+                        "id": "wrong_id",
+                        "attributes": {
+                            "reference": "NL66ABNA123456789",
+                        },
+                    },
+                    {
+                        "id": "2ad3df83-be01-47cf-a6be-cf0de5cb4c99",
+                        "attributes": {
+                            "reference": "NL66RABO123456789",
+                        },
+                    },
+                ],
+            }
+        )
+        requests_get.return_value = mock_response
+        # Start of actual test.
+        access_data = self._get_access_dict(include_account=False)
+        interface_model = self.env["ponto.interface"]
+        interface_model._set_access_account(access_data, "NL66RABO123456789")
+        self.assertIn("ponto_account", access_data)
+        self.assertEqual(
+            access_data["ponto_account"],
+            "2ad3df83-be01-47cf-a6be-cf0de5cb4c99"
+        )
+
+    @patch("requests.post")
+    def test_ponto_synchronisation(self, requests_post):
+        """Test requesting Ponto synchronization."""
+        mock_response = MagicMock()
+        mock_response.status_code = 400
+        mock_response.text = json.dumps(
+            {
+                "errors": [
+                    {
+                        "code": "accountRecentlySynchronized",
+                        "detail":
+                            "This type of synchronization was already created recently"
+                            " for the account. Try again later or on the Dashboard.",
+                        "meta": {}
+                    }
+                ]
+            }
+        )
+        requests_post.return_value = mock_response
+        # Start of actual test (succeeds if no Exceptions occur).
+        access_data = self._get_access_dict()
+        interface_model = self.env["ponto.interface"]
+        interface_model._ponto_synchronisation(access_data)
+
+    @patch("requests.get")
+    def test_get_transactions(self, requests_get):
+        """Test getting transactions from Ponto."""
+        mock_response = MagicMock()
+        mock_response.status_code = 200
+        # Key "data" will contain a list of transactions.
+        mock_response.text = json.dumps({"data": THREE_TRANSACTIONS})
+        requests_get.return_value = mock_response
+        # Start of actual test.
+        access_data = self._get_access_dict()
+        interface_model = self.env["ponto.interface"]
+        transactions = interface_model._get_transactions(access_data, False)
+        self.assertEqual(len(transactions), 3)
+        self.assertEqual(transactions[2]["id"], "b21a6c65-1c52-4ba6-8cbc-127d2b2d85ff")
+        self.assertEqual(
+            transactions[2]["attributes"]["counterpartReference"],
+            "BE10325927501996"
+        )
+
+    def _get_access_dict(self, include_account=True):
+        """Get access dict that caches login/account information."""
+        token_expiration = fields.Datetime.now() + relativedelta(seconds=1800)
+        access_data = {
+            "username": "uncle_john",
+            "password": "secret",
+            "access_token": "live_the_token",
+            "token_expiration": token_expiration,
+        }
+        if include_account:
+            access_data["ponto_account"] = "2ad3df83-be01-47cf-a6be-cf0de5cb4c99"
+        return access_data