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.
92 lines
4.0 KiB
Python
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']))
|