# -*- 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() def test_test_connection_guards_missing_dsn(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.action_test_connection()