5605012245
Resolves findings from the post-build review: - C1: a partial import was indistinguishable from success. action_run_import now logs failed rows at ERROR (survives nexa's log_level=warn) and the wizard shows red/amber banners with failed/skipped counts. - H3: an unrecognized billing_cycle silently fell back to monthly (wrong plan AND price). Now raised per-row -> failed[], never silently mis-billed. - M5: a NULL plan price silently became a $0 line. Prices now preserve NULL-vs-0.0; a missing price for the subscription's cycle is failed[]. - H2: post-connect query/schema errors now become a clean UserError, not a raw SQL traceback (matches the connection-error path). - M4: per-row failures now record the exception type and log a traceback. - MED#3: charge plan_id set explicitly False so re-runs re-assert the shadow-safe NULL even if it was changed between runs. - HIGH-edge: re-run only rewrites x_fc_* on existing subs; partner_id/plan_id/ line are set at creation only (never rewrite immutable fields). - account_link: partner email match is now case-insensitive (=ilike) to avoid duplicate partners against a differently-cased pre-existing partner. Shadow-safety invariant unchanged and re-confirmed. 52/52 green on odoo-trial.
246 lines
12 KiB
Python
246 lines
12 KiB
Python
# -*- coding: utf-8 -*-
|
|
# Copyright 2026 Nexa Systems Inc.
|
|
# License OPL-1
|
|
from odoo.exceptions import UserError
|
|
from odoo.tests.common import TransactionCase, tagged
|
|
|
|
|
|
def _fixture():
|
|
"""Two users, one plan, two subscriptions (monthly + yearly) — the canonical
|
|
NexaCloud row dicts the importer consumes."""
|
|
return {
|
|
"users": [
|
|
{"id": "u-1", "email": "ar@acme.test", "full_name": "Acme Inc",
|
|
"company": "Acme", "billing_email": "billing@acme.test",
|
|
"billing_address": "1 Main St", "billing_city": "Toronto",
|
|
"billing_state": "ON", "billing_postal_code": "M1M1M1",
|
|
"billing_country": "CA", "tax_id": "123456789RT0001",
|
|
"stripe_customer_id": "cus_ACME"},
|
|
{"id": "u-2", "email": "ops@globex.test", "full_name": "Globex",
|
|
"company": "Globex", "billing_email": None, "billing_address": None,
|
|
"billing_city": None, "billing_state": None, "billing_postal_code": None,
|
|
"billing_country": None, "tax_id": None, "stripe_customer_id": "cus_GLBX"},
|
|
],
|
|
"plans": [
|
|
{"id": "p-1", "name": "Starter", "price_monthly": 20.0,
|
|
"price_yearly": 200.0, "cpu_seconds_quota": 18000.0, "is_active": True},
|
|
],
|
|
"subscriptions": [
|
|
{"id": "s-1", "user_id": "u-1", "deployment_id": "d-1", "plan_id": "p-1",
|
|
"status": "active", "billing_cycle": "monthly",
|
|
"current_period_start": "2026-05-01", "current_period_end": "2026-06-01"},
|
|
{"id": "s-2", "user_id": "u-2", "deployment_id": "d-2", "plan_id": "p-1",
|
|
"status": "active", "billing_cycle": "yearly",
|
|
"current_period_start": "2026-05-01", "current_period_end": "2027-05-01"},
|
|
],
|
|
}
|
|
|
|
|
|
@tagged('post_install', '-at_install')
|
|
class TestImporterIdentity(TransactionCase):
|
|
|
|
def setUp(self):
|
|
super().setUp()
|
|
self.Wizard = self.env['fusion.billing.import.wizard'].sudo()
|
|
self.Link = self.env['fusion.billing.account.link'].sudo()
|
|
|
|
def test_imports_users_as_partners_and_links(self):
|
|
self.Wizard._import_rows({'users': _fixture()['users']})
|
|
svc = self.env['fusion.billing.service'].search([('code', '=', 'nexacloud')])
|
|
self.assertTrue(svc, "importer must find-or-create the nexacloud service")
|
|
link1 = self.Link.search([('service_id', '=', svc.id), ('external_id', '=', 'u-1')])
|
|
self.assertEqual(len(link1), 1)
|
|
self.assertEqual(link1.partner_id.email, 'billing@acme.test') # billing_email wins
|
|
self.assertEqual(link1.partner_id.city, 'Toronto')
|
|
self.assertEqual(link1.partner_id.vat, '123456789RT0001')
|
|
self.assertEqual(link1.partner_id.x_fc_stripe_customer_id, 'cus_ACME')
|
|
self.assertEqual(link1.partner_id.country_id.code, 'CA')
|
|
link2 = self.Link.search([('service_id', '=', svc.id), ('external_id', '=', 'u-2')])
|
|
self.assertEqual(link2.partner_id.email, 'ops@globex.test') # falls back to email
|
|
|
|
|
|
@tagged('post_install', '-at_install')
|
|
class TestImporterCatalog(TransactionCase):
|
|
|
|
def setUp(self):
|
|
super().setUp()
|
|
self.Wizard = self.env['fusion.billing.import.wizard'].sudo()
|
|
|
|
def test_imports_plan_as_charge_with_null_plan_id(self):
|
|
self.Wizard._import_rows({'plans': _fixture()['plans']})
|
|
metric = self.env['fusion.billing.metric'].search([('code', '=', 'cpu_seconds')])
|
|
self.assertTrue(metric)
|
|
charge = self.env['fusion.billing.charge'].search([('plan_code', '=', 'p-1')])
|
|
self.assertEqual(len(charge), 1)
|
|
self.assertEqual(charge.metric_id, metric)
|
|
self.assertEqual(charge.included_quota, 18000.0) # = plan.cpu_seconds_quota
|
|
self.assertEqual(charge.unit_batch, 3600.0) # one core-hour
|
|
self.assertAlmostEqual(charge.price_per_unit, 0.0075) # CAD per core-hour
|
|
self.assertEqual(charge.charge_model, 'standard')
|
|
self.assertFalse(charge.plan_id, "shadow: charge.plan_id must be NULL so the "
|
|
"rating cron never auto-mutates order lines")
|
|
self.assertTrue(charge.product_id, "charge needs an overage product")
|
|
# the subscription product is a recurring product (so orders using it are subs)
|
|
sub_product = self.env['product.product'].search(
|
|
[('default_code', '=', 'NC-PLAN-p-1')])
|
|
self.assertTrue(sub_product.recurring_invoice)
|
|
|
|
def test_charge_math_matches_nexacloud(self):
|
|
# 18000 quota + 2 core-hours overage (7200s) -> 2 batches * $0.0075 = $0.015
|
|
self.Wizard._import_rows({'plans': _fixture()['plans']})
|
|
charge = self.env['fusion.billing.charge'].search([('plan_code', '=', 'p-1')])
|
|
_overage, amount = charge._compute_billable(18000.0 + 7200.0)
|
|
self.assertAlmostEqual(amount, 0.015, places=4)
|
|
|
|
|
|
@tagged('post_install', '-at_install')
|
|
class TestImporterSubscriptions(TransactionCase):
|
|
|
|
def setUp(self):
|
|
super().setUp()
|
|
self.Wizard = self.env['fusion.billing.import.wizard'].sudo()
|
|
|
|
def test_imports_one_draft_shadow_subscription_per_deployment(self):
|
|
self.Wizard._import_rows(_fixture())
|
|
SaleOrder = self.env['sale.order']
|
|
sub1 = SaleOrder.search([('x_fc_nexacloud_subscription_id', '=', 's-1')])
|
|
self.assertEqual(len(sub1), 1)
|
|
self.assertTrue(sub1.is_subscription)
|
|
self.assertTrue(sub1.x_fc_shadow)
|
|
self.assertEqual(sub1.x_fc_nexacloud_deployment_id, 'd-1')
|
|
self.assertNotEqual(sub1.subscription_state, '3_progress') # left in draft
|
|
plan_line = sub1.order_line.filtered(
|
|
lambda l: l.product_id.default_code == 'NC-PLAN-p-1')
|
|
self.assertEqual(len(plan_line), 1)
|
|
self.assertAlmostEqual(plan_line.price_unit, 20.0) # price_monthly
|
|
sub2 = SaleOrder.search([('x_fc_nexacloud_subscription_id', '=', 's-2')])
|
|
line2 = sub2.order_line.filtered(lambda l: l.product_id.default_code == 'NC-PLAN-p-1')
|
|
self.assertAlmostEqual(line2.price_unit, 200.0) # price_yearly
|
|
self.assertEqual(sub2.plan_id.billing_period_unit, 'year')
|
|
|
|
def test_subscription_skipped_when_user_or_plan_unresolved(self):
|
|
data = _fixture()
|
|
data['subscriptions'].append(
|
|
{"id": "s-3", "user_id": "u-missing", "deployment_id": "d-3", "plan_id": "p-1",
|
|
"status": "active", "billing_cycle": "monthly",
|
|
"current_period_start": "2026-05-01", "current_period_end": "2026-06-01"})
|
|
summary = self.Wizard._import_rows(data)
|
|
self.assertFalse(self.env['sale.order'].search(
|
|
[('x_fc_nexacloud_subscription_id', '=', 's-3')]))
|
|
self.assertTrue(any(s.get('id') == 's-3' for s in summary['skipped']))
|
|
|
|
|
|
@tagged('post_install', '-at_install')
|
|
class TestImporterIdempotencyDryRun(TransactionCase):
|
|
|
|
def setUp(self):
|
|
super().setUp()
|
|
self.Wizard = self.env['fusion.billing.import.wizard'].sudo()
|
|
|
|
def _counts(self):
|
|
return (
|
|
self.env['fusion.billing.account.link'].search_count([]),
|
|
self.env['fusion.billing.charge'].search_count([]),
|
|
self.env['sale.order'].search_count([('x_fc_shadow', '=', True)]),
|
|
)
|
|
|
|
def test_rerun_updates_not_duplicates(self):
|
|
self.Wizard._import_rows(_fixture())
|
|
before = self._counts()
|
|
data = _fixture()
|
|
data['plans'][0]['cpu_seconds_quota'] = 99999.0
|
|
self.Wizard._import_rows(data)
|
|
self.assertEqual(self._counts(), before, "re-run must upsert, not duplicate")
|
|
charge = self.env['fusion.billing.charge'].search([('plan_code', '=', 'p-1')])
|
|
self.assertEqual(charge.included_quota, 99999.0)
|
|
|
|
def test_dry_run_writes_nothing(self):
|
|
summary = self.Wizard._import_rows(_fixture(), dry_run=True)
|
|
self.assertTrue(summary.get('dry_run'))
|
|
self.assertEqual(self._counts(), (0, 0, 0), "dry-run must not persist anything")
|
|
self.assertFalse(
|
|
self.env['fusion.billing.service'].search([('code', '=', 'nexacloud')]))
|
|
|
|
|
|
@tagged('post_install', '-at_install')
|
|
class TestImporterShadowSafety(TransactionCase):
|
|
|
|
def setUp(self):
|
|
super().setUp()
|
|
self.Wizard = self.env['fusion.billing.import.wizard'].sudo()
|
|
|
|
def test_import_creates_no_invoice_and_no_payment_token(self):
|
|
self.Wizard._import_rows(_fixture())
|
|
subs = self.env['sale.order'].search([('x_fc_shadow', '=', True)])
|
|
self.assertTrue(subs)
|
|
partners = subs.mapped('partner_id')
|
|
invoices = self.env['account.move'].search([
|
|
('partner_id', 'in', partners.ids), ('move_type', '=', 'out_invoice')])
|
|
self.assertFalse(invoices, "shadow import must not create any invoice")
|
|
tokens = self.env['payment.token'].search([('partner_id', 'in', partners.ids)])
|
|
self.assertFalse(tokens, "shadow import must not attach a payment token")
|
|
charges = self.env['fusion.billing.charge'].search([('plan_code', '=', 'p-1')])
|
|
self.assertTrue(charges)
|
|
self.assertFalse(any(charges.mapped('plan_id')))
|
|
|
|
def test_rating_cron_leaves_shadow_subscriptions_untouched(self):
|
|
self.Wizard._import_rows(_fixture())
|
|
subs = self.env['sale.order'].search([('x_fc_shadow', '=', True)])
|
|
lines_before = sum(len(s.order_line) for s in subs)
|
|
self.env['fusion.billing.usage']._cron_rate_open_periods()
|
|
subs.invalidate_recordset()
|
|
lines_after = sum(len(s.order_line) for s in subs)
|
|
self.assertEqual(lines_before, lines_after,
|
|
"charges with NULL plan_id must keep the rating cron a no-op")
|
|
|
|
|
|
@tagged('post_install', '-at_install')
|
|
class TestImporterErrorIsolation(TransactionCase):
|
|
|
|
def setUp(self):
|
|
super().setUp()
|
|
self.Wizard = self.env['fusion.billing.import.wizard'].sudo()
|
|
|
|
def test_one_bad_user_does_not_abort_the_batch(self):
|
|
data = _fixture()
|
|
# a row with no id -> str(urow['id']) raises KeyError, must be caught per-row
|
|
data['users'].insert(0, {"email": "broken@x.test"})
|
|
summary = self.Wizard._import_rows(data)
|
|
self.assertEqual(
|
|
self.env['fusion.billing.account.link'].search_count([]), 2)
|
|
self.assertTrue(summary['failed'], "the bad row must be recorded in failed[]")
|
|
self.assertTrue(any(f['kind'] == 'user' for f in summary['failed']))
|
|
|
|
def test_unknown_billing_cycle_is_failed_not_silently_monthly(self):
|
|
data = _fixture()
|
|
data['subscriptions'][0]['billing_cycle'] = 'annual' # not monthly/yearly
|
|
summary = self.Wizard._import_rows(data)
|
|
self.assertFalse(self.env['sale.order'].search(
|
|
[('x_fc_nexacloud_subscription_id', '=', 's-1')]),
|
|
"an unrecognized billing_cycle must NOT silently create a monthly sub")
|
|
self.assertTrue(any(f['kind'] == 'subscription' and f['id'] == 's-1'
|
|
for f in summary['failed']))
|
|
|
|
def test_missing_price_for_cycle_is_failed_not_zero(self):
|
|
data = _fixture()
|
|
data['plans'][0]['price_yearly'] = None # s-2 is yearly -> no price for it
|
|
summary = self.Wizard._import_rows(data)
|
|
# the yearly sub fails (no silent $0 line); the monthly one still imports
|
|
self.assertFalse(self.env['sale.order'].search(
|
|
[('x_fc_nexacloud_subscription_id', '=', 's-2')]),
|
|
"a missing price for the cycle must NOT silently create a $0 line")
|
|
self.assertTrue(self.env['sale.order'].search(
|
|
[('x_fc_nexacloud_subscription_id', '=', 's-1')]))
|
|
self.assertTrue(any(f['kind'] == 'subscription' and f['id'] == 's-2'
|
|
for f in summary['failed']))
|
|
|
|
|
|
@tagged('post_install', '-at_install')
|
|
class TestImporterReadGuard(TransactionCase):
|
|
|
|
def test_missing_dsn_raises_usererror(self):
|
|
self.env['ir.config_parameter'].sudo().set_param('fusion_billing.nexacloud_dsn', '')
|
|
wiz = self.env['fusion.billing.import.wizard'].sudo().create({'dry_run': True})
|
|
with self.assertRaises(UserError):
|
|
wiz._read_nexacloud_rows()
|