[IMP] Adapt account_bank_statement_import_online to 14.0

Also Rename account_bank_statement_import_online => account_statement_import_online
This commit is contained in:
Ronald Portier
2021-05-28 14:57:23 +02:00
parent cf39daaea0
commit 5d62d497d9
52 changed files with 1300 additions and 1367 deletions

View File

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

View File

@@ -0,0 +1,77 @@
# Copyright 2019-2020 Brainbean Apps (https://brainbeanapps.com)
# Copyright 2019-2020 Dataplug (https://dataplug.io)
# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl.html).
from datetime import datetime, timedelta
from random import randrange
from dateutil.relativedelta import relativedelta
from pytz import timezone
from odoo import fields, models
class OnlineBankStatementProviderDummy(models.Model):
_inherit = "online.bank.statement.provider"
def _obtain_statement_data(self, date_since, date_until):
self.ensure_one()
if self.service != "dummy":
return super()._obtain_statement_data(
date_since,
date_until,
) # pragma: no cover
if self.env.context.get("crash", False):
exception = self.env.context.get("exception", Exception("Expected"))
raise exception
line_step_options = self.env.context.get("step", {"minutes": 5})
line_step = relativedelta(**line_step_options)
expand_by = self.env.context.get("expand_by", 0)
data_since = self.env.context.get("data_since", date_since)
data_until = self.env.context.get("data_until", date_until)
data_since -= expand_by * line_step
data_until += expand_by * line_step
balance_start = self.env.context.get(
"balance_start", randrange(-10000, 10000, 1) * 0.1
)
balance = balance_start
tz = self.env.context.get("tz")
if tz:
tz = timezone(tz)
timestamp_mode = self.env.context.get("timestamp_mode")
lines = []
date = data_since
while date < data_until:
amount = self.env.context.get("amount", randrange(-100, 100, 1) * 0.1)
transaction_date = date.replace(tzinfo=tz)
if timestamp_mode == "date":
transaction_date = transaction_date.date()
elif timestamp_mode == "str":
transaction_date = fields.Datetime.to_string(transaction_date)
lines.append(
{
"name": "payment",
"amount": amount,
"date": transaction_date,
"unique_import_id": str(
int((date - datetime(1970, 1, 1)) / timedelta(seconds=1))
),
"partner_name": "John Doe",
"account_number": "XX00 0000 0000 0000",
}
)
balance += amount
date += line_step
balance_end = balance
statement = {}
if self.env.context.get("balance", True):
statement.update(
{"balance_start": balance_start, "balance_end_real": balance_end}
)
return lines, statement

View File

@@ -0,0 +1,645 @@
# Copyright 2019-2020 Brainbean Apps (https://brainbeanapps.com)
# Copyright 2019-2020 Dataplug (https://dataplug.io)
# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl).
from datetime import date, datetime
from urllib.error import HTTPError
from dateutil.relativedelta import relativedelta
from psycopg2 import IntegrityError
from odoo import fields
from odoo.tests import common
from odoo.tools import mute_logger
class TestAccountBankAccountStatementImportOnline(common.TransactionCase):
def setUp(self):
super().setUp()
self.now = fields.Datetime.now()
self.AccountJournal = self.env["account.journal"]
self.OnlineBankStatementProvider = self.env["online.bank.statement.provider"]
self.OnlineBankStatementPullWizard = self.env[
"online.bank.statement.pull.wizard"
]
self.AccountBankStatement = self.env["account.bank.statement"]
self.AccountBankStatementLine = self.env["account.bank.statement.line"]
def test_provider_unlink_restricted(self):
journal = self.AccountJournal.create(
{"name": "Bank", "type": "bank", "code": "BANK"}
)
with common.Form(journal) as journal_form:
journal_form.bank_statements_source = "online"
journal_form.online_bank_statement_provider = "dummy"
journal_form.save()
with self.assertRaises(IntegrityError), mute_logger("odoo.sql_db"):
journal.online_bank_statement_provider_id.unlink()
def test_cascade_unlink(self):
journal = self.AccountJournal.create(
{"name": "Bank", "type": "bank", "code": "BANK"}
)
with common.Form(journal) as journal_form:
journal_form.bank_statements_source = "online"
journal_form.online_bank_statement_provider = "dummy"
journal_form.save()
self.assertTrue(journal.online_bank_statement_provider_id)
journal.unlink()
self.assertFalse(self.OnlineBankStatementProvider.search([]))
def test_source_change_cleanup(self):
journal = self.AccountJournal.create(
{"name": "Bank", "type": "bank", "code": "BANK"}
)
with common.Form(journal) as journal_form:
journal_form.bank_statements_source = "online"
journal_form.online_bank_statement_provider = "dummy"
journal_form.save()
self.assertTrue(journal.online_bank_statement_provider_id)
with common.Form(journal) as journal_form:
journal_form.bank_statements_source = "undefined"
journal_form.save()
self.assertFalse(journal.online_bank_statement_provider_id)
self.assertFalse(self.OnlineBankStatementProvider.search([]))
def test_pull_mode_daily(self):
journal = self.AccountJournal.create(
{
"name": "Bank",
"type": "bank",
"code": "BANK",
"bank_statements_source": "online",
"online_bank_statement_provider": "dummy",
}
)
provider = journal.online_bank_statement_provider_id
provider.active = True
provider.statement_creation_mode = "daily"
provider.with_context(step={"hours": 2})._pull(
self.now - relativedelta(days=1),
self.now,
)
self.assertEqual(
len(self.AccountBankStatement.search([("journal_id", "=", journal.id)])), 2
)
def test_pull_mode_weekly(self):
journal = self.AccountJournal.create(
{
"name": "Bank",
"type": "bank",
"code": "BANK",
"bank_statements_source": "online",
"online_bank_statement_provider": "dummy",
}
)
provider = journal.online_bank_statement_provider_id
provider.active = True
provider.statement_creation_mode = "weekly"
provider.with_context(step={"hours": 8})._pull(
self.now - relativedelta(weeks=1),
self.now,
)
self.assertEqual(
len(self.AccountBankStatement.search([("journal_id", "=", journal.id)])), 2
)
def test_pull_mode_monthly(self):
journal = self.AccountJournal.create(
{
"name": "Bank",
"type": "bank",
"code": "BANK",
"bank_statements_source": "online",
"online_bank_statement_provider": "dummy",
}
)
provider = journal.online_bank_statement_provider_id
provider.active = True
provider.statement_creation_mode = "monthly"
provider.with_context(step={"hours": 8})._pull(
self.now - relativedelta(months=1),
self.now,
)
self.assertEqual(
len(self.AccountBankStatement.search([("journal_id", "=", journal.id)])), 2
)
def test_pull_scheduled(self):
journal = self.AccountJournal.create(
{
"name": "Bank",
"type": "bank",
"code": "BANK",
"bank_statements_source": "online",
"online_bank_statement_provider": "dummy",
}
)
provider = journal.online_bank_statement_provider_id
provider.active = True
provider.next_run = self.now - relativedelta(days=15)
self.assertFalse(
self.AccountBankStatement.search([("journal_id", "=", journal.id)])
)
provider.with_context(step={"hours": 8})._scheduled_pull()
statement = self.AccountBankStatement.search([("journal_id", "=", journal.id)])
self.assertEqual(len(statement), 1)
def test_pull_skip_duplicates_by_unique_import_id(self):
journal = self.AccountJournal.create(
{
"name": "Bank",
"type": "bank",
"code": "BANK",
"bank_statements_source": "online",
"online_bank_statement_provider": "dummy",
}
)
provider = journal.online_bank_statement_provider_id
provider.active = True
provider.statement_creation_mode = "weekly"
provider.with_context(
step={"hours": 8},
data_since=self.now - relativedelta(weeks=2),
data_until=self.now,
)._pull(
self.now - relativedelta(weeks=2),
self.now,
)
self.assertEqual(
len(
self.AccountBankStatementLine.search([("journal_id", "=", journal.id)])
),
14 * (24 / 8),
)
provider.with_context(
step={"hours": 8},
data_since=self.now - relativedelta(weeks=3),
data_until=self.now - relativedelta(weeks=1),
)._pull(
self.now - relativedelta(weeks=3),
self.now - relativedelta(weeks=1),
)
self.assertEqual(
len(
self.AccountBankStatementLine.search([("journal_id", "=", journal.id)])
),
21 * (24 / 8),
)
provider.with_context(
step={"hours": 8},
data_since=self.now - relativedelta(weeks=1),
data_until=self.now,
)._pull(
self.now - relativedelta(weeks=1),
self.now,
)
self.assertEqual(
len(
self.AccountBankStatementLine.search([("journal_id", "=", journal.id)])
),
21 * (24 / 8),
)
def test_interval_type_minutes(self):
journal = self.AccountJournal.create(
{
"name": "Bank",
"type": "bank",
"code": "BANK",
"bank_statements_source": "online",
"online_bank_statement_provider": "dummy",
}
)
provider = journal.online_bank_statement_provider_id
provider.active = True
provider.interval_type = "minutes"
provider._compute_update_schedule()
def test_interval_type_hours(self):
journal = self.AccountJournal.create(
{
"name": "Bank",
"type": "bank",
"code": "BANK",
"bank_statements_source": "online",
"online_bank_statement_provider": "dummy",
}
)
provider = journal.online_bank_statement_provider_id
provider.active = True
provider.interval_type = "hours"
provider._compute_update_schedule()
def test_interval_type_days(self):
journal = self.AccountJournal.create(
{
"name": "Bank",
"type": "bank",
"code": "BANK",
"bank_statements_source": "online",
"online_bank_statement_provider": "dummy",
}
)
provider = journal.online_bank_statement_provider_id
provider.active = True
provider.interval_type = "days"
provider._compute_update_schedule()
def test_interval_type_weeks(self):
journal = self.AccountJournal.create(
{
"name": "Bank",
"type": "bank",
"code": "BANK",
"bank_statements_source": "online",
"online_bank_statement_provider": "dummy",
}
)
provider = journal.online_bank_statement_provider_id
provider.active = True
provider.interval_type = "weeks"
provider._compute_update_schedule()
def test_pull_no_crash(self):
journal = self.AccountJournal.create(
{
"name": "Bank",
"type": "bank",
"code": "BANK",
"bank_statements_source": "online",
"online_bank_statement_provider": "dummy",
}
)
provider = journal.online_bank_statement_provider_id
provider.active = True
provider.statement_creation_mode = "weekly"
provider.with_context(crash=True, scheduled=True)._pull(
self.now - relativedelta(hours=1),
self.now,
)
self.assertFalse(
self.AccountBankStatement.search([("journal_id", "=", journal.id)])
)
def test_pull_crash(self):
journal = self.AccountJournal.create(
{
"name": "Bank",
"type": "bank",
"code": "BANK",
"bank_statements_source": "online",
"online_bank_statement_provider": "dummy",
}
)
provider = journal.online_bank_statement_provider_id
provider.active = True
provider.statement_creation_mode = "weekly"
with self.assertRaises(Exception):
provider.with_context(crash=True)._pull(
self.now - relativedelta(hours=1),
self.now,
)
def test_pull_httperror(self):
journal = self.AccountJournal.create(
{
"name": "Bank",
"type": "bank",
"code": "BANK",
"bank_statements_source": "online",
"online_bank_statement_provider": "dummy",
}
)
provider = journal.online_bank_statement_provider_id
provider.active = True
provider.statement_creation_mode = "weekly"
with self.assertRaises(HTTPError):
provider.with_context(
crash=True,
exception=HTTPError(None, 500, "Error", None, None),
)._pull(
self.now - relativedelta(hours=1),
self.now,
)
def test_pull_no_balance(self):
journal = self.AccountJournal.create(
{
"name": "Bank",
"type": "bank",
"code": "BANK",
"bank_statements_source": "online",
"online_bank_statement_provider": "dummy",
}
)
provider = journal.online_bank_statement_provider_id
provider.active = True
provider.statement_creation_mode = "daily"
provider.with_context(
step={"hours": 2},
balance_start=0,
amount=100.0,
balance=False,
)._pull(
self.now - relativedelta(days=1),
self.now,
)
statements = self.AccountBankStatement.search(
[("journal_id", "=", journal.id)],
order="date asc",
)
self.assertFalse(statements[0].balance_start)
self.assertFalse(statements[0].balance_end_real)
self.assertTrue(statements[0].balance_end)
self.assertTrue(statements[1].balance_start)
self.assertFalse(statements[1].balance_end_real)
def test_wizard(self):
journal = self.AccountJournal.create(
{
"name": "Bank",
"type": "bank",
"code": "BANK",
"bank_statements_source": "online",
"online_bank_statement_provider": "dummy",
}
)
action = journal.action_online_bank_statements_pull_wizard()
self.assertTrue(action["context"]["default_provider_ids"][0][2])
wizard = self.OnlineBankStatementPullWizard.with_context(
action["context"]
).create(
{"date_since": self.now - relativedelta(hours=1), "date_until": self.now}
)
self.assertTrue(wizard.provider_ids)
wizard.action_pull()
self.assertTrue(
self.AccountBankStatement.search([("journal_id", "=", journal.id)])
)
def test_pull_statement_partially(self):
journal = self.AccountJournal.create(
{
"name": "Bank",
"type": "bank",
"code": "BANK",
"bank_statements_source": "online",
"online_bank_statement_provider": "dummy",
}
)
provider = journal.online_bank_statement_provider_id
provider.active = True
provider.statement_creation_mode = "monthly"
provider_context = {
"step": {"hours": 24},
"data_since": datetime(2020, 1, 1),
"amount": 1.0,
"balance_start": 0,
}
provider.with_context(
**provider_context,
data_until=datetime(2020, 1, 31),
)._pull(
datetime(2020, 1, 1),
datetime(2020, 1, 31),
)
statements = self.AccountBankStatement.search(
[("journal_id", "=", journal.id)],
order="date asc",
)
self.assertEqual(len(statements), 1)
self.assertEqual(statements[0].balance_start, 0.0)
self.assertEqual(statements[0].balance_end_real, 30.0)
provider.with_context(
**provider_context,
data_until=datetime(2020, 2, 15),
)._pull(
datetime(2020, 1, 1),
datetime(2020, 2, 29),
)
statements = self.AccountBankStatement.search(
[("journal_id", "=", journal.id)],
order="date asc",
)
self.assertEqual(len(statements), 2)
self.assertEqual(statements[0].balance_start, 0.0)
self.assertEqual(statements[0].balance_end_real, 31.0)
self.assertEqual(statements[1].balance_start, 31.0)
self.assertEqual(statements[1].balance_end_real, 45.0)
provider.with_context(
**provider_context,
data_until=datetime(2020, 2, 29),
)._pull(
datetime(2020, 1, 1),
datetime(2020, 2, 29),
)
statements = self.AccountBankStatement.search(
[("journal_id", "=", journal.id)],
order="date asc",
)
self.assertEqual(len(statements), 2)
self.assertEqual(statements[0].balance_start, 0.0)
self.assertEqual(statements[0].balance_end_real, 31.0)
self.assertEqual(statements[1].balance_start, 31.0)
self.assertEqual(statements[1].balance_end_real, 59.0)
def test_tz_utc(self):
journal = self.AccountJournal.create(
{
"name": "Bank",
"type": "bank",
"code": "BANK",
"bank_statements_source": "online",
"online_bank_statement_provider": "dummy",
}
)
provider = journal.online_bank_statement_provider_id
provider.active = True
provider.tz = "UTC"
provider.with_context(
step={"hours": 1},
data_since=datetime(2020, 4, 17, 22, 0),
data_until=datetime(2020, 4, 18, 2, 0),
tz="UTC",
)._pull(
datetime(2020, 4, 17, 22, 0),
datetime(2020, 4, 18, 2, 0),
)
statement = self.AccountBankStatement.search([("journal_id", "=", journal.id)])
self.assertEqual(len(statement), 2)
lines = statement.mapped("line_ids").sorted()
self.assertEqual(len(lines), 4)
self.assertEqual(lines[0].date, date(2020, 4, 17))
self.assertEqual(lines[1].date, date(2020, 4, 17))
self.assertEqual(lines[2].date, date(2020, 4, 18))
self.assertEqual(lines[3].date, date(2020, 4, 18))
def test_tz_non_utc(self):
journal = self.AccountJournal.create(
{
"name": "Bank",
"type": "bank",
"code": "BANK",
"bank_statements_source": "online",
"online_bank_statement_provider": "dummy",
}
)
provider = journal.online_bank_statement_provider_id
provider.active = True
provider.tz = "Etc/GMT-2"
provider.with_context(
step={"hours": 1},
data_since=datetime(2020, 4, 17, 22, 0),
data_until=datetime(2020, 4, 18, 2, 0),
tz="UTC",
)._pull(
datetime(2020, 4, 17, 22, 0),
datetime(2020, 4, 18, 2, 0),
)
statement = self.AccountBankStatement.search([("journal_id", "=", journal.id)])
self.assertEqual(len(statement), 2)
lines = statement.mapped("line_ids").sorted()
self.assertEqual(len(lines), 4)
self.assertEqual(lines[0].date, date(2020, 4, 18))
self.assertEqual(lines[1].date, date(2020, 4, 18))
self.assertEqual(lines[2].date, date(2020, 4, 18))
self.assertEqual(lines[3].date, date(2020, 4, 18))
def test_other_tz_to_utc(self):
journal = self.AccountJournal.create(
{
"name": "Bank",
"type": "bank",
"code": "BANK",
"bank_statements_source": "online",
"online_bank_statement_provider": "dummy",
}
)
provider = journal.online_bank_statement_provider_id
provider.active = True
provider.with_context(
step={"hours": 1},
tz="Etc/GMT-2",
data_since=datetime(2020, 4, 18, 0, 0),
data_until=datetime(2020, 4, 18, 4, 0),
)._pull(
datetime(2020, 4, 17, 22, 0),
datetime(2020, 4, 18, 2, 0),
)
statement = self.AccountBankStatement.search([("journal_id", "=", journal.id)])
self.assertEqual(len(statement), 2)
lines = statement.mapped("line_ids").sorted()
self.assertEqual(len(lines), 4)
self.assertEqual(lines[0].date, date(2020, 4, 17))
self.assertEqual(lines[1].date, date(2020, 4, 17))
self.assertEqual(lines[2].date, date(2020, 4, 18))
self.assertEqual(lines[3].date, date(2020, 4, 18))
def test_timestamp_date_only_date(self):
journal = self.AccountJournal.create(
{
"name": "Bank",
"type": "bank",
"code": "BANK",
"bank_statements_source": "online",
"online_bank_statement_provider": "dummy",
}
)
provider = journal.online_bank_statement_provider_id
provider.active = True
provider.with_context(step={"hours": 1}, timestamp_mode="date")._pull(
datetime(2020, 4, 18, 0, 0),
datetime(2020, 4, 18, 4, 0),
)
statement = self.AccountBankStatement.search([("journal_id", "=", journal.id)])
self.assertEqual(len(statement), 1)
lines = statement.line_ids
self.assertEqual(len(lines), 24)
for line in lines:
self.assertEqual(line.date, date(2020, 4, 18))
def test_timestamp_date_only_str(self):
journal = self.AccountJournal.create(
{
"name": "Bank",
"type": "bank",
"code": "BANK",
"bank_statements_source": "online",
"online_bank_statement_provider": "dummy",
}
)
provider = journal.online_bank_statement_provider_id
provider.active = True
provider.with_context(
step={"hours": 1},
data_since=datetime(2020, 4, 18, 0, 0),
data_until=datetime(2020, 4, 18, 4, 0),
timestamp_mode="str",
)._pull(
datetime(2020, 4, 18, 0, 0),
datetime(2020, 4, 18, 4, 0),
)
statement = self.AccountBankStatement.search([("journal_id", "=", journal.id)])
self.assertEqual(len(statement), 1)
lines = statement.line_ids
self.assertEqual(len(lines), 4)
self.assertEqual(lines[0].date, date(2020, 4, 18))
self.assertEqual(lines[1].date, date(2020, 4, 18))
self.assertEqual(lines[2].date, date(2020, 4, 18))
self.assertEqual(lines[3].date, date(2020, 4, 18))