e36318f7a5
The NexaCloud->Odoo ledger now verifies every new invoice against its
SOURCE billing system before posting, instead of trusting NexaCloud's
unreliable created_at/status/paid_at:
- _fc_verify routes by stripe_invoice_id prefix (in_ -> Stripe REST,
lago: -> Lago REST) and returns source-truth
{invoice_date, void, draft, paid, paid_at, amount_paid}, or None when it
can't be determined/reached (left for the next run).
- _ingest_invoices(post=True, verified=...) uses the source invoice date
(and accounting date), and reconciles a payment ONLY when the source
confirms paid.
- _cron_sync_verified posts only finalized invoices; skips void + draft,
logs unverified for retry. Replaces the old _cron_ingest_recent.
Cron cron_fc_invoice_ledger is enabled daily on nexamain. First live run:
23 already-posted, 1 void + 2 Stripe drafts + 5 zero-amount all skipped,
0 new posted, ledger intact at $3,403.46.
Tests: routing/guards (no network), verified date+reconcile, and the cron's
void/draft/unverified filtering (sources patched). FCB_EXIT=0 on odoo-trial.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
264 lines
14 KiB
Python
264 lines
14 KiB
Python
# -*- coding: utf-8 -*-
|
|
# Copyright 2026 Nexa Systems Inc.
|
|
# License OPL-1
|
|
from unittest.mock import patch
|
|
|
|
from odoo.exceptions import UserError
|
|
from odoo.tests.common import TransactionCase, tagged
|
|
|
|
|
|
def _inv_fixture():
|
|
return [{
|
|
'id': 'inv-1', 'stripe_invoice_id': 'in_test1', 'invoice_number': 'NEX-0001',
|
|
'user_external_id': 'u-1', 'partner_name': 'Acme', 'partner_email': 'ar@acme.test',
|
|
'invoice_date': '2026-05-01', 'currency': 'CAD', 'status': 'open',
|
|
'subtotal': 100.0, 'tax': 13.0, 'amount_paid': 0.0, 'paid_at': None,
|
|
'items': [{'description': 'Odoo ERP Hosting (2026-05-01 to 2026-06-01)',
|
|
'quantity': 1.0, 'unit_price': 100.0, 'amount': 100.0}],
|
|
}]
|
|
|
|
|
|
@tagged('post_install', '-at_install')
|
|
class TestLedgerFamily(TransactionCase):
|
|
|
|
def setUp(self):
|
|
super().setUp()
|
|
self.W = self.env['fusion.billing.invoice.ledger.wizard'].sudo()
|
|
|
|
def test_family_classification(self):
|
|
f = self.W._fc_family_for
|
|
self.assertEqual(f('Odoo ERP Hosting (2026-05-01 to 2026-06-01)'), 'hosting')
|
|
self.assertEqual(f('WordPress Website Hosting - Managed (at $50.00 / month)'), 'hosting')
|
|
self.assertEqual(f('Managed Odoo - Standard (at $49.99 / month)'), 'managed')
|
|
self.assertEqual(f('Daily Backup Protection'), 'addons')
|
|
self.assertEqual(f('Remaining time on Daily Backup Protection after 27 May 2026'), 'addons')
|
|
self.assertEqual(f('Something Unmapped'), 'other')
|
|
|
|
def test_income_account_per_family_distinct(self):
|
|
a_host = self.W._fc_income_account('hosting')
|
|
a_add = self.W._fc_income_account('addons')
|
|
self.assertEqual(a_host.account_type, 'income')
|
|
self.assertNotEqual(a_host, a_add)
|
|
self.assertEqual(self.W._fc_income_account('hosting'), a_host) # idempotent
|
|
|
|
|
|
@tagged('post_install', '-at_install')
|
|
class TestLedgerTax(TransactionCase):
|
|
|
|
def setUp(self):
|
|
super().setUp()
|
|
self.W = self.env['fusion.billing.invoice.ledger.wizard'].sudo()
|
|
|
|
def test_tax_for_13pct_is_a_13_percent_sale_tax(self):
|
|
tax = self.W._fc_tax_for(100.0, 13.0)
|
|
self.assertTrue(tax, "expected an HST/13% sale tax on the Canadian COA")
|
|
self.assertEqual(tax.type_tax_use, 'sale')
|
|
res = tax.compute_all(100.0)
|
|
self.assertAlmostEqual(res['total_included'] - res['total_excluded'], 13.0, places=2)
|
|
|
|
def test_tax_for_zero_is_zero_or_empty(self):
|
|
tax = self.W._fc_tax_for(100.0, 0.0)
|
|
if tax:
|
|
res = tax.compute_all(100.0)
|
|
self.assertAlmostEqual(res['total_included'] - res['total_excluded'], 0.0, places=2)
|
|
|
|
|
|
@tagged('post_install', '-at_install')
|
|
class TestLedgerIngest(TransactionCase):
|
|
|
|
def setUp(self):
|
|
super().setUp()
|
|
self.W = self.env['fusion.billing.invoice.ledger.wizard'].sudo()
|
|
self.Move = self.env['account.move']
|
|
|
|
def test_ingest_creates_draft_invoice_with_right_totals(self):
|
|
self.W._ingest_invoices(_inv_fixture(), post=False)
|
|
mv = self.Move.search([('x_fc_nexacloud_invoice_id', '=', 'inv-1')])
|
|
self.assertEqual(len(mv), 1)
|
|
self.assertEqual(mv.move_type, 'out_invoice')
|
|
self.assertEqual(mv.state, 'draft')
|
|
self.assertAlmostEqual(mv.amount_untaxed, 100.0, places=2)
|
|
self.assertAlmostEqual(mv.amount_tax, 13.0, places=2) # equals source tax
|
|
self.assertAlmostEqual(mv.amount_total, 113.0, places=2)
|
|
self.assertEqual(mv.partner_id.email, 'ar@acme.test')
|
|
self.assertEqual(mv.invoice_line_ids.account_id, self.W._fc_income_account('hosting'))
|
|
|
|
def test_ingest_is_idempotent(self):
|
|
self.W._ingest_invoices(_inv_fixture(), post=False)
|
|
self.W._ingest_invoices(_inv_fixture(), post=False)
|
|
self.assertEqual(self.Move.search_count(
|
|
[('x_fc_nexacloud_invoice_id', '=', 'inv-1')]), 1)
|
|
|
|
def test_paid_invoice_is_reconciled_and_shows_paid(self):
|
|
data = _inv_fixture()
|
|
data[0].update({'status': 'paid', 'amount_paid': 113.0, 'paid_at': '2026-05-02'})
|
|
self.W._ingest_invoices(data, post=True)
|
|
mv = self.Move.search([('x_fc_nexacloud_invoice_id', '=', 'inv-1')])
|
|
self.assertEqual(mv.state, 'posted')
|
|
self.assertIn(mv.payment_state, ('paid', 'in_payment'))
|
|
|
|
def test_post_ingested_posts_drafts(self):
|
|
self.W._ingest_invoices(_inv_fixture(), post=False)
|
|
n = self.W._post_ingested()
|
|
mv = self.Move.search([('x_fc_nexacloud_invoice_id', '=', 'inv-1')])
|
|
self.assertEqual(mv.state, 'posted')
|
|
self.assertGreaterEqual(n, 1)
|
|
|
|
def test_read_invoices_guards_missing_dsn(self):
|
|
self.env['ir.config_parameter'].sudo().set_param('fusion_billing.nexacloud_dsn', '')
|
|
with self.assertRaises(UserError):
|
|
self.W._read_nexacloud_invoices()
|
|
|
|
def test_unitemized_subtotal_gets_reconciling_line(self):
|
|
data = [{
|
|
'id': 'inv-base', 'stripe_invoice_id': 'in_base', 'invoice_number': 'NEX-BASE',
|
|
'user_external_id': 'u-2', 'partner_name': 'Globex', 'partner_email': 'ops@globex.test',
|
|
'invoice_date': '2026-05-01', 'currency': 'CAD', 'status': 'open',
|
|
'subtotal': 200.0, 'tax': 0.0, 'amount_paid': 0.0, 'paid_at': None,
|
|
'items': [], # base plan billed via Stripe only — no line items
|
|
}]
|
|
self.W._ingest_invoices(data, post=False)
|
|
mv = self.Move.search([('x_fc_nexacloud_invoice_id', '=', 'inv-base')])
|
|
self.assertAlmostEqual(mv.amount_untaxed, 200.0, places=2) # captured via reconciling line
|
|
self.assertTrue(any('base/unitemized' in (l.name or '') for l in mv.invoice_line_ids))
|
|
|
|
def test_zero_amount_invoice_skipped(self):
|
|
data = [{'id': 'inv-zero', 'stripe_invoice_id': 'in_z', 'invoice_number': 'NEX-ZERO',
|
|
'user_external_id': 'u-1', 'partner_name': 'Acme', 'partner_email': 'ar@acme.test',
|
|
'invoice_date': '2026-05-01', 'currency': 'CAD', 'status': 'paid',
|
|
'subtotal': 0.0, 'tax': 0.0, 'amount_paid': 0.0, 'paid_at': None, 'items': []}]
|
|
summary = self.W._ingest_invoices(data, post=False)
|
|
self.assertFalse(self.Move.search([('x_fc_nexacloud_invoice_id', '=', 'inv-zero')]))
|
|
self.assertTrue(any(s.get('reason') == 'zero-amount invoice' for s in summary['skipped']))
|
|
|
|
def test_post_and_reconcile_paid_only(self):
|
|
base = _inv_fixture()[0]
|
|
paid = dict(base, id='inv-paid', invoice_number='NEX-PAID',
|
|
status='paid', amount_paid=113.0, paid_at='2026-05-02',
|
|
invoice_date='2026-05-01')
|
|
unpaid = dict(base, id='inv-unpaid', invoice_number='NEX-UNPAID',
|
|
status='open', amount_paid=0.0, invoice_date='2026-04-01')
|
|
self.W._ingest_invoices([paid, unpaid], post=False)
|
|
summary = self.W._post_and_reconcile_paid([paid, unpaid])
|
|
pm = self.Move.search([('x_fc_nexacloud_invoice_id', '=', 'inv-paid')])
|
|
um = self.Move.search([('x_fc_nexacloud_invoice_id', '=', 'inv-unpaid')])
|
|
self.assertEqual(pm.state, 'posted')
|
|
self.assertIn(pm.payment_state, ('paid', 'in_payment'))
|
|
self.assertEqual(str(pm.invoice_date), '2026-05-01') # original invoice date kept
|
|
self.assertEqual(um.state, 'draft') # unpaid stays draft
|
|
self.assertEqual(summary['posted'], 1)
|
|
self.assertEqual(summary['skipped_unpaid'], 1)
|
|
|
|
def test_partner_named_by_company_not_person(self):
|
|
data = _inv_fixture()
|
|
data[0]['partner_company'] = 'Acme Holdings Inc' # full_name is "Acme"; company wins
|
|
self.W._ingest_invoices(data, post=False)
|
|
mv = self.Move.search([('x_fc_nexacloud_invoice_id', '=', 'inv-1')])
|
|
self.assertEqual(mv.partner_id.name, 'Acme Holdings Inc')
|
|
self.assertTrue(mv.partner_id.is_company)
|
|
|
|
def test_prune_shadow_removes_shadow_subs_only(self):
|
|
p = self.env['res.partner'].sudo().create({'name': 'X'})
|
|
shadow = self.env['sale.order'].sudo().create({'partner_id': p.id, 'x_fc_shadow': True})
|
|
counts = self.W._fc_prune_metered_shadow()
|
|
self.assertFalse(shadow.exists())
|
|
self.assertGreaterEqual(counts.get('subscriptions', 0), 1)
|
|
|
|
|
|
@tagged('post_install', '-at_install')
|
|
class TestLedgerVerifiedSync(TransactionCase):
|
|
"""The go-forward path: invoice date + paid status come from the SOURCE billing
|
|
system (Stripe/Lago), never NexaCloud's own fields. HTTP is never hit in tests —
|
|
routing short-circuits when no API credentials are configured, and the cron is
|
|
exercised with _read_nexacloud_invoices / _fc_verify patched out."""
|
|
|
|
def setUp(self):
|
|
super().setUp()
|
|
self.W = self.env['fusion.billing.invoice.ledger.wizard'].sudo()
|
|
self.Move = self.env['account.move']
|
|
ICP = self.env['ir.config_parameter'].sudo()
|
|
# ensure no real credentials -> verify helpers short-circuit, never touch network
|
|
ICP.set_param('fusion_billing.stripe_api_key', '')
|
|
ICP.set_param('fusion_billing.lago_api_url', '')
|
|
ICP.set_param('fusion_billing.lago_api_key', '')
|
|
|
|
def test_ts_to_date_is_utc_and_none_safe(self):
|
|
self.assertEqual(self.W._fc_ts_to_date(0), '1970-01-01')
|
|
self.assertEqual(self.W._fc_ts_to_date(86400), '1970-01-02')
|
|
self.assertIsNone(self.W._fc_ts_to_date(None))
|
|
|
|
def test_verify_routes_and_guards_without_network(self):
|
|
# Stripe id with no key, Lago id with no config, and an unroutable id all -> None
|
|
self.assertIsNone(self.W._fc_verify({'stripe_invoice_id': 'in_abc'}))
|
|
self.assertIsNone(self.W._fc_verify({'stripe_invoice_id': 'lago:xyz'}))
|
|
self.assertIsNone(self.W._fc_verify({'stripe_invoice_id': 'mystery'}))
|
|
self.assertIsNone(self.W._fc_verify({'stripe_invoice_id': None}))
|
|
|
|
def test_verified_paid_uses_source_date_and_reconciles(self):
|
|
v = {'inv-1': {'invoice_date': '2026-02-10', 'void': False, 'paid': True,
|
|
'paid_at': '2026-02-12', 'amount_paid': 113.0}}
|
|
self.W._ingest_invoices(_inv_fixture(), post=True, verified=v)
|
|
mv = self.Move.search([('x_fc_nexacloud_invoice_id', '=', 'inv-1')])
|
|
self.assertEqual(mv.state, 'posted')
|
|
self.assertEqual(str(mv.invoice_date), '2026-02-10') # source date, not NexaCloud's
|
|
self.assertEqual(str(mv.date), str(mv.invoice_date)) # accounting date tracks it
|
|
self.assertIn(mv.payment_state, ('paid', 'in_payment'))
|
|
|
|
def test_verified_unpaid_posts_but_is_not_reconciled(self):
|
|
v = {'inv-1': {'invoice_date': '2026-04-01', 'void': False, 'paid': False,
|
|
'paid_at': None, 'amount_paid': 0.0}}
|
|
self.W._ingest_invoices(_inv_fixture(), post=True, verified=v)
|
|
mv = self.Move.search([('x_fc_nexacloud_invoice_id', '=', 'inv-1')])
|
|
self.assertEqual(mv.state, 'posted')
|
|
self.assertEqual(str(mv.invoice_date), '2026-04-01')
|
|
self.assertEqual(mv.payment_state, 'not_paid')
|
|
|
|
def test_cron_skips_void_draft_unverified_posts_only_finalized(self):
|
|
base = _inv_fixture()[0]
|
|
fixtures = [
|
|
dict(base, id='inv-paid', invoice_number='NEX-P', stripe_invoice_id='in_paid'),
|
|
dict(base, id='inv-void', invoice_number='NEX-V', stripe_invoice_id='in_void'),
|
|
dict(base, id='inv-draft', invoice_number=None, stripe_invoice_id='in_draft'),
|
|
dict(base, id='inv-unver', invoice_number='NEX-U', stripe_invoice_id='weird'),
|
|
]
|
|
verdicts = {
|
|
'inv-paid': {'invoice_date': '2026-03-01', 'void': False, 'draft': False,
|
|
'paid': True, 'paid_at': '2026-03-02', 'amount_paid': 113.0},
|
|
'inv-void': {'invoice_date': '2026-03-01', 'void': True, 'draft': False,
|
|
'paid': False, 'paid_at': None, 'amount_paid': 0.0},
|
|
'inv-draft': {'invoice_date': '2026-03-01', 'void': False, 'draft': True,
|
|
'paid': False, 'paid_at': None, 'amount_paid': 0.0},
|
|
}
|
|
cls = type(self.W)
|
|
with patch.object(cls, '_read_nexacloud_invoices', return_value=fixtures), \
|
|
patch.object(cls, '_fc_verify',
|
|
side_effect=lambda inv: verdicts.get(str(inv.get('id')))):
|
|
summary = self.W._cron_sync_verified()
|
|
self.assertEqual(summary['skipped_void'], 1)
|
|
self.assertEqual(summary['skipped_draft'], 1)
|
|
self.assertEqual(summary['unverified'], ['inv-unver'])
|
|
self.assertEqual(summary['posted'], 1)
|
|
self.assertEqual(summary['reconciled'], 1)
|
|
paid = self.Move.search([('x_fc_nexacloud_invoice_id', '=', 'inv-paid')])
|
|
self.assertEqual(paid.state, 'posted')
|
|
self.assertEqual(str(paid.invoice_date), '2026-03-01')
|
|
self.assertIn(paid.payment_state, ('paid', 'in_payment'))
|
|
for skipped in ('inv-void', 'inv-draft', 'inv-unver'):
|
|
self.assertFalse(self.Move.search([('x_fc_nexacloud_invoice_id', '=', skipped)]))
|
|
|
|
def test_cron_leaves_already_posted_untouched(self):
|
|
# first run posts inv-paid; second run must not re-touch it (idempotent)
|
|
base = _inv_fixture()[0]
|
|
fixtures = [dict(base, id='inv-x', invoice_number='NEX-X', stripe_invoice_id='in_x')]
|
|
verdict = {'invoice_date': '2026-03-01', 'void': False, 'paid': True,
|
|
'paid_at': '2026-03-02', 'amount_paid': 113.0}
|
|
cls = type(self.W)
|
|
with patch.object(cls, '_read_nexacloud_invoices', return_value=fixtures), \
|
|
patch.object(cls, '_fc_verify', side_effect=lambda inv: verdict):
|
|
self.W._cron_sync_verified()
|
|
summary2 = self.W._cron_sync_verified()
|
|
self.assertEqual(summary2['already_posted'], 1)
|
|
self.assertEqual(summary2['posted'], 0)
|
|
self.assertEqual(self.Move.search_count(
|
|
[('x_fc_nexacloud_invoice_id', '=', 'inv-x')]), 1)
|