You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
645 lines
21 KiB
645 lines
21 KiB
# 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 dateutil.relativedelta import relativedelta
|
|
from psycopg2 import IntegrityError
|
|
from urllib.error import HTTPError
|
|
|
|
from odoo.tests import common
|
|
from odoo.tools import mute_logger
|
|
from odoo import fields
|
|
|
|
|
|
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_boundary(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(
|
|
expand_by=1,
|
|
)._pull(
|
|
self.now - relativedelta(hours=1),
|
|
self.now,
|
|
)
|
|
|
|
statement = self.AccountBankStatement.search(
|
|
[('journal_id', '=', journal.id)],
|
|
)
|
|
self.assertEqual(len(statement), 1)
|
|
self.assertEqual(len(statement.line_ids), 12)
|
|
|
|
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})._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})._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})._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),
|
|
'data_until': datetime(2020, 2, 29),
|
|
'amount': 1.0,
|
|
'balance_start': 0,
|
|
}
|
|
|
|
provider.with_context(**provider_context)._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)._pull(
|
|
datetime(2020, 1, 1),
|
|
datetime(2020, 2, 15),
|
|
)
|
|
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)._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},
|
|
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},
|
|
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(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), 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_timestamp_date_only(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='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))
|