d770c0c3a9
- C1/H4: rating cron only rates subs on the charge's own plan_id
- C1: _fc_rate_usage skips creating a line when amount is 0 (still updates existing)
- C2/C4: /usage authorizes each event (exists + is_subscription + linked customer)
- C3: API handlers validate input and return 4xx-shaped errors instead of raising;
controller maps status=='error' to HTTP 400
- H1: cron uses real billing window [last_invoice_date or start_date, next_invoice_date)
- H2: _aggregate uses half-open window anchored on period_start
- H3: idempotency scoped to (subscription_id, metric_id, idempotency_key)
- H5: webhook stores canonical body, signs+POSTs it verbatim, adds X-Fusion-Event-Id,
caps backoff at 2**min(attempts,10)
- H6: SSRF guard rejects non-https / localhost / private / link-local webhook_url
- M7: charge_model reduced to standard/package (dropped unimplemented graduated/volume)
- L1: currency_id required on charge + reconciliation
- L2: charge price non-negative + unit_batch positive DB constraints
Adds 17 regression tests (suite 22 -> 39, all green via fcb_test_on_trial.sh).
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
140 lines
7.4 KiB
Python
140 lines
7.4 KiB
Python
# -*- 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)]))
|