a5db0fe71e
- SaleOrder._fc_rate_usage: aggregates usage, computes overage via charge._compute_billable, upserts sale.order.line for the overage product - FusionBillingUsage._cron_rate_open_periods: hourly cron iterates active charges × in-progress subscriptions, calls _fc_rate_usage - data/ir_cron.xml: two crons — rate usage (hourly), dispatch webhooks (2 min) - __manifest__.py: registers data/ir_cron.xml in data list - test_usage.py: test_rate_open_period_creates_overage_line (TDD, FCB_EXIT=0) Reference: _create_recurring_invoice / _get_invoiceable_lines confirmed in Enterprise sale_subscription/models/sale_order.py — overage line goes onto sale.order so native invoicing picks it up via _get_invoiceable_lines.
68 lines
3.6 KiB
Python
68 lines
3.6 KiB
Python
# -*- coding: utf-8 -*-
|
|
from odoo.tests.common import TransactionCase, tagged
|
|
|
|
|
|
@tagged('post_install', '-at_install')
|
|
class TestUsageIngestion(TransactionCase):
|
|
|
|
def setUp(self):
|
|
super().setUp()
|
|
self.metric = self.env['fusion.billing.metric'].sudo().create(
|
|
{'name': 'CPU seconds', 'code': 'cpu_seconds', 'aggregation': 'sum'})
|
|
self.plan = self.env['sale.subscription.plan'].sudo().create(
|
|
{'name': 'Monthly', 'billing_period_value': 1, 'billing_period_unit': 'month'})
|
|
self.partner = self.env['res.partner'].sudo().create({'name': 'Acme'})
|
|
self.sub = self.env['sale.order'].sudo().create({
|
|
'partner_id': self.partner.id, 'is_subscription': True, 'plan_id': self.plan.id,
|
|
})
|
|
self.Usage = self.env['fusion.billing.usage'].sudo()
|
|
|
|
def test_record_usage_creates_row(self):
|
|
u = self.Usage._record_usage(
|
|
self.sub, 'cpu_seconds', 120.0,
|
|
'2026-05-01 00:00:00', '2026-06-01 00:00:00', idem='nexacloud:cpu:sub1:2026-05-01')
|
|
self.assertEqual(u.quantity, 120.0)
|
|
self.assertEqual(u.metric_id, self.metric)
|
|
|
|
def test_idempotent_key_updates_not_duplicates(self):
|
|
k = 'nexacloud:cpu:sub1:2026-05-01'
|
|
self.Usage._record_usage(self.sub, 'cpu_seconds', 100.0, '2026-05-01', '2026-06-01', idem=k)
|
|
self.Usage._record_usage(self.sub, 'cpu_seconds', 175.0, '2026-05-01', '2026-06-01', idem=k)
|
|
rows = self.Usage.search([('idempotency_key', '=', k)])
|
|
self.assertEqual(len(rows), 1) # no duplicate
|
|
self.assertEqual(rows.quantity, 175.0) # last value wins for the same key
|
|
|
|
def test_aggregate_sum(self):
|
|
for i, q in enumerate([10.0, 20.0, 30.0]):
|
|
self.Usage._record_usage(self.sub, 'cpu_seconds', q,
|
|
'2026-05-01', '2026-06-01', idem='cpu-%d' % i)
|
|
total = self.Usage._aggregate(self.sub, self.metric, '2026-05-01', '2026-06-01')
|
|
self.assertEqual(total, 60.0)
|
|
|
|
def test_aggregate_max(self):
|
|
self.metric.aggregation = 'max'
|
|
for i, q in enumerate([10.0, 55.0, 30.0]):
|
|
self.Usage._record_usage(self.sub, 'cpu_seconds', q,
|
|
'2026-05-01', '2026-06-01', idem='m-%d' % i)
|
|
self.assertEqual(self.Usage._aggregate(self.sub, self.metric, '2026-05-01', '2026-06-01'), 55.0)
|
|
|
|
def test_aggregate_excludes_other_periods(self):
|
|
self.Usage._record_usage(self.sub, 'cpu_seconds', 99.0, '2026-04-01', '2026-05-01', idem='apr')
|
|
self.Usage._record_usage(self.sub, 'cpu_seconds', 5.0, '2026-05-01', '2026-06-01', idem='may')
|
|
self.assertEqual(self.Usage._aggregate(self.sub, self.metric, '2026-05-01', '2026-06-01'), 5.0)
|
|
|
|
def test_rate_open_period_creates_overage_line(self):
|
|
product = self.env['product.product'].sudo().create(
|
|
{'name': 'API overage', 'type': 'service', 'list_price': 0.0})
|
|
charge = self.env['fusion.billing.charge'].sudo().create({
|
|
'name': 'overage', 'plan_code': 'p', 'metric_id': self.metric.id,
|
|
'product_id': product.id, 'included_quota': 100.0,
|
|
'price_per_unit': 0.10, 'unit_batch': 1000.0, 'charge_model': 'standard'})
|
|
self.Usage._record_usage(self.sub, 'cpu_seconds', 1100.0,
|
|
'2026-05-01', '2026-06-01', idem='r1')
|
|
amount = self.sub._fc_rate_usage(charge, '2026-05-01', '2026-06-01')
|
|
# 1100 - 100 = 1000 overage = 1 batch * $0.10 = $0.10
|
|
self.assertAlmostEqual(amount, 0.10, places=2)
|
|
line = self.sub.order_line.filtered(lambda l: l.product_id == product)
|
|
self.assertTrue(line)
|