audrius/bank-statement-import
1
0
Fork 1
mirror of https://github.com/OCA/bank-statement-import.git synced 2025-01-20 12:37:43 +02:00
Code Issues Packages Projects Releases Wiki Activity

[MIG] account_statement_import_online: Migration to 16.0

Browse Source
This commit is contained in:
Ronald Portier (Therp BV)
2023-01-27 22:53:52 +01:00
committed by Carolina Fernandez
parent d905b32731
commit 2fc9b110bb
17 changed files with 523 additions and 809 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
account_statement_import_online/__manifest__.py
View File

@@ -1,10 +1,11 @@
# Copyright 2019-2020 Brainbean Apps (https://brainbeanapps.com)
# Copyright 2020 CorporateHub (https://corporatehub.eu)
# Copyright 2023 Therp BV (https://therp.nl)
# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl).
{
"name": "Online Bank Statements",
"version": "15.0.1.0.0",
"version": "16.0.1.0.0",
"author": "CorporateHub, Odoo Community Association (OCA)",
"maintainers": ["alexey-pelykh"],
"website": "https://github.com/OCA/bank-statement-import",
@@ -14,7 +15,6 @@
"external_dependencies": {"python": ["odoo_test_helper"]},
"depends": [
"account_statement_import_base",
"web_widget_dropdown_dynamic",
],
"data": [
"data/account_statement_import_online.xml",

10
account_statement_import_online/i18n/account_statement_import_online.pot
View File

@@ -29,11 +29,6 @@ msgstr ""
msgid "Active"
msgstr ""
#. module: account_statement_import_online
#: model:ir.model.fields,field_description:account_statement_import_online.field_online_bank_statement_provider__allow_empty_statements
msgid "Allow empty statements"
msgstr ""
#. module: account_statement_import_online
#: model:ir.model.fields,field_description:account_statement_import_online.field_online_bank_statement_provider__api_base
msgid "Api Base"
@@ -359,11 +354,6 @@ msgstr ""
msgid "Provider"
msgstr ""
#. module: account_statement_import_online
#: model:ir.model.fields,field_description:account_statement_import_online.field_online_bank_statement_pull_wizard__provider_ids
msgid "Providers"
msgstr ""
#. module: account_statement_import_online
#: model_terms:ir.ui.view,arch_db:account_statement_import_online.online_bank_statement_pull_wizard_form
msgid "Pull"

10
account_statement_import_online/i18n/it.po
View File

@@ -32,11 +32,6 @@ msgstr ""
msgid "Active"
msgstr ""
#. module: account_statement_import_online
#: model:ir.model.fields,field_description:account_statement_import_online.field_online_bank_statement_provider__allow_empty_statements
msgid "Allow empty statements"
msgstr ""
#. module: account_statement_import_online
#: model:ir.model.fields,field_description:account_statement_import_online.field_online_bank_statement_provider__api_base
msgid "Api Base"
@@ -362,11 +357,6 @@ msgstr "Password"
msgid "Provider"
msgstr ""
#. module: account_statement_import_online
#: model:ir.model.fields,field_description:account_statement_import_online.field_online_bank_statement_pull_wizard__provider_ids
msgid "Providers"
msgstr ""
#. module: account_statement_import_online
#: model_terms:ir.ui.view,arch_db:account_statement_import_online.online_bank_statement_pull_wizard_form
msgid "Pull"

10
account_statement_import_online/i18n/nl.po
View File

@@ -32,11 +32,6 @@ msgstr "Actie vereist"
msgid "Active"
msgstr "Actief"
#. module: account_statement_import_online
#: model:ir.model.fields,field_description:account_statement_import_online.field_online_bank_statement_provider__allow_empty_statements
msgid "Allow empty statements"
msgstr ""
#. module: account_statement_import_online
#: model:ir.model.fields,field_description:account_statement_import_online.field_online_bank_statement_provider__api_base
msgid "Api Base"
@@ -365,11 +360,6 @@ msgstr "Wachtwoord"
msgid "Provider"
msgstr "Leverancier"
#. module: account_statement_import_online
#: model:ir.model.fields,field_description:account_statement_import_online.field_online_bank_statement_pull_wizard__provider_ids
msgid "Providers"
msgstr "Leveranciers"
#. module: account_statement_import_online
#: model_terms:ir.ui.view,arch_db:account_statement_import_online.online_bank_statement_pull_wizard_form
msgid "Pull"

112
account_statement_import_online/models/account_journal.py
View File

@@ -1,7 +1,7 @@
# Copyright 2019-2020 Brainbean Apps (https://brainbeanapps.com)
# Copyright 2019-2020 Dataplug (https://dataplug.io)
# Copyright 2023 Therp BV (https://therp.nl)
# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl.html).
import logging
from odoo import _, api, fields, models
@@ -12,20 +12,21 @@ _logger = logging.getLogger(__name__)
class AccountJournal(models.Model):
_inherit = "account.journal"
@api.model
def _selection_service(self):
OnlineBankStatementProvider = self.env["online.bank.statement.provider"]
return OnlineBankStatementProvider._get_available_services() + [
("dummy", "Dummy")
]
# Keep provider fields for compatibility with other modules.
online_bank_statement_provider = fields.Selection(
selection=lambda self: self.env[
"account.journal"
]._selection_online_bank_statement_provider(),
help="Select the type of service provider (a model)",
selection=lambda self: self._selection_service(),
)
online_bank_statement_provider_id = fields.Many2one(
string="Statement Provider",
comodel_name="online.bank.statement.provider",
ondelete="restrict",
copy=False,
help="Select the actual instance of a configured provider (a record).\n"
"Selecting a type of provider will automatically create a provider"
" record linked to this journal.",
)
def __get_bank_statements_available_sources(self):
@@ -33,52 +34,65 @@ class AccountJournal(models.Model):
result.append(("online", _("Online (OCA)")))
return result
@api.model
def _selection_online_bank_statement_provider(self):
return self.env["online.bank.statement.provider"]._get_available_services() + [
("dummy", "Dummy")
]
def _update_providers(self):
"""Automatically create service.
@api.model
def values_online_bank_statement_provider(self):
"""Return values for provider type selection in the form view."""
res = self.env["online.bank.statement.provider"]._get_available_services()
if self.user_has_groups("base.group_no_one"):
res += [("dummy", "Dummy")]
return res
def _update_online_bank_statement_provider_id(self):
"""Keep provider synchronized with journal."""
This method exists for compatibility reasons. The preferred method
to create an online provider is directly through the menu,
"""
OnlineBankStatementProvider = self.env["online.bank.statement.provider"]
for journal in self.filtered("id"):
provider_id = journal.online_bank_statement_provider_id
if journal.bank_statements_source != "online":
journal.online_bank_statement_provider_id = False
if provider_id:
provider_id.unlink()
for journal in self.filtered("online_bank_statement_provider"):
service = journal.online_bank_statement_provider
if (
journal.online_bank_statement_provider_id
and service == journal.online_bank_statement_provider_id.service
):
_logger.info(
"Journal %s already linked to service %s", journal.name, service
)
# Provider already exists.
continue
if provider_id.service == journal.online_bank_statement_provider:
continue
journal.online_bank_statement_provider_id = False
if provider_id:
provider_id.unlink()
# fmt: off
journal.online_bank_statement_provider_id = \
OnlineBankStatementProvider.create({
# Use existing or create new provider for service.
provider = OnlineBankStatementProvider.search(
[
("journal_id", "=", journal.id),
("service", "=", service),
],
limit=1,
) or OnlineBankStatementProvider.create(
{
"journal_id": journal.id,
"service": journal.online_bank_statement_provider,
})
# fmt: on
"service": service,
}
)
journal.online_bank_statement_provider_id = provider
_logger.info("Journal %s now linked to service %s", journal.name, service)
@api.model
def create(self, vals):
rec = super().create(vals)
if "bank_statements_source" in vals or "online_bank_statement_provider" in vals:
rec._update_online_bank_statement_provider_id()
return rec
@api.model_create_multi
def create(self, vals_list):
for vals in vals_list:
self._update_vals(vals)
journals = super().create(vals_list)
journals._update_providers()
return journals
def write(self, vals):
self._update_vals(vals)
res = super().write(vals)
if "bank_statements_source" in vals or "online_bank_statement_provider" in vals:
self._update_online_bank_statement_provider_id()
self._update_providers()
return res
def _update_vals(self, vals):
"""Ensure consistent values."""
if (
"bank_statements_source" in vals
and vals.get("bank_statements_source") != "online"
):
vals["online_bank_statement_provider"] = False
vals["online_bank_statement_provider_id"] = False
def action_online_bank_statements_pull_wizard(self):
"""This method is also kept for compatibility reasons."""
self.ensure_one()
provider = self.online_bank_statement_provider_id
return provider.action_online_bank_statements_pull_wizard()

304
account_statement_import_online/models/online_bank_statement_provider.py
View File

@@ -1,5 +1,6 @@
# Copyright 2019-2020 Brainbean Apps (https://brainbeanapps.com)
# Copyright 2019-2020 Dataplug (https://dataplug.io)
# Copyright 2022-2023 Therp BV (https://therp.nl)
# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl.html).
import logging
@@ -28,7 +29,6 @@ class OnlineBankStatementProvider(models.Model):
journal_id = fields.Many2one(
comodel_name="account.journal",
required=True,
readonly=True,
ondelete="cascade",
domain=[("type", "=", "bank")],
)
@@ -48,7 +48,6 @@ class OnlineBankStatementProvider(models.Model):
service = fields.Selection(
selection=lambda self: self._selection_service(),
required=True,
readonly=True,
)
interval_type = fields.Selection(
selection=[
@@ -93,7 +92,6 @@ class OnlineBankStatementProvider(models.Model):
certificate_public_key = fields.Text()
certificate_private_key = fields.Text()
certificate_chain = fields.Text()
allow_empty_statements = fields.Boolean()
_sql_constraints = [
(
@@ -108,6 +106,46 @@ class OnlineBankStatementProvider(models.Model):
),
]
@api.model
def create(self, vals):
"""Set provider_id on journal after creation."""
records = super().create(vals)
records._update_journals()
return records
def write(self, vals):
"""Set provider_id on journal after creation."""
result = super().write(vals)
self._update_journals()
return result
def _update_journals(self):
"""Update journal with this provider.
This is for compatibility reasons.
"""
for this in self:
this.journal_id.write(
{
"online_bank_statement_provider_id": this.id,
"online_bank_statement_provider": this.service,
"bank_statements_source": "online",
}
)
def unlink(self):
"""Reset journals."""
journals = self.mapped("journal_id")
if journals:
vals = {
"bank_statements_source": "undefined",
"online_bank_statement_provider": False,
"online_bank_statement_provider_id": False,
}
journals.write(vals)
result = super().unlink()
return result
@api.model
def _get_available_services(self):
"""Hook for extension"""
@@ -145,10 +183,14 @@ class OnlineBankStatementProvider(models.Model):
}
def _pull(self, date_since, date_until):
"""Pull data for all providers within requested period."""
is_scheduled = self.env.context.get("scheduled")
for provider in self:
statement_date_since = provider._get_statement_date_since(date_since)
while statement_date_since < date_until:
# Note that statement_date_until is exclusive, while date_until is
# inclusive. So if we have daily statements date_until might
# be 2020-01-31, while statement_date_until is 2020-02-01.
statement_date_until = (
statement_date_since + provider._get_statement_date_step()
)
@@ -156,32 +198,13 @@ class OnlineBankStatementProvider(models.Model):
data = provider._obtain_statement_data(
statement_date_since, statement_date_until
)
except BaseException as e:
if is_scheduled:
_logger.warning(
'Online Bank Statement Provider "%s" failed to'
" obtain statement data since %s until %s"
% (
provider.name,
statement_date_since,
statement_date_until,
),
exc_info=True,
)
provider.message_post(
body=_(
"Failed to obtain statement data for period "
"since {since} until {until}: {exception}. See server logs for "
"more details."
).format(
since=statement_date_since,
until=statement_date_until,
exception=escape(str(e)) or _("N/A"),
),
subject=_("Issue with Online Bank Statement Provider"),
)
break
except BaseException as exception:
if not is_scheduled:
raise
provider._log_provider_exception(
exception, statement_date_since, statement_date_until
)
break # Continue with next provider.
provider._create_or_update_statement(
data, statement_date_since, statement_date_until
)
@@ -189,69 +212,112 @@ class OnlineBankStatementProvider(models.Model):
if is_scheduled:
provider._schedule_next_run()
def _log_provider_exception(
self, exception, statement_date_since, statement_date_until
):
"""Both log error, and post a message on the provider record."""
self.ensure_one()
_logger.warning(
_(
'Online Bank Statement provider "%(name)s" failed to'
" obtain statement data since %(since)s until %(until)s"
),
dict(
name=self.name,
since=statement_date_since,
until=statement_date_until,
),
exc_info=True,
)
self.message_post(
body=_(
"Failed to obtain statement data for period "
"since {since} until {until}: {exception}. See server logs for "
"more details."
).format(
since=statement_date_since,
until=statement_date_until,
exception=escape(str(exception)) or _("N/A"),
),
subject=_("Issue with Online Bank Statement self"),
)
def _create_or_update_statement(
self, data, statement_date_since, statement_date_until
):
"""Create or update bank statement with the data retrieved from provider."""
"""Create or update bank statement with the data retrieved from provider.
We can not use statement.date as a unique key within the statements
of a journal, because this is now a computed field based on the last date in
the statement lines.
However we can still ensure unique and predictable names, so we wil use that
to find existing statements.
"""
self.ensure_one()
if not data:
data = ([], {})
unfiltered_lines, statement_values = data
if not unfiltered_lines:
unfiltered_lines = []
if not statement_values:
statement_values = {}
statement_values["name"] = self.make_statement_name(statement_date_since)
filtered_lines = self._get_statement_filtered_lines(
unfiltered_lines,
statement_values,
statement_date_since,
statement_date_until,
)
if not filtered_lines:
return self.env["account.bank.statement"]
if filtered_lines:
statement_values.update(
{"line_ids": [[0, False, line] for line in filtered_lines]}
)
self._update_statement_balances(statement_values)
statement = self._statement_create_or_write(statement_values)
return statement
def make_statement_name(self, statement_date_since):
"""Make name for statement using date and journal name."""
self.ensure_one()
return "%s/%s" % (
self.journal_id.code,
statement_date_since.strftime("%Y-%m-%d"),
)
def _statement_create_or_write(self, statement_values):
"""Final creation of statement if new, else write."""
AccountBankStatement = self.env["account.bank.statement"]
is_scheduled = self.env.context.get("scheduled")
if is_scheduled:
AccountBankStatement = AccountBankStatement.with_context(
tracking_disable=True,
)
if not data:
data = ([], {})
if not data[0] and not data[1] and not self.allow_empty_statements:
return
lines_data, statement_values = data
if not lines_data:
lines_data = []
if not statement_values:
statement_values = {}
statement_date = self._get_statement_date(
statement_date_since,
statement_date_until,
)
statement_name = statement_values["name"]
statement = AccountBankStatement.search(
[
("journal_id", "=", self.journal_id.id),
("state", "=", "open"),
("date", "=", statement_date),
("name", "=", statement_name),
],
limit=1,
)
if not statement:
statement_values.update(
{
"name": "%s/%s"
% (self.journal_id.code, statement_date.strftime("%Y-%m-%d")),
"journal_id": self.journal_id.id,
"date": statement_date,
}
)
statement_values["journal_id"] = self.journal_id.id
statement = AccountBankStatement.with_context(
journal_id=self.journal_id.id,
).create(
# NOTE: This is needed since create() alters values
statement_values.copy()
)
filtered_lines = self._get_statement_filtered_lines(
lines_data, statement_values, statement_date_since, statement_date_until
)
statement_values.update(
{"line_ids": [[0, False, line] for line in filtered_lines]}
)
if "balance_start" in statement_values:
statement_values["balance_start"] = float(statement_values["balance_start"])
if "balance_end_real" in statement_values:
statement_values["balance_end_real"] = float(
statement_values["balance_end_real"]
)
).create(statement_values)
else:
statement.write(statement_values)
return statement
def _get_statement_filtered_lines(
self, lines_data, statement_values, statement_date_since, statement_date_until
self,
unfiltered_lines,
statement_values,
statement_date_since,
statement_date_until,
):
"""Get lines from line data, but only for the right date."""
AccountBankStatementLine = self.env["account.bank.statement.line"]
@@ -259,7 +325,10 @@ class OnlineBankStatementProvider(models.Model):
journal = self.journal_id
speeddict = journal._statement_line_import_speeddict()
filtered_lines = []
for line_values in lines_data:
lines_before_since = 0
lines_after_until = 0
lines_not_unique = 0
for line_values in unfiltered_lines:
date = line_values["date"]
if not isinstance(date, datetime):
date = fields.Datetime.from_string(date)
@@ -271,12 +340,14 @@ class OnlineBankStatementProvider(models.Model):
statement_values["balance_start"] = Decimal(
statement_values["balance_start"]
) + Decimal(line_values["amount"])
lines_before_since += 1
continue
elif date >= statement_date_until:
if "balance_end_real" in statement_values:
statement_values["balance_end_real"] = Decimal(
statement_values["balance_end_real"]
) - Decimal(line_values["amount"])
lines_after_until += 1
continue
date = date.replace(tzinfo=utc)
date = date.astimezone(provider_tz).replace(tzinfo=None)
@@ -289,13 +360,56 @@ class OnlineBankStatementProvider(models.Model):
if AccountBankStatementLine.sudo().search(
[("unique_import_id", "=", unique_import_id)], limit=1
):
lines_not_unique += 1
continue
if not line_values.get("payment_ref"):
line_values["payment_ref"] = line_values.get("ref")
line_values["journal_id"] = self.journal_id.id
journal._statement_line_import_update_hook(line_values, speeddict)
filtered_lines.append(line_values)
if unfiltered_lines:
if len(unfiltered_lines) == len(filtered_lines):
_logger.debug(_("All lines passed filtering"))
else:
_logger.debug(
_(
"Of %(lines_provided)s lines provided"
", %(before)s where before %(since)s"
", %(after)s where on or after %(until)s"
"and %(duplicate)s where not unique."
),
dict(
lines_provided=len(unfiltered_lines),
before=lines_before_since,
since=statement_date_since,
after=lines_after_until,
until=statement_date_until,
duplicate=lines_not_unique,
),
)
return filtered_lines
def _update_statement_balances(self, statement_values):
"""Update statement balance_ start/end/end_real."""
AccountBankStatement = self.env["account.bank.statement"]
if "balance_start" in statement_values:
statement_values["balance_start"] = float(statement_values["balance_start"])
else:
# Take balance_end of previous statement as start of this one.
previous_statement = AccountBankStatement.search(
[
("journal_id", "=", self.journal_id.id),
("name", "<", statement_values["name"]),
],
limit=1,
)
if previous_statement and previous_statement.balance_end:
statement_values["balance_start"] = previous_statement.balance_end
if "balance_end_real" in statement_values:
statement_values["balance_end_real"] = float(
statement_values["balance_end_real"]
)
def _schedule_next_run(self):
self.ensure_one()
self.last_successful_run = self.next_run
@@ -334,15 +448,6 @@ class OnlineBankStatementProvider(models.Model):
microsecond=0,
)
def _get_statement_date(self, date_since, date_until):
self.ensure_one()
# NOTE: Statement date is treated by Odoo as start of period. Details
# - addons/account/models/account_journal_dashboard.py
# - def get_line_graph_datas()
tz = timezone(self.tz) if self.tz else utc
date_since = date_since.replace(tzinfo=utc).astimezone(tz)
return date_since.date()
def _get_next_run_period(self):
self.ensure_one()
if self.interval_type == "minutes":
@@ -356,17 +461,19 @@ class OnlineBankStatementProvider(models.Model):
@api.model
def _scheduled_pull(self):
_logger.info("Scheduled pull of online bank statements...")
_logger.info(_("Scheduled pull of online bank statements..."))
providers = self.search(
[("active", "=", True), ("next_run", "<=", fields.Datetime.now())]
)
if providers:
_logger.info(
"Pulling online bank statements of: %s"
% ", ".join(providers.mapped("journal_id.name"))
_("Pulling online bank statements of: %(provider_names)s"),
dict(provider_names=", ".join(providers.mapped("journal_id.name"))),
)
for provider in providers.with_context(**{"scheduled": True}):
for provider in providers.with_context(
scheduled=True, tracking_disable=True
):
provider._adjust_schedule()
date_since = (
(provider.last_successful_run)
if provider.last_successful_run
@@ -374,11 +481,38 @@ class OnlineBankStatementProvider(models.Model):
)
date_until = provider.next_run
provider._pull(date_since, date_until)
_logger.info(_("Scheduled pull of online bank statements complete."))
_logger.info("Scheduled pull of online bank statements complete.")
def _adjust_schedule(self):
"""Make sure next_run is current.
Current means adding one more period would put if after the
current moment. This will be done at the end of the run.
The net effect of this method and the adjustment after the run
will be for the next_run to be in the future.
"""
self.ensure_one()
delta = self._get_next_run_period()
now = datetime.now()
next_run = self.next_run + delta
while next_run < now:
self.next_run = next_run
next_run = self.next_run + delta
def _obtain_statement_data(self, date_since, date_until):
"""Hook for extension"""
# Check tests/online_bank_statement_provider_dummy.py for reference
self.ensure_one()
return []
def action_online_bank_statements_pull_wizard(self):
self.ensure_one()
WIZARD_MODEL = "online.bank.statement.pull.wizard"
wizard = self.env[WIZARD_MODEL].create([{"provider_ids": [(6, 0, [self.id])]}])
return {
"type": "ir.actions.act_window",
"res_model": WIZARD_MODEL,
"res_id": wizard.id,
"view_mode": "form",
"target": "new",
}

20
account_statement_import_online/readme/CONFIGURE.rst
View File

@@ -1,23 +1,7 @@
To configure online bank statements provider:
#. Go to *Invoicing > Configuration > Bank Accounts*
#. Open bank account to configure and edit it
#. Set *Bank Feeds* to *Online*
#. Select online bank statements provider in *Online Bank Statements (OCA)*
section
#. Save the bank account
#. Click on provider and configure provider-specific settings.
or, alternatively:
#. Go to *Invoicing > Overview*
#. Open settings of the corresponding journal account
#. Switch to *Bank Account* tab
#. Set *Bank Feeds* to *Online*
#. Select online bank statements provider in *Online Bank Statements (OCA)*
section
#. Save the bank account
#. Click on provider and configure provider-specific settings.
#. Go to *Invoicing > Configuration > Online Bank Statement Providers*
#. Create a provider and configure provider-specific settings.
If you want to allow empty bank statements to be created every time the
information is pulled, you can check the option "Allow empty statements"

6
account_statement_import_online/readme/USAGE.rst
View File

@@ -1,8 +1,8 @@
To pull historical bank statements:
#. Go to *Invoicing > Configuration > Bank Accounts*
#. Select specific bank accounts
#. Launch *Actions > Online Bank Statements Pull Wizard*
#. Go to *Invoicing > Configuration > Online Bank Statement Providers*
#. Select a specific provider
#. Click on *PULL ONLINE BANK STATEMENT*
#. Configure date interval and click *Pull*
**NOTE**: To access these features, user needs to belong to

1
account_statement_import_online/tests/__init__.py
View File

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

13
account_statement_import_online/tests/online_bank_statement_provider_dummy.py
View File

@@ -29,10 +29,11 @@ class OnlineBankStatementProviderDummy(models.Model):
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
# Override date_since and date_until from context.
override_date_since = self.env.context.get("override_date_since", date_since)
override_date_until = self.env.context.get("override_date_until", date_until)
override_date_since -= expand_by * line_step
override_date_until += expand_by * line_step
balance_start = self.env.context.get(
"balance_start", randrange(-10000, 10000, 1) * 0.1
@@ -46,8 +47,8 @@ class OnlineBankStatementProviderDummy(models.Model):
timestamp_mode = self.env.context.get("timestamp_mode")
lines = []
date = data_since
while date < data_until:
date = override_date_since
while date < override_date_until:
amount = self.env.context.get("amount", randrange(-100, 100, 1) * 0.1)
transaction_date = date.replace(tzinfo=tz)
if timestamp_mode == "date":

709
account_statement_import_online/tests/test_account_bank_statement_import_online.py
View File

@@ -1,18 +1,19 @@
# Copyright 2019-2020 Brainbean Apps (https://brainbeanapps.com)
# Copyright 2019-2020 Dataplug (https://dataplug.io)
# Copyright 2022-2023 Therp BV (https://therp.nl)
# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl).
import logging
from datetime import date, datetime
from unittest import mock
from urllib.error import HTTPError
from dateutil.relativedelta import relativedelta
from odoo_test_helper import FakeModelLoader
from psycopg2 import IntegrityError
from odoo import fields
from odoo import _, fields
from odoo.tests import common
from odoo.tools import mute_logger
_logger = logging.getLogger(__name__)
mock_obtain_statement_data = (
"odoo.addons.account_statement_import_online.tests."
@@ -37,352 +38,131 @@ class TestAccountBankAccountStatementImportOnline(common.TransactionCase):
cls.loader.update_registry((OnlineBankStatementProviderDummy,))
cls.now = fields.Datetime.now()
cls.AccountAccount = cls.env["account.account"]
cls.AccountJournal = cls.env["account.journal"]
cls.OnlineBankStatementProvider = cls.env["online.bank.statement.provider"]
cls.OnlineBankStatementPullWizard = cls.env["online.bank.statement.pull.wizard"]
cls.AccountBankStatement = cls.env["account.bank.statement"]
cls.AccountBankStatementLine = cls.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)
save_provider_id = journal.online_bank_statement_provider_id.id
journal.unlink()
self.assertFalse(
self.OnlineBankStatementProvider.search(
[
("id", "=", save_provider_id),
]
)
)
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)
save_provider_id = journal.online_bank_statement_provider_id.id
# Stuff should not change when doing unrelated write.
journal.write({"code": "BIGBANK"})
self.assertTrue(journal.online_bank_statement_provider_id)
self.assertEqual(journal.online_bank_statement_provider_id.id, save_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(
[
("id", "=", save_provider_id),
]
cls.journal = cls.AccountJournal.create(
{
"name": "Bank",
"type": "bank",
"code": "BANK",
"bank_statements_source": "online",
}
)
cls.provider = cls.OnlineBankStatementProvider.create(
{
"name": "Dummy Provider",
"service": "dummy",
"journal_id": cls.journal.id,
"statement_creation_mode": "daily",
}
)
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.provider.statement_creation_mode = "daily"
self.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
)
self._getExpectedStatements(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.provider.statement_creation_mode = "weekly"
self.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
)
self._getExpectedStatements(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.provider.statement_creation_mode = "monthly"
self.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
)
self._getExpectedStatements(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)
self.provider.next_run = self.now - relativedelta(days=15)
self._getExpectedStatements(0)
self.provider.with_context(step={"hours": 8})._scheduled_pull()
self._getExpectedStatements(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(
self.provider.statement_creation_mode = "weekly"
# Get for two weeks of data.
self.provider.with_context(
step={"hours": 8},
data_since=self.now - relativedelta(weeks=2),
data_until=self.now,
override_date_since=self.now - relativedelta(weeks=2),
override_date_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(
expected_count = 14 * (24 / 8)
self._getExpectedLines(expected_count)
# Get two weeks, but one overlapping with previous.
self.provider.with_context(
step={"hours": 8},
data_since=self.now - relativedelta(weeks=3),
data_until=self.now - relativedelta(weeks=1),
override_date_since=self.now - relativedelta(weeks=3),
override_date_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(
expected_count = 21 * (24 / 8)
self._getExpectedLines(expected_count)
# Get another day, but within statements already retrieved.
self.provider.with_context(
step={"hours": 8},
data_since=self.now - relativedelta(weeks=1),
data_until=self.now,
override_date_since=self.now - relativedelta(weeks=1),
override_date_until=self.now,
)._pull(
self.now - relativedelta(weeks=1),
self.now,
)
self.assertEqual(
len(
self.AccountBankStatementLine.search([("journal_id", "=", journal.id)])
),
21 * (24 / 8),
)
self._getExpectedLines(expected_count)
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()
self.provider.interval_type = "minutes"
self.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()
self.provider.interval_type = "hours"
self.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()
self.provider.interval_type = "days"
self.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()
self.provider.interval_type = "weeks"
self.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.provider.statement_creation_mode = "weekly"
self.provider.with_context(crash=True, scheduled=True)._pull(
self.now - relativedelta(hours=1),
self.now,
)
self.assertFalse(
self.AccountBankStatement.search([("journal_id", "=", journal.id)])
)
self._getExpectedStatements(0)
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"
self.provider.statement_creation_mode = "weekly"
with self.assertRaisesRegex(Exception, "Expected"):
provider.with_context(crash=True)._pull(
self.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"
self.provider.statement_creation_mode = "weekly"
with self.assertRaises(HTTPError):
provider.with_context(
self.provider.with_context(
crash=True,
exception=HTTPError(None, 500, "Error", None, None),
)._pull(
@@ -391,21 +171,7 @@ class TestAccountBankAccountStatementImportOnline(common.TransactionCase):
)
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(
self.provider.with_context(
step={"hours": 2},
balance_start=0,
amount=100.0,
@@ -414,135 +180,93 @@ class TestAccountBankAccountStatementImportOnline(common.TransactionCase):
self.now - relativedelta(days=1),
self.now,
)
statements = self.AccountBankStatement.search(
[("journal_id", "=", journal.id)],
order="date asc",
)
statements = self._getExpectedStatements(2)
self.assertFalse(statements[0].balance_start)
self.assertTrue(statements[0].balance_end)
self.assertTrue(statements[1].balance_start)
def test_wizard(self):
journal = self.AccountJournal.create(
{
"name": "Bank",
"type": "bank",
"code": "BANK",
"bank_statements_source": "online",
"online_bank_statement_provider": "dummy",
vals = {
"date_since": self.now - relativedelta(hours=1),
"date_until": self.now,
}
)
vals = self.OnlineBankStatementPullWizard.with_context(
active_model="account.journal", active_id=journal.id
).default_get(fields_list=["provider_ids"])
vals["date_since"] = self.now - relativedelta(hours=1)
vals["date_until"] = self.now
wizard = self.OnlineBankStatementPullWizard.create(vals)
self.assertTrue(wizard.provider_ids)
wizard = self.OnlineBankStatementPullWizard.with_context(
active_model=self.provider._name, active_id=self.provider.id
).create(vals)
wizard.action_pull()
self.assertTrue(
self.AccountBankStatement.search([("journal_id", "=", journal.id)])
)
self._getExpectedStatements(1)
def test_wizard_on_journal(self):
vals = {
"date_since": self.now - relativedelta(hours=1),
"date_until": self.now,
}
wizard = self.OnlineBankStatementPullWizard.with_context(
active_model=self.journal._name, active_id=self.journal.id
).create(vals)
wizard.action_pull()
self._getExpectedStatements(1)
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"
self.provider.statement_creation_mode = "monthly"
provider_context = {
"step": {"hours": 24},
"data_since": datetime(2020, 1, 1),
"override_date_since": datetime(2020, 1, 1),
"amount": 1.0,
"balance_start": 0,
}
provider.with_context(
# Should create statement for first 30 days of january.
self.provider.with_context(
**provider_context,
data_until=datetime(2020, 1, 31),
override_date_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)
statements = self._getExpectedStatements(1)
self.assertEqual(statements[0].balance_start, 0.0)
self.assertEqual(statements[0].balance_end_real, 30.0)
provider.with_context(
# Should create statement for first 14 days of february,
# and add one line to statement for january.
self.provider.with_context(
**provider_context,
data_until=datetime(2020, 2, 15),
override_date_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)
statements = self._getExpectedStatements(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(
# Getting data for rest of februari should not create new statement.
self.provider.with_context(
**provider_context,
data_until=datetime(2020, 2, 29),
override_date_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)
statements = self._getExpectedStatements(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(
self.provider.tz = "UTC"
self.provider.with_context(
step={"hours": 1},
data_since=datetime(2020, 4, 17, 22, 0),
data_until=datetime(2020, 4, 18, 2, 0),
override_date_since=datetime(2020, 4, 17, 22, 0),
override_date_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()
statements = self._getExpectedStatements(2)
lines = statements.mapped("line_ids").sorted(key=lambda r: r.id)
self.assertEqual(len(lines), 4)
self.assertEqual(lines[0].date, date(2020, 4, 17))
self.assertEqual(lines[1].date, date(2020, 4, 17))
@@ -550,33 +274,23 @@ class TestAccountBankAccountStatementImportOnline(common.TransactionCase):
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",
}
)
"""Test situation where the provider is west of Greenwich.
provider = journal.online_bank_statement_provider_id
provider.active = True
provider.tz = "Etc/GMT-2"
provider.with_context(
In this case, when it is 22:00 according to the provider, it is
00:00 the next day according to GMT/UTZ.
"""
self.provider.tz = "Etc/GMT-2"
self.provider.with_context(
step={"hours": 1},
data_since=datetime(2020, 4, 17, 22, 0),
data_until=datetime(2020, 4, 18, 2, 0),
override_date_since=datetime(2020, 4, 17, 22, 0),
override_date_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()
statements = self._getExpectedStatements(2)
lines = statements.mapped("line_ids").sorted(key=lambda r: r.id)
self.assertEqual(len(lines), 4)
self.assertEqual(lines[0].date, date(2020, 4, 18))
self.assertEqual(lines[1].date, date(2020, 4, 18))
@@ -584,32 +298,25 @@ class TestAccountBankAccountStatementImportOnline(common.TransactionCase):
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",
}
)
"""Test the situation where we are tot the west of the provider.
provider = journal.online_bank_statement_provider_id
provider.active = True
provider.with_context(
Provider will be GMT/UTC, we will be two hours to the west.
When we pull data from 22:00 on the 17th of april, for
the provider this will be from 00:00 on the 18th.
We will translate the provider times back to our time.
"""
self.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),
override_date_since=datetime(2020, 4, 18, 0, 0),
override_date_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()
statements = self._getExpectedStatements(2)
lines = statements.mapped("line_ids").sorted(key=lambda r: r.id)
self.assertEqual(len(lines), 4)
self.assertEqual(lines[0].date, date(2020, 4, 17))
self.assertEqual(lines[1].date, date(2020, 4, 17))
@@ -617,58 +324,28 @@ class TestAccountBankAccountStatementImportOnline(common.TransactionCase):
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(
self.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
statements = self._getExpectedStatements(1)
lines = statements.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(
self.provider.with_context(
step={"hours": 1},
data_since=datetime(2020, 4, 18, 0, 0),
data_until=datetime(2020, 4, 18, 4, 0),
override_date_since=datetime(2020, 4, 18, 0, 0),
override_date_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
statements = self._getExpectedStatements(1)
lines = statements.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))
@@ -692,18 +369,6 @@ class TestAccountBankAccountStatementImportOnline(common.TransactionCase):
statements ('Allow empty statements' field is uncheck at the
provider level.).
"""
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"
with mock.patch(mock_obtain_statement_data) as mock_data:
mock_data.side_effect = [
self._get_statement_line_data(date(2021, 8, 10)),
@@ -711,57 +376,53 @@ class TestAccountBankAccountStatementImportOnline(common.TransactionCase):
([], {}), # August 9th, doesn't have statement
self._get_statement_line_data(date(2021, 8, 13)),
]
provider._pull(datetime(2021, 8, 10), datetime(2021, 8, 14))
statements = self.AccountBankStatement.search([("journal_id", "=", journal.id)])
self.assertEqual(len(statements), 2)
self.assertEqual(statements[1].balance_start, 0)
self.assertEqual(statements[1].balance_end_real, 100)
self.assertEqual(len(statements[1].line_ids), 1)
self.assertEqual(statements[0].balance_start, 100)
self.assertEqual(statements[0].balance_end_real, 200)
self.provider._pull(datetime(2021, 8, 10), datetime(2021, 8, 14))
statements = self._getExpectedStatements(2)
self.assertEqual(statements[0].balance_start, 0)
self.assertEqual(statements[0].balance_end, 100)
self.assertEqual(len(statements[0].line_ids), 1)
def test_create_empty_statements(self):
"""Test creating empty bank statements
('Allow empty statements' field is check at the provider level).
"""
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.allow_empty_statements = True
provider.statement_creation_mode = "daily"
with mock.patch(mock_obtain_statement_data) as mock_data:
mock_data.side_effect = [
self._get_statement_line_data(date(2021, 8, 10)),
([], {}), # August 8th, doesn't have statement
([], {}), # August 9th, doesn't have statement
self._get_statement_line_data(date(2021, 8, 13)),
]
provider._pull(datetime(2021, 8, 10), datetime(2021, 8, 14))
statements = self.AccountBankStatement.search([("journal_id", "=", journal.id)])
# 4 Statements: 2 with movements and 2 empty
self.assertEqual(len(statements), 4)
# With movement
self.assertEqual(statements[3].balance_start, 0)
self.assertEqual(statements[3].balance_end_real, 100)
self.assertEqual(len(statements[3].line_ids), 1)
# Empty
self.assertEqual(statements[2].balance_start, 100)
self.assertEqual(statements[2].balance_end_real, 100)
self.assertEqual(len(statements[2].line_ids), 0)
# Empty
self.assertEqual(statements[1].balance_start, 100)
self.assertEqual(statements[1].balance_end_real, 100)
self.assertEqual(len(statements[1].line_ids), 0)
# With movement
self.assertEqual(statements[0].balance_start, 100)
self.assertEqual(statements[0].balance_end_real, 200)
self.assertEqual(len(statements[0].line_ids), 1)
self.assertEqual(statements[1].balance_end, 200)
self.assertEqual(len(statements[1].line_ids), 1)
def test_unlink_provider(self):
"""Unlink provider should clear fields on journal."""
self.provider.unlink()
self.assertEqual(self.journal.bank_statements_source, "undefined")
self.assertEqual(self.journal.online_bank_statement_provider, False)
self.assertEqual(self.journal.online_bank_statement_provider_id.id, False)
def _getExpectedStatements(self, expected_length):
"""Check for length of statement recordset, with helpfull logging."""
statements = self.AccountBankStatement.search(
[("journal_id", "=", self.journal.id)], order="date asc"
)
actual_length = len(statements)
# If length not expected, log information about statements.
if actual_length != expected_length:
if actual_length == 0:
_logger.warning(
_("No statements found in journal"),
)
else:
_logger.warning(
_("Names and dates for statements found: %(statements)s"),
dict(
statements=", ".join(
["%s - %s" % (stmt.name, stmt.date) for stmt in statements]
)
),
)
# Now do the normal assert.
self.assertEqual(len(statements), expected_length)
# If we got expected number, return them.
return statements
def _getExpectedLines(self, expected_length):
"""Check number of lines created."""
lines = self.AccountBankStatementLine.search(
[("journal_id", "=", self.journal.id)]
)
self.assertEqual(len(lines), expected_length)
# If we got expected number, return them.
return lines

22
account_statement_import_online/tests/test_account_journal.py
View File

@@ -1,22 +0,0 @@
# Copyright 2021 Therp BV (https://therp.nl).
# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl).
from unittest.mock import patch
from odoo.tests import common
class TestAccountJournal(common.TransactionCase):
"""Test some functions adde d to account.journal model."""
def setUp(self):
super().setUp()
self.AccountJournal = self.env["account.journal"]
def test_values_online_bank_statement_provider(self):
"""Check method to retrieve provider types."""
# Make sure the users seems to have the group_no_one.
with patch.object(
self.AccountJournal.__class__, "user_has_groups", return_value=True
):
values = self.AccountJournal.values_online_bank_statement_provider()
self.assertIn("dummy", [entry[0] for entry in values])

12
account_statement_import_online/views/account_journal.xml
View File

@@ -11,7 +11,10 @@
<field name="model">account.journal</field>
<field name="inherit_id" ref="account.view_account_journal_form" />
<field name="arch" type="xml">
<xpath expr="//field[@name='bank_statements_source']/.." position="after">
<xpath
expr="//notebook//field[@name='bank_statements_source']/.."
position="after"
>
<group
name="online_bank_statements"
string="Online Bank Statements (OCA)"
@@ -31,8 +34,6 @@
attrs="{'required': [('bank_statements_source', '=', 'online')]}"
class="oe_edit_only"
groups="account.group_account_user"
widget="dynamic_dropdown"
values="values_online_bank_statement_provider"
/>
<label
for="online_bank_statement_provider_id"
@@ -48,15 +49,14 @@
/>
</group>
</xpath>
<xpath expr="/form/sheet" position="before">
<header>
<xpath expr="//div[@name='button_box']" position="inside">
<button
type="action"
name="%(action_online_bank_statements_pull_wizard)d"
attrs="{'invisible': [('online_bank_statement_provider', '=', False)]}"
string="Pull Online Bank Statement"
groups="account.group_account_user"
/>
</header>
</xpath>
</field>
</record>

4
account_statement_import_online/views/actions.xml
View File

@@ -12,8 +12,8 @@
<field name="res_model">online.bank.statement.pull.wizard</field>
<field name="target">new</field>
<field name="view_mode">form</field>
<field name="binding_model_id" ref="account.model_account_journal" />
<field name="binding_view_types">list</field>
<field name="binding_model_id" eval="False" />
<field name="binding_view_types" eval="False" />
<field name="groups_id" eval="[(4, ref('account.group_account_user'))]" />
</record>
</odoo>

22
account_statement_import_online/views/online_bank_statement_provider.xml
View File

@@ -2,6 +2,7 @@
<!--
Copyright 2019 Brainbean Apps (https://brainbeanapps.com)
Copyright 2019 Dataplug (https://dataplug.io)
Copyright 2021-2023 Therp BV (https://therp.nl)
License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl).
-->
<odoo>
@@ -12,8 +13,8 @@
<search string="Online Bank Statement Providers">
<field name="journal_id" />
<filter
name="active"
string="Inactive"
name="archived"
string="Archived"
domain="[('active', '=', False)]"
/>
<group name="groupby">
@@ -45,6 +46,15 @@
<field name="model">online.bank.statement.provider</field>
<field name="arch" type="xml">
<form string="Online Bank Statement Provider">
<field name="active" invisible="1" />
<header>
<button
type="action"
name="%(action_online_bank_statements_pull_wizard)d"
attrs="{'invisible': [('active', '=', False)]}"
string="Pull Online Bank Statement"
/>
</header>
<sheet>
<widget
name="web_ribbon"
@@ -61,12 +71,7 @@
name="company_id"
groups="base.group_multi_company"
/>
<field
name="service"
widget="dynamic_dropdown"
values="values_service"
/>
<field name="active" invisible="1" />
<field name="service" />
</group>
<group name="pull" string="Scheduled Pull">
<label for="interval_number" />
@@ -79,7 +84,6 @@
<group name="configuration" string="Configuration">
<field name="statement_creation_mode" />
<field name="tz" />
<field name="allow_empty_statements" />
</group>
</group>
</sheet>

49
account_statement_import_online/wizards/online_bank_statement_pull_wizard.py
View File

@@ -1,8 +1,9 @@
# Copyright 2019-2020 Brainbean Apps (https://brainbeanapps.com)
# Copyright 2019-2020 Dataplug (https://dataplug.io)
# Copyright 2023 Therp BV (https://therp.nl)
# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl).
from odoo import api, fields, models
from odoo import fields, models
class OnlineBankStatementPullWizard(models.TransientModel):
@@ -19,44 +20,18 @@ class OnlineBankStatementPullWizard(models.TransientModel):
required=True,
default=fields.Datetime.now,
)
# The link to providers is Many2many, because you can select multiple
# journals for the action to pull statements.
provider_ids = fields.Many2many(
string="Providers",
comodel_name="online.bank.statement.provider",
column1="wizard_id",
column2="provider_id",
relation="online_bank_statement_provider_pull_wizard_rel",
)
@api.model
def default_get(self, fields_list):
"""Retrieve providers from the journals for which this wizard is launched."""
res = super().default_get(fields_list)
journal_ids = []
if self.env.context.get("active_model") == "account.journal":
if self.env.context.get("active_ids"):
journal_ids = self.env.context["active_ids"]
elif self.env.context.get("active_id"):
journal_ids = [self.env.context["active_id"]]
if journal_ids:
journals = self.env["account.journal"].browse(journal_ids)
res["provider_ids"] = [journals.online_bank_statement_provider_id.id]
return res
def action_pull(self):
"""Pull statements from providers and then show list of statements."""
"""Pull statements from provider and then show list of statements."""
self.ensure_one()
self.with_context(active_test=False).provider_ids._pull(
self.date_since, self.date_until
)
action = self.env.ref("account.action_bank_statement_tree").sudo().read([])[0]
if len(self.provider_ids) == 1:
action["context"] = {
"search_default_journal_id": self.provider_ids[0].journal_id.id
}
active_model = self.env.context.get("active_model")
active_id = self.env.context.get("active_id")
active_record = self.env[active_model].browse(active_id)
if active_model == "account.journal":
provider = active_record.online_bank_statement_provider_id
else:
action["domain"] = [
("journal_id", "in", [o.journal_id.id for o in self.provider_ids])
]
provider = active_record
provider._pull(self.date_since, self.date_until)
action = self.env.ref("account.action_bank_statement_tree").sudo().read([])[0]
action["domain"] = [("journal_id", "=", provider.journal_id.id)]
return action

6
account_statement_import_online/wizards/online_bank_statement_pull_wizard.xml
View File

@@ -11,12 +11,6 @@
<field name="arch" type="xml">
<form>
<group name="main">
<field
name="provider_ids"
widget="many2many_tags"
required="1"
options="{'create': false, 'create_edit': false}"
/>
<field name="date_since" />
<field name="date_until" />
</group>