# -*- 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