Odoo-Modules/fusion_centralize_billing/tests/test_importer.py

# -*- 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_records_nexacloud_plan_id(self):
        self.Wizard._import_rows(_fixture())
        sub1 = self.env['sale.order'].search([('x_fc_nexacloud_subscription_id', '=', 's-1')])
        self.assertEqual(sub1.x_fc_nexacloud_plan_id, 'p-1')

    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()


@tagged('post_install', '-at_install')
class TestUsageApiSourceId(TransactionCase):
    """The /usage API must resolve a subscription by NexaCloud's OWN id, so usage can be
    pushed against shadow subs the importer created from UUIDs (the flip-day gap)."""

    def setUp(self):
        super().setUp()
        self.env['fusion.billing.import.wizard'].sudo()._import_rows(_fixture())
        self.service = self.env['fusion.billing.service'].search([('code', '=', 'nexacloud')])

    def test_record_usage_resolves_by_nexacloud_subscription_id(self):
        res = self.service._api_record_usage({'events': [{
            'subscription_external_id': 's-1',   # NexaCloud UUID, not the Odoo id
            'metric_code': 'cpu_seconds', 'quantity': 3600.0,
            'period_start': '2026-05-01', 'period_end': '2026-06-01',
            'idempotency_key': 'nc:s-1:2026-05'}]})
        self.assertEqual(res['status'], 'ok')
        self.assertEqual(res['accepted'], 1)
        sub = self.env['sale.order'].search([('x_fc_nexacloud_subscription_id', '=', 's-1')])
        usage = self.env['fusion.billing.usage'].search([('subscription_id', '=', sub.id)])
        self.assertEqual(usage.quantity, 3600.0)