a82f09ea50
- CRITICAL: reconciliation upsert keyed on (service, partner, period) collided when one customer has two deployments (two subs) in a period — the second overwrote the first. Add external_subscription_id to the model + a UNIQUE(service_id, external_subscription_id, period) constraint, and key the upsert per subscription. New test proves two subs for one partner keep two rows. - raise a clear error if the nexacloud service is missing (was a confusing per-row failure). - _fc_resolve_subscription: the integer fallback no longer reaches a different service's tagged subscription (latent multi-service IDOR); live untagged subs stay resolvable and the partner-link authz is unchanged. Full suite green on odoo-trial.
112 lines
5.1 KiB
Python
112 lines
5.1 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']))
|
|
|
|
def test_two_subscriptions_same_partner_period_do_not_collide(self):
|
|
# A customer with two deployments -> two subscriptions in the same period.
|
|
data = _fixture()
|
|
data['subscriptions'].append(
|
|
{"id": "s-1b", "user_id": "u-1", "deployment_id": "d-1b", "plan_id": "p-1",
|
|
"status": "active", "billing_cycle": "monthly",
|
|
"current_period_start": "2026-05-01", "current_period_end": "2026-06-01"})
|
|
self.env['fusion.billing.import.wizard'].sudo()._import_rows(data)
|
|
self.Recon._reconcile_rows([
|
|
{'subscription_external_id': 's-1', 'period': '2026-05',
|
|
'cpu_seconds': 0.0, 'external_amount': 20.0},
|
|
{'subscription_external_id': 's-1b', 'period': '2026-05',
|
|
'cpu_seconds': 0.0, 'external_amount': 99.0},
|
|
])
|
|
partner = self._partner_of('s-1')
|
|
rows = self.Recon.search(
|
|
[('partner_id', '=', partner.id), ('period', '=', '2026-05')])
|
|
self.assertEqual(len(rows), 2, "two subs for one partner must keep two rows")
|
|
self.assertEqual(set(rows.mapped('external_subscription_id')), {'s-1', 's-1b'})
|