Files
Odoo-Modules/fusion_centralize_billing/tests/test_reconciliation.py
gsinghpal 2bdf4ef6a0 feat(billing): 2d dual-run reconciliation (Odoo-computed vs NexaCloud-actual)
fusion.billing.reconciliation gains the compute: _compute_reconciliation
(flat + charge overage vs external, status match/delta at a tolerance) and
_reconcile_rows (resolve shadow sub -> flat + charge, upsert one row per
service/partner/period, per-row isolated). The wizard gains a read-only
_read_reconciliation_rows (NexaCloud usage cpu_hours*3600 + invoice-item
subtotals per YYYY-MM) and a "Run Reconciliation" button. 2a amended to
stamp x_fc_nexacloud_plan_id on shadow subs so reconciliation can find the
charge. Read-only on NexaCloud; writes only reconciliation rows (shadow
guarantees intact). 8 new tests, full suite green on odoo-trial.
2026-05-27 14:34:23 -04:00

92 lines
4.0 KiB
Python

# -*- coding: utf-8 -*-
# Copyright 2026 Nexa Systems Inc.
# License OPL-1
from odoo.tests.common import TransactionCase, tagged
from .test_importer import _fixture
@tagged('post_install', '-at_install')
class TestReconciliationMath(TransactionCase):
def setUp(self):
super().setUp()
self.Recon = self.env['fusion.billing.reconciliation'].sudo()
self.metric = self.env['fusion.billing.metric'].sudo().create(
{'name': 'CPU seconds', 'code': 'cpu_seconds', 'aggregation': 'sum'})
self.charge = self.env['fusion.billing.charge'].sudo().create({
'name': 'CPU', 'plan_code': 'p-1', 'metric_id': self.metric.id,
'included_quota': 18000.0, 'price_per_unit': 0.0075,
'unit_batch': 3600.0, 'charge_model': 'standard'})
def test_match_within_tolerance(self):
odoo_amt, delta, status = self.Recon._compute_reconciliation(
20.0, self.charge, 10000.0, 20.0, 0.01) # under quota, no overage
self.assertAlmostEqual(odoo_amt, 20.0)
self.assertEqual(status, 'match')
def test_overage_match(self):
# flat 20 + 2 core-hours overage (7200s -> $0.015) = 20.015; external 20.02 (cent)
odoo_amt, delta, status = self.Recon._compute_reconciliation(
20.0, self.charge, 18000.0 + 7200.0, 20.02, 0.01)
self.assertEqual(status, 'match')
def test_delta_flags_mismatch(self):
odoo_amt, delta, status = self.Recon._compute_reconciliation(
20.0, self.charge, 18000.0, 25.0, 0.01) # external 25 vs odoo 20
self.assertAlmostEqual(delta, -5.0, places=2)
self.assertEqual(status, 'delta')
def test_no_charge_is_flat_only(self):
odoo_amt, delta, status = self.Recon._compute_reconciliation(
20.0, self.env['fusion.billing.charge'], 999999.0, 20.0, 0.01)
self.assertAlmostEqual(odoo_amt, 20.0)
self.assertEqual(status, 'match')
@tagged('post_install', '-at_install')
class TestReconcileRows(TransactionCase):
def setUp(self):
super().setUp()
self.Wizard = self.env['fusion.billing.import.wizard'].sudo()
self.Wizard._import_rows(_fixture()) # shadow subs s-1/s-2 + p-1 charge
self.Recon = self.env['fusion.billing.reconciliation'].sudo()
self.SaleOrder = self.env['sale.order']
def _partner_of(self, sub_ext):
return self.SaleOrder.search(
[('x_fc_nexacloud_subscription_id', '=', sub_ext)]).partner_id
def test_creates_one_row_per_subscription_with_status(self):
summary = self.Recon._reconcile_rows([
{'subscription_external_id': 's-1', 'period': '2026-05',
'cpu_seconds': 0.0, 'external_amount': 20.0}, # flat 20 == 20 -> match
{'subscription_external_id': 's-2', 'period': '2026-05',
'cpu_seconds': 0.0, 'external_amount': 250.0}, # flat 200 vs 250 -> delta
])
rows = self.Recon.search([('period', '=', '2026-05')])
self.assertEqual(len(rows), 2)
s1 = rows.filtered(lambda r: r.odoo_amount == 20.0)
self.assertEqual(s1.status, 'match')
s2 = rows.filtered(lambda r: r.odoo_amount == 200.0)
self.assertEqual(s2.status, 'delta')
self.assertAlmostEqual(s2.delta, -50.0, places=2)
self.assertEqual(summary['match'], 1)
self.assertEqual(summary['delta'], 1)
def test_rerun_upserts(self):
row = [{'subscription_external_id': 's-1', 'period': '2026-05',
'cpu_seconds': 0.0, 'external_amount': 20.0}]
self.Recon._reconcile_rows(row)
self.Recon._reconcile_rows(row)
self.assertEqual(self.Recon.search_count([
('period', '=', '2026-05'),
('partner_id', '=', self._partner_of('s-1').id)]), 1)
def test_unknown_subscription_is_skipped(self):
summary = self.Recon._reconcile_rows([
{'subscription_external_id': 'nope', 'period': '2026-05',
'cpu_seconds': 0.0, 'external_amount': 1.0}])
self.assertTrue(any(s['id'] == 'nope' for s in summary['skipped']))