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>
100 lines
4.3 KiB
Python
100 lines
4.3 KiB
Python
# -*- coding: utf-8 -*-
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
from unittest.mock import patch
|
|
|
|
from odoo.exceptions import ValidationError
|
|
from odoo.tests.common import TransactionCase, tagged
|
|
|
|
|
|
@tagged('post_install', '-at_install')
|
|
class TestWebhookEngine(TransactionCase):
|
|
|
|
def setUp(self):
|
|
super().setUp()
|
|
self.service = self.env['fusion.billing.service'].sudo().create({
|
|
'name': 'NexaCloud', 'code': 'nexacloud',
|
|
'webhook_url': 'https://api.vps.nexasystems.ca/billing/webhook',
|
|
'webhook_secret': 'whsec_test',
|
|
})
|
|
self.Webhook = self.env['fusion.billing.webhook'].sudo()
|
|
|
|
def test_enqueue_signs_payload(self):
|
|
wh = self.Webhook._enqueue(self.service, 'invoice.payment_failed', {'invoice': 'INV-1'})
|
|
self.assertEqual(wh.state, 'pending')
|
|
body = json.dumps({'invoice': 'INV-1'}, sort_keys=True, separators=(',', ':'))
|
|
expected = hmac.new(b'whsec_test', body.encode(), hashlib.sha256).hexdigest()
|
|
self.assertEqual(wh.signature, expected)
|
|
|
|
def test_dispatch_marks_sent_on_2xx(self):
|
|
wh = self.Webhook._enqueue(self.service, 'invoice.paid', {'invoice': 'INV-2'})
|
|
|
|
class _Resp:
|
|
status_code = 200
|
|
text = 'ok'
|
|
|
|
with patch('odoo.addons.fusion_centralize_billing.models.webhook.requests.post',
|
|
return_value=_Resp()) as mock_post:
|
|
self.Webhook._cron_dispatch()
|
|
self.assertTrue(mock_post.called)
|
|
self.assertEqual(wh.state, 'sent')
|
|
|
|
def test_dispatch_retries_then_deadletters(self):
|
|
wh = self.Webhook._enqueue(self.service, 'invoice.paid', {'invoice': 'INV-3'})
|
|
wh.write({'attempts': 7}) # already past max
|
|
|
|
class _Resp:
|
|
status_code = 500
|
|
text = 'err'
|
|
|
|
with patch('odoo.addons.fusion_centralize_billing.models.webhook.requests.post',
|
|
return_value=_Resp()):
|
|
self.Webhook._cron_dispatch()
|
|
self.assertEqual(wh.state, 'dead')
|
|
|
|
# ── item 8 (H5): dispatch POSTs the stored body verbatim + event-id header ──
|
|
def test_dispatch_posts_stored_body_and_event_id(self):
|
|
wh = self.Webhook._enqueue(self.service, 'invoice.payment_failed', {'invoice': 'INV-9'})
|
|
|
|
class _Resp:
|
|
status_code = 200
|
|
text = 'ok'
|
|
|
|
with patch('odoo.addons.fusion_centralize_billing.models.webhook.requests.post',
|
|
return_value=_Resp()) as mock_post:
|
|
self.Webhook._cron_dispatch()
|
|
self.assertTrue(mock_post.called)
|
|
_args, kwargs = mock_post.call_args
|
|
# the exact stored body is POSTed (not a re-serialized payload)
|
|
self.assertEqual(kwargs['data'], wh.body)
|
|
self.assertEqual(wh.body, json.dumps(
|
|
{'invoice': 'INV-9'}, sort_keys=True, separators=(',', ':')))
|
|
# signature matches the bytes on the wire
|
|
expected = hmac.new(b'whsec_test', wh.body.encode(), hashlib.sha256).hexdigest()
|
|
self.assertEqual(kwargs['headers']['X-Fusion-Signature'], expected)
|
|
# event id header present and correct
|
|
self.assertEqual(kwargs['headers']['X-Fusion-Event-Id'], str(wh.id))
|
|
|
|
# ── item 9 (H6): SSRF guard on webhook_url ──
|
|
def test_webhook_url_rejects_loopback(self):
|
|
with self.assertRaises(ValidationError):
|
|
self.env['fusion.billing.service'].sudo().create({
|
|
'name': 'Evil', 'code': 'evil', 'webhook_url': 'http://127.0.0.1/x'})
|
|
|
|
def test_webhook_url_rejects_private_and_http(self):
|
|
for bad in ('http://10.0.0.5/hook', # private + non-https
|
|
'https://192.168.1.10/hook', # private
|
|
'https://localhost/hook', # localhost host
|
|
'https://169.254.169.254/latest', # link-local metadata
|
|
'http://api.example.com/hook'): # non-https
|
|
with self.assertRaises(ValidationError):
|
|
self.env['fusion.billing.service'].sudo().create({
|
|
'name': 'Bad', 'code': 'bad-%s' % bad, 'webhook_url': bad})
|
|
|
|
def test_webhook_url_allows_public_https(self):
|
|
svc = self.env['fusion.billing.service'].sudo().create({
|
|
'name': 'Good', 'code': 'good',
|
|
'webhook_url': 'https://api.vps.nexasystems.ca/billing/webhook'})
|
|
self.assertTrue(svc.id)
|