admin/Odoo-Modules
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(fusion_accounting_bank_rec): reconcile engine 6-method public API

Browse Source 
Adds fusion.reconcile.engine — the AbstractModel orchestrator for all
bank-line reconciliations. Six public methods (reconcile_one,
reconcile_batch, suggest_matches, accept_suggestion, write_off,
unreconcile) form the only sanctioned write path to
account.partial.reconcile from the rest of the module (controllers, AI
tools, wizards).

Implementation follows V19's bank_rec_widget pattern: rewrite the bank
move's suspense line into one counterpart per matched invoice (or a
write-off line) on the appropriate receivable / payable / write-off
account, then call account.move.line.reconcile() on each pair. Records
a precedent row per reconcile for downstream pattern learning.

16 new unit tests cover all six methods across happy paths, the
precedent side effect, suggestion lifecycle, batch auto-strategy, and
write-off line clearance. 67 total tests, 0 failed.

Made-with: Cursor
This commit is contained in:
gsinghpal
2026-04-19 10:50:46 -04:00
parent 920a624cd1
commit 80b8100232
5 changed files with 773 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
fusion_accounting_bank_rec/__manifest__.py
View File

@@ -1,6 +1,6 @@
{
'name': 'Fusion Accounting — Bank Reconciliation',
'version': '19.0.1.0.4',
'version': '19.0.1.0.5',
'category': 'Accounting/Accounting',
'sequence': 28,
'summary': 'Native V19 bank reconciliation widget with AI confidence scoring + behavioural learning.',

1
fusion_accounting_bank_rec/models/__init__.py
View File

@@ -4,3 +4,4 @@ from . import fusion_reconcile_suggestion
from . import fusion_bank_rec_widget
from . import account_bank_statement_line
from . import account_reconcile_model
from . import fusion_reconcile_engine

422
fusion_accounting_bank_rec/models/fusion_reconcile_engine.py Normal file
View File

@@ -0,0 +1,422 @@
"""The reconcile engine — orchestrator for all bank-line reconciliations.
Public API: 6 methods. All other code (controllers, AI tools, wizards)
must go through these methods; no direct ORM writes to
``account.partial.reconcile`` from anywhere else.
V19 mechanics (per Enterprise's bank_rec_widget pattern):
A bank statement line creates an ``account.move`` with two journal
items: a *liquidity* line on the journal's default account, and a
*suspense* line on the journal's suspense account. Reconciliation
replaces the suspense line with one or more *counterpart* lines posted
to the matched invoices' receivable / payable accounts (or the write-off
account), then calls Odoo's standard ``account.move.line.reconcile()``
on each counterpart + invoice pair.
Internal pipeline (per spec Section 3.3):
1. Validate (period not locked, mandatory args present).
2. Compute counterpart vals from ``against_lines`` and optional write-off.
3. Rewrite the bank move ``line_ids``: keep liquidity, drop suspense +
any prior other lines, append the new counterparts.
4. Reconcile each counterpart with its matched invoice line.
5. Audit (``mail.message``) + record precedent for future learning.
"""
import logging
from odoo import _, api, fields, models
from odoo.exceptions import ValidationError
from odoo.fields import Command
from ..services.matching_strategies import (
AmountExactStrategy,
Candidate,
FIFOStrategy,
MultiInvoiceStrategy,
)
from ..services.confidence_scoring import score_candidates
from ..services.memo_tokenizer import tokenize_memo
_logger = logging.getLogger(__name__)
class FusionReconcileEngine(models.AbstractModel):
_name = "fusion.reconcile.engine"
_description = "Fusion Bank Reconciliation Engine"
# ============================================================
# PUBLIC API (6 methods)
# ============================================================
@api.model
def reconcile_one(self, statement_line, *, against_lines=None,
write_off_vals=None):
"""Reconcile one bank line against a set of journal items.
Returns: ``{'partial_ids': [...], 'exchange_diff_move_id': int|None,
'write_off_move_id': int|None}``
"""
if not statement_line:
raise ValidationError(_("statement_line is required"))
statement_line.ensure_one()
AML = self.env['account.move.line']
against_lines = against_lines or AML
if not against_lines and not write_off_vals:
raise ValidationError(
_("Either against_lines or write_off_vals required"))
self._validate_reconcile(statement_line, against_lines)
bank_move = statement_line.move_id
liquidity_lines, suspense_lines, other_lines = (
statement_line._seek_for_lines())
# Build the new counterpart lines that replace suspense.
new_counterpart_vals = []
for inv_line in against_lines:
new_counterpart_vals.append(self._build_counterpart_vals(
statement_line, inv_line))
write_off_move_id = None
if write_off_vals:
new_counterpart_vals.append(self._build_write_off_vals(
statement_line, write_off_vals, against_lines))
# Replace the bank move line_ids: keep liquidity, drop everything
# else, append new counterparts.
ops = []
for line in (suspense_lines | other_lines):
ops.append(Command.unlink(line.id))
for vals in new_counterpart_vals:
ops.append(Command.create(vals))
editable_move = bank_move.with_context(
force_delete=True, skip_readonly_check=True)
prior_line_ids = set(bank_move.line_ids.ids)
editable_move.write({'line_ids': ops})
new_lines = bank_move.line_ids.filtered(
lambda line: line.id not in prior_line_ids)
# Reconcile each new counterpart with its matched invoice line.
Partial = self.env['account.partial.reconcile']
new_partial_ids = []
for new_line, inv_line in zip(
new_lines[:len(against_lines)], against_lines):
pair = new_line | inv_line
existing = set(Partial.search([
'|',
('debit_move_id', 'in', pair.ids),
('credit_move_id', 'in', pair.ids),
]).ids)
pair.reconcile()
added = Partial.search([
'|',
('debit_move_id', 'in', pair.ids),
('credit_move_id', 'in', pair.ids),
]).filtered(lambda p: p.id not in existing)
new_partial_ids.extend(added.ids)
self._post_audit(
statement_line, new_partial_ids, source='engine.reconcile_one')
if against_lines:
self._record_precedent(statement_line, against_lines)
return {
'partial_ids': new_partial_ids,
'exchange_diff_move_id': None,
'write_off_move_id': write_off_move_id,
}
@api.model
def reconcile_batch(self, statement_lines, *, strategy='auto'):
"""Bulk-reconcile a recordset using the chosen strategy.
Returns: ``{'reconciled_count': int, 'skipped': int,
'errors': [...]}``
"""
reconciled = 0
skipped = 0
errors = []
for line in statement_lines:
if line.is_reconciled:
skipped += 1
continue
try:
candidates = self._fetch_candidates(line)
picked = self._apply_strategy(line, candidates, strategy)
if picked:
self.reconcile_one(line, against_lines=picked)
reconciled += 1
else:
skipped += 1
except Exception as e: # noqa: BLE001
errors.append({'line_id': line.id, 'error': str(e)})
_logger.warning(
"reconcile_batch failed for line %s: %s", line.id, e)
return {
'reconciled_count': reconciled,
'skipped': skipped,
'errors': errors,
}
@api.model
def suggest_matches(self, statement_lines, *, limit_per_line=3):
"""Compute and persist AI suggestions per line.
Returns: dict mapping ``line_id`` -> list of suggestion dicts.
"""
out = {}
Suggestion = self.env['fusion.reconcile.suggestion']
for line in statement_lines:
candidates_records = self._fetch_candidates(line)
if not candidates_records:
continue
candidates_dataclasses = self._records_to_candidates(
line, candidates_records)
scored = score_candidates(
self.env,
statement_line=line,
candidates=candidates_dataclasses,
k=limit_per_line,
use_ai=True,
)
Suggestion.search([
('statement_line_id', '=', line.id),
('state', '=', 'pending'),
]).write({'state': 'superseded'})
line_suggestions = []
for rank, s in enumerate(scored, start=1):
sug = Suggestion.create({
'company_id': line.company_id.id,
'statement_line_id': line.id,
'proposed_move_line_ids': [(6, 0, [s.candidate_id])],
'confidence': s.confidence,
'rank': rank,
'reasoning': s.reasoning,
'score_amount_match': s.score_amount_match,
'score_partner_pattern': s.score_partner_pattern,
'score_precedent_similarity': s.score_precedent_similarity,
'score_ai_rerank': s.score_ai_rerank,
'generated_by': 'on_demand',
'state': 'pending',
})
line_suggestions.append({
'id': sug.id,
'rank': rank,
'confidence': s.confidence,
'reasoning': s.reasoning,
'candidate_id': s.candidate_id,
})
out[line.id] = line_suggestions
return out
@api.model
def accept_suggestion(self, suggestion):
"""User clicked Accept on a suggestion -> reconcile via its proposal.
Returns: same shape as ``reconcile_one``.
"""
if isinstance(suggestion, int):
suggestion = self.env['fusion.reconcile.suggestion'].browse(
suggestion)
suggestion.ensure_one()
line = suggestion.statement_line_id
against = suggestion.proposed_move_line_ids
result = self.reconcile_one(line, against_lines=against)
suggestion.write({
'state': 'accepted',
'accepted_at': fields.Datetime.now(),
'accepted_by': self.env.uid,
})
return result
@api.model
def write_off(self, statement_line, *, account, amount, label, tax_id=None):
"""Create a write-off move + reconcile the bank line against it.
Returns: same shape as ``reconcile_one``.
"""
write_off_vals = {
'account_id': account.id if hasattr(account, 'id') else account,
'amount': amount,
'tax_id': (tax_id.id if (tax_id and hasattr(tax_id, 'id'))
else tax_id),
'label': label,
}
return self.reconcile_one(
statement_line, against_lines=None, write_off_vals=write_off_vals)
@api.model
def unreconcile(self, partial_reconciles):
"""Reverse a reconciliation. Handles full vs. partial chains.
Returns: ``{'unreconciled_line_ids': [...]}``
"""
partial_reconciles = partial_reconciles.exists()
if not partial_reconciles:
return {'unreconciled_line_ids': []}
all_lines = (
partial_reconciles.mapped('debit_move_id')
| partial_reconciles.mapped('credit_move_id')
)
line_ids = all_lines.ids
partial_reconciles.unlink()
return {'unreconciled_line_ids': line_ids}
# ============================================================
# PRIVATE HELPERS
# ============================================================
def _validate_reconcile(self, statement_line, against_lines):
"""Phase 2: structural + safety checks."""
if not statement_line.exists():
raise ValidationError(_("Statement line does not exist"))
company = statement_line.company_id
line_date = statement_line.date
lock_date = company.fiscalyear_lock_date
if lock_date and line_date and line_date <= lock_date:
raise ValidationError(_(
"Cannot reconcile: line date %(line)s is on or before fiscal "
"year lock date %(lock)s",
line=line_date,
lock=lock_date,
))
def _build_counterpart_vals(self, statement_line, inv_line):
"""Build the vals for one counterpart line that mirrors an invoice
line on the bank move."""
return {
'name': inv_line.name or statement_line.payment_ref or '',
'account_id': inv_line.account_id.id,
'partner_id': (inv_line.partner_id.id
if inv_line.partner_id else False),
'currency_id': inv_line.currency_id.id,
'amount_currency': -inv_line.amount_residual_currency,
'balance': -inv_line.amount_residual,
}
def _build_write_off_vals(self, statement_line, write_off_vals,
against_lines):
"""Build the vals for a write-off counterpart line on the bank move.
The write-off absorbs the (signed) residual not covered by
``against_lines``: ``residual = bank_amount - sum(against_lines.balance)``.
We post that residual to the write-off account, with the opposite
sign so the bank move stays balanced.
"""
bank_amount = statement_line.amount
already_covered = sum(
-line.amount_residual for line in against_lines)
residual = bank_amount - already_covered
# The counterpart on the bank move must offset the liquidity line,
# so its balance is -residual.
wo_balance = -residual
# If the user explicitly passed an amount, prefer it (overrides).
if write_off_vals.get('amount') is not None and not against_lines:
wo_balance = -write_off_vals['amount']
vals = {
'name': write_off_vals.get('label') or _('Write-off'),
'account_id': write_off_vals['account_id'],
'partner_id': (statement_line.partner_id.id
if statement_line.partner_id else False),
'balance': wo_balance,
}
if write_off_vals.get('tax_id'):
vals['tax_ids'] = [(6, 0, [write_off_vals['tax_id']])]
return vals
def _fetch_candidates(self, statement_line):
"""SQL pre-filter: open journal items matching partner + reconcilable
account."""
domain = [
('parent_state', '=', 'posted'),
('account_id.reconcile', '=', True),
('reconciled', '=', False),
('display_type', 'not in', ('line_section', 'line_note')),
]
if statement_line.partner_id:
domain.append(('partner_id', '=', statement_line.partner_id.id))
return self.env['account.move.line'].search(domain, limit=200)
def _records_to_candidates(self, statement_line, records):
"""Convert ``account.move.line`` recordset to ``Candidate`` dataclasses."""
today = fields.Date.today()
result = []
for c in records:
ref_date = c.date_maturity or c.date or today
age_days = (today - ref_date).days
result.append(Candidate(
id=c.id,
amount=abs(c.amount_residual) or abs(c.balance),
partner_id=c.partner_id.id if c.partner_id else 0,
age_days=age_days,
))
return result
def _apply_strategy(self, line, candidate_records, strategy):
"""Apply the named strategy. Returns matching ``account.move.line``
recordset, or empty recordset if nothing matched."""
AML = self.env['account.move.line']
if not candidate_records:
return AML
candidate_dcs = self._records_to_candidates(line, candidate_records)
bank_amount = abs(line.amount)
if strategy == 'auto':
for strat_class in (AmountExactStrategy,
MultiInvoiceStrategy,
FIFOStrategy):
result = strat_class().match(
bank_amount=bank_amount, candidates=candidate_dcs)
if result.picked_ids:
return AML.browse(result.picked_ids)
return AML
def _post_audit(self, statement_line, partial_ids, source):
"""Append an audit log to the bank-line move's chatter."""
if not statement_line.move_id:
return
try:
statement_line.move_id.message_post(
body=_(
"Reconciled via %(source)s; %(count)d partial(s) created: "
"%(ids)s",
source=source,
count=len(partial_ids),
ids=partial_ids,
),
)
except Exception as e: # noqa: BLE001
_logger.debug(
"Audit log skipped for line %s: %s", statement_line.id, e)
def _record_precedent(self, statement_line, against_lines):
"""Append a precedent for future pattern learning. Best-effort."""
if not against_lines:
return
try:
self.env['fusion.reconcile.precedent'].sudo().create({
'company_id': statement_line.company_id.id,
'partner_id': (statement_line.partner_id.id
if statement_line.partner_id else False),
'amount': abs(statement_line.amount),
'currency_id': statement_line.currency_id.id,
'date': statement_line.date,
'memo_tokens': ','.join(
tokenize_memo(statement_line.payment_ref)),
'journal_id': statement_line.journal_id.id,
'matched_move_line_count': len(against_lines),
'matched_account_ids': ','.join(
str(i) for i in against_lines.mapped('account_id').ids),
'reconciler_user_id': self.env.uid,
'reconciled_at': fields.Datetime.now(),
'source': 'manual',
})
except Exception as e: # noqa: BLE001
_logger.warning(
"Failed to record precedent for line %s: %s",
statement_line.id, e)

1
fusion_accounting_bank_rec/tests/__init__.py
View File

@@ -5,3 +5,4 @@ from . import test_ai_suggestion_lifecycle
from . import test_precedent_lookup
from . import test_pattern_extraction
from . import test_confidence_scoring
from . import test_reconcile_engine_unit

348
fusion_accounting_bank_rec/tests/test_reconcile_engine_unit.py Normal file
View File

@@ -0,0 +1,348 @@
"""Unit tests for fusion.reconcile.engine — the 6-method public API.
Test layers:
- Layer 1: API surface (registry + method existence)
- Layer 2: unreconcile
- Layer 3: reconcile_one happy path
- Layer 4: accept_suggestion
- Layer 5: suggest_matches
- Layer 6: reconcile_batch
- Layer 7: write_off
Tests share a common setUpClass fixture providing a partner, bank
journal, statement, receivable account, and a small helper to mint a
posted customer invoice + bank statement line at given amounts.
"""
from datetime import date
from odoo.exceptions import ValidationError
from odoo.tests.common import TransactionCase, tagged
@tagged('post_install', '-at_install')
class TestReconcileEngineBase(TransactionCase):
"""Shared fixtures for engine tests."""
@classmethod
def setUpClass(cls):
super().setUpClass()
cls.engine = cls.env['fusion.reconcile.engine']
cls.company = cls.env.company
cls.currency = cls.company.currency_id
cls.partner = cls.env['res.partner'].create({
'name': 'Engine Test Partner',
})
cls.bank_journal = cls.env['account.journal'].create({
'name': 'Engine Test Bank',
'type': 'bank',
'code': 'ETBK',
'company_id': cls.company.id,
})
cls.sales_journal = cls.env['account.journal'].search([
('type', '=', 'sale'),
('company_id', '=', cls.company.id),
], limit=1)
if not cls.sales_journal:
cls.sales_journal = cls.env['account.journal'].create({
'name': 'Engine Test Sales',
'type': 'sale',
'code': 'ETSAL',
'company_id': cls.company.id,
})
cls.receivable_account = cls.env['account.account'].search([
('account_type', '=', 'asset_receivable'),
('company_ids', 'in', cls.company.id),
], limit=1)
cls.income_account = cls.env['account.account'].search([
('account_type', '=', 'income'),
('company_ids', 'in', cls.company.id),
], limit=1)
cls.expense_account = cls.env['account.account'].search([
('account_type', '=', 'expense'),
('company_ids', 'in', cls.company.id),
], limit=1)
def _make_statement_line(self, amount, *, partner=None, ref='ENGTEST',
line_date=None):
statement = self.env['account.bank.statement'].create({
'name': 'Engine Test Statement',
'journal_id': self.bank_journal.id,
})
return self.env['account.bank.statement.line'].create({
'statement_id': statement.id,
'journal_id': self.bank_journal.id,
'date': line_date or date.today(),
'payment_ref': ref,
'amount': amount,
'partner_id': (partner or self.partner).id,
})
def _make_invoice(self, amount, *, partner=None, inv_date=None):
"""Create + post a customer invoice for the given amount."""
inv = self.env['account.move'].create({
'move_type': 'out_invoice',
'partner_id': (partner or self.partner).id,
'invoice_date': inv_date or date.today(),
'journal_id': self.sales_journal.id,
'invoice_line_ids': [(0, 0, {
'name': 'Engine test product',
'quantity': 1,
'price_unit': amount,
'account_id': self.income_account.id,
'tax_ids': [(6, 0, [])],
})],
})
inv.action_post()
return inv
def _receivable_line(self, invoice):
return invoice.line_ids.filtered(
lambda line: line.account_id.account_type == 'asset_receivable'
)
# ============================================================
# Layer 1: API surface
# ============================================================
@tagged('post_install', '-at_install')
class TestReconcileEngineApi(TestReconcileEngineBase):
"""Layer 1: the engine class exists in the registry and exposes the
six expected methods."""
def test_engine_in_registry(self):
self.assertIn('fusion.reconcile.engine', self.env.registry)
def test_engine_is_abstract_model(self):
engine = self.env['fusion.reconcile.engine']
self.assertTrue(engine._abstract)
def test_six_public_methods_callable(self):
engine = self.env['fusion.reconcile.engine']
for name in ('reconcile_one', 'reconcile_batch', 'suggest_matches',
'accept_suggestion', 'write_off', 'unreconcile'):
self.assertTrue(callable(getattr(engine, name, None)),
f"engine.{name} must be callable")
def test_reconcile_one_requires_arguments(self):
line = self._make_statement_line(100.0)
with self.assertRaises(ValidationError):
self.engine.reconcile_one(line)
# ============================================================
# Layer 2: unreconcile
# ============================================================
@tagged('post_install', '-at_install')
class TestReconcileEngineUnreconcile(TestReconcileEngineBase):
def test_unreconcile_removes_partial_reconcile(self):
line = self._make_statement_line(100.0)
invoice = self._make_invoice(100.0)
receivable = self._receivable_line(invoice)
result = self.engine.reconcile_one(
line, against_lines=receivable)
self.assertTrue(result['partial_ids'],
"reconcile_one should produce partial_ids")
partials = self.env['account.partial.reconcile'].browse(
result['partial_ids']).exists()
self.assertTrue(partials)
out = self.engine.unreconcile(partials)
self.assertIn('unreconciled_line_ids', out)
self.assertTrue(out['unreconciled_line_ids'])
self.assertFalse(partials.exists(),
"Partials should be deleted after unreconcile")
receivable.invalidate_recordset(['reconciled', 'amount_residual'])
self.assertFalse(receivable.reconciled)
def test_unreconcile_empty_recordset_returns_empty(self):
empty = self.env['account.partial.reconcile']
out = self.engine.unreconcile(empty)
self.assertEqual(out, {'unreconciled_line_ids': []})
# ============================================================
# Layer 3: reconcile_one happy path
# ============================================================
@tagged('post_install', '-at_install')
class TestReconcileEngineReconcileOne(TestReconcileEngineBase):
def test_reconcile_one_simple_invoice_match(self):
line = self._make_statement_line(250.0)
invoice = self._make_invoice(250.0)
receivable = self._receivable_line(invoice)
self.assertFalse(receivable.reconciled)
result = self.engine.reconcile_one(
line, against_lines=receivable)
self.assertIsInstance(result, dict)
self.assertIn('partial_ids', result)
self.assertIn('exchange_diff_move_id', result)
self.assertIn('write_off_move_id', result)
self.assertTrue(result['partial_ids'])
receivable.invalidate_recordset(['reconciled', 'amount_residual'])
self.assertTrue(receivable.reconciled)
self.assertAlmostEqual(receivable.amount_residual, 0.0, places=2)
def test_reconcile_one_creates_precedent(self):
line = self._make_statement_line(125.0, ref='Engine REF#42')
invoice = self._make_invoice(125.0)
receivable = self._receivable_line(invoice)
before = self.env['fusion.reconcile.precedent'].search_count([
('partner_id', '=', self.partner.id),
])
self.engine.reconcile_one(line, against_lines=receivable)
after = self.env['fusion.reconcile.precedent'].search_count([
('partner_id', '=', self.partner.id),
])
self.assertEqual(after, before + 1)
# ============================================================
# Layer 4: accept_suggestion
# ============================================================
@tagged('post_install', '-at_install')
class TestReconcileEngineAcceptSuggestion(TestReconcileEngineBase):
def test_accept_suggestion_reconciles_and_marks_accepted(self):
line = self._make_statement_line(310.0)
invoice = self._make_invoice(310.0)
receivable = self._receivable_line(invoice)
sug = self.env['fusion.reconcile.suggestion'].create({
'company_id': self.company.id,
'statement_line_id': line.id,
'proposed_move_line_ids': [(6, 0, receivable.ids)],
'confidence': 0.97,
'rank': 1,
'reasoning': 'Exact amount match',
'state': 'pending',
})
result = self.engine.accept_suggestion(sug)
self.assertTrue(result['partial_ids'])
self.assertEqual(sug.state, 'accepted')
self.assertTrue(sug.accepted_at)
self.assertEqual(sug.accepted_by, self.env.user)
def test_accept_suggestion_by_id(self):
line = self._make_statement_line(75.0)
invoice = self._make_invoice(75.0)
receivable = self._receivable_line(invoice)
sug = self.env['fusion.reconcile.suggestion'].create({
'company_id': self.company.id,
'statement_line_id': line.id,
'proposed_move_line_ids': [(6, 0, receivable.ids)],
'confidence': 0.91,
'rank': 1,
'reasoning': 'OK',
'state': 'pending',
})
result = self.engine.accept_suggestion(sug.id)
self.assertTrue(result['partial_ids'])
self.assertEqual(sug.state, 'accepted')
# ============================================================
# Layer 5: suggest_matches
# ============================================================
@tagged('post_install', '-at_install')
class TestReconcileEngineSuggestMatches(TestReconcileEngineBase):
def test_suggest_matches_persists_pending_suggestions(self):
line = self._make_statement_line(420.0)
invoice = self._make_invoice(420.0)
# second open invoice for same partner — also a candidate
self._make_invoice(99.0)
out = self.engine.suggest_matches(line)
self.assertIn(line.id, out)
self.assertTrue(out[line.id])
suggestions = self.env['fusion.reconcile.suggestion'].search([
('statement_line_id', '=', line.id),
('state', '=', 'pending'),
])
self.assertTrue(suggestions)
# Top suggestion should reference the matching invoice's receivable
top = max(suggestions, key=lambda s: s.confidence)
receivable = self._receivable_line(invoice)
self.assertIn(receivable.id, top.proposed_move_line_ids.ids)
def test_suggest_matches_supersedes_prior_pending(self):
line = self._make_statement_line(180.0)
self._make_invoice(180.0)
old_sug = self.env['fusion.reconcile.suggestion'].create({
'company_id': self.company.id,
'statement_line_id': line.id,
'confidence': 0.5,
'rank': 1,
'reasoning': 'prior',
'state': 'pending',
})
self.engine.suggest_matches(line)
old_sug.invalidate_recordset(['state'])
self.assertEqual(old_sug.state, 'superseded')
def test_suggest_matches_returns_empty_for_no_candidates(self):
partner = self.env['res.partner'].create({'name': 'Empty Partner'})
line = self._make_statement_line(10.0, partner=partner)
out = self.engine.suggest_matches(line)
self.assertEqual(out, {})
# ============================================================
# Layer 6: reconcile_batch
# ============================================================
@tagged('post_install', '-at_install')
class TestReconcileEngineBatch(TestReconcileEngineBase):
def test_reconcile_batch_auto_strategy_matches_n_lines(self):
amounts = [100.0, 200.0, 333.33]
lines = self.env['account.bank.statement.line']
for amt in amounts:
invoice = self._make_invoice(amt)
self.assertTrue(invoice)
lines |= self._make_statement_line(amt, ref=f'BATCH-{amt}')
result = self.engine.reconcile_batch(lines, strategy='auto')
self.assertEqual(result['reconciled_count'], len(amounts))
self.assertEqual(result['skipped'], 0)
self.assertEqual(result['errors'], [])
def test_reconcile_batch_skips_already_reconciled(self):
line = self._make_statement_line(50.0)
invoice = self._make_invoice(50.0)
receivable = self._receivable_line(invoice)
self.engine.reconcile_one(line, against_lines=receivable)
result = self.engine.reconcile_batch(line, strategy='auto')
self.assertEqual(result['reconciled_count'], 0)
self.assertEqual(result['skipped'], 1)
# ============================================================
# Layer 7: write_off
# ============================================================
@tagged('post_install', '-at_install')
class TestReconcileEngineWriteOff(TestReconcileEngineBase):
def test_write_off_clears_bank_line(self):
line = self._make_statement_line(40.0, ref='Bank fee')
# No invoices exist; write off the whole amount to expense.
result = self.engine.write_off(
line,
account=self.expense_account,
amount=40.0,
label='Bank fees',
)
self.assertIn('write_off_move_id', result)
line.invalidate_recordset(['is_reconciled'])
self.assertTrue(line.is_reconciled)