# -*- coding: utf-8 -*- from odoo.tests.common import TransactionCase, tagged @tagged('post_install', '-at_install') class TestApiHandlers(TransactionCase): def setUp(self): super().setUp() self.service = self.env['fusion.billing.service'].sudo().create( {'name': 'NexaMaps', 'code': 'nexamaps'}) self.env['fusion.billing.metric'].sudo().create( {'name': 'API Calls', 'code': 'api_calls', 'aggregation': 'sum'}) self.plan = self.env['sale.subscription.plan'].sudo().create( {'name': 'Monthly', 'billing_period_value': 1, 'billing_period_unit': 'month'}) def test_api_upsert_customer(self): res = self.service._api_upsert_customer( {'external_id': 'client-9', 'name': 'Globex', 'email': 'billing@globex.test'}) self.assertEqual(res['status'], 'ok') link = self.env['fusion.billing.account.link'].search( [('service_id', '=', self.service.id), ('external_id', '=', 'client-9')]) self.assertEqual(link.partner_id.name, 'Globex') def test_api_record_usage_batch(self): self.service._api_upsert_customer({'external_id': 'client-9', 'name': 'Globex'}) partner = self.env['fusion.billing.account.link'].search( [('external_id', '=', 'client-9')]).partner_id sub = self.env['sale.order'].sudo().create( {'partner_id': partner.id, 'is_subscription': True, 'plan_id': self.plan.id}) res = self.service._api_record_usage({'events': [{ 'subscription_external_id': str(sub.id), 'metric_code': 'api_calls', 'quantity': 1234.0, 'period_start': '2026-05-01', 'period_end': '2026-06-01', 'idempotency_key': 'maps:client-9:2026-05-01', }]}) self.assertEqual(res['accepted'], 1) usage = self.env['fusion.billing.usage'].search([('subscription_id', '=', sub.id)]) self.assertEqual(usage.quantity, 1234.0) def test_api_catalog_lists_active_charges(self): self.env['fusion.billing.charge'].sudo().create({ 'name': 'Maps overage', 'plan_code': 'maps-business', 'metric_id': self.env['fusion.billing.metric'].search([('code', '=', 'api_calls')]).id, 'included_quota': 5_000_000.0, 'price_per_unit': 0.10, 'unit_batch': 1000.0}) cat = self.service._api_catalog() codes = [c['plan_code'] for c in cat['charges']] self.assertIn('maps-business', codes) def test_api_create_subscription(self): self.service._api_upsert_customer({'external_id': 'client-9', 'name': 'Globex'}) product = self.env['product.product'].sudo().create( {'name': 'Maps Business', 'type': 'service', 'recurring_invoice': True, 'list_price': 249.0}) res = self.service._api_create_subscription({ 'external_customer_id': 'client-9', 'plan_id': self.plan.id, 'lines': [{'product_id': product.id, 'quantity': 1}], }) self.assertEqual(res['status'], 'ok') sub = self.env['sale.order'].browse(res['subscription_id']) self.assertTrue(sub.is_subscription) self.assertEqual(sub.plan_id, self.plan) self.assertEqual(sub.subscription_state, '3_progress') # ── item 4 (C3): malformed input returns a 4xx-shaped error, never raises ── def test_record_usage_missing_metric_code_returns_error(self): self.service._api_upsert_customer({'external_id': 'client-9', 'name': 'Globex'}) partner = self.env['fusion.billing.account.link'].search( [('external_id', '=', 'client-9')]).partner_id sub = self.env['sale.order'].sudo().create( {'partner_id': partner.id, 'is_subscription': True, 'plan_id': self.plan.id}) # metric_code intentionally omitted — must return an error dict, not raise res = self.service._api_record_usage({'events': [{ 'subscription_external_id': str(sub.id), 'quantity': 10.0, 'period_start': '2026-05-01', 'period_end': '2026-06-01', }]}) self.assertEqual(res['status'], 'error') # no usage row written usage = self.env['fusion.billing.usage'].search([('subscription_id', '=', sub.id)]) self.assertFalse(usage) @tagged('post_install', '-at_install') class TestUsageAuthorization(TransactionCase): """/usage must only accept events for subscriptions the calling service is linked to, and reject unknown / non-subscription targets (items 3/C2/C4).""" def setUp(self): super().setUp() self.metric = self.env['fusion.billing.metric'].sudo().create( {'name': 'API Calls', 'code': 'api_calls', 'aggregation': 'sum'}) self.plan = self.env['sale.subscription.plan'].sudo().create( {'name': 'Monthly', 'billing_period_value': 1, 'billing_period_unit': 'month'}) self.service_a = self.env['fusion.billing.service'].sudo().create( {'name': 'Service A', 'code': 'svc_a'}) self.service_b = self.env['fusion.billing.service'].sudo().create( {'name': 'Service B', 'code': 'svc_b'}) # Service A owns customer + subscription self.service_a._api_upsert_customer({'external_id': 'cust-a', 'name': 'Cust A'}) self.partner_a = self.env['fusion.billing.account.link'].search( [('service_id', '=', self.service_a.id), ('external_id', '=', 'cust-a')]).partner_id self.sub_a = self.env['sale.order'].sudo().create( {'partner_id': self.partner_a.id, 'is_subscription': True, 'plan_id': self.plan.id}) self.Usage = self.env['fusion.billing.usage'].sudo() def _event(self, sub_id, idem): return {'events': [{ 'subscription_external_id': str(sub_id), 'metric_code': 'api_calls', 'quantity': 42.0, 'period_start': '2026-05-01', 'period_end': '2026-06-01', 'idempotency_key': idem, }]} def test_cross_service_usage_rejected(self): """Service B pushing usage onto Service A's subscription is rejected, no row.""" res = self.service_b._api_record_usage(self._event(self.sub_a.id, 'cross-1')) self.assertEqual(res['status'], 'error') self.assertEqual(res['error'], 'unknown subscription') self.assertFalse(self.Usage.search([('subscription_id', '=', self.sub_a.id)])) def test_same_service_usage_accepted(self): """Positive control: Service A pushing onto its own subscription is accepted.""" res = self.service_a._api_record_usage(self._event(self.sub_a.id, 'ok-1')) self.assertEqual(res['status'], 'ok') self.assertEqual(res['accepted'], 1) self.assertTrue(self.Usage.search([('subscription_id', '=', self.sub_a.id)])) def test_nonexistent_subscription_rejected(self): res = self.service_a._api_record_usage(self._event(999_999_999, 'ghost-1')) self.assertEqual(res['status'], 'error') self.assertEqual(res['error'], 'unknown subscription') def test_non_subscription_order_rejected(self): """A plain (non-subscription) sale.order owned by the linked customer is rejected.""" plain = self.env['sale.order'].sudo().create({'partner_id': self.partner_a.id}) self.assertFalse(plain.is_subscription) res = self.service_a._api_record_usage(self._event(plain.id, 'plain-1')) self.assertEqual(res['status'], 'error') self.assertEqual(res['error'], 'unknown subscription') self.assertFalse(self.Usage.search([('subscription_id', '=', plain.id)]))